Should squash with earlier commit.

* Renames the option to memtable_max_range_deletions
* Tries to add to options/. Our build doesn't validate it properly.
* checking moved to RangeDelete path rather than Read Path

db/memtable.cc

@@ -116,7 +116,8 @@ MemTable::MemTable(const InternalKeyComparator& cmp,
       oldest_key_time_(std::numeric_limits<uint64_t>::max()),
       atomic_flush_seqno_(kMaxSequenceNumber),
       approximate_memory_usage_(0),
-      max_tombstones_count_(mutable_cf_options.max_tombstones_count_) {
+      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());
@@ -172,8 +173,9 @@ size_t MemTable::ApproximateMemoryUsage() {
 }
 
 bool MemTable::ShouldFlushNow() {
-  // Read path will set this if too many tombstones have been generated.
-  if (max_tombstones_reached_) {
+  // 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;
   }
 
@@ -762,9 +764,14 @@ Status MemTable::Add(SequenceNumber s, ValueType type,
       num_deletes_.store(num_deletes_.load(std::memory_order_relaxed) + 1,
                          std::memory_order_relaxed);
     } else if (type == kTypeRangeDeletion) {
-      num_range_deletes_.store(
-          num_range_deletes_.load(std::memory_order_relaxed) + 1,
-          std::memory_order_relaxed);
+      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_ &&
@@ -1275,13 +1282,6 @@ bool MemTable::Get(const LookupKey& key, std::string* value,
     return false;
   }
 
-  // Arrange for a flush if too many tombstones have been created.
-  if (max_tombstones_count_ > 0 && max_tombstones_reached_ == false &&
-      num_range_deletes_.load(std::memory_order_relaxed) >
-          max_tombstones_count_) {
-    max_tombstones_reached_ = true;
-  }
-
   PERF_TIMER_GUARD(get_from_memtable_time);
 
   std::unique_ptr<FragmentedRangeTombstoneIterator> range_del_iter(

db/memtable.h

@@ -615,11 +615,13 @@ class MemTable {
   // Gets refreshed inside `ApproximateMemoryUsage()` or `ShouldFlushNow`
   std::atomic<uint64_t> approximate_memory_usage_;
 
-  // max tombstones before flushing, 0 for unlimited.
-  uint32_t max_tombstones_count_ = 0;
-  // Read will turn this flag on, if max_tombstones_count_ is reached.
-  // ShouldFlushNow() will check this.
-  bool max_tombstones_reached_ = false;
+  // 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;
+
   // Flush job info of the current memtable.
   std::unique_ptr<FlushJobInfo> flush_job_info_;
 

include/rocksdb/options.h

@@ -331,10 +331,10 @@ struct ColumnFamilyOptions : public AdvancedColumnFamilyOptions {
   // Default: nullptr
   std::shared_ptr<SstPartitionerFactory> sst_partitioner_factory = nullptr;
 
-  // Automatic flush after tombstone count in memtable hits this limit.
-  // helps with delete-range + read workloads
-  //  0 to disable it completely
-  uint32_t max_tombstones_count = 0;
+  // 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();

java/rocksjni/options.cc

@@ -3906,24 +3906,24 @@ jbyte Java_org_rocksdb_Options_prepopulateBlobCache(JNIEnv*, jobject,
 
 /*
  * Class:     org_rocksdb_Options
- * Method:    setMaxTombstonesCount
+ * Method:    setMemtableMaxRangeDeletions
  * Signature: (JI)V
  */
-void Java_org_rocksdb_Options_setMaxTombstonesCount(
-    JNIEnv*, jobject, jlong jhandle, jint jmax_tombstones_count) {
+void Java_org_rocksdb_Options_setMemtableMaxRangeDeletions(
+    JNIEnv*, jobject, jlong jhandle, jint jmemtable_max_range_deletions) {
   auto* opts = reinterpret_cast<ROCKSDB_NAMESPACE::Options*>(jhandle);
-  opts->max_tombstones_count = jmax_tombstones_count;
+  opts->memtable_max_range_deletions = jmemtable_max_range_deletions;
 }
 
 /*
  * Class:     org_rocksdb_Options
- * Method:    maxTombstonesCount
+ * Method:    memtableMaxRangeDeletions
  * Signature: (J)I
  */
-jint Java_org_rocksdb_Options_maxTombstonesCount(JNIEnv*, jobject,
-                                                 jlong jhandle) {
+jint Java_org_rocksdb_Options_memtableMaxRangeDeletions(JNIEnv*, jobject,
+                                                        jlong jhandle) {
   auto* opts = reinterpret_cast<ROCKSDB_NAMESPACE::Options*>(jhandle);
-  return static_cast<jint>(opts->max_tombstones_count);
+  return static_cast<jint>(opts->memtable_max_range_deletions);
 }
 
 //////////////////////////////////////////////////////////////////////////////
@@ -5794,26 +5794,26 @@ jbyte Java_org_rocksdb_ColumnFamilyOptions_prepopulateBlobCache(JNIEnv*,
 
 /*
  * Class:     org_rocksdb_ColumnFamilyOptions
- * Method:    setMaxTombstonesCount
+ * Method:    setMemtableMaxRangeDeletions
  * Signature: (JI)V
  */
-void Java_org_rocksdb_ColumnFamilyOptions_setMaxTombstonesCount(
-    JNIEnv*, jobject, jlong jhandle, jint jmax_tombstones_count) {
+void Java_org_rocksdb_ColumnFamilyOptions_setMemtableMaxRangeDeletions(
+    JNIEnv*, jobject, jlong jhandle, jint jmemtable_max_range_deletions) {
   auto* opts =
       reinterpret_cast<ROCKSDB_NAMESPACE::ColumnFamilyOptions*>(jhandle);
-  opts->max_tombstones_count = jmax_tombstones_count;
+  opts->memtable_max_range_deletions = jmemtable_max_range_deletions;
 }
 
 /*
  * Class:     org_rocksdb_ColumnFamilyOptions
- * Method:    maxTombstonesCount
+ * Method:    memtableMaxRangeDeletions
  * Signature: (J)I
  */
-jint Java_org_rocksdb_ColumnFamilyOptions_maxTombstonesCount(JNIEnv*, jobject,
-                                                             jlong jhandle) {
+jint Java_org_rocksdb_ColumnFamilyOptions_memtableMaxRangeDeletions(
+    JNIEnv*, jobject, jlong jhandle) {
   auto* opts =
       reinterpret_cast<ROCKSDB_NAMESPACE::ColumnFamilyOptions*>(jhandle);
-  return static_cast<jint>(opts->max_tombstones_count);
+  return static_cast<jint>(opts->memtable_max_range_deletions);
 }
 
 /////////////////////////////////////////////////////////////////////

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

@@ -960,14 +960,14 @@ public SstPartitionerFactory sstPartitionerFactory() {
   }
 
   @Override
-  public ColumnFamilyOptions setMaxTombstonesCount(final int count) {
-    setMaxTombstonesCount(nativeHandle_, count);
+  public ColumnFamilyOptions setMemtableMaxRangeDeletions(final int count) {
+    setMemtableMaxRangeDeletions(nativeHandle_, count);
     return this;
   }
 
   @Override
-  public int maxTombstonesCount() {
-    return maxTombstonesCount(nativeHandle_);
+  public int memtableMaxRangeDeletions() {
+    return memtableMaxRangeDeletions(nativeHandle_);
   }
 
   //
@@ -1509,8 +1509,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 setMaxTombstonesCount(final long handle, final int count);
-  private native int maxTombstonesCount(final long handle);
+  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

@@ -507,24 +507,21 @@ T setCompressionOptions(
   SstPartitionerFactory sstPartitionerFactory();
 
   /**
-   * Sets the maximum tombstones ( range delete calls ) allowed
-   * in a delete-range + read work load.
-   * The value is only checked when a Get is called.
+   * 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 setMaxTombstonesCount(final int count);
+  T setMemtableMaxRangeDeletions(final int count);
 
   /**
-   * Gets the current setting of max tombstones allowed in a
-   * delete-range + read work load.
+   * Gets the current setting of maximum range deletes allowed
    * 0(default) indicates that feature is disabled.
    *
-   * @return current value of max_tombstones_count
+   * @return current value of memtable_max_range_deletions
    */
-  int maxTombstonesCount();
+  int memtableMaxRangeDeletions();
 
   /**
    * Compaction concurrent thread limiter for the column family.

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

@@ -1988,14 +1988,14 @@ public SstPartitionerFactory sstPartitionerFactory() {
   }
 
   @Override
-  public Options setMaxTombstonesCount(final int count) {
-    setMaxTombstonesCount(nativeHandle_, count);
+  public Options setMemtableMaxRangeDeletions(final int count) {
+    setMemtableMaxRangeDeletions(nativeHandle_, count);
     return this;
   }
 
   @Override
-  public int maxTombstonesCount() {
-    return maxTombstonesCount(nativeHandle_);
+  public int memtableMaxRangeDeletions() {
+    return memtableMaxRangeDeletions(nativeHandle_);
   }
 
   @Override
@@ -2517,8 +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 setMaxTombstonesCount(final long handle, final int count);
-  private native int maxTombstonesCount(final long handle);
+  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

@@ -713,12 +713,12 @@ public void cfPaths() throws IOException {
   }
 
   @Test
-  public void maxTombstonesCount() {
+  public void memtableMaxRangeDeletions() {
     try (final ColumnFamilyOptions options = new ColumnFamilyOptions()) {
-      assertThat(options.maxTombstonesCount()).isEqualTo(0);
+      assertThat(options.memtableMaxRangeDeletions()).isEqualTo(0);
       final int val = 32;
-      assertThat(options.setMaxTombstonesCount(val)).isEqualTo(options);
-      assertThat(options.maxTombstonesCount()).isEqualTo(val);
+      assertThat(options.setMemtableMaxRangeDeletions(val)).isEqualTo(options);
+      assertThat(options.memtableMaxRangeDeletions()).isEqualTo(val);
     }
   }
 }

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

@@ -1457,12 +1457,12 @@ public void skipCheckingSstFileSizesOnDbOpen() {
   }
 
   @Test
-  public void maxTombstonesCount() {
+  public void memtableMaxRangeDeletions() {
     try (final Options options = new Options()) {
-      assertThat(options.maxTombstonesCount()).isEqualTo(0);
+      assertThat(options.memtableMaxRangeDeletions()).isEqualTo(0);
       final int val = 32;
-      assertThat(options.setMaxTombstonesCount(val)).isEqualTo(options);
-      assertThat(options.maxTombstonesCount()).isEqualTo(val);
+      assertThat(options.setMemtableMaxRangeDeletions(val)).isEqualTo(options);
+      assertThat(options.memtableMaxRangeDeletions()).isEqualTo(val);
     }
   }
 

options/cf_options.cc

@@ -533,6 +533,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

@@ -176,7 +176,7 @@ struct MutableCFOptions {
         sample_for_compression(
             options.sample_for_compression),  // TODO: is 0 fine here?
         compression_per_level(options.compression_per_level),
-        max_tombstones_count_(options.max_tombstones_count) {
+        memtable_max_range_deletions(options.memtable_max_range_deletions) {
     RefreshDerivedOptions(options.num_levels, options.compaction_style);
   }
 
@@ -226,7 +226,7 @@ struct MutableCFOptions {
         memtable_protection_bytes_per_key(0),
         block_protection_bytes_per_key(0),
         sample_for_compression(0),
-        max_tombstones_count_(0) {}
+        memtable_max_range_deletions(0) {}
 
   explicit MutableCFOptions(const Options& options);
 
@@ -320,7 +320,7 @@ struct MutableCFOptions {
 
   uint64_t sample_for_compression;
   std::vector<CompressionType> compression_per_level;
-  uint32_t max_tombstones_count_;
+  uint32_t memtable_max_range_deletions;
 
   // Derived options
   // Per-level target file size.

options/options.cc

@@ -450,6 +450,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

@@ -270,6 +270,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

@@ -553,7 +553,8 @@ TEST_F(OptionsSettableTest, ColumnFamilyOptionsAllFieldsSettable) {
       "blob_cache=1M;"
       "memtable_protection_bytes_per_key=2;"
       "persist_user_defined_timestamps=true;"
-      "block_protection_bytes_per_key=1;",
+      "block_protection_bytes_per_key=1;"
+      "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) {
       {"prepopulate_blob_cache", "kDisable"},
       {"last_level_temperature", "kWarm"},
       {"persist_user_defined_timestamps", "true"},
+      {"memtable_max_range_deletions", "0"},
   };
 
   std::unordered_map<std::string, std::string> db_options_map = {
@@ -271,6 +272,7 @@ TEST_F(OptionsTest, GetOptionsFromMapTest) {
   ASSERT_EQ(new_cf_opt.last_level_temperature, Temperature::kWarm);
   ASSERT_EQ(new_cf_opt.bottommost_temperature, Temperature::kWarm);
   ASSERT_EQ(new_cf_opt.persist_user_defined_timestamps, true);
+  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,
@@ -2323,6 +2325,7 @@ TEST_F(OptionsOldApiTest, GetOptionsFromMapTest) {
       {"prepopulate_blob_cache", "kDisable"},
       {"last_level_temperature", "kWarm"},
       {"persist_user_defined_timestamps", "true"},
+      {"memtable_max_range_deletions", "0"},
   };
 
   std::unordered_map<std::string, std::string> db_options_map = {
@@ -2463,6 +2466,7 @@ TEST_F(OptionsOldApiTest, GetOptionsFromMapTest) {
   ASSERT_EQ(new_cf_opt.last_level_temperature, Temperature::kWarm);
   ASSERT_EQ(new_cf_opt.bottommost_temperature, Temperature::kWarm);
   ASSERT_EQ(new_cf_opt.persist_user_defined_timestamps, true);
+  ASSERT_EQ(new_cf_opt.memtable_max_range_deletions, 0);
 
   cf_options_map["write_buffer_size"] = "hello";
   ASSERT_NOK(GetColumnFamilyOptionsFromMap(cf_config_options, base_cf_opt,