Repository navigation
Structurally eliminate rg_plan / RowSelection drift in the parquet runtime prune path (retain_row_groups) #24358
Description
Follow-up to #24352 / #24354 (and the second instance #24355), from @adriangb's review on #24354.
Problem
The parquet runtime row-group prune path in PushDecoderStreamState maintains state parallel to the arrow-rs push decoder and updates it independently of the decoder's own frontier:
rg_plan: VecDeque<RgPlanEntry>— a DataFusion-side queue of pending row groups, and- a carried flat
RowSelectionover the concatenation of the remaining row groups (arrow-rs side).
Both must stay aligned with the decoder's row-group frontier, but nothing enforces that structurally, and each drift is a silent wrong-results bug:
- Wrong results: TopK dynamic filter + pushdown_filters re-reads already-delivered row groups (duplicate rows displace the true top-k) #24352 — a row group whose post-predicate selection is empty is silently finished without handing back a reader, so
rg_plantrails the decoder by one and a later rebuild re-reads an already-delivered row group. Fixed in fix(parquet): sync rg_plan to decoder frontier — fix wrong TopK results from re-reading already-delivered row groups (#24352) #24354 by syncingrg_planto the frontier viapeek_next_row_group(). - Wrong results: dynamic row-group pruning drops row groups without slicing the carried RowSelection #24355 —
into_builder().with_row_groups(new_indices)drops row groups without slicing the carriedRowSelectionto match, so selectors intended for a dropped RG are applied to the next surviving one.
The #24354 fix also couples rg_plan's correctness to row_group_pruner.is_some() (it syncs only when a pruner exists) — sound today because the pruner is the only consumer that rebuilds from rg_plan, but fragile: the invariant is enforced only where a pruner happens to be present.
Proposed direction
Give the decoder ownership of the drop so the row-group queue and the RowSelection cannot be updated independently. As @adriangb suggested:
-
arrow-rs: add
ParquetPushDecoder::retain_row_groups(impl FnMut(usize) -> bool)(orremaining_row_groups() -> impl ExactSizeIterator<Item = usize>) that filters the frontier in place, slicing theRowSelectionalongside the row-group queue. -
DataFusion: replace
rg_plan/RgPlanEntry/sync_rg_plan_to_decoder_frontier/advance_rg_plan_to/ theData-arm pop / and theinto_builder()+is_at_row_group_boundary()dance with a singledecoder.retain_row_groups(|rg| !pruner.should_prune(&[rg]));
This removes the parallel state entirely, so there is no alignment left to drift — closing the whole family (#24352, #24355) at the root and dropping the pruner.is_some() coupling.
Either repo can host the change; the in-place retain really wants to live in arrow-rs, which owns both the queue and the selection.
Activity
zhuqi-lucas commentedon Aug 14, 2026
Upstream dependency: apache/arrow-rs#10624 already tracks the arrow-rs-side API for this — @haohuaijin proposes
with_row_group_selections(vec![RowGroupSelection { row_group_index, selection }]), i.e. the push decoder accepts row-group-local selections directly, and explicitly "rebuilding a push decoder preserves the remaining local selections."That's a cleaner shape than the
retain_row_groups(FnMut)I sketched above, and it root-causes the whole drift family: with per-RG selections, dropping a row group on rebuild keeps every survivor's selection aligned by construction, so there's no global-selection slicing to get wrong (#24355) and no parallelrg_planto drift (#24352).So the plan for this issue is:
with_row_group_selections(+ rebuild preserving local selections).into_overall_row_selection→ globalwith_row_selection+rg_plan/sync_rg_plan_to_decoder_frontier/ theinto_builder().with_row_groups()dance with the row-group-local API.Have linked this from arrow-rs#10624. Deferring the design here to that issue rather than adding a competing
retain_row_groupsproposal.fix(parquet): don't runtime-prune row groups while a page-index RowSe…
alamb commentedon Aug 14, 2026
Yeah, I agree, let's focus on apache/arrow-rs#10624 (thanks @haohuaijin for that). And thank you for this issue @zhuqi-lucas
[branch-55]: don't runtime-prune row groups while a page-index RowSel…
fix(parquet): don't runtime-prune row groups while a page-index RowSe…
fix(parquet): don't runtime-prune row groups while a page-index RowSe…
refactor(parquet): strip empty row groups so PreparedAccessPlan stays…
haohuaijin commentedon Sep 22, 2026
Hi @zhuqi-lucas, have you already started working on the DataFusion integration of
with_row_group_selections? My work on #23883 also depends on this, so I’d be happy to take on the integration if you haven’t started yet. Just wanted to check with you first to avoid duplicating work. Thanks!zhuqi-lucas commentedon Sep 22, 2026
Hi @haohuaijin i haven’t started yet.
haohuaijin commentedon Sep 22, 2026
Thanks @zhuqi-lucas! I’ll take on the integration of
with_row_group_selections.refactor: use row-group-local selections for Parquet scans (apache#25608
refactor: use row-group-local selections for Parquet scans (apache#25608
refactor: use row-group-local selections for Parquet scans (apache#25608