[WIP] ...

db/c.cc

@@ -2508,6 +2508,10 @@ void rocksdb_options_set_disable_auto_compactions(rocksdb_options_t* opt, int di
   opt->rep.disable_auto_compactions = disable;
 }
 
+void rocksdb_options_set_enable_lazy_compaction(rocksdb_options_t* opt, int enable) {
+  opt->rep.enable_lazy_compaction = enable;
+}
+
 void rocksdb_options_set_optimize_filters_for_hits(rocksdb_options_t* opt, int v) {
   opt->rep.optimize_filters_for_hits = v;
 }

db/column_family.cc

@@ -910,7 +910,8 @@ void ColumnFamilyData::CreateNewMemtable(
 }
 
 bool ColumnFamilyData::NeedsCompaction() const {
-  return compaction_picker_->NeedsCompaction(current_->storage_info());
+  return !current_->storage_info()->IsPickFail() &&
+         compaction_picker_->NeedsCompaction(current_->storage_info());
 }
 
 Compaction* ColumnFamilyData::PickCompaction(
@@ -919,6 +920,8 @@ Compaction* ColumnFamilyData::PickCompaction(
       GetName(), mutable_options, current_->storage_info(), log_buffer);
   if (result != nullptr) {
     result->SetInputVersion(current_);
+  } else {
+    current_->storage_info()->SetPickFail();
   }
   return result;
 }

db/compaction.cc

@@ -139,6 +139,8 @@ Compaction::Compaction(VersionStorageInfo* vstorage,
                        std::vector<FileMetaData*> _grandparents,
                        bool _manual_compaction, double _score,
                        bool _deletion_compaction,
+                       CompactionVarieties _compaction_varieties,
+                       const std::vector<InternalKeyRange>& _input_range,
                        CompactionReason _compaction_reason)
     : input_vstorage_(vstorage),
       start_level_(_inputs[0].level),
