Refactor marked_for_compaction in order to improve management (#142)

mm304321141 · mm304321141 · commit c7da45504ca0 · 2021-10-14T20:38:07.000+08:00
diff --git a/db/builder.cc b/db/builder.cc
@@ -209,7 +209,6 @@ Status BuildTable(
       blob_meta->prop.purpose = kEssenceSst;
       blob_meta->prop.flags |= TablePropertyCache::kNoRangeDeletions;
       status = blob_builder->Finish(&blob_meta->prop, nullptr);
-      blob_meta->marked_for_compaction = blob_builder->NeedCompact();
       TableProperties& tp = *separate_helper.current_prop;
       if (status.ok()) {
         blob_meta->fd.file_size = blob_builder->FileSize();
@@ -422,7 +421,8 @@ Status BuildTable(
     if (s.ok() && !empty) {
       uint64_t file_size = builder->FileSize();
       sst_meta()->fd.file_size = file_size;
-      sst_meta()->marked_for_compaction = builder->NeedCompact();
+      sst_meta()->marked_for_compaction =
+          builder->NeedCompact() ? FileMetaData::kMarkedFromTableBuilder : 0;
       sst_meta()->prop.num_entries = builder->NumEntries();
       assert(sst_meta()->fd.GetFileSize() > 0);
       // refresh now that builder is finished
diff --git a/db/compaction.cc b/db/compaction.cc
@@ -196,7 +196,7 @@ void ProcessFileMetaData(const char* job_info, FileMetaData* meta,
 
   if (tp->num_range_deletions > 0 && mopt->optimize_range_deletion &&
       !iopt->enable_lazy_compaction) {
-    meta->marked_for_compaction = true;
+    meta->marked_for_compaction |= FileMetaData::kMarkedFromRangeDeletion;
   }
   if (iopt->ttl_extractor_factory != nullptr) {
     GetCompactionTimePoint(tp->user_collected_properties,
diff --git a/db/compaction.h b/db/compaction.h
@@ -181,7 +181,7 @@ struct CompactionWorkerResult {
 
     // use UserProperties["User.Collected.Transient.Stat"] to reduce complexity
     // std::string stat_one;
-    bool marked_for_compaction;
+    uint8_t marked_for_compaction;
   };
   std::vector<FileInfo> files;
   std::string stat_all;
diff --git a/db/compaction_dispatcher.cc b/db/compaction_dispatcher.cc
@@ -931,7 +931,8 @@ std::string RemoteCompactionDispatcher::Worker::DoCompaction(Slice data) {
       }
     }
     if (s.ok()) {
-      meta.marked_for_compaction = builder->NeedCompact();
+      meta.marked_for_compaction =
+          builder->NeedCompact() ? FileMetaData::kMarkedFromTableBuilder : 0;
       meta.prop.num_entries = builder->NumEntries();
       for (auto& pair : dependence) {
         meta.prop.dependence.emplace_back(Dependence{pair.first, pair.second});
@@ -976,7 +977,7 @@ std::string RemoteCompactionDispatcher::Worker::DoCompaction(Slice data) {
       file_info.smallest_seqno = meta.fd.smallest_seqno;
       file_info.largest_seqno = meta.fd.largest_seqno;
       file_info.file_size = meta.fd.file_size;
-      file_info.marked_for_compaction |= builder->NeedCompact();
+      file_info.marked_for_compaction = meta.marked_for_compaction;
       result.files.emplace_back(file_info);
     }
     meta = FileMetaData();
diff --git a/db/compaction_job.cc b/db/compaction_job.cc
@@ -108,8 +108,18 @@ const char* GetCompactionReasonString(CompactionReason compaction_reason) {
       return "FIFOTtl";
     case CompactionReason::kManualCompaction:
       return "ManualCompaction";
-    case CompactionReason::kFilesMarkedForCompaction:
-      return "FilesMarkedForCompaction";
+    case CompactionReason::kFilesMarkedFromUser:
+      return "FilesMarkedFromUser";
+    case CompactionReason::kFilesMarkedFromTableBuilder:
+      return "FilesMarkedFromTableBuilder";
+    case CompactionReason::kFilesMarkedFromRangeDeletion:
+      return "FilesMarkedFromRangeDeletion";
+    case CompactionReason::kFilesMarkedFromTTL:
+      return "FilesMarkedFromTTL";
+    case CompactionReason::kFilesMarkedFromFileSystem:
+      return "FilesMarkedFromFileSystem";
+    case CompactionReason::kFilesMarkedFromUpdateBlob:
+      return "FilesMarkedFromUpdateBlob";
     case CompactionReason::kBottommostFiles:
       return "BottommostFiles";
     case CompactionReason::kFlush:
@@ -2428,7 +2438,9 @@ Status CompactionJob::FinishCompactionOutputFile(
     }
   }
   if (s.ok()) {
-    meta->marked_for_compaction = sub_compact->builder->NeedCompact();
+    meta->marked_for_compaction = sub_compact->builder->NeedCompact()
+                                      ? FileMetaData::kMarkedFromTableBuilder
+                                      : 0;
     meta->prop.num_entries = sub_compact->builder->NumEntries();
     for (auto& pair : dependence) {
       meta->prop.dependence.emplace_back(Dependence{pair.first, pair.second});
@@ -2558,7 +2570,6 @@ Status CompactionJob::FinishCompactionOutputBlob(
   auto meta = &sub_compact->current_blob_output()->meta;
   assert(meta != nullptr);
   if (s.ok()) {
-    meta->marked_for_compaction = sub_compact->blob_builder->NeedCompact();
     meta->prop.num_entries = sub_compact->blob_builder->NumEntries();
     meta->prop.inheritance = InheritanceTreeToSet(inheritance_tree);
     assert(std::is_sorted(meta->prop.inheritance.begin(),
diff --git a/db/compaction_picker.cc b/db/compaction_picker.cc
@@ -156,6 +156,7 @@ void AssignUserKey(std::string& key, const Slice& ikey) {
   Slice ukey = ExtractUserKey(ikey);
   key.assign(ukey.data(), ukey.size());
 };
+
 }  // anonymous namespace
 
 bool FindIntraL0Compaction(const std::vector<FileMetaData*>& level_files,
@@ -259,6 +260,29 @@ CompactionPicker::CompactionPicker(TableCache* table_cache,
 
 CompactionPicker::~CompactionPicker() {}
 
+CompactionReason CompactionPicker::ConvertCompactionReason(
+    uint8_t marked, CompactionReason default_reason) {
+  if (marked & FileMetaData::kMarkedFromUser) {
+    return CompactionReason::kFilesMarkedFromUser;
+  }
+  if (marked & FileMetaData::kMarkedFromTableBuilder) {
+    return CompactionReason::kFilesMarkedFromTableBuilder;
+  }
+  if (marked & FileMetaData::kMarkedFromRangeDeletion) {
+    return CompactionReason::kFilesMarkedFromRangeDeletion;
+  }
+  if (marked & FileMetaData::kMarkedFromTTL) {
+    return CompactionReason::kFilesMarkedFromTTL;
+  }
+  if (marked & FileMetaData::kMarkedFromFileSystem) {
+    return CompactionReason::kFilesMarkedFromFileSystem;
+  }
+  if (marked & FileMetaData::kMarkedFromUpdateBlob) {
+    return CompactionReason::kFilesMarkedFromUpdateBlob;
+  }
+  return default_reason;
+}
+
 double CompactionPicker::GetQ(std::vector<double>::const_iterator b,
                               std::vector<double>::const_iterator e, size_t g) {
   double S = std::accumulate(b, e, 0.0);
@@ -2030,8 +2054,12 @@ void CompactionPicker::PickFilesMarkedForCompaction(
     // files as being_compacted, but didn't call ComputeCompactionScore()
     assert(!level_file.second->being_compacted);
     *start_level = level_file.first;
-    *output_level =
-        (*start_level == 0) ? vstorage->base_level() : *start_level + 1;
+    if (level_file.second->is_output_to_parent_level()) {
+      *output_level =
+          (*start_level == 0) ? vstorage->base_level() : *start_level + 1;
+    } else {
+      *output_level = level_file.first;
+    }
 
     if (*start_level == 0 && !level0_compactions_in_progress()->empty()) {
       return false;
@@ -2229,7 +2257,9 @@ void LevelCompactionBuilder::SetupInitialFiles() {
         cf_name_, vstorage_, &start_level_, &output_level_,
         &start_level_inputs_);
     if (!start_level_inputs_.empty()) {
-      compaction_reason_ = CompactionReason::kFilesMarkedForCompaction;
+      compaction_reason_ = CompactionPicker::ConvertCompactionReason(
+          start_level_inputs_.files.front()->marked_for_compaction,
+          CompactionReason::kUnknown);
       return;
     }
 
@@ -2252,7 +2282,9 @@ void LevelCompactionBuilder::SetupInitialFiles() {
       start_level_inputs_.clear();
     } else {
       assert(!start_level_inputs_.empty());
-      compaction_reason_ = CompactionReason::kBottommostFiles;
+      compaction_reason_ = CompactionPicker::ConvertCompactionReason(
+          start_level_inputs_.files.front()->marked_for_compaction,
+          CompactionReason::kBottommostFiles);
       return;
     }
 
diff --git a/db/compaction_picker.h b/db/compaction_picker.h
@@ -80,6 +80,9 @@ class CompactionPicker {
     bool skip_composite;
   };
 
+  static CompactionReason ConvertCompactionReason(
+      uint8_t marked, CompactionReason default_reason);
+
   static double GetQ(std::vector<double>::const_iterator b,
                      std::vector<double>::const_iterator e, size_t g);
 
diff --git a/db/compaction_picker_universal.cc b/db/compaction_picker_universal.cc
@@ -1097,6 +1097,7 @@ Compaction* UniversalCompactionPicker::PickDeleteTriggeredCompaction(
   CompactionInputFiles start_level_inputs;
   int output_level;
   std::vector<CompactionInputFiles> inputs;
+  uint8_t marked = 0;
 
   if (vstorage->num_levels() == 1) {
     // This is single level universal. Since we're basically trying to reclaim
@@ -1111,6 +1112,7 @@ Compaction* UniversalCompactionPicker::PickDeleteTriggeredCompaction(
     for (FileMetaData* f : vstorage->LevelFiles(0)) {
       if (f->marked_for_compaction && !f->being_compacted) {
         compact = true;
+        marked = f->marked_for_compaction;
       }
       if (compact) {
         start_level_inputs.files.push_back(f);
@@ -1132,6 +1134,7 @@ Compaction* UniversalCompactionPicker::PickDeleteTriggeredCompaction(
     if (start_level_inputs.empty()) {
       return nullptr;
     }
+    marked = start_level_inputs.files.front()->marked_for_compaction;
 
     // Pick the first non-empty level after the start_level
     for (output_level = start_level + 1; output_level < vstorage->num_levels();
@@ -1216,7 +1219,8 @@ Compaction* UniversalCompactionPicker::PickDeleteTriggeredCompaction(
   params.manual_compaction = true;
   params.score = score;
   params.compaction_type = compaction_type;
-  params.compaction_reason = CompactionReason::kFilesMarkedForCompaction;
+  params.compaction_reason =
+      ConvertCompactionReason(marked, CompactionReason::kUnknown);
 
   return new Compaction(std::move(params));
 }
diff --git a/db/db_impl.cc b/db/db_impl.cc
@@ -945,15 +945,18 @@ void DBImpl::ScheduleTtlGC() {
           continue;
         }
         ++total_count;
-        old_mark_count += meta->marked_for_compaction;
+        bool marked =
+            !!(meta->marked_for_compaction & FileMetaData::kMarkedFromTTL);
+        old_mark_count += marked;
         TEST_SYNC_POINT("DBImpl:Exist-SST");
-        if (!meta->marked_for_compaction &&
+        if (!marked &&
             should_marked_for_compacted(
                 l, meta->fd.GetNumber(), meta->prop.earliest_time_begin_compact,
                 meta->prop.latest_time_end_compact, now)) {
-          meta->marked_for_compaction = true;
+          meta->marked_for_compaction |= FileMetaData::kMarkedFromTTL;
+          marked = true;
         }
-        if (meta->marked_for_compaction) {
+        if (marked) {
           new_mark_count++;
           TEST_SYNC_POINT("DBImpl:ScheduleTtlGC-mark");
         }
@@ -1104,13 +1107,15 @@ void DBImpl::ScheduleZNSGC() {
           continue;
         }
         ++total_count;
-        old_mark_count += meta->marked_for_compaction;
+        bool marked = !!(meta->marked_for_compaction &
+                         FileMetaData::kMarkedFromFileSystem);
+        old_mark_count += marked;
         TEST_SYNC_POINT("DBImpl:Exist-SST");
-        if (!meta->marked_for_compaction &&
-            mark_for_gc.count(meta->fd.GetNumber()) > 0) {
-          meta->marked_for_compaction = true;
+        if (!marked && mark_for_gc.count(meta->fd.GetNumber()) > 0) {
+          meta->marked_for_compaction |= FileMetaData::kMarkedFromFileSystem;
+          marked = true;
         }
-        if (meta->marked_for_compaction) {
+        if (marked) {
           new_mark_count++;
           TEST_SYNC_POINT("DBImpl:ScheduleZNSGC-mark");
         }
diff --git a/db/db_impl_experimental.cc b/db/db_impl_experimental.cc
@@ -47,7 +47,7 @@ Status DBImpl::SuggestCompactRange(ColumnFamilyHandle* column_family,
           level, begin == nullptr ? nullptr : &start_key,
           end == nullptr ? nullptr : &end_key, &inputs);
       for (auto f : inputs) {
-        f->marked_for_compaction = true;
+        f->marked_for_compaction |= FileMetaData::kMarkedFromUser;
       }
     }
     // Since we have some more files to compact, we should also recompute
@@ -68,7 +68,7 @@ Status DBImpl::SuggestCompactColumnFamily(ColumnFamilyHandle* column_family) {
     auto vstorage = cfd->current()->storage_info();
     for (int level = -1; level < vstorage->num_non_empty_levels(); ++level) {
       for (auto f : vstorage->LevelFiles(level)) {
-        f->marked_for_compaction = true;
+        f->marked_for_compaction |= FileMetaData::kMarkedFromUser;
       }
     }
     // Since we have some more files to compact, we should also recompute
diff --git a/db/db_test.cc b/db/db_test.cc
@@ -4871,7 +4871,7 @@ TEST_F(DBTest, SuggestCompactRangeTest) {
       "Compaction::Compaction::Start", [&](void* arg) {
         Compaction* c = (Compaction*)arg;
         ASSERT_EQ(c->compaction_reason(),
-                  CompactionReason::kFilesMarkedForCompaction);
+                  CompactionReason::kFilesMarkedFromUser);
         ASSERT_FALSE(c->is_manual_compaction());
       });
   // disable trival move make current test happy
diff --git a/db/map_builder.cc b/db/map_builder.cc
@@ -135,7 +135,7 @@ struct RangeWithDepend {
     include[0] = true;
     include[1] = true;
     has_delete_range = false;
-    marked_for_compaction = f->marked_for_compaction;
+    marked_for_compaction = !!f->marked_for_compaction;
     stable = false;
     dependence.emplace_back(MapSstElement::LinkTarget{f->fd.GetNumber(), 0});
   }
diff --git a/db/version_builder.cc b/db/version_builder.cc
@@ -391,6 +391,7 @@ class VersionBuilder::Rep {
                   item.f = f;
                 }
                 item.f->gc_status = FileMetaData::kGarbageCollectionPermitted;
+                item.f->marked_for_compaction &= FileMetaData::kMarkedFromUser;
               }
               break;
             case FileMetaData::kGarbageCollectionCandidate:
@@ -421,7 +422,8 @@ class VersionBuilder::Rep {
         old_file_queue.pop();
       }
       while (!old_file_queue.empty()) {
-        dependence_map[old_file_queue.top()].f->marked_for_compaction = true;
+        dependence_map[old_file_queue.top()].f->marked_for_compaction |=
+            FileMetaData::kMarkedFromUpdateBlob;
         old_file_queue.pop();
       }
     }
diff --git a/db/version_edit.cc b/db/version_edit.cc
@@ -17,6 +17,28 @@
 #include "util/string_util.h"
 #include "util/sync_point.h"
 
+namespace {
+uint8_t reverse_byte(uint8_t v) {
+  // normal
+  // return static_cast<uint8_t>(((v & (1 << 0)) << 7) | ((v & (1 << 1)) << 6) |
+  //                             ((v & (1 << 2)) << 5) | ((v & (1 << 3)) << 4) |
+  //                             ((v & (1 << 4)) << 3) | ((v & (1 << 5)) << 2) |
+  //                             ((v & (1 << 6)) << 1) | ((v & (1 << 7)) << 0));
+
+  // hack
+  // return ((v * 0x0802LU & 0x22110LU) | (v * 0x8020LU & 0x88440LU)) *
+  //            0x10101LU >>
+  //        16;
+
+  // lookup table
+  constexpr static uint8_t lookup[16] = {
+      0x0, 0x8, 0x4, 0xc, 0x2, 0xa, 0x6, 0xe,
+      0x1, 0x9, 0x5, 0xd, 0x3, 0xb, 0x7, 0xf,
+  };
+  return (lookup[v & 0b1111] << 4) | lookup[v >> 4];
+}
+}  // namespace
+
 namespace TERARKDB_NAMESPACE {
 
 // Tag numbers for serialized VersionEdit.  These numbers are written to
@@ -174,7 +196,7 @@ bool VersionEdit::EncodeTo(std::string* dst) const {
     }
     if (f.marked_for_compaction) {
       PutVarint32(dst, CustomTag::kNeedCompaction);
-      char p = static_cast<char>(1);
+      char p = static_cast<char>(reverse_byte(f.marked_for_compaction));
       PutLengthPrefixedSlice(dst, Slice(&p, 1));
     }
     if (has_min_log_number_to_keep_ && !min_log_num_written) {
@@ -305,7 +327,8 @@ const char* VersionEdit::DecodeNewFile4From(Slice* input) {
           if (field.size() != 1) {
             return "need_compaction field wrong size";
           }
-          f.marked_for_compaction = (field[0] == 1);
+          f.marked_for_compaction =
+              reverse_byte(static_cast<uint8_t>(field[0]));
           break;
         case kMinLogNumberToKeepHack:
           // This is a hack to encode kMinLogNumberToKeep in a
diff --git a/db/version_edit.h b/db/version_edit.h
diff --git a/db/version_set.cc b/db/version_set.cc
diff --git a/include/rocksdb/listener.h b/include/rocksdb/listener.h

Original file line number	Diff line number	Diff line change
`@@ -196,7 +196,7 @@ void ProcessFileMetaData(const char* job_info, FileMetaData* meta,`
`196`	`196`
`197`	`197`	`if (tp->num_range_deletions > 0 && mopt->optimize_range_deletion &&`
`198`	`198`	`!iopt->enable_lazy_compaction) {`
`199`		`- meta->marked_for_compaction = true;`
	`199`	`+ meta->marked_for_compaction \|= FileMetaData::kMarkedFromRangeDeletion;`
`200`	`200`	`}`
`201`	`201`	`if (iopt->ttl_extractor_factory != nullptr) {`
`202`	`202`	`GetCompactionTimePoint(tp->user_collected_properties,`
Original file line number	Diff line number	Diff line change
`@@ -931,7 +931,8 @@ std::string RemoteCompactionDispatcher::Worker::DoCompaction(Slice data) {`
`931`	`931`	`}`
`932`	`932`	`}`
`933`	`933`	`if (s.ok()) {`
`934`		`- meta.marked_for_compaction = builder->NeedCompact();`
	`934`	`+ meta.marked_for_compaction =`
	`935`	`+ builder->NeedCompact() ? FileMetaData::kMarkedFromTableBuilder : 0;`
`935`	`936`	`meta.prop.num_entries = builder->NumEntries();`
`936`	`937`	`for (auto& pair : dependence) {`
`937`	`938`	`meta.prop.dependence.emplace_back(Dependence{pair.first, pair.second});`
`@@ -976,7 +977,7 @@ std::string RemoteCompactionDispatcher::Worker::DoCompaction(Slice data) {`
`976`	`977`	`file_info.smallest_seqno = meta.fd.smallest_seqno;`
`977`	`978`	`file_info.largest_seqno = meta.fd.largest_seqno;`
`978`	`979`	`file_info.file_size = meta.fd.file_size;`
`979`		`- file_info.marked_for_compaction \|= builder->NeedCompact();`
	`980`	`+ file_info.marked_for_compaction = meta.marked_for_compaction;`
`980`	`981`	`result.files.emplace_back(file_info);`
`981`	`982`	`}`
`982`	`983`	`meta = FileMetaData();`
Original file line number	Diff line number	Diff line change
`@@ -47,7 +47,7 @@ Status DBImpl::SuggestCompactRange(ColumnFamilyHandle* column_family,`
`47`	`47`	`level, begin == nullptr ? nullptr : &start_key,`
`48`	`48`	`end == nullptr ? nullptr : &end_key, &inputs);`
`49`	`49`	`for (auto f : inputs) {`
`50`		`- f->marked_for_compaction = true;`
	`50`	`+ f->marked_for_compaction \|= FileMetaData::kMarkedFromUser;`
`51`	`51`	`}`
`52`	`52`	`}`
`53`	`53`	`// Since we have some more files to compact, we should also recompute`
`@@ -68,7 +68,7 @@ Status DBImpl::SuggestCompactColumnFamily(ColumnFamilyHandle* column_family) {`
`68`	`68`	`auto vstorage = cfd->current()->storage_info();`
`69`	`69`	`for (int level = -1; level < vstorage->num_non_empty_levels(); ++level) {`
`70`	`70`	`for (auto f : vstorage->LevelFiles(level)) {`
`71`		`- f->marked_for_compaction = true;`
	`71`	`+ f->marked_for_compaction \|= FileMetaData::kMarkedFromUser;`
`72`	`72`	`}`
`73`	`73`	`}`
`74`	`74`	`// Since we have some more files to compact, we should also recompute`