Describe the bug
With datafusion.execution.parquet.pushdown_filters = true and dynamic filter pushdown enabled (on by default), a query of the shape SELECT b FROM t WHERE <predicate on a> ORDER BY b LIMIT k can silently return wrong results: rows satisfying the predicate are dropped and replaced by later ones. No error, no warning.
This is a second, independent instance of the failure family in #24352, and it is not fixed by #24354. Verified against main at 0a429a37db and against #24354 at 44f7fc71c5; both return the wrong answer.
Root cause
The push decoder carries one flat RowSelection covering the concatenation of the remaining row groups, in order. RowGroupFrontier::next_readable_row_group consumes it positionally (selection.split_off(row_count) per row group), so alignment between the selection and the row-group queue is a hard invariant.
At a row-group boundary, PushDecoderStreamState::transition prunes row groups the dynamic predicate proves unwinnable and rebuilds the decoder:
let rebuilt = decoder.into_builder()?.with_row_groups(new_indices).build();
builder_from_remaining correctly carries the unconsumed selection into the rebuilt builder, but with_row_groups(new_indices) then removes row groups from under it without slicing the selection to match. The selectors intended for a dropped row group are applied to the next surviving one.
The plan-building path already knows about this coupling: PreparedAccessPlan::reorder_by_statistics refuses to run when a row selection is present ("Skipping RG reorder: row_selection present") because remapping the selection is too complex. The runtime rebuild makes the same kind of remap-free change to the row-group set with no equivalent guard.
To Reproduce
Deterministic, 400 rows, 4 row groups, 100% reproducible. In datafusion-cli:
COPY (
SELECT
CAST(CASE WHEN i / 100 = 1 THEN i % 100 ELSE 100 + (i % 100) END AS BIGINT) AS a,
CAST(CASE
WHEN i < 100 THEN 1000 + i
WHEN i < 200 THEN 2000 + (i - 100)
WHEN i < 300 THEN 3000 + (i - 200)
ELSE (i - 300)
END AS BIGINT) AS b
FROM generate_series(0, 399) AS t(i)
)
TO 'rgsel.parquet'
STORED AS PARQUET
OPTIONS (
'format.max_row_group_size' '100',
'format.data_page_row_count_limit' '10',
'format.write_batch_size' '10'
);
SET datafusion.execution.target_partitions = 1;
SET datafusion.execution.parquet.pushdown_filters = true;
CREATE EXTERNAL TABLE t (a BIGINT NOT NULL, b BIGINT NOT NULL)
STORED AS PARQUET LOCATION 'rgsel.parquet';
SELECT b FROM t WHERE a >= 50 ORDER BY b ASC LIMIT 5;
format.write_batch_size is required: data_page_row_count_limit is only enforced at write-batch boundaries, so at the default of 1024 each 100-row row group becomes a single page and the page index cannot prune within it.
Layout:
| RG |
b (sort col) |
a (filter col) |
a >= 50 selects |
| 0 |
1000..1099 |
100..199 |
all 100 rows |
| 1 |
2000..2099 |
0..99 |
rows 50..99 only (partial) |
| 2 |
3000..3099 |
100..199 |
all 100 rows |
| 3 |
0..99 |
100..199 |
all 100 rows |
Expected behavior
Actual
Controls (same binary, same file, one setting changed):
enable_dynamic_filter_pushdown |
parquet.pushdown_filters |
result |
| true |
true |
wrong |
| false |
true |
correct |
| true |
false |
correct |
| false |
false |
correct |
Mechanism
- Page-index pruning on
a >= 50 prunes the first five pages of RG 1 only. The decoder receives a RowSelection of [RG0: select 100][RG1: skip 50, select 50][RG2: select 100][RG3: select 100].
- RG 0 is read first and its 100 selectors are consumed. The remaining selection covers RG 1, 2, 3.
ORDER BY b ASC LIMIT 5 tightens the TopK threshold to about b < 1004. At the boundary the runtime pruner proves RG 1 (min 2000) and RG 2 (min 3000) unwinnable and rebuilds with row_groups = [3].
- The carried selection is unchanged, so RG 3 is decoded under RG 1's
skip 50, select 50. Rows b = 0..49 are skipped despite satisfying a >= 50, and they are exactly the correct answer.
Instrumenting the rebuild branch confirms the state: kept=[3] pruned=2.
The filter column has to differ from the sort column for the same reason as in #24352: otherwise the partially-selected row group is also the one statistics prune first.
Suggested fix directions
- Slice the carried
RowSelection alongside the row groups on rebuild. This is only really correct inside arrow-rs, which owns both the queue and the selection; doing it in DataFusion means reimplementing split_off positioning outside the crate that holds the invariant.
- Better: give the decoder a
retain_row_groups(impl FnMut(usize) -> bool) that filters the frontier in place, so the selection and the queue cannot be updated independently. That would also remove the into_builder / is_at_row_group_boundary dance entirely.
- Failing that, decline to prune at runtime while a row selection is live, matching what
reorder_by_statistics already does.
Note on test coverage
dynamic_rg_pruning_coexists_with_page_index_row_selection claims to cover exactly this ("Tests that the into_builder rebuild preserves the RowSelection derived from page-index pruning across RG drops") and passes. Its filter column is its sort column, so ORDER BY v DESC prunes every remaining row group at once and the rebuild branch is never entered. Instrumenting that branch shows it executes zero times across all nine tests in dynamic_row_group_pruning.rs, on both main and #24354. cargo-mutants agrees: mutating pruned_count += 1 to pruned_count *= 1 disables the rebuild and the early exit entirely while leaving the metric increment intact, and the whole suite still passes.
cc @zhuqi-lucas
Describe the bug
With
datafusion.execution.parquet.pushdown_filters = trueand dynamic filter pushdown enabled (on by default), a query of the shapeSELECT b FROM t WHERE <predicate on a> ORDER BY b LIMIT kcan silently return wrong results: rows satisfying the predicate are dropped and replaced by later ones. No error, no warning.This is a second, independent instance of the failure family in #24352, and it is not fixed by #24354. Verified against
mainat0a429a37dband against #24354 at44f7fc71c5; both return the wrong answer.Root cause
The push decoder carries one flat
RowSelectioncovering the concatenation of the remaining row groups, in order.RowGroupFrontier::next_readable_row_groupconsumes it positionally (selection.split_off(row_count)per row group), so alignment between the selection and the row-group queue is a hard invariant.At a row-group boundary,
PushDecoderStreamState::transitionprunes row groups the dynamic predicate proves unwinnable and rebuilds the decoder:builder_from_remainingcorrectly carries the unconsumed selection into the rebuilt builder, butwith_row_groups(new_indices)then removes row groups from under it without slicing the selection to match. The selectors intended for a dropped row group are applied to the next surviving one.The plan-building path already knows about this coupling:
PreparedAccessPlan::reorder_by_statisticsrefuses to run when a row selection is present ("Skipping RG reorder: row_selection present") because remapping the selection is too complex. The runtime rebuild makes the same kind of remap-free change to the row-group set with no equivalent guard.To Reproduce
Deterministic, 400 rows, 4 row groups, 100% reproducible. In
datafusion-cli:format.write_batch_sizeis required:data_page_row_count_limitis only enforced at write-batch boundaries, so at the default of 1024 each 100-row row group becomes a single page and the page index cannot prune within it.Layout:
b(sort col)a(filter col)a >= 50selectsExpected behavior
Actual
Controls (same binary, same file, one setting changed):
enable_dynamic_filter_pushdownparquet.pushdown_filtersMechanism
a >= 50prunes the first five pages of RG 1 only. The decoder receives aRowSelectionof[RG0: select 100][RG1: skip 50, select 50][RG2: select 100][RG3: select 100].ORDER BY b ASC LIMIT 5tightens the TopK threshold to aboutb < 1004. At the boundary the runtime pruner proves RG 1 (min 2000) and RG 2 (min 3000) unwinnable and rebuilds withrow_groups = [3].skip 50, select 50. Rowsb = 0..49are skipped despite satisfyinga >= 50, and they are exactly the correct answer.Instrumenting the rebuild branch confirms the state:
kept=[3] pruned=2.The filter column has to differ from the sort column for the same reason as in #24352: otherwise the partially-selected row group is also the one statistics prune first.
Suggested fix directions
RowSelectionalongside the row groups on rebuild. This is only really correct inside arrow-rs, which owns both the queue and the selection; doing it in DataFusion means reimplementingsplit_offpositioning outside the crate that holds the invariant.retain_row_groups(impl FnMut(usize) -> bool)that filters the frontier in place, so the selection and the queue cannot be updated independently. That would also remove theinto_builder/is_at_row_group_boundarydance entirely.reorder_by_statisticsalready does.Note on test coverage
dynamic_rg_pruning_coexists_with_page_index_row_selectionclaims to cover exactly this ("Tests that theinto_builderrebuild preserves theRowSelectionderived from page-index pruning across RG drops") and passes. Its filter column is its sort column, soORDER BY v DESCprunes every remaining row group at once and the rebuild branch is never entered. Instrumenting that branch shows it executes zero times across all nine tests indynamic_row_group_pruning.rs, on bothmainand #24354.cargo-mutantsagrees: mutatingpruned_count += 1topruned_count *= 1disables the rebuild and the early exit entirely while leaving the metric increment intact, and the whole suite still passes.cc @zhuqi-lucas