Skip to content

Commit 44f7fc7

Browse files
committed
Address review (@hhhizzz): gate sync on pruner + assert prune path in tests
- push_decoder: gate sync_rg_plan_to_decoder_frontier on row_group_pruner.is_some(). Only the runtime pruner rebuilds from rg_plan, and this avoids the O(remaining row groups) peek_next_row_group() cost on ordinary scans that never rebuild. - rust test: assert row_groups_pruned_dynamic_filter >= 1 so the test provably exercises the prune/rebuild path (#24352), not just the final output. - slt: enable both dynamic-filter switches explicitly instead of relying on defaults.
1 parent 5617063 commit 44f7fc7

3 files changed

Lines changed: 40 additions & 7 deletions

File tree

‎datafusion/core/tests/parquet/dynamic_row_group_pruning.rs‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -667,6 +667,20 @@ async fn topk_pushdown_does_not_reread_delivered_row_group() {
667667
// source row emitted twice. The correct answer is the 10 smallest-
668668
// `event_time` non-empty phrases, matching DuckDB / pushdown-off.
669669
assert_eq!(output.result_rows, 10, "{}", output.description());
670+
671+
// The test must actually exercise the runtime prune/rebuild path that
672+
// caused #24352 (not just a happy-path scan), otherwise a future default or
673+
// optimizer change could let it pass without the bug's precondition. Assert
674+
// the dynamic filter pruned at least one row group.
675+
let pruned = output
676+
.row_groups_pruned_dynamic_filter()
677+
.expect("`row_groups_pruned_dynamic_filter` metric must be registered");
678+
assert!(
679+
pruned >= 1,
680+
"test must exercise dynamic RG pruning (the #24352 path); pruned={pruned}\n{}",
681+
output.description(),
682+
);
683+
670684
let formatted = output.pretty_results();
671685
for p in [
672686
"p0", "p4096", "p4097", "p4098", "p4099", "p4100", "p4101", "p4102", "p4103",

‎datafusion/datasource-parquet/src/push_decoder.rs‎

Lines changed: 12 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -342,13 +342,18 @@ impl PushDecoderStreamState {
342342
.as_ref()
343343
.expect("decoder present")
344344
.is_at_row_group_boundary();
345-
// Before pruning/rebuilding, align `rg_plan` with the row group the
346-
// decoder will actually emit next. arrow-rs silently finishes row
347-
// groups whose post-predicate selection is empty without handing back
348-
// a reader, so without this sync `rg_plan` trails the decoder by one
349-
// and a later rebuild can re-read an already-delivered row group
350-
// (#24352).
351-
if at_boundary && let Err(e) = self.sync_rg_plan_to_decoder_frontier() {
345+
// Only the runtime pruner rebuilds the decoder from `rg_plan`, so
346+
// only it needs `rg_plan` kept in sync with the decoder frontier.
347+
// arrow-rs silently finishes row groups whose post-predicate
348+
// selection is empty without handing back a reader, so without this
349+
// sync `rg_plan` trails the decoder by one and a rebuild re-reads an
350+
// already-delivered row group (#24352). Gating on the pruner also
351+
// avoids the O(remaining row groups) cost of `peek_next_row_group()`
352+
// on ordinary scans that never rebuild.
353+
if at_boundary
354+
&& self.row_group_pruner.is_some()
355+
&& let Err(e) = self.sync_rg_plan_to_decoder_frontier()
356+
{
352357
return Some((Err(e), self));
353358
}
354359
if at_boundary && !self.rg_plan.is_empty() {

‎datafusion/sqllogictest/test_files/dynamic_row_group_pruning.slt‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -126,6 +126,14 @@ set datafusion.execution.parquet.pushdown_filters = true;
126126
statement ok
127127
set datafusion.execution.target_partitions = 1;
128128

129+
# Both dynamic-filter switches are on by default; set them explicitly so this
130+
# test keeps exercising the prune/rebuild path even if the defaults change.
131+
statement ok
132+
set datafusion.optimizer.enable_dynamic_filter_pushdown = true;
133+
134+
statement ok
135+
set datafusion.optimizer.enable_topk_dynamic_filter_pushdown = true;
136+
129137
statement ok
130138
CREATE TABLE q26_src AS
131139
SELECT
@@ -177,3 +185,9 @@ RESET datafusion.execution.parquet.pushdown_filters;
177185

178186
statement ok
179187
RESET datafusion.execution.target_partitions;
188+
189+
statement ok
190+
RESET datafusion.optimizer.enable_dynamic_filter_pushdown;
191+
192+
statement ok
193+
RESET datafusion.optimizer.enable_topk_dynamic_filter_pushdown;

0 commit comments

Comments
 (0)