fix: preserve TopK dynamic filters across recursive CTE iterations

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,31 @@ 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<()> {
+        let new_expr = Self::remap_children(
+            &self.children,
+            self.remapped_children.as_ref(),
+            lit(true),
+        )?;
+
+        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,
+            is_complete: false,
+        };
+        drop(current);
+
+        let _ = self.state_watch.send(FilterState::InProgress {
+            generation: new_generation,
+        });
+        Ok(())
+    }
+
     /// Mark this dynamic filter as complete and broadcast to all waiters.
     ///
     /// This signals that all expected updates have been received.
@@ -1032,6 +1058,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<dyn PhysicalExpr>);
+        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

datafusion/physical-plan/src/aggregates/mod.rs

@@ -2617,6 +2617,19 @@ impl ExecutionPlan for AggregateExec {
         )
     }
 
+    fn reset_state(self: Arc<Self>) -> Result<Arc<dyn ExecutionPlan>> {
+        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,
@@ -9814,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::<AggregateExec>()
+            .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(())
+    }
 }

datafusion/physical-plan/src/sorts/sort.rs

@@ -1440,7 +1440,12 @@ 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))

datafusion/sqllogictest/test_files/cte.slt

@@ -1727,3 +1727,117 @@ 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;
+
+# 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;