[mikamikasuki]

Which issue does this PR close?

Closes #26054

Rationale for this change

In a recursive CTE, an uncorrelated MIN/MAX over a Parquet table returns the correct value on the first iteration but NULL or an incorrect value on later iterations because the aggregate reuses the dynamic filter tightened during the first scan.

What changes are included in this PR?

Reset AggregateExec's dynamic filter expression and shared accumulator bounds when the plan is reused, allowing each recursive iteration to scan with fresh bounds.

What is the testing strategy for this PR?

Added a cte.slt regression for the reported recursive MAX query. Validation passed: cargo test -p datafusion-sqllogictest --test sqllogictests -- cte.slt; cargo test -p datafusion-physical-plan --lib (2,441 passed); cargo test -p datafusion-physical-expr --lib (1,686 passed, 2 ignored); cargo fmt --all -- --check; and targeted Clippy with -A clippy::ignore_without_reason -D warnings. The unsuppressed Rust 1.99 Clippy run reports the existing ignore_without_reason lint in datafusion/physical-expr/src/planner.rs.

Are there any user-facing changes?

Recursive CTE iterations now return the correct aggregate result consistently while retaining dynamic-filter pruning.

[adriangb]

Are there any issues / situations where we'd reset the filter on the aggregate but not the scan?

Otherwise looks good to me!

[adriangb]

What's the point of this line? Do we need to add some sort of FilterState::Reset?

[jayzhan211]

Thanks @mikamikasuki, this looks right to me. Resetting the filter in place (same DynamicFilterPhysicalExpr, new value) is what keeps the scan, which holds a reference to that filter, working across iterations.

I checked out 1e1b704 and verified:

• cte.slt and the dynamic_filters unit tests pass.
• Without dynamic_filter.filter.reset(), the new cte.slt case fails (iterations 2 and 3 return NULL), and on main it returns 0 NULL, 1 128, 2 NULL, 3 NULL.

Two small suggestions:

1. The shared_bound reset is not covered by a test. If I drop the loop that sets shared_bound back to ScalarValue::Null, cte.slt still passes, including with several row groups and batch_size = 8. Results stay correct, but pruning silently stops after the first iteration: iteration 2 starts with the stale bound (128), so scalar_max(128, local) never changes, bounds_changed stays false, and the filter is never published again. A small unit test that runs reset_state and asserts every shared_bound is NULL (and the filter expression is true) would pin this.

2. Nit: reset() publishes twice. It calls update(lit(true)), which broadcasts InProgress while is_complete is still true, then takes the lock again to clear is_complete and broadcasts a second time. Writing the new Inner (expression true, generation + 1, is_complete: false) under one write lock and sending once would make the reset a single step. It is harmless today because true never prunes, so this is optional.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 93.93939% with 2 lines in your changes missing coverage. Please review.
✅ Project coverage is 82.68%. Comparing base (b40d696) to head (1e1b704).
⚠️ Report is 150 commits behind head on main.

Files with missing lines	Patch %	Lines
...ysical-expr/src/expressions/dynamic_filters/mod.rs	95.45%	0 Missing and 1 partial ⚠️
datafusion/physical-plan/src/aggregates/mod.rs	90.90%	0 Missing and 1 partial ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #26056      +/-   ##
==========================================
- Coverage   82.69%   82.68%   -0.01%     
==========================================
  Files        1147     1147              
  Lines      447204   447237      +33     
  Branches   447204   447237      +33     
==========================================
+ Hits       369793   369814      +21     
- Misses      55001    55007       +6     
- Partials    22410    22416       +6     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[adriangb]

SortExec::reset_state makes a new filter with create_filter(), and HashJoinExec::reset_state sets dynamic_filter = None. This is now a 3rd pattern. Do we need 3 patterns? If not which ones should we go with and why?

[alamb]

@mikamikasuki can you please help finish this PR by responding to the feedback?

[adriangb]

Btw I think this may be the right answer, but I want to us to have a plan (e.g. update the other 2 approaches to use this one) before we add a 3rd.

[jayzhan211]

Thanks @adriangb for the suggestion.

I think in-place reset should be the only pattern, and HashJoinExec and SortExec should move to it.

The filter's identity is what connects the producer to the scan, and reset_state can't reconnect it. reset_plan_states resets the scan before its producer, and the scan keeps the old filter. So dropping or recreating the filter leaves the scan with one nobody updates: a recursive join prunes 0 rows today, versus 98/100 outside a CTE (#19905). The only alternative is a second pass to reconnect filters (PushedDynamicFilters in #19905), which is much more machinery than one reset().

The cost is that a reset copy can't run at the same time as the original. Main already has that limitation, since the scan shares the filter either way, and RecursiveQueryExec runs iterations one after another. I'll document it on reset_state.

I'll open follow-up issues to move HashJoinExec and SortExec to in-place reset. Each conversion is a few lines, plus a cte.slt check that the recursive case prunes.

[adriangb]

Some minor nits

[adriangb]

why not reset the generation to 0? @jayzhan211 is it obvious to you? if it's not lets add a comment.

[jayzhan211]

It isn't obvious, so a comment would help. The generation has to keep increasing across a reset. current() caches the remapped expression by generation, so if reset() went back to 1 (or 0), later updates would reach the cached generation again and current() would return the previous run's expression. I tried it: with the generation reset to 1, the first update after the reset returns the old expression.

@mikamikasuki could you add a comment like:

// Keep the generation increasing: `current()` caches by generation, so going
// back would let a later update hit the previous run's cached expression.

[adriangb]

Same here, why send this message and why this generation vs. e.g. 0?

[jayzhan211]

The watch channel still holds Complete from the previous run. Without this send, wait_complete() sees Complete and returns right away, before this run's filter is ready. I checked this with a subscriber held across the reset. It sends new_generation so the channel matches inner.

@mikamikasuki and a comment here like:

// The channel still holds `Complete` from the previous run; publish
// `InProgress` so `wait_complete()` waits for this run's filter.

[alamb]

This PR appears to be waiting on the author to respond to reviewer feedback so marking as draft. Please mark ready to review when it is ready for another look

[jayzhan211]

How about I add the comments and merge this

[adriangb]

How about I add the comments and merge this

sounds good to me!