Structurally eliminate rg_plan / RowSelection drift in the parquet runtime prune path (retain_row_groups)

[zhuqi-lucas]

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 RowSelection over 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_plan trails 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 syncing rg_plan to the frontier via peek_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 carried RowSelection to 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) (or remaining_row_groups() -> impl ExactSizeIterator<Item = usize>) that filters the frontier in place, slicing the RowSelection alongside the row-group queue.

• DataFusion: replace rg_plan / RgPlanEntry / sync_rg_plan_to_decoder_frontier / advance_rg_plan_to / the Data-arm pop / and the into_builder() + is_at_row_group_boundary() dance with a single

  decoder.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.

cc @alamb @adriangb @hhhizzz

[zhuqi-lucas]

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 parallel rg_plan to drift (#24352).

So the plan for this issue is:

1. arrow-rs#10624 lands with_row_group_selections (+ rebuild preserving local selections).
2. DataFusion bumps parquet, then replaces into_overall_row_selection → global with_row_selection + rg_plan / sync_rg_plan_to_decoder_frontier / the into_builder().with_row_groups() dance with the row-group-local API.
3. Remove the stop-the-bleeding guards (fix(parquet): sync rg_plan to decoder frontier — fix wrong TopK results from re-reading already-delivered row groups (#24352) #24354, fix(parquet): don't runtime-prune row groups while a page-index RowSelection is live (#24355) #24359).

Have linked this from arrow-rs#10624. Deferring the design here to that issue rather than adding a competing retain_row_groups proposal.

[alamb]

Yeah, I agree, let's focus on apache/arrow-rs#10624 (thanks @haohuaijin for that). And thank you for this issue @zhuqi-lucas