[andygrove]

Which issue does this PR close?

Part of #7459.

Rationale for this change

Spark 4.2.0 reads Parquet from S3 through the S3 Analytics Accelerator (AAL). That made the Spark baseline about 30% faster on TPC-DS SF1000 and shrank Comet's speedup (#7451).

AAL mostly brings Spark's per-file request pattern close to what Comet's native scan already does. In two places it still does better:

• AAL reads a small file with one request. Comet's scan takes two.
• AAL fetches a large read as parallel 8 MiB requests. Comet's scan fetches it with one request.

#7459 has the comparison.

What changes are included in this PR?

• Footer-read reuse (spark.comet.parquet.reuseFooterRead.enabled, default true).
  • The footer read fetches the last 512 KiB of a file. The scan now keeps those bytes and serves the data and Bloom filter ranges inside them from memory, so a file smaller than 512 KiB costs one request instead of two.
  • DataFusion's opener reads Bloom filters through the reader that loaded the metadata, then decodes through a new reader from the factory, which never loads metadata. So the factory holds each open file's bytes, keyed by path and split, until the readers using them drop.
  • The next-file metadata prefetch from perf: load the next Parquet file's footer and page index while the current file is read #7094 hands its bytes over the same way.
• Whole-file reads (spark.comet.parquet.wholeFileReadThreshold, default 0, off). The footer read fetches a file at or below the threshold whole, so the file costs one request. The cost is reading columns the query does not need. AAL uses 8 MiB.
• Split range requests (spark.comet.parquet.rangeRequestSplitSize, default 0, off). On a remote object store, a merged range longer than the split size is fetched as parts of at most that size, up to 16 at once. Ranges merge across gaps of up to 1 MiB, as in object_store::coalesce_ranges.
• Metrics: scan_io_data_wait_time (time spent waiting for data-page reads), scan_io_footer_read_reused_bytes and scan_io_metadata_storage_reads.
• Docs: the three configs in parquet.md and the three metrics in metrics.md.

Open questions:

• Memory accounting. The kept bytes are not charged to the memory pool. That is up to 512 KiB per open file by default, or up to the whole-file threshold when one is set. A scan holds at most two files' bytes at a time: the file it is reading and the next file's prefetch.
• Defaults. The EKS runs in perf: native Parquet scans make two S3 requests for a small file and fetch large merged ranges with one GET #7459 should decide whether reuse stays on by default, and the defaults of the other two options.

How are these changes tested?

Rust unit tests in eager_page_index_reader_factory.rs count GETs against an in-memory store. They check that:

• A file inside the footer read is read with one GET, against two without reuse.
• The footer read serves the row groups it covers: 3 GETs instead of 5 for a file with four row groups.
• A reader that DataFusion makes after the metadata load reuses that load's bytes.
• A filtered native scan built with init_datasource_exec reads a small remote file with one GET, against two without reuse.
• A prefetched small file is read with one GET.
• A file within the whole-file threshold is read with one GET, and a file one byte over the threshold is not.
• Ranges longer than the split size are fetched as parts that are all in flight at once, and return the requested bytes.
• The data wait time and the metadata storage reads are recorded.

CometExecSuite tests:

• "native scan serves a small file's data from its footer read" checks the default and the disabled setting through the Spark config.
• "Comet native metrics: scan" checks the new metric names.

Run locally:

• All datafusion-comet lib tests.
• The native scan tests in CometExecSuite on the default Spark profile (4.1).
• cargo clippy --all-targets --workspace -- -D warnings and make format.

I also simulated S3 with object_store's ThrottledStore, at 30 ms per request and about 100 MB/s per connection. The simulation is not committed.

Read	Before	After
Tiny single-file table	70 ms, 2 GETs	33 ms, 1 GET (reuse)
4 MiB file, 4 row groups	233 ms, 5 GETs	90 ms, 1 GET (whole-file threshold of 8 MiB)
One 32 MiB range	369 ms, 1 GET	120 ms, 4 GETs (8 MiB parts)

Not run yet: TPC-DS SF1000 on EKS against S3, which #7459 covers.

[andygrove]

TPC-DS benchmark of this PR (head 718f2d6cc), one run.

Setup: Spark 4.2.0 (Scala 2.13), TPC-DS SF 1000 Parquet (zstd), 103 queries, 2 iterations (mean), 32 executors x 16 cores on EKS with data in S3. The Spark properties are the same as in the Comet 1.2.0 run, apart from the Comet version.

Total time	Speedup over Spark
Spark 4.2.0: 994 s
Comet 1.2.0: 488 s	2.04x
This PR: 541 s	1.84x
Comet 1.1.0: 601 s	1.65x

• Compared with Comet 1.2.0, this PR is 0.90x in total (541 s vs 488 s) and 0.90x per query in geomean. 45 of 103 queries are more than 10% slower and 12 are more than 10% faster.
• Largest absolute slowdowns: q23b 30 s to 39 s, q88 10 s to 14 s, q24a 17 s to 20 s, q28 10 s to 12 s, q72 10.3 s to 12.4 s. Largest relative slowdowns are short queries: q56 1.0 s to 2.7 s, q94 2.9 s to 6.2 s, q58 1.1 s to 2.0 s.
• Largest gains: q54 6.3 s to 4.1 s, q47 5.8 s to 4.4 s, q01 6.4 s to 5.5 s, q68, q07, q08.
• The S3-heavy queries did not improve: q30 unchanged (1.8 s), q09 9.6 s to 10.4 s, q88 10.2 s to 13.9 s, q90 2.6 s to 3.3 s, q53 1.5 s to 2.3 s, q99 about the same.

Caveats:

• This is one run, and short queries vary by 10-20% between identical runs, so the per-query differences under about 1 s may be noise. I have not repeated either side.
• The 488 s run is the 1.2.0 release, and this PR is main at a4d716b63 (75 commits after 1.2.0) plus 4 commits. So the comparison includes everything merged since 1.2.0, not only this PR. I have not run main at a4d716b63 to separate them.