Refresh time-based compaction trigger after option changes

db/db_impl/db_impl.cc

@@ -1508,6 +1508,14 @@ Status DBImpl::StartPeriodicTaskScheduler() {
   return s;
 }
 
+Status DBImpl::RegisterTriggerCompactionTask() {
+  options_mutex_.AssertHeld();
+  return periodic_task_scheduler_.Register(
+      PeriodicTaskType::kTriggerCompaction,
+      periodic_task_functions_.at(PeriodicTaskType::kTriggerCompaction),
+      ComputeTriggerCompactionPeriod(), /*run_immediately=*/false);
+}
+
 Status DBImpl::RegisterRecordSeqnoTimeWorker() {
   options_mutex_.AssertHeld();
 
@@ -2005,6 +2013,8 @@ Status DBImpl::SetOptions(
       s = RegisterRecordSeqnoTimeWorker();
     }
   }
+  // Earlier CFs can have changed even if a later CF's SetOptions failed.
+  s.UpdateIfOk(RegisterTriggerCompactionTask());
 
   ROCKS_LOG_INFO(immutable_db_options_.info_log,
                  "SetOptions() on [%zu] column families, inputs:",
@@ -2209,6 +2219,9 @@ Status DBImpl::SetDBOptions(
       persist_options_status.PermitUncheckedError();
     }
   }