@@ -162,6 +164,8 @@ Compaction::Compaction(VersionStorageInfo* vstorage,
       is_full_compaction_(IsFullCompaction(vstorage, inputs_)),
       is_manual_compaction_(_manual_compaction),
       is_trivial_move_(false),
+      compaction_varieties_(_compaction_varieties),
+      input_range_(_input_range),
       compaction_reason_(_compaction_reason) {
   MarkFilesBeingCompacted(true);
   if (is_manual_compaction_) {
@@ -220,6 +224,12 @@ bool Compaction::IsTrivialMove() const {
   // If start_level_== output_level_, the purpose is to force compaction
   // filter to be applied to that level, and thus cannot be a trivial move.
 
+  // kUniversalTrivialMove don't need check
+  if (compaction_reason_ == CompactionReason::kUniversalTrivialMove) {
+    assert(is_trivial_move_);
+    return true;
+  }
+
   // Check if start level have files with overlapping ranges
   if (start_level_ == 0 && input_vstorage_->level0_non_overlapping() == false) {
     // We cannot move files from L0 to L1 if the files are overlapping

db/compaction.h

@@ -31,6 +31,12 @@ class ColumnFamilyData;
 class VersionStorageInfo;
 class CompactionFilter;
 
+enum CompactionVarieties : int {
+  kGeneralCompaction,
+  kLinkCompaction,
+  kMapCompaction,
+};
+
 // A Compaction encapsulates information about a compaction.
 class Compaction {
  public:
@@ -44,6 +50,8 @@ class Compaction {
              std::vector<FileMetaData*> grandparents,
              bool manual_compaction = false, double score = -1,
              bool deletion_compaction = false,
+             CompactionVarieties compaction_varieties = kGeneralCompaction,
+             const std::vector<InternalKeyRange>& input_range = {},
              CompactionReason compaction_reason = CompactionReason::kUnknown);
 
   // No copying allowed
@@ -134,6 +142,16 @@ class Compaction {
   // If true, then the compaction can be done by simply deleting input files.
   bool deletion_compaction() const { return deletion_compaction_; }
 
+  // Compaction varieties
+  CompactionVarieties compaction_varieties() const {
+    return compaction_varieties_;
+  }
+
+  // Range limit for inputs
+  const std::vector<InternalKeyRange>& input_range() const {
+    return input_range_;
+  };
+
   // Add all inputs to this compaction as delete operations to *edit.
   void AddInputDeletions(VersionEdit* edit);
 
@@ -292,6 +310,12 @@ class Compaction {
   // If true, then the comaction can be done by simply deleting input files.
   const bool deletion_compaction_;
 
+  // Compaction varieties
+  const CompactionVarieties compaction_varieties_;
+
+  // Range limit for inputs
+  const std::vector<InternalKeyRange> input_range_;
+
   // Compaction input files organized by level. Constant after construction
   const std::vector<CompactionInputFiles> inputs_;
 

db/compaction_picker.cc

@@ -1358,7 +1358,8 @@ Compaction* LevelCompactionBuilder::GetCompaction() {
                          output_level_, vstorage_->base_level()),
       GetCompressionOptions(ioptions_, vstorage_, output_level_),
       /* max_subcompactions */ 0, std::move(grandparents_), is_manual_,
-      start_level_score_, false /* deletion_compaction */, compaction_reason_);
+      start_level_score_, false /* deletion_compaction */,
+      kGeneralCompaction, {}, compaction_reason_);
 
   // If it's level 0 compaction, make sure we don't execute any other level 0
   // compactions in parallel
@@ -1608,7 +1609,8 @@ Compaction* FIFOCompactionPicker::PickTTLCompaction(
       vstorage, ioptions_, mutable_cf_options, std::move(inputs), 0, 0, 0, 0,
       kNoCompression, ioptions_.compression_opts, /* max_subcompactions */ 0,
       {}, /* is manual */ false, vstorage->CompactionScore(0),
-      /* is deletion compaction */ true, CompactionReason::kFIFOTtl);
+      /* is deletion compaction */ true, kGeneralCompaction, {},
+      CompactionReason::kFIFOTtl);
   return c;
 }
 
@@ -1648,7 +1650,7 @@ Compaction* FIFOCompactionPicker::PickSizeCompaction(
             0 /* output path ID */, mutable_cf_options.compression,
             ioptions_.compression_opts, 0 /* max_subcompactions */, {},
             /* is manual */ false, vstorage->CompactionScore(0),
-            /* is deletion compaction */ false,
+            /* is deletion compaction */ false, kGeneralCompaction, {},
             CompactionReason::kFIFOReduceNumFiles);
         return c;
       }
@@ -1696,7 +1698,8 @@ Compaction* FIFOCompactionPicker::PickSizeCompaction(
       vstorage, ioptions_, mutable_cf_options, std::move(inputs), 0, 0, 0, 0,
       kNoCompression, ioptions_.compression_opts, /* max_subcompactions */ 0,
       {}, /* is manual */ false, vstorage->CompactionScore(0),
-      /* is deletion compaction */ true, CompactionReason::kFIFOMaxSize);
+      /* is deletion compaction */ true, kGeneralCompaction, {},
+      CompactionReason::kFIFOMaxSize);
   return c;
 }
 

db/compaction_picker_universal.cc

@@ -179,9 +179,10 @@ void UniversalCompactionPicker::SortedRun::Dump(char* out_buf,
     if (file->fd.GetPathId() == 0 || !print_path) {
       snprintf(out_buf, out_buf_size, "file %" PRIu64, file->fd.GetNumber());
     } else {
-      snprintf(out_buf, out_buf_size, "file %" PRIu64
-                                      "(path "
-                                      "%" PRIu32 ")",
+      snprintf(out_buf, out_buf_size,
+               "file %" PRIu64
+               "(path "
+               "%" PRIu32 ")",
                file->fd.GetNumber(), file->fd.GetPathId());
     }
   } else {
@@ -352,8 +353,8 @@ Compaction* UniversalCompactionPicker::PickCompaction(
     return nullptr;
   }
 
-  if (mutable_cf_options.compaction_options_universal.allow_trivial_move ==
-      true) {
+  if (c->compaction_reason() != CompactionReason::kUniversalTrivialMove &&
+      mutable_cf_options.compaction_options_universal.allow_trivial_move) {
     c->set_is_trivial_move(IsInputFilesNonOverlapping(c));
   }
 
@@ -440,6 +441,101 @@ uint32_t UniversalCompactionPicker::GetPathId(
   return p;
 }
 
+Compaction* UniversalCompactionPicker::PickTrivialMove(
+    const std::string& cf_name, const MutableCFOptions& mutable_cf_options,
+    VersionStorageInfo* vstorage, LogBuffer* log_buffer) {
+  if (!mutable_cf_options.compaction_options_universal.allow_trivial_move) {
+    return nullptr;
+  }
+  int output_level = vstorage->num_levels() - 1;
+  // last level is reserved for the files ingested behind
+  if (ioptions_.allow_ingest_behind) {
+    --output_level;
+  }
+  int start_level = 0;
+  while (true) {
+    // found an empty level
+    for (; output_level >= 1; --output_level) {
+      if (!vstorage->LevelFiles(output_level).empty()) {
+        continue;
+      }
+      bool invalid = false;
+      for (auto c : compactions_in_progress_) {
+        if (c->output_level() == output_level) {
+          invalid = true;
+          break;
+        }
+      }
+      if (invalid) {
+        continue;
+      }
+      break;
+    }
+    if (output_level < 1) {
+      return nullptr;
+    }
+    bool invalid = false;
+    // found an non empty level
+    for (start_level = output_level - 1; start_level > 0; --start_level) {
+      invalid = false;
+      for (auto c : compactions_in_progress_) {
+        if (c->output_level() == start_level) {
+          invalid = true;
+          break;
+        }
+      }
+      if (!invalid && vstorage->LevelFiles(start_level).empty()) {
+        continue;
+      }
+      break;
+    }
+    if (!invalid) {
+      // will move lv0 last sst
+      if (start_level == 0) {
+        break;
+      }
+      auto& files = vstorage->LevelFiles(start_level);
+      for (size_t i = 0; i < files.size(); i++) {
+        if (files[i]->being_compacted) {
+          invalid = true;
+        }
+      }
+      if (!invalid) {
+        break;
+      }
+    }
+    output_level = start_level - 1;
+  }
+  CompactionInputFiles inputs;
+  inputs.level = start_level;
+  uint32_t path_id = 0;
+  if (start_level == 0) {
+    auto& level0_files = vstorage->LevelFiles(0);
+    if (level0_files.empty() || level0_files.back()->being_compacted) {
+      return nullptr;
+    }
+    FileMetaData* meta = level0_files.back();
+    inputs.files = {meta};
+    path_id = meta->fd.GetPathId();
+  } else {
+    inputs.files = vstorage->LevelFiles(start_level);
+    path_id = inputs.files.front()->fd.GetPathId();
+  }
+  assert(!AreFilesInCompaction(inputs.files));
+  auto c = new Compaction(
+      vstorage, ioptions_, mutable_cf_options, {std::move(inputs)},
+      output_level,
+      MaxFileSizeForLevel(mutable_cf_options, output_level,
+                          kCompactionStyleUniversal),
+      LLONG_MAX, path_id, kNoCompression,
+      GetCompressionOptions(ioptions_, vstorage, output_level), 0,
+      /* grandparents */ {}, /* is manual */ false, 0,
+      false /* deletion_compaction */, kGeneralCompaction, {},
+      CompactionReason::kUniversalTrivialMove);
+  c->set_is_trivial_move(true);
+  return c;
+}
+
 //
 // Consider compaction files based on their size differences with
 // the next file in time order.
@@ -639,7 +735,8 @@ Compaction* UniversalCompactionPicker::PickCompactionToReduceSortedRuns(
       GetCompressionOptions(ioptions_, vstorage, start_level,
                             enable_compression),
       /* max_subcompactions */ 0, /* grandparents */ {}, /* is manual */ false,
-      score, false /* deletion_compaction */, compaction_reason);
+      score, false /* deletion_compaction */, kGeneralCompaction, {},
+      compaction_reason);
 }
 
 // Look at overall size amplification. If size amplification
@@ -779,7 +876,7 @@ Compaction* UniversalCompactionPicker::PickCompactionToReduceSizeAmp(
                          1),
       GetCompressionOptions(ioptions_, vstorage, output_level),
       /* max_subcompactions */ 0, /* grandparents */ {}, /* is manual */ false,
-      score, false /* deletion_compaction */,
+      score, false /* deletion_compaction */, kGeneralCompaction, {},
       CompactionReason::kUniversalSizeAmplification);
 }
 
