fix: preserve TopK dynamic filters across recursive CTE iterations

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

@@ -1577,7 +1577,10 @@ 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());
+        if let Some(filter) = &new_sort.filter {
+            filter.read().expr().reset();
+        }
+        new_sort.rebuild_filter_for_current_partitioning();
         new_sort.metrics_set = ExecutionPlanMetricsSet::new();
 
         Ok(Arc::new(new_sort))

datafusion/sqllogictest/test_files/cte.slt

@@ -2006,3 +2006,86 @@ SET datafusion.execution.target_partitions = 4;
 
 statement ok
 RESET datafusion.execution.parquet.pushdown_filters;
+
+# 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;N) 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, 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;