adriangb commented on PR #11165: URL: https://github.com/apache/arrow-rs/pull/11165#issuecomment-5780778399
Thanks @sunchao. I worked with an agent to compare to the work I had done in https://github.com/apache/datafusion/pull/24086 / https://github.com/apache/arrow-rs/pull/10555 and generated this feedback from that discussion. It's a mix of agent written and hand written. Your design is much better in that the plan comes from the decoder's own frontier, so it cannot drift from what the decoder reads. We should keep that. My POC of the same idea is at *page* granularity in https://github.com/apache/arrow-rs/pull/10555, driven from https://github.com/apache/datafusion/pull/24086. I think we should go with page granularity. ## Why page granularity, not row groups Measured in apache/datafusion#24086 with simulated object-store latency and a 100 MB budget per stream: | workload | row-group readahead | page-granular | |---|---|---| | TPC-H SF10 | 4.9x | **6.35x** (22/22 queries faster) | | TPC-H SF1 | 1.44x | **1.69x** | | ClickBench, same files rewritten with a page index | 1.33x | **1.54x** (35 faster, 0 slower) | | one 300 MB row group: time to first batch | 700 ms | **112 ms** | | one 300 MB row group: peak memory | 283 MB | **100 MB** | GET count and bytes fetched were the same for both. The gain comes from decoding a row group while the rest of it is still in flight. A row-group preview cannot give this: the decoder still waits for the whole row group before it hands out a reader. Memory is important here: pages can fundamentally be bounded in size, row groups are bounded by row count (and bounding by size does not work well if you have large variations in per-row sizes between columns). ## Proposed shape ```rust impl ParquetPushDecoder { /// The bytes the remaining scan will read, in the order decoding needs them. /// /// A pure function of the decoder's plan (row groups, projection, /// selections, offset/limit, filters). Independent of buffered data. /// Never does I/O. Stable until the decoder is rebuilt. pub fn scan_plan(&self) -> impl Iterator<Item = PlannedRange> + '_; } pub struct PlannedRange { pub range: Range<u64>, /// Selected-row space, after offset/limit. The bytes can be released /// once the decode cursor passes `last_row`. pub first_row: u64, pub last_row: u64, pub row_group: usize, /// Leaf column index. pub column: usize, /// Dictionary | Data | ColumnChunk (file has no offset index) pub kind: PageKind, /// Predicate index in the `RowFilter`, or `Projection`. pub stage: Stage, /// `false` for stage 0 and for unfiltered scans. `true` when an earlier /// stage may filter out every row this page serves. pub conditional: bool, } ``` Two channels, with different guarantees: ```mermaid flowchart LR D["Push decoder"] -->|"scan_plan: upper bound, row-tagged, for speculation"| S["Caller scheduler"] D -->|"NeedsData: exact, ordered, always correct"| S S -->|"push_data, any byte shape"| D ``` ## Requested changes | # | Now | Proposed | Why | |---|---|---|---| | 1 | One entry per row group | One entry per page, with `first_row` / `last_row` | Row tags are what make batch-granular fetch and eviction possible | | 2 | Depth 1 or 2 | Lazy iterator over the rest of the scan. The caller stops at its own budget | Depth is caller policy. This also removes the per-boundary clone (@viirya's quadratic point) | | 3 | Omits ranges already buffered | Independent of buffer state. Stable for the decoder's lifetime | The caller tracks its cache (@alamb's point). No snapshot-invalidation rules to document | | 4 | `None` without offset index | `kind: ColumnChunk`, one entry per column chunk | Same API, coarser. The docs tell users to write a page index | | 5 | `None` with row filters | Upper bound, with `stage` and `conditional` tags | The caller chooses how speculative to be (table below) | | 6 | No page position | Report `row_group`, `column`, `kind` | The caller can group by row group or column chunk to choose its own granularity | | 7 | `into_inner()` in this PR | Separate PR | Unrelated to planning, and already accepted in principle | ## How different callers use one plan | storage | policy | reads from the plan | |---|---|---| | archival, very high cost per request | fetch everything, coalesce into a few GETs | all entries, including `conditional` | | balanced object storage | fill an N MB window in `(first_row, stage)` order | certain entries first, conditional ones while budget remains | | fast SSD, seeks are free | fetch only what the next batch needs | `NeedsData` at batch granularity (follow-up 1) | Predicate order stays with the caller: the `stage` index is the position in the `RowFilter` the caller built. ## Tests Exact equality with `NeedsData` no longer holds (pages vs merged ranges). The invariant to keep: per row group, the union of planned bytes equals the union of bytes that `NeedsData` requests. The cases here (offset/limit, per-group selections, reversed and duplicate groups, empty selections) are the right ones. #10555 has the raw-row to selected-row mapping and its tests. ## Follow-ups, not for this PR 1. **Batch-granular `NeedsData` inside the decoder**, with incremental `RowFilter` evaluation. Then the plan is exact and authoritative. Working branch: pydantic/arrow-rs `claude/push-decoder-batch-granular` (`2603e2ef66`, randomized equivalence tests). I will open it after this lands. 2. **Release buffers by row position, whatever shape they were pushed in.** Today `clear_ranges` frees exact-match ranges only, so a coalesced or partly-selected push is held for the life of the stream. This is what silently disables the budget in https://github.com/apache/datafusion/pull/23492 under page-skipping selections. Small separate PR. 3. `into_inner()`. -- 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]
