fix: keep a CoalescePartitionsExec required by a SinglePartition child - #23948
fix: keep a CoalescePartitionsExec required by a SinglePartition child#23948adriangb wants to merge 2 commits into
Conversation
`parallelize_sorts` (phase 3a of `EnsureRequirements`) removes `CoalescePartitionsExec`s that only act as a parallelism bottleneck below a global sort. `remove_bottleneck_in_subplan` decides that positionally: if `children[0]` is a `CoalescePartitionsExec`, it is removed, without consulting the parent's own distribution requirement for that child. The traversal reaches such a parent because `update_coalesce_ctx_children` marks a node as connected when *any* of its children is linked to a coalesce below, and it deliberately excludes children that require `Distribution::SinglePartition` only from *setting* that flag. So a `CollectLeft` `HashJoinExec` whose probe side is connected is descended into, and the coalesce satisfying the build side's `SinglePartition` requirement is then taken out of the plan. Nothing re-runs distribution enforcement afterwards, so the first thing to notice is `SanityCheckPlan`, which rejects the plan with `does not satisfy distribution requirements: SinglePartition`. Consult the per-child distribution requirement before removing, both for `children[0]` and when recursing into other children. The node `parallelize_sorts` is itself rewriting (the root of the call) is exempt: the caller drops that node and rebuilds the sort cascade around the result, so its requirement does not constrain the removal. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Covers the same regression end to end from SQL: a multi-file (and therefore multi-partition) parquet scan on the build side of a `CollectLeft` join, with a `DISTINCT ON` aggregate on the probe side supplying the coalesce link that makes the traversal descend into the join. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
|
@zhuqi-lucas @2010YOUY01 would one of you mind taking a look at this change please? |
There was a problem hiding this comment.
Pull request overview
Note
Copilot couldn't run its full agentic review because it didn't start before the timeout. Make sure your repository has a runner available, or add a copilot-code-review.yml file specifying one with the runs-on attribute. See the docs for more details.
Fixes a regression in EnsureRequirements sort-parallelization where a required CoalescePartitionsExec (needed to satisfy Distribution::SinglePartition for CollectLeft joins) could be removed, leading to SanityCheckPlan failures.
Changes:
- Prevent
remove_bottleneck_in_subplanfrom removingCoalescePartitionsExecwhen the parent requiresSinglePartitionfor that child. - Add end-to-end sqllogictest coverage for the
CollectLeftjoin + sort-parallelization reproducer. - Add physical optimizer regression tests covering both
UnknownPartitioning(n)andHash([k], n)build-side shapes.
Reviewed changes
Copilot reviewed 3 out of 3 changed files in this pull request and generated 4 comments.
| File | Description |
|---|---|
| datafusion/physical-optimizer/src/ensure_requirements/enforce_sorting/mod.rs | Adds per-child distribution checks to avoid removing required coalesces during sort parallelization. |
| datafusion/sqllogictest/test_files/joins.slt | Adds an end-to-end regression test reproducing the CollectLeft + SinglePartition failure. |
| datafusion/core/tests/physical_optimizer/ensure_requirements.rs | Adds optimizer-level regression tests and snapshots to ensure required coalesces are retained. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| let removable = |idx: usize| { | ||
| is_root | ||
| || !matches!( | ||
| plan.input_distribution_requirements() | ||
| .child_distribution(idx), | ||
| Some(Distribution::SinglePartition) | ||
| ) | ||
| }; | ||
| let remove_from_first_child = | ||
| is_coalesce_partitions(&requirements.children[0].plan) && removable(0); |
| let removable = |idx: usize| { | ||
| is_root | ||
| || !matches!( | ||
| plan.input_distribution_requirements() | ||
| .child_distribution(idx), | ||
| Some(Distribution::SinglePartition) | ||
| ) | ||
| }; |
| assert_snapshot!(plan_string(&optimized), @r" | ||
| SortPreservingMergeExec: [a@0 ASC NULLS LAST] | ||
| SortExec: expr=[a@0 ASC NULLS LAST], preserve_partitioning=[true] | ||
| HashJoinExec: mode=CollectLeft, join_type=Left, on=[(a@0, a@0)] | ||
| CoalescePartitionsExec | ||
| MockMultiPartitionExec | ||
| RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=4 | ||
| MockMultiPartitionExec | ||
| "); |
| query I | ||
| COPY (SELECT * FROM collect_left_src) TO 'test_files/scratch/joins/collect_left/0.parquet' STORED AS PARQUET; | ||
| ---- | ||
| 3 |
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## main #23948 +/- ##
==========================================
- Coverage 80.69% 80.69% -0.01%
==========================================
Files 1095 1095
Lines 372626 372641 +15
Branches 372626 372641 +15
==========================================
- Hits 300700 300692 -8
- Misses 53978 53994 +16
- Partials 17948 17955 +7 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
| /// which `SanityCheckPlan` then rejected with "does not satisfy distribution requirements: | ||
| /// SinglePartition". | ||
| #[test] | ||
| fn test_collect_left_join_keeps_build_side_coalesce() { |
There was a problem hiding this comment.
I verified this locally and I'm afraid these two unit tests don't actually cover the regression: with the fix commit reverted (and cargo clean -p datafusion-physical-optimizer to force a rebuild), both tests still pass. I then instrumented the pre-fix children[0]-coalesce-removal branch with an eprintln! probe — it never fires for these mock plans, so they never reach the buggy path at all. The snapshots are identical pre/post fix.
(The sqllogictest is fine — on the reverted code it fails with exactly the reported does not satisfy distribution requirements: SinglePartition error, and passes with the fix.)
My guess at the cause: by the time parallelize_sorts runs, the distribution phase has already restructured the probe side, so the coalesce link that makes the traversal descend into the join in the real reproducer isn't there in this mock shape. Could you either make these construct the pre-phase-3a plan shape directly (so the removal branch demonstrably fires — an eprintln probe on main is a quick way to confirm), or drop them and let the sqllogictest carry the regression? As-is they'd stay green if the bug were reintroduced.
| let removable = |idx: usize| { | ||
| is_root | ||
| || !matches!( | ||
| plan.input_distribution_requirements() | ||
| .child_distribution(idx), | ||
| Some(Distribution::SinglePartition) | ||
| ) |
There was a problem hiding this comment.
Question (non-blocking): should this also protect Distribution::HashPartitioned children? A single-partition input trivially satisfies a hash requirement, so a coalesce sitting under a hash-requiring child is load-bearing in the same way — removing it exposes whatever multi-partitioning is below, which may not be hashed on the right keys. I don't think EnsureRequirements itself ever places a coalesce there (hash requirements get a RepartitionExec), so it's probably unreachable via this rule's own output — but a plan built by other rules or by hand could hit it. Fine to leave as-is if you agree it's unreachable; a one-line comment saying so would help the next reader.
| remove_bottleneck_in_subplan(node) | ||
| .enumerate() | ||
| .map(|(idx, node)| { | ||
| if node.data && removable(idx) { |
There was a problem hiding this comment.
Worth a short comment that this is deliberately conservative: skipping the descent entirely also skips legitimate cleanups below the protected child (e.g. a redundant second coalesce underneath the load-bearing one). Correctness-first is the right call for a fix — just flagging that the gate could later be narrowed to "descend, but protect only the top-level coalesce" if anyone cares about the missed optimization.
Which issue does this PR close?
Rationale for this change
A valid query can be planned into a physical plan that
SanityCheckPlanthen rejects:The
HashJoinExecis inCollectLeftmode, which requiresDistribution::SinglePartitionon its build (left) child, but child 0 is a bare 4-partitionDataSourceExecwith noCoalescePartitionsExecabove it.Self-contained reproducer with
datafusion-cli(the fourCOPYstatements are what make the scan multi-partition):Setting
datafusion.optimizer.repartition_sorts = falsemakes it plan fine, which points at the sort-parallelization phase.EnsureRequirementsdoes insert the coalesce for theSinglePartitionrequirement (enforce_distribution.rs,Distribution::SinglePartition => add_merge_on_top(...)). Its own phase 3a (parallelize_sorts) then takes it back out:remove_bottleneck_in_subplanremoves aCoalescePartitionsExecfound atchildren[0]positionally, without consulting the parent's distribution requirement for that child.That parent is reached because
update_coalesce_ctx_childrenmarks a node as connected when any child qualifies. It correctly excludes aSinglePartition-requiring child from setting the flag, but the join's other child (UnspecifiedDistribution, connected to a coalesce below) sets it, so the traversal descends into the join and rewrites child 0 anyway. Nothing re-enforces distribution afterwards, soSanityCheckPlanis the first thing to notice. Note the survivingCoalescePartitionsExecon the probe side in the plan above: it is what propagated the flag, and it is untouched because theifreturns without recursing into child 1.The sibling helper on the phase 2b path already does consult the requirement (
update_child_to_remove_unnecessary_sort/remove_corresponding_sort_from_sub_planre-add a merge using the per-childchild_distribution(child_idx)); only this path is missing it.The same failure shows up with a build child that is already hash-partitioned on the join key (
Child-0 output partitioning: Hash([k@0], 8)), which is what aJoinSelectioninput swap leaves behind — aCollectLeftjoin reported asjoin_type=Rightwith an embedded projection.What changes are included in this PR?
remove_bottleneck_in_subplannow checks the parent's per-child distribution requirement before removing a coalesce, both forchildren[0]and when recursing into the other children.The node
parallelize_sortsis itself rewriting (the root of the call) is exempt, since the caller drops that node and rebuilds the sort cascade around the result — that is the rule's intended transformation, and gating it too would disable sort parallelization below a global sort. This is threaded through as anis_rootflag on a private_implfunction; the public entry point keeps its signature.Are these changes tested?
Yes, at two levels:
datafusion/sqllogictest/test_files/joins.sltreproducing it from SQL (the reproducer above, with the data written byCOPYinside the test). Onmainit fails with exactly the distribution error above.datafusion/core/tests/physical_optimizer/ensure_requirements.rscovering both shapes of the build child (UnknownPartitioning(n)andHash([k], n)), running the fullEnsureRequirementsrule and thenSanityCheckPlanvia the existingoptimize_and_sanity_checkhelper, plus the idempotency check.cargo test -p datafusion-physical-optimizer,cargo test -p datafusion --test core_integration -- physical_optimizer(530 tests) and the fullsqllogictestsuite (498 files) pass.Are there any user-facing changes?
No API changes. Plans that were previously rejected by
SanityCheckPlannow plan and execute; a coalesce that is genuinely required is retained where it was previously (incorrectly) removed.