Skip to content

fix: defer InterleaveExec partitioning check to InvariantLevel::Executable - #24963

Open
jayzhan211 wants to merge 2 commits into
apache:mainfrom
jayzhan211:interleave-deferred-invariant
Open

fix: defer InterleaveExec partitioning check to InvariantLevel::Executable#24963
jayzhan211 wants to merge 2 commits into
apache:mainfrom
jayzhan211:interleave-deferred-invariant

Conversation

@jayzhan211

Copy link
Copy Markdown
Contributor

Which issue does this PR close?

Rationale for this change

InterleaveExec is only valid while every child shares the same hash (or
range) partitioning. EnsureRequirements creates it from a UnionExec when
that happens to hold. Any later rule that rewrites a child and changes its
output partitioning, for example JoinSelection swapping a join's sides,
makes the optimizer's tree walk rebuild the interleave over children that no
longer match. InterleaveExec::replace_children asserted on that, so the
whole rule failed:

join_selection
caused by
Internal error: Assertion failed: can_interleave(children.iter()):
Can not create InterleaveExec: new children can not be interleaved.

This is the chain reported in #21826: distribution enforcement builds the
interleave, JoinSelection breaks it, and no rule can repair it because the
failure happens inside the rebuild, before any rule's closure runs.

#24959 made EnsureRequirements itself immune by normalizing interleaves back
to unions before it runs, but that does not help when the rebuild happens
inside another rule.

The root cause is that InterleaveExec enforces a cross-child distribution
property one level earlier than the rest of the framework does. Co-partitioning
for joins is validated by InputDistributionRequirements::check_invariants
only at InvariantLevel::Executable; a hash join rebuilt over children that
lost their partitioning does not error at rebuild time, it becomes temporarily
unsatisfied and is either repaired by a later distribution pass or rejected by
SanityCheckPlan. This PR makes InterleaveExec behave the same way.

This is not the fallback proposed in #21827. Nothing is substituted for a
different node and no error is downgraded to a log. The node stays an
InterleaveExec, the error is the same, and it is raised at the executable
checkpoint instead of at rebuild time. The decision on how to recover is left
to the optimizer rule, which #24959 already implements.

What changes are included in this PR?

In datafusion/physical-plan/src/union.rs:

  • InterleaveExec::try_new still asserts interleavability, so explicit
    construction and proto decoding keep the strict check.
  • A private try_new_unchecked builds the node without the check.
    replace_children in Recompute mode now uses it.
  • compute_properties only claims the shared hash / range partitioning when
    every child still has it, and reports UnknownPartitioning otherwise, so
    nothing downstream can rely on a layout the node does not have.
  • InterleaveExec::check_invariants runs the can_interleave check at
    InvariantLevel::Executable, with the same message as before.

The physical planner already checks Always after each rule and Executable
at the end, so an unrepaired interleave is still rejected, with the same
error, by SanityCheckPlan.

What is the testing strategy for this PR?

  • test_interleave_rebuild_defers_partitioning_invariant (unit test in
    union.rs): rebuilding over one round-robin child succeeds, try_new on
    the same children still fails, the node reports unknown partitioning, the
    Always check passes, and the Executable check fails with the original
    message.
  • issue_21826_join_selection_after_distribution_pass
    (core/tests/physical_optimizer/enforce_distribution.rs): the exact
    reported chain. A union of two partitioned hash joins with statistics chosen
    so JoinSelection swaps only one of them. On main this fails inside
    JoinSelection with the error above. Here JoinSelection completes, the
    interleave reports unknown partitioning, SanityCheckPlan rejects the
    unrepaired plan, and a second EnsureRequirements pass produces a valid
    union that the sanity checker accepts (snapshotted).
  • interleave_broken_by_later_rewrite_is_repaired_on_next_pass: the same
    mechanism where the disruption is a removable CoalescePartitionsExec, so
    the repair restores the interleave rather than demoting it, showing the
    optimization is not permanently lost.

Are there any user-facing changes?

No API changes. Behaviorally, an InterleaveExec whose children lose their
shared partitioning during optimization is now reported as an invariant
failure by SanityCheckPlan at the end of the pipeline instead of as an error
from the rule that rebuilt it. The diagnostic is therefore coarser (it names
the node, not the rule), which is the trade-off for letting rules complete and
a later distribution pass repair the plan. Custom pipelines that run rules
between two distribution passes no longer need a workaround.

@github-actions github-actions Bot added core Core DataFusion crate physical-plan Changes to the physical-plan crate labels Sep 6, 2026
}

#[test]
fn issue_21826_join_selection_after_distribution_pass() -> Result<()> {

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The case in #21826

@codecov-commenter

codecov-commenter commented Sep 6, 2026

Copy link
Copy Markdown

Codecov Report

❌ Patch coverage is 75.67568% with 18 lines in your changes missing coverage. Please review.
✅ Project coverage is 81.67%. Comparing base (3266eaa) to head (9abad8d).
⚠️ Report is 4 commits behind head on main.

Files with missing lines Patch % Lines
datafusion/physical-plan/src/union.rs 75.67% 5 Missing and 13 partials ⚠️
Additional details and impacted files
@@            Coverage Diff             @@
##             main   #24963      +/-   ##
==========================================
+ Coverage   81.64%   81.67%   +0.03%     
==========================================
  Files        1124     1126       +2     
  Lines      413173   414594    +1421     
  Branches   413173   414594    +1421     
==========================================
+ Hits       337320   338607    +1287     
- Misses      55995    56060      +65     
- Partials    19858    19927      +69     

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

Comment thread datafusion/physical-plan/src/union.rs Outdated
let output_partitioning = if can_interleave(inputs.iter()) {
first_partitioning.clone()
} else {
Partitioning::UnknownPartitioning(first_partitioning.partition_count())

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

One non-blocking nit: in the degraded branch, UnknownPartitioning(first_partitioning.partition_count()) takes the count from inputs[0]. If another child has more partitions, those are never executed —
silent row loss rather than an error. Unreachable through the planner since the Executable check rejects the plan first, but max() over the children would make the bypass path fail loudly instead. One line, and it costs nothing?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This makes sense

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

core Core DataFusion crate physical-plan Changes to the physical-plan crate

Projects

None yet

Development

Successfully merging this pull request may close these issues.

InterleaveExec::with_new_children panics when optimizer rewrites change children's partitioning

3 participants