@@ -899,7 +996,7 @@ Compaction* UniversalCompactionPicker::PickDeleteTriggeredCompaction(
                          1),
       GetCompressionOptions(ioptions_, vstorage, output_level),
       /* max_subcompactions */ 0, /* grandparents */ {}, /* is manual */ true,
-      score, false /* deletion_compaction */,
+      score, false /* deletion_compaction */, kGeneralCompaction, {},
       CompactionReason::kFilesMarkedForCompaction);
 }
 }  // namespace rocksdb

db/compaction_picker_universal.h

@@ -60,6 +60,11 @@ class UniversalCompactionPicker : public CompactionPicker {
     bool being_compacted;
   };
 
+  // Pick universal trivial move for push sst to bottom level
+  Compaction* PickTrivialMove(
+      const std::string& cf_name, const MutableCFOptions& mutable_cf_options,
+      VersionStorageInfo* vstorage, LogBuffer* log_buffer);
+
   // Pick Universal compaction to limit read amplification
   Compaction* PickCompactionToReduceSortedRuns(
       const std::string& cf_name, const MutableCFOptions& mutable_cf_options,

db/db_impl_compaction_flush.cc

@@ -886,7 +886,7 @@ Status DBImpl::ReFitLevel(ColumnFamilyData* cfd, int level, int target_level) {
       edit.AddFile(to_level, f->fd.GetNumber(), f->fd.GetPathId(),
                    f->fd.GetFileSize(), f->smallest, f->largest,
                    f->smallest_seqno, f->largest_seqno,
-                   f->marked_for_compaction);
+                   f->marked_for_compaction, f->file_gene);
     }
     ROCKS_LOG_DEBUG(immutable_db_options_.info_log,
                     "[%s] Apply version edit:\n%s", cfd->GetName().c_str(),
@@ -1805,7 +1805,7 @@ Status DBImpl::BackgroundCompaction(bool* made_progress,
         c->edit()->AddFile(c->output_level(), f->fd.GetNumber(),
                            f->fd.GetPathId(), f->fd.GetFileSize(), f->smallest,
                            f->largest, f->smallest_seqno, f->largest_seqno,
-                           f->marked_for_compaction);
+                           f->marked_for_compaction, f->file_gene);
 
         ROCKS_LOG_BUFFER(
             log_buffer,

db/db_impl_experimental.cc

@@ -132,7 +132,7 @@ Status DBImpl::PromoteL0(ColumnFamilyHandle* column_family, int target_level) {
       edit.AddFile(target_level, f->fd.GetNumber(), f->fd.GetPathId(),
                    f->fd.GetFileSize(), f->smallest, f->largest,
                    f->smallest_seqno, f->largest_seqno,
-                   f->marked_for_compaction);
+                   f->marked_for_compaction, f->file_gene);
     }
 
     status = versions_->LogAndApply(cfd, *cfd->GetLatestMutableCFOptions(),

db/db_impl_open.cc

@@ -970,7 +970,7 @@ Status DBImpl::WriteLevel0TableForRecovery(int job_id, ColumnFamilyData* cfd,
     edit->AddFile(level, meta.fd.GetNumber(), meta.fd.GetPathId(),
                   meta.fd.GetFileSize(), meta.smallest, meta.largest,
                   meta.smallest_seqno, meta.largest_seqno,
-                  meta.marked_for_compaction);
+                  meta.marked_for_compaction, meta.file_gene);
   }
 
   InternalStats::CompactionStats stats(CompactionReason::kFlush, 1);

db/dbformat.h

@@ -250,6 +250,11 @@ class InternalKey {
   std::string DebugString(bool hex = false) const;
 };
 
+// range smallest .. largest (close interval)
+struct InternalKeyRange {
+  InternalKey smallest, largest;
+};
+
 inline int InternalKeyComparator::Compare(
     const InternalKey& a, const InternalKey& b) const {
   return Compare(a.Encode(), b.Encode());

db/external_sst_file_ingestion_job.cc

@@ -204,7 +204,7 @@ Status ExternalSstFileIngestionJob::Run() {
     edit_.AddFile(f.picked_level, f.fd.GetNumber(), f.fd.GetPathId(),
                   f.fd.GetFileSize(), f.smallest_internal_key(),
                   f.largest_internal_key(), f.assigned_seqno, f.assigned_seqno,
-                  false);
+                  false, 0);
   }
 
   if (consumed_seqno) {

db/flush_job.cc

@@ -390,7 +390,7 @@ Status FlushJob::WriteLevel0Table() {
     edit_->AddFile(0 /* level */, meta_.fd.GetNumber(), meta_.fd.GetPathId(),
                    meta_.fd.GetFileSize(), meta_.smallest, meta_.largest,
                    meta_.smallest_seqno, meta_.largest_seqno,
-                   meta_.marked_for_compaction);
+                   meta_.marked_for_compaction, meta_.file_gene);
   }
 
   // Note that here we treat flush as level 0 compaction in internal stats

db/range_del_aggregator.cc

-//  Copyright (c) 2016-present, Facebook, Inc.  All rights reserved.
+//  Copyright (c) 2016-present, Facebook, Inc.  All rights reserved.
 //  This source code is licensed under both the GPLv2 (found in the
 //  COPYING file in the root directory) and Apache 2.0 License
 //  (found in the LICENSE.Apache file in the root directory).

db/repair.cc

@@ -569,7 +569,8 @@ class Repairer {
         edit.AddFile(0, table->meta.fd.GetNumber(), table->meta.fd.GetPathId(),
                      table->meta.fd.GetFileSize(), table->meta.smallest,
                      table->meta.largest, table->min_sequence,
-                     table->max_sequence, table->meta.marked_for_compaction);
+                     table->max_sequence, table->meta.marked_for_compaction,
+                     table->meta.file_gene);
       }
       assert(next_file_number_ > 0);
       vset_.MarkFileNumberUsed(next_file_number_ - 1);

