Support row-group-local row selections in the Parquet push decoder

Is your feature request related to a problem or challenge?

The Parquet reader currently accepts selected row groups and one global RowSelection:

builder
    .with_row_groups(row_group_indexes)
    .with_row_selection(row_selection)

Current DataFusion flow

DataFusion stores the scan decision for each row group separately in ParquetAccessPlan. Row-level filters update each entry through scan_selection:

row group 0: bitmap selection
row group 1: skip
row group 2: selector selection
row group 3: scan all

Before constructing the Parquet reader, prepare calls into_overall_row_selection to convert this plan into two values:

row groups: [0, 2, 3]
selection:  one global RowSelection covering row groups 0, 2, and 3

Full row groups become select(row_group_rows), while partial selections are appended to the same selection stream. Mixed bitmap and selector inputs must also be converted to one common representation.

DataFusion then passes both values to arrow-rs in DecoderBuilderConfig::build:

builder
    .with_row_groups(row_group_indexes)
    .with_row_selection(global_selection)

When arrow-rs starts reading, RowGroupFrontier::next_readable_row_group uses the row-group metadata to partition this global selection back into row-group-local selections. The complete flow is therefore:

per-row-group selections
    -> one global RowSelection
    -> per-row-group selections during decoding

Describe the solution you'd like

Allow the push decoder to accept row-group-local selections directly:

pub struct RowGroupSelection {
    row_group_index: usize,
    selection: Option<RowSelection>,
}

builder.with_row_group_selections(vec![
    RowGroupSelection::new(0, Some(bitmap_selection)),
    RowGroupSelection::new(2, Some(selector_selection)),
    RowGroupSelection::new(3, None),
]);

Proposed behavior:

Entries are decoded in the supplied order.
Omitted row groups are skipped.
None reads the entire row group.
Each row group retains its bitmap or selector representation.
Existing with_row_groups and with_row_selection APIs remain supported.
Rebuilding a push decoder preserves the remaining local selections.

Describe alternatives you've considered

keep current behaviour

Additional context

find this when work on apache/datafusion#24186

@alamb

cc @alamb , what do you think about this, this can remove the each row group row selection -> full row selection -> each row group row selection convert, but for e2e performance it do not imporve in normal query i think.

but it have two bennefit for the user that already have a RowSelection.

user able to choose bitamp or rowselecter represention in row_group level, current only support file level
it can improve some performance when user have a fragmented rowselector(the case describe in Allow RowSelection to be backed by a BooleanBuffer to reduce memory usage #10140 (comment)). but for fragmented case, we already have the bitmap...

i think this is a good idea (and a very clear ticket)

pub struct RowGroupSelection {
    row_group_index: usize,
    selection: Option<RowSelection>,
}

builder.with_row_group_selections(vec![
    RowGroupSelection::new(0, Some(bitmap_selection)),
    RowGroupSelection::new(2, Some(selector_selection)),
    RowGroupSelection::new(3, None),
]);

In my mind the trick will be how to mesh the existing apis (with_row_groups and with_row_selection) and make sure we have a coherent documentation of what will happen if the user calls each one

@haohuaijin

Just a note that this missing capability is also the root cause of a wrong-results bug family in DataFusion, not only a performance/ergonomics improvement.

Because DataFusion collapses the per-row-group selections into one global RowSelection and then rebuilds the push decoder to drop row groups at runtime (into_builder().with_row_groups(...)), the global selection is not sliced to match the dropped row groups — selectors intended for a dropped RG get applied to the next surviving one, silently returning wrong rows:

Wrong results: dynamic row-group pruning drops row groups without slicing the carried RowSelection datafusion#24355 — runtime dynamic row-group pruning drops RGs without slicing the carried RowSelection.
Wrong results: TopK dynamic filter + pushdown_filters re-reads already-delivered row groups (duplicate rows displace the true top-k) datafusion#24352 — a related drift in the same parallel-state path.

The behavior you list — "Rebuilding a push decoder preserves the remaining local selections" — is exactly what makes both bugs structurally impossible: if selections stay row-group-local, dropping a RG on rebuild keeps every survivor's selection aligned by construction, and DataFusion can also delete the parallel rg_plan bookkeeping it maintains today.

DataFusion has shipped minimal stop-the-bleeding fixes for both (apache/datafusion#24354, apache/datafusion#24359), and is tracking the structural fix on its side in apache/datafusion#24358, which depends on this API landing here. Happy to help with the DataFusion-side migration once with_row_group_selections is available. Thanks @haohuaijin for writing this up.