[rocksdb] Pass file metadata to IngestExternalFile to improve ingestion latency

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 populates the full `ExternalSstFileInfo` (its `smallest_internal_key`, `largest_internal_key`, and `table_properties`), and a new `IngestExternalFileArg::file_infos` field carries one `ExternalSstFileInfo` 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). Range tombstones 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).

## Benchmark results

`db_bench` `ingestexternalfile` benchmark, release build, on SSD (btrfs). Files are linked (`move_files`) so the measurement isolates the ingest path rather than file-copy throughput.

| Config | Ingest latency / call (P50) |
| --- | --- |
| Baseline (open + scan each file) | 28,111 us |
| `--ingest_external_file_use_file_info=true` | 18,444 us |

~34% lower median ingest latency by skipping the per-file open + scan (a warm/page-cache number; the gain is larger for cold files).

Command:

```
make clean && DEBUG_LEVEL=0 make -j$(nproc) db_bench
B=/data/$USER/ingest_bench
for fi in false true; do
  rm -rf $B
  ./db_bench --benchmarks=ingestexternalfile --num=1000000 --value_size=100 \
    --compression_type=none --ingest_external_file_batch_size=10 \
    --ingest_external_file_num_batches=10 --disable_auto_compactions=1 \
    --statistics --db=$B --ingest_external_file_use_file_info=$fi 2>&1 \
    | grep rocksdb.ingest.external.file.micros
done
```

