hhhizzz opened a new issue, #24352:
URL: https://github.com/apache/datafusion/issues/24352
### Describe the bug
With `datafusion.execution.parquet.pushdown_filters = true` and TopK dynamic
filter pushdown enabled (both dynamic-filter switches are **on by
default**), a
query of the shape
```sql
SELECT b FROM t WHERE <predicate on a> ORDER BY <b> LIMIT k
```
can silently return **wrong results**: one source row is emitted several
times
and the true tail of the top-k is missing. No error, no warning.
Out of the box DataFusion is unaffected because `pushdown_filters` defaults
to
`false`, but any user who enables that documented performance option is
exposed, and the returned rows themselves are well-formed — only the *set* is
wrong — so the corruption is hard to notice.
The root cause is a bookkeeping desync in
`datafusion/datasource-parquet/src/push_decoder.rs`, amplified by the runtime
row-group prune/rebuild path added in #22450.
### To Reproduce
Deterministic, 8192 rows, 4 row groups, ~72 KB, runs in about a second, 100%
reproducible (verified 5/5 on an unmodified `datafusion-cli`).
Create the fixture (DuckDB is used only as an independent writer and oracle;
2048 is its minimum row-group size):
```sql
COPY (
SELECT
CASE
WHEN i < 2048 THEN i * 1000 -- F
WHEN i < 4096 THEN (CASE WHEN i = 2048 THEN 50 ELSE 20000 + i END) -- E
WHEN i < 6144 THEN 100 + (i - 4096) -- N
ELSE 5000 + (i - 6144) -- P
END::BIGINT AS event_time,
CASE WHEN i = 2048 THEN '' ELSE 'p' || i::VARCHAR END AS search_phrase
FROM range(8192) AS t(i)
)
TO 'q26.parquet' (FORMAT PARQUET, ROW_GROUP_SIZE 2048);
```
Then in `datafusion-cli`:
```sql
SET datafusion.execution.target_partitions = 1;
SET datafusion.execution.parquet.pushdown_filters = true;
SET datafusion.optimizer.enable_dynamic_filter_pushdown = true;
SET datafusion.optimizer.enable_topk_dynamic_filter_pushdown = true;
CREATE EXTERNAL TABLE hits (
event_time BIGINT NOT NULL,
search_phrase VARCHAR NOT NULL
) STORED AS PARQUET LOCATION 'q26.parquet';
SELECT search_phrase FROM hits
WHERE search_phrase <> ''
ORDER BY event_time
LIMIT 10;
```
Expected (and what DuckDB, and DataFusion with either switch off, return):
```text
p0 p4096 p4097 p4098 p4099 p4100 p4101 p4102 p4103 p4104
```
Actual:
```text
p0 p4096 p4096 p4097 p4097 p4098 p4098 p4099 p4099 p4100
```
`search_phrase` is unique per row, so the repeats are the *same source row*
emitted more than once; `p4101`–`p4104` are lost.
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 |
Deterministic, not a race: identical wrong output at `target_partitions` = 1,
4 and 8, and 5/5 identical across repeated runs.
### Expected behavior
_No response_
### Additional context
### Mechanism
Three things have to line up, and the fixture arranges them in order:
1. **A row group whose post-predicate selection is empty, in a way statistics
cannot see.** In RG `E` the only small `event_time` (50) sits on a row
whose
`search_phrase` is `''`. Row-group statistics therefore say "keep"
(min 50 < current threshold), but after applying `search_phrase <> ''` and
the dynamic `event_time < threshold`, *no rows survive*.
arrow-rs finishes such a row group without handing back a reader —
`!plan_builder.selects_any()` → `RowGroupDecoderState::Finished` in
`parquet/src/arrow/push_decoder/reader_builder/mod.rs` (there is a second,
equivalent early-finish for the limit/offset case in the same file).
2. **`rg_plan` desyncs.** `PushDecoderStreamState::transition` maintains its
own
`VecDeque` of pending row groups and pops it **only** when the decoder
hands
back a reader (`push_decoder.rs`, in the `DecodeResult::Data(reader)`
arm),
with the comment "`rg_plan.front()` is the RG the decoder is about to
read".
That invariant does not hold for silently-finished row groups: RG `E`
consumes no pop, so from then on every pop removes the *previous* entry
and
`rg_plan` trails the decoder by one.
3. **The runtime prune path rebuilds from the stale plan.** When the dynamic
threshold makes RG `P` prunable, the boundary handler rebuilds the decoder
with `decoder.into_builder().with_row_groups(new_indices)` where
`new_indices` comes from `rg_plan` — which still lists RG `N`, *already
delivered*. `N` is read again and its rows are emitted a second time.
`TopK` does not de-duplicate (nor should it have to), so the repeated rows
take
heap slots and evict the genuine tail.
Instrumented trace of the fixture (three added `eprintln!`s: initial plan,
each
pop, each rebuild; plus first/last `event_time` of every batch reaching
TopK):
```text
Q26_PLAN0 head=[0, 1, 2, 3] len=4
Q26_POP rg=Some(0)
Q26_BATCH n=0 rows=2048 first=0 last=2047000 <- RG F, threshold :=
9000
Q26_POP rg=Some(1)
Q26_BATCH n=1 rows=2048 first=100 last=2147 <- RG N's data, popped
E's entry
Q26_REBUILD pruned=1 new_head=[2] new_len=1 <- P pruned; plan still
lists N
Q26_POP rg=Some(2)
Q26_BATCH n=2 rows=8 first=100 last=107 <- N re-delivered
```
Causal check: gating only the step-2 prune/rebuild block behind an env var
(same binary, same data, nothing else changed) restores byte-correct output.
This also explains why the filter column has to differ from the sort column:
if
they are the same, every row group that would produce an empty selection also
has `min >= threshold` and is removed by statistics pruning first, so no
silent skip — and no desync — can occur.
### Environment
- Upstream `main` at `0a429a37db`, `datafusion-cli 54.1.0`, arrow/parquet
`59.2.0` from crates.io, no patches — clean checkout, default build.
- macOS/arm64 for the fixture above; the original finding was on Linux/AMD64
against ClickBench `hits.parquet`, so both platforms are affected.
- Reproduces at `target_partitions` = 1, 4 and 8.
- Relevant upstream work: runtime row-group pruning and decoder rebuild
(#22450), row-group statistics reordering for TopK (#21956, #23888).
### Suggested fix directions
Either side can restore the invariant:
- **arrow-rs**: let the push decoder report row groups it finished without
producing a reader (or tag the returned reader with its row-group index) so
callers can keep their own bookkeeping in sync.
- **DataFusion**: stop maintaining a parallel `rg_plan`; ask the decoder for
the
set of row groups still outstanding when rebuilding, so a rebuild can never
re-include an already-delivered group.
A defensive assertion that a rebuild never re-includes a delivered row group
would have caught this at the boundary.
--
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]
