init

Author: joshkang97

db/compaction/compaction_job.cc

@@ -103,6 +103,8 @@ const char* GetCompactionReasonString(CompactionReason compaction_reason) {
       return "RoundRobinTtl";
     case CompactionReason::kRefitLevel:
       return "RefitLevel";
+    case CompactionReason::kReadTriggered:
+      return "ReadTriggered";
     case CompactionReason::kNumOfReasons:
       // fall through
     default:

db/compaction/compaction_picker_level.cc

@@ -36,6 +36,9 @@ bool LevelCompactionPicker::NeedsCompaction(
   if (!vstorage->FilesMarkedForForcedBlobGC().empty()) {
     return true;
   }
+  if (!vstorage->ReadTriggeredCompactionFiles().empty()) {
+    return true;
+  }
   for (int i = 0; i <= vstorage->MaxInputLevel(); i++) {
     if (vstorage->CompactionScore(i) >= 1) {
       return true;
@@ -325,6 +328,14 @@ void LevelCompactionBuilder::SetupInitialFiles() {
     compaction_reason_ = CompactionReason::kForcedBlobGC;
     return;
   }
+
+  // Read-triggered compaction
+  PickFileToCompact(vstorage_->ReadTriggeredCompactionFiles(),
+                    CompactToNextLevel::kYes);
+  if (!start_level_inputs_.empty()) {
+    compaction_reason_ = CompactionReason::kReadTriggered;
+    return;
+  }
 }
 
 bool LevelCompactionBuilder::SetupOtherL0FilesIfNeeded() {

db/compaction/compaction_picker_test.cc

@@ -982,6 +982,150 @@ TEST_F(CompactionPickerTest, UniversalIncrementalSpace1) {
   ASSERT_EQ(14U, compaction->input(1, 3)->fd.GetNumber());
 }
 
+TEST_F(CompactionPickerTest, ReadTriggeredCompactionDisabled) {
+  NewVersionStorage(6, kCompactionStyleLevel);
+  // threshold=0 means disabled, no files should be marked
+  mutable_cf_options_.read_triggered_compaction_threshold = 0.0;
+  Add(0, 1U, "100", "200", 1000U);
+  file_map_[1U].first->stats.num_reads_sampled.store(999999);
+  UpdateVersionStorageInfo();
+  ASSERT_TRUE(vstorage_->ReadTriggeredCompactionFiles().empty());
+}
+
+TEST_F(CompactionPickerTest, ReadTriggeredCompactionBelowThreshold) {
+  NewVersionStorage(6, kCompactionStyleLevel);
+  mutable_cf_options_.read_triggered_compaction_threshold = 1.0;
+  // file_size=1000, reads=500 => reads_per_byte=0.5 < 1.0
+  Add(0, 1U, "100", "200", 1000U);
+  file_map_[1U].first->stats.num_reads_sampled.store(500);
+  UpdateVersionStorageInfo();
+  ASSERT_TRUE(vstorage_->ReadTriggeredCompactionFiles().empty());
+}
+
+TEST_F(CompactionPickerTest, ReadTriggeredCompactionAboveThreshold) {
+  NewVersionStorage(6, kCompactionStyleLevel);
+  mutable_cf_options_.read_triggered_compaction_threshold = 0.5;
+  // file_size=1000, reads=600 => reads_per_byte=0.6 > 0.5
+  Add(0, 1U, "100", "200", 1000U);
+  file_map_[1U].first->stats.num_reads_sampled.store(600);
+  // file_size=1000, reads=300 => reads_per_byte=0.3 < 0.5 (not marked)
+  Add(1, 2U, "300", "400", 1000U);
+  file_map_[2U].first->stats.num_reads_sampled.store(300);
+  // file_size=1000, reads=800 => reads_per_byte=0.8 > 0.5 (hottest)
+  Add(2, 3U, "500", "600", 1000U);
+  file_map_[3U].first->stats.num_reads_sampled.store(800);
+  // Add a file at the bottom so L2 is not the last non-empty level
+  Add(4, 4U, "700", "800", 1000U);
+  UpdateVersionStorageInfo();
+
+  const auto& marked = vstorage_->ReadTriggeredCompactionFiles();
+  ASSERT_EQ(marked.size(), 2);
+  // Sorted by reads_per_byte descending: file 3 (0.8) then file 1 (0.6)
+  ASSERT_EQ(marked[0].second->fd.GetNumber(), 3U);
+  ASSERT_EQ(marked[1].second->fd.GetNumber(), 1U);
+}
+
+TEST_F(CompactionPickerTest, NeedsCompactionReadTriggered) {
+  NewVersionStorage(6, kCompactionStyleLevel);
+  mutable_cf_options_.read_triggered_compaction_threshold = 0.1;
+  Add(1, 1U, "100", "200", 1000U);
+  file_map_[1U].first->stats.num_reads_sampled.store(500);
+  Add(3, 2U, "300", "400", 1000U);
+  UpdateVersionStorageInfo();
+
+  ASSERT_FALSE(vstorage_->ReadTriggeredCompactionFiles().empty());
+  ASSERT_TRUE(level_compaction_picker.NeedsCompaction(vstorage_.get()));
+}
+
+TEST_F(CompactionPickerTest, ReadTriggeredPicksFile) {
+  NewVersionStorage(6, kCompactionStyleLevel);
+  mutable_cf_options_.read_triggered_compaction_threshold = 0.1;
+  Add(1, 1U, "100", "200", 1000U);
+  file_map_[1U].first->stats.num_reads_sampled.store(500);
+  Add(3, 2U, "300", "400", 1000U);
+  UpdateVersionStorageInfo();
+
+  std::unique_ptr<Compaction> compaction(level_compaction_picker.PickCompaction(
+      cf_name_, mutable_cf_options_, mutable_db_options_,
+      /*existing_snapshots=*/{}, /*snapshot_checker=*/nullptr, vstorage_.get(),
+      &log_buffer_, /*full_history_ts_low=*/""));
+  ASSERT_TRUE(compaction.get() != nullptr);
+  ASSERT_EQ(compaction->compaction_reason(), CompactionReason::kReadTriggered);
+}
+
+TEST_F(CompactionPickerTest, UniversalReadTriggeredCompaction) {
+  const uint64_t kFileSize = 100000;
+
+  mutable_cf_options_.read_triggered_compaction_threshold = 0.001;
+  // Set trigger high so size amp / sorted run pickers don't fire
+  mutable_cf_options_.level0_file_num_compaction_trigger = 10;
+  UniversalCompactionPicker universal_compaction_picker(ioptions_, &icmp_);
+
+  NewVersionStorage(5, kCompactionStyleUniversal);
+
+  // Hot file at L2 with data at L4 below it
+  Add(0, 1U, "150", "200", kFileSize, 0, 500, 550);
+  Add(2, 2U, "301", "350", kFileSize, 0, 201, 250);
+  Add(4, 3U, "301", "400", kFileSize, 0, 101, 150);
+
+  // Mark file 2 (L2) as having high reads
+  file_map_[2U].first->stats.num_reads_sampled.store(kFileSize);
+  UpdateVersionStorageInfo();
+
+  ASSERT_TRUE(universal_compaction_picker.NeedsCompaction(vstorage_.get()));
+
+  std::unique_ptr<Compaction> compaction(
+      universal_compaction_picker.PickCompaction(
+          cf_name_, mutable_cf_options_, mutable_db_options_,
+          /*existing_snapshots=*/{}, /*snapshot_checker=*/nullptr,
+          vstorage_.get(), &log_buffer_, /*full_history_ts_low=*/""));
+
+  ASSERT_TRUE(compaction);
+  ASSERT_EQ(compaction->compaction_reason(), CompactionReason::kReadTriggered);
+  ASSERT_EQ(compaction->start_level(), 2);
+  ASSERT_EQ(compaction->output_level(), 4);
+}
+
+TEST_F(CompactionPickerTest, ReadTriggeredSkipsLastLevel) {
+  const uint64_t kFileSize = 100000;
+
+  mutable_cf_options_.read_triggered_compaction_threshold = 0.001;
+  mutable_cf_options_.level0_file_num_compaction_trigger = 10;
+  UniversalCompactionPicker universal_compaction_picker(ioptions_, &icmp_);
+
+  NewVersionStorage(5, kCompactionStyleUniversal);
+
+  Add(0, 1U, "150", "200", kFileSize, 0, 500, 550);
+  Add(4, 3U, "301", "350", kFileSize, 0, 101, 150);
+
+  // File 3 is at the last non-empty level — should NOT be marked for
+  // read-triggered compaction. Bottommost file cleanup is handled
+  // separately by ComputeBottommostFilesMarkedForCompaction().
+  file_map_[3U].first->stats.num_reads_sampled.store(kFileSize);
+  UpdateVersionStorageInfo();
+
+  ASSERT_TRUE(vstorage_->ReadTriggeredCompactionFiles().empty());
+}
+
+TEST_F(CompactionPickerTest, UniversalReadTriggeredNoPickWhenNotMarked) {
+  const uint64_t kFileSize = 100000;
+
+  mutable_cf_options_.read_triggered_compaction_threshold = 0.001;
+  UniversalCompactionPicker universal_compaction_picker(ioptions_, &icmp_);
+
+  NewVersionStorage(5, kCompactionStyleUniversal);
+
+  Add(0, 1U, "150", "200", kFileSize, 0, 500, 550);
+  Add(4, 3U, "301", "350", kFileSize, 0, 101, 150);
+
+  // No reads on any file
+  UpdateVersionStorageInfo();
+
+  ASSERT_TRUE(vstorage_->ReadTriggeredCompactionFiles().empty());
+  // Not enough sorted runs to trigger compaction either
+  ASSERT_FALSE(universal_compaction_picker.NeedsCompaction(vstorage_.get()));
+}
+
 TEST_F(CompactionPickerTest, UniversalIncrementalSpace2) {
   const uint64_t kFileSize = 100000;
 

db/compaction/compaction_picker_universal.cc

@@ -275,6 +275,21 @@ class UniversalCompactionBuilder {
     return c;
   }
 
+  Compaction* MaybePickReadTriggeredCompaction(
+      Compaction* const prev_picked_c) {
+    if (prev_picked_c != nullptr ||
+        vstorage_->ReadTriggeredCompactionFiles().empty()) {
+      return prev_picked_c;
+    }
+    Compaction* c = PickReadTriggeredCompaction();
+    if (c != nullptr) {
+      ROCKS_LOG_BUFFER(log_buffer_,
+                       "[%s] Universal: picked for read triggered compaction\n",
+                       cf_name_.c_str());
+    }
+    return c;
+  }
+
   // Pick Universal compaction to limit read amplification
   Compaction* PickCompactionToReduceSortedRuns(
       unsigned int ratio, unsigned int max_number_of_files_to_compact);
@@ -292,6 +307,8 @@ class UniversalCompactionBuilder {
 
   Compaction* PickDeleteTriggeredCompaction();
 
+  Compaction* PickReadTriggeredCompaction();
+
   // Returns true if this given file (that is marked be compaction) should be
   // skipped from being picked for now. We do this to best use standalone range
   // tombstone files.
@@ -594,6 +611,9 @@ bool UniversalCompactionPicker::NeedsCompaction(
   if (!vstorage->FilesMarkedForCompaction().empty()) {
     return true;
   }
+  if (!vstorage->ReadTriggeredCompactionFiles().empty()) {
+    return true;
+  }
   return false;
 }
 
@@ -759,6 +779,7 @@ Compaction* UniversalCompactionBuilder::PickCompaction() {
   if (sorted_runs_.size() == 0 ||
       (vstorage_->FilesMarkedForPeriodicCompaction().empty() &&
        vstorage_->FilesMarkedForCompaction().empty() &&
+       vstorage_->ReadTriggeredCompactionFiles().empty() &&
        sorted_runs_.size() < (unsigned int)file_num_compaction_trigger)) {
     ROCKS_LOG_BUFFER(log_buffer_, "[%s] Universal: nothing to do\n",
                      cf_name_.c_str());
@@ -782,6 +803,7 @@ Compaction* UniversalCompactionBuilder::PickCompaction() {
   c = MaybePickCompactionToReduceSortedRuns(c, file_num_compaction_trigger,
                                             ratio);
   c = MaybePickDeleteTriggeredCompaction(c);
+  c = MaybePickReadTriggeredCompaction(c);
 
   if (c == nullptr) {
     TEST_SYNC_POINT_CALLBACK(
@@ -1782,6 +1804,112 @@ Compaction* UniversalCompactionBuilder::PickPeriodicCompaction() {
   return c;
 }
 
+Compaction* UniversalCompactionBuilder::PickReadTriggeredCompaction() {
+  ROCKS_LOG_BUFFER(log_buffer_, "[%s] Universal: Read Triggered Compaction",
+                   cf_name_.c_str());
+
+  // Directly pick the hot file and compact it with the overlapping next level,
+  // similar to PickDeleteTriggeredCompaction for multi-level universal.
+  CompactionInputFiles start_level_inputs;
+  int start_level = -1;
+  int output_level;
+  std::vector<CompactionInputFiles> inputs;
+  std::vector<FileMetaData*> grandparents;
+
+  // Find the first non-being-compacted read-triggered file.
+  for (const auto& level_file : vstorage_->ReadTriggeredCompactionFiles()) {
+    assert(!level_file.second->being_compacted);
+    start_level = level_file.first;
+
+    if (start_level == 0 &&
+        !picker_->level0_compactions_in_progress()->empty()) {
+      continue;
+    }
+
+    start_level_inputs.files = {level_file.second};
+    start_level_inputs.level = start_level;
+    if (picker_->ExpandInputsToCleanCut(cf_name_, vstorage_,
+                                        &start_level_inputs)) {
+      break;
+    }
+    start_level_inputs.files.clear();
+  }
+
+  if (start_level_inputs.empty()) {
+    return nullptr;
+  }
+
+  int max_output_level = vstorage_->MaxOutputLevel(allow_ingest_behind_);
+
+  // Pick the first non-empty level after start_level as output.
+  for (output_level = start_level + 1; output_level <= max_output_level;
+       output_level++) {
+    if (vstorage_->NumLevelFiles(output_level) != 0) {
+      break;
+    }
+  }
+
+  // Only non-last level files can be picked as inputs
+  assert(output_level > start_level);
+
+  if (!MeetsOutputLevelRequirements(output_level)) {
+    return nullptr;
+  }
+
+  if (start_level == 0) {
+    if (!picker_->GetOverlappingL0Files(vstorage_, &start_level_inputs,
+                                        output_level, nullptr)) {
+      return nullptr;
+    }
+  }
+
+  CompactionInputFiles output_level_inputs;
+  int parent_index = -1;
+
+  output_level_inputs.level = output_level;
+  if (!picker_->SetupOtherInputs(cf_name_, mutable_cf_options_, vstorage_,
+                                 &start_level_inputs, &output_level_inputs,
+                                 &parent_index, -1, false)) {
+    return nullptr;
+  }
+
+  inputs.push_back(start_level_inputs);
+  if (!output_level_inputs.empty()) {
+    inputs.push_back(output_level_inputs);
+  }
+
+  if (picker_->FilesRangeOverlapWithCompaction(
+          inputs, output_level,
+          Compaction::EvaluateProximalLevel(vstorage_, mutable_cf_options_,
+                                            ioptions_, start_level,
+                                            output_level))) {
+    return nullptr;
+  }
+
+  picker_->GetGrandparents(vstorage_, start_level_inputs, output_level_inputs,
+                           &grandparents);
+
+  uint64_t estimated_total_size = 0;
+  for (FileMetaData* f : vstorage_->LevelFiles(output_level)) {
+    estimated_total_size += f->fd.GetFileSize();
+  }
+  uint32_t path_id =
+      GetPathId(ioptions_, mutable_cf_options_, estimated_total_size);
+  return new Compaction(
+      vstorage_, ioptions_, mutable_cf_options_, mutable_db_options_,
+      std::move(inputs), output_level,
+      MaxFileSizeForLevel(mutable_cf_options_, output_level,
+                          kCompactionStyleUniversal),
+      GetMaxOverlappingBytes(), path_id,
+      GetCompressionType(vstorage_, mutable_cf_options_, output_level, 1),
+      GetCompressionOptions(mutable_cf_options_, vstorage_, output_level),
+      Temperature::kUnknown,
+      /* max_subcompactions */ 0, grandparents, earliest_snapshot_,
+      snapshot_checker_, CompactionReason::kReadTriggered,
+      /* trim_ts */ "", score_,
+      /* l0_files_might_overlap */ true);
+}
+
 uint64_t UniversalCompactionBuilder::GetMaxOverlappingBytes() const {
   if (!mutable_cf_options_.compaction_options_universal.incremental) {
     return std::numeric_limits<uint64_t>::max();