diff --git a/db/db_impl/db_impl.cc b/db/db_impl/db_impl.cc index fea401477c..a144f11c53 100644 --- a/db/db_impl/db_impl.cc +++ b/db/db_impl/db_impl.cc @@ -784,7 +784,76 @@ void DBImpl::PrintStatistics() { } } +// Computes the minimum time-based compaction interval for a CF based on +// various options. +// Returns 0 if all time-based compaction options are disabled. +static uint64_t GetMinTimeBasedCompactionInterval( + const ColumnFamilyOptions& cf_opts) { + uint64_t min_interval = UINT64_MAX; + if (cf_opts.periodic_compaction_seconds > 0) { + min_interval = std::min(min_interval, cf_opts.periodic_compaction_seconds); + } + if (cf_opts.ttl > 0) { + min_interval = std::min(min_interval, cf_opts.ttl); + } + const auto& fifo_thresholds = + cf_opts.compaction_options_fifo.file_temperature_age_thresholds; + if (!fifo_thresholds.empty() && fifo_thresholds[0].age > 0) { + // Thresholds are in increasing order by age, so first is smallest + min_interval = std::min(min_interval, fifo_thresholds[0].age); + } + if (cf_opts.bottommost_file_compaction_delay > 0) { + // NOTE: 0 does not exactly mean "disabled" in this case but it does mean + // there's no time component to the relevant compaction picking. + min_interval = std::min(min_interval, + uint64_t{cf_opts.bottommost_file_compaction_delay}); + } + // Note: Assume sentinel values like UINT64_MAX - 1 (used by ttl and + // periodic_compaction_seconds) are like disabling, if they reach here + // unsanitized. + return min_interval > UINT64_MAX / 2 ? 0 : min_interval; +} + +uint64_t DBImpl::ComputeTriggerCompactionPeriod() { + // Start with a maximum period of every 12 hours. + uint64_t period_sec = 12 * 60 * 60; + + // 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 + // problem, as it should be less overhead than these. + if (mutable_db_options_.stats_dump_period_sec > 0) { + period_sec = std::min(period_sec, + (uint64_t)mutable_db_options_.stats_dump_period_sec); + } + if (mutable_db_options_.stats_persist_period_sec > 0) { + period_sec = std::min( + period_sec, (uint64_t)mutable_db_options_.stats_persist_period_sec); + } + + // Consider per-CF settings that can trigger compaction based on time. + uint64_t compaction_trigger_sec = UINT64_MAX; + for (auto cfd : *versions_->GetColumnFamilySet()) { + if (cfd->IsDropped()) { + continue; + } + uint64_t cf_min = + GetMinTimeBasedCompactionInterval(cfd->GetLatestCFOptions()); + if (cf_min > 0) { + compaction_trigger_sec = std::min(compaction_trigger_sec, cf_min); + } + } + // We might not align with those timings perfectly, so we tolerate only some + // proportional delay, up to 1/kTriggerDivisor ~= 20%. + constexpr uint64_t kTriggerDivisor = 5; + period_sec = std::min(period_sec, compaction_trigger_sec / kTriggerDivisor); + // But must be > 0 + period_sec = std::max(period_sec, uint64_t{1}); + + return period_sec; +} + Status DBImpl::StartPeriodicTaskScheduler() { + Status s; #ifndef NDEBUG // It only used by test to disable scheduler bool disable_scheduler = false; @@ -792,7 +861,7 @@ Status DBImpl::StartPeriodicTaskScheduler() { "DBImpl::StartPeriodicTaskScheduler:DisableScheduler", &disable_scheduler); if (disable_scheduler) { - return Status::OK(); + return s; } { @@ -803,7 +872,7 @@ Status DBImpl::StartPeriodicTaskScheduler() { #endif // !NDEBUG if (mutable_db_options_.stats_dump_period_sec > 0) { - Status s = periodic_task_scheduler_.Register( + s = periodic_task_scheduler_.Register( PeriodicTaskType::kDumpStats, periodic_task_functions_.at(PeriodicTaskType::kDumpStats), mutable_db_options_.stats_dump_period_sec, @@ -813,7 +882,7 @@ Status DBImpl::StartPeriodicTaskScheduler() { } } if (mutable_db_options_.stats_persist_period_sec > 0) { - Status s = periodic_task_scheduler_.Register( + s = periodic_task_scheduler_.Register( PeriodicTaskType::kPersistStats, periodic_task_functions_.at(PeriodicTaskType::kPersistStats), mutable_db_options_.stats_persist_period_sec, @@ -823,18 +892,19 @@ Status DBImpl::StartPeriodicTaskScheduler() { } } - Status s = periodic_task_scheduler_.Register( + s = periodic_task_scheduler_.Register( PeriodicTaskType::kFlushInfoLog, periodic_task_functions_.at(PeriodicTaskType::kFlushInfoLog), /*run_immediately=*/true); - - if (s.ok()) { - s = periodic_task_scheduler_.Register( - PeriodicTaskType::kTriggerCompaction, - periodic_task_functions_.at(PeriodicTaskType::kTriggerCompaction), - /*run_immediately=*/false); + if (!s.ok()) { + return s; } + s = periodic_task_scheduler_.Register( + PeriodicTaskType::kTriggerCompaction, + periodic_task_functions_.at(PeriodicTaskType::kTriggerCompaction), + ComputeTriggerCompactionPeriod(), /*run_immediately=*/false); + return s; } @@ -6916,8 +6986,17 @@ void DBImpl::TriggerPeriodicCompaction() { if (cfd->IsDropped()) { continue; } - if (cfd->GetLatestCFOptions().periodic_compaction_seconds && - !cfd->queued_for_compaction()) { + 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) { + TEST_SYNC_POINT_CALLBACK( + "DBImpl::TriggerPeriodicCompaction:BeforeComputeCompactionScore", + cfd); cfd->current()->storage_info()->ComputeCompactionScore( cfd->ioptions(), cfd->GetLatestMutableCFOptions(), cfd->GetFullHistoryTsLow()); diff --git a/db/db_impl/db_impl.h b/db/db_impl/db_impl.h index c3c432bec8..ded8bfda17 100644 --- a/db/db_impl/db_impl.h +++ b/db/db_impl/db_impl.h @@ -2506,6 +2506,13 @@ class DBImpl : public DB { // Schedule background tasks Status StartPeriodicTaskScheduler(); + // 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.) + uint64_t ComputeTriggerCompactionPeriod(); + // Cancel scheduled periodic tasks Status CancelPeriodicTaskScheduler(); diff --git a/db/periodic_task_scheduler.cc b/db/periodic_task_scheduler.cc index ee3f07b91e..848096d98e 100644 --- a/db/periodic_task_scheduler.cc +++ b/db/periodic_task_scheduler.cc @@ -7,6 +7,7 @@ #include "db/periodic_task_scheduler.h" #include "rocksdb/system_clock.h" +#include "test_util/sync_point.h" namespace ROCKSDB_NAMESPACE { @@ -26,7 +27,7 @@ static const std::map kDefaultPeriodSeconds = { {PeriodicTaskType::kPersistStats, kInvalidPeriodSec}, {PeriodicTaskType::kFlushInfoLog, 10}, {PeriodicTaskType::kRecordSeqnoTime, kInvalidPeriodSec}, - {PeriodicTaskType::kTriggerCompaction, 12 * 60 * 60} // 12 hours + {PeriodicTaskType::kTriggerCompaction, kInvalidPeriodSec}, }; static const std::map kPeriodicTaskTypeNames = { @@ -85,6 +86,14 @@ Status PeriodicTaskScheduler::Register(PeriodicTaskType task_type, if (!result.second) { return Status::Aborted("Failed to add periodic task"); } +#ifndef NDEBUG + { + std::pair task_info{task_type, + repeat_period_seconds}; + TEST_SYNC_POINT_CALLBACK("PeriodicTaskScheduler::Register:TaskRegistered", + &task_info); + } +#endif // NDEBUG return Status::OK(); } diff --git a/db/periodic_task_scheduler_test.cc b/db/periodic_task_scheduler_test.cc index 5575333b09..aed3f87dcb 100644 --- a/db/periodic_task_scheduler_test.cc +++ b/db/periodic_task_scheduler_test.cc @@ -6,6 +6,9 @@ #include "db/periodic_task_scheduler.h" +#include + +#include "db/column_family.h" #include "db/db_test_util.h" #include "env/composite_env_wrapper.h" #include "test_util/mock_time_env.h" @@ -124,15 +127,289 @@ TEST_F(PeriodicTaskSchedulerTest, Basic) { ASSERT_EQ(3, pst_st_counter); ASSERT_EQ(5, flush_info_log_counter); - ASSERT_EQ(0, trigger_compaction_counter); - dbfull()->TEST_WaitForPeriodicTaskRun([&] { - mock_clock_->MockSleepForSeconds(static_cast(12 * 60 * 60)); - }); - ASSERT_EQ(1, trigger_compaction_counter); - dbfull()->TEST_WaitForPeriodicTaskRun([&] { - mock_clock_->MockSleepForSeconds(static_cast(12 * 60 * 60)); - }); - ASSERT_EQ(2, trigger_compaction_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); + + // Sleep one more period and verify it increments + dbfull()->TEST_WaitForPeriodicTaskRun( + [&] { mock_clock_->MockSleepForSeconds(static_cast(kPeriodSec)); }); + ASSERT_EQ(5, trigger_compaction_counter); + + Close(); +} + +TEST_F(PeriodicTaskSchedulerTest, TriggerCompactionPeriodComputation) { + // Test that ComputeTriggerCompactionPeriod() returns the expected values + // for different configurations. + Close(); + + // Helper to capture the trigger compaction period on Reopen + uint64_t trigger_compaction_period = 0; + SyncPoint::GetInstance()->SetCallBack( + "PeriodicTaskScheduler::Register:TaskRegistered", [&](void* arg) { + auto* task_info = + static_cast*>(arg); + if (task_info->first == PeriodicTaskType::kTriggerCompaction) { + trigger_compaction_period = task_info->second; + } + }); + SyncPoint::GetInstance()->EnableProcessing(); + + // Helper to test the trigger compaction period with optional second CF + auto test_period = [&](Options& options, uint64_t expected_period, + const ColumnFamilyOptions* cf2_options = nullptr) { + options.create_if_missing = true; + options.env = mock_env_.get(); + trigger_compaction_period = 0; + + if (cf2_options == nullptr) { + Reopen(options); + } else { + // Open with two column families + // First destroy DB to start fresh + ASSERT_OK(DestroyDB(dbname_, options)); + // Open and create cf2 + ASSERT_OK(TryReopen(options)); + CreateColumnFamilies({"cf2"}, options); + Close(); + // Now reopen with both CFs + std::vector cf_descs; + cf_descs.emplace_back(kDefaultColumnFamilyName, options); + cf_descs.emplace_back("cf2", *cf2_options); + std::vector handles; + ASSERT_OK(DB::Open(options, dbname_, cf_descs, &handles, &db_)); + for (auto* h : handles) { + delete h; + } + } + ASSERT_EQ(expected_period, trigger_compaction_period); + Close(); + // Destroy DB to clean up for next test case + ASSERT_OK(DestroyDB(dbname_, options)); + }; + + // Case 1: 12-hour cap when stats_dump_period is longer + { + Options options; + options.stats_dump_period_sec = 24 * 60 * 60; // 24 hours + options.stats_persist_period_sec = 0; + test_period(options, 12 * 60 * 60); // Expect 12 hours (cap) + } + + // Case 2: stats_dump_period_sec sets the period (when smaller than 12 hours) + { + Options options; + options.stats_dump_period_sec = 3600; // 1 hour + options.stats_persist_period_sec = 0; + test_period(options, 3600); + } + + // Case 3: stats_persist_period_sec sets the period + { + Options options; + options.stats_dump_period_sec = 0; + options.stats_persist_period_sec = 1800; // 30 minutes + test_period(options, 1800); + } + + // Case 4: periodic_compaction_seconds / kTriggerDivisor + { + Options options; + options.stats_dump_period_sec = 24 * 60 * 60; + options.stats_persist_period_sec = 0; + options.periodic_compaction_seconds = 500; // 500 / 5 = 100 + test_period(options, 100); + } + + // Case 5: ttl / kTriggerDivisor + { + Options options; + options.stats_dump_period_sec = 24 * 60 * 60; + options.stats_persist_period_sec = 0; + options.ttl = 600; // 600 / 5 = 120 + test_period(options, 120); + } + + // Case 6: bottommost_file_compaction_delay / kTriggerDivisor + { + Options options; + options.stats_dump_period_sec = 24 * 60 * 60; + options.stats_persist_period_sec = 0; + options.bottommost_file_compaction_delay = 250; // 250 / 5 = 50 + test_period(options, 50); + } + + // Case 7: file_temperature_age_thresholds / kTriggerDivisor (FIFO) + { + Options options; + options.stats_dump_period_sec = 24 * 60 * 60; + options.stats_persist_period_sec = 0; + options.compaction_style = kCompactionStyleFIFO; + options.num_levels = 1; // Required for file_temperature_age_thresholds + options.compaction_options_fifo.file_temperature_age_thresholds = { + {Temperature::kWarm, 1000}}; // 1000 / 5 = 200 + // FIFO requires max_table_files_size to be set + options.compaction_options_fifo.max_table_files_size = 1024 * 1024 * 1024; + test_period(options, 200); + } + + // Case 8: Minimum of multiple CF options is used + { + Options options; + options.stats_dump_period_sec = 24 * 60 * 60; + options.stats_persist_period_sec = 0; + options.periodic_compaction_seconds = 1000; // 1000 / 5 = 200 + + ColumnFamilyOptions cf2_opts; + cf2_opts.ttl = 500; // 500 / 5 = 100 (smaller, should be used) + + test_period(options, 100, &cf2_opts); + } + + // Case 9: Second CF has larger value, first CF's value is used + { + Options options; + options.stats_dump_period_sec = 24 * 60 * 60; + options.stats_persist_period_sec = 0; + options.ttl = 300; // 300 / 5 = 60 + + ColumnFamilyOptions cf2_opts; + cf2_opts.periodic_compaction_seconds = 1000; // 1000 / 5 = 200 (larger) + + test_period(options, 60, &cf2_opts); + } + + // Case 10: Minimum is 1 second (never 0) + { + Options options; + options.stats_dump_period_sec = 0; + options.stats_persist_period_sec = 0; + options.ttl = 3; // 3 / 5 = 0, but clamped to 1 + test_period(options, 1); + } +} + +// 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 +// only periodic_compaction_seconds was checked, causing CFs with ttl, +// bottommost_file_compaction_delay, or file_temperature_age_thresholds to not +// be considered. +// +// NOTE: This test uses a separate test class without mock time because the +// PeriodicTaskSchedulerTest fixture's mock clock setup does not integrate +// properly with opening the DB via DB::Open() with multiple column families +// having different options. +class TriggerCompactionTest : public DBTestBase { + public: + TriggerCompactionTest() + : DBTestBase("trigger_compaction_test", /*env_do_fsync=*/true) {} +}; + +TEST_F(TriggerCompactionTest, QueuesAllTimeBasedOptions) { + Close(); + + // Track which CFs get their compaction score computed during + // TriggerPeriodicCompaction + std::set cfs_with_score_computed; + SyncPoint::GetInstance()->SetCallBack( + "DBImpl::TriggerPeriodicCompaction:BeforeComputeCompactionScore", + [&](void* arg) { + auto* cfd = static_cast(arg); + cfs_with_score_computed.insert(cfd->GetName()); + }); + SyncPoint::GetInstance()->EnableProcessing(); + + // Use small period so the test completes quickly with real time + constexpr unsigned int kPeriodSec = 1; + + // Default CF: no time-based compaction options + Options options; + options.create_if_missing = true; + options.stats_dump_period_sec = kPeriodSec; + options.stats_persist_period_sec = 0; + options.periodic_compaction_seconds = 0; + options.ttl = 0; + options.bottommost_file_compaction_delay = 0; + + // Create the DB with default CF first + DestroyAndReopen(options); + + // 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); + Close(); + + // Now reopen with specific options for each CF + std::vector cf_names = { + kDefaultColumnFamilyName, "cf_periodic", "cf_ttl", + "cf_bottommost", "cf_fifo", "cf_none"}; + std::vector cf_options; + + // default: no time-based options + cf_options.push_back(options); + + // cf_periodic: periodic_compaction_seconds + Options opt_periodic = options; + opt_periodic.periodic_compaction_seconds = 100; + cf_options.push_back(opt_periodic); + + // cf_ttl: ttl + Options opt_ttl = options; + opt_ttl.ttl = 100; + cf_options.push_back(opt_ttl); + + // cf_bottommost: bottommost_file_compaction_delay + Options opt_bottommost = options; + opt_bottommost.bottommost_file_compaction_delay = 100; + cf_options.push_back(opt_bottommost); + + // cf_fifo: file_temperature_age_thresholds (requires FIFO compaction) + Options opt_fifo = options; + opt_fifo.compaction_style = kCompactionStyleFIFO; + opt_fifo.num_levels = 1; + opt_fifo.compaction_options_fifo.max_table_files_size = 1024 * 1024 * 1024; + opt_fifo.compaction_options_fifo.file_temperature_age_thresholds = { + {Temperature::kWarm, 100}}; + cf_options.push_back(opt_fifo); + + // cf_none: explicitly no time-based options + Options opt_none = options; + opt_none.periodic_compaction_seconds = 0; + opt_none.ttl = 0; + opt_none.bottommost_file_compaction_delay = 0; + cf_options.push_back(opt_none); + + ReopenWithColumnFamilies(cf_names, cf_options); + + // Wait for TriggerPeriodicCompaction to run using TEST_WaitForPeriodicTaskRun + // with a real sleep callback to advance real time + dbfull()->TEST_WaitForPeriodicTaskRun( + [&] { Env::Default()->SleepForMicroseconds(kPeriodSec * 1000000); }); + + // Verify CFs with time-based options had their compaction score computed + EXPECT_GT(cfs_with_score_computed.count("cf_periodic"), 0u) + << "Expected cf_periodic to have compaction score computed"; + EXPECT_GT(cfs_with_score_computed.count("cf_ttl"), 0u) + << "Expected cf_ttl to have compaction score computed"; + EXPECT_GT(cfs_with_score_computed.count("cf_bottommost"), 0u) + << "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"; + + // CFs without time-based options should NOT have score computed + EXPECT_EQ(cfs_with_score_computed.count("default"), 0u) + << "Expected default CF to NOT have compaction score computed"; + EXPECT_EQ(cfs_with_score_computed.count("cf_none"), 0u) + << "Expected cf_none to NOT have compaction score computed"; Close(); }