andygrove opened a new issue, #7093: URL: https://github.com/apache/datafusion-comet/issues/7093
### What is the problem the feature request solves? A native Parquet scan has no I/O in flight while it decodes, so on high-latency storage every fetch's round trip adds straight to task time. Two places in DataFusion 55.2 serialize the fetches. Between files, DataFusion 54 rebuilt `FileStream` on the morsel API. Its `ScanState` keeps a single I/O outstanding ([scan_state.rs#L42-L43](https://github.com/apache/datafusion/blob/55.2.0/datafusion/datasource/src/file_stream/scan_state.rs#L42-L43)) and plans the next file only after the current reader has returned its last batch ([scan_state.rs#L268](https://github.com/apache/datafusion/blob/55.2.0/datafusion/datasource/src/file_stream/scan_state.rs#L268)). DataFusion 52 and 53 started opening the next file as soon as the current one was open ([file_stream.rs#L100-L107](https://github.com/apache/datafusion/blob/53.0.0/datafusion/datasource/src/file_stream.rs#L100-L107)). Comet lost that overlap when it moved to DataFusion 54 in #4062, so 1.0.0 and 1.1.0 load each file's footer and page index, and its Bloom filters when a predicate uses them, only after the previous file is finished. Partitions made of many small files pay the most. Spark packs up to about 32 of them into one 12 8 MB partition, since each file costs `spark.sql.files.openCostInBytes` (4 MB), and every one adds a serial metadata round trip before its data fetch. Within a file, DataFusion's per-file driver decodes row group N one batch per poll and asks for row group N+1's byte ranges only once N is exhausted ([push_decoder.rs#L396-L421](https://github.com/apache/datafusion/blob/55.2.0/datafusion/datasource-parquet/src/push_decoder.rs#L396-L421)). Nothing is fetched while N is decoded and passed through the operators above the scan. Two things already overlap. A plan with no JVM input runs on a Tokio worker and hands batches to the Spark task thread through a two-slot channel ([jni_api.rs#L1270-L1291](https://github.com/apache/datafusion-comet/blob/a0cbb0fa06515acd5647538025a3a894be71b9bd/native/core/src/execution/jni_api.rs#L1270-L1291)), and on remote stores the coalesced ranges of a single fetch go out ten at a time. ### Describe the potential solution Both changes fit in the `ParquetFileReaderFactory` and `AsyncFileReader` that Comet already supplies to DataFusion ([eager_page_index_reader_factory.rs](https://github.com/apache/datafusion-comet/blob/a0cbb0fa06515acd5647538025a3a894be71b9bd/native/core/src/parquet/eager_page_index_reader_factory.rs)), so neither needs a DataFusion or arrow-rs change. 1. Read ahead the next file's metadata. The factory gets the scan's ordered file list. Once a file's metadata has loaded, it starts a background task that loads the next file's footer and page index into the plan's `FileMetadataCache`, and the next file's open waits for that task instead of fetching again. A follow-up can also fetch the next file's first row group inside its split. 2. Read ahead the next row group. When row group N's ranges return, map them to the row group's column chunks, start fetching row group N+1's chunks for the same columns, and serve the next request from that fetch when it covers it, reading directly otherwise. The guess has to mirror DataFusion's `prune_by_range`, or the end of every split fetches the next split's row group for nothing. Statistics, Bloom filter, page index and dynamic pruning can still skip N+1, so mispredictions need a cap, and the buffer should come from the plan's memory pool, with the read-ahead skipped when the pool can't cover it. This only helps tasks that read several row groups, and not when the executor's network is already saturated. A metric for the time spent waiting in `get_byte_ranges` would bound what the second change can save. `metadata_load_time` already bounds the first. Some alternatives I set aside. Wrapping `ParquetSource`'s morselizer would open the next file exactly rather than by guessing, but Comet's dynamic filter pushdown downcasts the scan's source to `ParquetSource`. A `spawn_buffered` stage between the scan and its parent uses a second core, and a buffer of one or two batches can't hide a row-group-sized fetch. The native Iceberg scan reads through iceberg-rust's own `AsyncFileReader`, and `spark.comet.scan.icebergNative.dataFileConcurrencyLimit` already overlaps files there. ### Additional context #4695 proposes asynchronous prefetch alongside a node-local fragment cache, and its prefetch piece hasn't been built. If #4828 (object-store data cache) lands, read-ahead could fill that cache. -- 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]
