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

@Xuanwo

Describe the bug

Reusing a limit=1 FTS plan concurrently can return an empty result because SortExec::reset_state() now reuses dynamic-filter state (fix: preserve TopK dynamic filters across recursive CTE iterations #26670).

I am filing a ticket to track this work

To Reproduce

Not sure yet

Expected behavior

No response

Additional context

No response

Willingness to contribute

None

take

untake

Reproducer

I could not find a way to reproduce this with SQL in datafusion-cli: each statement plans a fresh SortExec, so two executions never share a dynamic filter. The only SQL path that calls reset_state is a recursive CTE, and its iterations run one after another (the case #26670 fixed). The Lance scenario is a cached physical plan that is reset and executed by two queries at the same time, so here is a Rust test that models that.

Add the following to the tests module at the end of datafusion/physical-plan/src/sorts/sort.rs (it compiles unchanged on both 55.0.0 and main):

    /// Regression test for <https://github.com/apache/datafusion/issues/27130>
    ///
    /// Plans derived from the same TopK plan via `reset_state` must not share
    /// dynamic filter state. This models a cached `LIMIT 1` plan that is reset
    /// and executed by two queries at the same time: both copies are created
    /// before either executes, and the first one to finish must not cause the
    /// other to return an empty result.
    #[tokio::test]
    async fn test_topk_reset_state_copies_are_independent() -> Result<()> {
        let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, false)]));
        let batch = RecordBatch::try_new(
            Arc::clone(&schema),
            vec![Arc::new(Int32Array::from(vec![3, 1, 2]))],
        )?;
        let input =
            TestMemoryExec::try_new_exec(&[vec![batch]], Arc::clone(&schema), None)?;
        let sort: Arc<dyn ExecutionPlan> = Arc::new(
            SortExec::new(
                [PhysicalSortExpr::new_default(Arc::new(Column::new("a", 0)))].into(),
                input,
            )
            .with_fetch(Some(1)),
        );

        // Both executions are prepared before either runs
        let first = Arc::clone(&sort).reset_state()?;
        let second = Arc::clone(&sort).reset_state()?;

        let task_ctx = Arc::new(TaskContext::default());
        let first_result = collect(first, Arc::clone(&task_ctx)).await?;
        let second_result = collect(second, task_ctx).await?;

        assert_snapshot!(batches_to_string(&first_result), @r"
        +---+
        | a |
        +---+
        | 1 |
        +---+
        ");
        assert_snapshot!(batches_to_string(&second_result), @r"
        +---+
        | a |
        +---+
        | 1 |
        +---+
        ");

        Ok(())
    }

Run with

cargo test -p datafusion-physical-plan --lib test_topk_reset_state_copies_are_independent

Result on `55.0.0`

test sorts::sort::tests::test_topk_reset_state_copies_are_independent ... ok

Result on `main` (`70f2ed0`)

The second execution returns no rows:

━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ Snapshot Summary ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
Snapshot: topk_reset_state_copies_are_independent-2
Source: datafusion/physical-plan/src/sorts/sort.rs:4807
────────────────────────────────────────────────────────────────────────────────
Expression: batches_to_string(&second_result)
────────────────────────────────────────────────────────────────────────────────
-old snapshot
+new results
────────────┬───────────────────────────────────────────────────────────────────
    1       │-+---+
    2       │-| a |
    3       │-+---+
    4       │-| 1 |
    5       │-+---+
          1 │+++
          2 │+++
────────────┴───────────────────────────────────────────────────────────────────

Root cause (speculation)

Before #26670, SortExec::reset_state gave the new SortExec a brand new dynamic filter. Now it calls reset() on the existing DynamicFilterPhysicalExpr and only rebuilds the TopKDynamicFilters wrapper, so every reset copy (and the original) shares one DynamicFilterPhysicalExpr.

When the first copy finishes a LIMIT 1 it publishes a < 1 into that shared expression. The second copy's TopK evaluates the shared expression against its own input in TopK::insert_batch (the TopK pre-filters rows itself, so no scan / filter pushdown is needed to hit this), every row is dropped, and it emits nothing.

Note that the doc comment on reset_plan_states on main already says a plan that uses dynamic filters may not be reset while the original or another reset copy is still executing, so this is currently documented as a limitation. It is still a behavior change from 55 though.

take

I can reproduce this with the test above on main (b6b2e0e): the first copy returns 1, the second returns no rows.

Question before I write the fix: #26670 keeps the same DynamicFilterPhysicalExpr across reset_state on purpose, so the scan that holds it keeps seeing updates. Two copies made by reset_state therefore share one filter. I see two options:

Give each copy a fresh filter in reset_plan_states and point the scan's predicate at it, so the recursive CTE case from fix: preserve TopK dynamic filters across recursive CTE iterations #26670 still prunes.
For 56, go back to creating a new filter in reset_state (as before fix: preserve TopK dynamic filters across recursive CTE iterations #26670), and handle the recursive CTE case separately.

HashJoinExec::reset_state reuses its filter the same way, so I'd check it too. Which would you prefer?

take

let's revert to the 55 behavior

@raghavyadav01

@raghavyadav01 thank you -- can you open a PR to propose reverting the 56 change?

Specifically, revert #26670

@alamb

@alamb Sorry, my bad. #26670 assumed there's only ever one reset copy running, which is true for recursive CTEs but not for cached plans. +1 on the revert for 56.

FYI though, AggregateExec has the same problem on main. I tested SELECT max(x) FROM t WHERE x % 2 = 0 on a parquet file (x = 1..1000, 10 row groups), reset the plan twice and ran both copies. The first returns 1000, the second returns NULL because its scan picks up the first copy's x > 1000 filter. Running them concurrently is worse, most of the time you get wrong but plausible values like 900 or 800.

It's not new in 56 though. On 55.2.0 AggregateExec didn't override reset_state at all, so even reset -> run -> reset -> run gave NULL (#26054). #26056 fixed that part, so I don't think it should be reverted along with #26670.

	55.2.0	main
topk limit 1, reset x2 then run both	ok	2nd: no rows
max, reset/run/reset/run	2nd: NULL	ok
max, reset x2 then run both	2nd: NULL	2nd: NULL
max, 2 copies concurrently (50 runs)	40 wrong	45 wrong

Haven't tested HashJoin, but since its filter just describes the build side I'd guess it's fine.

@raghavyadav01 I think option 1 is the real fix as you mention, since it would cover both topk and aggregate and keep the CTE pruning.

@alamb

@alamb Sorry, my bad. #26670 assumed there's only ever one reset copy running, which is true for recursive CTEs but not for cached plans. +1 on the revert for 56.

No worries! We'll just revert this one temporarily to stop the regression

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

Then we can work on the real fix for 57.

fix: give each SortExec reset_state copy its own TopK dynamic filter …

[branch-56] fix: give each SortExec reset_state copy its own TopK dyn…

Navigation Menu

Repository navigation

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

Describe the bug

To Reproduce

Expected behavior

Additional context

Willingness to contribute

Reproducer

Result on `55.0.0`

Result on `main` (`70f2ed0`)

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

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

Description

Describe the bug

To Reproduce

Expected behavior

Additional context

Willingness to contribute

Activity

1fanwang commented on Nov 5, 2026

1fanwang commented on Nov 5, 2026

alamb commented on Nov 5, 2026

Reproducer

Result on 55.0.0

Result on main (70f2ed0)

raghavyadav01 commented on Nov 5, 2026

raghavyadav01 commented on Nov 5, 2026

alamb commented on Nov 5, 2026

alamb commented on Nov 5, 2026

alamb commented on Nov 5, 2026

Kristofferjoha commented on Nov 5, 2026

alamb commented on Nov 5, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions

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

Result on `55.0.0`

Result on `main` (`70f2ed0`)