fix: preserve TopK dynamic filters across recursive CTE iterations by Kristofferjoha · Pull Request #26670 · apache/datafusion

Kristofferjoha · 2026-10-23T16:26:40Z

Which issue does this PR close?

Rationale for this change

TopK dynamic filters inside recursive CTEs stop updating the scan because SortExec::reset_state() creates a new filter on each iteration, while the scan still references the original one. This prevents pruning during recursive queries.

What changes are included in this PR?

Reset the existing shared dynamic filter instead of replacing it.
Rebuild the TopK filter state using the existing DynamicFilterPhysicalExpr.
Add a recursive TopK regression test in cte.slt.

Depends on #26056 for DynamicFilterPhysicalExpr::reset(). It hasn't been merged yet, but it has been approved, so I based this change on its implementation.

What is the testing strategy for this PR?

The new cte.slt case checks that the recursive query returns 1–4 and that the Parquet scan outputs 30 rows instead of 300, pruning 27 row groups across three iterations.

Are there any user-facing changes?

No.

…red_bound reset

jayzhan211 · 2026-10-24T12:02:06Z

            .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());


nit: after the reset this is the same as the existing rebuild_filter_for_current_partitioning() helper, so it could reuse it:

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

My bad, missed the existing helper. reset_state() uses it now.

jayzhan211 · 2026-10-25T03:45:11Z

        // 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()?;


nit: #26669 makes DynamicFilterPhysicalExpr::reset() infallible, so this ? won't compile once both PRs are in. The two PRs touch different files, so git won't flag a conflict. Could you drop the ? in whichever of the two merges second (or rebase this one on #26669)?

- filter.read().expr().reset()?; + filter.read().expr().reset();

let me address this as well

sorry didnt see it until now, Thanks

jayzhan211 · 2026-10-25T11:06:03Z

Thanks @Kristofferjoha

alamb · 2026-11-05T17:24:26Z

This PR appears to have introduced a 56 regression:

Reusing a limit=1 FTS plan concurrently can return an empty result #27130

alamb · 2026-11-05T20:51:31Z

We are going to revert this change for 56, and then we can work on the longer term fix for 57.

@Kristofferjoha

