Add an optional min file size requirement for deletion triggered compaction (#13707)

Summary:
add the `min_file_size` parameter to CompactOnDeletionCollector. A file must be at least this size for it to qualify for DTC. This is useful when a user wants to specific a min file size requirement that is larger than the size constraint imposed by the sliding window's `deletion_trigger` requirement.

Added some comment explaining that the file_size provided to table property collector only includes data blocks and may not be up-to-date. This PR also updates DTC to consider SingleDelete and DeletionWithTimestamp as tombstones.

Pull Request resolved: https://github.com/facebook/rocksdb/pull/13707

Test Plan:
- new unit test for when min_file_size is specified.
- existing unit test for when min_file_size is not specified.

Reviewed By: hx235, pdillinger

Differential Revision: D76837231

Pulled By: cbi42

fbshipit-source-id: 0782144e75aef9961bf03da2a2c4b3c613ce5db3

include/rocksdb/table_properties.h

@@ -109,6 +109,10 @@ class TablePropertiesCollector {
   // table.
   // @params key    the user key that is inserted into the table.
   // @params value  the value that is inserted into the table.
+  // @params file_size the current file size. For BlockBasedTable, this
+  //         includes all the data blocks written so far, upto but not including
+  //         the current block being built. With parallel compression, data
+  //         blocks are written async so it depends on the compression progress.
   virtual Status AddUserKey(const Slice& key, const Slice& value,
                             EntryType /*type*/, SequenceNumber /*seq*/,
                             uint64_t /*file_size*/) {
@@ -143,7 +147,7 @@ class TablePropertiesCollector {
   // The name of the properties collector can be used for debugging purpose.
   virtual const char* Name() const = 0;
 
-  // EXPERIMENTAL Return whether the output file should be further compacted
+  // Return whether the output file should be further compacted
   virtual bool NeedCompact() const { return false; }
 
   // For internal use only.

include/rocksdb/utilities/table_properties_collectors.h

@@ -23,15 +23,20 @@ class CompactOnDeletionCollectorFactory
   // A factory of a table property collector that marks a SST
   // file as need-compaction when it observe at least "D" deletion
   // entries in any "N" consecutive entries, or the ratio of tombstone
-  // entries >= deletion_ratio.
+  // entries >= deletion_ratio for the entire file.
   //
   // @param sliding_window_size "N"
   // @param deletion_trigger "D"
   // @param deletion_ratio, if <= 0 or > 1, disable triggering compaction
   //     based on deletion ratio.
+  // @param min_file_size, a file needs to be at least this size to be marked
+  //     for compaction. See comments above
+  //     TablePropertiesCollector::AddUserKey() for limitations/inaccuracies on
+  //     the file size.
   CompactOnDeletionCollectorFactory(size_t sliding_window_size,
                                     size_t deletion_trigger,
-                                    double deletion_ratio);
+                                    double deletion_ratio,
+                                    uint64_t min_file_size = 0);
 
   ~CompactOnDeletionCollectorFactory() override {}
 
@@ -59,6 +64,12 @@ class CompactOnDeletionCollectorFactory
   }
 
   double GetDeletionRatio() const { return deletion_ratio_.load(); }
+
+  uint64_t GetMinFileSize() const { return min_file_size_.load(); }
+  void SetMinFileSize(uint64_t min_file_size) {
+    min_file_size_.store(min_file_size);
+  }
+
   static const char* kClassName() { return "CompactOnDeletionCollector"; }
   const char* Name() const override { return kClassName(); }
 
@@ -68,6 +79,7 @@ class CompactOnDeletionCollectorFactory
   std::atomic<size_t> sliding_window_size_;
   std::atomic<size_t> deletion_trigger_;
   std::atomic<double> deletion_ratio_;
+  std::atomic<uint64_t> min_file_size_;
 };
 
 // Creates a factory of a table property collector that marks a SST
@@ -85,7 +97,8 @@ class CompactOnDeletionCollectorFactory
 std::shared_ptr<CompactOnDeletionCollectorFactory>
 NewCompactOnDeletionCollectorFactory(size_t sliding_window_size,
                                      size_t deletion_trigger,
-                                     double deletion_ratio = 0);
+                                     double deletion_ratio = 0,
+                                     uint64_t min_file_size = 0);
 
 // A factory of a table property collector that marks a SST file as
 // need-compaction when for the tiering use case, it observes, among all the