Differential Revision: [D107721261](https://our.internmc.facebook.com/intern/diff/D107721261/)

**NOTE FOR REVIEWERS**: This PR has internal Meta-specific changes or comments, please review them on [Phabricator](https://our.internmc.facebook.com/intern/diff/D107721261/)!
ghstack-source-id: 390599675
Pull-Request: https://github.com/facebook/rocksdb/pull/14835

db/db_impl/db_impl.cc

@@ -6675,6 +6675,13 @@ Status DBImpl::IngestExternalFile(
 Status DBImpl::IngestExternalFiles(
     const std::vector<IngestExternalFileArg>& args) {
   PERF_TIMER_GUARD(file_ingestion_nanos);
+  // Prepare/run ingestion latency, recorded only on success (below).
+  const bool record_ingest_micros =
+      stats_ != nullptr &&
+      stats_->get_stats_level() > StatsLevel::kExceptTimers;
+  const uint64_t ingest_start_micros =
+      record_ingest_micros ? immutable_db_options_.clock->NowMicros() : 0;
+  uint64_t prepare_micros = 0;
   // TODO: plumb Env::IOActivity, Env::IOPriority
   const WriteOptions write_options;
 
@@ -6701,6 +6708,27 @@ Status DBImpl::IngestExternalFiles(
           "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");
+      }
+      for (const auto* file_info : args[i].file_infos) {
+        if (file_info == nullptr) {
+          return Status::InvalidArgument(
+              "file_infos[" + std::to_string(i) +
+              "] contains a null PreparedFileInfo pointer; each entry must "
+              "point to a handle from SstFileWriter::Finish");
+        }
+      }
+      // file_infos avoids opening the file, so it cannot write a global seqno
+      // back into it. (write_global_seqno is deprecated and defaults to false.)
+      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.");
@@ -6794,8 +6822,9 @@ Status DBImpl::IngestExternalFiles(
             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;
@@ -6810,8 +6839,9 @@ Status DBImpl::IngestExternalFiles(
             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;
     }
@@ -6826,6 +6856,13 @@ Status DBImpl::IngestExternalFiles(
     return status;
   }
 
+  // End of prepare phase; run phase starts here.
+  uint64_t run_start_micros = 0;
+  if (record_ingest_micros) {
+    run_start_micros = immutable_db_options_.clock->NowMicros();
+    prepare_micros = run_start_micros - ingest_start_micros;
+  }
+
   std::vector<SuperVersionContext> sv_ctxs;
   for (size_t i = 0; i != num_cfs; ++i) {
     sv_ctxs.emplace_back(true /* create_superversion */);
@@ -7067,6 +7104,14 @@ Status DBImpl::IngestExternalFiles(
       }
     }
   }
+  // Record latency only for successful ingestions.
+  if (record_ingest_micros && status.ok()) {
+    RecordTimeToHistogram(stats_, INGEST_EXTERNAL_FILE_PREPARE_TIME,
+                          prepare_micros);
+    RecordTimeToHistogram(
+        stats_, INGEST_EXTERNAL_FILE_RUN_TIME,
+        immutable_db_options_.clock->NowMicros() - run_start_micros);
+  }
   return status;
 }
 

db/external_sst_file_ingestion_job.cc

@@ -19,6 +19,7 @@
 #include "logging/logging.h"
 #include "monitoring/statistics_impl.h"
 #include "table/merging_iterator.h"
+#include "table/prepared_file_info.h"
 #include "table/sst_file_writer_collectors.h"
 #include "table/table_builder.h"
 #include "table/unique_id_impl.h"
@@ -31,18 +32,22 @@ Status ExternalSstFileIngestionJob::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) {
   Status status;
 
-  // Read the information of files we are ingesting
-  for (const std::string& file_path : external_files_paths) {
+  // Read the information of files we are ingesting.
+  for (size_t i = 0; i < external_files_paths.size(); i++) {
+    const std::string& file_path = external_files_paths[i];
     IngestedFileInfo file_to_ingest;
     // For temperature, first assume it matches provided hint
     file_to_ingest.file_temperature = file_temperature;
-    status =
-        GetIngestedFileInfo(file_path, next_file_number++, &file_to_ingest, sv);
+    const PreparedFileInfo* file_info =
+        file_infos.empty() ? nullptr : file_infos[i];
+    status = GetIngestedFileInfo(file_path, next_file_number++, file_info,
+                                 &file_to_ingest, sv);
     if (!status.ok()) {
       ROCKS_LOG_WARN(db_options_.info_log,
                      "Failed to get ingested file info: %s: %s",
@@ -970,18 +975,14 @@ Status ExternalSstFileIngestionJob::ResetTableReader(
 }
 
 Status ExternalSstFileIngestionJob::SanityCheckTableProperties(
-    const std::string& external_file, uint64_t new_file_number,
-    SuperVersion* sv, IngestedFileInfo* file_to_ingest,
-    std::unique_ptr<TableReader>* table_reader) {
-  // Get the external file properties
-  auto props = table_reader->get()->GetTableProperties();
-  assert(props.get());
-  const auto& uprops = props->user_collected_properties;
+    const std::string& external_file, const TableProperties& props,
+    IngestedFileInfo* file_to_ingest) {
+  const auto& uprops = props.user_collected_properties;
 
   // Get table version
   auto version_iter = uprops.find(ExternalSstFilePropertyNames::kVersion);
   if (version_iter == uprops.end()) {
-    assert(!SstFileWriter::CreatedBySstFileWriter(*props));
+    assert(!SstFileWriter::CreatedBySstFileWriter(props));
     if (!ingestion_options_.allow_db_generated_files) {
       return Status::Corruption("External file version not found");
     } else {
@@ -990,7 +991,7 @@ Status ExternalSstFileIngestionJob::SanityCheckTableProperties(
       file_to_ingest->version = 0;
     }
   } else {
-    assert(SstFileWriter::CreatedBySstFileWriter(*props));
+    assert(SstFileWriter::CreatedBySstFileWriter(props));
     file_to_ingest->version = DecodeFixed32(version_iter->second.c_str());
   }
 
@@ -1004,12 +1005,17 @@ Status ExternalSstFileIngestionJob::SanityCheckTableProperties(
 
     // Set the global sequence number
     file_to_ingest->original_seqno = DecodeFixed64(seqno_iter->second.c_str());
-    if (props->external_sst_file_global_seqno_offset == 0) {
-      file_to_ingest->global_seqno_offset = 0;
+    file_to_ingest->global_seqno_offset =
+        static_cast<size_t>(props.external_sst_file_global_seqno_offset);
+    // The on-disk offset is only needed if we will write the global seqno back
+    // into the file (write_global_seqno). The metadata fast-path does not open
+    // the file, and its in-memory table properties do not carry the offset
+    // (it is only computed while reading the file back); that is fine as long
+    // as write_global_seqno is not requested.
+    if (ingestion_options_.write_global_seqno &&
+        file_to_ingest->global_seqno_offset == 0) {
       return Status::Corruption("Was not able to find file global seqno field");
     }
-    file_to_ingest->global_seqno_offset =
-        static_cast<size_t>(props->external_sst_file_global_seqno_offset);
   } else if (file_to_ingest->version == 1) {
     // SST file V1 should not have global seqno field
     assert(seqno_iter == uprops.end());
@@ -1030,23 +1036,20 @@ Status ExternalSstFileIngestionJob::SanityCheckTableProperties(
                                    " is not supported");
   }
 
-  file_to_ingest->cf_id = static_cast<uint32_t>(props->column_family_id);
-  // This assignment works fine even though `table_reader` may later be reset,
-  // since that will not affect how table properties are parsed, and this
-  // assignment is making a copy.
-  file_to_ingest->table_properties = *props;
+  file_to_ingest->cf_id = static_cast<uint32_t>(props.column_family_id);
+  file_to_ingest->table_properties = props;
 
   // Get number of entries in table
-  file_to_ingest->num_entries = props->num_entries;
-  file_to_ingest->num_range_deletions = props->num_range_deletions;
+  file_to_ingest->num_entries = props.num_entries;
+  file_to_ingest->num_range_deletions = props.num_range_deletions;
 
   // Validate table properties related to comparator name and user defined
   // timestamps persisted flag.
   file_to_ingest->user_defined_timestamps_persisted =
-      static_cast<bool>(props->user_defined_timestamps_persisted);
+      static_cast<bool>(props.user_defined_timestamps_persisted);
   bool mark_sst_file_has_no_udt = false;
   Status s = ValidateUserDefinedTimestampsOptions(
-      cfd_->user_comparator(), props->comparator_name,
+      cfd_->user_comparator(), props.comparator_name,
       cfd_->ioptions().persist_user_defined_timestamps,
       file_to_ingest->user_defined_timestamps_persisted,
       &mark_sst_file_has_no_udt);
@@ -1054,7 +1057,9 @@ Status ExternalSstFileIngestionJob::SanityCheckTableProperties(
     // A column family that enables user-defined timestamps in Memtable only
     // feature can also ingest external files created by a setting that disables
     // user-defined timestamps. In that case, we need to re-mark the
-    // user_defined_timestamps_persisted flag for the file.
+    // user_defined_timestamps_persisted flag for the file. The open-and-scan
+    // caller is then responsible for reopening its `TableReader` with the
+    // updated flag.
     file_to_ingest->user_defined_timestamps_persisted = false;
   } else if (!s.ok()) {
     ROCKS_LOG_WARN(
@@ -1064,22 +1069,88 @@ Status ExternalSstFileIngestionJob::SanityCheckTableProperties(
     return s;
   }
 
-  // `TableReader` is initialized with `user_defined_timestamps_persisted` flag
-  // to be true. If its value changed to false after this sanity check, we
-  // need to reset the `TableReader`.
-  if (ucmp_->timestamp_size() > 0 &&
-      !file_to_ingest->user_defined_timestamps_persisted) {
-    s = ResetTableReader(external_file, new_file_number,
-                         file_to_ingest->user_defined_timestamps_persisted, sv,
-                         file_to_ingest, table_reader);
-  }
   return s;
 }
 
-Status ExternalSstFileIngestionJob::GetIngestedFileInfo(
+Status ExternalSstFileIngestionJob::GetIngestedFileInfoFromFileInfo(
+    const std::string& external_file, const PreparedFileInfo& file_info,
+    IngestedFileInfo* file_to_ingest) {
+  // Boundaries, size, and table properties were obtained without opening the
+  // file (e.g. produced by SstFileWriter::Finish). We reuse them directly.
+  file_to_ingest->file_size = file_info.file_size;
+  file_to_ingest->smallest_internal_key.DecodeFrom(
+      file_info.smallest_internal_key);
+  file_to_ingest->largest_internal_key.DecodeFrom(
+      file_info.largest_internal_key);
+  Status status = SanityCheckTableProperties(
+      external_file, file_info.table_properties, file_to_ingest);
+  if (!status.ok()) {
+    ROCKS_LOG_WARN(
+        db_options_.info_log,
+        "Failed to sanity check table properties for external file %s: %s",
+        external_file.c_str(), status.ToString().c_str());
+    return status;
+  }
+
+  // When the column family enables user-defined timestamps but this file was
+  // written without persisting them (the "UDT in Memtables only" feature), the
+  // supplied boundary keys carry no timestamp (SstFileWriter strips it). Pad
+  // them with the min timestamp so they match the format the open-and-scan path
+  // produces (its reopened `TableReader` pads keys the same way) and the bounds
+  // recorded in the new FileMetaData.
+  const size_t ts_sz = ucmp_->timestamp_size();
+  if (ts_sz > 0 && !file_to_ingest->user_defined_timestamps_persisted) {
+    std::string padded;
+    PadInternalKeyWithMinTimestamp(
+        &padded, file_to_ingest->smallest_internal_key.Encode(), ts_sz);
+    file_to_ingest->smallest_internal_key.DecodeFrom(padded);
+    padded.clear();
+    PadInternalKeyWithMinTimestamp(
+        &padded, file_to_ingest->largest_internal_key.Encode(), ts_sz);
+    file_to_ingest->largest_internal_key.DecodeFrom(padded);
+  }
+
+  // Extend the key bounds to cover any range tombstones. Their bounds are not
+  // part of the point-key internal keys, but the file's recorded range must
+  // include them (the open-and-scan path does this while scanning tombstones).
+  // SstFileWriter writes range tombstones at sequence number 0.
+  if (!file_info.smallest_range_del_key.empty()) {
+    RangeTombstone tombstone(Slice(file_info.smallest_range_del_key),
+                             Slice(file_info.largest_range_del_key),
+                             /*sn=*/0);
+    file_range_checker_.MaybeUpdateRange(
+        tombstone.SerializeKey(), tombstone.SerializeEndKey(), file_to_ingest);
+  }
+
+  if (ingestion_options_.allow_db_generated_files) {
+    // Sequence numbers are preserved (not reassigned), so the bounds come from
+    // the supplied table properties; the GetSeqnoBoundaryForFile scan is
+    // skipped.
+    file_to_ingest->smallest_seqno =
+        file_to_ingest->table_properties.key_smallest_seqno;
+    file_to_ingest->largest_seqno =
+        file_to_ingest->table_properties.key_largest_seqno;
+  } else {
+    // Normal ingestion reassigns a global sequence number later, so the file's
+    // keys must currently be at seqno 0 (mirror of the open-and-scan check).
+    SequenceNumber largest_seqno =
+        file_to_ingest->table_properties.key_largest_seqno;
+    // UINT64_MAX means unknown and the file is generated before table property
+    // `key_largest_seqno` is introduced.
+    if (largest_seqno != UINT64_MAX && largest_seqno > 0) {
+      return Status::Corruption(
+          "External file has non zero largest sequence number " +
+          std::to_string(largest_seqno));
+    }
+  }
+  return Status::OK();
+}
+
+Status ExternalSstFileIngestionJob::GetIngestedFileInfoFromFile(
     const std::string& external_file, uint64_t new_file_number,
-    IngestedFileInfo* file_to_ingest, SuperVersion* sv) {
-  file_to_ingest->external_file_path = external_file;
+    IngestedFileInfo* file_to_ingest, SuperVersion* sv,
+    std::unique_ptr<TableReader>* out_table_reader) {
+  TEST_SYNC_POINT("ExternalSstFileIngestionJob::GetIngestedFileInfo:ReadPath");
 
   // Get external file size
   Status status = fs_->GetFileSize(external_file, IOOptions(),
@@ -1091,15 +1162,11 @@ Status ExternalSstFileIngestionJob::GetIngestedFileInfo(
     return status;
   }
 
-  // Assign FD with number
-  file_to_ingest->fd =
-      FileDescriptor(new_file_number, 0, file_to_ingest->file_size);
-
-  // Create TableReader for external file
+  // Create TableReader for external file.
   std::unique_ptr<TableReader> table_reader;
   // Initially create the `TableReader` with flag
-  // `user_defined_timestamps_persisted` to be true since that's the most common
-  // case
+  // `user_defined_timestamps_persisted` to be true since that's the most
+  // common case
   status = ResetTableReader(external_file, new_file_number,
                             /*user_defined_timestamps_persisted=*/true, sv,
                             file_to_ingest, &table_reader);
@@ -1110,8 +1177,8 @@ Status ExternalSstFileIngestionJob::GetIngestedFileInfo(
     return status;
   }
 
-  status = SanityCheckTableProperties(external_file, new_file_number, sv,
-                                      file_to_ingest, &table_reader);
+  status = SanityCheckTableProperties(
+      external_file, *table_reader->GetTableProperties(), file_to_ingest);
   if (!status.ok()) {
     ROCKS_LOG_WARN(
         db_options_.info_log,
@@ -1120,6 +1187,23 @@ Status ExternalSstFileIngestionJob::GetIngestedFileInfo(
     return status;
   }
 
+  // The `TableReader` above was opened with `user_defined_timestamps_persisted`
+  // assumed true. If the sanity check determined the file has no persisted
+  // timestamps (UDT-in-Memtable-only feature), reopen it with the corrected
+  // flag so keys are parsed properly by the scan below.
+  if (ucmp_->timestamp_size() > 0 &&
+      !file_to_ingest->user_defined_timestamps_persisted) {
+    status = ResetTableReader(external_file, new_file_number,
+                              file_to_ingest->user_defined_timestamps_persisted,
+                              sv, file_to_ingest, &table_reader);
+    if (!status.ok()) {
+      ROCKS_LOG_WARN(db_options_.info_log,
+                     "Failed to reset table reader for external file %s: %s",
+                     external_file.c_str(), status.ToString().c_str());
+      return status;
+    }
+  }
+
   const bool allow_data_in_errors = db_options_.allow_data_in_errors;
   ParsedInternalKey key;
   if (ingestion_options_.allow_db_generated_files) {
@@ -1141,8 +1225,8 @@ Status ExternalSstFileIngestionJob::GetIngestedFileInfo(
   } else {
     SequenceNumber largest_seqno =
         table_reader.get()->GetTableProperties()->key_largest_seqno;
-    // UINT64_MAX means unknown and the file is generated before table property
-    // `key_largest_seqno` is introduced.
+    // UINT64_MAX means unknown and the file is generated before table
+    // property `key_largest_seqno` is introduced.
     if (largest_seqno != UINT64_MAX && largest_seqno > 0) {
       return Status::Corruption(
           "External file has non zero largest sequence number " +
@@ -1150,24 +1234,6 @@ Status ExternalSstFileIngestionJob::GetIngestedFileInfo(
     }
   }
 
-  if (ingestion_options_.verify_checksums_before_ingest) {
-    // If customized readahead size is needed, we can pass a user option
-    // all the way to here. Right now we just rely on the default readahead
-    // to keep things simple.
-    // TODO: plumb Env::IOActivity, Env::IOPriority
-    ReadOptions ro;
-    ro.readahead_size = ingestion_options_.verify_checksums_readahead_size;
-    ro.fill_cache = ingestion_options_.fill_cache;
-    status = table_reader->VerifyChecksum(
-        ro, TableReaderCaller::kExternalSSTIngestion);
-    if (!status.ok()) {
-      ROCKS_LOG_WARN(db_options_.info_log,
-                     "Failed to verify checksum for table reader: %s",
-                     status.ToString().c_str());
-      return status;
-    }
-  }
-
   // TODO: plumb Env::IOActivity, Env::IOPriority
   ReadOptions ro;
   ro.fill_cache = ingestion_options_.fill_cache;
@@ -1254,6 +1320,75 @@ Status ExternalSstFileIngestionJob::GetIngestedFileInfo(
     }
   }
 
+  *out_table_reader = std::move(table_reader);
+  return Status::OK();
+}
+
+Status ExternalSstFileIngestionJob::GetIngestedFileInfo(
+    const std::string& external_file, uint64_t new_file_number,
+    const PreparedFileInfo* file_info, IngestedFileInfo* file_to_ingest,
+    SuperVersion* sv) {
+  file_to_ingest->external_file_path = external_file;
+
+  std::unique_ptr<TableReader> table_reader;
+  Status status =
+      file_info != nullptr
+          ? GetIngestedFileInfoFromFileInfo(external_file, *file_info,
+                                            file_to_ingest)
+          : GetIngestedFileInfoFromFile(external_file, new_file_number,
+                                        file_to_ingest, sv, &table_reader);
+  if (!status.ok()) {
+    return status;
+  }
+
+  assert(file_to_ingest->file_size > 0);
+  assert(!file_to_ingest->unset());
+  assert(file_to_ingest->table_properties.num_entries > 0 ||
+         file_to_ingest->table_properties.num_range_deletions > 0);
+  if (ingestion_options_.allow_db_generated_files) {
+    // These files keep their original sequence numbers (derived, not
+    // reassigned), so the bounds must be valid here.
+    assert(file_to_ingest->smallest_seqno <= file_to_ingest->largest_seqno);
+    assert(file_to_ingest->largest_seqno < kMaxSequenceNumber);
+  }
+
+  // Verify the file checksum if requested. The open-and-scan path already has a
+  // `TableReader` open; the fast-path opens one here only when verification is
+  // requested -- otherwise it performs no file I/O.
+  if (ingestion_options_.verify_checksums_before_ingest) {
+    if (table_reader == nullptr) {
+      status =
+          ResetTableReader(external_file, new_file_number,
+                           file_to_ingest->user_defined_timestamps_persisted,
+                           sv, file_to_ingest, &table_reader);
+      if (!status.ok()) {
+        ROCKS_LOG_WARN(db_options_.info_log,
+                       "Failed to reset table reader for external file %s: %s",
+                       external_file.c_str(), status.ToString().c_str());
+        return status;
+      }
+    }
+    // If customized readahead size is needed, we can pass a user option all the
+    // way to here. Right now we just rely on the default readahead to keep
+    // things simple.
+    // TODO: plumb Env::IOActivity, Env::IOPriority
+    ReadOptions ro;
+    ro.readahead_size = ingestion_options_.verify_checksums_readahead_size;
+    ro.fill_cache = ingestion_options_.fill_cache;
+    status = table_reader->VerifyChecksum(
+        ro, TableReaderCaller::kExternalSSTIngestion);
+    if (!status.ok()) {
+      ROCKS_LOG_WARN(db_options_.info_log,
+                     "Failed to verify checksum for external file %s: %s",
+                     external_file.c_str(), status.ToString().c_str());
+      return status;
+    }
+  }
+
+  // Assign FD with number.
+  file_to_ingest->fd =
+      FileDescriptor(new_file_number, 0, file_to_ingest->file_size);
+
   const size_t ts_sz = ucmp_->timestamp_size();
   Slice smallest = file_to_ingest->smallest_internal_key.user_key();
   Slice largest = file_to_ingest->largest_internal_key.user_key();

db/external_sst_file_ingestion_job.h

@@ -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);
@@ -315,17 +316,36 @@ 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 `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* 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& 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

db/external_sst_file_test.cc

@@ -152,7 +152,8 @@ class ExternalSSTFileTest
       bool verify_checksums_before_ingest = true, bool ingest_behind = false,
       bool sort_data = false,
       std::map<std::string, std::string>* true_data = nullptr,
-      ColumnFamilyHandle* cfh = nullptr, bool fill_cache = false) {
+      ColumnFamilyHandle* cfh = nullptr, bool fill_cache = false,
+      bool ingest_with_file_info = false) {
     // Generate a file id if not provided
     if (file_id == -1) {
       file_id = last_file_id_ + 1;
@@ -188,7 +189,13 @@ class ExternalSSTFileTest
         return s;
       }
     }
-    s = sst_file_writer.Finish();
+    ExternalSstFileInfo file_info;
+    std::shared_ptr<const PreparedFileInfo> prepared_file_info;
+    if (ingest_with_file_info) {
+      s = sst_file_writer.Finish(&file_info, &prepared_file_info);
+    } else {
+      s = sst_file_writer.Finish();
+    }
 
     if (s.ok()) {
       IngestExternalFileOptions ifo;
@@ -197,7 +204,14 @@ class ExternalSSTFileTest
       ifo.verify_checksums_before_ingest = verify_checksums_before_ingest;
       ifo.ingest_behind = ingest_behind;
       ifo.fill_cache = fill_cache;
-      if (cfh) {
+      if (ingest_with_file_info) {
+        IngestExternalFileArg arg;
+        arg.column_family = cfh ? cfh : db_->DefaultColumnFamily();
+        arg.external_files = {file_path};
+        arg.file_infos = {prepared_file_info.get()};
+        arg.options = ifo;
+        s = db_->IngestExternalFiles({arg});
+      } else if (cfh) {
         s = db_->IngestExternalFile(cfh, {file_path}, ifo);
       } else {
         s = db_->IngestExternalFile({file_path}, ifo);
@@ -298,6 +312,181 @@ class ExternalSSTFileTest
   int last_file_id_ = 0;
 };
 
+TEST_F(ExternalSSTFileTest, IngestionTimingHistogram) {
+  Options options = CurrentOptions();
+  options.statistics = CreateDBStatistics();
+  options.statistics->set_stats_level(StatsLevel::kAll);  // enable timers
+  DestroyAndReopen(options);
+
+  ASSERT_OK(GenerateAndAddExternalFile(
+      options, {{"k1", "v1"}, {"k2", "v2"}, {"k3", "v3"}}, /*file_id=*/-1,
+      /*allow_global_seqno=*/true, /*write_global_seqno=*/false,
+      /*verify_checksums_before_ingest=*/true, /*ingest_behind=*/false,
+      /*sort_data=*/true));
+
+  auto hist = [&](Histograms h) {
+    HistogramData hd;
+    options.statistics->histogramData(h, &hd);
+    return hd;
+  };
+
+  // One sample per call (not per CF) in each phase histogram.
+  HistogramData prepare_hd = hist(INGEST_EXTERNAL_FILE_PREPARE_TIME);
+  HistogramData run_hd = hist(INGEST_EXTERNAL_FILE_RUN_TIME);
+  ASSERT_EQ(1, prepare_hd.count);
+  ASSERT_EQ(1, run_hd.count);
+  ASSERT_GT(prepare_hd.max, 0.0);
+  ASSERT_GT(run_hd.max, 0.0);
+
+  // A second call adds exactly one more sample to each histogram.
+  ASSERT_OK(GenerateAndAddExternalFile(
+      options, {{"k4", "v4"}, {"k5", "v5"}}, /*file_id=*/-1,
+      /*allow_global_seqno=*/true, /*write_global_seqno=*/false,
+      /*verify_checksums_before_ingest=*/true, /*ingest_behind=*/false,
+      /*sort_data=*/true));
+  ASSERT_EQ(2, hist(INGEST_EXTERNAL_FILE_PREPARE_TIME).count);
+  ASSERT_EQ(2, hist(INGEST_EXTERNAL_FILE_RUN_TIME).count);
+
+  // A failed ingestion records nothing (latency is logged only on success).
+  ASSERT_NOK(db_->IngestExternalFile({"/path/does/not/exist.sst"},
+                                     IngestExternalFileOptions()));
+  ASSERT_EQ(2, hist(INGEST_EXTERNAL_FILE_PREPARE_TIME).count);
+  ASSERT_EQ(2, hist(INGEST_EXTERNAL_FILE_RUN_TIME).count);
+}
+
+// Ingestion that reuses the SstFileWriter's metadata via
+// IngestExternalFileArg::file_infos skips the open-and-scan path and still
+// produces correct data, including overlapping data that is reassigned a new
+// global sequence number.
+TEST_F(ExternalSSTFileTest, IngestWithFileInfo) {
+  Options options = CurrentOptions();
+  DestroyAndReopen(options);
+  std::map<std::string, std::string> true_data;
+
+  // Count entries into the open-and-scan path so we can assert it is skipped.
+  std::atomic<int> read_path_count{0};
+  SyncPoint::GetInstance()->SetCallBack(
+      "ExternalSstFileIngestionJob::GetIngestedFileInfo:ReadPath",
+      [&](void*) { read_path_count.fetch_add(1); });
+  SyncPoint::GetInstance()->EnableProcessing();
+
+  // Reuse the writer's metadata, with the normal ingestion options: a global
+  // sequence number is assigned (allow_db_generated_files is NOT set) and the
+  // checksum is verified (the fast-path opens the file only to verify it, which
+  // is not a metadata scan).
+  auto ingest_fi = [&](std::vector<std::pair<std::string, std::string>> data) {
+    return GenerateAndAddExternalFile(
+        options, std::move(data), /*file_id=*/-1, /*allow_global_seqno=*/true,
+        /*write_global_seqno=*/false, /*verify_checksums_before_ingest=*/true,
+        /*ingest_behind=*/false, /*sort_data=*/true, &true_data,
+        /*cfh=*/nullptr, /*fill_cache=*/false, /*ingest_with_file_info=*/true);
+  };
+
+  // Non-overlapping ingest: data reads back and the file is not scanned.
+  ASSERT_OK(ingest_fi({{Key(1), "a1"}, {Key(2), "a2"}, {Key(3), "a3"}}));
+  ASSERT_EQ(read_path_count.load(), 0);
+
+  // Overlap existing (flushed) data so a non-zero global seqno is assigned and
+  // the newer values win -- the normal seqno-reassignment path through
+  // file_infos.
+  ASSERT_OK(db_->Put(WriteOptions(), Key(2), "old2"));
+  true_data[Key(2)] = "old2";
+  ASSERT_OK(Flush());
+  ASSERT_OK(ingest_fi({{Key(2), "new2"}, {Key(4), "new4"}}));
+  ASSERT_EQ(read_path_count.load(), 0);
+
+  // Control: ingesting the same way but WITHOUT file_infos does enter the scan.
+  ASSERT_OK(GenerateAndAddExternalFile(
+      options, {{Key(8), "c8"}}, /*file_id=*/-1, /*allow_global_seqno=*/true,
+      /*write_global_seqno=*/false, /*verify_checksums_before_ingest=*/true,
+      /*ingest_behind=*/false, /*sort_data=*/true, &true_data));
+  ASSERT_GT(read_path_count.load(), 0);
+
+  SyncPoint::GetInstance()->DisableProcessing();
+  SyncPoint::GetInstance()->ClearAllCallBacks();
+  size_t kcnt = 0;
+  VerifyDBFromMap(true_data, &kcnt, false);
+}
+
+// Range deletions are honored on the file_infos fast-path: the writer's
+// range-del bounds are carried in the metadata, so a DeleteRange covering
+// existing keys still shadows them.
+TEST_F(ExternalSSTFileTest, IngestWithFileInfoRangeDeletion) {
+  Options options = CurrentOptions();
+  DestroyAndReopen(options);
+  ASSERT_OK(db_->Put(WriteOptions(), Key(20), "live20"));
+  ASSERT_OK(db_->Put(WriteOptions(), Key(25), "live25"));
+  ASSERT_OK(Flush());
+
+  const std::string f = sst_files_dir_ + "rangedel.sst";
+  SstFileWriter w(EnvOptions(), options);
+  ASSERT_OK(w.Open(f));
+  ASSERT_OK(w.Put(Key(30), "v30"));
+  ASSERT_OK(w.DeleteRange(Key(20), Key(26)));  // covers keys 20..25
+  std::shared_ptr<const PreparedFileInfo> info;
+  ASSERT_OK(w.Finish(/*file_info=*/nullptr, &info));
+
+  IngestExternalFileArg arg;
+  arg.column_family = db_->DefaultColumnFamily();
+  arg.external_files = {f};
+  arg.file_infos = {info.get()};
+  arg.options.allow_global_seqno = true;
+  ASSERT_OK(db_->IngestExternalFiles({arg}));
+
+  ReadOptions ro;
+  std::string val;
+  ASSERT_TRUE(db_->Get(ro, Key(20), &val).IsNotFound());
+  ASSERT_TRUE(db_->Get(ro, Key(25), &val).IsNotFound());
+  ASSERT_OK(db_->Get(ro, Key(30), &val));
+  ASSERT_EQ(val, "v30");
+}
+
+// verify_checksums_before_ingest is honored on the file_infos fast-path: it
+// opens the file just to verify, so corruption is still detected.
+TEST_F(ExternalSSTFileTest, IngestWithFileInfoVerifiesChecksum) {
+  Options options = CurrentOptions();
+  DestroyAndReopen(options);
+
+  const std::string f = sst_files_dir_ + "corrupt.sst";
+  SstFileWriter w(EnvOptions(), options);
+  ASSERT_OK(w.Open(f));
+  for (int k = 0; k < 100; k++) {
+    ASSERT_OK(w.Put(Key(k), "v" + Key(k)));
+  }
+  std::shared_ptr<const PreparedFileInfo> info;
+  ASSERT_OK(w.Finish(/*file_info=*/nullptr, &info));
+  ASSERT_OK(test::CorruptFile(options.env, f, /*offset=*/64,
+                              /*bytes_to_corrupt=*/8));
+
+  IngestExternalFileArg arg;
+  arg.column_family = db_->DefaultColumnFamily();
+  arg.external_files = {f};
+  arg.file_infos = {info.get()};
+  arg.options.verify_checksums_before_ingest = true;
+  ASSERT_TRUE(db_->IngestExternalFiles({arg}).IsCorruption());
+}
+
+// write_global_seqno is incompatible with file_infos: the file is not opened,
+// so the seqno cannot be written back into it.
+TEST_F(ExternalSSTFileTest, IngestWithFileInfoRejectsWriteGlobalSeqno) {
+  Options options = CurrentOptions();
+  DestroyAndReopen(options);
+
+  const std::string f = sst_files_dir_ + "wgs.sst";
+  SstFileWriter w(EnvOptions(), options);
+  ASSERT_OK(w.Open(f));
+  ASSERT_OK(w.Put(Key(1), "v1"));
+  std::shared_ptr<const PreparedFileInfo> info;
+  ASSERT_OK(w.Finish(/*file_info=*/nullptr, &info));
+
+  IngestExternalFileArg arg;
+  arg.column_family = db_->DefaultColumnFamily();
+  arg.external_files = {f};
+  arg.file_infos = {info.get()};
+  arg.options.write_global_seqno = true;
+  ASSERT_TRUE(db_->IngestExternalFiles({arg}).IsInvalidArgument());
+}
+
 TEST_F(ExternalSSTFileTest, ComparatorMismatch) {
   Options options = CurrentOptions();
   Options options_diff_ucmp = options;
@@ -1816,7 +2005,8 @@ TEST_P(ExternalSSTFileTest, IngestFileWithGlobalSeqnoRandomized) {
     for (int i = 0; i < 500; i++) {
       std::vector<std::pair<std::string, std::string>> random_data;
       for (int j = 0; j < 100; j++) {
-        std::string k = rnd.RandomString(rnd.Next() % 20);
+        std::string k =
+            rnd.RandomString(rnd.Next() % 20 + 1);  // requires non-empty keys
         std::string v = rnd.RandomString(rnd.Next() % 50);
         random_data.emplace_back(k, v);
       }

db_stress_tool/db_stress_common.h

@@ -228,6 +228,7 @@ DECLARE_uint64(backup_max_size);
 DECLARE_int32(checkpoint_one_in);
 DECLARE_int32(ingest_external_file_one_in);
 DECLARE_int32(ingest_external_file_width);
+DECLARE_int32(ingest_external_file_use_file_info_one_in);
 DECLARE_int32(compact_files_one_in);
 DECLARE_int32(compact_range_one_in);
 DECLARE_int32(promote_l0_one_in);

db_stress_tool/db_stress_gflags.cc

@@ -862,6 +862,12 @@ DEFINE_int32(ingest_external_file_one_in, 0,
 DEFINE_int32(ingest_external_file_width, 100,
              "The width of the ingested external files.");
 
+DEFINE_int32(ingest_external_file_use_file_info_one_in, 0,
+             "If non-zero, the ingestexternalfile flow reuses each file's "
+             "metadata via IngestExternalFileArg::file_infos (from "
+             "SstFileWriter::Finish) once every N ingestions on average, so "
+             "ingestion skips re-opening and scanning the files.");
+
 DEFINE_int32(compact_files_one_in, 0,
              "If non-zero, then CompactFiles() will be called once for every N "
              "operations on average.  0 indicates CompactFiles() is disabled.");

db_stress_tool/no_batched_ops_stress.cc

@@ -2326,6 +2326,12 @@ class NonBatchedOpsStressTest : public StressTest {
     // deletion file's compaction input optimization.
     bool test_standalone_range_deletion = thread->rand.OneInOpt(
         FLAGS_test_ingest_standalone_range_deletion_one_in);
+    // When true, reuse the writer's metadata via IngestExternalFileArg's
+    // file_infos so ingestion skips re-opening and scanning the file. Not
+    // combined with the standalone range deletion mode (a range-del-only file).
+    bool use_file_info =
+        !test_standalone_range_deletion &&
+        thread->rand.OneInOpt(FLAGS_ingest_external_file_use_file_info_one_in);
     std::vector<std::string> external_files;
     const std::string sst_filename =
         GetDbPath() + "/." + std::to_string(thread->tid) + ".sst";
@@ -2369,6 +2375,7 @@ class NonBatchedOpsStressTest : public StressTest {
     SstFileWriter sst_file_writer(EnvOptions(options_), options_);
     SstFileWriter standalone_rangedel_sst_file_writer(EnvOptions(options_),
                                                       options_);
+    std::shared_ptr<const PreparedFileInfo> sst_file_info;
     if (s.ok()) {
       s = sst_file_writer.Open(sst_filename);
     }
@@ -2462,7 +2469,7 @@ class NonBatchedOpsStressTest : public StressTest {
       }
     }
     if (s.ok() && !keys.empty()) {
-      s = sst_file_writer.Finish();
+      s = sst_file_writer.Finish(/*file_info=*/nullptr, &sst_file_info);
     }
 
     if (s.ok() && total_keys != 0 && test_standalone_range_deletion) {
@@ -2494,9 +2501,19 @@ class NonBatchedOpsStressTest : public StressTest {
                          << ingest_options.verify_checksums_readahead_size
                          << ", fill_cache: " << ingest_options.fill_cache
                          << ", test_standalone_range_deletion: "
-                         << test_standalone_range_deletion;
-      s = db_->IngestExternalFile(column_families_[column_family],
-                                  external_files, ingest_options);
+                         << test_standalone_range_deletion
+                         << ", use_file_info: " << use_file_info;
+      if (use_file_info && sst_file_info) {
+        IngestExternalFileArg arg;
+        arg.column_family = column_families_[column_family];
+        arg.external_files = external_files;
+        arg.file_infos = {sst_file_info.get()};
+        arg.options = ingest_options;
+        s = db_->IngestExternalFiles({arg});
+      } else {
+        s = db_->IngestExternalFile(column_families_[column_family],
+                                    external_files, ingest_options);
+      }
     }
     if (!s.ok()) {
       for (PendingExpectedValue& pending_expected_value :

include/rocksdb/options.h

@@ -2871,6 +2871,14 @@ struct IngestExternalFileOptions {
   bool fill_cache = true;
 };
 
+// PreparedFileInfo is an opaque description of an SST file produced by
+// SstFileWriter::Finish(). A borrowed pointer to one can be placed in
+// IngestExternalFileArg::file_infos to let IngestExternalFiles() skip
+// re-opening and scanning the file when computing its metadata, reducing
+// ingestion I/O. Its contents are an internal implementation detail; the public
+// API only ever refers to it through this forward declaration.
+struct PreparedFileInfo;
+
 // It is valid that files_checksums and files_checksum_func_names are both
 // empty (no checksum information is provided for ingestion). Otherwise,
 // their sizes should be the same as external_files. The file order should
@@ -2907,6 +2915,15 @@ struct IngestExternalFileArg {
   // exclusive, so it is best not to depend on one or the other until it is
   // sorted out.
   std::optional<RangeOpt> atomic_replace_range;
+
+  // Optional per-file opaque metadata, parallel to `external_files` (same size
+  // and order). Each entry is a borrowed pointer to a PreparedFileInfo produced
+  // by SstFileWriter::Finish; the owning handle must stay alive until ingestion
+  // completes. When provided, ingestion reuses this metadata instead of
+  // re-opening and scanning each file to recompute it, avoiding that extra I/O.
+  // Not compatible with options.write_global_seqno (the file is not opened, so
+  // a global seqno cannot be written back into it).
+  std::vector<const PreparedFileInfo*> file_infos;
 };
 
 enum TraceFilterType : uint64_t {

include/rocksdb/sst_file_writer.h

@@ -168,11 +168,20 @@ class SstFileWriter {
   Status DeleteRange(const Slice& begin_key, const Slice& end_key,
                      const Slice& timestamp);
 
-  // Finalize writing to sst file and close file.
+  // Finalize writing to sst file and close file. Calling Finish() with no
+  // argument simply finalizes the file.
   //
-  // An optional ExternalSstFileInfo pointer can be passed to the function
-  // which will be populated with information about the created sst file.
-  Status Finish(ExternalSstFileInfo* file_info = nullptr);
+  // If `file_info` is non-null, it is populated with human-readable information
+  // about the created sst file.
+  //
+  // If `prepared_file_info` is non-null, it receives an opaque owning handle
+  // describing the file. A borrowed pointer to it can be placed in
+  // IngestExternalFileArg::file_infos so ingestion reuses this metadata instead
+  // of re-opening and scanning the file; the handle must be kept alive until
+  // ingestion completes.
+  Status Finish(
+      ExternalSstFileInfo* file_info = nullptr,
+      std::shared_ptr<const PreparedFileInfo>* prepared_file_info = nullptr);
 
   // Return the current file size.
   uint64_t FileSize();

include/rocksdb/statistics.h

@@ -765,6 +765,16 @@ enum Histograms : uint32_t {
   // blocks for uniform key distribution tracking.
   BLOCK_KEY_DISTRIBUTION_CV,
 
+  // Time (microseconds) in the "prepare" phase of an IngestExternalFile(s) call
+  // (validation and reading/linking the files; does not block writes). One
+  // sample per successful call; requires stats level > kExceptTimers.
+  INGEST_EXTERNAL_FILE_PREPARE_TIME,
+
+  // Time (microseconds) in the "run" phase of an IngestExternalFile(s) call
+  // (work under the DB mutex while writes are blocked). One sample per
+  // successful call; requires stats level > kExceptTimers.
+  INGEST_EXTERNAL_FILE_RUN_TIME,
+
   HISTOGRAM_ENUM_MAX
 };
 

java/rocksjni/portal.h

@@ -5998,6 +5998,10 @@ class HistogramTypeJni {
         return 0x41;
       case ROCKSDB_NAMESPACE::Histograms::BLOCK_KEY_DISTRIBUTION_CV:
         return 0x42;
+      case ROCKSDB_NAMESPACE::Histograms::INGEST_EXTERNAL_FILE_PREPARE_TIME:
+        return 0x43;
+      case ROCKSDB_NAMESPACE::Histograms::INGEST_EXTERNAL_FILE_RUN_TIME:
+        return 0x44;
       case ROCKSDB_NAMESPACE::Histograms::HISTOGRAM_ENUM_MAX:
         // 0x3E is reserved for backwards compatibility on current minor
         // version.
@@ -6151,6 +6155,10 @@ class HistogramTypeJni {
         return ROCKSDB_NAMESPACE::Histograms::MULTISCAN_BLOCKS_PER_PREPARE;
       case 0x42:
         return ROCKSDB_NAMESPACE::Histograms::BLOCK_KEY_DISTRIBUTION_CV;
+      case 0x43:
+        return ROCKSDB_NAMESPACE::Histograms::INGEST_EXTERNAL_FILE_PREPARE_TIME;
+      case 0x44:
+        return ROCKSDB_NAMESPACE::Histograms::INGEST_EXTERNAL_FILE_RUN_TIME;
       case 0x3E:
         // 0x3E is reserved for backwards compatibility on current minor
         // version.

java/src/main/java/org/rocksdb/HistogramType.java

@@ -231,6 +231,16 @@ public enum HistogramType {
    */
   BLOCK_KEY_DISTRIBUTION_CV((byte) 0x42),
 
+  /**
+   * Time (us) in the prepare phase of an IngestExternalFile(s) call.
+   */
+  INGEST_EXTERNAL_FILE_PREPARE_TIME((byte) 0x43),
+
+  /**
+   * Time (us) in the run phase of an IngestExternalFile(s) call.
+   */
+  INGEST_EXTERNAL_FILE_RUN_TIME((byte) 0x44),
+
   // 0x3E is reserved for backwards compatibility on current minor version.
   HISTOGRAM_ENUM_MAX((byte) 0x3E);
 

monitoring/statistics.cc

@@ -386,6 +386,9 @@ const std::vector<std::pair<Histograms, std::string>> HistogramsNameMap = {
     {MULTISCAN_PREPARE_MICROS, "rocksdb.multiscan.prepare.micros"},
     {MULTISCAN_BLOCKS_PER_PREPARE, "rocksdb.multiscan.blocks.per.prepare"},
     {BLOCK_KEY_DISTRIBUTION_CV, "rocksdb.block.key.distribution.cv"},
+    {INGEST_EXTERNAL_FILE_PREPARE_TIME,
+     "rocksdb.ingest.external.file.prepare.micros"},
+    {INGEST_EXTERNAL_FILE_RUN_TIME, "rocksdb.ingest.external.file.run.micros"},
 };
 
 std::shared_ptr<Statistics> CreateDBStatistics() {

table/prepared_file_info.h

@@ -0,0 +1,33 @@
+//  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
+//  This source code is licensed under both the GPLv2 (found in the
+//  COPYING file in the root directory) and Apache 2.0 License
+//  (found in the LICENSE.Apache file in the root directory).
+
+#pragma once
+
+#include <cstdint>
+#include <string>
+
+#include "rocksdb/table_properties.h"
+
+namespace ROCKSDB_NAMESPACE {
+
+// Definition of the type that is opaque to public API users (declared only as a
+// forward declaration in rocksdb/options.h and rocksdb/sst_file_writer.h).
+// Produced by SstFileWriter::Finish and consumed by ExternalSstFileIngestionJob
+// so ingestion can skip re-opening and scanning the file to recompute its
+// metadata.
+struct PreparedFileInfo {
+  uint64_t file_size = 0;
+  // Encoded internal keys (user key + 8-byte footer), with the user-defined
+  // timestamp stripped when timestamps are not persisted.
+  std::string smallest_internal_key;
+  std::string largest_internal_key;
+  // Range-deletion boundary user keys (timestamp-stripped when applicable);
+  // empty when the file has no range deletions.
+  std::string smallest_range_del_key;
+  std::string largest_range_del_key;
+  TableProperties table_properties;
+};
+
+}  // namespace ROCKSDB_NAMESPACE

table/sst_file_writer.cc

@@ -15,6 +15,7 @@
 #include "rocksdb/file_system.h"
 #include "rocksdb/table.h"
 #include "table/block_based/block_based_table_builder.h"
+#include "table/prepared_file_info.h"
 #include "table/sst_file_writer_collectors.h"
 #include "test_util/sync_point.h"
 
@@ -57,6 +58,11 @@ struct SstFileWriter::Rep {
   WriteOptions write_options;
   InternalKeyComparator internal_comparator;
   ExternalSstFileInfo file_info;
+  // Encoded internal keys of the smallest/largest point key, tracked here (not
+  // on the public ExternalSstFileInfo) so a PreparedFileInfo can be built.
+  // Empty when the file has no point keys.
+  std::string smallest_internal_key;
+  std::string largest_internal_key;
   InternalKey ikey;
   std::string column_family_name;
   ColumnFamilyHandle* cfh;
@@ -124,8 +130,15 @@ struct SstFileWriter::Rep {
     builder->Add(ikey.Encode(), value);
 
     // update file info
+    Slice encoded_ikey = ikey.Encode();
+    if (file_info.num_entries == 0) {
+      smallest_internal_key.assign(encoded_ikey.data(), encoded_ikey.size());
+    }
     file_info.num_entries++;
     file_info.largest_key.assign(user_key.data(), user_key.size());
+    // Keys are added in strictly increasing order, so the last point key is the
+    // largest.
+    largest_internal_key.assign(encoded_ikey.data(), encoded_ikey.size());
     file_info.file_size = builder->FileSize();
 
     InvalidatePageCache(false /* closing */).PermitUncheckedError();
@@ -436,6 +449,8 @@ Status SstFileWriter::Open(const std::string& file_path, Temperature temp) {
   r->file_info = ExternalSstFileInfo();
   r->file_info.file_path = file_path;
   r->file_info.version = 2;
+  r->smallest_internal_key.clear();
+  r->largest_internal_key.clear();
   return s;
 }
 
@@ -476,7 +491,9 @@ Status SstFileWriter::DeleteRange(const Slice& begin_key, const Slice& end_key,
   return rep_->DeleteRange(begin_key, end_key, timestamp);
 }
 
-Status SstFileWriter::Finish(ExternalSstFileInfo* file_info) {
+Status SstFileWriter::Finish(
+    ExternalSstFileInfo* file_info,
+    std::shared_ptr<const PreparedFileInfo>* prepared_file_info) {
   Rep* r = rep_.get();
   if (!r->builder) {
     return Status::InvalidArgument("File is not opened");
@@ -484,12 +501,15 @@ Status SstFileWriter::Finish(ExternalSstFileInfo* file_info) {
   if (r->file_info.num_entries == 0 &&
       r->file_info.num_range_del_entries == 0) {
     r->builder->status().PermitUncheckedError();
+    // Leave the builder intact; ~SstFileWriter will Abandon() it.
     return Status::InvalidArgument("Cannot create sst file with no entries");
   }
 
+  // Finalize the table builder and underlying file. The builder is reset only
+  // at the end of this function, so its table properties stay readable for
+  // prepared_file_info below.
   Status s = r->builder->Finish();
   r->file_info.file_size = r->builder->FileSize();
-
   IOOptions opts;
   if (s.ok()) {
     s = WritableFileWriter::PrepareIOOptions(r->write_options, opts);
@@ -541,6 +561,44 @@ Status SstFileWriter::Finish(ExternalSstFileInfo* file_info) {
     }
   }
 
+  if (s.ok() && prepared_file_info != nullptr) {
+    auto prepared = std::make_shared<PreparedFileInfo>();
+    prepared->file_size = r->file_info.file_size;
+    prepared->table_properties = r->builder->GetTableProperties();
+    prepared->smallest_internal_key = r->smallest_internal_key;
+    prepared->largest_internal_key = r->largest_internal_key;
+    prepared->smallest_range_del_key = r->file_info.smallest_range_del_key;
+    prepared->largest_range_del_key = r->file_info.largest_range_del_key;
+    // Strip user-defined timestamps when they should not be persisted, so the
+    // boundaries match the format the open-and-scan ingestion path produces.
+    if (r->strip_timestamp) {
+      if (!prepared->smallest_internal_key.empty()) {
+        assert(prepared->smallest_internal_key.size() >=
+               kNumInternalBytes + r->ts_sz);
+        assert(prepared->largest_internal_key.size() >=
+               kNumInternalBytes + r->ts_sz);
+        // The timestamp sits just before the 8-byte internal-key footer.
+        prepared->smallest_internal_key.erase(
+            prepared->smallest_internal_key.size() - kNumInternalBytes -
+                r->ts_sz,
+            r->ts_sz);
+        prepared->largest_internal_key.erase(
+            prepared->largest_internal_key.size() - kNumInternalBytes -
+                r->ts_sz,
+            r->ts_sz);
+      }
+      if (!prepared->smallest_range_del_key.empty()) {
+        assert(prepared->smallest_range_del_key.size() >= r->ts_sz);
+        assert(prepared->largest_range_del_key.size() >= r->ts_sz);
+        prepared->smallest_range_del_key.resize(
+            prepared->smallest_range_del_key.size() - r->ts_sz);
+        prepared->largest_range_del_key.resize(
+            prepared->largest_range_del_key.size() - r->ts_sz);
+      }
+    }
+    *prepared_file_info = std::move(prepared);
+  }
+
   r->builder.reset();
   return s;
 }

tools/db_bench_tool.cc

@@ -62,6 +62,7 @@
 #include "rocksdb/secondary_cache.h"
 #include "rocksdb/slice.h"
 #include "rocksdb/slice_transform.h"
+#include "rocksdb/sst_file_writer.h"
 #include "rocksdb/stats_history.h"
 #include "rocksdb/table.h"
 #include "rocksdb/tool_hooks.h"
@@ -235,6 +236,8 @@ DEFINE_string(
     "\tcompact1  -- compact L1 into L2\n"
     "\twaitforcompaction - pause until compaction is (probably) done\n"
     "\tflush - flush the memtable\n"
+    "\tingestexternalfile -- Create external SST files and ingest them into "
+    "the DB via IngestExternalFile\n"
     "\topenandcompact -- Open DB and compact all files to bottommost level, "
     "writing output to separate directory without modifying source DB. "
     "Designed for remote compaction service testing\n"
@@ -1309,6 +1312,21 @@ DEFINE_string(backup_dir, "",
 DEFINE_string(restore_dir, "",
               "If not empty string, use the given dir for restore.");
 
+DEFINE_int32(ingest_external_file_batch_size, 10,
+             "Number of SST files ingested per IngestExternalFile call (one "
+             "batch) in the ingestexternalfile benchmark.");
+
+DEFINE_int32(ingest_external_file_num_batches, 1,
+             "Number of IngestExternalFile calls (batches) in the "
+             "ingestexternalfile benchmark. Total files ingested is "
+             "batch_size * num_batches; each file holds --num keys.");
+
+DEFINE_bool(
+    ingest_external_file_use_file_info, false,
+    "If true, the ingestexternalfile benchmark passes each file's metadata "
+    "(from SstFileWriter::Finish) via IngestExternalFileArg::file_infos, so "
+    "ingestion reuses it instead of re-opening and scanning the files.");
+
 DEFINE_uint64(
     initial_auto_readahead_size,
     ROCKSDB_NAMESPACE::BlockBasedTableOptions().initial_auto_readahead_size,
@@ -3972,6 +3990,9 @@ class Benchmark {
         method = &Benchmark::WriteSeqSeekSeq;
       } else if (name == "compact") {
         method = &Benchmark::Compact;
+      } else if (name == "ingestexternalfile") {
+        num_threads = 1;
+        method = &Benchmark::IngestExternalFile;
       } else if (name == "compactall") {
         CompactAll();
       } else if (name == "compact0") {
@@ -9288,6 +9309,120 @@ class Benchmark {
     db->CompactRange(cro, nullptr, nullptr);
   }
 
+  // Creates SST files with SstFileWriter and ingests them into the DB via
+  // IngestExternalFile, reporting throughput. Batching is used only to group
+  // multiple files into a single IngestExternalFile call:
+  // --ingest_external_file_batch_size files per call,
+  // --ingest_external_file_num_batches calls. Each file holds --num keys over
+  // the same key range, so the files overlap and are assigned new global
+  // sequence numbers. With --ingest_external_file_use_file_info the writer's
+  // metadata is passed via IngestExternalFileArg::file_infos so ingestion
+  // reuses it instead of re-opening and scanning the files. Single threaded.
+  //
+  // NOTE: db_bench's reported micros/op includes file generation; use the
+  // rocksdb.ingest.external.file.micros histogram (--statistics) for the
+  // isolated per-call ingest latency.
+  void IngestExternalFile(ThreadState* thread) {
+    DB* db = SelectDB(thread);
+
+    const int batch_size = std::max(1, FLAGS_ingest_external_file_batch_size);
+    const int num_batches = std::max(1, FLAGS_ingest_external_file_num_batches);
+    const bool use_file_info = FLAGS_ingest_external_file_use_file_info;
+    // --num is the number of keys in each file; all files cover this same key
+    // range, so they overlap (which is fine -- a global seqno is assigned).
+    const int64_t keys_per_file = num_;
+    const int64_t key_space = keys_per_file;
+
+    const std::string tmp_dir = FLAGS_db + "/ingest_tmp";
+    Status s = FLAGS_env->CreateDirIfMissing(tmp_dir);
+    if (!s.ok()) {
+      fprintf(stderr, "CreateDirIfMissing(%s) failed: %s\n", tmp_dir.c_str(),
+              s.ToString().c_str());
+      db_bench_exit(1);
+    }
+
+    RandomGenerator gen;
+    std::unique_ptr<const char[]> key_guard;
+    Slice key = AllocateKey(&key_guard);
+    int64_t bytes = 0;
+
+    for (int b = 0; b < num_batches; ++b) {
+      // Write batch_size SstFileWriter files (all covering the same key range);
+      // they are batched into a single ingest call below.
+      std::vector<std::string> files;
+      std::vector<std::shared_ptr<const PreparedFileInfo>> prepared_infos;
+      for (int j = 0; j < batch_size; ++j) {
+        std::string file_path = tmp_dir + "/ingest_" + std::to_string(b) + "_" +
+                                std::to_string(j) + ".sst";
+        SstFileWriter sst_file_writer(EnvOptions(), open_options_);
+        s = sst_file_writer.Open(file_path);
+        if (!s.ok()) {
+          fprintf(stderr, "SstFileWriter::Open(%s) failed: %s\n",
+                  file_path.c_str(), s.ToString().c_str());
+          db_bench_exit(1);
+        }
+        for (int64_t i = 0; i < keys_per_file; ++i) {
+          GenerateKeyFromInt(i, key_space, &key);
+          Slice value = gen.Generate();
+          s = sst_file_writer.Put(key, value);
+          if (!s.ok()) {
+            sst_file_writer.Finish().PermitUncheckedError();
+            fprintf(stderr, "SstFileWriter::Put failed: %s\n",
+                    s.ToString().c_str());
+            db_bench_exit(1);
+          }
+          bytes += key.size() + value.size();
+        }
+        std::shared_ptr<const PreparedFileInfo> info;
+        s = sst_file_writer.Finish(/*file_info=*/nullptr, &info);
+        if (!s.ok()) {
+          fprintf(stderr, "SstFileWriter::Finish(%s) failed: %s\n",
+                  file_path.c_str(), s.ToString().c_str());
+          db_bench_exit(1);
+        }
+        files.emplace_back(file_path);
+        prepared_infos.emplace_back(std::move(info));
+      }
+
+      // Link the files into the DB instead of copying them, so the benchmark
+      // measures the ingest path itself (which the file_info fast-path speeds
+      // up by skipping the re-open and scan of each file) rather than file-copy
+      // throughput. allow_global_seqno (default) handles the overlap.
+      IngestExternalFileOptions ingest_options;
+      ingest_options.move_files = true;
+      if (use_file_info) {
+        // Reuse the writer's metadata so ingestion skips re-opening/scanning.
+        IngestExternalFileArg arg;
+        arg.column_family = db->DefaultColumnFamily();
+        arg.external_files = files;
+        for (const auto& prepared : prepared_infos) {
+          arg.file_infos.push_back(prepared.get());
+        }
+        arg.options = ingest_options;
+        s = db->IngestExternalFiles({arg});
+      } else {
+        s = db->IngestExternalFile(files, ingest_options);
+      }
+      if (!s.ok()) {
+        fprintf(stderr, "IngestExternalFile failed: %s\n",
+                s.ToString().c_str());
+        db_bench_exit(1);
+      }
+      thread->stats.FinishedOps(&db_, db, 1, kWrite);
+
+      for (const auto& f : files) {
+        FLAGS_env->DeleteFile(f).PermitUncheckedError();
+      }
+    }
+
+    thread->stats.AddBytes(bytes);
+    char msg[100];
+    snprintf(msg, sizeof(msg),
+             "(%d batches x %d files, %" PRId64 " keys/file%s)", num_batches,
+             batch_size, keys_per_file, use_file_info ? ", file_info" : "");
+    thread->stats.AddMessage(msg);
+  }
+
   void CompactAll() {
     CompactRangeOptions cro;
     cro.max_subcompactions = static_cast<uint32_t>(FLAGS_subcompactions);

tools/db_bench_tool_test.cc

@@ -85,6 +85,21 @@ class DBBenchTest : public testing::Test {
     ASSERT_EQ(0, db_bench_tool(argc(), argv()));
   }
 
+  // Every flag is passed explicitly because gflags state persists across
+  // db_bench_tool() calls within the same test process.
+  void RunIngestBench(int batch_size, int num_batches, bool use_file_info) {
+    ResetArgs();
+    AppendArgs(
+        {"./db_bench", "--benchmarks=ingestexternalfile", "--use_existing_db=0",
+         "--num=2000", "--compression_type=none", "--db=" + db_path_,
+         "--wal_dir=" + wal_path_,
+         "--ingest_external_file_batch_size=" + std::to_string(batch_size),
+         "--ingest_external_file_num_batches=" + std::to_string(num_batches),
+         "--ingest_external_file_use_file_info=" +
+             std::string(use_file_info ? "true" : "false")});
+    ASSERT_EQ(0, db_bench_tool(argc(), argv()));
+  }
+
   void VerifyOptions(const Options& opt) {
     DBOptions loaded_db_opts;
     ConfigOptions config_opts;
@@ -144,6 +159,16 @@ TEST_F(DBBenchTest, OptionsFile) {
   VerifyOptions(opt);
 }
 
+TEST_F(DBBenchTest, IngestExternalFile) {
+  RunIngestBench(/*batch_size=*/3, /*num_batches=*/2,
+                 /*use_file_info=*/false);
+}
+
+TEST_F(DBBenchTest, IngestExternalFileWithFileInfo) {
+  RunIngestBench(/*batch_size=*/3, /*num_batches=*/2,
+                 /*use_file_info=*/true);
+}
+
 TEST_F(DBBenchTest, OptionsFileUniversal) {
   const std::string kOptionsFileName = test_path_ + "/OPTIONS_test";
 

tools/db_crashtest.py

@@ -215,6 +215,7 @@ default_params = {
     "index_block_search_type": lambda: random.choice([0, 1, 2]),
     "uniform_cv_threshold": lambda: random.choice([-1, 0.2, 1000]),
     "ingest_external_file_one_in": lambda: random.choice([1000, 1000000]),
+    "ingest_external_file_use_file_info_one_in": lambda: random.choice([0, 2]),
     "test_ingest_standalone_range_deletion_one_in": lambda: random.choice([0, 5, 10]),
     "iterpercent": 10,
     "lock_wal_one_in": lambda: random.choice([10000, 1000000]),

unreleased_history/new_features/ingest_external_file_phase_histograms.md

@@ -0,0 +1 @@
+Added two stats histograms reporting the per-call latency (in microseconds) of each successful `IngestExternalFile`/`IngestExternalFiles` call, split by phase: `rocksdb.ingest.external.file.prepare.micros` (argument validation and reading/validating/linking the external files, which does not block live writes) and `rocksdb.ingest.external.file.run.micros` (the ingestion performed under the DB mutex while live writes are blocked). Failed ingestions are not recorded.

unreleased_history/public_api_changes/ingest_external_files_file_infos.md

@@ -0,0 +1 @@
+`IngestExternalFileArg` gained a `file_infos` field: a vector of borrowed `const PreparedFileInfo*` pointers, parallel to `external_files`, each referring to an opaque handle produced via the new `std::shared_ptr<const PreparedFileInfo>*` output parameter of `SstFileWriter::Finish` (the handle must outlive the ingestion). When set, `IngestExternalFiles()` reuses the supplied per-file metadata instead of re-opening and scanning each file to recompute it, avoiding that extra I/O.