alamb opened a new pull request, #26387: URL: https://github.com/apache/datafusion/pull/26387
## Which issue does this PR close? - Part of https://github.com/apache/datafusion/issues/24393 - Related to https://github.com/apache/datafusion/issues/20324 and https://github.com/apache/datafusion/issues/3463 - Supersedes the prototype in https://github.com/apache/datafusion/pull/24395 **Note: this PR builds on https://github.com/apache/datafusion/pull/26384 (pin to arrow-rs `main`) and can not be merged until the next arrow release, because it uses the new `ParquetPushDecoder::scan_plan` API from apache/arrow-rs#10555.** Only the last commit is specific to this PR. ## Rationale for this change Enabling `pushdown_filters` today also changes the I/O pattern: instead of one request per row group, the reader fetches the columns for each filter predicate, then (after evaluating the filters) the remaining projected columns. When a file has no offset index (e.g. the ClickBench dataset), those extra round trips cannot prune any bytes, so they are pure overhead. This is one of the things that has blocked enabling `pushdown_filters` by default. This PR separates the I/O pattern from filter evaluation with a new config option, `datafusion.execution.parquet.progressive_io`: | `pushdown_filters` | `progressive_io` | behavior | |---|---|---| | false | (any) | unchanged: one request per row group | | true | **false (default)** | filters evaluated during decode, but everything the row group needs (filter columns and projected columns) is fetched with **one request per row group** | | true | true | progressive fetch (filter columns first, then the projected columns of the rows that passed), but only when the file has an offset index, where it can actually prune bytes; otherwise one request per row group | ## What changes are included in this PR? - New `progressive_io` config option (config, proto, docs, `ParquetSource::with_progressive_io`), resolved table-or-session like `pushdown_filters` - `PushDecoderStreamState` gets a `one_shot_io` flag. When set, at each row-group boundary where a real `RowFilter` is installed, the stream asks the decoder for its own `ScanPlan`, takes the ranges of the next row group, coalesces the pages of each column chunk into contiguous runs, fetches them with a single `get_byte_ranges` call and pushes them into the decoder *before* driving it. The decoder then finds everything buffered and never pauses for I/O between evaluating the filters and decoding the output. Compared to the earlier prototype (#24395), which computed whole-column-chunk ranges itself from the union of the filter and projection masks, using the decoder's `ScanPlan` means: - The ranges are exactly what the decoder will read, so the one-shot request fetches only the pages a page-index row selection keeps. The prototype had to fall back to progressive I/O whenever the page index pruned rows; that limitation is gone (see the new `*_page_index` tests). - No DataFusion-side knowledge of the decoder's fetch rules (dictionary pages, predicate cache, fused predicates, ...) is needed, and the plan automatically tracks decoder rebuilds at row-group boundaries (dynamic row group pruning, the per-row-group `fully_matched` filter toggle) because it is taken fresh at each boundary. - Anything the plan does not cover is simply requested by the decoder and fetched as before, so correctness never depends on the plan. Pages of one column chunk are merged into contiguous runs because the decoder requests whole column chunks when there is no row selection and only uses a requested range that lies within a single pushed buffer. Column chunks are deliberately not merged with each other, so the request has the same shape (one range per column chunk) as a scan without a row filter; `object_store` coalesces adjacent ranges anyway. Memory: the decoder releases a row group's buffered bytes (including pushed bytes it did not request) when it is done with the row group (apache/arrow-rs#11235), so peak buffered bytes per row group are the same as for the non-pushdown reader. ## What is the testing strategy for this PR? - New request-count tests in `object_store_access.rs`: - `query_single_parquet_file_pushdown_filters_one_shot_io`: with `pushdown_filters=true` and the default `progressive_io=false`, each row group is fetched with exactly one request (and row group 2, which is fully matched by statistics, reads only the projected column) - `query_single_parquet_file_pushdown_filters_progressive_io`: `progressive_io=true` restores the progressive pattern - `*_page_index` variants on a file with five pages per column chunk, showing the one-shot request contains only the pages the page index keeps, plus a no-pushdown baseline for comparison - Unit tests for the range coalescing in `push_decoder.rs` - Existing filter pushdown tests (`parquet_integration`, 245 tests), `datafusion-datasource-parquet` tests and the parquet sqllogictests pass unchanged ### An observation While writing the page-index tests I noticed that once a `RowFilter` is installed, the arrow-rs decoder fetches every data page of the *output* columns, even pages the page-index row selection skips (compare `query_single_parquet_file_multi_page_no_pushdown`, which skips `b`'s first two pages, with the `*_page_index` tests, which read all of `b`). This happens with both `progressive_io` settings and with `force_filter_selections` on or off, so it is independent of this PR, but it may be worth a look upstream. ## Are there any user-facing changes? Yes: a new configuration option, and (intentionally) a changed default I/O pattern for users who enable `pushdown_filters`. Benchmarks (`clickbench_pushdown`) to follow. -- 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]