db/version_edit.cc

@@ -50,6 +50,7 @@ enum CustomTag : uint32_t {
   // kMinLogNumberToKeep as part of a CustomTag as a hack. This should be
   // removed when manifest becomes forward-comptabile.
   kMinLogNumberToKeepHack = 3,
+  kFileGene = 64,
   kPathId = 65,
 };
 // If this bit for the custom tag is set, opening DB should fail if
@@ -117,7 +118,8 @@ bool VersionEdit::EncodeTo(std::string* dst) const {
       return false;
     }
     bool has_customized_fields = false;
-    if (f.marked_for_compaction || has_min_log_number_to_keep_) {
+    if (f.marked_for_compaction || has_min_log_number_to_keep_ ||
+        f.file_gene != 0) {
       PutVarint32(dst, kNewFile4);
       has_customized_fields = true;
     } else if (f.fd.GetPathId() == 0) {
@@ -180,6 +182,11 @@ bool VersionEdit::EncodeTo(std::string* dst) const {
         PutLengthPrefixedSlice(dst, Slice(varint_log_number));
         min_log_num_written = true;
       }
+      if (f.file_gene) {
+        PutVarint32(dst, CustomTag::kFileGene);
+        char p = f.file_gene;
+        PutLengthPrefixedSlice(dst, Slice(&p, 1));
+      }
       TEST_SYNC_POINT_CALLBACK("VersionEdit::EncodeTo:NewFile4:CustomizeFields",
                                dst);
 
@@ -278,6 +285,12 @@ const char* VersionEdit::DecodeNewFile4From(Slice* input) {
           }
           has_min_log_number_to_keep_ = true;
           break;
+        case kFileGene:
+          if (field.size() != 1) {
+            return "need_compaction field wrong size";
+          }
+          f.file_gene = field[0];
+          break;
         default:
           if ((custom_tag & kCustomTagNonSafeIgnoreMask) != 0) {
             // Should not proceed if cannot understand it

db/version_edit.h

@@ -73,6 +73,12 @@ struct FileSampledStats {
   mutable std::atomic<uint64_t> num_reads_sampled;
 };
 
+enum class FileGene {
+  kSST,
+  kLink,
+  kMap
+};
+
 struct FileMetaData {
   FileDescriptor fd;
   InternalKey smallest;            // Smallest internal key served by table
@@ -107,6 +113,8 @@ struct FileMetaData {
   bool marked_for_compaction;  // True if client asked us nicely to compact this
                                // file.
 
+  uint8_t file_gene;           // Zero for plain sst
+
   FileMetaData()
       : smallest_seqno(kMaxSequenceNumber),
         largest_seqno(0),
@@ -119,7 +127,8 @@ struct FileMetaData {
         refs(0),
         being_compacted(false),
         init_stats_from_file(false),
-        marked_for_compaction(false) {}
+        marked_for_compaction(false),
+        file_gene(0) {}
 
   // REQUIRED: Keys must be given to the function in sorted order (it expects
   // the last key to be the largest).
@@ -230,7 +239,7 @@ class VersionEdit {
                uint64_t file_size, const InternalKey& smallest,
                const InternalKey& largest, const SequenceNumber& smallest_seqno,
                const SequenceNumber& largest_seqno,
-               bool marked_for_compaction) {
+               bool marked_for_compaction, uint8_t file_gene) {
     assert(smallest_seqno <= largest_seqno);
     FileMetaData f;
     f.fd = FileDescriptor(file, file_path_id, file_size);
@@ -239,6 +248,7 @@ class VersionEdit {
     f.smallest_seqno = smallest_seqno;
     f.largest_seqno = largest_seqno;
     f.marked_for_compaction = marked_for_compaction;
+    f.file_gene = file_gene;
     new_files_.emplace_back(level, std::move(f));
   }
 

db/version_set.cc

@@ -1114,6 +1114,7 @@ VersionStorageInfo::VersionStorageInfo(
       current_num_samples_(0),
       estimated_compaction_needed_bytes_(0),
       finalized_(false),
+      is_pick_fail_(false),
       force_consistency_checks_(_force_consistency_checks) {
   if (ref_vstorage != nullptr) {
     accumulated_file_size_ = ref_vstorage->accumulated_file_size_;
@@ -1658,6 +1659,7 @@ void VersionStorageInfo::ComputeCompactionScore(
       }
     }
   }
+  is_pick_fail_ = false;
   ComputeFilesMarkedForCompaction();
   ComputeBottommostFilesMarkedForCompaction();
   if (mutable_cf_options.ttl > 0) {
@@ -3964,7 +3966,7 @@ Status VersionSet::WriteSnapshot(log::Writer* log) {
           edit.AddFile(level, f->fd.GetNumber(), f->fd.GetPathId(),
                        f->fd.GetFileSize(), f->smallest, f->largest,
                        f->smallest_seqno, f->largest_seqno,
-                       f->marked_for_compaction);
+                       f->marked_for_compaction, f->file_gene);
         }
       }
       edit.SetLogNumber(cfd->GetLogNumber());

db/version_set.h

@@ -242,6 +242,12 @@ class VersionStorageInfo {
   bool HasOverlappingUserKey(const std::vector<FileMetaData*>* inputs,
                              int level);
 
+  // Set picker compaction fail
+  void SetPickFail() { is_pick_fail_ = true; }
+
+  // Is picker compaction fail
+  bool IsPickFail() const { return is_pick_fail_; }
+
   int num_levels() const { return num_levels_; }
 
   // REQUIRES: This version has been saved (see VersionSet::SaveTo)
@@ -514,6 +520,7 @@ class VersionStorageInfo {
   uint64_t estimated_compaction_needed_bytes_;
 
   bool finalized_;
+  bool is_pick_fail_;
 
   // If set to true, we will run consistency checks even if RocksDB
   // is compiled in release mode

include/rocksdb/listener.h

@@ -89,6 +89,8 @@ enum class CompactionReason : int {
   kExternalSstIngestion,
   // total number of compaction reasons, new reasons must be added above this.
   kNumOfReasons,
+  // [Universal] trivial move files
+  kUniversalTrivialMove,
 };
 
 enum class FlushReason : int {

include/rocksdb/options.h

@@ -273,6 +273,9 @@ struct ColumnFamilyOptions : public AdvancedColumnFamilyOptions {
   // Dynamically changeable through SetOptions() API
   bool disable_auto_compactions = false;
 
+  // Enable map or link compaction
+  bool enable_lazy_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

options/cf_options.cc

@@ -147,6 +147,8 @@ void MutableCFOptions::Dump(Logger* log) const {
       prefix_extractor == nullptr ? "nullptr" : prefix_extractor->Name());
   ROCKS_LOG_INFO(log, "                 disable_auto_compactions: %d",
                  disable_auto_compactions);
+  ROCKS_LOG_INFO(log, "                   enable_lazy_compaction: %d",
+                 enable_lazy_compaction);
   ROCKS_LOG_INFO(log, "      soft_pending_compaction_bytes_limit: %" PRIu64,
                  soft_pending_compaction_bytes_limit);
   ROCKS_LOG_INFO(log, "      hard_pending_compaction_bytes_limit: %" PRIu64,

options/cf_options.h

@@ -218,6 +218,7 @@ struct MutableCFOptions {
 
   // Compaction related options
   bool disable_auto_compactions;
+  bool enable_lazy_compaction;
   uint64_t soft_pending_compaction_bytes_limit;
   uint64_t hard_pending_compaction_bytes_limit;
   int level0_file_num_compaction_trigger;

options/options.cc

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

options/options_helper.cc

@@ -150,6 +150,8 @@ ColumnFamilyOptions BuildColumnFamilyOptions(
   // Compaction related options
   cf_opts.disable_auto_compactions =
       mutable_cf_options.disable_auto_compactions;
+  cf_opts.enable_lazy_compaction =
+      mutable_cf_options.enable_lazy_compaction;
   cf_opts.soft_pending_compaction_bytes_limit =
       mutable_cf_options.soft_pending_compaction_bytes_limit;
   cf_opts.hard_pending_compaction_bytes_limit =
@@ -1659,6 +1661,10 @@ std::unordered_map<std::string, OptionTypeInfo>
          {offset_of(&ColumnFamilyOptions::disable_auto_compactions),
           OptionType::kBoolean, OptionVerificationType::kNormal, true,
           offsetof(struct MutableCFOptions, disable_auto_compactions)}},
+        {"enable_lazy_compactions",
+         {offset_of(&ColumnFamilyOptions::enable_lazy_compaction),
+          OptionType::kBoolean, OptionVerificationType::kNormal, true,
+          offsetof(struct MutableCFOptions, enable_lazy_compaction)}},
         {"filter_deletes",
          {0, OptionType::kBoolean, OptionVerificationType::kDeprecated, true,
           0}},

tools/db_bench_tool.cc

@@ -936,6 +936,8 @@ DEFINE_bool(readonly, false, "Run read only benchmarks.");
 
 DEFINE_bool(disable_auto_compactions, false, "Do not auto trigger compactions");
 
+DEFINE_bool(enable_lazy_compaction, false, "Enable map or link compaction");
+
 DEFINE_uint64(wal_ttl_seconds, 0, "Set the TTL for the WAL Files in seconds.");
 DEFINE_uint64(wal_size_limit_MB, 0, "Set the size limit for the WAL Files"
               " in MB.");
@@ -3318,6 +3320,7 @@ void VerifyDBFromDB(std::string& truth_db_name) {
     options.table_cache_numshardbits = FLAGS_table_cache_numshardbits;
     options.max_compaction_bytes = FLAGS_max_compaction_bytes;
     options.disable_auto_compactions = FLAGS_disable_auto_compactions;
+    options.enable_lazy_compaction = FLAGS_enable_lazy_compaction;
     options.optimize_filters_for_hits = FLAGS_optimize_filters_for_hits;
 
     // fill storage options

util/testutil.cc

@@ -302,6 +302,7 @@ void RandomInitCFOptions(ColumnFamilyOptions* cf_opt, Random* rnd) {
   // boolean options
   cf_opt->report_bg_io_stats = rnd->Uniform(2);
   cf_opt->disable_auto_compactions = rnd->Uniform(2);
+  cf_opt->enable_lazy_compaction = rnd->Uniform(2);
   cf_opt->inplace_update_support = rnd->Uniform(2);
   cf_opt->level_compaction_dynamic_level_bytes = rnd->Uniform(2);
   cf_opt->optimize_filters_for_hits = rnd->Uniform(2);