hhhizzz commented on code in PR #24354:
URL: https://github.com/apache/datafusion/pull/24354#discussion_r3781212472
##########
datafusion/datasource-parquet/src/push_decoder.rs:
##########
@@ -342,6 +342,15 @@ impl PushDecoderStreamState {
.as_ref()
.expect("decoder present")
.is_at_row_group_boundary();
+ // Before pruning/rebuilding, align `rg_plan` with the row group
the
+ // decoder will actually emit next. arrow-rs silently finishes row
+ // groups whose post-predicate selection is empty without handing
back
+ // a reader, so without this sync `rg_plan` trails the decoder by
one
+ // and a later rebuild can re-read an already-delivered row group
+ // (#24352).
+ if at_boundary && let Err(e) =
self.sync_rg_plan_to_decoder_frontier() {
Review Comment:
Could we gate this synchronization on `self.row_group_pruner.is_some()`?
`peek_next_row_group()` is not O(1): in parquet 59.2.0 it clones the
remaining row-group indices and optional `RowSelection`, and documents a
per-call cost of `O(remaining row groups + selectors)`. With the current
`at_boundary`-only guard, ordinary Parquet scans also call it at every
row-group boundary, adding O(R²) cloning and allocation for a file with R row
groups, even though no decoder rebuild can occur.
The `rg_plan` desynchronization is only observable when the plan is later
used to rebuild the decoder, which requires `row_group_pruner` to be `Some`.
Could we use:
```rust
if at_boundary
&& self.row_group_pruner.is_some()
&& let Err(e) = self.sync_rg_plan_to_decoder_frontier()
{
return Some((Err(e), self));
}
```
This should preserve the #24352 fix while leaving ordinary Parquet scans
unchanged. A high-row-group-count benchmark would also help validate this path.
Arrow complexity reference:
https://github.com/apache/arrow-rs/blob/59.2.0/parquet/src/arrow/push_decoder/remaining.rs#L319-L334
##########
datafusion/sqllogictest/test_files/dynamic_row_group_pruning.slt:
##########
@@ -110,3 +110,70 @@ RESET datafusion.execution.parquet.pushdown_filters;
statement ok
RESET datafusion.explain.analyze_level;
+
+# Regression test for #24352: TopK dynamic filter + `pushdown_filters` must not
Review Comment:
Could this q26 regression explicitly enable both dynamic-filter settings and
assert its own `dynamic_rg_pruning=eligible` plan or non-zero
`row_groups_pruned_dynamic_filter` metric? It currently relies on defaults and
final output, so a future optimizer/default change could let the test pass
without exercising the prune/rebuild path that caused #24352.
--
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]
