Allow skipping lmax index and filter block prefetches for external file ingestion (#14859)

Summary:

External SST ingestion can spend noticeable foreground commit time opening the newly ingested table reader and prefetching index/filter blocks, especially for large files going directly to the last level. This adds `IngestExternalFileOptions::prefetch_lmax_index_and_filter_blocks`, defaulting to existing behavior, so bulk-load callers can skip that commit-time metadata prefetch when they do not need the cache warmed immediately.

The option is threaded through prepared ingestion handles into the manifest writer path and drives the existing `LoadTableHandlers()` `prefetch_index_and_filter_in_cache` argument. 

Benchmarks:

```
db_bench --benchmarks=ingestexternalfile --num=2200000 --ingest_external_file_num_batches=1 --ingest_external_file_batch_size=1 --ingest_external_file_use_file_info=true --ingest_external_file_fill_cache=true --cache_index_and_filter_blocks=true --bloom_bits=10 --statistics=true --stats_level=3 --ingest_external_file_prefetch_lmax_index_and_filter_blocks=<true|false>
```

| `ingest_external_file_prefetch_lmax_index_and_filter_blocks` | `rocksdb.ingest.external.file.run.micros` | `rocksdb.table.open.io.micros` | index/filter cache adds | last-level read bytes |
| `true` | `3459 us` | `2560 us` | `1 / 1` | `3551559` |
| `false` | `2112 us` | `1145 us` | `0 / 0` | `1333` |

Differential Revision: D108678511

Author: joshkang97

db/external_sst_file_ingestion_job.cc

@@ -44,6 +44,8 @@ Status ExternalSstFileIngestionJob::Prepare(
     IngestedFileInfo file_to_ingest;
     // For temperature, first assume it matches provided hint
     file_to_ingest.file_temperature = file_temperature;
+    file_to_ingest.prefetch_lmax_index_and_filter_blocks =
+        ingestion_options_.prefetch_lmax_index_and_filter_blocks;
     const PreparedFileInfo* prepared_file_info =
         file_infos.empty() ? nullptr : file_infos[i];
     status = GetIngestedFileInfo(file_path, next_file_number++,
@@ -748,6 +750,9 @@ Status ExternalSstFileIngestionJob::AssignLevelsForOneBatch(
         tail_size, file->user_defined_timestamps_persisted, "", "");
     f_metadata.temperature = file->file_temperature;
     f_metadata.marked_for_compaction = marked_for_compaction;
+    f_metadata.skip_index_and_filter_blocks_prefetch =
+        !file->prefetch_lmax_index_and_filter_blocks &&
+        file->picked_level == cfd_->NumberLevels() - 1;
     // Extract min/max timestamps from table properties for UDT support.
     // This ensures ingested files have proper timestamp ranges in FileMetaData,
     // similar to files created by flush and compaction.

db/external_sst_file_ingestion_job.h

@@ -181,6 +181,9 @@ struct IngestedFileInfo : public KeyRangeInfo {
   // setting.
   bool user_defined_timestamps_persisted = true;
 
+  // Whether Lmax commit-time table opening should prefetch index/filter blocks.
+  bool prefetch_lmax_index_and_filter_blocks = true;
+
   SequenceNumber largest_seqno = kMaxSequenceNumber;
   SequenceNumber smallest_seqno = kMaxSequenceNumber;
 };

db/external_sst_file_test.cc

@@ -465,6 +465,85 @@ TEST_F(ExternalSSTFileTest, ParallelFileOpenWithFileOpeningThreads) {
   }
 }
 
+TEST_F(ExternalSSTFileTest, LmaxPrefetchSkipDoesNotDisableL0Prefetch) {
+  LRUCacheOptions co;
+  co.capacity = 32 << 20;
+  std::shared_ptr<Cache> cache = NewLRUCache(co);
+  BlockBasedTableOptions table_options;
+  table_options.block_cache = cache;
+  table_options.cache_index_and_filter_blocks = true;
+  table_options.filter_policy.reset(NewBloomFilterPolicy(10));
+
+  Options options = CurrentOptions();
+  options.disable_auto_compactions = true;
+  options.max_open_files = -1;
+  options.num_levels = 2;
+  options.optimize_filters_for_hits = false;
+  options.statistics = CreateDBStatistics();
+  options.table_factory.reset(NewBlockBasedTableFactory(table_options));
+  DestroyAndReopen(options);
+
+  const auto write_file = [&](const std::string& file_path,
+                              const std::string& value,
+                              ExternalSstFileInfo* file_info) {
+    SstFileWriter writer(EnvOptions(), options);
+    ASSERT_OK(writer.Open(file_path));
+    ASSERT_OK(writer.Put(Key(10), value));
+    ASSERT_OK(writer.Finish(file_info));
+  };
+
+  const auto ingest_file = [&](const std::string& file_path,
+                               const ExternalSstFileInfo& file_info,
+                               bool fail_if_not_bottommost_level) {
+    IngestExternalFileArg arg;
+    arg.column_family = db_->DefaultColumnFamily();
+    arg.external_files = {file_path};
+    arg.file_infos = {file_info.prepared_file_info.get()};
+    arg.options.fail_if_not_bottommost_level = fail_if_not_bottommost_level;
+    arg.options.verify_checksums_before_ingest = false;
+    arg.options.prefetch_lmax_index_and_filter_blocks = false;
+    ASSERT_OK(db_->IngestExternalFiles({arg}));
+  };
+
+  const std::string lmax_file_path = sst_files_dir_ + "lazy_lmax.sst";
+  ExternalSstFileInfo lmax_file_info;
+  write_file(lmax_file_path, "v10", &lmax_file_info);
+
+  ASSERT_EQ(0, options.statistics->getAndResetTickerCount(
+                   Tickers::BLOCK_CACHE_INDEX_ADD));
+  ASSERT_EQ(0, options.statistics->getAndResetTickerCount(
+                   Tickers::BLOCK_CACHE_FILTER_ADD));
+
+  ingest_file(lmax_file_path, lmax_file_info,
+              true /* fail_if_not_bottommost_level */);
+  ASSERT_EQ("0,1", FilesPerLevel());
+
+  ASSERT_EQ(0, options.statistics->getAndResetTickerCount(
+                   Tickers::BLOCK_CACHE_INDEX_ADD));
+  ASSERT_EQ(0, options.statistics->getAndResetTickerCount(
+                   Tickers::BLOCK_CACHE_FILTER_ADD));
+
+  const std::string l0_file_path = sst_files_dir_ + "l0.sst";
+  ExternalSstFileInfo l0_file_info;
+  write_file(l0_file_path, "v11", &l0_file_info);
+
+  ASSERT_EQ(0, options.statistics->getAndResetTickerCount(
+                   Tickers::BLOCK_CACHE_INDEX_ADD));
+  ASSERT_EQ(0, options.statistics->getAndResetTickerCount(
+                   Tickers::BLOCK_CACHE_FILTER_ADD));
+
+  ingest_file(l0_file_path, l0_file_info,
+              false /* fail_if_not_bottommost_level */);
+  ASSERT_EQ("1,1", FilesPerLevel());
+
+  EXPECT_GT(options.statistics->getAndResetTickerCount(
+                Tickers::BLOCK_CACHE_INDEX_ADD),
+            0);
+  EXPECT_GT(options.statistics->getAndResetTickerCount(
+                Tickers::BLOCK_CACHE_FILTER_ADD),
+            0);
+}
+
 TEST_F(ExternalSSTFileTest, AbortPreparedIngestion) {
   Options options = CurrentOptions();
   DestroyAndReopen(options);

db/version_edit.h

@@ -337,6 +337,11 @@ struct FileMetaData {
   // via FileOptions::file_metadata on subsequent opens. Empty if not available.
   std::string file_open_metadata;
 
+  // Skips prefetching index and filter blocks into block cache on file open,
+  // not persisted in MANIFEST. NOTE: false does not guarantee prefetching
+  // either (e.g. when index and filter blocks are not cached in block cache).
+  bool skip_index_and_filter_blocks_prefetch = false;
+
   FileMetaData() = default;
 
   FileMetaData(uint64_t file, uint32_t file_path_id, uint64_t file_size,

db/version_util.cc

@@ -62,11 +62,14 @@ Status LoadTableHandlersHelper(
 
       TableCache::TypedHandle* handle = nullptr;
       TableReader* table_reader = nullptr;
+      const bool prefetch_index_and_filter_for_file =
+          prefetch_index_and_filter_in_cache &&
+          !file_meta->skip_index_and_filter_blocks_prefetch;
       auto status = table_cache->FindTable(
           read_options, file_options, internal_comparator, *file_meta, &handle,
           mutable_cf_options, &table_reader, false /* no_io */,
           internal_stats->GetFileReadHist(level), false /* skip_filters */,
-          level, prefetch_index_and_filter_in_cache,
+          level, prefetch_index_and_filter_for_file,
           max_file_size_for_l0_meta_pin, file_meta->temperature,
           true /* pin_table_handle */);
 

db_stress_tool/no_batched_ops_stress.cc

@@ -2511,6 +2511,8 @@ class NonBatchedOpsStressTest : public StressTest {
           thread->rand.OneInOpt(2) ? 1024 * 1024 : 0;
       ingest_options.fill_cache = thread->rand.OneInOpt(4);
       ingest_options.file_opening_threads = 1 + thread->rand.Uniform(4);
+      ingest_options.prefetch_lmax_index_and_filter_blocks =
+          !thread->rand.OneInOpt(4);
       const bool use_prepare_commit = thread->rand.OneInOpt(
           FLAGS_ingest_external_file_prepare_commit_one_in);
       const bool use_separate_prepare_calls = use_prepare_commit &&
@@ -2524,6 +2526,8 @@ class NonBatchedOpsStressTest : public StressTest {
                          << ", fill_cache: " << ingest_options.fill_cache
                          << ", file_opening_threads: "
                          << ingest_options.file_opening_threads
+                         << ", prefetch_lmax_index_and_filter_blocks: "
+                         << ingest_options.prefetch_lmax_index_and_filter_blocks
                          << ", ingest_external_file_data_file_count: "
                          << data_file_count
                          << ", num_external_files: " << external_files.size()

include/rocksdb/options.h

@@ -2952,6 +2952,13 @@ struct IngestExternalFileOptions {
   // ingestion options.
   bool fill_cache = true;
 
+  // Controls whether external file ingestion should prefetch index and filter
+  // blocks while opening table readers during commit. Setting this to false can
+  // reduce commit latency for bulk loads into Lmax when
+  // (BlockBasedTableOptions::cache_index_and_filter_blocks=true or partitioned
+  // filters/indexes are enabled).
+  bool prefetch_lmax_index_and_filter_blocks = true;
+
   // Maximum number of threads used to open table readers for the files being
   // ingested during commit, can speed up ingestion performance, when ingesting
   // multiple files at once.

tools/db_bench_tool.cc

@@ -1352,6 +1352,12 @@ DEFINE_bool(ingest_external_file_fill_cache,
             "If true, the ingestexternalfile benchmark allows file ingestion "
             "reads to populate block cache.");
 
+DEFINE_bool(ingest_external_file_prefetch_lmax_index_and_filter_blocks,
+            ROCKSDB_NAMESPACE::IngestExternalFileOptions()
+                .prefetch_lmax_index_and_filter_blocks,
+            "If true, the ingestexternalfile benchmark prefetches index and "
+            "filter blocks while opening table readers during commit.");
+
 DEFINE_uint64(
     initial_auto_readahead_size,
     ROCKSDB_NAMESPACE::BlockBasedTableOptions().initial_auto_readahead_size,
@@ -9710,6 +9716,8 @@ class Benchmark {
       ingest_options.move_files = true;
       ingest_options.file_opening_threads = file_opening_threads;
       ingest_options.fill_cache = FLAGS_ingest_external_file_fill_cache;
+      ingest_options.prefetch_lmax_index_and_filter_blocks =
+          FLAGS_ingest_external_file_prefetch_lmax_index_and_filter_blocks;
       if (use_file_info) {
         // Reuse the writer's metadata so ingestion skips re-opening/scanning.
         IngestExternalFileArg arg;

unreleased_history/performance_improvements/ingest_external_file_lmax_prefetch.md

@@ -0,0 +1 @@
+Reduced commit latency for large external file ingestions into the last level by adding `IngestExternalFileOptions::prefetch_lmax_index_and_filter_blocks`, which can skip commit-time index and filter block prefetching for cache-backed table metadata.