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 d53522f5de7c..02408fdae17e 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 e528cf467ad9..b4706a4af872 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 376fb82a27b3..c470a23ed6f9 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 02408fdae17e..26a87d449061 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 b4706a4af872..faff96804096 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 84f38fc9d467a5ac91cd9287bfdf443a876a86d2 Mon Sep 17 00:00:00 2001 From: Kristofferjoha Date: Fri, 23 Oct 2026 17:29:38 +0200 Subject: [PATCH 3/4] fixed expr state bug --- datafusion/physical-plan/src/sorts/sort.rs | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/datafusion/physical-plan/src/sorts/sort.rs b/datafusion/physical-plan/src/sorts/sort.rs index 7c34545b7733..851360ae3648 100644 --- a/datafusion/physical-plan/src/sorts/sort.rs +++ b/datafusion/physical-plan/src/sorts/sort.rs @@ -1440,7 +1440,15 @@ impl ExecutionPlan for SortExec { .expect("rebuilt SortExec with new children") .clone(); // Our dynamic filter and execution metrics are the state we need to reset. - new_sort.filter = Some(new_sort.create_filter()); + let expr = new_sort + .filter + .as_ref() + .map(|filter| filter.read().expr()); + if let Some(expr) = expr { + expr.reset()?; + new_sort.filter = Some(new_sort.create_filter_with_expr(expr)); + } + new_sort.metrics_set = ExecutionPlanMetricsSet::new(); Ok(Arc::new(new_sort)) From afa570445d9e9f744e20207fceadee7a05dcaeda Mon Sep 17 00:00:00 2001 From: Kristofferjoha Date: Fri, 23 Oct 2026 17:43:39 +0200 Subject: [PATCH 4/4] cte.slt test --- datafusion/physical-plan/src/sorts/sort.rs | 5 +- datafusion/sqllogictest/test_files/cte.slt | 83 ++++++++++++++++++++++ 2 files changed, 84 insertions(+), 4 deletions(-) diff --git a/datafusion/physical-plan/src/sorts/sort.rs b/datafusion/physical-plan/src/sorts/sort.rs index 851360ae3648..f2d351b3c1a3 100644 --- a/datafusion/physical-plan/src/sorts/sort.rs +++ b/datafusion/physical-plan/src/sorts/sort.rs @@ -1440,10 +1440,7 @@ impl ExecutionPlan for SortExec { .expect("rebuilt SortExec with new children") .clone(); // Our dynamic filter and execution metrics are the state we need to reset. - let expr = new_sort - .filter - .as_ref() - .map(|filter| filter.read().expr()); + let expr = new_sort.filter.as_ref().map(|filter| filter.read().expr()); if let Some(expr) = expr { expr.reset()?; new_sort.filter = Some(new_sort.create_filter_with_expr(expr)); diff --git a/datafusion/sqllogictest/test_files/cte.slt b/datafusion/sqllogictest/test_files/cte.slt index c470a23ed6f9..589982dffefb 100644 --- a/datafusion/sqllogictest/test_files/cte.slt +++ b/datafusion/sqllogictest/test_files/cte.slt @@ -1758,3 +1758,86 @@ SELECT n, m FROM w; statement ok DROP TABLE dynamic_filter_recursive; + +# TopK must reset its dynamic filter in place so the Parquet scan keeps receiving +# updated bounds after each recursive iteration (#26642). The minimum is in the +# first row group, so each iteration should scan only 10 rows instead of 100. +statement ok +SET datafusion.execution.target_partitions = 1; + +statement ok +SET datafusion.execution.batch_size = 10; + +statement ok +SET datafusion.execution.parquet.pushdown_filters = true; + +statement ok +COPY ( + SELECT CASE WHEN value = 1 THEN 1 ELSE 102 - value END AS k + FROM generate_series(1, 100) +) TO 'test_files/scratch/cte/topk_dynamic_filter.parquet' +OPTIONS ('format.max_row_group_size' '10'); + +statement ok +CREATE EXTERNAL TABLE topk_dynamic_filter STORED AS PARQUET +LOCATION 'test_files/scratch/cte/topk_dynamic_filter.parquet'; + +query I +WITH RECURSIVE w(n) AS ( + SELECT 1 + UNION ALL + SELECT w.n + s.k + FROM w CROSS JOIN ( + SELECT k FROM topk_dynamic_filter ORDER BY k LIMIT 1 + ) s + WHERE w.n < 4 +) +SELECT * FROM w ORDER BY n; +---- +1 +2 +3 +4 + +statement ok +SET datafusion.explain.analyze_categories = 'rows'; + +query TT +EXPLAIN ANALYZE +WITH RECURSIVE w(n) AS ( + SELECT 1 + UNION ALL + SELECT w.n + s.k + FROM w CROSS JOIN ( + SELECT k FROM topk_dynamic_filter ORDER BY k LIMIT 1 + ) s + WHERE w.n < 4 +) +SELECT * FROM w ORDER BY n; +---- +Plan with Metrics +01)SortExec: expr=[n@0 ASC NULLS LAST], preserve_partitioning=[false], metrics=[output_rows=4, output_batches=1, spill_count=0, spilled_rows=0] +02)--RecursiveQueryExec: name=w, is_distinct=false, metrics=[output_rows=4, output_batches=4] +03)----ProjectionExec: expr=[CAST(1 AS Int64) as n], metrics=[output_rows=1, output_batches=1] +04)------PlaceholderRowExec, metrics=[] +05)----ProjectionExec: expr=[n@0 + k@1 as n], metrics=[] +06)------CrossJoinExec, metrics=[] +07)--------FilterExec: n@0 < 4, metrics=[] +08)----------WorkTableExec: name=w, metrics=[] +09)--------SortExec: TopK(fetch=1), expr=[k@0 ASC NULLS LAST], preserve_partitioning=[false], metrics=[] +10)----------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/cte/topk_dynamic_filter.parquet]]}, projection=[k], file_type=parquet, predicate=DynamicFilter [ true ], sort_order_for_reorder=[k@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, metrics=[output_rows=30, output_batches=3, files_ranges_pruned_statistics=3 total → 3 matched, row_groups_pruned_statistics=30 total → 30 matched, row_groups_pruned_bloom_filter=30 total → 30 matched, page_index_pages_pruned=0 total → 0 matched, page_index_rows_pruned=0 total → 0 matched, limit_pruned_row_groups=0 total → 0 matched, batches_split=0, file_open_errors=0, file_scan_errors=0, files_opened=3, files_processed=3, num_predicate_creation_errors=0, predicate_evaluation_errors=0, pushdown_rows_matched=30, pushdown_rows_pruned=0, predicate_cache_inner_records=30, predicate_cache_records=30, scan_efficiency_ratio=9.9% (279/2.82 K)] + +statement ok +DROP TABLE topk_dynamic_filter; + +statement ok +RESET datafusion.explain.analyze_categories; + +statement ok +SET datafusion.execution.target_partitions = 4; + +statement ok +RESET datafusion.execution.batch_size; + +statement ok +RESET datafusion.execution.parquet.pushdown_filters;