tools/db_crashtest.py

@@ -723,7 +723,7 @@ def finalize_and_sanitize(src_params):
         else:
             dest_params["mock_direct_io"] = True
 
-    if dest_params["memtablerep"] == "vector":
+    if dest_params.get("memtablerep") == "vector":
         dest_params["inplace_update_support"] = 0
         dest_params["paranoid_memory_checks"] = 0
 

unreleased_history/new_features/concurrent_vector_insert.md

@@ -0,0 +1 @@
+* Vector based memtable now supports concurrent writers (DBOptions::allow_concurrent_memtable_write) #13675.

unreleased_history/public_api_changes/min-file-size-dtc.md

@@ -0,0 +1 @@
+* Add an optional min file size requirement for deletion triggered compaction. It can be specified when creating `CompactOnDeletionCollectorFactory`.

utilities/table_properties_collectors/compact_on_deletion_collector.cc

@@ -17,16 +17,19 @@
 namespace ROCKSDB_NAMESPACE {
 
 CompactOnDeletionCollector::CompactOnDeletionCollector(
-    size_t sliding_window_size, size_t deletion_trigger, double deletion_ratio)
+    size_t sliding_window_size, size_t deletion_trigger, double deletion_ratio,
+    uint64_t min_file_size)
     : bucket_size_((sliding_window_size + kNumBuckets - 1) / kNumBuckets),
       current_bucket_(0),
       num_keys_in_current_bucket_(0),
       num_deletions_in_observation_window_(0),
       deletion_trigger_(deletion_trigger),
       deletion_ratio_(deletion_ratio),
+      min_file_size_(min_file_size),
+      cur_file_size_(0),
+      max_deletion_in_window_(0),
       deletion_ratio_enabled_(deletion_ratio > 0 && deletion_ratio <= 1),
