database/rocksdb
1
0
Fork 0
mirror of https://github.com/facebook/rocksdb.git synced 2026-10-23 08:09:02 +03:00
Code Activity

Fix & improve compaction trigger on a "quiet" DB (#14396)

Browse Source 
Summary:
As a follow-up to https://github.com/facebook/rocksdb/issues/13736, allow a "quiet" DB to react much sooner to time-based compaction triggers. For details see DBImpl::ComputeTriggerCompactionPeriod() implementation.

Also based on review feedback, fixing a bug where only column families setting periodic compaction would be triggered, rather than any time-based compaction.

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

Test Plan: extended+added unit tests to cover much of the logic

Reviewed By: xingbowang

Differential Revision: D94626166

Pulled By: pdillinger

fbshipit-source-id: de9ca19e46bdba5d9715474efbc61805354d730d
This commit is contained in:
Peter Dillinger
2026-02-28 13:43:55 -08:00
committed by meta-codesync[bot]
parent 8707abfc11
commit da2c3c0ee6
4 changed files with 394 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

103
db/db_impl/db_impl.cc
View File

@@ -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());

7
db/db_impl/db_impl.h
View File

@@ -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();

11
db/periodic_task_scheduler.cc
View File

@@ -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<PeriodicTaskType, uint64_t> kDefaultPeriodSeconds = {
{PeriodicTaskType::kPersistStats, kInvalidPeriodSec},
{PeriodicTaskType::kFlushInfoLog, 10},
{PeriodicTaskType::kRecordSeqnoTime, kInvalidPeriodSec},
{PeriodicTaskType::kTriggerCompaction, 12 * 60 * 60} // 12 hours
{PeriodicTaskType::kTriggerCompaction, kInvalidPeriodSec},
};
static const std::map<PeriodicTaskType, std::string> 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<PeriodicTaskType, uint64_t> task_info{task_type,
repeat_period_seconds};
TEST_SYNC_POINT_CALLBACK("PeriodicTaskScheduler::Register:TaskRegistered",
&task_info);
}
#endif // NDEBUG
return Status::OK();
}

295
db/periodic_task_scheduler_test.cc
View File

@@ -6,6 +6,9 @@
#include "db/periodic_task_scheduler.h"
#include <set>
#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<int>(12 * 60 * 60));
});
ASSERT_EQ(1, trigger_compaction_counter);
dbfull()->TEST_WaitForPeriodicTaskRun([&] {
mock_clock_->MockSleepForSeconds(static_cast<int>(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<int>(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<std::pair<PeriodicTaskType, uint64_t>*>(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<ColumnFamilyDescriptor> cf_descs;
cf_descs.emplace_back(kDefaultColumnFamilyName, options);
cf_descs.emplace_back("cf2", *cf2_options);
std::vector<ColumnFamilyHandle*> 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<std::string> cfs_with_score_computed;
SyncPoint::GetInstance()->SetCallBack(
"DBImpl::TriggerPeriodicCompaction:BeforeComputeCompactionScore",
[&](void* arg) {
auto* cfd = static_cast<ColumnFamilyData*>(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<std::string> cf_names = {
kDefaultColumnFamilyName, "cf_periodic", "cf_ttl",
"cf_bottommost", "cf_fifo", "cf_none"};
std::vector<Options> 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();
}