+  if (s.ok()) {
+    s = RegisterTriggerCompactionTask();
+  }
   ROCKS_LOG_INFO(immutable_db_options_.info_log, "SetDBOptions(), inputs:");
   for (const auto& o : options_map) {
     ROCKS_LOG_INFO(immutable_db_options_.info_log, "%s: %s\n", o.first.c_str(),
@@ -4508,6 +4521,7 @@ Status DBImpl::WrapUpCreateColumnFamilies(
   if (register_worker) {
     s.UpdateIfOk(RegisterRecordSeqnoTimeWorker());
   }
+  s.UpdateIfOk(RegisterTriggerCompactionTask());
   return s;
 }
 
@@ -4968,6 +4982,9 @@ Status DBImpl::DropColumnFamilyWithOptionsFile(
   if (preserve_info.IsEnabled()) {
     s = RegisterRecordSeqnoTimeWorker();
   }
+  if (cfd->IsDropped()) {
+    s.UpdateIfOk(RegisterTriggerCompactionTask());
+  }
 
   if (s.ok()) {
     // Note that here we erase the associated cf_info of the to-be-dropped

db/db_impl/db_impl.h

@@ -3136,11 +3136,14 @@ class DBImpl : public DB
 
   // Compute the repeat period for the kTriggerCompaction task, which ensures
   // compactions not dependent on writes (flushes) are eventually triggered when
-  // there are no writes (flushes). NOT thread safe; only called during DB open
-  // (StartPeriodicTaskScheduler). KNOWN LIMITATION: doesn't get updated with
-  // dynamic option updates. (Probably not worth the extra complexity.)
+  // there are no writes (flushes). Call only during DB open or while holding
+  // options_mutex_, which serializes changes to the options and CF set.
   uint64_t ComputeTriggerCompactionPeriod();
 
+  // Refresh the task after a change to its DB or CF options, or the CF set.
+  // Requires options_mutex_.
+  Status RegisterTriggerCompactionTask();
+
   // Cancel scheduled periodic tasks
   Status CancelPeriodicTaskScheduler();
 

db/periodic_task_scheduler_test.cc

@@ -98,6 +98,8 @@ TEST_F(PeriodicTaskSchedulerTest, Basic) {
   ASSERT_EQ(3, dump_st_counter);
   ASSERT_EQ(3, pst_st_counter);
   ASSERT_EQ(3, flush_info_log_counter);
+  const int trigger_compaction_count_before_disable =
+      trigger_compaction_counter;
 
   // Disable scheduler with SetOption
   ASSERT_OK(dbfull()->SetDBOptions(
@@ -111,6 +113,8 @@ TEST_F(PeriodicTaskSchedulerTest, Basic) {
   ASSERT_EQ(3, dump_st_counter);
   ASSERT_EQ(3, pst_st_counter);
   ASSERT_EQ(4, flush_info_log_counter);
+  ASSERT_EQ(trigger_compaction_count_before_disable,
+            trigger_compaction_counter);
 
   ASSERT_EQ(2u, scheduler.TEST_GetValidTaskNum());
 
@@ -127,19 +131,16 @@ TEST_F(PeriodicTaskSchedulerTest, Basic) {
   ASSERT_EQ(3, pst_st_counter);
   ASSERT_EQ(5, flush_info_log_counter);
 
-  // With the new ComputeTriggerCompactionPeriod() logic, since
-  // stats_dump_period_sec = stats_persist_period_sec = kPeriodSec, the trigger
-  // compaction period is also kPeriodSec. Unlike other periodic tasks,
-  // kTriggerCompaction uses run_immediately=false, so it doesn't fire on
-  // initial registration. It fires after each period elapses.
-  // By now we've had: initial 9s + 10s + 10s + 10s + 10s = 49s elapsed.
-  // kTriggerCompaction fires at t=10, 20, 30, 40 = 4 times.
-  ASSERT_EQ(4, trigger_compaction_counter);
+  // Disabling the stats tasks also lengthens the compaction trigger period.
+  // Re-enabling the dump task shortens it again, starting a new schedule.
+  ASSERT_GT(trigger_compaction_counter,
+            trigger_compaction_count_before_disable);
+  const int trigger_compaction_count = trigger_compaction_counter;
 
-  // Sleep one more period and verify it increments
+  // Sleep one more period and verify the new schedule keeps running.
   dbfull()->TEST_WaitForPeriodicTaskRun(
       [&] { mock_clock_->MockSleepForSeconds(static_cast<int>(kPeriodSec)); });
-  ASSERT_EQ(5, trigger_compaction_counter);
+  ASSERT_GT(trigger_compaction_counter, trigger_compaction_count);
 
   Close();
 }
@@ -322,6 +323,98 @@ TEST_F(PeriodicTaskSchedulerTest, TriggerCompactionPeriodComputation) {
   }
 }
 
+TEST_F(PeriodicTaskSchedulerTest, TriggerCompactionPeriodFollowsOptionChanges) {
+  Close();
+  uint64_t trigger_compaction_period = 0;
+  SyncPoint::GetInstance()->SetCallBack(
+      "PeriodicTaskScheduler::Register:TaskRegistered", [&](void* arg) {
+        auto* task_info =
+            static_cast<std::pair<PeriodicTaskType, uint64_t>*>(arg);
+        if (task_info->first == PeriodicTaskType::kTriggerCompaction) {
+          trigger_compaction_period = task_info->second;
+        }
+      });
+  SyncPoint::GetInstance()->EnableProcessing();
+
+  Options options;
+  options.create_if_missing = true;
+  options.env = mock_env_.get();
+  options.stats_dump_period_sec = 0;
+  options.stats_persist_period_sec = 0;
+  Reopen(options);
+  ASSERT_EQ(12 * 60 * 60, trigger_compaction_period);
+
+  ASSERT_OK(dbfull()->SetOptions({{"ttl", "100"}}));
+  ASSERT_EQ(20, trigger_compaction_period);
+  ASSERT_OK(dbfull()->SetOptions({{"ttl", "0"}}));
+  ASSERT_EQ(12 * 60 * 60, trigger_compaction_period);
+
+  ASSERT_OK(dbfull()->SetDBOptions(
+      {{"max_compaction_trigger_wakeup_seconds", "30"}}));
+  ASSERT_EQ(30, trigger_compaction_period);
+  ASSERT_OK(dbfull()->SetDBOptions({{"stats_dump_period_sec", "10"}}));
+  ASSERT_EQ(10, trigger_compaction_period);
+  ASSERT_OK(dbfull()->SetDBOptions({{"stats_dump_period_sec", "0"}}));
+  ASSERT_EQ(30, trigger_compaction_period);
+
+  ColumnFamilyOptions cf_options;
+  cf_options.ttl = 50;
+  ColumnFamilyHandle* cf_handle = nullptr;
+  ASSERT_OK(db_->CreateColumnFamily(cf_options, "ttl_cf", &cf_handle));
+  ASSERT_EQ(10, trigger_compaction_period);
+  ASSERT_OK(
+      db_->SetOptions(cf_handle, {{"periodic_compaction_seconds", "20"}}));
+  ASSERT_EQ(4, trigger_compaction_period);
+  ASSERT_OK(db_->SetOptions(cf_handle, {{"periodic_compaction_seconds", "0"}}));
+  ASSERT_EQ(10, trigger_compaction_period);
+  ASSERT_OK(db_->DropColumnFamily(cf_handle));
+  ASSERT_EQ(30, trigger_compaction_period);
+  delete cf_handle;
+  Close();
+}
+
+TEST_F(PeriodicTaskSchedulerTest, TriggerCompactionPeriodAfterPartialUpdate) {
+  Close();
+  uint64_t trigger_compaction_period = 0;
+  SyncPoint::GetInstance()->SetCallBack(
+      "PeriodicTaskScheduler::Register:TaskRegistered", [&](void* arg) {
+        auto* task_info =
+            static_cast<std::pair<PeriodicTaskType, uint64_t>*>(arg);
+        if (task_info->first == PeriodicTaskType::kTriggerCompaction) {
+          trigger_compaction_period = task_info->second;
+        }
+      });
+  SyncPoint::GetInstance()->EnableProcessing();
+
+  Options options;
+  options.create_if_missing = true;
+  options.env = mock_env_.get();
+  options.stats_dump_period_sec = 0;
+  options.stats_persist_period_sec = 0;
+  Reopen(options);
+  ASSERT_EQ(12 * 60 * 60, trigger_compaction_period);
+
+  ColumnFamilyHandle* cf_handle = nullptr;
+  ASSERT_OK(db_->CreateColumnFamily(options, "other_cf", &cf_handle));
+  std::unordered_map<ColumnFamilyHandle*,
+                     std::unordered_map<std::string, std::string>>
+      cf_options;
+  cf_options.emplace(db_->DefaultColumnFamily(),
+                     std::unordered_map<std::string, std::string>{});
+  cf_options.emplace(cf_handle, std::unordered_map<std::string, std::string>{});
+  // Assign by iteration order so the valid change is applied before the
+  // invalid one, regardless of how the handles hash on this platform.
+  auto it = cf_options.begin();
+  it->second = {{"ttl", "100"}};
+  (++it)->second = {{"unknown_option", "1"}};
+  ASSERT_NOK(db_->SetOptions(cf_options));
+  ASSERT_EQ(20, trigger_compaction_period);
+
+  ASSERT_OK(db_->DropColumnFamily(cf_handle));
+  delete cf_handle;
+  Close();
+}
+
 // Test that TriggerPeriodicCompaction() properly considers CFs for compaction
 // based on all time-based compaction options (not just
 // periodic_compaction_seconds). This is a regression test for the bug where

unreleased_history/bug_fixes/dynamic_time_based_compaction_trigger.md

@@ -0,0 +1 @@
+Fixed time-based compactions not waking up on idle DBs after their intervals are changed with SetOptions or SetDBOptions, or after column families are added or dropped.