format lint, add EnableIntraL0Compaction

db/compaction/compaction_picker_level.cc

@@ -902,6 +902,9 @@ bool LevelCompactionBuilder::PickIntraL0Compaction() {
   start_level_inputs_.clear();
   const std::vector<FileMetaData*>& level_files =
       vstorage_->LevelFiles(0 /* level */);
+  if (mutable_cf_options_.disable_intra_l0_compaction == true) {
+    return false;
+  }
   if (level_files.size() <
           static_cast<size_t>(
               mutable_cf_options_.level0_file_num_compaction_trigger + 2) ||

db/db_impl/db_impl.h

@@ -452,6 +452,8 @@ class DBImpl : public DB {
 
   Status EnableAutoCompaction(
       const std::vector<ColumnFamilyHandle*>& column_family_handles) override;
+  Status EnableIntraL0Compaction(
+      const std::vector<ColumnFamilyHandle*>& column_family_handles) override;
 
   void EnableManualCompaction() override;
   void DisableManualCompaction() override;

db/db_impl/db_impl_compaction_flush.cc

@@ -2793,7 +2793,19 @@ Status DBImpl::EnableAutoCompaction(
 
   return s;
 }
+Status DBImpl::EnableIntraL0Compaction(
+    const std::vector<ColumnFamilyHandle*>& column_family_handles) {
+  Status s;
+  for (auto cf_ptr : column_family_handles) {
+    Status status =
+        this->SetOptions(cf_ptr, {{"disable_intra_l0_compaction", "false"}});
+    if (!status.ok()) {
+      s = status;
+    }
+  }
 
+  return s;
+}
 // NOTE: Calling DisableManualCompaction() may overwrite the
 // user-provided canceled variable in CompactRangeOptions
 void DBImpl::DisableManualCompaction() {

include/rocksdb/db.h

@@ -1692,6 +1692,8 @@ class DB {
   virtual Status EnableAutoCompaction(
       const std::vector<ColumnFamilyHandle*>& column_family_handles) = 0;
 
+  virtual Status EnableIntraL0Compaction(
+      const std::vector<ColumnFamilyHandle*>& column_family_handles) = 0;
   // After this function call, CompactRange() or CompactFiles() will not
   // run compactions and fail. Calling this function will tell outstanding
   // manual compactions to abort and will wait for them to finish or abort

include/rocksdb/options.h

@@ -313,6 +313,9 @@ struct ColumnFamilyOptions : public AdvancedColumnFamilyOptions {
   // Dynamically changeable through SetOptions() API
   bool disable_auto_compactions = false;
 
+  // Disable Intra Level 0 Compaction for low write amplification.
+  bool disable_intra_l0_compaction = false;
+
   // This is a factory that provides TableFactory objects.
   // Default: a block-based table factory that provides a default
   // implementation of TableBuilder and TableReader with default

include/rocksdb/utilities/stackable_db.h

@@ -369,6 +369,11 @@ class StackableDB : public DB {
     return db_->EnableAutoCompaction(column_family_handles);
   }
 
+  Status EnableIntraL0Compaction(
+      const std::vector<ColumnFamilyHandle*>& column_family_handles) override {
+    return db_->EnableIntraL0Compaction(column_family_handles);
+  }
+
   void EnableManualCompaction() override {
     return db_->EnableManualCompaction();
   }

java/rocksjni/rocksjni.cc

@@ -2972,6 +2972,21 @@ void Java_org_rocksdb_RocksDB_enableAutoCompaction(JNIEnv* env, jclass,
   db->EnableAutoCompaction(cf_handles);
 }
 
+void Java_org_rocksdb_RocksDB_enableIntraL0Compaction(JNIEnv* env, jclass,
+                                                      jlong jdb_handle,
+                                                      jlongArray jcf_handles) {
+  auto* db = reinterpret_cast<ROCKSDB_NAMESPACE::DB*>(jdb_handle);
+  jboolean has_exception = JNI_FALSE;
+  const std::vector<ROCKSDB_NAMESPACE::ColumnFamilyHandle*> cf_handles =
+      ROCKSDB_NAMESPACE::JniUtil::fromJPointers<
+          ROCKSDB_NAMESPACE::ColumnFamilyHandle>(env, jcf_handles,
+                                                 &has_exception);
+  if (has_exception == JNI_TRUE) {
+    // exception occurred
+    return;
+  }
+  db->EnableIntraL0Compaction(cf_handles);
+}
 /*
  * Class:     org_rocksdb_RocksDB
  * Method:    numberLevels

options/cf_options.cc

@@ -366,6 +366,10 @@ static std::unordered_map<std::string, OptionTypeInfo>
          {offsetof(struct MutableCFOptions, disable_auto_compactions),
           OptionType::kBoolean, OptionVerificationType::kNormal,
           OptionTypeFlags::kMutable}},
+        {"disable_intra_l0_compaction",
+         {offsetof(struct MutableCFOptions, disable_intra_l0_compaction),
+          OptionType::kBoolean, OptionVerificationType::kNormal,
+          OptionTypeFlags::kMutable}},
         {"table_factory",
          {offsetof(struct MutableCFOptions, table_factory),
           OptionType::kCustomizable, OptionVerificationType::kByName,

options/cf_options.h

@@ -120,6 +120,7 @@ struct MutableCFOptions {
         experimental_mempurge_threshold(
             options.experimental_mempurge_threshold),
         disable_auto_compactions(options.disable_auto_compactions),
+        disable_intra_l0_compaction(options.disable_intra_l0_compaction),
         table_factory(options.table_factory),
         soft_pending_compaction_bytes_limit(
             options.soft_pending_compaction_bytes_limit),
@@ -298,6 +299,8 @@ struct MutableCFOptions {
 
   // Compaction related options
   bool disable_auto_compactions;
+  bool disable_intra_l0_compaction;
+
   std::shared_ptr<TableFactory> table_factory;
   uint64_t soft_pending_compaction_bytes_limit;
   uint64_t hard_pending_compaction_bytes_limit;

options/options.cc

@@ -310,6 +310,8 @@ void ColumnFamilyOptions::Dump(Logger* log) const {
                    hard_pending_compaction_bytes_limit);
   ROCKS_LOG_HEADER(log, "               Options.disable_auto_compactions: %d",
                    disable_auto_compactions);
+  ROCKS_LOG_HEADER(log, "  Options.disable_intra_l0_compaction: %d",
+                   disable_intra_l0_compaction);
 
   const auto& it_compaction_style =
       compaction_style_to_string.find(compaction_style);

options/options_helper.cc

@@ -239,6 +239,7 @@ void UpdateColumnFamilyOptions(const MutableCFOptions& moptions,
 
   // Compaction related options
   cf_opts->disable_auto_compactions = moptions.disable_auto_compactions;
+  cf_opts->disable_intra_l0_compaction = moptions.disable_intra_l0_compaction;
   cf_opts->table_factory = moptions.table_factory;
   cf_opts->soft_pending_compaction_bytes_limit =
       moptions.soft_pending_compaction_bytes_limit;