Improve performance the mix of range-delete and read workloads.

The idea is to force flush after few range-delete operations.
Experimentation shows 32 or 64 giving good performance.

memtable_max_range_deletions is available as a ColumnFamilyOption.
The default is 0, which disables. it.

db/memtable.cc

@@ -95,6 +95,7 @@ MemTable::MemTable(const InternalKeyComparator& cmp,
       data_size_(0),
       num_entries_(0),
       num_deletes_(0),
+      num_range_deletes_(0),
       write_buffer_size_(mutable_cf_options.write_buffer_size),
       flush_in_progress_(false),
       flush_completed_(false),
@@ -114,7 +115,9 @@ MemTable::MemTable(const InternalKeyComparator& cmp,
           ioptions.memtable_insert_with_hint_prefix_extractor.get()),
       oldest_key_time_(std::numeric_limits<uint64_t>::max()),
       atomic_flush_seqno_(kMaxSequenceNumber),
-      approximate_memory_usage_(0) {
+      approximate_memory_usage_(0),
+      memtable_max_range_deletions_(
+          mutable_cf_options.memtable_max_range_deletions) {
   UpdateFlushState();
   // something went wrong if we need to flush before inserting anything
   assert(!ShouldScheduleFlush());
@@ -170,6 +173,12 @@ size_t MemTable::ApproximateMemoryUsage() {
 }
 
 bool MemTable::ShouldFlushNow() {
+  // This is set if memtable_max_range_deletions is > 0,
+  // and that many range deletions are done
+  if (memtable_max_range_deletions_reached_) {
+    return true;
+  }
+
   size_t write_buffer_size = write_buffer_size_.load(std::memory_order_relaxed);
   // In a lot of times, we cannot allocate arena blocks that exactly matches the
   // buffer size. Thus we have to decide if we should over-allocate or
@@ -787,6 +796,15 @@ Status MemTable::Add(SequenceNumber s, ValueType type,
         type == kTypeDeletionWithTimestamp) {
       num_deletes_.store(num_deletes_.load(std::memory_order_relaxed) + 1,
                          std::memory_order_relaxed);
+    } else if (type == kTypeRangeDeletion) {
+      uint64_t val = num_range_deletes_.load(std::memory_order_relaxed) + 1;
+      num_range_deletes_.store(val, std::memory_order_relaxed);
+      // Arrange for a flush if too many delete ranges have been issued.
+      if (memtable_max_range_deletions_ > 0 &&
+          memtable_max_range_deletions_reached_ == false &&
+          val > (uint64_t)memtable_max_range_deletions_) {
+        memtable_max_range_deletions_reached_ = true;
+      }
     }
 
     if (bloom_filter_ && prefix_extractor_ &&
@@ -1294,6 +1312,7 @@ bool MemTable::Get(const LookupKey& key, std::string* value,
     // Avoiding recording stats for speed.
     return false;
   }
+
   PERF_TIMER_GUARD(get_from_memtable_time);
 
   std::unique_ptr<FragmentedRangeTombstoneIterator> range_del_iter(

db/memtable.h

@@ -555,6 +555,7 @@ class MemTable {
   std::atomic<uint64_t> data_size_;
   std::atomic<uint64_t> num_entries_;
   std::atomic<uint64_t> num_deletes_;
+  std::atomic<uint64_t> num_range_deletes_;
 
   // Dynamically changeable memtable option
   std::atomic<size_t> write_buffer_size_;
@@ -616,6 +617,13 @@ class MemTable {
   // Gets refreshed inside `ApproximateMemoryUsage()` or `ShouldFlushNow`
   std::atomic<uint64_t> approximate_memory_usage_;
 
+  // max range deletions in a memtable,  before automatic flushing, 0 for
+  // unlimited.
+  uint32_t memtable_max_range_deletions_ = 0;
+  // Range-delete will turn this flag on, if memtable_max_range_deletions_ is
+  // reached. ShouldFlushNow() will check this.
+  bool memtable_max_range_deletions_reached_ = false;
+
 #ifndef ROCKSDB_LITE
   // Flush job info of the current memtable.
   std::unique_ptr<FlushJobInfo> flush_job_info_;

include/rocksdb/options.h

@@ -334,6 +334,11 @@ struct ColumnFamilyOptions : public AdvancedColumnFamilyOptions {
   // Default: nullptr
   std::shared_ptr<SstPartitionerFactory> sst_partitioner_factory = nullptr;
 
+  // Automatic flush after range deletions count in memtable hits this limit.
+  // helps with workloads having lot of range deletes.
+  // 0 to disable it completely
+  uint32_t memtable_max_range_deletions = 0;
+
   // Create ColumnFamilyOptions with default values for all fields
   ColumnFamilyOptions();
   // Create ColumnFamilyOptions from Options

java/rocksjni/options.cc

@@ -3905,6 +3905,28 @@ jbyte Java_org_rocksdb_Options_prepopulateBlobCache(JNIEnv*, jobject,
       opts->prepopulate_blob_cache);
 }
 
+/*
+ * Class:     org_rocksdb_Options
+ * Method:    setMemtableMaxRangeDeletions
+ * Signature: (JI)V
+ */
+void Java_org_rocksdb_Options_setMemtableMaxRangeDeletions(
+    JNIEnv*, jobject, jlong jhandle, jint jmemtable_max_range_deletions) {
+  auto* opts = reinterpret_cast<ROCKSDB_NAMESPACE::Options*>(jhandle);
+  opts->memtable_max_range_deletions = jmemtable_max_range_deletions;
+}
+
+/*
+ * Class:     org_rocksdb_Options
+ * Method:    memtableMaxRangeDeletions
+ * Signature: (J)I
+ */
+jint Java_org_rocksdb_Options_memtableMaxRangeDeletions(JNIEnv*, jobject,
+                                                        jlong jhandle) {
+  auto* opts = reinterpret_cast<ROCKSDB_NAMESPACE::Options*>(jhandle);
+  return static_cast<jint>(opts->memtable_max_range_deletions);
+}
+
 //////////////////////////////////////////////////////////////////////////////
 // ROCKSDB_NAMESPACE::ColumnFamilyOptions
 
@@ -5767,6 +5789,30 @@ jbyte Java_org_rocksdb_ColumnFamilyOptions_prepopulateBlobCache(JNIEnv*,
       opts->prepopulate_blob_cache);
 }
 
+/*
+ * Class:     org_rocksdb_ColumnFamilyOptions
+ * Method:    setMemtableMaxRangeDeletions
+ * Signature: (JI)V
+ */
+void Java_org_rocksdb_ColumnFamilyOptions_setMemtableMaxRangeDeletions(
+    JNIEnv*, jobject, jlong jhandle, jint jmemtable_max_range_deletions) {
+  auto* opts =
+      reinterpret_cast<ROCKSDB_NAMESPACE::ColumnFamilyOptions*>(jhandle);
+  opts->memtable_max_range_deletions = jmemtable_max_range_deletions;
+}
+
+/*
+ * Class:     org_rocksdb_ColumnFamilyOptions
+ * Method:    memtableMaxRangeDeletions
+ * Signature: (J)I
+ */
+jint Java_org_rocksdb_ColumnFamilyOptions_memtableMaxRangeDeletions(
+    JNIEnv*, jobject, jlong jhandle) {
+  auto* opts =
+      reinterpret_cast<ROCKSDB_NAMESPACE::ColumnFamilyOptions*>(jhandle);
+  return static_cast<jint>(opts->memtable_max_range_deletions);
+}
+
 /////////////////////////////////////////////////////////////////////
 // ROCKSDB_NAMESPACE::DBOptions
 

java/src/main/java/org/rocksdb/ColumnFamilyOptions.java

@@ -956,6 +956,17 @@ public SstPartitionerFactory sstPartitionerFactory() {
     return sstPartitionerFactory_;
   }
 
+  @Override
+  public ColumnFamilyOptions setMemtableMaxRangeDeletions(final int count) {
+    setMemtableMaxRangeDeletions(nativeHandle_, count);
+    return this;
+  }
+
+  @Override
+  public int memtableMaxRangeDeletions() {
+    return memtableMaxRangeDeletions(nativeHandle_);
+  }
+
   //
   // BEGIN options for blobs (integrated BlobDB)
   //
@@ -1495,6 +1506,8 @@ private native void setForceConsistencyChecks(final long handle,
   private native void setSstPartitionerFactory(long nativeHandle_, long newFactoryHandle);
   private static native void setCompactionThreadLimiter(
       final long nativeHandle_, final long compactionThreadLimiterHandle);
+  private native void setMemtableMaxRangeDeletions(final long handle, final int count);
+  private native int memtableMaxRangeDeletions(final long handle);
 
   private native void setEnableBlobFiles(final long nativeHandle_, final boolean enableBlobFiles);
   private native boolean enableBlobFiles(final long nativeHandle_);

java/src/main/java/org/rocksdb/ColumnFamilyOptionsInterface.java

@@ -506,6 +506,23 @@ T setCompressionOptions(
   @Experimental("Caution: this option is experimental")
   SstPartitionerFactory sstPartitionerFactory();
 
+  /**
+   * Sets the maximum range delete calls, after which memtable is flushed.
+   * This applies to the mutable memtable.
+   *
+   * @param a positive integer, 0 (default) to disable the feature.
+   * @return the reference of the current options.
+   */
+  T setMemtableMaxRangeDeletions(final int count);
+
+  /**
+   * Gets the current setting of maximum range deletes allowed
+   * 0(default) indicates that feature is disabled.
+   *
+   * @return current value of memtable_max_range_deletions
+   */
+  int memtableMaxRangeDeletions();
+
   /**
    * Compaction concurrent thread limiter for the column family.
    * If non-nullptr, use given concurrent thread limiter to control

java/src/main/java/org/rocksdb/Options.java

@@ -1987,6 +1987,17 @@ public SstPartitionerFactory sstPartitionerFactory() {
     return sstPartitionerFactory_;
   }
 
+  @Override
+  public Options setMemtableMaxRangeDeletions(final int count) {
+    setMemtableMaxRangeDeletions(nativeHandle_, count);
+    return this;
+  }
+
+  @Override
+  public int memtableMaxRangeDeletions() {
+    return memtableMaxRangeDeletions(nativeHandle_);
+  }
+
   @Override
   public Options setCompactionThreadLimiter(final ConcurrentTaskLimiter compactionThreadLimiter) {
     setCompactionThreadLimiter(nativeHandle_, compactionThreadLimiter.nativeHandle_);
@@ -2506,6 +2517,8 @@ private native void setAtomicFlush(final long handle,
       final boolean atomicFlush);
   private native boolean atomicFlush(final long handle);
   private native void setSstPartitionerFactory(long nativeHandle_, long newFactoryHandle);
+  private native void setMemtableMaxRangeDeletions(final long handle, final int count);
+  private native int memtableMaxRangeDeletions(final long handle);
   private static native void setCompactionThreadLimiter(
       final long nativeHandle_, final long newLimiterHandle);
   private static native void setAvoidUnnecessaryBlockingIO(

java/src/test/java/org/rocksdb/ColumnFamilyOptionsTest.java

@@ -711,4 +711,14 @@ public void cfPaths() throws IOException {
       assertThat(options.cfPaths()).isEqualTo(paths);
     }
   }
+
+  @Test
+  public void memtableMaxRangeDeletions() {
+    try (final ColumnFamilyOptions options = new ColumnFamilyOptions()) {
+      assertThat(options.memtableMaxRangeDeletions()).isEqualTo(0);
+      final int val = 32;
+      assertThat(options.setMemtableMaxRangeDeletions(val)).isEqualTo(options);
+      assertThat(options.memtableMaxRangeDeletions()).isEqualTo(val);
+    }
+  }
 }

java/src/test/java/org/rocksdb/OptionsTest.java

@@ -1456,6 +1456,16 @@ public void skipCheckingSstFileSizesOnDbOpen() {
     }
   }
 
+  @Test
+  public void memtableMaxRangeDeletions() {
+    try (final Options options = new Options()) {
+      assertThat(options.memtableMaxRangeDeletions()).isEqualTo(0);
+      final int val = 32;
+      assertThat(options.setMemtableMaxRangeDeletions(val)).isEqualTo(options);
+      assertThat(options.memtableMaxRangeDeletions()).isEqualTo(val);
+    }
+  }
+
   @Test
   public void eventListeners() {
     final AtomicBoolean wasCalled1 = new AtomicBoolean();

options/cf_options.cc

@@ -525,6 +525,11 @@ static std::unordered_map<std::string, OptionTypeInfo>
                }
              })},
         // End special case properties
+        {"memtable_max_range_deletions",
+         {offsetof(struct MutableCFOptions, memtable_max_range_deletions),
+          OptionType::kUInt32T, OptionVerificationType::kNormal,
+          OptionTypeFlags::kNone}},
+
 };
 
 static std::unordered_map<std::string, OptionTypeInfo>

options/cf_options.h

@@ -172,7 +172,8 @@ struct MutableCFOptions {
             options.memtable_protection_bytes_per_key),
         sample_for_compression(
             options.sample_for_compression),  // TODO: is 0 fine here?
-        compression_per_level(options.compression_per_level) {
+        compression_per_level(options.compression_per_level),
+        memtable_max_range_deletions(options.memtable_max_range_deletions) {
     RefreshDerivedOptions(options.num_levels, options.compaction_style);
   }
 
@@ -220,7 +221,8 @@ struct MutableCFOptions {
         bottommost_compression(kDisableCompressionOption),
         last_level_temperature(Temperature::kUnknown),
         memtable_protection_bytes_per_key(0),
-        sample_for_compression(0) {}
+        sample_for_compression(0),
+        memtable_max_range_deletions(0) {}
 
   explicit MutableCFOptions(const Options& options);
 
@@ -313,6 +315,7 @@ struct MutableCFOptions {
 
   uint64_t sample_for_compression;
   std::vector<CompressionType> compression_per_level;
+  uint32_t memtable_max_range_deletions;
 
   // Derived options
   // Per-level target file size.

options/options.cc

@@ -444,6 +444,8 @@ void ColumnFamilyOptions::Dump(Logger* log) const {
     }
     ROCKS_LOG_HEADER(log, "Options.experimental_mempurge_threshold: %f",
                      experimental_mempurge_threshold);
+    ROCKS_LOG_HEADER(log, "   Options.memtable_max_range_deletions: %d",
+                     memtable_max_range_deletions);
 }  // ColumnFamilyOptions::Dump
 
 void Options::Dump(Logger* log) const {

options/options_helper.cc

@@ -278,6 +278,7 @@ void UpdateColumnFamilyOptions(const MutableCFOptions& moptions,
   cf_opts->compression_per_level = moptions.compression_per_level;
   cf_opts->last_level_temperature = moptions.last_level_temperature;
   cf_opts->bottommost_temperature = moptions.last_level_temperature;
+  cf_opts->memtable_max_range_deletions = moptions.memtable_max_range_deletions;
 }
 
 void UpdateColumnFamilyOptions(const ImmutableCFOptions& ioptions,

options/options_settable_test.cc

@@ -539,7 +539,8 @@ TEST_F(OptionsSettableTest, ColumnFamilyOptionsAllFieldsSettable) {
       "compaction_options_fifo={max_table_files_size=3;allow_"
       "compaction=false;age_for_warm=1;};"
       "blob_cache=1M;"
-      "memtable_protection_bytes_per_key=2;",
+      "memtable_protection_bytes_per_key=2;"
+      "memtable_max_range_deletions=999999;",
       new_options));
 
   ASSERT_NE(new_options->blob_cache.get(), nullptr);

options/options_test.cc

@@ -129,6 +129,7 @@ TEST_F(OptionsTest, GetOptionsFromMapTest) {
       {"blob_file_starting_level", "1"},
       {"prepopulate_blob_cache", "kDisable"},
       {"last_level_temperature", "kWarm"},
+      {"memtable_max_range_deletions", "0"},
   };
 
   std::unordered_map<std::string, std::string> db_options_map = {
@@ -270,6 +271,7 @@ TEST_F(OptionsTest, GetOptionsFromMapTest) {
   ASSERT_EQ(new_cf_opt.prepopulate_blob_cache, PrepopulateBlobCache::kDisable);
   ASSERT_EQ(new_cf_opt.last_level_temperature, Temperature::kWarm);
   ASSERT_EQ(new_cf_opt.bottommost_temperature, Temperature::kWarm);
+  ASSERT_EQ(new_cf_opt.memtable_max_range_deletions, 0);
 
   cf_options_map["write_buffer_size"] = "hello";
   ASSERT_NOK(GetColumnFamilyOptionsFromMap(exact, base_cf_opt, cf_options_map,
@@ -2372,6 +2374,7 @@ TEST_F(OptionsOldApiTest, GetOptionsFromMapTest) {
       {"blob_file_starting_level", "1"},
       {"prepopulate_blob_cache", "kDisable"},
       {"last_level_temperature", "kWarm"},
+      {"memtable_max_range_deletions", "0"},
   };
 
   std::unordered_map<std::string, std::string> db_options_map = {
@@ -2508,6 +2511,7 @@ TEST_F(OptionsOldApiTest, GetOptionsFromMapTest) {
   ASSERT_EQ(new_cf_opt.prepopulate_blob_cache, PrepopulateBlobCache::kDisable);
   ASSERT_EQ(new_cf_opt.last_level_temperature, Temperature::kWarm);
   ASSERT_EQ(new_cf_opt.bottommost_temperature, Temperature::kWarm);
+  ASSERT_EQ(new_cf_opt.memtable_max_range_deletions, 0);
 
   cf_options_map["write_buffer_size"] = "hello";
   ASSERT_NOK(GetColumnFamilyOptionsFromMap(