adriangb opened a new pull request, #11334: URL: https://github.com/apache/arrow-rs/pull/11334
Stacked on https://github.com/apache/arrow-rs/pull/10555. Review only the commits after ec3b0d2c50. # Which issue does this PR close? - Part of https://github.com/apache/arrow-rs/issues/6946. # Rationale for this change A caller that reads ahead asks the plan for its first range: ```rust let decoder = ParquetPushDecoderBuilder::new_with_metadata(metadata).build()?; let first = decoder.scan_plan().next(); // what the first fetch waits for ``` Before the first range, the planner starts one cursor for each column of the stage. On a wide file this cost is O(columns), and each column used about 200 bytes of heap memory. This PR makes each column cost less time and less memory. The plan output (order and contents) does not change. Time of `scan_plan().next()`, one row group of 1,000 rows, every column, no selection (criterion, `push_decoder/scan_plan/first_range/all`): | Columns × pages | Before | After | Change | |---|---|---|---| | 100 × 10 | 2.08 µs | 0.84 µs | −59% | | 1,000 × 10 | 20.2 µs | 6.39 µs | −68% | | 10,000 × 10 | 179 µs | 63.9 µs | −64% | | 100,000 × 10 | 3.16 ms | 1.83 ms | −41% | | 1,000 × 100 | 17.7 µs | 6.77 µs | −62% | Heap memory of one `scan_plan().next()` call (counting global allocator, same files): | Columns | Allocations before | Allocations after | Bytes before | Bytes after | |---|---|---|---|---| | 100 | 13 | 7 | 22,420 | 4,788 | | 1,000 | 16 | 7 | 203,064 | 41,688 | | 10,000 | 20 | 7 | 2,419,104 | 410,688 | | 100,000 | 23 | 7 | 22,089,136 | 4,100,688 | At 100,000 columns the time is limited by memory reads of the column metadata and the offset index (about 1.2–1.4 ms for only these reads), so the gain there is smaller. # What changes are included in this PR? Terms: * **Cursor**: the state that gives the next ranges of one column chunk, in file order. * **Stage**: one decoding step of a row group (a predicate, or the output projection). The planner merges the cursors of a stage with a heap, by first row. | Commit | Change | |---|---| | bench | Adds a 100,000 column case to the `push_decoder/scan_plan` benchmark. | | perf | A cursor is now only its heap entry: the sort key of its next range, the position of the column, and which range of the column chunk is next (dictionary page, data page position, or whole chunk). When an entry is popped, the planner computes the end and kind of the range again from the page locations. A stage keeps the list of selected pages of a column only when the row selection removes pages. | Before, each cursor held a full `PlannedRange` for its next range (about 128 bytes), and the heap held a copy of its sort key. The planner also looked up the offset index two times for each column. `ColumnFetch` and `StageSchedule` are still the only source of the bytes that the decoder fetches. The planner does not copy decoder logic. # Are these changes tested? Yes, by the existing tests. The tests that compare the plan order, that check that the plan covers every later decoder request, and the lazy-plan test pass. The lazy-plan test now counts heap entries instead of cursors, because the cursor list is gone (one entry for each projected column, as before). Benchmarks (`cargo bench -p parquet --features arrow,async --bench push_decoder -- scan_plan`, baseline saved on the bench commit, Apple M-series laptop). `first_range` is the first `next()`, `whole_scan` is `count()`: | Case | first_range before | first_range after | whole_scan before | whole_scan after | |---|---|---|---|---| | all/100cols_10pages | 2.08 µs | 0.84 µs | 30.1 µs | 22.6 µs | | selection/100cols_10pages | 23.7 µs | 21.7 µs | 61.8 µs | 51.2 µs | | narrow/100cols_10pages | 574 ns | 349 ns | 2.57 µs | 1.42 µs | | all/1000cols_10pages | 20.2 µs | 6.39 µs | 383 µs | 320 µs | | selection/1000cols_10pages | 227 µs | 216 µs | 642 µs | 589 µs | | narrow/1000cols_10pages | 1.06 µs | 647 ns | 2.70 µs | 1.71 µs | | all/10000cols_10pages | 179 µs | 63.9 µs | 5.26 ms | 4.42 ms | | selection/10000cols_10pages | 2.27 ms | 2.19 ms | 7.77 ms | 7.05 ms | | narrow/10000cols_10pages | 3.50 µs | 3.81 µs | 5.44 µs | 4.95 µs | | all/100000cols_10pages | 3.16 ms | 1.83 ms | 67.2 ms | 58.1 ms | | selection/100000cols_10pages | 23.7 ms | 22.8 ms | 91.3 ms | 84.3 ms | | narrow/100000cols_10pages | 32.1 µs | 32.1 µs | 31.1 µs | 32.3 µs | | all/1000cols_100pages | 17.7 µs | 6.77 µs | 3.84 ms | 3.45 ms | | selection/1000cols_100pages | 325 µs | 326 µs | 749 µs | 711 µs | | narrow/1000cols_100pages | 837 ns | 676 ns | 20.1 µs | 12.3 µs | The `narrow` cases read 10 columns, so they do not depend on this change. Three alternating reruns of `narrow` at 10,000 and 100,000 columns gave equal or faster times after the change (the +9% in the table is noise on this machine). The `selection` cases stay slow: `ColumnFetch::new` calls `RowSelection::scan_ranges` for each column (about 6 allocations per column). This PR does not change that. # Are there any user-facing changes? No. The public API and the plan output do not change. 🤖 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]
