fix: skip HashJoinExec nodes carrying dynamic filters in JoinSelection

[siddubakka]

Which issue does this PR close?

• Closes JoinSelection should skip HashJoinExec nodes that already carry a dynamic filter instead of failing in swap_inputs #26106.

Rationale for this change

JoinSelection is not safe to run on plans that already carry dynamic filters. This happens during re-optimization passes (e.g. downstream pipelines that wrap an already-optimized plan into a writer sink and re-run physical optimization).

Once FilterPushdown has wired up a dynamic filter between the build side and probe side of a HashJoinExec, swapping the inputs via swap_inputs panics with:

Internal error: Cannot swap HashJoinExec inputs after dynamic filters have been constructed.

The join's build side is already committed at that stage, and swapping inputs would invalidate the dynamic filter expressions that reference probe-side columns. JoinSelection should detect this and leave the HashJoinExec unchanged instead of failing the plan.

What changes are included in this PR?

• In JoinSelection::statistical_join_selection_subrule, check if !hash_join.dynamic_expressions_produced().is_empty() and return None (leaving the plan unchanged).
• In can_swap_hash_join, guard against swapping when dynamic expressions are produced.
• In hash_join_swap_subrule, guard against swapping unbounded left inputs when dynamic expressions are produced.
• Added tests in datafusion/core/tests/physical_optimizer/join_selection.rs verifying that JoinSelection skips HashJoinExec carrying dynamic filters in both CollectLeft and Partitioned modes.

What is the testing strategy for this PR?

Added test_join_selection_skips_hash_join_with_dynamic_filter in datafusion/core/tests/physical_optimizer/join_selection.rs verifying that JoinSelection leaves the plan unchanged for both CollectLeft and Partitioned modes without error.

Are there any user-facing changes?

No API changes. Fixes an internal error when re-optimizing plans that contain dynamic filters.

[zhuqi-lucas]

@siddubakka The CI has some fails, need to make it pass first.

[siddubakka]

@zhuqi-lucas fixed the fmt and test failure, CI should be good now

[codecov-commenter]

Codecov Report

❌ Patch coverage is 69.56522% with 7 lines in your changes missing coverage. Please review.
✅ Project coverage is 82.83%. Comparing base (45de64c) to head (c5f7217).
⚠️ Report is 1 commits behind head on main.

Files with missing lines	Patch %	Lines
...atafusion/physical-optimizer/src/join_selection.rs	69.56%	3 Missing and 4 partials ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #26123      +/-   ##
==========================================
- Coverage   82.83%   82.83%   -0.01%     
==========================================
  Files        1149     1149              
  Lines      452706   452709       +3     
  Branches   452706   452709       +3     
==========================================
  Hits       375007   375007              
  Misses      54876    54876              
- Partials    22823    22826       +3     

☔ 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.

[mikamikasuki]

This attribute makes the current clippy --all-targets --workspace ... -D warnings job fail: Clippy 1.99 denies clippy::allow_attributes and explicitly suggests #[expect]. Replacing it with #[expect(deprecated)] preserves the intended scoped deprecation handling and also warns if the deprecation is later removed.

[siddubakka]

@mikamikasuki thank you for your help i have made the changes now

[siddubakka]

@zhuqi-lucas fixed the clippy failure and the codecov issue (removed an unreachable branch that was dropping coverage). also added a test for it, so everything should be passing now.

[siddubakka]

hey @zhuqi-lucas @xudong963, i pushed the latest updates yesterday. could one of you approve the workflow runs when you have a moment so the CI checks can kick off? thanks!

[rgbuilds]

cargo fmt --all -- --check currently fails on the signature of test_join_selection_skips_unbounded_hash_join_with_dynamic_filter. Please run cargo fmt --all; this is the only merge-blocking issue I found in the implementation

[rgbuilds]

Could we add a regression that lets FilterPushdown::new_post_optimization() create the dynamic filter and probe-side consumer, then reruns JoinSelection and executes the plan? The current tests correctly exercise both guards, but manually calling with_dynamic_filter_expr does not verify the producer/consumer connection or query results.

[siddubakka]

@rgbuilds hey thanks for the review, fixed the fmt failure on that test signature

and yeah you're right about the manual with_dynamic_filter_expr test — added one that runs FilterPushdown::new_post_optimization to wire up a real filter, then checks JoinSelection leaves it alone and the plan still returns the right rows

[rgbuilds]

@rgbuilds hey thanks for the review, fixed the fmt failure on that test signature

and yeah you're right about the manual with_dynamic_filter_expr test — added one that runs FilterPushdown::new_post_optimization to wire up a real filter, then checks JoinSelection leaves it alone and the plan still returns the right rows

Thanks—the formatting issue is fixed, and the real FilterPushdown wiring and execution check are helpful.

One issue remains: both TestScanBuilder inputs report the same 123-byte size, so should_swap_join_order returns false. I confirmed that this test still passes when the statistical dynamic-filter guard is removed, because JoinSelection never attempts the problematic swap.

Could we give the left/build side larger reported statistics than the right/probe side, or otherwise force should_swap_join_order to return true?

It would also be helpful to assert that the original left and right children are preserved. Then the test should fail without the guard and pass with it.

[siddubakka]

@rgbuilds ah yeah missed that, both sides were 123 bytes so the swap never triggered. forced the build side bigger with a registry now and added ptr_eq checks on both children, should actually exercise the guard now

[rgbuilds]

@rgbuilds ah yeah missed that, both sides were 123 bytes so the swap never triggered. forced the build side bigger with a registry now and added ptr_eq checks on both children, should actually exercise the guard now

Thanks—the build-side statistics override and the Arc::ptr_eq assertions are useful improvements. However, I repeated the guard-removal check locally at commit 1643c176, and the regression test still passes when the statistical dynamic-filter guard is disabled.