…(revert apache#26670) (apache#27142) ## Which issue does this PR close? - Closes apache#27130. ## Rationale for this change As discussed in apache#27130, this reverts apache#26670 so copies made by `SortExec::reset_state()` no longer share one TopK dynamic filter. When a cached `LIMIT 1` plan was reset twice and both copies ran, the first one to finish tightened the shared filter and the second returned no rows. A plain revert brings back a different wrong result, though: a plan that has run once and is then reset still has its old bound in the scan's predicate (the scan keeps the filter it got at planning time). For `SELECT a FROM t ORDER BY a LIMIT 1` over a parquet file with pushdown enabled, the rerun returned no rows with the plain revert. So this keeps the in-place `reset()` from apache#26056 on the old filter and then gives the new plan its own filter, as @Kristofferjoha suggested keeping. | | main | plain revert | this PR | |---|---|---|---| | reset twice, run both copies (apache#27130) | 2nd: no rows | ok | ok | | run, reset, run again (parquet, pushdown) | ok | 2nd: no rows | ok | ## What changes are included in this PR? - `SortExec::reset_state` resets the old dynamic filter in place, then creates a new one for the reset plan. The `sort_concurrency` reset from apache#26242 is kept. (The plain revert did not apply cleanly because apache#26242 changed the same lines.) - Removes the recursive CTE pruning test apache#26670 added to `cte.slt`, since the scan now keeps the old filter (reset to `true`) and the pruning no longer happens, as in 55. - Adds two unit tests in `sorts/sort.rs`: the reproducer from apache#27130, and a test that the filter a scan would hold is back to `true` after `reset_state`. The second one fails with the plain revert. ## Are these changes tested? Yes. Both new tests fail on main or on the plain revert respectively, and pass here. All `datafusion-physical-plan` unit tests and sqllogictests pass. ## Are there any user-facing changes? The TopK dynamic filter inside a recursive CTE no longer prunes after the first iteration, as in 55 (the apache#26670 test's scan reads 300 rows instead of 30). Query results are unchanged. The proper fix (a fresh filter per copy that the scan also picks up, which would also cover `AggregateExec`) can follow after 56. This reopens apache#26642.

…amic filter (backport #27142) (#27174) ## Which issue does this PR close? - part of #24461 - Backport of #27142 to `branch-56` - Fixes the 56 regression reported in #27130. ## What changes are included in this PR? Clean cherry-pick of #27142: copies made by `SortExec::reset_state()` each get their own TopK dynamic filter again (reverting #26670), and the old filter is reset so a plan that is run, reset and run again doesn't keep the previous bound in the scan. ## Are these changes tested? Same tests as #27142. On `branch-56`: `datafusion-physical-plan` sort tests 164/164, sqllogictest 544/544, clippy and fmt clean. 🤖 Generated with [Claude Code](https://claude.com/claude-code)

mac and others added 5 commits October 5, 2026 07:46

fix: reset aggregate dynamic filters between recursive iterations

0de6c24

Merge branch 'main' into fix/26054-reset-aggregate-dynamic-filters

1e1b704

fix(expr): publish dynamic filter reset in a single step and test sha…

8a56599

…red_bound reset

fixed expr state bug

84f38fc

cte.slt test

afa5704

Kristofferjoha changed the title ~~Fix/topk recursive 26642~~ fix: preserve TopK dynamic filters across recursive CTE iterations Oct 23, 2026

jayzhan211 approved these changes Oct 24, 2026

View reviewed changes

Use existing helper

66d4b61

github-actions Bot added physical-expr Changes to the physical-expr crates sqllogictest SQL Logic Tests (.slt) physical-plan Changes to the physical-plan crate labels Oct 24, 2026

jayzhan211 approved these changes Oct 25, 2026

View reviewed changes

jayzhan211 added 2 commits October 25, 2026 18:56

Merge upstream/main into fix/topk-recursive-26642

75db65f

Update cte.slt TopK plan for current main

d44e0bf

github-actions Bot removed the physical-expr Changes to the physical-expr crates label Oct 25, 2026

jayzhan211 added this pull request to the merge queue Oct 25, 2026

Merged via the queue into apache:main with commit 6aef019 Oct 25, 2026
45 checks passed

github-actions Bot added the v56.0.0 label Oct 25, 2026

comphead mentioned this pull request Oct 27, 2026

fix: bound parallel batch sorting in SortExec #26242

Merged

Xuanwo mentioned this pull request Nov 5, 2026

Release DataFusion 56.0.0 (Oct/Nov 2026) #24461

Open

35 of 57 tasks

alamb mentioned this pull request Nov 5, 2026

Reusing a limit=1 FTS plan concurrently can return an empty result #27130

Closed

raghavyadav01 mentioned this pull request Nov 5, 2026

fix: give each SortExec reset_state copy its own TopK dynamic filter (revert #26670) #27142

Merged

raghavyadav01 mentioned this pull request Nov 6, 2026

[branch-56] fix: give each SortExec reset_state copy its own TopK dynamic filter (backport #27142) #27174

Merged

Navigation Menu

Repository navigation

fix: preserve TopK dynamic filters across recursive CTE iterations - #26670

fix: preserve TopK dynamic filters across recursive CTE iterations#26670
jayzhan211 merged 8 commits into
apache:mainfrom
Kristofferjoha:fix/topk-recursive-26642

Kristofferjoha commented Oct 23, 2026

Uh oh!

jayzhan211 Oct 24, 2026

Uh oh!

Kristofferjoha Oct 24, 2026

Uh oh!

jayzhan211 Oct 25, 2026

Uh oh!

jayzhan211 Oct 25, 2026

Uh oh!

Kristofferjoha Oct 25, 2026

Uh oh!

jayzhan211 commented Oct 25, 2026

Uh oh!

Uh oh!

alamb commented Nov 5, 2026

Uh oh!

alamb commented Nov 5, 2026

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

Conversation

Kristofferjoha commented Oct 23, 2026

Which issue does this PR close?

Rationale for this change

What changes are included in this PR?

What is the testing strategy for this PR?

Are there any user-facing changes?

Uh oh!

jayzhan211 Oct 24, 2026

Choose a reason for hiding this comment

Uh oh!

Kristofferjoha Oct 24, 2026

Choose a reason for hiding this comment

Uh oh!

jayzhan211 Oct 25, 2026

Choose a reason for hiding this comment

Uh oh!

jayzhan211 Oct 25, 2026

Choose a reason for hiding this comment

Uh oh!

Kristofferjoha Oct 25, 2026

Choose a reason for hiding this comment

Uh oh!

jayzhan211 commented Oct 25, 2026

Uh oh!

Uh oh!

alamb commented Nov 5, 2026

Uh oh!

alamb commented Nov 5, 2026

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants