From 0de6c24c8aa1d3d75d8ce55d0f156814e1ae06d8 Mon Sep 17 00:00:00 2001 From: mac Date: Mon, 5 Oct 2026 07:46:08 -0700 Subject: [PATCH 1/4] fix: reset aggregate dynamic filters between recursive iterations --- .../src/expressions/dynamic_filters/mod.rs | 30 ++++++++++++++++++ .../physical-plan/src/aggregates/mod.rs | 13 ++++++++ datafusion/sqllogictest/test_files/cte.slt | 31 +++++++++++++++++++ 3 files changed, 74 insertions(+) diff --git a/datafusion/physical-expr/src/expressions/dynamic_filters/mod.rs b/datafusion/physical-expr/src/expressions/dynamic_filters/mod.rs index d53522f5de7c8..02408fdae17ec 100644 --- a/datafusion/physical-expr/src/expressions/dynamic_filters/mod.rs +++ b/datafusion/physical-expr/src/expressions/dynamic_filters/mod.rs @@ -29,6 +29,7 @@ use datafusion_common::{ tree_node::{Transformed, TransformedResult, TreeNode}, }; +use crate::expressions::lit; use datafusion_expr::ColumnarValue; use datafusion_expr_common::dyn_eq::DynHash; @@ -343,6 +344,20 @@ impl DynamicFilterPhysicalExpr { Ok(()) } + /// Reset this dynamic filter so its producer can publish a new expression + /// during a later execution of the same plan. + pub fn reset(&self) -> Result<()> { + self.update(lit(true))?; + let mut current = self.inner.write(); + current.is_complete = false; + let generation = current.generation; + drop(current); + let _ = self + .state_watch + .send(FilterState::InProgress { generation }); + Ok(()) + } + /// Mark this dynamic filter as complete and broadcast to all waiters. /// /// This signals that all expected updates have been received. @@ -1032,6 +1047,21 @@ mod test { dynamic_filter.wait_complete().await; } + #[test] + fn test_reset_reopens_completed_filter() { + let dynamic_filter = + DynamicFilterPhysicalExpr::new(vec![], lit(false) as Arc); + dynamic_filter.mark_complete(); + let generation_before_reset = dynamic_filter.inner.read().generation; + + dynamic_filter.reset().unwrap(); + + let inner = dynamic_filter.inner.read(); + assert!(!inner.is_complete); + assert!(inner.generation > generation_before_reset); + assert_eq!(inner.expr.to_string(), "true"); + } + #[test] fn test_with_new_children_independence() { // Create a schema with columns a, b, c, d diff --git a/datafusion/physical-plan/src/aggregates/mod.rs b/datafusion/physical-plan/src/aggregates/mod.rs index e528cf467ad9c..b4706a4af872a 100644 --- a/datafusion/physical-plan/src/aggregates/mod.rs +++ b/datafusion/physical-plan/src/aggregates/mod.rs @@ -2617,6 +2617,19 @@ impl ExecutionPlan for AggregateExec { ) } + fn reset_state(self: Arc) -> Result> { + if let Some(dynamic_filter) = &self.dynamic_filter { + dynamic_filter.filter.reset()?; + for accumulator_filter in &dynamic_filter.accumulator_dyn_filter_info { + *accumulator_filter.shared_bound.lock() = ScalarValue::Null; + } + } + + let mut aggregate = (*self).clone(); + aggregate.metrics = ExecutionPlanMetricsSet::new(); + Ok(Arc::new(aggregate)) + } + fn execute( &self, partition: usize, diff --git a/datafusion/sqllogictest/test_files/cte.slt b/datafusion/sqllogictest/test_files/cte.slt index 376fb82a27b3d..c470a23ed6f98 100644 --- a/datafusion/sqllogictest/test_files/cte.slt +++ b/datafusion/sqllogictest/test_files/cte.slt @@ -1727,3 +1727,34 @@ order by 1, 2; statement ok drop table test; + +# Dynamic filters produced by a recursive term must be reset between +# iterations. Otherwise the second execution of this uncorrelated aggregate +# reuses the first iteration's MAX bound and scans no rows. +statement ok +COPY (SELECT CAST(value AS DOUBLE) AS r FROM generate_series(0, 128)) TO 'test_files/scratch/cte/dynamic_filter_recursive.parquet' STORED AS PARQUET; + +statement ok +CREATE EXTERNAL TABLE dynamic_filter_recursive STORED AS PARQUET LOCATION 'test_files/scratch/cte/dynamic_filter_recursive.parquet'; + +query IR rowsort +WITH RECURSIVE w(n, m) AS ( + SELECT 0, CAST(NULL AS DOUBLE) + UNION ALL + SELECT w.n + 1, x.m + FROM w CROSS JOIN ( + SELECT MAX(r) AS m + FROM dynamic_filter_recursive + WHERE r % 2 = 0 + ) x + WHERE w.n < 3 +) +SELECT n, m FROM w; +---- +0 NULL +1 128 +2 128 +3 128 + +statement ok +DROP TABLE dynamic_filter_recursive; From 8a5659987994c7cf8b9284e4d58c6c0b53a8065f Mon Sep 17 00:00:00 2001 From: mika Date: Tue, 20 Oct 2026 22:21:04 -0700 Subject: [PATCH 2/4] fix(expr): publish dynamic filter reset in a single step and test shared_bound reset --- .../src/expressions/dynamic_filters/mod.rs | 23 +++++--- .../physical-plan/src/aggregates/mod.rs | 54 +++++++++++++++++++ 2 files changed, 71 insertions(+), 6 deletions(-) diff --git a/datafusion/physical-expr/src/expressions/dynamic_filters/mod.rs b/datafusion/physical-expr/src/expressions/dynamic_filters/mod.rs index 02408fdae17ec..26a87d449061a 100644 --- a/datafusion/physical-expr/src/expressions/dynamic_filters/mod.rs +++ b/datafusion/physical-expr/src/expressions/dynamic_filters/mod.rs @@ -347,14 +347,25 @@ impl DynamicFilterPhysicalExpr { /// Reset this dynamic filter so its producer can publish a new expression /// during a later execution of the same plan. pub fn reset(&self) -> Result<()> { - self.update(lit(true))?; + let new_expr = Self::remap_children( + &self.children, + self.remapped_children.as_ref(), + lit(true), + )?; + let mut current = self.inner.write(); - current.is_complete = false; - let generation = current.generation; + let new_generation = current.generation + 1; + *current = Inner { + expression_id: current.expression_id, + generation: new_generation, + expr: new_expr, + is_complete: false, + }; drop(current); - let _ = self - .state_watch - .send(FilterState::InProgress { generation }); + + let _ = self.state_watch.send(FilterState::InProgress { + generation: new_generation, + }); Ok(()) } diff --git a/datafusion/physical-plan/src/aggregates/mod.rs b/datafusion/physical-plan/src/aggregates/mod.rs index b4706a4af872a..faff968040965 100644 --- a/datafusion/physical-plan/src/aggregates/mod.rs +++ b/datafusion/physical-plan/src/aggregates/mod.rs @@ -9827,4 +9827,58 @@ mod tests { assert!(agg.set_dynamic_filter(df).is_err()); Ok(()) } + + #[test] + fn test_aggregate_exec_reset_state() -> Result<()> { + let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int64, false)])); + let child = Arc::new(EmptyExec::new(Arc::clone(&schema))); + + let agg = AggregateExec::try_new( + AggregateMode::Partial, + PhysicalGroupBy::new_single(vec![]), + vec![Arc::new( + AggregateExprBuilder::new(min_udaf(), vec![col("a", &schema)?]) + .schema(Arc::clone(&schema)) + .alias("min_a") + .build()?, + )], + vec![None], + child, + Arc::clone(&schema), + )?; + + let df = Arc::new(DynamicFilterPhysicalExpr::new( + vec![col("a", &schema)?], + lit(false), + )); + let agg = Arc::new(agg.set_dynamic_filter(Arc::clone(&df))?); + + // Simulate an iteration setting a shared bound and completing the filter + if let Some(dyn_filter) = &agg.dynamic_filter { + for info in &dyn_filter.accumulator_dyn_filter_info { + *info.shared_bound.lock() = ScalarValue::Int64(Some(42)); + } + dyn_filter.filter.mark_complete(); + } + + // Call reset_state + let reset_plan = agg.reset_state()?; + let reset_agg = reset_plan + .downcast_ref::() + .expect("should downcast back to AggregateExec"); + + let dyn_filter = reset_agg + .dynamic_filter + .as_ref() + .expect("dynamic filter should be retained"); + + for info in &dyn_filter.accumulator_dyn_filter_info { + assert_eq!(*info.shared_bound.lock(), ScalarValue::Null); + } + + let current_expr = dyn_filter.filter.current()?; + assert_eq!(format!("{current_expr}"), format!("{}", lit(true))); + + Ok(()) + } } From 6cb6513589c79713a9c6da0553703f359bf325da Mon Sep 17 00:00:00 2001 From: Kristofferjoha Date: Fri, 23 Oct 2026 18:16:46 +0200 Subject: [PATCH 3/4] reset existing shared filter --- .../physical-plan/src/joins/hash_join/exec.rs | 14 +++- .../joins/hash_join/exec/prepared/tests.rs | 4 +- datafusion/sqllogictest/test_files/cte.slt | 73 +++++++++++++++++++ 3 files changed, 85 insertions(+), 6 deletions(-) diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs index 10a9fa70a8f76..d74f6567ef03d 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec.rs +++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs @@ -566,11 +566,17 @@ impl HashJoinExecBuilder { } /// Reset task-local runtime state while retaining the immutable prepared build. - pub fn reset_state(mut self) -> Self { + pub fn reset_state(mut self) -> Result { self.exec.left_fut = Default::default(); - self.exec.dynamic_filter = None; + + if let Some(dynamic_filter) = &mut self.exec.dynamic_filter { + dynamic_filter.filter.reset()?; + dynamic_filter.build_accumulator = OnceLock::new(); + } + self.exec.metrics = ExecutionPlanMetricsSet::new(); - self + + Ok(self) } /// Build result as a dyn execution plan. @@ -1687,7 +1693,7 @@ impl ExecutionPlan for HashJoinExec { } fn reset_state(self: Arc) -> Result> { - self.builder().reset_state().build_exec() + self.builder().reset_state()?.build_exec() } fn execute( diff --git a/datafusion/physical-plan/src/joins/hash_join/exec/prepared/tests.rs b/datafusion/physical-plan/src/joins/hash_join/exec/prepared/tests.rs index 48281e5f481b1..3500b282b6500 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec/prepared/tests.rs +++ b/datafusion/physical-plan/src/joins/hash_join/exec/prepared/tests.rs @@ -328,7 +328,7 @@ async fn prepared_build_reuses_data_with_independent_dynamic_filters() -> Result .build()?; let ordinary = task .builder() - .reset_state() + .reset_state()? .with_new_children(vec![ TestMemoryExec::try_new_exec( &[vec![build.clone()]], @@ -666,7 +666,7 @@ async fn prepared_build_matches_collect_left() -> Result<()> { .with_batch_size(batch_size), ), ); - let ordinary = base.builder().reset_state().build()?; + let ordinary = base.builder().reset_state()?.build()?; let attached = base .builder() .with_prepared_build(Arc::clone(&prepared)) diff --git a/datafusion/sqllogictest/test_files/cte.slt b/datafusion/sqllogictest/test_files/cte.slt index c470a23ed6f98..df856edf905e0 100644 --- a/datafusion/sqllogictest/test_files/cte.slt +++ b/datafusion/sqllogictest/test_files/cte.slt @@ -1758,3 +1758,76 @@ SELECT n, m FROM w; statement ok DROP TABLE dynamic_filter_recursive; + +# Hash join dynamic filters must remain shared with the probe scan across +# recursive iterations (#26641). Each iteration looks up a different key, so +# retaining a stale bound would also truncate the recursive result. +statement ok +SET datafusion.execution.target_partitions = 1; + +statement ok +SET datafusion.execution.parquet.pushdown_filters = true; + +statement ok +COPY (SELECT value AS k FROM generate_series(1, 100)) +TO 'test_files/scratch/cte/hash_join_dynamic_filter.parquet' STORED AS PARQUET; + +statement ok +CREATE EXTERNAL TABLE hash_join_dynamic_filter STORED AS PARQUET +LOCATION 'test_files/scratch/cte/hash_join_dynamic_filter.parquet'; + +query I +WITH RECURSIVE w(n) AS ( + SELECT 1 + UNION ALL + SELECT t.k + FROM w JOIN hash_join_dynamic_filter t ON t.k = w.n + 1 + WHERE w.n < 5 +) +SELECT * FROM w ORDER BY n; +---- +1 +2 +3 +4 +5 + +statement ok +SET datafusion.explain.analyze_categories = 'rows'; + +# The probe scan should output four rows and prune the other 396, rather than +# outputting all 400 rows across the four iterations. +query TT +EXPLAIN ANALYZE +WITH RECURSIVE w(n) AS ( + SELECT 1 + UNION ALL + SELECT t.k + FROM w JOIN hash_join_dynamic_filter t ON t.k = w.n + 1 + WHERE w.n < 5 +) +SELECT * FROM w ORDER BY n; +---- +Plan with Metrics +01)SortExec: expr=[n@0 ASC NULLS LAST], preserve_partitioning=[false], metrics=[output_rows=5, output_batches=1, spill_count=0, spilled_rows=0] +02)--RecursiveQueryExec: name=w, is_distinct=false, metrics=[output_rows=5, output_batches=5] +03)----ProjectionExec: expr=[1 as n], metrics=[output_rows=1, output_batches=1] +04)------PlaceholderRowExec, metrics=[] +05)----ProjectionExec: expr=[k@0 as n], metrics=[] +06)------HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(w.n + Int64(1)@1, k@0)], projection=[k@2], metrics=[] +07)--------ProjectionExec: expr=[n@0 as n, n@0 + 1 as w.n + Int64(1)], metrics=[] +08)----------FilterExec: n@0 < 5, metrics=[] +09)------------WorkTableExec: name=w, metrics=[] +10)--------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/cte/hash_join_dynamic_filter.parquet]]}, projection=[k], output_ordering=[k@0 ASC NULLS LAST], file_type=parquet, predicate=DynamicFilter [ true ], dynamic_rg_pruning=eligible, metrics=[output_rows=4, output_batches=4, files_ranges_pruned_statistics=4 total → 4 matched, row_groups_pruned_statistics=4 total → 4 matched, row_groups_pruned_bloom_filter=4 total → 4 matched, page_index_pages_pruned=4 total → 4 matched, page_index_rows_pruned=400 total → 400 matched, limit_pruned_row_groups=0 total → 0 matched, batches_split=0, file_open_errors=0, file_scan_errors=0, files_opened=4, files_processed=4, num_predicate_creation_errors=0, predicate_evaluation_errors=0, pushdown_rows_matched=4, pushdown_rows_pruned=396, predicate_cache_inner_records=400, predicate_cache_records=4, scan_efficiency_ratio=157.36% (1.14 K/727)] + +statement ok +DROP TABLE hash_join_dynamic_filter; + +statement ok +RESET datafusion.explain.analyze_categories; + +statement ok +SET datafusion.execution.target_partitions = 4; + +statement ok +RESET datafusion.execution.parquet.pushdown_filters; From 07091523affe61a0a6acf27cfbcc8626e0853f25 Mon Sep 17 00:00:00 2001 From: Kristofferjoha Date: Sat, 24 Oct 2026 15:07:55 +0200 Subject: [PATCH 4/4] remove unnecessary fallible operation --- .../src/expressions/dynamic_filters/mod.rs | 13 +++---------- datafusion/physical-plan/src/aggregates/mod.rs | 2 +- datafusion/physical-plan/src/execution_plan.rs | 4 +++- .../physical-plan/src/joins/hash_join/exec.rs | 8 ++++---- .../src/joins/hash_join/exec/prepared/tests.rs | 4 ++-- datafusion/sqllogictest/test_files/cte.slt | 5 +++-- 6 files changed, 16 insertions(+), 20 deletions(-) diff --git a/datafusion/physical-expr/src/expressions/dynamic_filters/mod.rs b/datafusion/physical-expr/src/expressions/dynamic_filters/mod.rs index 26a87d449061a..0556bf8a4f918 100644 --- a/datafusion/physical-expr/src/expressions/dynamic_filters/mod.rs +++ b/datafusion/physical-expr/src/expressions/dynamic_filters/mod.rs @@ -346,19 +346,13 @@ impl DynamicFilterPhysicalExpr { /// Reset this dynamic filter so its producer can publish a new expression /// during a later execution of the same plan. - pub fn reset(&self) -> Result<()> { - let new_expr = Self::remap_children( - &self.children, - self.remapped_children.as_ref(), - lit(true), - )?; - + pub fn reset(&self) { let mut current = self.inner.write(); let new_generation = current.generation + 1; *current = Inner { expression_id: current.expression_id, generation: new_generation, - expr: new_expr, + expr: lit(true), is_complete: false, }; drop(current); @@ -366,7 +360,6 @@ impl DynamicFilterPhysicalExpr { let _ = self.state_watch.send(FilterState::InProgress { generation: new_generation, }); - Ok(()) } /// Mark this dynamic filter as complete and broadcast to all waiters. @@ -1065,7 +1058,7 @@ mod test { dynamic_filter.mark_complete(); let generation_before_reset = dynamic_filter.inner.read().generation; - dynamic_filter.reset().unwrap(); + dynamic_filter.reset(); let inner = dynamic_filter.inner.read(); assert!(!inner.is_complete); diff --git a/datafusion/physical-plan/src/aggregates/mod.rs b/datafusion/physical-plan/src/aggregates/mod.rs index faff968040965..74c37716c68e5 100644 --- a/datafusion/physical-plan/src/aggregates/mod.rs +++ b/datafusion/physical-plan/src/aggregates/mod.rs @@ -2619,7 +2619,7 @@ impl ExecutionPlan for AggregateExec { fn reset_state(self: Arc) -> Result> { if let Some(dynamic_filter) = &self.dynamic_filter { - dynamic_filter.filter.reset()?; + dynamic_filter.filter.reset(); for accumulator_filter in &dynamic_filter.accumulator_dyn_filter_info { *accumulator_filter.shared_bound.lock() = ScalarValue::Null; } diff --git a/datafusion/physical-plan/src/execution_plan.rs b/datafusion/physical-plan/src/execution_plan.rs index 2f5680e659e0d..79af31b0e0cbb 100644 --- a/datafusion/physical-plan/src/execution_plan.rs +++ b/datafusion/physical-plan/src/execution_plan.rs @@ -1996,7 +1996,9 @@ pub fn check_not_null_constraints( /// /// While this function enables plan reuse, it does not allow the same plan to be executed if it (OR): /// -/// * uses dynamic filters, +/// * uses dynamic filters and the original plan (or another reset copy) is still +/// executing, because operators such as `HashJoinExec` and `AggregateExec` +/// reset their shared dynamic filter in place, /// * represents a recursive query. /// pub fn reset_plan_states(plan: Arc) -> Result> { diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs index d74f6567ef03d..7093fa2a90117 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec.rs +++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs @@ -566,17 +566,17 @@ impl HashJoinExecBuilder { } /// Reset task-local runtime state while retaining the immutable prepared build. - pub fn reset_state(mut self) -> Result { + pub fn reset_state(mut self) -> Self { self.exec.left_fut = Default::default(); if let Some(dynamic_filter) = &mut self.exec.dynamic_filter { - dynamic_filter.filter.reset()?; + dynamic_filter.filter.reset(); dynamic_filter.build_accumulator = OnceLock::new(); } self.exec.metrics = ExecutionPlanMetricsSet::new(); - Ok(self) + self } /// Build result as a dyn execution plan. @@ -1693,7 +1693,7 @@ impl ExecutionPlan for HashJoinExec { } fn reset_state(self: Arc) -> Result> { - self.builder().reset_state()?.build_exec() + self.builder().reset_state().build_exec() } fn execute( diff --git a/datafusion/physical-plan/src/joins/hash_join/exec/prepared/tests.rs b/datafusion/physical-plan/src/joins/hash_join/exec/prepared/tests.rs index 3500b282b6500..48281e5f481b1 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec/prepared/tests.rs +++ b/datafusion/physical-plan/src/joins/hash_join/exec/prepared/tests.rs @@ -328,7 +328,7 @@ async fn prepared_build_reuses_data_with_independent_dynamic_filters() -> Result .build()?; let ordinary = task .builder() - .reset_state()? + .reset_state() .with_new_children(vec![ TestMemoryExec::try_new_exec( &[vec![build.clone()]], @@ -666,7 +666,7 @@ async fn prepared_build_matches_collect_left() -> Result<()> { .with_batch_size(batch_size), ), ); - let ordinary = base.builder().reset_state()?.build()?; + let ordinary = base.builder().reset_state().build()?; let attached = base .builder() .with_prepared_build(Arc::clone(&prepared)) diff --git a/datafusion/sqllogictest/test_files/cte.slt b/datafusion/sqllogictest/test_files/cte.slt index df856edf905e0..36308c8fabb07 100644 --- a/datafusion/sqllogictest/test_files/cte.slt +++ b/datafusion/sqllogictest/test_files/cte.slt @@ -1760,8 +1760,9 @@ statement ok DROP TABLE dynamic_filter_recursive; # Hash join dynamic filters must remain shared with the probe scan across -# recursive iterations (#26641). Each iteration looks up a different key, so -# retaining a stale bound would also truncate the recursive result. +# recursive iterations (#26641). Each iteration looks up a different key; the +# EXPLAIN ANALYZE below checks that the filter is reset and republished, so the +# probe scan prunes rows on every iteration. statement ok SET datafusion.execution.target_partitions = 1;