[WIP] Add Link Compaction related interfaces

Author: royguo

db/column_family.cc

@@ -994,6 +994,13 @@ Compaction* ColumnFamilyData::PickCompaction(
   return result;
 }
 
+// TODO(guokuankuan@bytedance.com)
+Compaction* ColumnFamilyData::PickLinkCompaction(
+    const MutableCFOptions& mutable_options,
+    const std::vector<SequenceNumber>& snapshots, LogBuffer* log_buffer) {
+  return nullptr;
+}
+
 Compaction* ColumnFamilyData::PickGarbageCollection(
     const MutableCFOptions& mutable_options, LogBuffer* log_buffer) {
   StopWatch sw(ioptions_.env, ioptions_.statistics,

db/column_family.h

@@ -278,6 +278,12 @@ class ColumnFamilyData {
                              const std::vector<SequenceNumber>& snapshots,
                              LogBuffer* log_buffer);
 
+  // Pick a Link Compaction and set related version information to it.
+  Compaction* PickLinkCompaction(const MutableCFOptions& mutable_options,
+                             const std::vector<SequenceNumber>& snapshots,
+                             LogBuffer* log_buffer);
+
+
   Compaction* PickGarbageCollection(const MutableCFOptions& mutable_options,
                                     LogBuffer* log_buffer);
   // Check if the passed range overlap with any running compactions.

db/compaction.h

@@ -77,6 +77,7 @@ enum CompactionType {
   kKeyValueCompaction = 0,
   kMapCompaction = 1,
   kGarbageCollection = 2,
+  kLinkCompaction = 3,
 };
 
 struct CompactionParams {

db/compaction_iterator.cc

@@ -170,7 +170,7 @@ CompactionIterator::CompactionIterator(
     level_ptrs_ = std::vector<size_t>(compaction_->number_levels(), 0);
   }
 
-  if (snapshots_->size() == 0) {
+  if (snapshots_->empty()) {
     // optimize for fast path if there are no snapshots
     visible_at_tip_ = true;
     earliest_snapshot_ = kMaxSequenceNumber;

db/compaction_job.cc

@@ -304,7 +304,9 @@ struct CompactionJob::SubcompactionState {
   }
 
   struct RebuildBlobsInfo {
+    // File numbers
     chash_set<uint64_t> blobs;
+    // pop_count = planned file count - actual used file count.
     size_t pop_count;
   };
   struct BlobRefInfo {
@@ -1493,6 +1495,9 @@ void CompactionJob::ProcessCompaction(SubcompactionState* sub_compact) {
     case kGarbageCollection:
       ProcessGarbageCollection(sub_compact);
       break;
+    case kLinkCompaction:
+      ProcessLinkCompaction(sub_compact);
+      break;
     default:
       assert(false);
       break;
@@ -1765,6 +1770,9 @@ void CompactionJob::ProcessKeyValueCompaction(SubcompactionState* sub_compact) {
   if (!sub_compact->compaction->partial_compaction()) {
     dict_sample_data.reserve(kSampleBytes);
   }
+
+  // Represents how many records in target blob SST that are needed by the key
+  // SST
   std::unordered_map<uint64_t, uint64_t> dependence;
 
   size_t yield_count = 0;
@@ -2036,7 +2044,12 @@ void CompactionJob::ProcessKeyValueCompaction(SubcompactionState* sub_compact) {
   sub_compact->c_iter.reset();
   input.reset();
   sub_compact->status = status;
-}  // namespace TERARKDB_NAMESPACE
+}
+
+// TODO(guokuankuan@bytedance.com)
+void CompactionJob::ProcessLinkCompaction(SubcompactionState* sub_compact) {
+  return;
+}
 
 void CompactionJob::ProcessGarbageCollection(SubcompactionState* sub_compact) {
   assert(sub_compact != nullptr);

db/compaction_job.h

@@ -123,6 +123,7 @@ class CompactionJob {
   // kv-pairs
   void ProcessCompaction(SubcompactionState* sub_compact);
   void ProcessKeyValueCompaction(SubcompactionState* sub_compact);
+  void ProcessLinkCompaction(SubcompactionState* sub_compact);
   void ProcessGarbageCollection(SubcompactionState* sub_compact);
 
   Status FinishCompactionOutputFile(

db/compaction_picker.cc

@@ -994,6 +994,15 @@ Compaction* CompactionPicker::PickGarbageCollection(
   return c;
 }
 
+// TODO (guokuankuan@bytedance.com)
+Compaction* CompactionPicker::PickLinkCompaction(
+    const std::string& cf_name, const MutableCFOptions& mutable_cf_options,
+    VersionStorageInfo* vstorage,
+    const std::vector<SequenceNumber>& snapshots, LogBuffer* log_buffer) {
+  // Need to be implemented
+  return nullptr;
+}
+
 void CompactionPicker::InitFilesBeingCompact(
     const MutableCFOptions& mutable_cf_options, VersionStorageInfo* vstorage,
     const InternalKey* begin, const InternalKey* end,
@@ -2971,7 +2980,7 @@ bool LevelCompactionBuilder::PickFileToCompact() {
   // store where to start the iteration in the next call to PickCompaction
   vstorage_->SetNextCompactionIndex(start_level_, cmp_idx);
 
-  return start_level_inputs_.size() > 0;
+  return !start_level_inputs_.empty();
 }
 
 bool LevelCompactionBuilder::PickIntraL0Compaction() {

db/compaction_picker.h

@@ -106,6 +106,16 @@ class CompactionPicker {
       VersionStorageInfo* vstorage,
       const std::vector<SequenceNumber>& snapshots, LogBuffer* log_buffer) = 0;
 
+  // Gather target files (including blob SST(KV separated) and ordinary SST) and
+  // composite a compaction for future use.
+  // A Link Compaction contains two major objectives:
+  // (1) Link target SST files into a logical linked sst
+  // (2) GC old linked SST files
+  virtual Compaction* PickLinkCompaction(
+      const std::string& cf_name, const MutableCFOptions& mutable_cf_options,
+      VersionStorageInfo* vstorage,
+      const std::vector<SequenceNumber>& snapshots, LogBuffer* log_buffer) = 0;
+
   // Pick compaction which level has map or link sst
   Compaction* PickGarbageCollection(const std::string& cf_name,
                                     const MutableCFOptions& mutable_cf_options,

db/db_impl.h

@@ -1246,6 +1246,7 @@ class DBImpl : public DB {
                             const std::string& dir_to_sync, FileType type,
                             uint64_t number, int job_id);
   static void BGWorkCompaction(void* arg);
+  static void BGWorkLinkCompaction(void* arg);
   static void BGWorkGarbageCollection(void* arg);
   // Runs a pre-chosen universal compaction involving bottom level in a
   // separate, bottom-pri thread pool.
@@ -1255,12 +1256,18 @@ class DBImpl : public DB {
   static void UnscheduleCallback(void* arg);
   void BackgroundCallCompaction(PrepickedCompaction* prepicked_compaction,
                                 Env::Priority bg_thread_pri);
+  void BackgroundCallLinkCompaction(PrepickedCompaction* prepicked_compaction,
+                                    Env::Priority bg_thread_pri);
   void BackgroundCallGarbageCollection();
   void BackgroundCallFlush();
   void BackgroundCallPurge();
+
   Status BackgroundCompaction(bool* madeProgress, JobContext* job_context,
                               LogBuffer* log_buffer,
                               PrepickedCompaction* prepicked_compaction);
+  Status BackgroundLinkCompaction(bool* madeProgress, JobContext* job_context,
+                                  LogBuffer* log_buffer,
+                                  PrepickedCompaction* prepicked_compaction);
   Status BackgroundGarbageCollection(bool* madeProgress,
                                      JobContext* job_context,
                                      LogBuffer* log_buffer);

db/db_impl_compaction_flush.cc

@@ -1888,6 +1888,14 @@ void DBImpl::MaybeScheduleFlushOrCompaction() {
     env_->Schedule(&DBImpl::BGWorkCompaction, ca, Env::Priority::LOW, this,
                    &DBImpl::UnscheduleCallback);
   }
+
+  // TODO(guokuankuan@bytedance.com)
+  // We need to find out a reasonable opportunity to emit Link Compactions
+  while(false) {
+    CompactionArg* ca = new CompactionArg;
+    env_->Schedule(&DBImpl::BGWorkLinkCompaction, ca, Env::Priority::LOW, this,
+                   &DBImpl::UnscheduleCallback);
+  }
 }
 
 DBImpl::BGJobLimits DBImpl::GetBGJobLimits() const {
@@ -2072,6 +2080,18 @@ void DBImpl::BGWorkCompaction(void* arg) {
   delete prepicked_compaction;
 }
 
+void DBImpl::BGWorkLinkCompaction(void* arg) {
+  CompactionArg ca = *(reinterpret_cast<CompactionArg*>(arg));
+  delete reinterpret_cast<CompactionArg*>(arg);
+  IOSTATS_SET_THREAD_POOL_ID(Env::Priority::LOW);
+  TEST_SYNC_POINT("DBImpl::BGWorkLinkCompaction");
+  auto prepicked_compaction =
+      static_cast<PrepickedCompaction*>(ca.prepicked_compaction);
+  reinterpret_cast<DBImpl*>(ca.db)->BackgroundCallLinkCompaction(
+      prepicked_compaction, Env::Priority::LOW);
+  delete prepicked_compaction;
+}
+
 void DBImpl::BGWorkGarbageCollection(void* arg) {
   CompactionArg ca = *(reinterpret_cast<CompactionArg*>(arg));
   delete reinterpret_cast<CompactionArg*>(arg);
@@ -2408,6 +2428,14 @@ void DBImpl::BackgroundCallCompaction(PrepickedCompaction* prepicked_compaction,
   }
 }
 
+// TODO(guokuankuan@bytedance.com)
+void DBImpl::BackgroundCallLinkCompaction(PrepickedCompaction* prepicked_compaction,
+                                          Env::Priority bg_thread_pri) {
+//  Status s = BackgroundLinkCompaction(&made_progress, &job_context, &log_buffer,
+//                                      prepicked_compaction);
+  return;
+}
+
 void DBImpl::BackgroundCallGarbageCollection() {
   bool made_progress = false;
   JobContext job_context(next_job_id_.fetch_add(1), true);
@@ -2920,6 +2948,13 @@ Status DBImpl::BackgroundCompaction(bool* made_progress,
   return status;
 }
 
+Status DBImpl::BackgroundLinkCompaction(bool* made_progress,
+                                        JobContext* job_context,
+                                        LogBuffer* log_buffer,
+                                        PrepickedCompaction* prepicked_compaction) {
+  return Status::OK();
+}
+
 Status DBImpl::BackgroundGarbageCollection(bool* made_progress,
                                            JobContext* job_context,
                                            LogBuffer* log_buffer) {

db/dbformat.h

@@ -64,6 +64,9 @@ enum ValueType : unsigned char {
   // generated by WriteUnprepared write policy is not mistakenly read by
   // another.
   kTypeBeginUnprepareXID = 0x13,  // WAL only.
+  // Similar to kTypeValueIndex, this means current value belongs to
+  // a LinkSST and pointed to the actual value file.
+  kTypeLinkIndex = 0x14, // LinkSST only
   kMaxValue = 0x7F                // Not used for storing records.
 };
 
@@ -784,6 +787,12 @@ struct ParsedInternalKeyComparator {
   const InternalKeyComparator* cmp;
 };
 
+// This interface is used for transferring data format between KV separated and
+// combined kv pairs.
+// We should add an implementation instance while processing key value pairs
+// during compactions.
+// Note that, we also port `Version` to this interface since we may need to fetch
+// value from any version that has separated values.
 class SeparateHelper {
  public:
   virtual ~SeparateHelper() = default;

include/rocksdb/compaction_filter.h

@@ -200,6 +200,10 @@ class CompactionFilter
 
   // Determines whether value changed by compaction filter were stable.
   // Default as false, which means stability of outcome is not promised.
+  // "Stable" means the changed value will not change after the same
+  // operation is applied multiple times.
+  // Creators of the compaction filter should override this function, or
+  // the behavior of the stability checking is undefined.
   virtual bool IsStableChangeValue() const { return false; }
 
   // Returns a name that identifies this compaction filter.

util/iterator_cache.h

@@ -18,6 +18,7 @@ struct FileMetaData;
 class RangeDelAggregator;
 class TableReader;
 
+// <FileNumber -> FileMetaData>
 typedef chash_map<uint64_t, FileMetaData*> DependenceMap;
 
 class IteratorCache {