adriangb opened a new issue, #24355:
URL: https://github.com/apache/datafusion/issues/24355
### 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:
```rust
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`:
```sql
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
```text
0 1 2 3 4
```
### Actual
```text
50 51 52 53 54
```
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
1. 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]`.
2. RG 0 is read first and its 100 selectors are consumed. The remaining
selection covers RG 1, 2, 3.
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]`.
4. 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
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
