Pass file metadata to IngestExternalFile to improve ingestion latency (#14837)

Summary:
Pull Request resolved: #14837

External file ingestion (`DB::IngestExternalFile[s]`) re-opens every SST file and scans it -- footer, properties, index, filter, and the first/last data blocks -- to recompute the boundary keys, sequence-number bounds, and table properties before committing. For cold, I/O-bound files this scan dominates ingest latency, even when the file is moved/linked rather than copied.

This change lets a caller skip that work when it already has the file's metadata. `SstFileWriter::Finish()` now returns a `PreparedFileInfo` with the file size, table properties, and prepared internal `smallest`/`largest` bounds, and a new `IngestExternalFileArg::file_infos` field carries one `PreparedFileInfo` per file into `IngestExternalFiles()`. When set, ingestion reuses that metadata instead of re-opening and scanning each file. The file is still copied/linked, and the checksum is still verified when `verify_checksums_before_ingest` is set (the fast path opens the file only for that). Point-key and range-deletion bounds are folded into the same prepared bound pair, and user-defined-timestamp files (including the "UDT in Memtables only" format, whose boundary keys carry no timestamp) are handled.

Internally, ingestion-job metadata acquisition was split into `GetIngestedFileInfoFromFileInfo` (reuse caller metadata) and `GetIngestedFileInfoFromFile` (open + scan). Prepared boundary updates use RocksDB comparators, and the timestamp-stripping path pads prepared bounds back to the internal timestamp shape before installing the file.

The `ingestexternalfile` `db_bench` benchmark now also exposes `--ingest_external_file_fill_cache` so ingestion-read block-cache effects can be controlled independently while comparing the file-info fast path against the normal open-and-scan path.

A future PR will support creating `PreparedFileInfo` from a existing DB-generated file, so this optimization is not just limited to SstFileWriter.

## Benchmark Results

Existing `db_bench ingestexternalfile` benchmark, release build, using an XFS flash-backed filesystem. Files are linked (`move_files`) so the measurement isolates the ingest path rather than file-copy throughput. These numbers used direct reads and `fill_cache=false` to avoid warming the block cache while comparing the file-info fast path against the normal open-and-scan path.

Each run used 1M keys/SST, one ingest call, and `db_bench` reported a 62.9 MB estimated file size. The comparison varied the number of files in that ingest call across 10, 30, and 50 files.

| Files/batch | Config | Prepare P50 | Run P50 | Run P50/file | Total ingestion P50 | Total P50 drop |
| --- | --- | --- | --- | --- | --- | --- |
| 10 | Baseline | 10.909 ms | 7.593 ms | 0.759 ms/file | 18.502 ms | -- |
| 10 | `file_info=true` | 1.127 ms | 7.541 ms | 0.754 ms/file | 8.668 ms | 53.2% |
| 30 | Baseline | 29.049 ms | 23.161 ms | 0.772 ms/file | 52.210 ms | -- |
| 30 | `file_info=true` | 2.734 ms | 23.384 ms | 0.779 ms/file | 26.118 ms | 50.0% |
| 50 | Baseline | 49.036 ms | 39.224 ms | 0.784 ms/file | 88.260 ms | -- |
| 50 | `file_info=true` | 4.133 ms | 39.417 ms | 0.788 ms/file | 43.550 ms | 50.7% |

With `fill_cache=false` and direct reads, the metadata fast path cuts `Prepare()` by roughly 90-92%. `Run()` is roughly 0.75-0.79 ms per file across these runs. End-to-end `db_bench` micros/op is almost unchanged because this benchmark includes SST generation and compaction work; use `rocksdb.ingest.external.file.prepare.micros` and `rocksdb.ingest.external.file.run.micros` to isolate ingestion.

Benchmark args: `--benchmarks=ingestexternalfile --use_existing_db=0 --num=1000000 --compression_type=none --statistics --use_direct_reads=true --ingest_external_file_batch_size=<10|30|50> --ingest_external_file_num_batches=1 --ingest_external_file_use_file_info=<false|true> --ingest_external_file_fill_cache=false`.

Reviewed By: xingbowang

Differential Revision: D107721261

fbshipit-source-id: b06ecb7b35f260ec02acf837e7986057bc23cdbf

Author: joshkang97

db/db_impl/db_impl.cc

@@ -6815,6 +6815,17 @@ Status DBImpl::PrepareFileIngestion(
           "external_files[" + std::to_string(i) + "] is empty";
       return Status::InvalidArgument(err_msg);
     }
+    if (!args[i].file_infos.empty()) {
+      if (args[i].file_infos.size() != args[i].external_files.size()) {
+        return Status::InvalidArgument("file_infos[" + std::to_string(i) +
+                                       "] size must match external_files[" +
+                                       std::to_string(i) + "] size");
+      }
+      if (args[i].options.write_global_seqno) {
+        return Status::InvalidArgument(
+            "write_global_seqno is not supported when file_infos is set");
+      }
+    }
     if (i && args[i].options.fill_cache != args[i - 1].options.fill_cache) {
       return Status::InvalidArgument(
           "fill_cache should be the same across ingestion options.");
@@ -6909,8 +6920,9 @@ Status DBImpl::PrepareFileIngestion(
             this);
     Status es = ingestion_jobs[i].Prepare(
         args[i].external_files, args[i].files_checksums,
-        args[i].files_checksum_func_names, args[i].atomic_replace_range,
-        args[i].file_temperature, start_file_number, super_version);
+        args[i].files_checksum_func_names, args[i].file_infos,
+        args[i].atomic_replace_range, args[i].file_temperature,
+        start_file_number, super_version);
     // capture first error only
     if (!es.ok() && status.ok()) {
       status = es;
@@ -6925,8 +6937,9 @@ Status DBImpl::PrepareFileIngestion(
             this);
     Status es = ingestion_jobs[0].Prepare(
         args[0].external_files, args[0].files_checksums,
-        args[0].files_checksum_func_names, args[0].atomic_replace_range,
-        args[0].file_temperature, next_file_number, super_version);
+        args[0].files_checksum_func_names, args[0].file_infos,
+        args[0].atomic_replace_range, args[0].file_temperature,
+        next_file_number, super_version);
     if (!es.ok()) {
       status = es;
     }

db/external_sst_file_ingestion_job.cc

@@ -248,6 +248,7 @@ class ExternalSstFileIngestionJob {
   Status Prepare(const std::vector<std::string>& external_files_paths,
                  const std::vector<std::string>& files_checksums,
                  const std::vector<std::string>& files_checksum_func_names,
+                 const std::vector<const PreparedFileInfo*>& file_infos,
                  const std::optional<RangeOpt>& atomic_replace_range,
                  const Temperature& file_temperature, uint64_t next_file_number,
                  SuperVersion* sv);
@@ -323,17 +324,37 @@ class ExternalSstFileIngestionJob {
   // different options. For example: when external file does not contain
   // timestamps while column family enables UDT in Memtables only feature.
   Status SanityCheckTableProperties(const std::string& external_file,
-                                    uint64_t new_file_number, SuperVersion* sv,
-                                    IngestedFileInfo* file_to_ingest,
-                                    std::unique_ptr<TableReader>* table_reader);
+                                    const TableProperties& props,
+                                    IngestedFileInfo* file_to_ingest);
 
   // Open the external file and populate `file_to_ingest` with all the
-  // external information we need to ingest this file.
+  // external information we need to ingest this file. When
+  // `prepared_file_info` is non-null, its caller-supplied metadata is reused
+  // instead of opening and scanning the file.
   Status GetIngestedFileInfo(const std::string& external_file,
                              uint64_t new_file_number,
+                             const PreparedFileInfo* prepared_file_info,
                              IngestedFileInfo* file_to_ingest,
                              SuperVersion* sv);
 
+  // Acquire the per-file metadata from the caller-supplied opaque
+  // `PreparedFileInfo` (produced by SstFileWriter::Finish) instead of opening
+  // the file.
+  Status GetIngestedFileInfoFromFileInfo(
+      const std::string& external_file,
+      const PreparedFileInfo& prepared_file_info,
+      IngestedFileInfo* file_to_ingest);
+
+  // Acquire the per-file metadata by opening the external file and scanning it
+  // (table properties, sequence number bounds, and boundary keys including any
+  // range-tombstone extensions). Used when no file_info is available. The
+  // opened `TableReader` is returned via `*table_reader` so the caller can
+  // reuse it (e.g. to verify the file checksum) without re-opening the file.
+  Status GetIngestedFileInfoFromFile(
+      const std::string& external_file, uint64_t new_file_number,
+      IngestedFileInfo* file_to_ingest, SuperVersion* sv,
+      std::unique_ptr<TableReader>* out_table_reader);
+
   // If the input files' key range overlaps themselves, this function divides
   // them in the user specified order into multiple batches. Where the files
   // within a batch do not overlap with each other, but key range could overlap