Read-triggered compactions (#14426)

Summary:
Add read-triggered compaction, a new feature that reduces read amplification by compacting SST files that receive high read traffic. When an SST file's read frequency (`num_reads_sampled / file_size`) exceeds a configurable threshold, it is marked for compaction to a lower level.

The feature introduces two new options: a CF option `read_triggered_compaction_threshold` (default 0, disabled) and a DB option `max_periodic_compaction_trigger_seconds` (default 43200s) that controls how often the background thread re-evaluates compaction scores on quiet databases. Both options are dynamically changeable.

Lowering `max_periodic_compaction_trigger_seconds` does add some overhead, but generally is minimal, so running this every couple of minutes in a production environment seems fairly reasonable.

## Key changes

- **New CF option `read_triggered_compaction_threshold`** (`advanced_options.h`): When positive, files with `reads_per_byte > threshold` are marked for compaction. Files at the last non-empty level are skipped (bottommost compaction handles those separately). Marked files are sorted by hotness (reads_per_byte descending).
- **New DB option `max_periodic_compaction_trigger_seconds`** (`options.h`): Replaces the hardcoded 12-hour ceiling in `ComputeTriggerCompactionPeriod()`. Essential for read-triggered compaction on quiet DBs since there are no writes to trigger score re-evaluation.
- **Leveled compaction picker** (`compaction_picker_level.cc`): Adds read-triggered as the lowest-priority compaction reason in `SetupInitialFiles()`, using the existing `PickFileToCompact` helper.
- **Universal compaction picker** (`compaction_picker_universal.cc`): Adds `PickReadTriggeredCompaction` as lowest priority. Refactors shared "find output level + compute overlapping inputs + create Compaction" logic from both `PickDeleteTriggeredCompaction` and `PickReadTriggeredCompaction` into `BuildCompactionToNextLevel`, handling both single-level and multi-level universal cases.
- **Periodic trigger integration** (`db_impl.cc`): `TriggerPeriodicCompaction` now also fires for CFs with `read_triggered_compaction_threshold > 0`, even without time-based compaction configured.
- **Stress test & db_bench support**: Both `db_stress` and `db_bench` support the new options. `db_crashtest.py` randomly enables read-triggered compaction and sets a short periodic trigger interval when enabled.

Pull Request resolved: https://github.com/facebook/rocksdb/pull/14426

Test Plan:
**Unit tests**:
- `compaction_picker_test` — 7 new tests: `ReadTriggeredCompactionDisabled`, `ReadTriggeredCompactionBelowThreshold`, `ReadTriggeredCompactionAboveThreshold`, `NeedsCompactionReadTriggered`, `ReadTriggeredPicksFile`, `UniversalReadTriggeredCompaction`, `ReadTriggeredSkipsLastLevel`, `UniversalReadTriggeredNoPickWhenNotMarked`
- `db_compaction_test` — `ReadTriggeredCompaction` integration test verifying end-to-end behavior with sync points
- Stress test coverage

**Stress test**:
```
make V=1 -j "CRASH_TEST_EXT_ARGS=--duration=600 --max_key=2500000 --max_compaction_trigger_wakeup_seconds=10
  --read_triggered_compaction_threshold=0.0001 --interval=600" blackbox_crash_test
```
- confirmed read triggered compactions from LOGS

**Benchmark** (`db_bench`):

Setup: 5M keys (100B values, 16B keys), leveled compaction, 5 levels, 4MB target file size. DB fully compacted, then 2M overlapping keys written without compaction to create L0/L1 overlap (82 files, ~294MB).

LSM shape change during readrandom with read-triggered compaction:
```
BEFORE: L0=9 files (15MB), L1=4 (16MB), L2=20 (69MB), L3=49 (194MB) — 82 files, 294MB
AFTER:  L3=66 files (223MB)
```

| Benchmark | Config | avg ops/s | % change |
|-----------|--------|-----------|----------|
| readrandom (8 threads, 5M reads) | baseline (threshold=0) | 1,086,965 | — |
| readrandom (8 threads, 5M reads) | threshold=0.000001, trigger=5s | 1,453,697 | **+33.7%** |

Reviewed By: xingbowang

Differential Revision: D97838716

Pulled By: joshkang97

fbshipit-source-id: a21fcb270c7fadd4f78d98b9c821982f220dd3f0

db/c.cc

@@ -4747,6 +4747,26 @@ double rocksdb_options_get_blob_gc_force_threshold(rocksdb_options_t* opt) {
   return opt->rep.blob_garbage_collection_force_threshold;
 }
 
+void rocksdb_options_set_read_triggered_compaction_threshold(
+    rocksdb_options_t* opt, double val) {
+  opt->rep.read_triggered_compaction_threshold = val;
+}
+
+double rocksdb_options_get_read_triggered_compaction_threshold(
+    rocksdb_options_t* opt) {
+  return opt->rep.read_triggered_compaction_threshold;
+}
+
+void rocksdb_options_set_max_compaction_trigger_wakeup_seconds(
+    rocksdb_options_t* opt, uint64_t val) {
+  opt->rep.max_compaction_trigger_wakeup_seconds = val;
+}
+
+uint64_t rocksdb_options_get_max_compaction_trigger_wakeup_seconds(
+    rocksdb_options_t* opt) {
+  return opt->rep.max_compaction_trigger_wakeup_seconds;
+}
+
 void rocksdb_options_set_blob_compaction_readahead_size(rocksdb_options_t* opt,
                                                         uint64_t val) {
   opt->rep.blob_compaction_readahead_size = val;

db/c_test.c

@@ -2867,6 +2867,14 @@ int main(int argc, char** argv) {
     rocksdb_options_set_blob_gc_force_threshold(o, 0.75);
     CheckCondition(0.75 == rocksdb_options_get_blob_gc_force_threshold(o));
 
+    rocksdb_options_set_read_triggered_compaction_threshold(o, 0.001);
+    CheckCondition(0.001 ==
+                   rocksdb_options_get_read_triggered_compaction_threshold(o));
+
+    rocksdb_options_set_max_compaction_trigger_wakeup_seconds(o, 3600);
+    CheckCondition(
+        3600 == rocksdb_options_get_max_compaction_trigger_wakeup_seconds(o));
+
     rocksdb_options_set_blob_compaction_readahead_size(o, 262144);
     CheckCondition(262144 ==
                    rocksdb_options_get_blob_compaction_readahead_size(o));

db/column_family.cc

@@ -11,6 +11,7 @@
 
 #include <algorithm>
 #include <cinttypes>
+#include <cmath>
 #include <limits>
 #include <sstream>
 #include <string>
@@ -1563,6 +1564,14 @@ Status ColumnFamilyData::ValidateOptions(
     }
   }
 
+  if (cf_options.read_triggered_compaction_threshold < 0.0 ||
+      std::isnan(cf_options.read_triggered_compaction_threshold) ||
+      std::isinf(cf_options.read_triggered_compaction_threshold)) {
+    return Status::InvalidArgument(
+        "read_triggered_compaction_threshold must be >= 0.0 and finite. "
+        "Use 0.0 to disable.");
+  }
+
   if (cf_options.compaction_style == kCompactionStyleFIFO &&
       db_options.max_open_files != -1 && cf_options.ttl > 0) {
     return Status::NotSupported(

db/compaction/compaction_job.cc

@@ -104,6 +104,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

@@ -984,6 +984,186 @@ 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_collapsible_entry_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_collapsible_entry_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_collapsible_entry_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_collapsible_entry_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_collapsible_entry_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_collapsible_entry_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_collapsible_entry_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_collapsible_entry_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_collapsible_entry_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, UniversalReadTriggeredIntraL0) {
+  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_);
+
+  // Single-level universal with overlapping L0 files
+  NewVersionStorage(1, kCompactionStyleUniversal);
+
+  // Two L0 files with overlapping key ranges
+  Add(0, 1U, "100", "300", kFileSize, 0, 500, 550);
+  Add(0, 2U, "200", "400", kFileSize, 0, 400, 450);
+
+  // Mark file 1 as hot
+  file_map_[1U].first->stats.num_collapsible_entry_reads_sampled.store(
+      kFileSize);
+  UpdateVersionStorageInfo();
+
+  ASSERT_FALSE(vstorage_->ReadTriggeredCompactionFiles().empty());
+  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(), 0);
+  ASSERT_EQ(compaction->output_level(), 0);
+}
+
 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,14 @@ class UniversalCompactionBuilder {
 
   Compaction* PickDeleteTriggeredCompaction();
 
+  Compaction* PickReadTriggeredCompaction();
+
+  // Given already-selected start_level_inputs, find the first non-empty output
+  // level, compute overlapping inputs, and create a Compaction. Can produce
+  // intra-L0 compaction if there is only level 0.
+  Compaction* BuildCompactionToNextLevel(
+      CompactionInputFiles& start_level_inputs, CompactionReason reason);
+
   // 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 +617,9 @@ bool UniversalCompactionPicker::NeedsCompaction(
   if (!vstorage->FilesMarkedForCompaction().empty()) {
     return true;
   }
+  if (!vstorage->ReadTriggeredCompactionFiles().empty()) {
+    return true;
+  }
   return false;
 }
 
@@ -759,6 +785,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 +809,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(
@@ -1461,9 +1489,6 @@ Compaction* UniversalCompactionBuilder::PickIncrementalForReduceSizeAmp(
 // CompactOnDeleteCollector due to the presence of tombstones.
 Compaction* UniversalCompactionBuilder::PickDeleteTriggeredCompaction() {
   CompactionInputFiles start_level_inputs;
-  int output_level;
-  std::vector<CompactionInputFiles> inputs;
-  std::vector<FileMetaData*> grandparents;
 
   if (vstorage_->num_levels() == 1) {
     // This is single level universal. Since we're basically trying to reclaim
@@ -1474,7 +1499,6 @@ Compaction* UniversalCompactionBuilder::PickDeleteTriggeredCompaction() {
 
     start_level_inputs.level = 0;
     start_level_inputs.files.clear();
-    output_level = 0;
     // Find the first file marked for compaction. Ignore the last file
     for (size_t loop = 0; loop + 1 < sorted_runs_.size(); loop++) {
       SortedRun* sr = &sorted_runs_[loop];
@@ -1503,13 +1527,9 @@ Compaction* UniversalCompactionBuilder::PickDeleteTriggeredCompaction() {
       FileMetaData* f = vstorage_->LevelFiles(0)[loop];
       start_level_inputs.files.push_back(f);
     }
-    if (start_level_inputs.size() <= 1) {
-      // If only the last file in L0 is marked for compaction, ignore it
-      return nullptr;
-    }
-    inputs.push_back(start_level_inputs);
   } else {
     int start_level;
+    int output_level;
 
     // For multi-level universal, the strategy is to make this look more like
     // leveled. We pick one of the files marked for compaction and compact with
@@ -1522,100 +1542,10 @@ Compaction* UniversalCompactionBuilder::PickDeleteTriggeredCompaction() {
     if (start_level_inputs.empty()) {
       return nullptr;
     }
-
-    int max_output_level = vstorage_->MaxOutputLevel(allow_ingest_behind_);
-    // Pick the first non-empty level after the start_level
-    for (output_level = start_level + 1; output_level <= max_output_level;
-         output_level++) {
-      if (vstorage_->NumLevelFiles(output_level) != 0) {
-        break;
-      }
-    }
-
-    // If all higher levels are empty, pick the highest level as output level
-    if (output_level > max_output_level) {
-      if (start_level == 0) {
-        output_level = max_output_level;
-      } else {
-        // If start level is non-zero and all higher levels are empty, this
-        // compaction will translate into a trivial move. Since the idea is
-        // to reclaim space and trivial move doesn't help with that, we
-        // skip compaction in this case and return nullptr
-        return nullptr;
-      }
-    }
-    assert(output_level <= max_output_level);
-
-    if (!MeetsOutputLevelRequirements(output_level)) {
-      return nullptr;
-    }
-
-    if (output_level != 0) {
-      // For standalone range deletion, we don't want to compact it with newer
-      // L0 files that it doesn't cover.
-      const FileMetaData* starting_l0_file =
-          (start_level == 0 && start_level_inputs.size() == 1 &&
-           start_level_inputs.files[0]->FileIsStandAloneRangeTombstone())
-              ? start_level_inputs.files[0]
-              : nullptr;
-
-      if (start_level == 0) {
-        if (!picker_->GetOverlappingL0Files(vstorage_, &start_level_inputs,
-                                            output_level, nullptr,
-                                            starting_l0_file)) {
-          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,
-                                     starting_l0_file)) {
-        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);
-    } else {
-      inputs.push_back(start_level_inputs);
-    }
   }
 
-  uint64_t estimated_total_size = 0;
-  // Use size of the output level as estimated file size
-  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),
-      /* max_grandparent_overlap_bytes */ 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::kFilesMarkedForCompaction,
-      /* trim_ts */ "", score_,
-      /* l0_files_might_overlap */ true);
+  return BuildCompactionToNextLevel(
+      start_level_inputs, CompactionReason::kFilesMarkedForCompaction);
 }
 
 Compaction* UniversalCompactionBuilder::PickCompactionToOldest(
@@ -1782,6 +1712,153 @@ Compaction* UniversalCompactionBuilder::PickPeriodicCompaction() {
   return c;
 }
 
+Compaction* UniversalCompactionBuilder::PickReadTriggeredCompaction() {
+  ROCKS_LOG_BUFFER(log_buffer_, "[%s] Universal: Read Triggered Compaction",
+                   cf_name_.c_str());
+
+  if (vstorage_->ReadTriggeredCompactionFiles().empty()) {
+    return nullptr;
+  }
+
+  CompactionInputFiles start_level_inputs;
+  int start_level = -1;
+
+  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;
+  }
+
+  // For intra L0 compactions, pick up all overlapping files
+  if (vstorage_->num_levels() == 1 && start_level_inputs.level == 0) {
+    if (!picker_->GetOverlappingL0Files(vstorage_, &start_level_inputs,
+                                        /*output_level=*/0, nullptr, nullptr)) {
+      return nullptr;
+    }
+  }
+
+  return BuildCompactionToNextLevel(start_level_inputs,
+                                    CompactionReason::kReadTriggered);
+}
+
+Compaction* UniversalCompactionBuilder::BuildCompactionToNextLevel(
+    CompactionInputFiles& start_level_inputs, CompactionReason reason) {
+  int start_level = start_level_inputs.level;
+  int max_output_level = vstorage_->MaxOutputLevel(allow_ingest_behind_);
+
+  // Pick the first non-empty level after start_level as output.
+  int output_level;
+  for (output_level = start_level + 1; output_level <= max_output_level;
+       output_level++) {
+    if (vstorage_->NumLevelFiles(output_level) != 0) {
+      break;
+    }
+  }
+
+  // If all higher levels are empty, pick the highest level as output level
+  // for L0 start. For non-L0 start, a trivial move doesn't help reclaim
+  // space, so skip.
+  if (output_level > max_output_level) {
+    if (start_level == 0) {
+      output_level = max_output_level;
+    } else {
+      return nullptr;
+    }
+  }
+
+  if (start_level_inputs.size() <= 1 && output_level == 0) {
+    // If only the last file in L0 is marked for compaction, ignore it
+    return nullptr;
+  }
+
+  std::vector<CompactionInputFiles> inputs;
+  std::vector<FileMetaData*> grandparents;
+
+  if (output_level != 0) {
+    if (!MeetsOutputLevelRequirements(output_level)) {
+      return nullptr;
+    }
+
+    // For standalone range deletion, we don't want to compact it with newer
+    // L0 files that it doesn't cover.
+    const FileMetaData* starting_l0_file =
+        (start_level == 0 && start_level_inputs.size() == 1 &&
+         start_level_inputs.files[0]->FileIsStandAloneRangeTombstone())
+            ? start_level_inputs.files[0]
+            : nullptr;
+
+    if (start_level == 0) {
+      if (!picker_->GetOverlappingL0Files(vstorage_, &start_level_inputs,
+                                          output_level, nullptr,
+                                          starting_l0_file)) {
+        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, starting_l0_file)) {
+      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);
+  } else {
+    inputs.push_back(start_level_inputs);
+  }
+
+  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),
+      /* max_grandparent_overlap_bytes */ 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_, reason,
+      /* 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();

db/db_compaction_test.cc

@@ -11693,6 +11693,76 @@ TEST_F(DBCompactionTest, PeriodicTask) {
   Close();
 }
 
+TEST_F(DBCompactionTest, ReadTriggeredCompaction) {
+  Options options = CurrentOptions();
+  options.num_levels = 3;
+  options.level0_file_num_compaction_trigger = 10;
+  options.read_triggered_compaction_threshold = 0.001;
+  options.disable_auto_compactions = true;
+  DestroyAndReopen(options);
+
+  // Write data at L2 first so L1 is not the bottommost level
+  Random rnd(301);
+  for (int i = 0; i < 5; ++i) {
+    ASSERT_OK(Put(Key(i), rnd.RandomString(1024)));
+  }
+  ASSERT_OK(Flush());
+  MoveFilesToLevel(2);
+  ASSERT_EQ("0,0,1", FilesPerLevel());
+
+  // Write more data and move to L1 (the hot level)
+  for (int i = 5; i < 10; ++i) {
+    ASSERT_OK(Put(Key(i), rnd.RandomString(1024)));
+  }
+  ASSERT_OK(Flush());
+  MoveFilesToLevel(1);
+  ASSERT_EQ("0,1,1", FilesPerLevel());
+
+  ColumnFamilyMetaData cf_meta;
+  dbfull()->GetColumnFamilyMetaData(dbfull()->DefaultColumnFamily(), &cf_meta);
+  ASSERT_EQ(cf_meta.levels[1].files.size(), 1);
+  uint64_t file_size = cf_meta.levels[1].files[0].size;
+  // Set reads high enough to exceed threshold (0.001 * file_size)
+  uint64_t reads_needed =
+      static_cast<uint64_t>(0.002 * static_cast<double>(file_size));
+
+  {
+    auto* cfd = static_cast_with_check<ColumnFamilyHandleImpl>(
+                    dbfull()->DefaultColumnFamily())
+                    ->cfd();
+    auto* vstorage = cfd->current()->storage_info();
+    for (auto* f : vstorage->LevelFiles(1)) {
+      f->stats.num_collapsible_entry_reads_sampled.store(reads_needed);
+    }
+  }
+
+  std::atomic<int> read_triggered_compactions{0};
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
+      "LevelCompactionPicker::PickCompaction:Return", [&](void* arg) {
+        Compaction* compaction = static_cast<Compaction*>(arg);
+        if (compaction->compaction_reason() ==
+            CompactionReason::kReadTriggered) {
+          read_triggered_compactions.fetch_add(1, std::memory_order_relaxed);
+        }
+      });
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->EnableProcessing();
+
+  // Enable auto compactions to trigger
+  ASSERT_OK(dbfull()->SetOptions({{"disable_auto_compactions", "false"}}));
+  ASSERT_OK(dbfull()->TEST_WaitForCompact());
+
+  ASSERT_GE(read_triggered_compactions, 1);
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->DisableProcessing();
+
+  // Verify the L1 file was compacted down (L1 should be empty now)
+  ASSERT_EQ(0, NumTableFilesAtLevel(1));
+
+  // Verify data integrity: all keys are still readable
+  for (int i = 0; i < 10; ++i) {
+    ASSERT_NE(Get(Key(i)), "NOT_FOUND");
+  }
+}
+
 // Regression test for a bug in SetupOtherFilesWithRoundRobinExpansion where
 // duplicate files are added to the compaction input, corrupting
 // ExpandInputsToCleanCut and violating the clean-cut invariant. The bug

db/db_impl/db_impl.cc

@@ -822,8 +822,9 @@ static uint64_t GetMinTimeBasedCompactionInterval(
 }
 
 uint64_t DBImpl::ComputeTriggerCompactionPeriod() {
-  // Start with a maximum period of every 12 hours.
-  uint64_t period_sec = 12 * 60 * 60;
+  // Start with the configured maximum, then reduce based on other options.
+  uint64_t period_sec =
+      mutable_db_options_.max_compaction_trigger_wakeup_seconds;
 
   // Consider DB-level options that have the DB waking up periodically anyway.
   // Waking up to check for compactions at the same interval should be no
@@ -7036,11 +7037,12 @@ void DBImpl::TriggerPeriodicCompaction() {
       if (cfd->queued_for_compaction()) {
         continue;
       }
-      // Check if this CF has any time-based compaction trigger configured.
-      // This includes periodic_compaction_seconds, ttl, or FIFO temperature
-      // thresholds. Note: periodic_compaction_seconds may be 0 even when
-      // ttl or temperature thresholds are set, due to option sanitization.
-      if (GetMinTimeBasedCompactionInterval(cfd->GetLatestCFOptions()) > 0) {
+      // Check if this CF has any time-based or read-triggered compaction
+      // configured. This includes periodic_compaction_seconds, ttl, FIFO
+      // temperature thresholds, or read_triggered_compaction_threshold.
+      if (GetMinTimeBasedCompactionInterval(cfd->GetLatestCFOptions()) > 0 ||
+          cfd->GetLatestMutableCFOptions().read_triggered_compaction_threshold >
+              0.0) {
         TEST_SYNC_POINT_CALLBACK(
             "DBImpl::TriggerPeriodicCompaction:BeforeComputeCompactionScore",
             cfd);

db/periodic_task_scheduler_test.cc

@@ -293,6 +293,33 @@ TEST_F(PeriodicTaskSchedulerTest, TriggerCompactionPeriodComputation) {
     options.ttl = 3;  // 3 / 5 = 0, but clamped to 1
     test_period(options, 1);
   }
+
+  // Case 11: max_compaction_trigger_wakeup_seconds caps the period
+  {
+    Options options;
+    options.stats_dump_period_sec = 24 * 60 * 60;  // 24 hours
+    options.stats_persist_period_sec = 0;
+    options.max_compaction_trigger_wakeup_seconds = 600;  // 10 minutes
+    test_period(options, 600);
+  }
+
+  // Case 12: max_compaction_trigger_wakeup_seconds lower than stats_dump
+  {
+    Options options;
+    options.stats_dump_period_sec = 3600;  // 1 hour
+    options.stats_persist_period_sec = 0;
+    options.max_compaction_trigger_wakeup_seconds = 300;  // 5 minutes
+    test_period(options, 300);  // max_periodic cap wins
+  }
+
+  // Case 13: max_compaction_trigger_wakeup_seconds larger than stats_dump
+  {
+    Options options;
+    options.stats_dump_period_sec = 300;  // 5 minutes
+    options.stats_persist_period_sec = 0;
+    options.max_compaction_trigger_wakeup_seconds = 3600;  // 1 hour
+    test_period(options, 300);  // stats_dump wins (smaller)
+  }
 }
 
 // Test that TriggerPeriodicCompaction() properly considers CFs for compaction
@@ -343,15 +370,19 @@ TEST_F(TriggerCompactionTest, QueuesAllTimeBasedOptions) {
 
   // Create the additional CFs (using same options, will set specific options
   // on reopen)
-  CreateColumnFamilies(
-      {"cf_periodic", "cf_ttl", "cf_bottommost", "cf_fifo", "cf_none"},
-      options);
+  CreateColumnFamilies({"cf_periodic", "cf_ttl", "cf_bottommost", "cf_fifo",
+                        "cf_read_triggered", "cf_none"},
+                       options);
   Close();
 
   // Now reopen with specific options for each CF
-  std::vector<std::string> cf_names = {
-      kDefaultColumnFamilyName, "cf_periodic", "cf_ttl",
-      "cf_bottommost",          "cf_fifo",     "cf_none"};
+  std::vector<std::string> cf_names = {kDefaultColumnFamilyName,
+                                       "cf_periodic",
+                                       "cf_ttl",
+                                       "cf_bottommost",
+                                       "cf_fifo",
+                                       "cf_read_triggered",
+                                       "cf_none"};
   std::vector<Options> cf_options;
 
   // default: no time-based options
@@ -381,6 +412,11 @@ TEST_F(TriggerCompactionTest, QueuesAllTimeBasedOptions) {
       {Temperature::kWarm, 100}};
   cf_options.push_back(opt_fifo);
 
+  // cf_read_triggered: read_triggered_compaction_threshold
+  Options opt_read_triggered = options;
+  opt_read_triggered.read_triggered_compaction_threshold = 0.001;
+  cf_options.push_back(opt_read_triggered);
+
   // cf_none: explicitly no time-based options
   Options opt_none = options;
   opt_none.periodic_compaction_seconds = 0;
@@ -404,6 +440,8 @@ TEST_F(TriggerCompactionTest, QueuesAllTimeBasedOptions) {
       << "Expected cf_bottommost to have compaction score computed";
   EXPECT_GT(cfs_with_score_computed.count("cf_fifo"), 0u)
       << "Expected cf_fifo to have compaction score computed";
+  EXPECT_GT(cfs_with_score_computed.count("cf_read_triggered"), 0u)
+      << "Expected cf_read_triggered to have compaction score computed";
 
   // CFs without time-based options should NOT have score computed
   EXPECT_EQ(cfs_with_score_computed.count("default"), 0u)

db/version_set.cc

@@ -3980,6 +3980,9 @@ void VersionStorageInfo::ComputeCompactionScore(
       mutable_cf_options.blob_garbage_collection_age_cutoff,
       mutable_cf_options.blob_garbage_collection_force_threshold,
       mutable_cf_options.enable_blob_garbage_collection);
+  ComputeFilesMarkedForReadTriggeredCompaction(
+      mutable_cf_options.read_triggered_compaction_threshold,
+      immutable_options.compaction_style);
 
   EstimateCompactionBytesNeeded(mutable_cf_options);
 }
@@ -4203,6 +4206,65 @@ void VersionStorageInfo::ComputeFilesMarkedForForcedBlobGC(
   }
 }
 
+void VersionStorageInfo::ComputeFilesMarkedForReadTriggeredCompaction(
+    double threshold, CompactionStyle compaction_style) {
+  read_triggered_compaction_files_.clear();
+  if (threshold <= 0.0 || compaction_style == kCompactionStyleFIFO) {
+    return;
+  }
+
+  // Skip files at the last non-empty level — there is no lower level to
+  // compact into. Exception: L0 files are allowed even when L0 is the last
+  // non-empty level, because in single-level universal or FIFO compaction, L0
+  // files can be compacted together (L0 → L0).
+  int last_non_empty = num_non_empty_levels_ - 1;
+
+  for (int level = 0; level < num_levels(); level++) {
+    if (level >= last_non_empty && level != 0) {
+      continue;
+    }
+    for (auto* f : files_[level]) {
+      if (f->being_compacted) {
+        continue;
+      }
+      uint64_t file_size = f->fd.GetFileSize();
+      if (file_size == 0) {
+        continue;
+      }
+      double reads_per_byte =
+          static_cast<double>(f->stats.num_collapsible_entry_reads_sampled.load(
+              std::memory_order_relaxed)) /
+          static_cast<double>(file_size);
+      if (reads_per_byte > threshold) {
+        read_triggered_compaction_files_.emplace_back(level, f);
+      }
+    }
+  }
+
+  // Snapshot reads_per_byte for each file so the sort comparator sees stable
+  // values.
+  std::unordered_map<const FileMetaData*, double> reads_per_byte_snapshot;
+  reads_per_byte_snapshot.reserve(read_triggered_compaction_files_.size());
+  for (const auto& entry : read_triggered_compaction_files_) {
+    const auto* f = entry.second;
+    uint64_t file_size = f->fd.GetFileSize();
+    reads_per_byte_snapshot[f] =
+        file_size > 0 ? static_cast<double>(
+                            f->stats.num_collapsible_entry_reads_sampled.load(
+                                std::memory_order_relaxed)) /
+                            static_cast<double>(file_size)
+                      : 0.0;
+  }
+
+  // Sort by reads_per_byte descending so the hottest files are picked first
+  std::sort(read_triggered_compaction_files_.begin(),
+            read_triggered_compaction_files_.end(),
+            [&reads_per_byte_snapshot](const auto& a, const auto& b) {
+              return reads_per_byte_snapshot.at(a.second) >
+                     reads_per_byte_snapshot.at(b.second);
+            });
+}
+
 namespace {
 
 // used to sort files by size

db/version_set.h

@@ -250,6 +250,13 @@ class VersionStorageInfo {
       double blob_garbage_collection_force_threshold,
       bool enable_blob_garbage_collection);
 
+  // This computes read_triggered_compaction_files_ and is called by
+  // ComputeCompactionScore()
+  //
+  // REQUIRES: DB mutex held
+  void ComputeFilesMarkedForReadTriggeredCompaction(
+      double threshold, CompactionStyle compaction_style);
+
   bool level0_non_overlapping() const { return level0_non_overlapping_; }
 
   // Updates the oldest snapshot and related internal state, like the bottommost
@@ -536,6 +543,19 @@ class VersionStorageInfo {
     return files_marked_for_forced_blob_gc_;
   }
 
+  // REQUIRES: ComputeCompactionScore has been called
+  // REQUIRES: DB mutex held during access
+  const autovector<std::pair<int, FileMetaData*>>&
+  ReadTriggeredCompactionFiles() const {
+    assert(finalized_);
+    return read_triggered_compaction_files_;
+  }
+
+  void TEST_AddFileMarkedForReadTriggeredCompaction(int level,
+                                                    FileMetaData* f) {
+    read_triggered_compaction_files_.emplace_back(level, f);
+  }
+
   int base_level() const { return base_level_; }
   double level_multiplier() const { return level_multiplier_; }
 
@@ -744,6 +764,8 @@ class VersionStorageInfo {
 
   autovector<std::pair<int, FileMetaData*>> files_marked_for_forced_blob_gc_;
 
+  autovector<std::pair<int, FileMetaData*>> read_triggered_compaction_files_;
+
   // Threshold for needing to mark another bottommost file. Maintain it so we
   // can quickly check when releasing a snapshot whether more bottommost files
   // became eligible for compaction. It's defined as the min of the max nonzero

db_stress_tool/db_stress_common.h

@@ -199,6 +199,7 @@ DECLARE_bool(statistics);
 DECLARE_bool(sync);
 DECLARE_bool(use_fsync);
 DECLARE_uint64(stats_dump_period_sec);
+DECLARE_uint64(max_compaction_trigger_wakeup_seconds);
 DECLARE_uint64(bytes_per_sync);
 DECLARE_uint64(wal_bytes_per_sync);
 DECLARE_int32(kill_random_test);
@@ -458,6 +459,7 @@ DECLARE_uint64(compaction_on_deletion_min_file_size);
 DECLARE_int32(compaction_on_deletion_trigger_count);
 DECLARE_int32(compaction_on_deletion_window_size);
 DECLARE_double(compaction_on_deletion_ratio);
+DECLARE_double(read_triggered_compaction_threshold);
 
 constexpr long KB = 1024;
 constexpr int kRandomValueMaxFactor = 3;

db_stress_tool/db_stress_gflags.cc

@@ -1321,6 +1321,11 @@ DEFINE_uint64(stats_dump_period_sec,
               ROCKSDB_NAMESPACE::Options().stats_dump_period_sec,
               "Gap between printing stats to log in seconds");
 
+DEFINE_uint64(
+    max_compaction_trigger_wakeup_seconds,
+    ROCKSDB_NAMESPACE::Options().max_compaction_trigger_wakeup_seconds,
+    "Sets DB option max_compaction_trigger_wakeup_seconds.");
+
 DEFINE_bool(verification_only, false,
             "If true, tests will only execute verification step");
 extern "C" bool RocksDbIOUringEnable() { return true; }
@@ -1597,6 +1602,10 @@ DEFINE_double(compaction_on_deletion_ratio, 0.5,
               "Deletion ratio threshold for triggering compaction. "
               "Default: 0.5 (50%)");
 
+DEFINE_double(read_triggered_compaction_threshold,
+              ROCKSDB_NAMESPACE::Options().read_triggered_compaction_threshold,
+              "Sets CF option read_triggered_compaction_threshold.");
+
 DEFINE_bool(
     auto_refresh_iterator_with_snapshot,
     ROCKSDB_NAMESPACE::ReadOptions().auto_refresh_iterator_with_snapshot,

db_stress_tool/db_stress_test_base.cc

@@ -4513,6 +4513,8 @@ void InitializeOptionsFromFlags(
   options.daily_offpeak_time_utc = FLAGS_daily_offpeak_time_utc;
   options.stats_dump_period_sec =
       static_cast<unsigned int>(FLAGS_stats_dump_period_sec);
+  options.max_compaction_trigger_wakeup_seconds =
+      FLAGS_max_compaction_trigger_wakeup_seconds;
   options.ttl = FLAGS_compaction_ttl;
   options.enable_pipelined_write = FLAGS_enable_pipelined_write;
   options.enable_write_thread_adaptive_yield =
@@ -4566,6 +4568,8 @@ void InitializeOptionsFromFlags(
       FLAGS_blob_garbage_collection_force_threshold;
   options.blob_compaction_readahead_size = FLAGS_blob_compaction_readahead_size;
   options.blob_file_starting_level = FLAGS_blob_file_starting_level;
+  options.read_triggered_compaction_threshold =
+      FLAGS_read_triggered_compaction_threshold;
 
   if (FLAGS_use_blob_cache) {
     if (FLAGS_use_shared_block_and_blob_cache) {

include/rocksdb/advanced_options.h

@@ -940,6 +940,61 @@ struct AdvancedColumnFamilyOptions {
   // Dynamically changeable through SetOptions() API
   uint64_t periodic_compaction_seconds = 0xfffffffffffffffe;
 
+  // When set to a positive value, enables read-triggered compaction. An SST
+  // file is marked for compaction when its estimated read frequency
+  // (estimated_reads / file_size) exceeds this threshold. This helps reduce
+  // read amplification for hot keys by compacting frequently-read files.
+  //
+  // Only "collapsible" reads are counted — lookups that return NotFound
+  // (bloom filter false positive), Delete/SingleDeletion (tombstone), or
+  // Merge (partial result). These are reads where the file contributed no
+  // final value and compaction would eliminate the wasted work.
+  //
+  // Choosing a value: the threshold balances read IO saved against the
+  // write amplification (WA) of an extra compaction. This assumes the
+  // block-based table format is being used,
+  //
+  // Break-even derivation (no block cache):
+  //   Let r = estimated_reads / file_size  (the threshold)
+  //       S = file_size
+  //       B = block_size             (typically 4 KB)
+  //       F = level fanout           (typically ~10)
+  //
+  //   Each collapsible read wastes one data-block read = B bytes of IO.
+  //   Total wasted read IO for a file = r * S * B.
+  //
+  //   Compaction cost: one level-L file overlaps ~F files in level L+1,
+  //   so we read (1 + F) files and write (1 + F) files.
+  //   Total compaction IO = 2 * (1 + F) * S.
+  //
+  //   Break-even when wasted read IO equals compaction IO:
+  //     r * S * B = 2 * (1 + F) * S
+  //     r = 2 * (1 + F) / B
+  //
+  //   With F = 10, B = 4096:  r = 22 / 4096 ≈ 0.005.
+  //
+  // With a block-cache hit rate h (0 ≤ h < 1), each collapsible read
+  // only costs (1 - h) * B bytes of actual disk IO, so:
+  //     r = 2 * (1 + F) / ((1 - h) * B)
+  //
+  //   h = 0   → r ≈ 0.005
+  //   h = 0.5 → r ≈ 0.01
+  //   h = 0.9 → r ≈ 0.05
+  //
+  // A recommended starting point is 0.01, which avoids triggering
+  // compactions that cost more IO than they save for most cache-friendly
+  // workloads, while still being responsive enough to compact files with
+  // significant wasted reads.
+  //
+  // For this feature to take effect on a "quiet" DB (no writes), the DB-level
+  // option `max_compaction_trigger_wakeup_seconds` must also be set to a
+  // non-zero value so the periodic background job can re-evaluate files.
+  //
+  // Valid range: >= 0.0 (must be finite). Use 0.0 to disable.
+  //
+  // Dynamically changeable through SetOptions() API
+  double read_triggered_compaction_threshold = 0.0;
+
   // If this option is set then 1 in N blocks are compressed
   // using a fast (lz4) and slow (zstd) compression algorithm.
   // The compressibility is reported as stats and the stored

include/rocksdb/c.h

@@ -1712,6 +1712,19 @@ extern ROCKSDB_LIBRARY_API void rocksdb_options_set_blob_gc_force_threshold(
 extern ROCKSDB_LIBRARY_API double rocksdb_options_get_blob_gc_force_threshold(
     rocksdb_options_t* opt);
 
+extern ROCKSDB_LIBRARY_API void
+rocksdb_options_set_read_triggered_compaction_threshold(rocksdb_options_t* opt,
+                                                        double val);
+extern ROCKSDB_LIBRARY_API double
+rocksdb_options_get_read_triggered_compaction_threshold(rocksdb_options_t* opt);
+
+extern ROCKSDB_LIBRARY_API void
+rocksdb_options_set_max_compaction_trigger_wakeup_seconds(
+    rocksdb_options_t* opt, uint64_t val);
+extern ROCKSDB_LIBRARY_API uint64_t
+rocksdb_options_get_max_compaction_trigger_wakeup_seconds(
+    rocksdb_options_t* opt);
+
 extern ROCKSDB_LIBRARY_API void
 rocksdb_options_set_blob_compaction_readahead_size(rocksdb_options_t* opt,
                                                    uint64_t val);

include/rocksdb/listener.h

@@ -157,6 +157,8 @@ enum class CompactionReason : int {
   // [InternalOnly] DBImpl::ReFitLevel treated as a compaction,
   // Used only for internal conflict checking with other compactions
   kRefitLevel,
+  // Compaction triggered by high read frequency on SST files
+  kReadTriggered,
   // total number of compaction reasons, new reasons must be added above this.
   kNumOfReasons,
 };

include/rocksdb/options.h

@@ -1709,6 +1709,28 @@ struct DBOptions {
   // this field blank. Default: Empty string (no offpeak).
   std::string daily_offpeak_time_utc = "";
 
+  // Maximum interval in seconds between periodic compaction trigger checks.
+  // The periodic trigger re-evaluates compaction scores for all column
+  // families, which is necessary for features like read-triggered compaction
+  // and time-based compaction to work on a "quiet" DB with no writes.
+  //
+  // This is an upper bound: the actual check interval may be reduced to
+  // align with stats_dump_period_sec, stats_persist_period_sec, or per-CF
+  // time-based compaction intervals (periodic_compaction_seconds, ttl, etc.).
+  //
+  // Note: this option controls how often RocksDB *checks* whether compaction
+  // is needed. It is different from the CF option `periodic_compaction_seconds`
+  // which controls the *age threshold* at which SST files become eligible for
+  // periodic compaction.
+  //
+  // The minimum effective period is 1 second (values below 1 are clamped to 1).
+  // Setting this to 0 results in the most aggressive 1-second polling.
+  //
+  // Default: 43200 (12 hours)
+  //
+  // Dynamically changeable through SetDBOptions() API.
+  uint64_t max_compaction_trigger_wakeup_seconds = 43200;
+
   // EXPERIMENTAL
 
   // When a RocksDB database is opened in follower mode, this option

java/rocksjni/options.cc

@@ -1486,6 +1486,29 @@ void Java_org_rocksdb_Options_setStatsDumpPeriodSec(
       static_cast<unsigned int>(jstats_dump_period_sec);
 }
 
+/*
+ * Class:     org_rocksdb_Options
+ * Method:    setMaxCompactionTriggerWakeupSeconds
+ * Signature: (JJ)V
+ */
+void Java_org_rocksdb_Options_setMaxCompactionTriggerWakeupSeconds(
+    JNIEnv*, jclass, jlong jhandle, jlong jval) {
+  reinterpret_cast<ROCKSDB_NAMESPACE::Options*>(jhandle)
+      ->max_compaction_trigger_wakeup_seconds = static_cast<uint64_t>(jval);
+}
+
+/*
+ * Class:     org_rocksdb_Options
+ * Method:    maxCompactionTriggerWakeupSeconds
+ * Signature: (J)J
+ */
+jlong Java_org_rocksdb_Options_maxCompactionTriggerWakeupSeconds(
+    JNIEnv*, jclass, jlong jhandle) {
+  return static_cast<jlong>(
+      reinterpret_cast<ROCKSDB_NAMESPACE::Options*>(jhandle)
+          ->max_compaction_trigger_wakeup_seconds);
+}
+
 /*
  * Class:     org_rocksdb_Options
  * Method:    statsPersistPeriodSec
@@ -3769,6 +3792,28 @@ jdouble Java_org_rocksdb_Options_blobGarbageCollectionForceThreshold(
   return static_cast<jdouble>(opts->blob_garbage_collection_force_threshold);
 }
 
+/*
+ * Class:     org_rocksdb_Options
+ * Method:    setReadTriggeredCompactionThreshold
+ * Signature: (JD)V
+ */
+void Java_org_rocksdb_Options_setReadTriggeredCompactionThreshold(
+    JNIEnv*, jclass, jlong jhandle, jdouble jval) {
+  auto* opts = reinterpret_cast<ROCKSDB_NAMESPACE::Options*>(jhandle);
+  opts->read_triggered_compaction_threshold = static_cast<double>(jval);
+}
+
+/*
+ * Class:     org_rocksdb_Options
+ * Method:    readTriggeredCompactionThreshold
+ * Signature: (J)D
+ */
+jdouble Java_org_rocksdb_Options_readTriggeredCompactionThreshold(
+    JNIEnv*, jclass, jlong jhandle) {
+  auto* opts = reinterpret_cast<ROCKSDB_NAMESPACE::Options*>(jhandle);
+  return static_cast<jdouble>(opts->read_triggered_compaction_threshold);
+}
+
 /*
  * Class:     org_rocksdb_Options
  * Method:    setBlobCompactionReadaheadSize
@@ -5677,6 +5722,30 @@ Java_org_rocksdb_ColumnFamilyOptions_blobGarbageCollectionForceThreshold(
   return static_cast<jdouble>(opts->blob_garbage_collection_force_threshold);
 }
 
+/*
+ * Class:     org_rocksdb_ColumnFamilyOptions
+ * Method:    setReadTriggeredCompactionThreshold
+ * Signature: (JD)V
+ */
+void Java_org_rocksdb_ColumnFamilyOptions_setReadTriggeredCompactionThreshold(
+    JNIEnv*, jclass, jlong jhandle, jdouble jval) {
+  auto* opts =
+      reinterpret_cast<ROCKSDB_NAMESPACE::ColumnFamilyOptions*>(jhandle);
+  opts->read_triggered_compaction_threshold = static_cast<double>(jval);
+}
+
+/*
+ * Class:     org_rocksdb_ColumnFamilyOptions
+ * Method:    readTriggeredCompactionThreshold
+ * Signature: (J)D
+ */
+jdouble Java_org_rocksdb_ColumnFamilyOptions_readTriggeredCompactionThreshold(
+    JNIEnv*, jclass, jlong jhandle) {
+  auto* opts =
+      reinterpret_cast<ROCKSDB_NAMESPACE::ColumnFamilyOptions*>(jhandle);
+  return static_cast<jdouble>(opts->read_triggered_compaction_threshold);
+}
+
 /*
  * Class:     org_rocksdb_ColumnFamilyOptions
  * Method:    setBlobCompactionReadaheadSize
@@ -6914,6 +6983,29 @@ jint Java_org_rocksdb_DBOptions_statsDumpPeriodSec(JNIEnv*, jclass,
       ->stats_dump_period_sec;
 }
 
+/*
+ * Class:     org_rocksdb_DBOptions
+ * Method:    setMaxCompactionTriggerWakeupSeconds
+ * Signature: (JJ)V
+ */
+void Java_org_rocksdb_DBOptions_setMaxCompactionTriggerWakeupSeconds(
+    JNIEnv*, jclass, jlong jhandle, jlong jval) {
+  reinterpret_cast<ROCKSDB_NAMESPACE::DBOptions*>(jhandle)
+      ->max_compaction_trigger_wakeup_seconds = static_cast<uint64_t>(jval);
+}
+
+/*
+ * Class:     org_rocksdb_DBOptions
+ * Method:    maxCompactionTriggerWakeupSeconds
+ * Signature: (J)J
+ */
+jlong Java_org_rocksdb_DBOptions_maxCompactionTriggerWakeupSeconds(
+    JNIEnv*, jclass, jlong jhandle) {
+  return static_cast<jlong>(
+      reinterpret_cast<ROCKSDB_NAMESPACE::DBOptions*>(jhandle)
+          ->max_compaction_trigger_wakeup_seconds);
+}
+
 /*
  * Class:     org_rocksdb_DBOptions
  * Method:    setStatsPersistPeriodSec

java/src/main/java/org/rocksdb/AdvancedMutableColumnFamilyOptionsInterface.java

@@ -756,6 +756,23 @@ public interface AdvancedMutableColumnFamilyOptionsInterface<
    */
   double blobGarbageCollectionForceThreshold();
 
+  /**
+   * Set threshold for read-triggered compaction. An SST file is marked for
+   * compaction when its sampled read frequency (sampled_reads / file_size)
+   * exceeds this value. Use 0.0 to disable.
+   *
+   * @param readTriggeredCompactionThreshold the threshold value
+   * @return the reference to the current options
+   */
+  T setReadTriggeredCompactionThreshold(double readTriggeredCompactionThreshold);
+
+  /**
+   * Get the current value for read-triggered compaction threshold.
+   *
+   * @return the current threshold
+   */
+  double readTriggeredCompactionThreshold();
+
   /**
    * Set compaction readahead for blob files.
    * <p>

java/src/main/java/org/rocksdb/ColumnFamilyOptions.java

@@ -1222,6 +1222,18 @@ public class ColumnFamilyOptions
     return blobGarbageCollectionForceThreshold(nativeHandle_);
   }
 
+  @Override
+  public ColumnFamilyOptions setReadTriggeredCompactionThreshold(
+      final double readTriggeredCompactionThreshold) {
+    setReadTriggeredCompactionThreshold(nativeHandle_, readTriggeredCompactionThreshold);
+    return this;
+  }
+
+  @Override
+  public double readTriggeredCompactionThreshold() {
+    return readTriggeredCompactionThreshold(nativeHandle_);
+  }
+
   /**
    * Set compaction readahead for blob files.
    * <p>
@@ -1496,6 +1508,9 @@ public class ColumnFamilyOptions
   private static native void setBlobGarbageCollectionForceThreshold(
       final long nativeHandle_, final double blobGarbageCollectionForceThreshold);
   private static native double blobGarbageCollectionForceThreshold(final long nativeHandle_);
+  private static native void setReadTriggeredCompactionThreshold(
+      final long nativeHandle_, final double readTriggeredCompactionThreshold);
+  private static native double readTriggeredCompactionThreshold(final long nativeHandle_);
   private static native void setBlobCompactionReadaheadSize(
       final long nativeHandle_, final long blobCompactionReadaheadSize);
   private static native long blobCompactionReadaheadSize(final long nativeHandle_);

java/src/main/java/org/rocksdb/DBOptions.java

@@ -691,6 +691,20 @@ public class DBOptions extends RocksObject
     return statsDumpPeriodSec(nativeHandle_);
   }
 
+  @Override
+  public DBOptions setMaxCompactionTriggerWakeupSeconds(
+      final long maxCompactionTriggerWakeupSeconds) {
+    assert (isOwningHandle());
+    setMaxCompactionTriggerWakeupSeconds(nativeHandle_, maxCompactionTriggerWakeupSeconds);
+    return this;
+  }
+
+  @Override
+  public long maxCompactionTriggerWakeupSeconds() {
+    assert (isOwningHandle());
+    return maxCompactionTriggerWakeupSeconds(nativeHandle_);
+  }
+
   @Override
   public DBOptions setStatsPersistPeriodSec(
       final int statsPersistPeriodSec) {
@@ -1335,6 +1349,9 @@ public class DBOptions extends RocksObject
   private static native boolean isFdCloseOnExec(long handle);
   private static native void setStatsDumpPeriodSec(long handle, int statsDumpPeriodSec);
   private static native int statsDumpPeriodSec(long handle);
+  private static native void setMaxCompactionTriggerWakeupSeconds(
+      long handle, long maxCompactionTriggerWakeupSeconds);
+  private static native long maxCompactionTriggerWakeupSeconds(long handle);
   private static native void setStatsPersistPeriodSec(
       final long handle, final int statsPersistPeriodSec);
   private static native int statsPersistPeriodSec(final long handle);

java/src/main/java/org/rocksdb/MutableColumnFamilyOptions.java

@@ -99,7 +99,8 @@ public class MutableColumnFamilyOptions extends AbstractMutableOptions {
     max_bytes_for_level_multiplier(ValueType.INT),
     max_bytes_for_level_multiplier_additional(ValueType.INT_ARRAY),
     ttl(ValueType.LONG),
-    periodic_compaction_seconds(ValueType.LONG);
+    periodic_compaction_seconds(ValueType.LONG),
+    read_triggered_compaction_threshold(ValueType.DOUBLE);
 
     private final ValueType valueType;
     CompactionOption(final ValueType valueType) {
@@ -585,6 +586,18 @@ public class MutableColumnFamilyOptions extends AbstractMutableOptions {
       return getDouble(BlobOption.blob_garbage_collection_force_threshold);
     }
 
+    @Override
+    public MutableColumnFamilyOptionsBuilder setReadTriggeredCompactionThreshold(
+        final double readTriggeredCompactionThreshold) {
+      return setDouble(
+          CompactionOption.read_triggered_compaction_threshold, readTriggeredCompactionThreshold);
+    }
+
+    @Override
+    public double readTriggeredCompactionThreshold() {
+      return getDouble(CompactionOption.read_triggered_compaction_threshold);
+    }
+
     @Override
     public MutableColumnFamilyOptionsBuilder setBlobCompactionReadaheadSize(
         final long blobCompactionReadaheadSize) {

java/src/main/java/org/rocksdb/MutableDBOptions.java

@@ -77,6 +77,7 @@ public class MutableDBOptions extends AbstractMutableOptions {
     wal_bytes_per_sync(ValueType.LONG),
     strict_bytes_per_sync(ValueType.BOOLEAN),
     compaction_readahead_size(ValueType.LONG),
+    max_compaction_trigger_wakeup_seconds(ValueType.LONG),
 
     daily_offpeak_time_utc(ValueType.STRING);
 
@@ -215,6 +216,18 @@ public class MutableDBOptions extends AbstractMutableOptions {
       return getInt(DBOption.stats_dump_period_sec);
     }
 
+    @Override
+    public MutableDBOptionsBuilder setMaxCompactionTriggerWakeupSeconds(
+        final long maxCompactionTriggerWakeupSeconds) {
+      return setLong(
+          DBOption.max_compaction_trigger_wakeup_seconds, maxCompactionTriggerWakeupSeconds);
+    }
+
+    @Override
+    public long maxCompactionTriggerWakeupSeconds() {
+      return getLong(DBOption.max_compaction_trigger_wakeup_seconds);
+    }
+
     @Override
     public MutableDBOptionsBuilder setStatsPersistPeriodSec(
         final int statsPersistPeriodSec) {

java/src/main/java/org/rocksdb/MutableDBOptionsInterface.java

@@ -270,6 +270,23 @@ public interface MutableDBOptionsInterface<T extends MutableDBOptionsInterface<T
    */
   int statsDumpPeriodSec();
 
+  /**
+   * Maximum interval in seconds between periodic compaction trigger checks.
+   * Default: 43200 (12 hours)
+   *
+   * @param maxCompactionTriggerWakeupSeconds time interval in seconds.
+   * @return the instance of the current object.
+   */
+  T setMaxCompactionTriggerWakeupSeconds(long maxCompactionTriggerWakeupSeconds);
+
+  /**
+   * Maximum interval in seconds between periodic compaction trigger checks.
+   * Default: 43200 (12 hours)
+   *
+   * @return time interval in seconds.
+   */
+  long maxCompactionTriggerWakeupSeconds();
+
   /**
    * If not zero, dump rocksdb.stats to RocksDB every
    * {@code statsPersistPeriodSec}

java/src/main/java/org/rocksdb/Options.java

@@ -780,6 +780,20 @@ public class Options extends RocksObject
     return this;
   }
 
+  @Override
+  public Options setMaxCompactionTriggerWakeupSeconds(
+      final long maxCompactionTriggerWakeupSeconds) {
+    assert (isOwningHandle());
+    setMaxCompactionTriggerWakeupSeconds(nativeHandle_, maxCompactionTriggerWakeupSeconds);
+    return this;
+  }
+
+  @Override
+  public long maxCompactionTriggerWakeupSeconds() {
+    assert (isOwningHandle());
+    return maxCompactionTriggerWakeupSeconds(nativeHandle_);
+  }
+
   @Override
   public Options setStatsPersistPeriodSec(
       final int statsPersistPeriodSec) {
@@ -2059,6 +2073,18 @@ public class Options extends RocksObject
     return blobGarbageCollectionForceThreshold(nativeHandle_);
   }
 
+  @Override
+  public Options setReadTriggeredCompactionThreshold(
+      final double readTriggeredCompactionThreshold) {
+    setReadTriggeredCompactionThreshold(nativeHandle_, readTriggeredCompactionThreshold);
+    return this;
+  }
+
+  @Override
+  public double readTriggeredCompactionThreshold() {
+    return readTriggeredCompactionThreshold(nativeHandle_);
+  }
+
   @Override
   public Options setBlobCompactionReadaheadSize(final long blobCompactionReadaheadSize) {
     setBlobCompactionReadaheadSize(nativeHandle_, blobCompactionReadaheadSize);
@@ -2230,6 +2256,9 @@ public class Options extends RocksObject
   private static native boolean isFdCloseOnExec(long handle);
   private static native void setStatsDumpPeriodSec(long handle, int statsDumpPeriodSec);
   private static native int statsDumpPeriodSec(long handle);
+  private static native void setMaxCompactionTriggerWakeupSeconds(
+      long handle, long maxCompactionTriggerWakeupSeconds);
+  private static native long maxCompactionTriggerWakeupSeconds(long handle);
   private static native void setStatsPersistPeriodSec(
       final long handle, final int statsPersistPeriodSec);
   private static native int statsPersistPeriodSec(final long handle);
@@ -2493,6 +2522,9 @@ public class Options extends RocksObject
   private static native void setBlobGarbageCollectionForceThreshold(
       final long nativeHandle_, final double blobGarbageCollectionForceThreshold);
   private static native double blobGarbageCollectionForceThreshold(final long nativeHandle_);
+  private static native void setReadTriggeredCompactionThreshold(
+      final long nativeHandle_, final double readTriggeredCompactionThreshold);
+  private static native double readTriggeredCompactionThreshold(final long nativeHandle_);
   private static native void setBlobCompactionReadaheadSize(
       final long nativeHandle_, final long blobCompactionReadaheadSize);
   private static native long blobCompactionReadaheadSize(final long nativeHandle_);

options/cf_options.cc

@@ -582,6 +582,11 @@ static std::unordered_map<std::string, OptionTypeInfo>
          {offsetof(struct MutableCFOptions, periodic_compaction_seconds),
           OptionType::kUInt64T, OptionVerificationType::kNormal,
           OptionTypeFlags::kMutable}},
+        {"read_triggered_compaction_threshold",
+         {offsetof(struct MutableCFOptions,
+                   read_triggered_compaction_threshold),
+          OptionType::kDouble, OptionVerificationType::kNormal,
+          OptionTypeFlags::kMutable}},
         {"preclude_last_level_data_seconds",
          {offsetof(struct MutableCFOptions, preclude_last_level_data_seconds),
           OptionType::kUInt64T, OptionVerificationType::kNormal,
@@ -1204,6 +1209,8 @@ void MutableCFOptions::Dump(Logger* log) const {
                  ttl);
   ROCKS_LOG_INFO(log, "              periodic_compaction_seconds: %" PRIu64,
                  periodic_compaction_seconds);
+  ROCKS_LOG_INFO(log, "    read_triggered_compaction_threshold: %f",
+                 read_triggered_compaction_threshold);
   ROCKS_LOG_INFO(log,
                  "              preclude_last_level_data_seconds: %" PRIu64,
                  preclude_last_level_data_seconds);

options/cf_options.h

@@ -140,6 +140,8 @@ struct MutableCFOptions {
         max_bytes_for_level_multiplier(options.max_bytes_for_level_multiplier),
         ttl(options.ttl),
         periodic_compaction_seconds(options.periodic_compaction_seconds),
+        read_triggered_compaction_threshold(
+            options.read_triggered_compaction_threshold),
         max_bytes_for_level_multiplier_additional(
             options.max_bytes_for_level_multiplier_additional),
         compaction_options_fifo(options.compaction_options_fifo),
@@ -216,6 +218,7 @@ struct MutableCFOptions {
         max_bytes_for_level_multiplier(0),
         ttl(0),
         periodic_compaction_seconds(0),
+        read_triggered_compaction_threshold(0.0),
         compaction_options_fifo(),
         preclude_last_level_data_seconds(0),
         preserve_internal_time_seconds(0),
@@ -316,6 +319,7 @@ struct MutableCFOptions {
   double max_bytes_for_level_multiplier;
   uint64_t ttl;
   uint64_t periodic_compaction_seconds;
+  double read_triggered_compaction_threshold;
   std::vector<int> max_bytes_for_level_multiplier_additional;
   CompactionOptionsFIFO compaction_options_fifo;
   CompactionOptionsUniversal compaction_options_universal;

options/db_options.cc

@@ -144,6 +144,11 @@ static std::unordered_map<std::string, OptionTypeInfo>
          {offsetof(struct MutableDBOptions, daily_offpeak_time_utc),
           OptionType::kString, OptionVerificationType::kNormal,
           OptionTypeFlags::kMutable}},
+        {"max_compaction_trigger_wakeup_seconds",
+         {offsetof(struct MutableDBOptions,
+                   max_compaction_trigger_wakeup_seconds),
+          OptionType::kUInt64T, OptionVerificationType::kNormal,
+          OptionTypeFlags::kMutable}},
 };
 
 static std::unordered_map<std::string, OptionTypeInfo>
@@ -1064,7 +1069,9 @@ MutableDBOptions::MutableDBOptions(const DBOptions& options)
       manifest_preallocation_size(options.manifest_preallocation_size),
       verify_manifest_content_on_close(
           options.verify_manifest_content_on_close),
-      daily_offpeak_time_utc(options.daily_offpeak_time_utc) {}
+      daily_offpeak_time_utc(options.daily_offpeak_time_utc),
+      max_compaction_trigger_wakeup_seconds(
+          options.max_compaction_trigger_wakeup_seconds) {}
 
 void MutableDBOptions::Dump(Logger* log) const {
   ROCKS_LOG_HEADER(log, "            Options.max_background_jobs: %d",
@@ -1121,6 +1128,9 @@ void MutableDBOptions::Dump(Logger* log) const {
                    verify_manifest_content_on_close);
   ROCKS_LOG_HEADER(log, "Options.daily_offpeak_time_utc: %s",
                    daily_offpeak_time_utc.c_str());
+  ROCKS_LOG_HEADER(log,
+                   "Options.max_compaction_trigger_wakeup_seconds: %" PRIu64,
+                   max_compaction_trigger_wakeup_seconds);
 }
 
 Status GetMutableDBOptionsFromStrings(

options/db_options.h

@@ -150,6 +150,7 @@ struct MutableDBOptions {
   size_t manifest_preallocation_size;
   bool verify_manifest_content_on_close;
   std::string daily_offpeak_time_utc;
+  uint64_t max_compaction_trigger_wakeup_seconds;
 };
 
 Status GetStringFromMutableDBOptions(const ConfigOptions& config_options,

options/options_helper.cc

@@ -192,6 +192,8 @@ void BuildDBOptions(const ImmutableDBOptions& immutable_db_options,
   options.verify_manifest_content_on_close =
       mutable_db_options.verify_manifest_content_on_close;
   options.daily_offpeak_time_utc = mutable_db_options.daily_offpeak_time_utc;
+  options.max_compaction_trigger_wakeup_seconds =
+      mutable_db_options.max_compaction_trigger_wakeup_seconds;
   options.follower_refresh_catchup_period_ms =
       immutable_db_options.follower_refresh_catchup_period_ms;
   options.follower_catchup_retry_count =
@@ -264,6 +266,8 @@ void UpdateColumnFamilyOptions(const MutableCFOptions& moptions,
       moptions.max_bytes_for_level_multiplier;
   cf_opts->ttl = moptions.ttl;
   cf_opts->periodic_compaction_seconds = moptions.periodic_compaction_seconds;
+  cf_opts->read_triggered_compaction_threshold =
+      moptions.read_triggered_compaction_threshold;
   cf_opts->preclude_last_level_data_seconds =
       moptions.preclude_last_level_data_seconds;
   cf_opts->preserve_internal_time_seconds =

options/options_settable_test.cc

@@ -476,6 +476,7 @@ TEST_F(OptionsSettableTest, DBOptionsAllFieldsSettable) {
       "allow_data_in_errors=false;"
       "enforce_single_del_contracts=false;"
       "daily_offpeak_time_utc=08:30-19:00;"
+      "max_compaction_trigger_wakeup_seconds=43200;"
       "follower_refresh_catchup_period_ms=123;"
       "follower_catchup_retry_count=456;"
       "follower_catchup_retry_wait_ms=789;"
@@ -662,6 +663,7 @@ TEST_F(OptionsSettableTest, ColumnFamilyOptionsAllFieldsSettable) {
       "disallow_memtable_writes=true;"
       "ttl=60;"
       "periodic_compaction_seconds=3600;"
+      "read_triggered_compaction_threshold=0.5;"
       "sample_for_compression=0;"
       "enable_blob_files=true;"
       "min_blob_size=256;"

tools/db_bench_tool.cc

@@ -1133,6 +1133,15 @@ DEFINE_uint64(blob_compaction_readahead_size,
                   .blob_compaction_readahead_size,
               "[Integrated BlobDB] Compaction readahead for blob files.");
 
+DEFINE_double(read_triggered_compaction_threshold,
+              ROCKSDB_NAMESPACE::AdvancedColumnFamilyOptions()
+                  .read_triggered_compaction_threshold,
+              "Threshold for read-triggered compaction. An SST file is marked "
+              "for compaction when its sampled read frequency "
+              "(sampled_reads / file_size) exceeds this value. Collapsible "
+              "reads (NotFound, Merge, Delete results) are sampled. "
+              "0 disables the feature.");
+
 DEFINE_int32(
     blob_file_starting_level,
     ROCKSDB_NAMESPACE::AdvancedColumnFamilyOptions().blob_file_starting_level,
@@ -1769,6 +1778,10 @@ DEFINE_bool(dump_malloc_stats, true, "Dump malloc stats in LOG ");
 DEFINE_uint64(stats_dump_period_sec,
               ROCKSDB_NAMESPACE::Options().stats_dump_period_sec,
               "Gap between printing stats to log in seconds");
+DEFINE_uint64(
+    max_compaction_trigger_wakeup_seconds,
+    ROCKSDB_NAMESPACE::Options().max_compaction_trigger_wakeup_seconds,
+    "Maximum interval in seconds between periodic compaction trigger checks.");
 DEFINE_uint64(stats_persist_period_sec,
               ROCKSDB_NAMESPACE::Options().stats_persist_period_sec,
               "Gap between persisting stats in seconds");
@@ -4415,6 +4428,8 @@ class Benchmark {
     options.dump_malloc_stats = FLAGS_dump_malloc_stats;
     options.stats_dump_period_sec =
         static_cast<unsigned int>(FLAGS_stats_dump_period_sec);
+    options.max_compaction_trigger_wakeup_seconds =
+        FLAGS_max_compaction_trigger_wakeup_seconds;
     options.stats_persist_period_sec =
         static_cast<unsigned int>(FLAGS_stats_persist_period_sec);
     options.persist_stats_to_disk = FLAGS_persist_stats_to_disk;
@@ -4994,6 +5009,8 @@ class Benchmark {
     options.blob_compaction_readahead_size =
         FLAGS_blob_compaction_readahead_size;
     options.blob_file_starting_level = FLAGS_blob_file_starting_level;
+    options.read_triggered_compaction_threshold =
+        FLAGS_read_triggered_compaction_threshold;
 
     if (FLAGS_readonly && FLAGS_transaction_db) {
       fprintf(stderr, "Cannot use readonly flag with transaction_db\n");

tools/db_crashtest.py

@@ -266,6 +266,8 @@ default_params = {
     "use_get_entity": lambda: random.choice([0] * 7 + [1]),
     "use_multi_get_entity": lambda: random.choice([0] * 7 + [1]),
     "periodic_compaction_seconds": lambda: random.choice([0, 0, 1, 2, 10, 100, 1000]),
+    "max_compaction_trigger_wakeup_seconds": lambda: random.choice([43200, 600, 30]),
+    "read_triggered_compaction_threshold": lambda: random.choice([0.0, 0.001, 0.01]),
     "daily_offpeak_time_utc": lambda: random.choice(
         ["", "", "00:00-23:59", "04:00-08:00", "23:30-03:15"]
     ),
@@ -1304,6 +1306,11 @@ def finalize_and_sanitize(src_params):
     if dest_params.get("remote_compaction_worker_threads", 0) == 0:
         dest_params["allow_resumption_one_in"] = 0
 
+    # When read-triggered compaction is enabled, use a short periodic trigger
+    # interval so that the feature gets exercised on a quiet DB.
+    if dest_params.get("read_triggered_compaction_threshold", 0) > 0:
+        dest_params["max_compaction_trigger_wakeup_seconds"] = 20
+
     return dest_params
 
 

unreleased_history/new_features/read_triggered_compaction.md

@@ -0,0 +1 @@
+Added read-triggered compaction: a new column family option `read_triggered_compaction_threshold` (default 0, disabled) that marks SST files for compaction when their read frequency (`num_collapsible_entry_reads_sampled / file_size`) exceeds the threshold. This helps reduce read amplification for frequently-read ("hot") keys. A new DB option `max_compaction_trigger_wakeup_seconds` (default 43200s / 12 hours) controls the maximum interval for periodic compaction score re-evaluation, which is necessary for this feature to work on quiet (no-write) databases.