perf: reuse native Parquet footer reads and split large range requests

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

@@ -231,6 +231,9 @@ execution metrics. Counters accumulate per scan operator; they do not instrument
 | `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.                                   |
+| `scan_io_metadata_storage_reads`           | Storage reads made by metadata loads, prefetched loads included. A load whose footer read holds the footer and page index takes one.                                                                                         |
+| `scan_io_footer_read_reused_bytes`         | Bytes served from a file's footer read instead of a new request (`spark.comet.parquet.reuseFooterRead.enabled`). They are also counted in `scan_io_data_bytes` or `scan_io_metadata_bytes`.                                  |
+| `scan_io_data_wait_time`                   | Time the scan waited for data-page reads from storage. Reads served from a footer read do not wait.                                                                                                                          |
 | `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.

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

@@ -256,6 +256,22 @@ The cost is one extra footer and page index read for a file that is never opened
 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.
 
+### Reducing Requests to High-Latency Storage
+
+A native Parquet scan reads a file's footer and page index with one read of the last 512 KiB of the file. For a
+file smaller than that, the read returns the whole file. These settings decide how many requests the rest of
+the file then takes:
+
+| Config                                        | Default | Effect                                                                                                                                                                                       |
+| --------------------------------------------- | ------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
+| `spark.comet.parquet.reuseFooterRead.enabled` | `true`  | Serves the data inside the footer read from memory instead of requesting it again, so a file smaller than 512 KiB costs one request.                                                         |
+| `spark.comet.parquet.wholeFileReadThreshold`  | `0`     | Reads a file of at most this size whole with its footer read, so it costs one request. It also reads columns the query doesn't need. Needs `reuseFooterRead`. `0` turns it off.              |
+| `spark.comet.parquet.rangeRequestSplitSize`   | `0`     | On a remote object store, splits a merged byte range longer than this into requests of at most this size, sent in parallel. One request is limited by a single connection. `0` turns it off. |
+
+The `scan_io_footer_read_reused_bytes` metric shows the bytes served from footer reads,
+`scan_io_metadata_storage_reads` counts the reads that loading metadata takes, and `scan_io_data_wait_time` shows
+how long the scan waited for data reads.
+
 ## 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

@@ -142,9 +142,12 @@ 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_PREFETCH_NEXT_FILE_METADATA_ENABLED, 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_RANGE_REQUEST_SPLIT_SIZE,
+    COMET_PARQUET_REUSE_FOOTER_READ_ENABLED, COMET_PARQUET_ROW_FILTER_PUSHDOWN_ENABLED,
+    COMET_PARQUET_WHOLE_FILE_READ_THRESHOLD, COMET_STRICT_TIMESTAMP_LABELS, COMET_TRACING_ENABLED,
+    SPARK_EXECUTOR_CORES,
 };
+use crate::parquet::eager_page_index_reader_factory::ScanIoOptions;
 use crate::parquet::encryption_support::{CometEncryptionFactory, ENCRYPTION_FACTORY_ID};
 use crate::parquet::parquet_exec::CometParquetScanOptions;
 use crate::parquet::parquet_support::CometObjectStoreRegistry;
@@ -1115,6 +1118,12 @@ pub(crate) fn comet_session_context(
     session_config = session_config.with_extension(Arc::new(CometParquetScanOptions {
         prefetch_next_file_metadata: spark_config
             .get_bool(COMET_PARQUET_PREFETCH_NEXT_FILE_METADATA_ENABLED),
+        scan_io: ScanIoOptions {
+            reuse_footer_read: spark_config.get_bool(COMET_PARQUET_REUSE_FOOTER_READ_ENABLED),
+            whole_file_read_threshold: spark_config
+                .get_u64(COMET_PARQUET_WHOLE_FILE_READ_THRESHOLD, 0),
+            range_split_size: spark_config.get_u64(COMET_PARQUET_RANGE_REQUEST_SPLIT_SIZE, 0),
+        },
     }));
 
     // Translate Comet's spill codec into DataFusion's spill compression, which DataFusion

native/core/src/execution/spark_config.rs

@@ -26,6 +26,12 @@ 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_PARQUET_REUSE_FOOTER_READ_ENABLED: &str =
+    "spark.comet.parquet.reuseFooterRead.enabled";
+pub(crate) const COMET_PARQUET_WHOLE_FILE_READ_THRESHOLD: &str =
+    "spark.comet.parquet.wholeFileReadThreshold";
+pub(crate) const COMET_PARQUET_RANGE_REQUEST_SPLIT_SIZE: &str =
+    "spark.comet.parquet.rangeRequestSplitSize";
 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 =