jayzhan211 opened a new pull request, #24963:
URL: https://github.com/apache/datafusion/pull/24963
## Which issue does this PR close?
- Closes #21826. Builds on #24959, which handled the `EnsureRequirements`
side.
## 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
