andygrove opened a new pull request, #7094: URL: https://github.com/apache/datafusion-comet/pull/7094
## Which issue does this PR close? Part of #7093. ## Rationale for this change DataFusion 54 rebuilt `FileStream` on the morsel API, and its `ScanState` plans a file only after the previous file's reader has returned its last batch. DataFusion 52 and 53 opened the next file while reading the current one, so since the DataFusion 54 upgrade (#4062) every file boundary in a native Parquet scan waits for a footer and page-index round trip before the file's data fetch can start. A task that reads many small files pays that latency once per file. ## What changes are included in this PR? `EagerPageIndexReaderFactory`, which already supplies Comet's `AsyncFileReader` to DataFusion, now gets the scan's file groups. Once a file's metadata has loaded, its reader spawns a task that loads the next file's footer and page index through the plan's metadata cache. When DataFusion opens that file, its `get_metadata` takes the task and waits for it instead of reading the footer again. A prefetch that fails or is cancelled yields nothing, and the open then loads the metadata itself, so any error still comes from the open. Dropping the scan aborts a load in flight. The metadata load moved into a `load_metadata` helper that the open and the prefetch share, so footer and metadata byte accounting is the same on both paths. The new `spark.comet.parquet.prefetchNextFileMetadata.enabled` config defaults to true. It reaches the scan through a `SessionConfig` extension, `CometParquetScanOptions`, so every scan that `init_datasource_exec` builds gets it, Delta's included, without a proto change. Encrypted scans never prefetch, because their metadata bypasses the cache. The new `scan_io_metadata_prefetches` metric counts loads read from storage ahead of their file's open, and the open that uses one counts as a `scan_io_metadata_cache_hits`. This PR only covers metadata. Fetching the next file's first row group ahead, and the next row group within a file, are the other parts of #7093. It composes with #7082, which shares the metadata cache across a stage's tasks, since the prefetch writes into whichever cache the plan has. ## How are these changes tested? New Rust tests in `eager_page_index_reader_factory.rs` cover the prefetch across three files, an open that waits for its load in flight instead of reading again (using a store that holds the read at a gate), no prefetch without the option or after the last file, the next-file lookup skipping further ranges of the same file, a failed prefetch leaving the open to report its own error, and dropping the scan aborting a load in flight. `prefetching_the_next_files_metadata_keeps_the_scan_result` in `parquet_exec.rs` runs a three-file `DataSourceExec` with and without the prefetch and checks that the batches match and the metrics show one storage metadata load instead of three. A new `CometExecSuite` test reads four files in one task with the config on and off. Locally, the native crate's unit tests pass, and so do `CometExecSuite`, `ParquetReadV1Suite` and `ParquetReadFromFakeHadoopFsSuite`. I also measured a full scan with an optimized build, using a scratch benchmark that isn't part of this PR. It reads local files through object_store's `ThrottledStore`, which adds a fixed latency to every request and optionally a per-byte cost. Each number is the median of five interleaved runs per arm. | Files | Latency per request | Bandwidth per request | Off | On | Speedup | | --- | --- | --- | ---: | ---: | ---: | | 32 of 2 MB, 1 row group each | none | unlimited | 11 ms | 10 ms | 1.13x | | 32 of 2 MB, 1 row group each | 10 ms | unlimited | 823 ms | 442 ms | 1.86x | | 32 of 2 MB, 1 row group each | 30 ms | unlimited | 2120 ms | 1106 ms | 1.92x | | 32 of 2 MB, 1 row group each | 10 ms | 100 MB/s | 1621 ms | 1079 ms | 1.50x | | 32 of 2 MB, 1 row group each | 30 ms | 100 MB/s | 2897 ms | 1730 ms | 1.67x | | 8 of 19 MB, 4 row groups each | none | 100 MB/s | 1759 ms | 1718 ms | 1.02x | | 8 of 19 MB, 4 row groups each | 30 ms | 100 MB/s | 2983 ms | 2709 ms | 1.10x | Small files gain the most, because each one needed two round trips, metadata and then data, and the prefetch hides the first. Larger files with several row groups still fetch their row groups one after another, which is the second part of #7093. -- 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]
