andygrove opened a new issue, #7459: URL: https://github.com/apache/datafusion-comet/issues/7459
### What is the problem the feature request solves? Hadoop 3.5.0 made the S3 Analytics Accelerator (AAL) the default S3A input stream, which made Spark 4.2.0 about 30% faster on TPC-DS SF1000 in S3 (#7451). Comet's native Parquet scan reads through `object_store` and does not use that stream. Reading the AAL, Hadoop, parquet-java and Spark source shows that AAL mostly brings Spark's per-file request pattern close to what Comet's native scan already did. These are serial S3 round trips per file, traced from source rather than measured: | Read | Spark classic | Spark + AAL | Comet | |---|---|---|---| | 5 KB single-file dimension table | HEAD + 4-5 GETs | HEAD + 1 GET | 2 GETs (footer, then data) | | 128 MB split, 3 of 20 columns, footer and page index within 512 KiB | 5 or more | 3 (HEAD, tail, data) | 2 (tail, data) | The Spark baseline caught up and Comet's time did not move. In the published 1.1.0 results (Spark 4.2.0 on AAL), Comet's lead is smallest on the scan-and-aggregate queries that AAL helped most: | Query | q09 | q28 | q88 | q90 | q53 | q99 | All queries | |---|---|---|---|---|---|---|---| | Comet speedup | 1.35x | 1.38x | 1.24x | 1.38x | 1.28x | 1.32x | 1.84x | AAL still does three things better than Comet's scan: 1. **Small files.** AAL reads an object of up to 8 MiB with one GET when it is opened. Comet's footer read (`METADATA_SIZE_HINT`, 512 KiB) already returns the whole file when the file is smaller than that. DataFusion drops those bytes once the footer is decoded (`datafusion-datasource-parquet` 55.2.0 `metadata.rs:305-319`), so the data read fetches them again. Dimension tables are read this way, and their scans feed the broadcasts that the fact-table stage waits for. 2. **Large reads.** AAL splits a contiguous read of more than about 11 MiB into 8 MiB GETs that run in parallel. `object_store` merges ranges whose gaps are at most 1 MiB, with no limit on the merged size, and fetches each merged range with one GET, 10 at a time. Adjacent projected columns of a large row group can become one GET of tens of MB, and a single S3 connection delivers on the order of 100 MB/s. 3. **Footer reads of large files.** AAL reads the last 32 KiB plus the 1 MiB before it in parallel, or 1 MiB plus 8 MiB for files over 1 GiB. Comet reads the last 512 KiB, so a larger footer and page index takes one or two more serial GETs. No metric shows how often that happens. There is also no metric for how long a scan waits on data reads. That wait bounds what most of these changes, and the read-ahead in #7093, can save. ### Describe the potential solution Each change is in `native/core/src/parquet/eager_page_index_reader_factory.rs`, behind a config so it can be compared on and off: - [ ] A metric for the wall time a scan spends waiting in `get_byte_ranges`, and a count of the GETs each metadata load makes. - [ ] Reuse the footer read's bytes. Keep the tail that the metadata load fetched, and serve data ranges inside it from memory. This reads no extra bytes. The next-file metadata prefetch from #7094 has to hand the tail over as well. - [ ] Read small files whole. For a file at or below a threshold, make the footer read cover the whole file, so the file costs one GET. AAL uses 8 MiB. The right default depends on how much of a file a query projects, so measure it. - [ ] Split large merged ranges into parallel GETs in `ScanIoObjectStore::get_ranges`. AAL uses 8 MiB parts above about 11 MiB. - [ ] Size the footer read by file size, if the metadata GET count shows extra round trips. Experiments, on TPC-DS SF1000 in S3 against Spark 4.2.0 on AAL: - [ ] From the existing event logs, collect `scan_io_*`, `metadata_load_time` and `time_elapsed_opening` for q09, q28, q88, q90, q53 and q99, and the durations of the stages that scan dimension tables. - [ ] Each config on and off, as the median of interleaved runs, for the whole run and for the queries above. - [ ] `scan_io_metadata_cache_hits` with #7082. On 32 executors, both splits of a two-split file run on the same executor about 1 time in 32, so its effect depends on how many splits each file has. - [ ] The same comparison on local disk, where the changes should be neutral. ### Additional context - #7451 has the investigation and the AAL details. - Overlapping I/O with decode is tracked in #7093 (apache/datafusion#23492 for row groups, #7094 for the next file's metadata). AAL does not do this for Spark 4.2. It never prefetches the next row group, and its Parquet column prefetch is inactive because parquet-java 1.17 reads row groups with `readVectored`. - #7082 shares footers and page indexes across the tasks of a stage. It is the counterpart of AAL's block cache, whose entries expire about a second after their last access. It does not cache data, so it does not overlap with the items above. - #4828 proposes a cross-query data cache, which AAL does not have. -- 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]
