fix: defer InterleaveExec partitioning check to InvariantLevel::Executable - #24963
fix: defer InterleaveExec partitioning check to InvariantLevel::Executable#24963jayzhan211 wants to merge 2 commits into
InterleaveExec partitioning check to InvariantLevel::Executable#24963Conversation
…hildren during rewrites
| } | ||
|
|
||
| #[test] | ||
| fn issue_21826_join_selection_after_distribution_pass() -> Result<()> { |
Codecov Report❌ Patch coverage is
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. 🚀 New features to boost your workflow:
|
| let output_partitioning = if can_interleave(inputs.iter()) { | ||
| first_partitioning.clone() | ||
| } else { | ||
| Partitioning::UnknownPartitioning(first_partitioning.partition_count()) |
There was a problem hiding this comment.
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?
…terleavable inputs
Which issue does this PR close?
InterleaveExectoUnionExecbefore enforcing distribution #24959, which handled theEnsureRequirementsside.Rationale for this change
InterleaveExecis only valid while every child shares the same hash (orrange) partitioning.
EnsureRequirementscreates it from aUnionExecwhenthat happens to hold. Any later rule that rewrites a child and changes its
output partitioning, for example
JoinSelectionswapping a join's sides,makes the optimizer's tree walk rebuild the interleave over children that no
longer match.
InterleaveExec::replace_childrenasserted on that, so thewhole rule failed:
This is the chain reported in #21826: distribution enforcement builds the
interleave,
JoinSelectionbreaks it, and no rule can repair it because thefailure happens inside the rebuild, before any rule's closure runs.
#24959 made
EnsureRequirementsitself immune by normalizing interleaves backto unions before it runs, but that does not help when the rebuild happens
inside another rule.
The root cause is that
InterleaveExecenforces a cross-child distributionproperty one level earlier than the rest of the framework does. Co-partitioning
for joins is validated by
InputDistributionRequirements::check_invariantsonly at
InvariantLevel::Executable; a hash join rebuilt over children thatlost 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 makesInterleaveExecbehave 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 executablecheckpoint 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_newstill asserts interleavability, so explicitconstruction and proto decoding keep the strict check.
try_new_uncheckedbuilds the node without the check.replace_childreninRecomputemode now uses it.compute_propertiesonly claims the shared hash / range partitioning whenevery child still has it, and reports
UnknownPartitioningotherwise, sonothing downstream can rely on a layout the node does not have.
InterleaveExec::check_invariantsruns thecan_interleavecheck atInvariantLevel::Executable, with the same message as before.The physical planner already checks
Alwaysafter each rule andExecutableat 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 inunion.rs): rebuilding over one round-robin child succeeds,try_newonthe same children still fails, the node reports unknown partitioning, the
Alwayscheck passes, and theExecutablecheck fails with the originalmessage.
issue_21826_join_selection_after_distribution_pass(
core/tests/physical_optimizer/enforce_distribution.rs): the exactreported chain. A union of two partitioned hash joins with statistics chosen
so
JoinSelectionswaps only one of them. Onmainthis fails insideJoinSelectionwith the error above. HereJoinSelectioncompletes, theinterleave reports unknown partitioning,
SanityCheckPlanrejects theunrepaired plan, and a second
EnsureRequirementspass produces a validunion that the sanity checker accepts (snapshotted).
interleave_broken_by_later_rewrite_is_repaired_on_next_pass: the samemechanism where the disruption is a removable
CoalescePartitionsExec, sothe 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
InterleaveExecwhose children lose theirshared partitioning during optimization is now reported as an invariant
failure by
SanityCheckPlanat the end of the pipeline instead of as an errorfrom 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.