perf: load the next Parquet file's footer and page index while the current file is read

docs/source/user-guide/latest/metrics.md

@@ -230,12 +230,17 @@ execution metrics. Counters accumulate per scan operator; they do not instrument
 | `scan_io_object_store_response_bytes_read` | Response bytes actually consumed at that API, including bytes fetched between coalesced ranges. Not HTTP wire bytes.                                                                                                         |
 | `scan_io_metadata_cache_hits`              | Successful, cache-eligible metadata opens requiring no storage reads.                                                                                                                                                        |
 | `scan_io_metadata_cache_misses`            | Successful, cache-eligible metadata opens requiring storage reads. Failed opens and encrypted opens, which bypass this shared cache, increment neither cache counter.                                                        |
+| `scan_io_metadata_prefetches`              | Opens that used a load which read storage ahead of them, started when the previous file opened (`spark.comet.parquet.prefetchNextFileMetadata.enabled`). Each also counts as a cache miss.                                   |
 | `row_groups_pruned_dynamic_filter`         | Row groups skipped by a runtime predicate whose value changes during execution, such as a local TopK threshold.                                                                                                              |
 
 Reader-level and object-store bytes are two views of the same reads; do not add them together.
 Likewise, footer bytes are a subset of metadata bytes, not a third reader-level category. A warm
-metadata-cache hit contributes no new metadata or footer I/O. A valid plaintext footer followed by
-a page-index failure still contributes footer bytes; an invalid plaintext footer does not.
+metadata-cache hit contributes no new metadata or footer I/O. A prefetched load's footer and
+metadata bytes count when the prefetch reads them, so a prefetch that no open uses, because the
+scan stopped early or pruned the file before opening it, adds bytes with no matching cache miss and
+no `scan_io_metadata_prefetches`.
+A valid plaintext footer followed by a page-index failure still contributes footer bytes; an
+invalid plaintext footer does not.
 
 Remote counters follow the backend selected during object-store construction, including native
 S3 (`s3`/`s3a`), GCS (`gs`), Azure (`az`, `adl`, `azure`, `abfs`, `abfss`), and HTTP(S) stores.

docs/source/user-guide/latest/parquet.md

@@ -215,6 +215,18 @@ more splits overall, so some idle tasks may remain — tune the value against yo
 See [#3817](https://github.com/apache/datafusion-comet/issues/3817#issuecomment-4193279630) for a
 worked example and further discussion.
 
+### Prefetching the Next File's Metadata
+
+A native Parquet scan opens a task's files one after another, and each open waits for a round trip to read
+the file's footer and page index before its data fetch can start. For tasks that read many small files from
+high-latency storage such as S3, that wait is paid once per file. By default, Comet loads the next file's
+footer and page index in the background while the current file is read. Set
+`spark.comet.parquet.prefetchNextFileMetadata.enabled` to `false` to turn this off.
+
+The cost is one extra footer and page index read for a file that is never opened, such as when a scan stops
+early or the file is pruned before it opens. Scans of encrypted Parquet files never prefetch. The
+`scan_io_metadata_prefetches` metric counts the opens that used a load which read storage ahead of them.
+
 ## Known Issues
 
 This section lists open issues in which Comet's Parquet reads and writes give a different result

native/core/src/execution/jni_api.rs

@@ -137,10 +137,11 @@ use crate::execution::spark_config::{
     SparkConfig, COMET_DEBUG_ENABLED, COMET_DEBUG_MEMORY,
     COMET_EXEC_AGGREGATE_SKIP_PARTIAL_ENABLED, COMET_EXEC_SPILL_COMPRESSION_CODEC,
     COMET_EXPLAIN_NATIVE_ENABLED, COMET_MAX_TEMP_DIRECTORY_SIZE,
-    COMET_PARQUET_ROW_FILTER_PUSHDOWN_ENABLED, COMET_STRICT_TIMESTAMP_LABELS,
-    COMET_TRACING_ENABLED, SPARK_EXECUTOR_CORES,
+    COMET_PARQUET_PREFETCH_NEXT_FILE_METADATA_ENABLED, COMET_PARQUET_ROW_FILTER_PUSHDOWN_ENABLED,
+    COMET_STRICT_TIMESTAMP_LABELS, COMET_TRACING_ENABLED, SPARK_EXECUTOR_CORES,
 };
 use crate::parquet::encryption_support::{CometEncryptionFactory, ENCRYPTION_FACTORY_ID};
+use crate::parquet::parquet_exec::CometParquetScanOptions;
 use crate::parquet::parquet_support::CometObjectStoreRegistry;
 use datafusion_comet_proto::spark_operator::operator::OpStruct;
 use log::{info, warn};
@@ -1029,6 +1030,13 @@ fn prepare_datafusion_session_context(
             session_config.set_str("datafusion.execution.parquet.reorder_filters", "true");
     }
 
+    // Native Parquet scan options that DataFusion has no setting for travel as a session config
+    // extension, which every scan in the plan reads.
+    session_config = session_config.with_extension(Arc::new(CometParquetScanOptions {
+        prefetch_next_file_metadata: spark_config
+            .get_bool(COMET_PARQUET_PREFETCH_NEXT_FILE_METADATA_ENABLED),
+    }));
+
     // Translate Comet's spill codec into DataFusion's spill compression, which DataFusion
     // otherwise leaves uncompressed. Set before the pass-through below for the same reason,
     // so an explicit `spark.comet.datafusion.execution.spill_compression` still wins.

native/core/src/execution/spark_config.rs

@@ -24,6 +24,8 @@ pub(crate) const COMET_MAX_TEMP_DIRECTORY_SIZE: &str = "spark.comet.maxTempDirec
 pub(crate) const COMET_DEBUG_MEMORY: &str = "spark.comet.debug.memory";
 pub(crate) const COMET_PARQUET_ROW_FILTER_PUSHDOWN_ENABLED: &str =
     "spark.comet.parquet.rowFilterPushdown.enabled";
+pub(crate) const COMET_PARQUET_PREFETCH_NEXT_FILE_METADATA_ENABLED: &str =
+    "spark.comet.parquet.prefetchNextFileMetadata.enabled";
 pub(crate) const COMET_EXEC_AGGREGATE_SKIP_PARTIAL_ENABLED: &str =
     "spark.comet.exec.aggregate.skipPartial.enabled";
 pub(crate) const COMET_EXEC_SPILL_COMPRESSION_CODEC: &str =