diff --git a/db/c.cc b/db/c.cc index cd6c995de6..66fcae824a 100644 --- a/db/c.cc +++ b/db/c.cc @@ -5062,6 +5062,16 @@ unsigned char rocksdb_options_get_use_direct_io_for_flush_and_compaction( return opt->rep.use_direct_io_for_flush_and_compaction; } +void rocksdb_options_set_use_direct_io_for_compaction_reads( + rocksdb_options_t* opt, unsigned char v) { + opt->rep.use_direct_io_for_compaction_reads = v; +} + +unsigned char rocksdb_options_get_use_direct_io_for_compaction_reads( + rocksdb_options_t* opt) { + return opt->rep.use_direct_io_for_compaction_reads; +} + void rocksdb_options_set_allow_mmap_reads(rocksdb_options_t* opt, unsigned char v) { opt->rep.allow_mmap_reads = v; diff --git a/db/c_test.c b/db/c_test.c index a2258d5cb3..b6f6457735 100644 --- a/db/c_test.c +++ b/db/c_test.c @@ -2766,6 +2766,10 @@ int main(int argc, char** argv) { CheckCondition( 1 == rocksdb_options_get_use_direct_io_for_flush_and_compaction(o)); + rocksdb_options_set_use_direct_io_for_compaction_reads(o, 1); + CheckCondition(1 == + rocksdb_options_get_use_direct_io_for_compaction_reads(o)); + rocksdb_options_set_is_fd_close_on_exec(o, 1); CheckCondition(1 == rocksdb_options_get_is_fd_close_on_exec(o)); @@ -2993,6 +2997,8 @@ int main(int argc, char** argv) { CheckCondition(1 == rocksdb_options_get_use_direct_reads(copy)); CheckCondition( 1 == rocksdb_options_get_use_direct_io_for_flush_and_compaction(copy)); + CheckCondition( + 1 == rocksdb_options_get_use_direct_io_for_compaction_reads(copy)); CheckCondition(1 == rocksdb_options_get_is_fd_close_on_exec(copy)); CheckCondition(18 == rocksdb_options_get_stats_dump_period_sec(copy)); CheckCondition(5 == rocksdb_options_get_stats_persist_period_sec(copy)); @@ -3271,6 +3277,12 @@ int main(int argc, char** argv) { CheckCondition( 1 == rocksdb_options_get_use_direct_io_for_flush_and_compaction(o)); + rocksdb_options_set_use_direct_io_for_compaction_reads(copy, 0); + CheckCondition( + 0 == rocksdb_options_get_use_direct_io_for_compaction_reads(copy)); + CheckCondition(1 == + rocksdb_options_get_use_direct_io_for_compaction_reads(o)); + rocksdb_options_set_is_fd_close_on_exec(copy, 0); CheckCondition(0 == rocksdb_options_get_is_fd_close_on_exec(copy)); CheckCondition(1 == rocksdb_options_get_is_fd_close_on_exec(o)); diff --git a/db/compaction/compaction_job.cc b/db/compaction/compaction_job.cc index aa23fcd119..4f0d7604ed 100644 --- a/db/compaction/compaction_job.cc +++ b/db/compaction/compaction_job.cc @@ -173,6 +173,7 @@ CompactionJob::CompactionJob( fs_(db_options.fs, io_tracer), file_options_for_read_( fs_->OptimizeForCompactionTableRead(file_options, db_options_)), + file_options_for_compaction_input_read_(file_options_for_read_), versions_(versions), shutting_down_(shutting_down), manual_compaction_canceled_(manual_compaction_canceled), @@ -203,6 +204,11 @@ CompactionJob::CompactionJob( assert(job_context); assert(job_context->snapshot_context_initialized); + if (db_options_.use_direct_io_for_compaction_reads && + !db_options_.use_direct_reads) { + file_options_for_compaction_input_read_.use_direct_reads = true; + } + const auto* cfd = compact_->compaction->column_family_data(); ThreadStatusUtil::SetEnableTracking(db_options_.enable_thread_tracking); ThreadStatusUtil::SetColumnFamily(cfd); @@ -1536,10 +1542,20 @@ InternalIterator* CompactionJob::CreateInputIterator( // Although the v2 aggregator is what the level iterator(s) know about, // the AddTombstones calls will be propagated down to the v1 aggregator. + const bool open_ephemeral_table_reader = + db_options_.use_direct_io_for_compaction_reads && + !db_options_.use_direct_reads; + FileOptions& input_file_options = + open_ephemeral_table_reader ? file_options_for_compaction_input_read_ + : file_options_for_read_; + TEST_SYNC_POINT_CALLBACK( + "CompactionJob::CreateInputIterator:InputFileOptions", + &input_file_options); iterators.raw_input = std::unique_ptr(versions_->MakeInputIterator( read_options, sub_compact->compaction, sub_compact->RangeDelAgg(), - file_options_for_read_, boundaries.start, boundaries.end)); + input_file_options, boundaries.start, boundaries.end, + open_ephemeral_table_reader)); InternalIterator* input = iterators.raw_input.get(); if (boundaries.start.has_value() || boundaries.end.has_value()) { diff --git a/db/compaction/compaction_job.h b/db/compaction/compaction_job.h index 21486f8953..140d15b01f 100644 --- a/db/compaction/compaction_job.h +++ b/db/compaction/compaction_job.h @@ -462,6 +462,8 @@ class CompactionJob { FileSystemPtr fs_; // env_option optimized for compaction table reads FileOptions file_options_for_read_; + // file_options_for_read_ plus compaction-input-only overrides. + FileOptions file_options_for_compaction_input_read_; VersionSet* versions_; const std::atomic* shutting_down_; const std::atomic& manual_compaction_canceled_; diff --git a/db/db_compaction_test.cc b/db/db_compaction_test.cc index 6053340a83..a6aba0b82f 100644 --- a/db/db_compaction_test.cc +++ b/db/db_compaction_test.cc @@ -8,11 +8,13 @@ // found in the LICENSE file. See the AUTHORS file for names of contributors. #include +#include #include "compaction/compaction_picker_universal.h" #include "db/blob/blob_index.h" #include "db/db_test_util.h" #include "db/dbformat.h" +#include "db/table_cache.h" #include "env/mock_env.h" #include "file/filename.h" #include "port/port.h" @@ -6651,6 +6653,572 @@ TEST_P(DBCompactionDirectIOTest, DirectIO) { INSTANTIATE_TEST_CASE_P(DBCompactionDirectIOTest, DBCompactionDirectIOTest, testing::Bool()); +// With use_direct_io_for_compaction_reads OFF, compaction reads must stay +// buffered: neither the compaction-input FileOptions nor a kernel O_DIRECT +// open should fire. Runs on every platform (the sync points just don't fire +// where O_DIRECT isn't reachable). Pairs with +// UseDirectIoForCompactionReadsEndToEnd for the on case. +TEST_F(DBCompactionTest, UseDirectIoForCompactionReadsOffStaysBuffered) { + Options options = CurrentOptions(); + Destroy(options); + options.create_if_missing = true; + options.disable_auto_compactions = true; + options.use_direct_reads = false; + options.use_direct_io_for_compaction_reads = false; + options.use_direct_io_for_flush_and_compaction = false; + + std::atomic observed_direct_compaction_read{false}; + std::atomic observed_callbacks{0}; + std::atomic observed_odirect_opens{0}; + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack( + "CompactionJob::CreateInputIterator:InputFileOptions", [&](void* arg) { + const auto* fo = static_cast(arg); + observed_callbacks.fetch_add(1, std::memory_order_relaxed); + if (fo->use_direct_reads) { + observed_direct_compaction_read.store(true, + std::memory_order_relaxed); + } + }); + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack( + "NewRandomAccessFile:O_DIRECT", [&](void* /*arg*/) { + observed_odirect_opens.fetch_add(1, std::memory_order_relaxed); + }); + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->EnableProcessing(); + + ASSERT_OK(TryReopen(options)); + + const std::string value(4096, 'v'); + for (int i = 0; i < 64; ++i) { + ASSERT_OK(Put(Key(i), value)); + } + ASSERT_OK(Flush()); + for (int i = 0; i < 64; ++i) { + ASSERT_OK(Put(Key(i), value)); + } + ASSERT_OK(Flush()); + + ASSERT_OK(dbfull()->CompactRange(CompactRangeOptions(), nullptr, nullptr)); + ASSERT_OK(dbfull()->TEST_WaitForCompact()); + + ASSERT_GT(observed_callbacks.load(), 0); + ASSERT_FALSE(observed_direct_compaction_read.load()); + ASSERT_EQ(0, observed_odirect_opens.load()); + + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->DisableProcessing(); + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->ClearAllCallBacks(); + Destroy(options); +} + +TEST_F(DBCompactionTest, + UseDirectIoForCompactionReadsUsesBoundedEphemeralReaders) { + auto fs = std::make_shared(Env::Default()->GetSystemClock(), + /*supports_direct_io=*/true); + std::unique_ptr mock_env = NewCompositeEnv(fs); + + Options options = CurrentOptions(); + options.env = mock_env.get(); + Destroy(options); + options.create_if_missing = true; + options.disable_auto_compactions = true; + options.use_direct_reads = false; + options.use_direct_io_for_compaction_reads = true; + options.use_direct_io_for_flush_and_compaction = false; + options.max_subcompactions = 3; + options.statistics = CreateDBStatistics(); + + BlockBasedTableOptions table_options; + table_options.block_cache = NewLRUCache(64 << 20); + table_options.cache_index_and_filter_blocks = true; + table_options.pin_l0_filter_and_index_blocks_in_cache = true; + table_options.filter_policy.reset(NewBloomFilterPolicy(10)); + options.table_factory.reset(NewBlockBasedTableFactory(table_options)); + + std::atomic fresh_reader_opens{0}; + std::atomic malformed_fresh_reader_options{0}; + std::atomic avoid_shared_metadata_cache_opens{0}; + std::atomic shared_metadata_cache_uses{0}; + std::atomic input_iterator_file_bound{0}; + + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack( + "TableCache::FindTable:FreshTableReader", [&](void* arg) { + const auto* open_options = static_cast(arg); + fresh_reader_opens.fetch_add(1, std::memory_order_relaxed); + if (!open_options->open_ephemeral_table_reader || + !open_options->avoid_shared_metadata_cache || + !open_options->skip_filters) { + malformed_fresh_reader_options.fetch_add(1, + std::memory_order_relaxed); + } + }); + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack( + "BlockBasedTable::PrefetchIndexAndFilterBlocks:SharedMetadataCache", + [&](void* arg) { + const auto* context = static_cast*>(arg); + const bool avoid_shared_metadata_cache = context->first; + const bool use_cache = context->second; + if (avoid_shared_metadata_cache) { + avoid_shared_metadata_cache_opens.fetch_add( + 1, std::memory_order_relaxed); + if (use_cache) { + shared_metadata_cache_uses.fetch_add(1, std::memory_order_relaxed); + } + } + }); + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack( + "VersionSet::MakeInputIterator:NewCompactionMergingIterator", + [&](void* arg) { + input_iterator_file_bound.fetch_add(*static_cast(arg), + std::memory_order_relaxed); + }); + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->EnableProcessing(); + + ASSERT_OK(TryReopen(options)); + + const std::string value(4096, 'v'); + for (int i = 0; i < 64; ++i) { + ASSERT_OK(Put(Key(i), value)); + } + ASSERT_OK(Flush()); + for (int i = 0; i < 64; ++i) { + ASSERT_OK(Put(Key(i), value)); + } + ASSERT_OK(Flush()); + + ASSERT_OK(dbfull()->CompactRange(CompactRangeOptions(), nullptr, nullptr)); + ASSERT_OK(dbfull()->TEST_WaitForCompact()); + + ASSERT_GT(fresh_reader_opens.load(), 0); + EXPECT_EQ(0, malformed_fresh_reader_options.load()); + ASSERT_GT(avoid_shared_metadata_cache_opens.load(), 0); + EXPECT_EQ(0, shared_metadata_cache_uses.load()); + ASSERT_GT(input_iterator_file_bound.load(), 0); + EXPECT_LE(static_cast(fresh_reader_opens.load()), + input_iterator_file_bound.load()); + + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->DisableProcessing(); + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->ClearAllCallBacks(); + Destroy(options); +} + +// End-to-end check that use_direct_io_for_compaction_reads opens compaction +// inputs with O_DIRECT while use_direct_reads stays off (user reads buffered). +// The NewRandomAccessFile:O_DIRECT sync point in env/fs_posix.cc fires once +// per fresh open with the O_DIRECT flag, so this proves the kernel path, not +// just the FileOptions. Only runs on platforms that take the O_DIRECT path. +#if !defined(OS_MACOSX) && !defined(OS_OPENBSD) && !defined(OS_SOLARIS) && \ + !defined(OS_WIN) +TEST_F(DBCompactionTest, UseDirectIoForCompactionReadsEndToEnd) { + if (!IsDirectIOSupported()) { + ROCKSDB_GTEST_BYPASS("Direct IO not supported"); + return; + } + + Options options = CurrentOptions(); + Destroy(options); + options.create_if_missing = true; + options.disable_auto_compactions = true; + // User reads stay buffered, compaction reads should switch to O_DIRECT. + options.use_direct_reads = false; + options.use_direct_io_for_compaction_reads = true; + // Isolate the read-side change; leave the compaction write path buffered. + options.use_direct_io_for_flush_and_compaction = false; + + // Sync-point callbacks fire on compaction threads while the test thread + // reads these counters, so use atomics to avoid a data race. + std::atomic observed_run_starts{0}; + std::atomic observed_odirect_opens{0}; + std::atomic observed_direct_compaction_read{false}; + std::atomic observed_callbacks{0}; + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->LoadDependency({}); + // Plumbing-level probe: the compaction-input FileOptions should carry + // use_direct_reads = true when the new flag is enabled. + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack( + "CompactionJob::CreateInputIterator:InputFileOptions", [&](void* arg) { + const auto* fo = static_cast(arg); + observed_callbacks.fetch_add(1, std::memory_order_relaxed); + if (fo != nullptr && fo->use_direct_reads) { + observed_direct_compaction_read.store(true, + std::memory_order_relaxed); + } + }); + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack( + "CompactionJob::Run():Start", [&](void* /*arg*/) { + observed_run_starts.fetch_add(1, std::memory_order_relaxed); + }); + // Kernel-level probe: this fires only when open() is issued with O_DIRECT, + // proving we change the actual cache mode, not just the FileOptions struct. + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack( + "NewRandomAccessFile:O_DIRECT", [&](void* /*arg*/) { + observed_odirect_opens.fetch_add(1, std::memory_order_relaxed); + }); + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->EnableProcessing(); + + Status s = TryReopen(options); + if (s.IsNotSupported() || s.IsInvalidArgument()) { + ROCKSDB_GTEST_BYPASS( + "Direct IO reads not supported in this test environment"); + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->DisableProcessing(); + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->ClearAllCallBacks(); + return; + } + ASSERT_OK(s); + + // Produce two L0 files with OVERLAPPING key ranges so that CompactRange has + // actual merge work to do (otherwise RocksDB performs a trivial file move + // and never constructs a CompactionJob). + const std::string value(4096, 'v'); + for (int i = 0; i < 64; ++i) { + ASSERT_OK(Put(Key(i), value)); + } + ASSERT_OK(Flush()); + for (int i = 0; i < 64; ++i) { + ASSERT_OK(Put(Key(i), value)); + } + ASSERT_OK(Flush()); + + // User reads should still go through the buffered path. Confirm that the + // option does not silently flip use_direct_reads for user reads. + for (int i = 0; i < 8; ++i) { + std::string actual; + ASSERT_OK(db_->Get(ReadOptions(), Key(i), &actual)); + ASSERT_EQ(value, actual); + } + + ASSERT_OK(dbfull()->CompactRange(CompactRangeOptions(), nullptr, nullptr)); + // Wait for compaction to complete and CompactionJob to be constructed. + ASSERT_OK(dbfull()->TEST_WaitForCompact()); + + // Confirm the compaction actually ran; otherwise the missing sync-point hits + // would be a test-setup problem, not an option regression. + ASSERT_GT(observed_run_starts.load(), 0) + << "CompactionJob::Run():Start never fired; CompactRange did not " + "schedule a compaction."; + ASSERT_GT(observed_callbacks.load(), 0); + ASSERT_TRUE(observed_direct_compaction_read.load()); + // At least one compaction-input open went through O_DIRECT. Without the + // TableCache bypass this would be zero, since compaction would reuse the + // buffered handles cached for user reads. + EXPECT_GT(observed_odirect_opens.load(), 0) + << "no compaction-input opens went through O_DIRECT; " + "observed_odirect_opens=" + << observed_odirect_opens.load(); + + // Quick sanity sweep after compaction to confirm data is intact. + for (int i = 0; i < 64; ++i) { + std::string actual; + ASSERT_OK(db_->Get(ReadOptions(), Key(i), &actual)); + ASSERT_EQ(value, actual); + } + + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->DisableProcessing(); + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->ClearAllCallBacks(); + Destroy(options); +} + +// Exercise the LevelIterator bypass path (L1+ compactions) with range +// tombstones, where the ephemeral TableReader's lifetime is coupled to the +// range_tombstone_iter the file iterator returns. The end-to-end test above +// only makes two L0 files, which take the direct NewIterator path and never +// hit LevelIterator. Here we build L1/L2 with tombstones and compact L1->L2 so +// LevelIterator::NewFileIterator drives the bypass; a wrong reader/iterator +// lifetime would crash or trip sanitizers as LevelIterator switches files. +// Correctness is checked by computing the expected state of every key and +// asserting Get() matches: wave-2 puts beat wave-1, and each key's most-recent +// covering tombstone shadows older puts. +TEST_F(DBCompactionTest, + UseDirectIoForCompactionReadsLevelIteratorWithTombstones) { + if (!IsDirectIOSupported()) { + ROCKSDB_GTEST_BYPASS("Direct IO not supported"); + return; + } + + Options options = CurrentOptions(); + Destroy(options); + options.create_if_missing = true; + options.disable_auto_compactions = true; + options.use_direct_reads = false; + options.use_direct_io_for_compaction_reads = true; + options.use_direct_io_for_flush_and_compaction = false; + // Small files / small level base so we can pack data into L1 and L2 with + // a few flushes and CompactRange calls instead of needing millions of keys. + options.write_buffer_size = 64 * 1024; + options.target_file_size_base = 64 * 1024; + options.max_bytes_for_level_base = 256 * 1024; + options.level0_file_num_compaction_trigger = 100; // never auto-trigger + + std::atomic observed_odirect_opens{0}; + std::atomic observed_run_starts{0}; + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack( + "NewRandomAccessFile:O_DIRECT", [&](void* /*arg*/) { + observed_odirect_opens.fetch_add(1, std::memory_order_relaxed); + }); + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack( + "CompactionJob::Run():Start", [&](void* /*arg*/) { + observed_run_starts.fetch_add(1, std::memory_order_relaxed); + }); + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->EnableProcessing(); + + Status s = TryReopen(options); + if (s.IsNotSupported() || s.IsInvalidArgument()) { + ROCKSDB_GTEST_BYPASS( + "Direct IO reads not supported in this test environment"); + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->DisableProcessing(); + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->ClearAllCallBacks(); + return; + } + ASSERT_OK(s); + + // Use distinguishable values per wave so we can verify which wave's put + // won the merge for each key, not just "some put won". + const std::string wave1_value(1024, '1'); + const std::string wave2_value(1024, '2'); + + auto write_batch = [&](int begin, int end, const std::string& value, + bool with_range_tombstone) { + for (int i = begin; i < end; ++i) { + ASSERT_OK(Put(Key(i), value)); + } + if (with_range_tombstone) { + // Drop a slice in the middle of the just-written range. The + // DeleteRange follows the puts in this batch, so its sequence number + // is higher and it shadows the puts on [del_lo, del_hi) within this + // SST. + ASSERT_OK(db_->DeleteRange(WriteOptions(), db_->DefaultColumnFamily(), + Key(begin + (end - begin) / 4), + Key(begin + 3 * (end - begin) / 4))); + } + ASSERT_OK(Flush()); + }; + + // Wave 1: four flushed SSTs covering [0, 800), each with a tombstone + // covering the middle half of its own range. Then compact down so the + // next phase exercises LevelIterator over L1+ files. + constexpr int kWave1Begin = 0; + constexpr int kWave1End = 800; + constexpr int kWave1BatchSize = 200; + for (int batch_begin = kWave1Begin; batch_begin < kWave1End; + batch_begin += kWave1BatchSize) { + write_batch(batch_begin, batch_begin + kWave1BatchSize, wave1_value, + /*with_range_tombstone=*/true); + } + ASSERT_OK(dbfull()->CompactRange(CompactRangeOptions(), nullptr, nullptr)); + ASSERT_OK(dbfull()->TEST_WaitForCompact()); + + // Wave 2: two more flushed SSTs at L0 that overlap wave 1, each with their + // own range tombstone. The puts in wave 2 have higher seqno than wave 1, + // so they win when not tombstoned. + struct Wave2Range { + int begin; + int end; + }; + const Wave2Range wave2_ranges[] = {{50, 250}, {350, 550}}; + for (const auto& r : wave2_ranges) { + write_batch(r.begin, r.end, wave2_value, /*with_range_tombstone=*/true); + } + + const int run_starts_before = observed_run_starts.load(); + const int odirect_before = observed_odirect_opens.load(); + + // Compact everything together, forcing a LevelIterator over the lower-level + // files on the bypass path. Wrong ephemeral reader / tombstone-iter + // lifetimes should trip sanitizers here. + ASSERT_OK(dbfull()->CompactRange(CompactRangeOptions(), nullptr, nullptr)); + ASSERT_OK(dbfull()->TEST_WaitForCompact()); + + ASSERT_GT(observed_run_starts.load(), run_starts_before) + << "expected at least one compaction to run during the L1+ phase"; + EXPECT_GT(observed_odirect_opens.load(), odirect_before) + << "no compaction-input opens went through O_DIRECT during L1+ " + "compaction; LevelIterator bypass path may be broken"; + + // Compute the precise expected state per key from the test parameters. + // For each k in the wave-1 range: + // - If k is inside a wave-2 batch: wave-2 wins. NotFound iff k is in + // that batch's tombstone range; otherwise present with wave2_value. + // - Else: wave 1 wins. NotFound iff k is in its batch's tombstone + // range; otherwise present with wave1_value. + enum class Expectation { kAbsent, kWave1Value, kWave2Value }; + auto classify = [&](int k) -> Expectation { + // Wave 2 first (it shadows wave 1 wherever it covers). + for (const auto& r : wave2_ranges) { + if (k >= r.begin && k < r.end) { + const int width = r.end - r.begin; + const int del_lo = r.begin + width / 4; + const int del_hi = r.begin + 3 * width / 4; + if (k >= del_lo && k < del_hi) { + return Expectation::kAbsent; + } + return Expectation::kWave2Value; + } + } + // Fall back to wave 1. + const int batch = (k - kWave1Begin) / kWave1BatchSize; + const int batch_begin = kWave1Begin + batch * kWave1BatchSize; + const int del_lo = batch_begin + kWave1BatchSize / 4; + const int del_hi = batch_begin + 3 * kWave1BatchSize / 4; + if (k >= del_lo && k < del_hi) { + return Expectation::kAbsent; + } + return Expectation::kWave1Value; + }; + + int present_w1 = 0; + int present_w2 = 0; + int absent = 0; + std::string actual; + for (int k = kWave1Begin; k < kWave1End; ++k) { + const Status get_s = db_->Get(ReadOptions(), Key(k), &actual); + const Expectation exp = classify(k); + switch (exp) { + case Expectation::kAbsent: + EXPECT_TRUE(get_s.IsNotFound()) + << "key " << k << " expected NotFound (covered by tombstone); " + << "got status=" << get_s.ToString() + << " value_len=" << (get_s.ok() ? actual.size() : 0); + ++absent; + break; + case Expectation::kWave1Value: + ASSERT_OK(get_s) << "key " << k << " expected wave-1 value"; + EXPECT_EQ(wave1_value, actual) + << "key " << k << " expected wave-1 value but got a different one"; + ++present_w1; + break; + case Expectation::kWave2Value: + ASSERT_OK(get_s) << "key " << k << " expected wave-2 value"; + EXPECT_EQ(wave2_value, actual) + << "key " << k << " expected wave-2 value but got a different one"; + ++present_w2; + break; + } + } + // All three buckets should be non-empty, so the test really exercises both + // tombstone paths and the wave-2-wins path. A zero here means the setup + // drifted and the setup (not these checks) should be fixed. + EXPECT_GT(present_w1, 0); + EXPECT_GT(present_w2, 0); + EXPECT_GT(absent, 0); + + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->DisableProcessing(); + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->ClearAllCallBacks(); + Destroy(options); +} + +// With the option enabled, run reader threads against the live DB while manual +// compactions run in the background. Each compaction-input file is open through +// an ephemeral O_DIRECT handle while the shared TableCache still serves user +// reads through a buffered handle, so the same SST is open in two cache modes +// at once. This stresses that coexistence under TSAN/ASAN/UBSAN; it asserts +// reads stay consistent and final values match the last writes, not anything +// about timing. +TEST_F(DBCompactionTest, UseDirectIoForCompactionReadsConcurrentReadStress) { + if (!IsDirectIOSupported()) { + ROCKSDB_GTEST_BYPASS("Direct IO not supported"); + return; + } + + Options options = CurrentOptions(); + Destroy(options); + options.create_if_missing = true; + options.disable_auto_compactions = true; + options.use_direct_reads = false; + options.use_direct_io_for_compaction_reads = true; + options.use_direct_io_for_flush_and_compaction = false; + options.write_buffer_size = 64 * 1024; + options.target_file_size_base = 64 * 1024; + options.max_bytes_for_level_base = 256 * 1024; + options.level0_file_num_compaction_trigger = 100; // never auto-trigger + + Status s = TryReopen(options); + if (s.IsNotSupported() || s.IsInvalidArgument()) { + ROCKSDB_GTEST_BYPASS( + "Direct IO reads not supported in this test environment"); + return; + } + ASSERT_OK(s); + + constexpr int kNumKeys = 512; + constexpr int kValueSize = 256; + const std::string base_value(kValueSize, 'b'); + + // Initial population, flushed into a few L0 SSTs. + for (int i = 0; i < kNumKeys; ++i) { + ASSERT_OK(Put(Key(i), base_value)); + if (i > 0 && i % 128 == 0) { + ASSERT_OK(Flush()); + } + } + ASSERT_OK(Flush()); + + std::atomic stop{false}; + std::atomic reads_observed{0}; + std::atomic read_errors{0}; + + // Reader threads loop until stop is set, recording any status that isn't OK + // or NotFound. The point is to surface Corruption/IOError/use-after-free, so + // the main thread can fail the test. + constexpr int kNumReaders = 4; + std::vector readers; + readers.reserve(kNumReaders); + for (int t = 0; t < kNumReaders; ++t) { + readers.emplace_back([&, t]() { + std::string value; + while (!stop.load(std::memory_order_acquire)) { + const int k = + (t * 7919 + + static_cast(reads_observed.load(std::memory_order_relaxed))) % + kNumKeys; + Status read_s = db_->Get(ReadOptions(), Key(k), &value); + if (!read_s.ok() && !read_s.IsNotFound()) { + read_errors.fetch_add(1, std::memory_order_relaxed); + } + reads_observed.fetch_add(1, std::memory_order_relaxed); + } + }); + } + + // Drive a few rounds of writes and compactions on the main thread while + // the readers hammer Get(). Each round overwrites every key with a + // round-tagged value and then compacts. + constexpr int kRounds = 4; + std::string last_value = base_value; + for (int round = 0; round < kRounds; ++round) { + const std::string round_value(kValueSize, + static_cast('A' + (round % 26))); + for (int i = 0; i < kNumKeys; ++i) { + ASSERT_OK(Put(Key(i), round_value)); + if (i > 0 && i % 64 == 0) { + ASSERT_OK(Flush()); + } + } + ASSERT_OK(Flush()); + ASSERT_OK(dbfull()->CompactRange(CompactRangeOptions(), nullptr, nullptr)); + ASSERT_OK(dbfull()->TEST_WaitForCompact()); + last_value = round_value; + } + + stop.store(true, std::memory_order_release); + for (auto& reader : readers) { + reader.join(); + } + + // Every key must now hold the final round's value. Catches wholesale + // corruption the sampling readers might miss, and confirms compaction + // finished under the bypass path. + std::string value; + for (int i = 0; i < kNumKeys; ++i) { + ASSERT_OK(db_->Get(ReadOptions(), Key(i), &value)); + EXPECT_EQ(last_value, value); + } + EXPECT_GT(reads_observed.load(), 0) + << "reader threads never observed a single read; test was a no-op"; + EXPECT_EQ(0, read_errors.load()) + << "concurrent reads against compaction with bypass-path saw " + << read_errors.load() << " non-OK/non-NotFound status returns"; + + Destroy(options); +} +#endif // !defined(OS_MACOSX) && !defined(OS_OPENBSD) && ... + class CompactionPriTest : public DBTestBase, public testing::WithParamInterface { public: diff --git a/db/db_impl/db_impl.cc b/db/db_impl/db_impl.cc index 764031f2de..8d3baa275c 100644 --- a/db/db_impl/db_impl.cc +++ b/db/db_impl/db_impl.cc @@ -1842,6 +1842,8 @@ Status DBImpl::SetDBOptions( // TODO(xiez): clarify why apply optimize for read to write options file_options_for_compaction_ = fs_->OptimizeForCompactionTableRead( file_options_for_compaction_, immutable_db_options_); + TEST_SYNC_POINT_CALLBACK("DBImpl::SetDBOptions:FileOptionsForCompaction", + &file_options_for_compaction_); if (wal_other_option_changed || wal_size_option_changed) { WriteThread::Writer w; write_thread_.EnterUnbatched(&w, &mutex_); diff --git a/db/db_impl/db_impl_open.cc b/db/db_impl/db_impl_open.cc index a56fe6bc41..d4655b9860 100644 --- a/db/db_impl/db_impl_open.cc +++ b/db/db_impl/db_impl_open.cc @@ -255,6 +255,17 @@ Status DBImpl::ValidateOptions(const DBOptions& db_options) { "then direct I/O reads (use_direct_reads) must be disabled. "); } + if (db_options.allow_mmap_reads && + db_options.use_direct_io_for_compaction_reads) { + // mmap reads and direct I/O share the same EnvOptions field, so enabling + // both would try to mmap and O_DIRECT the same reads. Reject it here rather + // than tripping a lower-level assert. + return Status::NotSupported( + "If memory mapped reads (allow_mmap_reads) are enabled " + "then compaction-only direct I/O reads " + "(use_direct_io_for_compaction_reads) must be disabled. "); + } + if (db_options.allow_mmap_writes && db_options.use_direct_io_for_flush_and_compaction) { return Status::NotSupported( @@ -506,7 +517,8 @@ Status DBImpl::Recover( std::unique_ptr idfile; FileOptions customized_fs(file_options_); customized_fs.use_direct_reads |= - immutable_db_options_.use_direct_io_for_flush_and_compaction; + immutable_db_options_.use_direct_io_for_flush_and_compaction || + immutable_db_options_.use_direct_io_for_compaction_reads; const std::string& fname = manifest_path.empty() ? current_fname : manifest_path; s = fs_->NewRandomAccessFile(fname, customized_fs, &idfile, nullptr); diff --git a/db/db_options_test.cc b/db/db_options_test.cc index 26b5a51f46..949a62e216 100644 --- a/db/db_options_test.cc +++ b/db/db_options_test.cc @@ -13,6 +13,7 @@ #include "db/column_family.h" #include "db/db_impl/db_impl.h" #include "db/db_test_util.h" +#include "env/mock_env.h" #include "options/options_helper.h" #include "port/stack_trace.h" #include "rocksdb/cache.h" @@ -1763,6 +1764,171 @@ TEST_F(DBOptionsTest, SetOptionsMultipleColumnFamilies) { ASSERT_TRUE(dbfull()->GetOptions(handles_[2]).disable_auto_compactions); } +// Confirms the default value and serialization/parse round-trip of the new +// option. No DB open required; exercises only the options-metadata layer. +TEST_F(DBOptionsTest, UseDirectIoForCompactionReadsRoundTrip) { + // Default value must remain false to preserve existing semantics. + ASSERT_FALSE(DBOptions().use_direct_io_for_compaction_reads); + + DBOptions parsed; + ConfigOptions config_options; + ASSERT_OK(GetDBOptionsFromString(config_options, DBOptions(), + "use_direct_io_for_compaction_reads=true", + &parsed)); + ASSERT_TRUE(parsed.use_direct_io_for_compaction_reads); + ASSERT_OK(GetDBOptionsFromString(config_options, DBOptions(), + "use_direct_io_for_compaction_reads=false", + &parsed)); + ASSERT_FALSE(parsed.use_direct_io_for_compaction_reads); +} + +// Validates that Open rejects the documented incompatible combination. +TEST_F(DBOptionsTest, UseDirectIoForCompactionReadsValidation) { + // mmap_reads + use_direct_io_for_compaction_reads is rejected at Open + // time, the same way mmap_reads + use_direct_reads has always been + // rejected. + Options bad_options = CurrentOptions(); + bad_options.create_if_missing = true; + bad_options.allow_mmap_reads = true; + bad_options.use_direct_io_for_compaction_reads = true; + Status bad_status = TryReopen(bad_options); + ASSERT_TRUE(bad_status.IsNotSupported()) << bad_status.ToString(); +} + +// Confirms the option is plumbed all the way to the live DB's options API: +// after opening with the flag set, GetDBOptions() reports it back. Only runs +// when the environment supports direct I/O. +TEST_F(DBOptionsTest, UseDirectIoForCompactionReadsLiveReopen) { + Options options = CurrentOptions(); + options.create_if_missing = true; + options.use_direct_io_for_compaction_reads = true; + // Use a buffered user-read setup so the new flag is the one doing the work. + options.use_direct_reads = false; + options.use_direct_io_for_flush_and_compaction = false; + Status s = TryReopen(options); + if (s.IsNotSupported() || s.IsInvalidArgument()) { + ROCKSDB_GTEST_BYPASS( + "Direct I/O not supported in this test environment; live reopen " + "cannot be exercised."); + return; + } + ASSERT_OK(s); + ASSERT_TRUE(dbfull()->GetDBOptions().use_direct_io_for_compaction_reads); + Close(); +} + +TEST_F(DBOptionsTest, UseDirectIoForCompactionReadsUnsupportedFileSystem) { + auto fs = std::make_shared(Env::Default()->GetSystemClock(), + /*supports_direct_io=*/false); + std::unique_ptr mock_env = NewCompositeEnv(fs); + + Options options = CurrentOptions(); + options.env = mock_env.get(); + options.create_if_missing = true; + options.use_direct_reads = false; + options.use_direct_io_for_flush_and_compaction = false; + options.use_direct_io_for_compaction_reads = true; + + Status s = TryReopen(options); + ASSERT_TRUE(s.IsInvalidArgument()) << s.ToString(); + ASSERT_NE(s.ToString().find("Direct I/O is not supported"), std::string::npos) + << s.ToString(); +} + +TEST_F( + DBOptionsTest, + UseDirectIoForCompactionReadsSetDBOptionsKeepsVerificationReadsBuffered) { + auto fs = std::make_shared(Env::Default()->GetSystemClock(), + /*supports_direct_io=*/true); + std::unique_ptr mock_env = NewCompositeEnv(fs); + + Options options = CurrentOptions(); + options.env = mock_env.get(); + options.create_if_missing = true; + options.use_direct_reads = false; + options.use_direct_io_for_flush_and_compaction = false; + options.use_direct_io_for_compaction_reads = true; + + ASSERT_OK(TryReopen(options)); + + bool observed = false; + FileOptions observed_file_options; + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack( + "DBImpl::SetDBOptions:FileOptionsForCompaction", [&](void* arg) { + observed = true; + observed_file_options = *static_cast(arg); + }); + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->EnableProcessing(); + + ASSERT_OK(dbfull()->SetDBOptions({{"bytes_per_sync", "1024"}})); + ASSERT_TRUE(observed); + ASSERT_FALSE(observed_file_options.use_direct_reads); + + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->DisableProcessing(); + ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->ClearAllCallBacks(); + Close(); +} + +// Exercises the FileSystem::OptimizeForCompactionTableRead and +// OptimizeForBlobFileRead helpers directly, asserting that the compaction-only +// flag does not affect shared FileSystem hooks. +TEST_F(DBOptionsTest, OptimizeForCompactionTableReadUsesGlobalDirectReadsOnly) { + FileOptions in_opts; + in_opts.use_direct_reads = false; + + { + Options check_options; + check_options.use_direct_reads = false; + check_options.use_direct_io_for_compaction_reads = true; + check_options.use_direct_io_for_flush_and_compaction = false; + ImmutableDBOptions immutable(check_options); + FileOptions sst_read = + env_->GetFileSystem()->OptimizeForCompactionTableRead(in_opts, + immutable); + FileOptions blob_read = + env_->GetFileSystem()->OptimizeForBlobFileRead(in_opts, immutable); + EXPECT_FALSE(sst_read.use_direct_reads); + EXPECT_FALSE(blob_read.use_direct_reads); + EXPECT_FALSE(sst_read.use_direct_writes); + } + + { + Options off_options; + off_options.use_direct_reads = false; + off_options.use_direct_io_for_compaction_reads = false; + off_options.use_direct_io_for_flush_and_compaction = false; + ImmutableDBOptions immutable_off(off_options); + FileOptions sst_read_off = + env_->GetFileSystem()->OptimizeForCompactionTableRead(in_opts, + immutable_off); + EXPECT_FALSE(sst_read_off.use_direct_reads); + } + + { + Options global_on_options; + global_on_options.use_direct_reads = true; + global_on_options.use_direct_io_for_compaction_reads = false; + global_on_options.use_direct_io_for_flush_and_compaction = false; + ImmutableDBOptions immutable_global(global_on_options); + FileOptions sst_read_global = + env_->GetFileSystem()->OptimizeForCompactionTableRead(in_opts, + immutable_global); + EXPECT_TRUE(sst_read_global.use_direct_reads); + } + + { + Options both_on; + both_on.use_direct_reads = true; + both_on.use_direct_io_for_compaction_reads = true; + both_on.use_direct_io_for_flush_and_compaction = false; + ImmutableDBOptions immutable_both(both_on); + FileOptions sst_read_both = + env_->GetFileSystem()->OptimizeForCompactionTableRead(in_opts, + immutable_both); + EXPECT_TRUE(sst_read_both.use_direct_reads); + } +} + } // namespace ROCKSDB_NAMESPACE int main(int argc, char** argv) { diff --git a/db/range_tombstone_fragmenter.cc b/db/range_tombstone_fragmenter.cc index 90e4da55eb..87a2007ef3 100644 --- a/db/range_tombstone_fragmenter.cc +++ b/db/range_tombstone_fragmenter.cc @@ -497,6 +497,9 @@ FragmentedRangeTombstoneIterator::SplitBySnapshot( splits; SequenceNumber lower = 0; SequenceNumber upper; + // Safe to dereference icmp_ below: SplitBySnapshot is construction-time work + // (called synchronously from the range-del aggregator while the reader that + // owns *icmp_ is alive). See the icmp_ note in the header. for (size_t i = 0; i <= snapshots.size(); i++) { if (i >= snapshots.size()) { upper = kMaxSequenceNumber; diff --git a/db/range_tombstone_fragmenter.h b/db/range_tombstone_fragmenter.h index 1fb68bd42e..0d79e7fbe6 100644 --- a/db/range_tombstone_fragmenter.h +++ b/db/range_tombstone_fragmenter.h @@ -199,7 +199,9 @@ class FragmentedRangeTombstoneIterator : public InternalIterator { RangeTombstone Tombstone() const { assert(Valid()); - if (icmp_->user_comparator()->timestamp_size()) { + // Use ucmp_ (long-lived; see the icmp_ note below) rather than + // icmp_->user_comparator(). They are equal for all constructors. + if (ucmp_->timestamp_size()) { return RangeTombstone(start_key(), end_key(), seq(), timestamp()); } return RangeTombstone(start_key(), end_key(), seq()); @@ -316,6 +318,12 @@ class FragmentedRangeTombstoneIterator : public InternalIterator { const RangeTombstoneStackStartComparator tombstone_start_cmp_; const RangeTombstoneStackEndComparator tombstone_end_cmp_; + // icmp_ is a borrowed pointer and is not guaranteed to outlive this iterator: + // it may be built from a comparator with a shorter lifetime than an iterator + // some owner keeps alive. Only dereference it during construction-time work + // (e.g. SplitBySnapshot), when the referent is known alive. Post-construction + // navigation must use ucmp_ (the long-lived user comparator) and tombstones_; + // do not add new icmp_ dereferences in Next()/Seek()/Tombstone(). const InternalKeyComparator* icmp_; const Comparator* ucmp_; std::shared_ptr tombstones_ref_; diff --git a/db/table_cache.cc b/db/table_cache.cc index 74c2307be2..6dec5baeeb 100644 --- a/db/table_cache.cc +++ b/db/table_cache.cc @@ -98,7 +98,7 @@ Status TableCache::GetTableReader( const MutableCFOptions& mutable_cf_options, bool skip_filters, int level, bool prefetch_index_and_filter_in_cache, size_t max_file_size_for_l0_meta_pin, Temperature file_temperature, - std::string* file_open_metadata) { + std::string* file_open_metadata, bool avoid_shared_metadata_cache) { std::string fname = TableFileName( ioptions_.cf_paths, file_meta.fd.GetNumber(), file_meta.fd.GetPathId()); std::unique_ptr file; @@ -190,7 +190,8 @@ Status TableCache::GetTableReader( block_cache_tracer_, max_file_size_for_l0_meta_pin, db_session_id_, file_meta.fd.GetNumber(), expected_unique_id, file_meta.fd.largest_seqno, file_meta.tail_size, - file_meta.user_defined_timestamps_persisted), + file_meta.user_defined_timestamps_persisted, + avoid_shared_metadata_cache), std::move(file_reader), file_meta.fd.GetFileSize(), table_reader, prefetch_index_and_filter_in_cache); TEST_SYNC_POINT("TableCache::GetTableReader:0"); @@ -213,11 +214,53 @@ Status TableCache::FindTable( const bool no_io, HistogramImpl* file_read_hist, bool skip_filters, int level, bool prefetch_index_and_filter_in_cache, size_t max_file_size_for_l0_meta_pin, Temperature file_temperature, - bool pin_table_handle, std::string* file_open_metadata) { + bool pin_table_handle, std::string* file_open_metadata, + std::unique_ptr* fresh_table_reader_owner, + const TableCacheOpenOptions& open_options) { assert(out_table_reader != nullptr && *out_table_reader == nullptr); assert(handle != nullptr && *handle == nullptr); + // open_ephemeral_table_reader requests a fresh reader; + // fresh_table_reader_owner is where we return it. The two must agree. + assert(open_options.open_ephemeral_table_reader == + (fresh_table_reader_owner != nullptr)); PERF_TIMER_GUARD_WITH_CLOCK(find_table_nanos, ioptions_.clock); + // Bypass path: open a fresh TableReader, skipping the pinned-reader fast path + // and the shared cache. The caller takes ownership via + // fresh_table_reader_owner. no_io is not allowed here since opening a new + // reader always needs I/O. + if (fresh_table_reader_owner != nullptr) { + assert(!no_io); + if (no_io) { + // Defensive in release builds: the caller violated the contract. + return Status::Incomplete( + "fresh TableReader requested but no_io is set; cannot open file"); + } + std::unique_ptr table_reader; + const bool effective_skip_filters = + skip_filters || open_options.skip_filters; + TEST_SYNC_POINT_CALLBACK("TableCache::FindTable:FreshTableReader", + const_cast(&open_options)); + Status s = GetTableReader( + ro, file_options, internal_comparator, file_meta, + false /* sequential mode */, file_read_hist, &table_reader, + mutable_cf_options, effective_skip_filters, level, + prefetch_index_and_filter_in_cache && + !open_options.avoid_shared_metadata_cache, + max_file_size_for_l0_meta_pin, file_temperature, file_open_metadata, + open_options.avoid_shared_metadata_cache); + if (!s.ok()) { + assert(table_reader == nullptr); + RecordTick(ioptions_.stats, NO_FILE_ERRORS); + IGNORE_STATUS_IF_ERROR(s); + return s; + } + *out_table_reader = table_reader.get(); + *fresh_table_reader_owner = std::move(table_reader); + *handle = nullptr; + return s; + } + // Fast path: if table reader is already pinned, return it directly without a // cache lookup. auto pinned_reader = file_meta.fd.pinned_reader.Get(); @@ -313,15 +356,23 @@ InternalIterator* TableCache::NewIterator( const InternalKey* largest_compaction_key, bool allow_unprepared_value, const SequenceNumber* read_seqno, std::unique_ptr* range_del_iter, - bool maybe_pin_table_handle, std::string* file_open_metadata) { + bool maybe_pin_table_handle, std::string* file_open_metadata, + const TableCacheOpenOptions& open_options) { PERF_TIMER_GUARD(new_table_iterator_nanos); Status s; TableReader* table_reader = nullptr; TypedHandle* handle = nullptr; + assert(!open_options.open_ephemeral_table_reader || + table_reader_ptr == nullptr); + // Holds ownership of a freshly-opened TableReader when the caller asked us + // to bypass the shared cache. When non-empty, the iterator we hand back must + // arrange to free it on destruction. + std::unique_ptr ephemeral_reader; if (table_reader_ptr != nullptr) { *table_reader_ptr = nullptr; } + const bool effective_skip_filters = skip_filters || open_options.skip_filters; bool for_compaction = caller == TableReaderCaller::kCompaction; TEST_SYNC_POINT_CALLBACK("TableCache::NewIterator::BeforeFindTable", const_cast(&file_meta.fd)); @@ -329,9 +380,13 @@ InternalIterator* TableCache::NewIterator( options, file_options, icomparator, file_meta, &handle, mutable_cf_options, &table_reader, options.read_tier == kBlockCacheTier /* no_io */, file_read_hist, - skip_filters, level, true /* prefetch_index_and_filter_in_cache */, - max_file_size_for_l0_meta_pin, file_meta.temperature, - maybe_pin_table_handle && should_pin_table_handles_, file_open_metadata); + effective_skip_filters, level, + /*prefetch_index_and_filter_in_cache=*/ + !open_options.avoid_shared_metadata_cache, max_file_size_for_l0_meta_pin, + file_meta.temperature, + maybe_pin_table_handle && should_pin_table_handles_, file_open_metadata, + open_options.open_ephemeral_table_reader ? &ephemeral_reader : nullptr, + open_options); InternalIterator* result = nullptr; if (s.ok()) { if (options.table_filter && @@ -340,13 +395,17 @@ InternalIterator* TableCache::NewIterator( } else { result = table_reader->NewIterator( options, mutable_cf_options.prefix_extractor.get(), arena, - skip_filters, caller, file_options.compaction_readahead_size, - allow_unprepared_value); + effective_skip_filters, caller, + file_options.compaction_readahead_size, allow_unprepared_value); } if (handle != nullptr) { cache_.RegisterReleaseAsCleanup(handle, *result); handle = nullptr; // prevent from releasing below } + // Don't hand ephemeral_reader to result's cleanup yet: range-del + // processing below can set s to non-OK, in which case result is replaced + // by an error iterator at function end. Transfer ownership only once we + // know s stays OK; otherwise ephemeral_reader's destructor frees it. if (for_compaction) { table_reader->SetupForCompaction(); @@ -398,8 +457,37 @@ InternalIterator* TableCache::NewIterator( if (handle != nullptr) { cache_.Release(handle); } + // Range-del processing is done and s is final. Hand the ephemeral reader's + // lifetime to the returned iterator; if s is non-OK, leave it for + // ephemeral_reader's destructor. RegisterCleanup gets the raw pointer before + // release(), so if its allocation throws the reader is still owned by + // ephemeral_reader and freed on unwind. + if (s.ok() && ephemeral_reader && result != nullptr) { + TableReader* raw = ephemeral_reader.get(); + result->RegisterCleanup( + [](void* arg1, void* /*arg2*/) { + delete static_cast(arg1); + }, + raw, nullptr); + // release() returns raw, which the cleanup now owns; drop the unique_ptr's + // ownership so the reader isn't freed twice. + [[maybe_unused]] TableReader* released = ephemeral_reader.release(); + assert(released == raw); + } if (!s.ok()) { + // Today result is always null here: it is only set when s was OK, and the + // only later change to s, new_range_del_iter->status(), is always OK for a + // FragmentedRangeTombstoneIterator. The assert documents that; the cleanup + // still disposes of any stray result before the error iterator replaces it. assert(result == nullptr); + if (result != nullptr) { + if (arena != nullptr) { + result->~InternalIterator(); + } else { + delete result; + } + result = nullptr; + } result = NewErrorInternalIterator(s, arena); } return result; diff --git a/db/table_cache.h b/db/table_cache.h index e19ae997ff..1df16c6d6a 100644 --- a/db/table_cache.h +++ b/db/table_cache.h @@ -35,6 +35,18 @@ struct FileDescriptor; class GetContext; class HistogramImpl; +struct TableCacheOpenOptions { + // Open a new TableReader owned by the returned iterator instead of reusing + // the pinned reader or shared TableCache entry. + bool open_ephemeral_table_reader = false; + + // Avoid inserting open-time metadata blocks into the shared block cache. + bool avoid_shared_metadata_cache = false; + + // Disable loading/accessing filter blocks for this open/iterator. + bool skip_filters = false; +}; + // Manages caching for TableReader objects for a column family. The actual // cache is allocated separately and passed to the constructor. TableCache // wraps around the underlying SST file readers by providing Get(), @@ -73,7 +85,9 @@ class TableCache { // underlying the returned iterator, or nullptr if no Table object underlies // the returned iterator. The returned "*table_reader_ptr" object is owned // by the cache and should not be deleted, and is valid for as long as the - // returned iterator is live. + // returned iterator is live. `table_reader_ptr` must be nullptr when + // `open_options.open_ephemeral_table_reader` is set because that reader is + // owned by iterator cleanup rather than by the cache. // If !options.ignore_range_deletions, and range_del_iter is non-nullptr, // then range_del_iter is set to a TruncatedRangeDelIterator for range // tombstones in the SST file corresponding to the specified file number. The @@ -82,12 +96,16 @@ class TableCache { // @param options Must outlive the returned iterator. // @param range_del_agg If non-nullptr, adds range deletions to the // aggregator. If an error occurs, returns it in a NewErrorInternalIterator - // @param for_compaction If true, a new TableReader may be allocated (but - // not cached), depending on the CF options + // @param caller Identifies the high-level caller for table-reader behavior + // such as compaction preparation. // @param skip_filters Disables loading/accessing the filter block // @param level The level this table is at, -1 for "not set / don't know" // @param range_del_read_seqno If non-nullptr, will be used to create // *range_del_iter. + // @param open_options Optional table-open policy overrides. The caller must + // not combine `open_ephemeral_table_reader` with + // `ReadOptions::read_tier = kBlockCacheTier` (no_io); + // opening an ephemeral reader inherently requires I/O. InternalIterator* NewIterator( const ReadOptions& options, const FileOptions& toptions, const InternalKeyComparator& internal_comparator, @@ -103,7 +121,8 @@ class TableCache { bool maybe_pin_table_handle = false, // If non-null, and the table reader is newly opened (not cached), // retrieves file open metadata via GetFileOpenMetadata(). - std::string* file_open_metadata = nullptr); + std::string* file_open_metadata = nullptr, + const TableCacheOpenOptions& open_options = TableCacheOpenOptions()); // If a seek to internal key "k" in specified file finds an entry, // call get_context->SaveValue() repeatedly until @@ -190,18 +209,30 @@ class TableCache { // @param pin_table_handle If true, pins the table reader on file_meta so // future lookups bypass the cache. *handle is set to nullptr // on return in this case. - Status FindTable(const ReadOptions& ro, const FileOptions& toptions, - const InternalKeyComparator& internal_comparator, - const FileMetaData& file_meta, TypedHandle**, - const MutableCFOptions& mutable_cf_options, - TableReader** table_reader, const bool no_io = false, - HistogramImpl* file_read_hist = nullptr, - bool skip_filters = false, int level = -1, - bool prefetch_index_and_filter_in_cache = true, - size_t max_file_size_for_l0_meta_pin = 0, - Temperature file_temperature = Temperature::kUnknown, - bool pin_table_handle = false, - std::string* file_open_metadata = nullptr); + // @param fresh_table_reader_owner If non-null, FindTable always opens a new + // TableReader (skipping the pinned-reader fast path and the + // shared cache) and writes ownership into this unique_ptr. + // `*handle` will be nullptr on return and `*table_reader` will + // point to the freshly allocated reader. Callers that pass this + // are responsible for keeping the unique_ptr alive for the + // lifetime of any iterators built on top. + // @param open_options Optional table-open policy overrides (e.g. + // avoid_shared_metadata_cache, skip_filters). + // `open_options.open_ephemeral_table_reader` must be set if and + // only if `fresh_table_reader_owner` is non-null. + Status FindTable( + const ReadOptions& ro, const FileOptions& toptions, + const InternalKeyComparator& internal_comparator, + const FileMetaData& file_meta, TypedHandle**, + const MutableCFOptions& mutable_cf_options, TableReader** table_reader, + const bool no_io = false, HistogramImpl* file_read_hist = nullptr, + bool skip_filters = false, int level = -1, + bool prefetch_index_and_filter_in_cache = true, + size_t max_file_size_for_l0_meta_pin = 0, + Temperature file_temperature = Temperature::kUnknown, + bool pin_table_handle = false, std::string* file_open_metadata = nullptr, + std::unique_ptr* fresh_table_reader_owner = nullptr, + const TableCacheOpenOptions& open_options = TableCacheOpenOptions()); // Get the table properties of a given table. // @no_io: indicates if we should load table to the cache if it is not present @@ -286,7 +317,8 @@ class TableCache { bool prefetch_index_and_filter_in_cache = true, size_t max_file_size_for_l0_meta_pin = 0, Temperature file_temperature = Temperature::kUnknown, - std::string* file_open_metadata = nullptr); + std::string* file_open_metadata = nullptr, + bool avoid_shared_metadata_cache = false); // Update the max_covering_tombstone_seq in the GetContext for each key based // on the range deletions in the table diff --git a/db/version_set.cc b/db/version_set.cc index 995ca752f4..4dd2048802 100644 --- a/db/version_set.cc +++ b/db/version_set.cc @@ -966,6 +966,15 @@ bool SomeFileOverlapsRange(const InternalKeyComparator& icmp, namespace { +TableCacheOpenOptions GetCompactionTableCacheOpenOptions( + bool open_ephemeral_table_reader) { + TableCacheOpenOptions options; + options.open_ephemeral_table_reader = open_ephemeral_table_reader; + options.avoid_shared_metadata_cache = open_ephemeral_table_reader; + options.skip_filters = open_ephemeral_table_reader; + return options; +} + class LevelIterator final : public InternalIterator { public: // NOTE: many of the const& parameters are saved in this object (so @@ -981,7 +990,8 @@ class LevelIterator final : public InternalIterator { bool allow_unprepared_value = false, std::unique_ptr*** range_tombstone_iter_ptr_ = nullptr, - Statistics* db_statistics = nullptr, SystemClock* clock = nullptr) + Statistics* db_statistics = nullptr, SystemClock* clock = nullptr, + bool open_ephemeral_table_reader = false) : table_cache_(table_cache), read_options_(read_options), file_options_(file_options), @@ -1007,7 +1017,9 @@ class LevelIterator final : public InternalIterator { to_return_sentinel_(false), scan_opts_(nullptr), db_statistics_(db_statistics), - clock_(clock) { + clock_(clock), + table_cache_open_options_( + GetCompactionTableCacheOpenOptions(open_ephemeral_table_reader)) { // Empty level is not supported. assert(flevel_ != nullptr && flevel_->num_files > 0); if (range_tombstone_iter_ptr_) { @@ -1314,7 +1326,9 @@ class LevelIterator final : public InternalIterator { /*max_file_size_for_l0_meta_pin=*/0, smallest_compaction_key, largest_compaction_key, allow_unprepared_value_, &read_seq_, range_tombstone_iter_, - /*maybe_pin_table_handle=*/true); + /*maybe_pin_table_handle=*/ + !table_cache_open_options_.open_ephemeral_table_reader, + /*file_open_metadata=*/nullptr, table_cache_open_options_); } // Check if current file being fully within iterate_lower_bound. @@ -1391,6 +1405,7 @@ class LevelIterator final : public InternalIterator { Statistics* db_statistics_ = nullptr; SystemClock* clock_ = nullptr; + TableCacheOpenOptions table_cache_open_options_; // Our stored scan_opts for each prefix std::unique_ptr file_to_scan_opts_ = nullptr; @@ -7859,7 +7874,7 @@ InternalIterator* VersionSet::MakeInputIterator( RangeDelAggregator* range_del_agg, const FileOptions& file_options_compactions, const std::optional& start, - const std::optional& end) { + const std::optional& end, bool open_ephemeral_table_reader) { auto cfd = c->column_family_data(); // Level-0 files have to be merged together. For other levels, // we will make a concatenating iterator per level. @@ -7878,6 +7893,8 @@ InternalIterator* VersionSet::MakeInputIterator( range_tombstones; size_t num = 0; [[maybe_unused]] size_t num_input_files = 0; + const TableCacheOpenOptions table_cache_open_options = + GetCompactionTableCacheOpenOptions(open_ephemeral_table_reader); for (size_t which = 0; which < c->num_input_levels(); which++) { const LevelFilesBrief* flevel = c->input_levels(which); num_input_files += flevel->num_files; @@ -7914,7 +7931,9 @@ InternalIterator* VersionSet::MakeInputIterator( /*largest_compaction_key=*/nullptr, /*allow_unprepared_value=*/false, /*range_del_read_seqno=*/nullptr, - /*range_del_iter=*/&range_tombstone_iter); + /*range_del_iter=*/&range_tombstone_iter, + /*maybe_pin_table_handle=*/false, + /*file_open_metadata=*/nullptr, table_cache_open_options); range_tombstones.emplace_back(std::move(range_tombstone_iter), nullptr); } @@ -7929,7 +7948,7 @@ InternalIterator* VersionSet::MakeInputIterator( TableReaderCaller::kCompaction, /*skip_filters=*/false, /*level=*/static_cast(c->level(which)), range_del_agg, c->boundaries(which), false, &tombstone_iter_ptr, - db_options_->statistics.get(), clock_); + db_options_->statistics.get(), clock_, open_ephemeral_table_reader); range_tombstones.emplace_back(nullptr, tombstone_iter_ptr); } } diff --git a/db/version_set.h b/db/version_set.h index 6a1d654714..e424316b43 100644 --- a/db/version_set.h +++ b/db/version_set.h @@ -1558,12 +1558,16 @@ class VersionSet { // The caller should delete the iterator when no longer needed. // @param read_options Must outlive the returned iterator. // @param start, end indicates compaction range + // @param open_ephemeral_table_reader When true, the per-file iterators + // bypass the shared TableCache and open fresh TableReaders + // using `file_options_compactions`. InternalIterator* MakeInputIterator( const ReadOptions& read_options, const Compaction* c, RangeDelAggregator* range_del_agg, const FileOptions& file_options_compactions, const std::optional& start, - const std::optional& end); + const std::optional& end, + bool open_ephemeral_table_reader = false); // Add all files listed in any live version to *live_table_files and // *live_blob_files. Note that these lists may contain duplicates. diff --git a/db_stress_tool/db_stress_common.h b/db_stress_tool/db_stress_common.h index f73de23129..abfa8a0e19 100644 --- a/db_stress_tool/db_stress_common.h +++ b/db_stress_tool/db_stress_common.h @@ -201,6 +201,7 @@ DECLARE_bool(verify_checksum); DECLARE_bool(mmap_read); DECLARE_bool(mmap_write); DECLARE_bool(use_direct_reads); +DECLARE_bool(use_direct_io_for_compaction_reads); DECLARE_bool(use_direct_io_for_flush_and_compaction); DECLARE_bool(mock_direct_io); DECLARE_bool(statistics); diff --git a/db_stress_tool/db_stress_gflags.cc b/db_stress_tool/db_stress_gflags.cc index 895ec1a016..4c658a9d7f 100644 --- a/db_stress_tool/db_stress_gflags.cc +++ b/db_stress_tool/db_stress_gflags.cc @@ -732,6 +732,11 @@ DEFINE_bool(mmap_write, ROCKSDB_NAMESPACE::Options().allow_mmap_writes, DEFINE_bool(use_direct_reads, ROCKSDB_NAMESPACE::Options().use_direct_reads, "Use O_DIRECT for reading data"); +DEFINE_bool(use_direct_io_for_compaction_reads, + ROCKSDB_NAMESPACE::Options().use_direct_io_for_compaction_reads, + "Use O_DIRECT for compaction-input SST reads only, while keeping " + "user reads buffered"); + DEFINE_bool(use_direct_io_for_flush_and_compaction, ROCKSDB_NAMESPACE::Options().use_direct_io_for_flush_and_compaction, "Use O_DIRECT for writing data"); diff --git a/db_stress_tool/db_stress_test_base.cc b/db_stress_tool/db_stress_test_base.cc index 83ad31ac31..72c62cda1c 100644 --- a/db_stress_tool/db_stress_test_base.cc +++ b/db_stress_tool/db_stress_test_base.cc @@ -5166,6 +5166,8 @@ void InitializeOptionsFromFlags( options.allow_mmap_reads = FLAGS_mmap_read; options.allow_mmap_writes = FLAGS_mmap_write; options.use_direct_reads = FLAGS_use_direct_reads; + options.use_direct_io_for_compaction_reads = + FLAGS_use_direct_io_for_compaction_reads; options.use_direct_io_for_flush_and_compaction = FLAGS_use_direct_io_for_flush_and_compaction; options.recycle_log_file_num = diff --git a/env/fs_posix.cc b/env/fs_posix.cc index 4aca3735f4..63a4cb8214 100644 --- a/env/fs_posix.cc +++ b/env/fs_posix.cc @@ -950,15 +950,14 @@ class PosixFileSystem : public FileSystem { FileOptions OptimizeForCompactionTableRead( const FileOptions& file_options, const ImmutableDBOptions& db_options) const override { - FileOptions fo = FileOptions(file_options); + FileOptions fo = + FileSystem::OptimizeForCompactionTableRead(file_options, db_options); #ifdef OS_LINUX // To fix https://github.com/facebook/rocksdb/issues/12038 - if (!file_options.use_direct_reads && - file_options.compaction_readahead_size > 0) { + if (!fo.use_direct_reads && fo.compaction_readahead_size > 0) { size_t system_limit = GetCompactionReadaheadSizeSystemLimit(db_options.db_paths); - if (system_limit > 0 && - file_options.compaction_readahead_size > system_limit) { + if (system_limit > 0 && fo.compaction_readahead_size > system_limit) { fo.compaction_readahead_size = system_limit; } } diff --git a/include/rocksdb/c.h b/include/rocksdb/c.h index 3d3d1c49f2..a00fa0942b 100644 --- a/include/rocksdb/c.h +++ b/include/rocksdb/c.h @@ -1908,6 +1908,11 @@ rocksdb_options_set_use_direct_io_for_flush_and_compaction(rocksdb_options_t*, unsigned char); extern ROCKSDB_LIBRARY_API unsigned char rocksdb_options_get_use_direct_io_for_flush_and_compaction(rocksdb_options_t*); +extern ROCKSDB_LIBRARY_API void +rocksdb_options_set_use_direct_io_for_compaction_reads(rocksdb_options_t*, + unsigned char); +extern ROCKSDB_LIBRARY_API unsigned char +rocksdb_options_get_use_direct_io_for_compaction_reads(rocksdb_options_t*); extern ROCKSDB_LIBRARY_API void rocksdb_options_set_is_fd_close_on_exec( rocksdb_options_t*, unsigned char); extern ROCKSDB_LIBRARY_API unsigned char diff --git a/include/rocksdb/options.h b/include/rocksdb/options.h index 7dfe55eb1b..cea9c20864 100644 --- a/include/rocksdb/options.h +++ b/include/rocksdb/options.h @@ -1155,7 +1155,46 @@ struct DBOptions { // Default: false bool use_direct_reads = false; - // Use O_DIRECT for writes in background flush and compactions. + // Use O_DIRECT for compaction-input SST reads only, leaving user reads + // buffered. Useful when sequential compaction reads would otherwise evict + // the hot user-read working set from the OS page cache. When this is true + // and use_direct_reads is false, compaction opens short-lived O_DIRECT + // readers for its input files instead of reusing the buffered readers cached + // for user reads. This is the read-side analogue of + // use_direct_io_for_flush_and_compaction, and the two are often paired on + // write-heavy workloads. + // + // Scope and limits: + // * DBOption scope (applies to all column families); no per-CF setting. + // * Covers compaction inputs only. Blob-file reads and compaction-output + // verification (paranoid_file_checks) still use the buffered path. + // * The ephemeral readers bypass the TableCache and are not counted against + // max_open_files. Non-L0 levels keep one reader open at a time; L0 opens + // all of a subcompaction's overlapping inputs at once, so with large L0 + // fan-in and many subcompactions, watch RLIMIT_NOFILE. + // * Every input file is reopened per compaction, so NO_FILE_OPENS and + // TABLE_OPEN_IO_MICROS rise while this is enabled. + // + // The same SST can be open through both a buffered handle (user reads) and an + // O_DIRECT handle (the compaction scan) at once; modern Linux handles this + // fine. The flag is neutral or slightly negative for in-memory DBs or + // uniform random reads, so measure before enabling. + // + // Has no effect when use_direct_reads is true (all reads are already + // O_DIRECT). Rejected at DB::Open when allow_mmap_reads is set. + // + // On a filesystem without O_DIRECT support (e.g. tmpfs), DB::Open fails: it + // probes by opening the MANIFEST with O_DIRECT. The probe only checks the + // filesystem holding the DB directory, so if SST files live elsewhere (via + // db_paths/cf_paths) without O_DIRECT, Open succeeds and the first compaction + // fails instead. + // + // Default: false + bool use_direct_io_for_compaction_reads = false; + + // Use O_DIRECT for writes in background flush and compactions. See also + // use_direct_io_for_compaction_reads, the read-side analogue often paired + // with this on write-heavy workloads. // Default: false bool use_direct_io_for_flush_and_compaction = false; diff --git a/options/db_options.cc b/options/db_options.cc index 4efaf476fb..13f7b8c137 100644 --- a/options/db_options.cc +++ b/options/db_options.cc @@ -192,6 +192,11 @@ static std::unordered_map {offsetof(struct ImmutableDBOptions, use_direct_reads), OptionType::kBoolean, OptionVerificationType::kNormal, OptionTypeFlags::kNone}}, + {"use_direct_io_for_compaction_reads", + {offsetof(struct ImmutableDBOptions, + use_direct_io_for_compaction_reads), + OptionType::kBoolean, OptionVerificationType::kNormal, + OptionTypeFlags::kNone}}, {"use_direct_writes", {0, OptionType::kBoolean, OptionVerificationType::kDeprecated, OptionTypeFlags::kNone}}, @@ -785,6 +790,8 @@ ImmutableDBOptions::ImmutableDBOptions(const DBOptions& options) allow_mmap_reads(options.allow_mmap_reads), allow_mmap_writes(options.allow_mmap_writes), use_direct_reads(options.use_direct_reads), + use_direct_io_for_compaction_reads( + options.use_direct_io_for_compaction_reads), use_direct_io_for_flush_and_compaction( options.use_direct_io_for_flush_and_compaction), allow_fallocate(options.allow_fallocate), @@ -911,8 +918,10 @@ void ImmutableDBOptions::Dump(Logger* log) const { ROCKS_LOG_HEADER(log, " Options.use_direct_reads: %d", use_direct_reads); ROCKS_LOG_HEADER(log, - " " - "Options.use_direct_io_for_flush_and_compaction: %d", + " Options.use_direct_io_for_compaction_reads: %d", + use_direct_io_for_compaction_reads); + ROCKS_LOG_HEADER(log, + " Options.use_direct_io_for_flush_and_compaction: %d", use_direct_io_for_flush_and_compaction); ROCKS_LOG_HEADER(log, " Options.create_missing_column_families: %d", create_missing_column_families); diff --git a/options/db_options.h b/options/db_options.h index e96bb83c18..56e3145fe3 100644 --- a/options/db_options.h +++ b/options/db_options.h @@ -57,6 +57,7 @@ struct ImmutableDBOptions { bool allow_mmap_reads; bool allow_mmap_writes; bool use_direct_reads; + bool use_direct_io_for_compaction_reads; bool use_direct_io_for_flush_and_compaction; bool allow_fallocate; bool is_fd_close_on_exec; diff --git a/options/options_helper.cc b/options/options_helper.cc index 6dfb3ffde0..c107622033 100644 --- a/options/options_helper.cc +++ b/options/options_helper.cc @@ -114,6 +114,8 @@ void BuildDBOptions(const ImmutableDBOptions& immutable_db_options, options.allow_mmap_reads = immutable_db_options.allow_mmap_reads; options.allow_mmap_writes = immutable_db_options.allow_mmap_writes; options.use_direct_reads = immutable_db_options.use_direct_reads; + options.use_direct_io_for_compaction_reads = + immutable_db_options.use_direct_io_for_compaction_reads; options.use_direct_io_for_flush_and_compaction = immutable_db_options.use_direct_io_for_flush_and_compaction; options.allow_fallocate = immutable_db_options.allow_fallocate; diff --git a/options/options_settable_test.cc b/options/options_settable_test.cc index 774cf06f29..7b03e2aadb 100644 --- a/options/options_settable_test.cc +++ b/options/options_settable_test.cc @@ -447,6 +447,7 @@ TEST_F(OptionsSettableTest, DBOptionsAllFieldsSettable) { "allow_fallocate=true;" "allow_mmap_reads=false;" "use_direct_reads=false;" + "use_direct_io_for_compaction_reads=false;" "use_direct_io_for_flush_and_compaction=false;" "max_log_file_size=4607;" "advise_random_on_open=true;" diff --git a/options/options_test.cc b/options/options_test.cc index deceb33908..a6a60e6390 100644 --- a/options/options_test.cc +++ b/options/options_test.cc @@ -171,6 +171,7 @@ TEST_F(OptionsTest, GetOptionsFromMapTest) { {"allow_mmap_reads", "true"}, {"allow_mmap_writes", "false"}, {"use_direct_reads", "false"}, + {"use_direct_io_for_compaction_reads", "false"}, {"use_direct_io_for_flush_and_compaction", "false"}, {"is_fd_close_on_exec", "true"}, {"skip_log_error_on_recovery", "false"}, @@ -357,6 +358,7 @@ TEST_F(OptionsTest, GetOptionsFromMapTest) { ASSERT_EQ(new_db_opt.allow_mmap_reads, true); ASSERT_EQ(new_db_opt.allow_mmap_writes, false); ASSERT_EQ(new_db_opt.use_direct_reads, false); + ASSERT_EQ(new_db_opt.use_direct_io_for_compaction_reads, false); ASSERT_EQ(new_db_opt.use_direct_io_for_flush_and_compaction, false); ASSERT_EQ(new_db_opt.is_fd_close_on_exec, true); ASSERT_EQ(new_db_opt.stats_dump_period_sec, 46U); @@ -2706,6 +2708,7 @@ TEST_F(OptionsOldApiTest, GetOptionsFromMapTest) { {"allow_mmap_reads", "true"}, {"allow_mmap_writes", "false"}, {"use_direct_reads", "false"}, + {"use_direct_io_for_compaction_reads", "false"}, {"use_direct_io_for_flush_and_compaction", "false"}, {"is_fd_close_on_exec", "true"}, {"skip_log_error_on_recovery", "false"}, @@ -2897,6 +2900,7 @@ TEST_F(OptionsOldApiTest, GetOptionsFromMapTest) { ASSERT_EQ(new_db_opt.allow_mmap_reads, true); ASSERT_EQ(new_db_opt.allow_mmap_writes, false); ASSERT_EQ(new_db_opt.use_direct_reads, false); + ASSERT_EQ(new_db_opt.use_direct_io_for_compaction_reads, false); ASSERT_EQ(new_db_opt.use_direct_io_for_flush_and_compaction, false); ASSERT_EQ(new_db_opt.is_fd_close_on_exec, true); ASSERT_EQ(new_db_opt.stats_dump_period_sec, 46U); diff --git a/table/block_based/block_based_table_factory.cc b/table/block_based/block_based_table_factory.cc index 230b88cb48..bf1fa416c4 100644 --- a/table/block_based/block_based_table_factory.cc +++ b/table/block_based/block_based_table_factory.cc @@ -622,7 +622,8 @@ Status BlockBasedTableFactory::NewTableReader( table_reader_options.max_file_size_for_l0_meta_pin, table_reader_options.cur_db_session_id, table_reader_options.cur_file_num, table_reader_options.unique_id, - table_reader_options.user_defined_timestamps_persisted); + table_reader_options.user_defined_timestamps_persisted, + table_reader_options.avoid_shared_metadata_cache); } TableBuilder* BlockBasedTableFactory::NewTableBuilder( diff --git a/table/block_based/block_based_table_reader.cc b/table/block_based/block_based_table_reader.cc index 52bff36230..77006a9a00 100644 --- a/table/block_based/block_based_table_reader.cc +++ b/table/block_based/block_based_table_reader.cc @@ -868,7 +868,8 @@ Status BlockBasedTable::Open( BlockCacheTracer* const block_cache_tracer, size_t max_file_size_for_l0_meta_pin, const std::string& cur_db_session_id, uint64_t cur_file_num, UniqueId64x2 expected_unique_id, - const bool user_defined_timestamps_persisted) { + const bool user_defined_timestamps_persisted, + const bool avoid_shared_metadata_cache) { table_reader->reset(); Status s; @@ -887,8 +888,10 @@ Status BlockBasedTable::Open( ro.io_activity = read_options.io_activity; ro.fill_cache = read_options.fill_cache; - // prefetch both index and filters, down to all partitions - const bool prefetch_all = prefetch_index_and_filter_in_cache || level == 0; + // Prefetch both index and filters, down to all partitions. The L0 fast path + // is skipped when open-time reads must avoid the shared metadata cache. + const bool prefetch_all = prefetch_index_and_filter_in_cache || + (level == 0 && !avoid_shared_metadata_cache); const bool preload_all = !table_options.cache_index_and_filter_blocks; if (!ioptions.allow_mmap_reads && !env_options.use_mmap_reads) { @@ -1081,7 +1084,8 @@ Status BlockBasedTable::Open( s = new_table->PrefetchIndexAndFilterBlocks( ro, prefetch_buffer.get(), metaindex_iter.get(), new_table.get(), prefetch_all, table_options, level, file_size, - max_file_size_for_l0_meta_pin, &lookup_context); + max_file_size_for_l0_meta_pin, avoid_shared_metadata_cache, + &lookup_context); if (s.ok()) { // Update tail prefetch stats @@ -1335,7 +1339,7 @@ Status BlockBasedTable::PrefetchIndexAndFilterBlocks( InternalIterator* meta_iter, BlockBasedTable* new_table, bool prefetch_all, const BlockBasedTableOptions& table_options, const int level, size_t file_size, size_t max_file_size_for_l0_meta_pin, - BlockCacheLookupContext* lookup_context) { + bool avoid_shared_metadata_cache, BlockCacheLookupContext* lookup_context) { // Find filter handle and filter type if (rep_->filter_policy) { auto name = rep_->filter_policy->CompatibilityName(); @@ -1373,10 +1377,16 @@ Status BlockBasedTable::PrefetchIndexAndFilterBlocks( BlockBasedTableOptions::IndexType index_type = rep_->index_type; - const bool use_cache = table_options.cache_index_and_filter_blocks; + const bool use_cache = table_options.cache_index_and_filter_blocks && + !avoid_shared_metadata_cache; + [[maybe_unused]] std::pair shared_metadata_cache_context = { + avoid_shared_metadata_cache, use_cache}; + TEST_SYNC_POINT_CALLBACK( + "BlockBasedTable::PrefetchIndexAndFilterBlocks:SharedMetadataCache", + &shared_metadata_cache_context); - const bool maybe_flushed = - level == 0 && file_size <= max_file_size_for_l0_meta_pin; + const bool maybe_flushed = !avoid_shared_metadata_cache && level == 0 && + file_size <= max_file_size_for_l0_meta_pin; std::function is_pinned = [maybe_flushed, &is_pinned](PinningTier pinning_tier, PinningTier fallback_pinning_tier) { @@ -1400,16 +1410,20 @@ Status BlockBasedTable::PrefetchIndexAndFilterBlocks( assert(false); return false; }; - const bool pin_top_level_index = is_pinned( - table_options.metadata_cache_options.top_level_index_pinning, - table_options.pin_top_level_index_and_filter ? PinningTier::kAll - : PinningTier::kNone); + const bool pin_top_level_index = + !avoid_shared_metadata_cache && + is_pinned(table_options.metadata_cache_options.top_level_index_pinning, + table_options.pin_top_level_index_and_filter + ? PinningTier::kAll + : PinningTier::kNone); const bool pin_partition = + !avoid_shared_metadata_cache && is_pinned(table_options.metadata_cache_options.partition_pinning, table_options.pin_l0_filter_and_index_blocks_in_cache ? PinningTier::kFlushedAndSimilar : PinningTier::kNone); const bool pin_unpartitioned = + !avoid_shared_metadata_cache && is_pinned(table_options.metadata_cache_options.unpartitioned_pinning, table_options.pin_l0_filter_and_index_blocks_in_cache ? PinningTier::kFlushedAndSimilar @@ -1510,7 +1524,8 @@ Status BlockBasedTable::PrefetchIndexAndFilterBlocks( // The partitions of partitioned index are always stored in cache. They // are hence follow the configuration for pin and prefetch regardless of // the value of cache_index_and_filter_blocks - if (s.ok() && (prefetch_all || pin_partition)) { + if (s.ok() && !avoid_shared_metadata_cache && + (prefetch_all || pin_partition)) { s = rep_->index_reader->CacheDependencies(ro, pin_partition, prefetch_buffer); } @@ -1535,7 +1550,7 @@ Status BlockBasedTable::PrefetchIndexAndFilterBlocks( if (filter) { // Refer to the comment above about paritioned indexes always being cached - if (prefetch_all || pin_partition) { + if (!avoid_shared_metadata_cache && (prefetch_all || pin_partition)) { s = filter->CacheDependencies(ro, pin_partition, prefetch_buffer); if (!s.ok()) { return s; diff --git a/table/block_based/block_based_table_reader.h b/table/block_based/block_based_table_reader.h index 01a0ebc1cf..75ac5a4ce8 100644 --- a/table/block_based/block_based_table_reader.h +++ b/table/block_based/block_based_table_reader.h @@ -145,6 +145,8 @@ class BlockBasedTable : public TableReader { // are set. // @param force_direct_prefetch if true, always prefetching to RocksDB // buffer, rather than calling RandomAccessFile::Prefetch(). + // @param avoid_shared_metadata_cache if true, open-time index/filter/ + // dictionary reads must not insert into the shared block cache. static Status Open( const ReadOptions& ro, const ImmutableOptions& ioptions, const EnvOptions& env_options, @@ -166,7 +168,8 @@ class BlockBasedTable : public TableReader { size_t max_file_size_for_l0_meta_pin = 0, const std::string& cur_db_session_id = "", uint64_t cur_file_num = 0, UniqueId64x2 expected_unique_id = {}, - const bool user_defined_timestamps_persisted = true); + const bool user_defined_timestamps_persisted = true, + bool avoid_shared_metadata_cache = false); bool PrefixRangeMayMatch(const Slice& internal_key, const ReadOptions& read_options, @@ -569,12 +572,15 @@ class BlockBasedTable : public TableReader { const InternalKeyComparator& internal_comparator, BlockCacheLookupContext* lookup_context); // If index and filter blocks do not need to be pinned, `prefetch_all` - // determines whether they will be read and add to cache. + // determines whether they will be read and added to cache. When + // `avoid_shared_metadata_cache` is set, open-time metadata reads avoid the + // shared block cache regardless of pinning/prefetch policy. Status PrefetchIndexAndFilterBlocks( const ReadOptions& ro, FilePrefetchBuffer* prefetch_buffer, InternalIterator* meta_iter, BlockBasedTable* new_table, bool prefetch_all, const BlockBasedTableOptions& table_options, const int level, size_t file_size, size_t max_file_size_for_l0_meta_pin, + bool avoid_shared_metadata_cache, BlockCacheLookupContext* lookup_context); static BlockType GetBlockTypeForMetaBlockByName(const Slice& meta_block_name); diff --git a/table/table_builder.h b/table/table_builder.h index ec9f61bbf9..e3cee1455d 100644 --- a/table/table_builder.h +++ b/table/table_builder.h @@ -45,7 +45,8 @@ struct TableReaderOptions { size_t _max_file_size_for_l0_meta_pin = 0, const std::string& _cur_db_session_id = "", uint64_t _cur_file_num = 0, UniqueId64x2 _unique_id = {}, SequenceNumber _largest_seqno = 0, - uint64_t _tail_size = 0, bool _user_defined_timestamps_persisted = true) + uint64_t _tail_size = 0, bool _user_defined_timestamps_persisted = true, + bool _avoid_shared_metadata_cache = false) : ioptions(_ioptions), prefix_extractor(_prefix_extractor), compression_manager(_compression_manager), @@ -63,7 +64,8 @@ struct TableReaderOptions { unique_id(_unique_id), block_protection_bytes_per_key(_block_protection_bytes_per_key), tail_size(_tail_size), - user_defined_timestamps_persisted(_user_defined_timestamps_persisted) {} + user_defined_timestamps_persisted(_user_defined_timestamps_persisted), + avoid_shared_metadata_cache(_avoid_shared_metadata_cache) {} const ImmutableOptions& ioptions; const std::shared_ptr& prefix_extractor; @@ -103,6 +105,10 @@ struct TableReaderOptions { // Whether the key in the table contains user-defined timestamps. bool user_defined_timestamps_persisted; + + // Open-time metadata reads should not insert index/filter/dictionary blocks + // into the shared block cache. + bool avoid_shared_metadata_cache; }; struct TableBuilderOptions : public TablePropertiesCollectorFactory::Context { diff --git a/test_util/testutil.cc b/test_util/testutil.cc index cd6ce58afa..7dabfaeace 100644 --- a/test_util/testutil.cc +++ b/test_util/testutil.cc @@ -301,6 +301,7 @@ void RandomInitDBOptions(DBOptions* db_opt, Random* rnd) { db_opt->allow_mmap_reads = rnd->Uniform(2); db_opt->allow_mmap_writes = rnd->Uniform(2); db_opt->use_direct_reads = rnd->Uniform(2); + db_opt->use_direct_io_for_compaction_reads = rnd->Uniform(2); db_opt->use_direct_io_for_flush_and_compaction = rnd->Uniform(2); db_opt->create_if_missing = rnd->Uniform(2); db_opt->create_missing_column_families = rnd->Uniform(2); diff --git a/tools/db_bench_tool.cc b/tools/db_bench_tool.cc index 45f30a286b..1138309b95 100644 --- a/tools/db_bench_tool.cc +++ b/tools/db_bench_tool.cc @@ -269,6 +269,15 @@ DEFINE_string( DEFINE_int64(num, 1000000, "Number of key/values to place in database"); +DEFINE_int64(bgwriter_num, 0, + "If > 0, the background-writer thread used by readwhilewriting / " + "readwhilemerging / multireadwhilewriting writes random keys over " + "[0, bgwriter_num) instead of [0, num). Lets the reader thread " + "group operate on a small hot subset (--num) while the writer " + "spreads its puts across a much larger keyspace, which is what " + "drives continuous flushes and compaction. Designed for " + "benchmarking compaction-time effects on user reads."); + DEFINE_int64(numdistinct, 1000, "Number of distinct keys to use. Used in RandomWithVerify to " "read/write on fewer keys so that gets are more likely to find the" @@ -1718,6 +1727,11 @@ DEFINE_bool(mmap_write, ROCKSDB_NAMESPACE::Options().allow_mmap_writes, DEFINE_bool(use_direct_reads, ROCKSDB_NAMESPACE::Options().use_direct_reads, "Use O_DIRECT for reading data"); +DEFINE_bool(use_direct_io_for_compaction_reads, + ROCKSDB_NAMESPACE::Options().use_direct_io_for_compaction_reads, + "Use O_DIRECT for compaction-input SST reads only, while keeping " + "user reads buffered"); + DEFINE_bool(use_direct_io_for_flush_and_compaction, ROCKSDB_NAMESPACE::Options().use_direct_io_for_flush_and_compaction, "Use O_DIRECT for background flush and compaction writes"); @@ -4750,6 +4764,8 @@ class Benchmark { options.allow_mmap_reads = FLAGS_mmap_read; options.allow_mmap_writes = FLAGS_mmap_write; options.use_direct_reads = FLAGS_use_direct_reads; + options.use_direct_io_for_compaction_reads = + FLAGS_use_direct_io_for_compaction_reads; options.use_direct_io_for_flush_and_compaction = FLAGS_use_direct_io_for_flush_and_compaction; options.manual_wal_flush = FLAGS_manual_wal_flush; @@ -8140,7 +8156,11 @@ class Benchmark { } } - GenerateKeyFromInt(thread->rand.Next() % FLAGS_num, FLAGS_num, &key); + const int64_t bg_keyspace = + FLAGS_bgwriter_num > 0 ? FLAGS_bgwriter_num : FLAGS_num; + const int64_t num_keys = + FLAGS_use_existing_keys ? FLAGS_num : bg_keyspace; + GenerateKeyFromInt(thread->rand.Next() % bg_keyspace, num_keys, &key); Status s; Slice val = gen.Generate(); @@ -9807,6 +9827,12 @@ int db_bench_tool(int argc, char** argv, ToolHooks& hooks) { "settable\n"); db_bench_exit(1); } + if (FLAGS_use_existing_keys && FLAGS_bgwriter_num > FLAGS_num) { + fprintf(stderr, + "`-bgwriter_num` must be less than or equal to `-num` when " + "`-use_existing_keys` is set\n"); + db_bench_exit(1); + } FLAGS_value_size_distribution_type_e = StringToDistributionType(FLAGS_value_size_distribution_type.c_str()); diff --git a/tools/db_crashtest.py b/tools/db_crashtest.py index 8707946166..ee4c3f6137 100644 --- a/tools/db_crashtest.py +++ b/tools/db_crashtest.py @@ -263,6 +263,7 @@ default_params = { "top_level_index_pinning": lambda: random.randint(0, 3), "unpartitioned_pinning": lambda: random.randint(0, 3), "use_direct_reads": lambda: random.randint(0, 1), + "use_direct_io_for_compaction_reads": lambda: random.randint(0, 1), "use_direct_io_for_flush_and_compaction": lambda: random.randint(0, 1), "use_sqfc_for_range_queries": lambda: random.choice([0, 1, 1, 1]), "mock_direct_io": False, @@ -933,6 +934,7 @@ def finalize_and_sanitize(src_params): if dest_params["mmap_read"] == 1: dest_params["use_direct_io_for_flush_and_compaction"] = 0 dest_params["use_direct_reads"] = 0 + dest_params["use_direct_io_for_compaction_reads"] = 0 dest_params["multiscan_use_async_io"] = 0 if dest_params.get("min_tombstones_for_range_conversion", 0) > 0: # SQFC range-query filtering installs ReadOptions::table_filter on @@ -945,13 +947,16 @@ def finalize_and_sanitize(src_params): if ( dest_params["use_direct_io_for_flush_and_compaction"] == 1 or dest_params["use_direct_reads"] == 1 + or dest_params["use_direct_io_for_compaction_reads"] == 1 ) and not is_direct_io_supported(dest_params["db"]): if is_release_mode(): print( - "{} does not support direct IO. Disabling use_direct_reads and " + "{} does not support direct IO. Disabling use_direct_reads, " + "use_direct_io_for_compaction_reads and " "use_direct_io_for_flush_and_compaction.\n".format(dest_params["db"]) ) dest_params["use_direct_reads"] = 0 + dest_params["use_direct_io_for_compaction_reads"] = 0 dest_params["use_direct_io_for_flush_and_compaction"] = 0 else: dest_params["mock_direct_io"] = True diff --git a/unreleased_history/behavior_changes/posix_optimize_for_compaction_table_read_delegates_to_base.md b/unreleased_history/behavior_changes/posix_optimize_for_compaction_table_read_delegates_to_base.md new file mode 100644 index 0000000000..03b64a4617 --- /dev/null +++ b/unreleased_history/behavior_changes/posix_optimize_for_compaction_table_read_delegates_to_base.md @@ -0,0 +1 @@ +`PosixFileSystem::OptimizeForCompactionTableRead` now delegates to the base `FileSystem::OptimizeForCompactionTableRead` implementation before applying its Linux-only compaction-readahead clamp (added for #12038). The returned `FileOptions::use_direct_reads` therefore reflects `DBOptions::use_direct_reads`, consistent with the base class and with other `FileSystem` implementations. Previously the override ignored the base implementation and keyed both the returned options and the Linux readahead clamp off the incoming `FileOptions` rather than `DBOptions`. At the in-tree call site the incoming `FileOptions::use_direct_reads` already matches `DBOptions::use_direct_reads`, so this is effectively a no-op for existing configurations; it removes a latent inconsistency where a caller passing `FileOptions` whose `use_direct_reads` disagreed with `DBOptions` would have had the global flag silently ignored for compaction-input reads on Linux. (Note: the `#12038` readahead clamp still keys off the global `use_direct_reads`; it is not skipped for the compaction-only `use_direct_io_for_compaction_reads` path, since that flag enables O_DIRECT after this hook runs.) diff --git a/unreleased_history/new_features/use_direct_io_for_compaction_reads.md b/unreleased_history/new_features/use_direct_io_for_compaction_reads.md new file mode 100644 index 0000000000..61690e83d0 --- /dev/null +++ b/unreleased_history/new_features/use_direct_io_for_compaction_reads.md @@ -0,0 +1 @@ +Added `DBOptions::use_direct_io_for_compaction_reads` (default false). When enabled, compaction-input SST reads use `O_DIRECT` while user reads remain buffered, avoiding page-cache eviction of hot user-read data by sequential compaction scans. Pair with `use_direct_io_for_flush_and_compaction = true` on write-heavy workloads for direct I/O on both compaction inputs and outputs. Rejected at Open when combined with `allow_mmap_reads = true`. No-op when `use_direct_reads = true` is already set (since all reads are already direct).