Dandandan opened a new pull request, #23285: URL: https://github.com/apache/datafusion/pull/23285
## Which issue does this PR close? - N/A (perf improvement, no tracking issue) ## Rationale for this change DataFusion parallelizes parquet scans by splitting files into byte ranges (`FileGroupPartitioner`), but each row group is assigned entirely to the single range that contains the offset of its first data page. Files with fewer row groups than `target_partitions` therefore cannot use all cores: a file with one large row group is decoded by exactly one partition, no matter how many cores are available. This PR implements "morsel"-style splitting *within* row groups: a byte range that partially overlaps a row group now reads the proportional slice of that row group's rows via a `RowSelection`, so all partitions decode disjoint slices of the same row group in parallel. ### Benchmark results (M-series, 10 cores) TPC-DS SF=1 rewritten with a single row group per file (e.g. `store_sales` = 2.88M rows / 100MB / 1 row group): - **2.2x faster overall** (16.8s → 7.5s total, 99 queries × 5 iterations) - 82/99 queries faster, up to **5.3x** (q22), 0 with meaningful regressions (worst "slower" query is a 4ms query changing by <1ms) - CPU utilization on q22 (scan of 11.7M-row single-row-group `inventory`): 0.95 cores → 7.4 cores process-wide average; instruction overhead from splitting is only +7% TPC-H SF=1 (standard multi-row-group files, checking the new default doesn't regress the common case): - **1.16x faster overall** (1258ms → 1084ms), 13/22 queries faster, worst regression 1.13x on a 35ms query ## What changes are included in this PR? - `RowGroupAccessPlanFilter::split_by_range`: maps a row group's byte span (first dictionary/data page offset + compressed size) linearly onto its rows and assigns each range the row interval proportional to its byte overlap. Boundaries are computed with identical flooring integer arithmetic on both sides of every range boundary, so ranges that tile a file assign every row to exactly one range - no duplicates, no gaps. - The parquet opener uses `split_by_range` instead of `prune_by_range` when the new option is enabled; the old midpoint behavior is preserved when disabled. - `should_load_page_index` now also loads the page index when the access plan contains row selections, so the reader fetches and decodes only the pages covering each partition's slice (instead of whole column chunks). - New config option `datafusion.execution.parquet.split_row_groups_by_range`, **default `true`**, with proto serialization, `information_schema` and config docs updates. Existing machinery composes unchanged: page-index pruning intersects with the range selection (`scan_selection`), limit pruning counts selection rows, `reverse()` remaps selections, and sorted-scan range splitting produces contiguous in-order slices. ## Are these changes tested? - Unit tests for `split_by_range`: exact tiling across ranges, full-coverage keeps `Scan`, no-overlap skips, splitting across multiple row groups, zero-row slices, and respecting already-skipped row groups. - End-to-end opener tests reading a single-row-group file through 4 byte ranges (with and without predicate pushdown + page index): every row returned exactly once, in order, and work actually distributed across ranges; with the option disabled the old single-range behavior is asserted. - Verified at scale with a sqllogictest against TPC-DS `store_sales` (2.88M rows, single row group): counts/sums/distinct aggregates identical to the multi-row-group original, and the (item, ticket) primary key appears exactly once per row. - Full TPC-DS benchmark run returns identical row counts for all 99 queries with the option on and off. ## Are there any user-facing changes? New config option `datafusion.execution.parquet.split_row_groups_by_range` (default `true`). With the default, partitions of a ranged parquet scan that previously returned no rows (their range contained no row group start) now return the rows proportional to their byte range; total scan output is unchanged. 🤖 Generated with [Claude Code](https://claude.com/claude-code) -- 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]
