Repository navigation
fix: normalize InterleaveExec to UnionExec before enforcing distribution - #24959
Conversation
…ution enforcement
|
@zhuqi-lucas WDYT about converting the interleave into a union like this? Are there concerns I'm missing? |
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #24959 +/- ##
==========================================
- Coverage 81.62% 81.62% -0.01%
==========================================
Files 1124 1124
Lines 412462 412474 +12
Branches 412462 412474 +12
==========================================
- Hits 336684 336682 -2
- Misses 55965 55973 +8
- Partials 19813 19819 +6 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
zhuqi-lucas
left a comment
There was a problem hiding this comment.
Thanks @jayzhan211 — approach matches what #21827 landed on.
One scope note: the invariant itself is unchanged. On your branch, rebuilding an interleave whose children lost their partitioning (what any transform_up does) still fails:
Internal error: Assertion failed: can_interleave(children.iter())
EnsureRequirements is safe because Phase 0 strips interleaves before anything rebuilds them, but other rules aren't. JoinSelection runs at optimizer.rs:101, before this rule at :136, so the case reported in #21826 — a custom chain running EnforceDistribution twice, JoinSelection then walking an interleave the earlier pass built — isn't covered here. Same goes for the ~10 rules after :136 that rebuild the tree while an interleave is live.
Not blocking — the idempotency fix is worth having. Just maybe Part of #21826 rather than Closes.
|
Let me find another solution to close #21827 |
…ecutable` (apache#24963) ## Which issue does this PR close? - Closes apache#21826. Builds on apache#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 apache#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. apache#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 apache#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 apache#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.
Which issue does this PR close?
Rationale for this change
EnsureRequirementsfails with an internal error when the plan it is givenalready contains an
InterleaveExecwhose children no longer share a hash(or range) partitioning:
InterleaveExecdeclares no distribution requirement of its own. It is onlyvalid while its children happen to share the same partitioning, which is why
ensure_distributioncreates it from aUnionExecin the first place. Onceit exists, several rewrites can take that partitioning away: this rule strips
RepartitionExecs that nothing requires, join key reordering changes the hashexpressions,
JoinSelectionside swaps change output partitioning, and so on.PlanContext::with_new_childrenrebuilds every parent from its updatedchildren while walking the tree, so the interleave is rebuilt over the changed
children before the rule's closure sees the node, and the rebuild fails.
This affects any pipeline that runs the rule more than once, re-optimizes a
deserialized plan, or runs
EnforceDistributionafter other rewrites (thesetup reported in #21826). It also breaks the idempotency the rule advertises.
The discussion on #21827 concluded that the fix belongs at the
EnforceDistributioncall site rather than as a silent fallback inInterleaveExec::with_new_children; this PR is that fix.What changes are included in this PR?
replace_interleave_with_unionhelper inensure_requirements/enforce_distribution.rs, which turns anInterleaveExecback into the equivalentUnionExec.EnsureRequirements::optimizeruns it as a top-down "Phase 0" before joinkey reordering. Phase 2 then re-derives interleaves from unions wherever the
children still qualify, exactly as it already re-derives
RepartitionExec,CoalescePartitionsExecandSortPreservingMergeExec.change its output partitioning and make the parent interleave fail to
rebuild.
A minimal reproducer that fails on
mainand passes here:What is the testing strategy for this PR?
Three new tests in
datafusion/core/tests/physical_optimizer/enforce_distribution.rs, each runthrough the existing multi-pass harness (
DISTRIB_DISTRIB_SORTandSORT_DISTRIB_DISTRIB) so idempotency is checked as well:interleave_falls_back_to_union_when_children_lose_partitioning: thereproducer above; the result is a
UnionExecover the scans.interleave_fallback_still_satisfies_parent_hash_requirement: same inputunder a
FinalPartitionedaggregate; the aggregate gets a single hashrepartition above the union.
existing_interleave_is_kept_when_children_stay_interleavable: aninterleave over hash partitioned aggregates is re-derived as an interleave,
and stays a union with
prefer_existing_union = true.The test harness's tree-node integrity block now applies the same pre-pass so
it mirrors the rule. Existing
union_to_interleaveandunion_not_to_interleavetests are unchanged, andunion.sltpasses.Are there any user-facing changes?
No API changes. One behavioral note: with
datafusion.optimizer.prefer_existing_union = true, anInterleaveExecpresent in the input plan is now emitted as a
UnionExec. The rule nevercreates interleaves under that setting, and plans produced by the default
pipeline are unaffected because
EnsureRequirementsis the only creator ofInterleaveExecand runs once.