Track blob garbage during flush of direct-write blob references (#14623)

Summary: - meter blob garbage during `BuildTable()` when flush input can contain direct-write blob references, so overwrite elision and flush-time compaction filters register garbage in the manifest - plumb `BlobFileGarbage` through flush and recovery manifest edits even when the flush produces no SST output - add wide-column direct-write tests covering TTL-filtered flushes with both mixed live/expired records and all-expired input Pull Request resolved: https://github.com/facebook/rocksdb/pull/14623 Reviewed By: anand1976 Differential Revision: D101212841 Pulled By: xingbowang fbshipit-source-id: e5803ddaf17b5dfd92312e42191b76e311257f3b
2026-07-07 14:47:40 +08:00 · 2026-04-16 16:45:13 -07:00
parent df5695cf24
commit 7b06a9ae7e
5 changed files with 324 additions and 19 deletions
@@ -13,7 +13,9 @@
 #include <deque>
 #include <vector>

+#include "db/blob/blob_counting_iterator.h"
 #include "db/blob/blob_file_builder.h"
+#include "db/blob/blob_garbage_meter.h"
 #include "db/compaction/compaction_iterator.h"
 #include "db/dbformat.h"
 #include "db/event_helpers.h"
@@ -87,7 +89,8 @@ Status BuildTable(
    Env::WriteLifeTimeHint write_hint, const std::string* full_history_ts_low,
    BlobFileCompletionCallback* blob_callback, Version* version,
    uint64_t* memtable_payload_bytes, uint64_t* memtable_garbage_bytes,
-    InternalStats::CompactionStats* flush_stats, bool fast_sst_open) {
+    InternalStats::CompactionStats* flush_stats,
+    std::vector<BlobFileGarbage>* blob_file_garbages, bool fast_sst_open) {
  assert((tboptions.column_family_id ==
          TablePropertiesCollectorFactory::Context::kUnknownColumnFamily) ==
         tboptions.column_family_name.empty());
@@ -99,7 +102,24 @@ Status BuildTable(
                                   /*enable_hash=*/paranoid_file_checks);
  Status s;
  meta->fd.file_size = 0;
-  iter->SeekToFirst();
+  if (blob_file_garbages != nullptr) {
+    blob_file_garbages->clear();
+  }
+
+  InternalIterator* input = iter;
+  std::unique_ptr<BlobGarbageMeter> blob_garbage_meter;
+  std::unique_ptr<BlobCountingIterator> blob_counting_iter;
+  if (blob_file_garbages != nullptr) {
+    // Flush can drop blob-backed records via overwrite elision and flush-time
+    // compaction filters. Track input and surviving output refs just like the
+    // compaction path so the manifest learns about the new garbage.
+    blob_garbage_meter.reset(new BlobGarbageMeter());
+    blob_counting_iter.reset(
+        new BlobCountingIterator(iter, blob_garbage_meter.get()));
+    input = blob_counting_iter.get();
+  }
+
+  input->SeekToFirst();
  std::unique_ptr<CompactionRangeDelAggregator> range_del_agg(
      new CompactionRangeDelAggregator(&tboptions.internal_comparator,
                                       snapshots, full_history_ts_low));
@@ -128,7 +148,7 @@ Status BuildTable(

  TableProperties tp;
  bool table_file_created = false;
-  if (iter->Valid() || !range_del_agg->IsEmpty()) {
+  if (input->Valid() || !range_del_agg->IsEmpty()) {
    std::unique_ptr<CompactionFilter> compaction_filter;
    if (ioptions.compaction_filter_factory != nullptr &&
        ioptions.compaction_filter_factory->ShouldFilterTableFileCreation(
@@ -211,7 +231,7 @@ Status BuildTable(

    const std::atomic<bool> kManualCompactionCanceledFalse{false};
    CompactionIterator c_iter(
-        iter, ucmp, &merge, kMaxSequenceNumber, &snapshots, earliest_snapshot,
+        input, ucmp, &merge, kMaxSequenceNumber, &snapshots, earliest_snapshot,
        earliest_write_conflict_snapshot, job_snapshot, snapshot_checker, env,
        ShouldReportDetailedTime(env, ioptions.stats), range_del_agg.get(),
        blob_file_builder.get(), ioptions.allow_data_in_errors,
@@ -268,6 +288,14 @@ Status BuildTable(
        flush_stats->num_output_records++;
      }

+      if (blob_garbage_meter != nullptr) {
+        s = blob_garbage_meter->ProcessOutFlow(key_after_flush,
+                                               value_after_flush);
+        if (!s.ok()) {
+          break;
+        }
+      }
+
      s = meta->UpdateBoundaries(key_after_flush, value_after_flush,
                                 ikey.sequence, ikey.type);
      if (!s.ok()) {
@@ -490,8 +518,20 @@ Status BuildTable(
  }

  // Check for input iterator errors
-  if (!iter->status().ok()) {
-    s = iter->status();
+  if (!input->status().ok()) {
+    s = input->status();
+  }
+
+  if (s.ok() && blob_file_garbages != nullptr &&
+      blob_garbage_meter != nullptr) {
+    for (const auto& pair : blob_garbage_meter->flows()) {
+      const uint64_t blob_file_number = pair.first;
+      const BlobGarbageMeter::BlobInOutFlow& flow = pair.second;
+      if (flow.HasGarbage()) {
+        blob_file_garbages->emplace_back(
+            blob_file_number, flow.GetGarbageCount(), flow.GetGarbageBytes());
+      }
+    }
  }

  if (!s.ok() || meta->fd.GetFileSize() == 0) {
@@ -31,6 +31,7 @@ struct FileMetaData;

 class VersionSet;
 class BlobFileAddition;
+class BlobFileGarbage;
 class SnapshotChecker;
 class TableCache;
 class TableBuilder;
@@ -80,6 +81,7 @@ Status BuildTable(
    Version* version = nullptr, uint64_t* memtable_payload_bytes = nullptr,
    uint64_t* memtable_garbage_bytes = nullptr,
    InternalStats::CompactionStats* flush_stats = nullptr,
+    std::vector<BlobFileGarbage>* blob_file_garbages = nullptr,
    bool fast_sst_open = false);

 }  // namespace ROCKSDB_NAMESPACE
@@ -889,6 +889,13 @@ Status FlushJob::WriteLevel0Table() {
      ts_sz > 0 && !cfd_->ioptions().persist_user_defined_timestamps;

  std::vector<BlobFileAddition> blob_file_additions;
+  std::vector<BlobFileGarbage> blob_file_garbages;
+  // Only direct-write memtables can carry blob indexes that reference
+  // pre-existing blob files. Plain flushes write new blob files from inline
+  // values, so there is no pre-existing blob garbage to meter on the input
+  // side.
+  std::vector<BlobFileGarbage>* const blob_file_garbages_for_filtering =
+      cfd_->blob_partition_manager() != nullptr ? &blob_file_garbages : nullptr;
  // Note that here we treat flush as level 0 compaction in internal stats
  InternalStats::CompactionStats flush_stats(CompactionReason::kFlush,
                                             1 /* count**/);
@@ -1033,7 +1040,7 @@ Status FlushJob::WriteLevel0Table() {
          seqno_to_time_mapping_.get(), event_logger_, job_context_->job_id,
          &table_properties_, write_hint, full_history_ts_low, blob_callback_,
          base_, &memtable_payload_bytes, &memtable_garbage_bytes, &flush_stats,
-          fast_sst_open_);
+          blob_file_garbages_for_filtering, fast_sst_open_);
      TEST_SYNC_POINT_CALLBACK("FlushJob::WriteLevel0Table:s", &s);
      // TODO: Cleanup io_status in BuildTable and table builders
      assert(!s.ok() || io_s.ok());
@@ -1104,21 +1111,27 @@ Status FlushJob::WriteLevel0Table() {
  }
  base_->Unref();

-  // Note that if file_size is zero, the file has been deleted and
-  // should not be added to the manifest.
+  // Note that if file_size is zero, the SST has been deleted and should not be
+  // added to the manifest. Blob metadata updates may still need to be
+  // committed for direct-write files or flush-time filtering.
  const bool has_output = meta_.fd.GetFileSize() > 0;

-  if (s.ok() && has_output) {
-    TEST_SYNC_POINT("DBImpl::FlushJob:SSTFileCreated");
-    // if we have more than 1 background thread, then we cannot
-    // insert files directly into higher levels because some other
-    // threads could be concurrently producing compacted files for
-    // that key range.
-    // Add file to L0
-    TEST_SYNC_POINT_CALLBACK("FileMetaData::FileMetaData", &meta_);
-    edit_->AddFile(0 /* level */, meta_);
-    edit_->SetBlobFileAdditions(std::move(blob_file_additions));
+  if (s.ok()) {
+    if (has_output) {
+      TEST_SYNC_POINT("DBImpl::FlushJob:SSTFileCreated");
+      // if we have more than 1 background thread, then we cannot
+      // insert files directly into higher levels because some other
+      // threads could be concurrently producing compacted files for
+      // that key range.
+      // Add file to L0
+      TEST_SYNC_POINT_CALLBACK("FileMetaData::FileMetaData", &meta_);
+      edit_->AddFile(0 /* level */, meta_);
+    }

+    edit_->SetBlobFileAdditions(std::move(blob_file_additions));
+    for (auto& garbage : blob_file_garbages) {
+      edit_->AddBlobFileGarbage(std::move(garbage));
+    }
    for (auto& addition : external_blob_file_additions_) {
      edit_->AddBlobFile(std::move(addition));
    }
@@ -15,6 +15,7 @@

 #include "db/blob/blob_file_partition_manager.h"
 #include "db/blob/blob_index.h"
+#include "db/blob/blob_log_format.h"
 #include "db/column_family.h"
 #include "db/db_test_util.h"
 #include "db/wide/wide_column_serialization.h"
@@ -342,6 +343,46 @@ class TTLOnlyLazyDropFilter : public CompactionFilter {
  std::atomic<int>* blob_columns_seen_;
 };

+class FlushOnlyTTLOnlyLazyDropFilterFactory : public CompactionFilterFactory {
+ public:
+  FlushOnlyTTLOnlyLazyDropFilterFactory(std::atomic<uint64_t>* ttl_cutoff,
+                                        std::atomic<int>* filter_call_count,
+                                        std::atomic<int>* ttl_columns_seen,
+                                        std::atomic<int>* ttl_bad_size_count,
+                                        std::atomic<int>* missing_ttl_count,
+                                        std::atomic<int>* blob_columns_seen)
+      : ttl_cutoff_(ttl_cutoff),
+        filter_call_count_(filter_call_count),
+        ttl_columns_seen_(ttl_columns_seen),
+        ttl_bad_size_count_(ttl_bad_size_count),
+        missing_ttl_count_(missing_ttl_count),
+        blob_columns_seen_(blob_columns_seen) {}
+
+  std::unique_ptr<CompactionFilter> CreateCompactionFilter(
+      const CompactionFilter::Context& /*context*/) override {
+    return std::make_unique<TTLOnlyLazyDropFilter>(
+        ttl_cutoff_, filter_call_count_, ttl_columns_seen_, ttl_bad_size_count_,
+        missing_ttl_count_, blob_columns_seen_);
+  }
+
+  bool ShouldFilterTableFileCreation(
+      TableFileCreationReason reason) const override {
+    return reason == TableFileCreationReason::kFlush;
+  }
+
+  const char* Name() const override {
+    return "FlushOnlyTTLOnlyLazyDropFilterFactory";
+  }
+
+ private:
+  std::atomic<uint64_t>* ttl_cutoff_;
+  std::atomic<int>* filter_call_count_;
+  std::atomic<int>* ttl_columns_seen_;
+  std::atomic<int>* ttl_bad_size_count_;
+  std::atomic<int>* missing_ttl_count_;
+  std::atomic<int>* blob_columns_seen_;
+};
+
 class NameBasedWideColumnPartitionStrategy : public BlobFilePartitionStrategy {
 public:
  using BlobFilePartitionStrategy::SelectPartition;
@@ -692,6 +733,214 @@ TEST_F(
  VerifySingleCfCoalescingIteratorMatchesDirectIterator(scenario);
 }

+TEST_F(DBWideBlobDirectWriteTest,
+       DirectWriteWideEntityLazyTTLFlushTracksExpiredBlobGarbage) {
+  std::atomic<int> filter_call_count{0};
+  std::atomic<int> ttl_columns_seen{0};
+  std::atomic<int> ttl_bad_size_count{0};
+  std::atomic<int> missing_ttl_count{0};
+  std::atomic<int> blob_columns_seen{0};
+  std::atomic<uint64_t> ttl_cutoff{1000};
+
+  auto filter_factory = std::make_shared<FlushOnlyTTLOnlyLazyDropFilterFactory>(
+      &ttl_cutoff, &filter_call_count, &ttl_columns_seen, &ttl_bad_size_count,
+      &missing_ttl_count, &blob_columns_seen);
+
+  Options options = GetDirectWriteOptions();
+  options.allow_concurrent_memtable_write = false;
+  options.blob_direct_write_partitions = 1;
+  options.min_blob_size = 64;
+  options.blob_file_size = 1 << 20;
+  options.disable_auto_compactions = true;
+  options.enable_blob_garbage_collection = false;
+  options.compaction_filter_factory = filter_factory;
+
+  Reopen(options);
+
+  const std::string expired_key = "expired_entity";
+  const std::string live_key = "live_entity";
+  const std::string expired_value = GenerateLargeValue(4096, 'E');
+  const std::string live_value = GenerateLargeValue(4096, 'L');
+  const std::string expired_ttl = EncodeFixedTTL(10);
+  const std::string live_ttl = EncodeFixedTTL(1000);
+  const auto expired_columns_data =
+      BuildTTLWideEntityData(expired_value, expired_ttl);
+  const WideColumns expired_columns = ToWideColumns(expired_columns_data);
+  const auto live_columns_data = BuildTTLWideEntityData(live_value, live_ttl);
+  const WideColumns live_columns = ToWideColumns(live_columns_data);
+
+  ASSERT_OK(db_->PutEntity(WriteOptions(), db_->DefaultColumnFamily(),
+                           expired_key, expired_columns));
+  ASSERT_OK(db_->PutEntity(WriteOptions(), db_->DefaultColumnFamily(), live_key,
+                           live_columns));
+
+  ASSERT_OK(Flush());
+
+  ASSERT_GE(filter_call_count.load(), 2);
+  ASSERT_EQ(ttl_columns_seen.load(), filter_call_count.load());
+  ASSERT_EQ(ttl_bad_size_count.load(), 0);
+  ASSERT_EQ(missing_ttl_count.load(), 0);
+  ASSERT_EQ(blob_columns_seen.load(), filter_call_count.load());
+
+  PinnableWideColumns result;
+  ASSERT_TRUE(db_->GetEntity(ReadOptions(), db_->DefaultColumnFamily(),
+                             expired_key, &result)
+                  .IsNotFound());
+  ASSERT_OK(db_->GetEntity(ReadOptions(), db_->DefaultColumnFamily(), live_key,
+                           &result));
+  ASSERT_EQ(result.columns(), live_columns);
+
+  const uint64_t expired_record_bytes =
+      expired_value.size() +
+      BlobLogRecord::CalculateAdjustmentForRecordHeader(expired_key.size());
+  const uint64_t live_record_bytes =
+      live_value.size() +
+      BlobLogRecord::CalculateAdjustmentForRecordHeader(live_key.size());
+
+  ColumnFamilyMetaData cf_meta;
+  db_->GetColumnFamilyMetaData(&cf_meta);
+  ASSERT_EQ(cf_meta.blob_file_count, 1U);
+  ASSERT_EQ(cf_meta.blob_files.size(), 1U);
+
+  const BlobMetaData& blob_meta = cf_meta.blob_files[0];
+  ASSERT_EQ(blob_meta.total_blob_count, 2U);
+  ASSERT_EQ(blob_meta.total_blob_bytes,
+            expired_record_bytes + live_record_bytes);
+  ASSERT_EQ(blob_meta.garbage_blob_count, 1U);
+  ASSERT_EQ(blob_meta.garbage_blob_bytes, expired_record_bytes);
+}
+
+TEST_F(DBWideBlobDirectWriteTest,
+       DirectWriteWideEntityFlushOverwriteElisionTracksBlobGarbage) {
+  Options options = GetDirectWriteOptions();
+  options.allow_concurrent_memtable_write = false;
+  options.blob_direct_write_partitions = 1;
+  options.min_blob_size = 64;
+  options.blob_file_size = 1 << 20;
+  options.disable_auto_compactions = true;
+  options.enable_blob_garbage_collection = false;
+
+  Reopen(options);
+
+  const std::string key = "overwritten_entity";
+  const std::string old_value = GenerateLargeValue(4096, 'O');
+  const std::string new_value = GenerateLargeValue(4096, 'N');
+  const auto old_columns_data =
+      WideColumnStringPairs{{"", old_value}, {"meta", "old_inline_meta"}};
+  const WideColumns old_columns = ToWideColumns(old_columns_data);
+  const auto new_columns_data =
+      WideColumnStringPairs{{"", new_value}, {"meta", "new_inline_meta"}};
+  const WideColumns new_columns = ToWideColumns(new_columns_data);
+
+  ASSERT_OK(db_->PutEntity(WriteOptions(), db_->DefaultColumnFamily(), key,
+                           old_columns));
+  ASSERT_OK(db_->PutEntity(WriteOptions(), db_->DefaultColumnFamily(), key,
+                           new_columns));
+
+  ASSERT_OK(Flush());
+
+  PinnableWideColumns result;
+  ASSERT_OK(
+      db_->GetEntity(ReadOptions(), db_->DefaultColumnFamily(), key, &result));
+  ASSERT_EQ(result.columns(), new_columns);
+
+  const uint64_t old_record_bytes =
+      old_value.size() +
+      BlobLogRecord::CalculateAdjustmentForRecordHeader(key.size());
+  const uint64_t new_record_bytes =
+      new_value.size() +
+      BlobLogRecord::CalculateAdjustmentForRecordHeader(key.size());
+
+  ColumnFamilyMetaData cf_meta;
+  db_->GetColumnFamilyMetaData(&cf_meta);
+  ASSERT_EQ(cf_meta.blob_file_count, 1U);
+  ASSERT_EQ(cf_meta.blob_files.size(), 1U);
+
+  const BlobMetaData& blob_meta = cf_meta.blob_files[0];
+  ASSERT_EQ(blob_meta.total_blob_count, 2U);
+  ASSERT_EQ(blob_meta.total_blob_bytes, old_record_bytes + new_record_bytes);
+  ASSERT_EQ(blob_meta.garbage_blob_count, 1U);
+  ASSERT_EQ(blob_meta.garbage_blob_bytes, old_record_bytes);
+}
+
+TEST_F(DBWideBlobDirectWriteTest,
+       DirectWriteWideEntityLazyTTLFlushAllExpiredDoesNotLeakBlobGeneration) {
+  std::atomic<int> filter_call_count{0};
+  std::atomic<int> ttl_columns_seen{0};
+  std::atomic<int> ttl_bad_size_count{0};
+  std::atomic<int> missing_ttl_count{0};
+  std::atomic<int> blob_columns_seen{0};
+  std::atomic<uint64_t> ttl_cutoff{1000};
+
+  auto filter_factory = std::make_shared<FlushOnlyTTLOnlyLazyDropFilterFactory>(
+      &ttl_cutoff, &filter_call_count, &ttl_columns_seen, &ttl_bad_size_count,
+      &missing_ttl_count, &blob_columns_seen);
+
+  Options options = GetDirectWriteOptions();
+  options.allow_concurrent_memtable_write = false;
+  options.blob_direct_write_partitions = 1;
+  options.min_blob_size = 64;
+  options.blob_file_size = 1 << 20;
+  options.disable_auto_compactions = true;
+  options.enable_blob_garbage_collection = false;
+  options.compaction_filter_factory = filter_factory;
+
+  Reopen(options);
+
+  const std::string expired_key = "expired_only_entity";
+  const std::string expired_value = GenerateLargeValue(4096, 'X');
+  const std::string expired_ttl = EncodeFixedTTL(10);
+  const auto expired_columns_data =
+      BuildTTLWideEntityData(expired_value, expired_ttl);
+  const WideColumns expired_columns = ToWideColumns(expired_columns_data);
+
+  ASSERT_OK(db_->PutEntity(WriteOptions(), db_->DefaultColumnFamily(),
+                           expired_key, expired_columns));
+
+  ASSERT_OK(Flush());
+
+  ASSERT_GE(filter_call_count.load(), 1);
+  ASSERT_EQ(ttl_columns_seen.load(), filter_call_count.load());
+  ASSERT_EQ(ttl_bad_size_count.load(), 0);
+  ASSERT_EQ(missing_ttl_count.load(), 0);
+  ASSERT_EQ(blob_columns_seen.load(), filter_call_count.load());
+
+  PinnableWideColumns result;
+  ASSERT_TRUE(db_->GetEntity(ReadOptions(), db_->DefaultColumnFamily(),
+                             expired_key, &result)
+                  .IsNotFound());
+
+  ColumnFamilyMetaData cf_meta;
+  db_->GetColumnFamilyMetaData(&cf_meta);
+  ASSERT_EQ(cf_meta.blob_file_count, 0U);
+  ASSERT_TRUE(cf_meta.blob_files.empty());
+
+  const std::string live_key = "live_after_empty_flush";
+  const std::string live_value = GenerateLargeValue(4096, 'Y');
+  const std::string live_ttl = EncodeFixedTTL(5000);
+  const auto live_columns_data = BuildTTLWideEntityData(live_value, live_ttl);
+  const WideColumns live_columns = ToWideColumns(live_columns_data);
+
+  ASSERT_OK(db_->PutEntity(WriteOptions(), db_->DefaultColumnFamily(), live_key,
+                           live_columns));
+  ASSERT_OK(Flush());
+  ASSERT_OK(db_->GetEntity(ReadOptions(), db_->DefaultColumnFamily(), live_key,
+                           &result));
+  ASSERT_EQ(result.columns(), live_columns);
+
+  const uint64_t live_record_bytes =
+      live_value.size() +
+      BlobLogRecord::CalculateAdjustmentForRecordHeader(live_key.size());
+
+  db_->GetColumnFamilyMetaData(&cf_meta);
+  ASSERT_EQ(cf_meta.blob_file_count, 1U);
+  ASSERT_EQ(cf_meta.blob_files.size(), 1U);
+  ASSERT_EQ(cf_meta.blob_files[0].total_blob_count, 1U);
+  ASSERT_EQ(cf_meta.blob_files[0].total_blob_bytes, live_record_bytes);
+  ASSERT_EQ(cf_meta.blob_files[0].garbage_blob_count, 0U);
+  ASSERT_EQ(cf_meta.blob_files[0].garbage_blob_bytes, 0U);
+}
+
 TEST_F(DBWideBlobDirectWriteTest,
       MultiCfCoalescingIteratorResolvesBlobBackedColumnsAfterSeekRefresh) {
  constexpr int kNumKeys = 32;
@@ -0,0 +1 @@
+Fix blob garbage accounting for blob direct-write flushes so flush-time filtering and overwrite elision correctly register obsolete blob bytes in blob metadata.
				`@@ -0,0 +1 @@`
				`Fix blob garbage accounting for blob direct-write flushes so flush-time filtering and overwrite elision correctly register obsolete blob bytes in blob metadata.`