Skip to content

Wrong results: dynamic row-group pruning drops row groups without slicing the carried RowSelection #24355

Description

@adriangb

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

0 1 2 3 4

Actual

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

Metadata

Metadata

Assignees

Labels

bugSomething isn't working

Type

No type

Projects

No projects

Milestone

No milestone

Relationships

None yet

Development

No branches or pull requests

Issue actions