For the ablation, I temporarily changed:

if !hash_join.dynamic_expressions_produced().is_empty() {

to:

if false && !hash_join.dynamic_expressions_produced().is_empty() {

I then reran:

cargo test -p datafusion \
  --test core_integration \
  test_join_selection_skips_real_dynamic_filter_pushdown \
  -- --nocapture

The test still passed.

The build-side statistics provider is being applied, but after FilterPushdown wires the dynamic filter into the probe side, the statistics visible to should_swap_join_order are:

left/build:
  num_rows        = Inexact(262144)
  total_byte_size = Inexact(8388608)

right/probe:
  num_rows        = Absent
  total_byte_size = Absent

should_swap_join_order first compares total_byte_size when both sides have values. If that is not possible, it compares num_rows when both sides have values. Because the probe side has neither value, both comparisons are unavailable and the function returns false.

Therefore, with the guard removed:

1. JoinSelection still does not call swap_inputs.
2. The left and right children remain unchanged.
3. The Arc::ptr_eq assertions pass.
4. The test succeeds even though the guard is absent.

Could we also register small statistics for plans whose schema contains the probe_ fields? For example:

registry.register(Arc::new(ClosureStatisticsProvider::with_matches(
    |plan| {
        plan.schema()
            .fields()
            .iter()
            .any(|f| f.name().starts_with("probe_"))
    },
    |_plan, _child_stats| {
        Ok(StatisticsResult::Computed(small_statistics().into()))
    },
)));

I tested this locally. With the additional probe-side provider, the statistics become comparable:

build total_byte_size = 8,388,608
probe total_byte_size = 32,768

With those statistics and the guard disabled, should_swap_join_order returns true, JoinSelection reaches swap_inputs, and the test fails with the expected error:

Cannot swap HashJoinExec inputs after dynamic filters have been constructed

That gives the regression test the behavior we want:

guard present
→ dynamically filtered join is skipped
→ original children are preserved
→ execution succeeds

guard removed
→ statistics request a swap
→ swap_inputs rejects the dynamically filtered join
→ test fails

[siddubakka]

@rgbuilds added the probe side provider too — build reports big stats and probe reports small now so the swap is actually requested. checked the logic, without the guard it goes to swap_inputs and errors out, with it the join is skipped and the rows still come back right

[rgbuilds]

@rgbuilds added the probe side provider too — build reports big stats and probe reports small now so the swap is actually requested. checked the logic, without the guard it goes to swap_inputs and errors out, with it the join is skipped and the rows still come back right

Thanks—I verified the latest update locally at 1f56be8.

The focused regression passes with the guard present and fails at swap_inputs as expected when I disable the guard. This addresses my concern.

[github-actions]

Thank you for opening this pull request!

Reviewer note: cargo-semver-checks reported the current version number is not SemVer-compatible with the changes in this pull request (compared against the base branch).

Details

     Cloning apache/main
    Building datafusion v55.1.0 (current)
       Built [  56.044s] (current)
     Parsing datafusion v55.1.0 (current)
      Parsed [   0.031s] (current)
    Building datafusion v55.1.0 (baseline)
       Built [  54.991s] (baseline)
     Parsing datafusion v55.1.0 (baseline)
      Parsed [   0.033s] (baseline)
    Checking datafusion v55.1.0 -> v55.1.0 (no change; assume patch)
     Checked [   0.608s] 223 checks: 223 pass, 31 skip
     Summary no semver update required
    Finished [ 113.619s] datafusion
    Building datafusion-physical-optimizer v55.1.0 (current)
       Built [  37.259s] (current)
     Parsing datafusion-physical-optimizer v55.1.0 (current)
      Parsed [   0.020s] (current)
    Building datafusion-physical-optimizer v55.1.0 (baseline)
       Built [  37.025s] (baseline)
     Parsing datafusion-physical-optimizer v55.1.0 (baseline)
      Parsed [   0.020s] (baseline)
    Checking datafusion-physical-optimizer v55.1.0 -> v55.1.0 (no change; assume patch)
     Checked [   0.104s] 223 checks: 222 pass, 1 fail, 0 warn, 31 skip

--- failure function_missing: pub fn removed or renamed ---

Description:
A publicly-visible function cannot be imported by its prior path. A `pub use` may have been removed, or the function itself may have been renamed or removed entirely.
        ref: https://doc.rust-lang.org/cargo/reference/semver.html#item-remove
       impl: https://github.com/obi1kenobi/cargo-semver-checks/tree/v0.50.0/src/lints/function_missing.ron

Failed in:
  function datafusion_physical_optimizer::limit_pushdown::pushdown_limit_helper_with_stats, previously in file /home/runner/work/datafusion/datafusion/target/semver-checks/git-apache_main/17e0fde672add5ee0fe5c99925a455131065fc52/datafusion/physical-optimizer/src/limit_pushdown.rs:184

     Summary semver requires new major version: 1 major and 0 minor checks failed
    Finished [  75.742s] datafusion-physical-optimizer

[siddubakka]

hi @zhuqi-lucas @xudong963 — this is ready for review whenever you get a chance. for context, @rgbuilds reviewed the earlier rounds and approved after verifying the regression test passes with the guard and fails at swap_inputs without it.

I also just resolved the merge conflict with main (upstream's StatisticsContext refactor touched the same function, re-applied the guards on top — cargo check passes on the lib and core_integration target).

two things worth knowing: the semver bot failure is pre-existing on main (it's about pushdown_limit_helper_with_stats being removed in limit_pushdown.rs, not this PR — this diff only touches join_selection.rs + its test), and the red X on my merge commit is just a cancelled superseded run. the CI runs are also still waiting on approval.

thanks!