diff --git a/datafusion/physical-plan/src/sorts/sort.rs b/datafusion/physical-plan/src/sorts/sort.rs index 13065bed078c0..da272e2ff3341 100644 --- a/datafusion/physical-plan/src/sorts/sort.rs +++ b/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)) diff --git a/datafusion/sqllogictest/test_files/cte.slt b/datafusion/sqllogictest/test_files/cte.slt index a097312cdcb29..400eb45e394a3 100644 --- a/datafusion/sqllogictest/test_files/cte.slt +++ b/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;