-      need_compaction_(false),
-      finished_(false) {
+      need_compaction_(false) {
   memset(num_deletions_in_buckets_, 0, sizeof(size_t) * kNumBuckets);
 }
 
@@ -39,7 +42,7 @@ Status CompactOnDeletionCollector::AddUserKey(const Slice& /*key*/,
                                               const Slice& /*value*/,
                                               EntryType type,
                                               SequenceNumber /*seq*/,
-                                              uint64_t /*file_size*/) {
+                                              uint64_t file_size) {
   assert(!finished_);
   if (!bucket_size_ && !deletion_ratio_enabled_) {
     // This collector is effectively disabled
@@ -51,11 +54,14 @@ Status CompactOnDeletionCollector::AddUserKey(const Slice& /*key*/,
     return Status::OK();
   }
 
+  const bool is_delete = (type == kEntryDelete || type == kEntrySingleDelete ||
+                          type == kEntryDeleteWithTimestamp);
   if (deletion_ratio_enabled_) {
     total_entries_++;
-    if (type == kEntryDelete) {
+    if (is_delete) {
       deletion_entries_++;
     }
+    cur_file_size_ = file_size;
   }
 
   if (bucket_size_) {
@@ -76,13 +82,20 @@ Status CompactOnDeletionCollector::AddUserKey(const Slice& /*key*/,
     }
 
     num_keys_in_current_bucket_++;
-    if (type == kEntryDelete) {
+    if (is_delete) {
       num_deletions_in_observation_window_++;
       num_deletions_in_buckets_[current_bucket_]++;
-      if (num_deletions_in_observation_window_ >= deletion_trigger_) {
-        need_compaction_ = true;
+      if (num_deletions_in_observation_window_ >= max_deletion_in_window_) {
+        max_deletion_in_window_ = num_deletions_in_observation_window_;
       }
     }
+
+    // The file may qualify for compaction based on file size constraints,
+    // even if max_deletion_in_window_ is not updated.
+    if (max_deletion_in_window_ >= deletion_trigger_ &&
+        file_size >= min_file_size_) {
+      need_compaction_ = true;
+    }
   }
 
   return Status::OK();
@@ -90,7 +103,8 @@ Status CompactOnDeletionCollector::AddUserKey(const Slice& /*key*/,
 
 Status CompactOnDeletionCollector::Finish(
     UserCollectedProperties* /*properties*/) {
-  if (!need_compaction_ && deletion_ratio_enabled_ && total_entries_ > 0) {
+  if (!need_compaction_ && deletion_ratio_enabled_ && total_entries_ > 0 &&
+      cur_file_size_ >= min_file_size_) {
     double ratio = static_cast<double>(deletion_entries_) / total_entries_;
     need_compaction_ = ratio >= deletion_ratio_;
   }
@@ -153,23 +167,43 @@ static std::unordered_map<std::string, OptionTypeInfo>
             return Status::OK();
           },
           nullptr}},
+        {"min_file_size",
+         {0, OptionType::kUnknown, OptionVerificationType::kNormal,
+          OptionTypeFlags::kCompareNever | OptionTypeFlags::kMutable,
+          [](const ConfigOptions&, const std::string&, const std::string& value,
+             void* addr) {
+            auto* factory =
+                static_cast<CompactOnDeletionCollectorFactory*>(addr);
+            factory->SetMinFileSize(ParseUint64(value));
+            return Status::OK();
+          },
+          [](const ConfigOptions&, const std::string&, const void* addr,
+             std::string* value) {
+            const auto* factory =
+                static_cast<const CompactOnDeletionCollectorFactory*>(addr);
+            *value = std::to_string(factory->GetMinFileSize());
+            return Status::OK();
+          },
+          nullptr}},
 
 };
 
 CompactOnDeletionCollectorFactory::CompactOnDeletionCollectorFactory(
-    size_t sliding_window_size, size_t deletion_trigger, double deletion_ratio)
+    size_t sliding_window_size, size_t deletion_trigger, double deletion_ratio,
+    uint64_t min_file_size)
     : sliding_window_size_(sliding_window_size),
       deletion_trigger_(deletion_trigger),
-      deletion_ratio_(deletion_ratio) {
+      deletion_ratio_(deletion_ratio),
+      min_file_size_(min_file_size) {
   RegisterOptions("", this, &on_deletion_collector_type_info);
 }
 
 TablePropertiesCollector*
 CompactOnDeletionCollectorFactory::CreateTablePropertiesCollector(
     TablePropertiesCollectorFactory::Context /*context*/) {
-  return new CompactOnDeletionCollector(sliding_window_size_.load(),
-                                        deletion_trigger_.load(),
-                                        deletion_ratio_.load());
+  return new CompactOnDeletionCollector(
+      sliding_window_size_.load(), deletion_trigger_.load(),
+      deletion_ratio_.load(), min_file_size_.load());
 }
 
 std::string CompactOnDeletionCollectorFactory::ToString() const {
@@ -183,10 +217,12 @@ std::string CompactOnDeletionCollectorFactory::ToString() const {
 std::shared_ptr<CompactOnDeletionCollectorFactory>
 NewCompactOnDeletionCollectorFactory(size_t sliding_window_size,
                                      size_t deletion_trigger,
-                                     double deletion_ratio) {
+                                     double deletion_ratio,
+                                     uint64_t min_file_size) {
   return std::shared_ptr<CompactOnDeletionCollectorFactory>(
       new CompactOnDeletionCollectorFactory(sliding_window_size,
-                                            deletion_trigger, deletion_ratio));
+                                            deletion_trigger, deletion_ratio,
+                                            min_file_size));
 }
 
 namespace {

utilities/table_properties_collectors/compact_on_deletion_collector.h

@@ -11,7 +11,8 @@ namespace ROCKSDB_NAMESPACE {
 class CompactOnDeletionCollector : public TablePropertiesCollector {
  public:
   CompactOnDeletionCollector(size_t sliding_window_size,
-                             size_t deletion_trigger, double deletion_raatio);
+                             size_t deletion_trigger, double deletion_ratio,
+                             uint64_t min_file_size);
 
   // AddUserKey() will be called when a new key/value pair is inserted into the
   // table.
@@ -36,7 +37,7 @@ class CompactOnDeletionCollector : public TablePropertiesCollector {
   // The name of the properties collector can be used for debugging purpose.
   const char* Name() const override { return "CompactOnDeletionCollector"; }
 
-  // EXPERIMENTAL Return whether the output file should be further compacted
+  // Return whether the output file should be further compacted
   bool NeedCompact() const override { return need_compaction_; }
 
   static const int kNumBuckets = 128;
@@ -48,18 +49,21 @@ class CompactOnDeletionCollector : public TablePropertiesCollector {
   // "bucket_size_" keys.
   size_t num_deletions_in_buckets_[kNumBuckets];
   // the number of keys in a bucket
-  size_t bucket_size_;
+  const size_t bucket_size_;
 
   size_t current_bucket_;
   size_t num_keys_in_current_bucket_;
   size_t num_deletions_in_observation_window_;
-  size_t deletion_trigger_;
+  const size_t deletion_trigger_;
   const double deletion_ratio_;
-  const bool deletion_ratio_enabled_;
   size_t total_entries_ = 0;
   size_t deletion_entries_ = 0;
+  const size_t min_file_size_;
+  size_t cur_file_size_;
+  size_t max_deletion_in_window_;
+  const bool deletion_ratio_enabled_;
   // true if the current SST file needs to be compacted.
   bool need_compaction_;
-  bool finished_;
+  bool finished_ = false;
 };
 }  // namespace ROCKSDB_NAMESPACE

utilities/table_properties_collectors/compact_on_deletion_collector_test.cc

@@ -232,6 +232,61 @@ TEST(CompactOnDeletionCollector, SlidingWindow) {
   }
 }
 
+TEST(CompactOnDeletionCollector, MinFileSize) {
+  TablePropertiesCollectorFactory::Context context;
+  context.column_family_id =
+      TablePropertiesCollectorFactory::Context::kUnknownColumnFamily;
+  context.last_level_inclusive_max_seqno_threshold = kMaxSequenceNumber;
+
+  const size_t kWindowSize = 1000;
+  const size_t kDeletionTrigger = 800;
+  const double kDeletionRatio = 0.9;
+  const uint64_t kMinFileSize = 1 << 20;
+
+  for (uint64_t file_size : {(uint64_t)0, kMinFileSize - 1, kMinFileSize}) {
+    {
+      auto factory = NewCompactOnDeletionCollectorFactory(
+          kWindowSize, kDeletionTrigger, 0, kMinFileSize);
+      std::unique_ptr<TablePropertiesCollector> collector(
+          factory->CreateTablePropertiesCollector(context));
+
+      // Add enough deletions to meet the sliding window triggers
+      for (size_t i = 0; i < kWindowSize; i++) {
+        if (i < kDeletionTrigger) {
+          ASSERT_OK(collector->AddUserKey("key", "value", kEntryDelete, 0,
+                                          file_size));
+        } else {
+          ASSERT_OK(
+              collector->AddUserKey("key", "value", kEntryPut, 0, file_size));
+        }
+      }
+      ASSERT_OK(collector->Finish(nullptr));
+      ASSERT_EQ(collector->NeedCompact(), file_size >= kMinFileSize);
+    }
+
+    {
+      auto factory = NewCompactOnDeletionCollectorFactory(
+          kWindowSize, kDeletionTrigger, kDeletionRatio, kMinFileSize);
+
+      std::unique_ptr<TablePropertiesCollector> collector(
+          factory->CreateTablePropertiesCollector(context));
+
+      const size_t kTotalEntries = 100;
+      // Add all deletions to maximize tombstone ratio
+      for (size_t i = 0; i < kTotalEntries - 1; i++) {
+        ASSERT_OK(
+            collector->AddUserKey("key", "value", kEntrySingleDelete, 0, 0));
+      }
+      // Give update file size
+      ASSERT_OK(collector->AddUserKey("key", "value", kEntrySingleDelete, 0,
+                                      file_size));
+
+      ASSERT_OK(collector->Finish(nullptr));
+      ASSERT_EQ(collector->NeedCompact(), file_size >= kMinFileSize);
+    }
+  }
+}
+
 }  // namespace ROCKSDB_NAMESPACE
 
 int main(int argc, char** argv) {