andygrove opened a new pull request, #7460:
URL: https://github.com/apache/datafusion-comet/pull/7460
## 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 #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 #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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
