Uh oh!
There was an error while loading. Please reload this page.
fix: keep a CoalescePartitionsExec required by a SinglePartition child - #23948
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>
adriangb
commented
Jul 28, 2026
@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.85% +0.16%
==========================================
Files 1095 1099 +4 Lines 372626 374345 +1719 Branches 372626 374345 +1719 ==========================================
+ Hits 300700 302695 +1995 + Misses 53978 53605 -373 - Partials 17948 18045 +97 ☔ 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.
There was a problem hiding this comment.
Yep — confirmed, and your hypothesis was exactly right. Phase 2a drops the probe-side coalesce in the mock shape, so the join never gets marked as connected and the walk never descends into it.
Rebuilt in 8fb3d30: the tests now construct the pre-phase-3a shape directly (probe-side coalesce intact) and drive parallelize_sorts via transform_up, then check with SanityCheckPlan. Both fail with the reported SinglePartition error when the fix is reverted.
| 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.
There was a problem hiding this comment.
Agreed on both counts — same hazard in principle, but ensure_distribution satisfies a hash requirement with a RepartitionExec, never a coalesce, so widening the match would be dead code. Left as-is with a comment saying so.
| 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.
There was a problem hiding this comment.
Yes, that's the intended trade — the missed cleanup is a redundant coalesce under a load-bearing one, which I'd rather leave than risk narrowing wrong in a fix. Added a comment recording that, with "descend, but protect only the topmost coalesce" as the follow-up.
- Hoist `input_distribution_requirements()` out of the `removable` closure so it is computed once per node instead of once per child check. - Use `children.first()` instead of indexing, so the traversal is robust if it is ever reached on a node with no children. - Document why only `SinglePartition` is protected (a `HashPartitioned` child is in the same position in principle, but `ensure_distribution` never puts a coalesce there) and why skipping the descent entirely is deliberate. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
…oval The two optimizer-level tests went through the full `EnsureRequirements` rule from a freshly built plan. By the time phase 3a ran, the earlier phases had already dropped the probe-side `CoalescePartitionsExec`, so the join was never marked as connected and `remove_bottleneck_in_subplan` never descended into it. Both tests passed with the fix reverted. Build the pre-phase-3a plan shape directly instead — the shape the reproducer actually produces, with a coalesce on both the build and probe sides of the join — and drive `parallelize_sorts` on its own, then check the result with `SanityCheckPlan`. Both tests now fail with the fix reverted, with the same "does not satisfy distribution requirements: SinglePartition" error the PR reports. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Uh oh!
There was an error while loading. Please reload this page.
apache#23948) ## Which issue does this PR close? - None filed; happy to open one if preferred. ## Rationale for this change A valid query can be planned into a physical plan that `SanityCheckPlan` then rejects: ``` SanityCheckPlan caused by Error during planning: Plan: ["HashJoinExec: mode=CollectLeft, join_type=Left, on=[(id@0, id@0)], projection=[id@0]", " DataSourceExec: file_groups={4 groups: [...]}, projection=[id], file_type=parquet", " RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1", " CoalescePartitionsExec", " ProjectionExec: expr=[first_value(t.id) ORDER BY [...]@1 as id]", " AggregateExec: mode=FinalPartitioned, gby=[id@0 as id], aggr=[first_value(t.id) ORDER BY [...]]", " RepartitionExec: partitioning=Hash([id@0], 8), input_partitions=4", " AggregateExec: mode=Partial, gby=[id@1 as id], aggr=[first_value(t.id) ORDER BY [...]]", " DataSourceExec: file_groups={4 groups: [...]}, projection=[ts, id], file_type=parquet"] does not satisfy distribution requirements: SinglePartition. Child-0 output partitioning: UnknownPartitioning(4) ``` The `HashJoinExec` is in `CollectLeft` mode, which requires `Distribution::SinglePartition` on its build (left) child, but child 0 is a bare 4-partition `DataSourceExec` with no `CoalescePartitionsExec` above it. Self-contained reproducer with `datafusion-cli` (the four `COPY` statements are what make the scan multi-partition): ```sql set datafusion.execution.target_partitions = 8; set datafusion.optimizer.repartition_file_scans = false; create table src (id int, ts int) as values (1, 10), (2, 20), (3, 30); copy (select * from src) to 'data/0.parquet' stored as parquet; copy (select * from src) to 'data/1.parquet' stored as parquet; copy (select * from src) to 'data/2.parquet' stored as parquet; copy (select * from src) to 'data/3.parquet' stored as parquet; create external table t stored as parquet location 'data/'; select a.id from t a left join (select distinct on (id) id, ts from t order by id, ts) f on a.id = f.id order by a.id; ``` Setting `datafusion.optimizer.repartition_sorts = false` makes it plan fine, which points at the sort-parallelization phase. `EnsureRequirements` does insert the coalesce for the `SinglePartition` requirement (`enforce_distribution.rs`, `Distribution::SinglePartition => add_merge_on_top(...)`). Its own phase 3a (`parallelize_sorts`) then takes it back out: `remove_bottleneck_in_subplan` removes a `CoalescePartitionsExec` found at `children[0]` positionally, without consulting the parent's distribution requirement for that child. That parent is reached because `update_coalesce_ctx_children` marks a node as connected when *any* child qualifies. It correctly excludes a `SinglePartition`-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, so `SanityCheckPlan` is the first thing to notice. Note the surviving `CoalescePartitionsExec` on the probe side in the plan above: it is what propagated the flag, and it is untouched because the `if` returns 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_plan` re-add a merge using the per-child `child_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 a `JoinSelection` input swap leaves behind — a `CollectLeft` join reported as `join_type=Right` with an embedded projection. ## What changes are included in this PR? `remove_bottleneck_in_subplan` now checks the parent's per-child distribution requirement before removing a coalesce, both for `children[0]` and when recursing into the other children. The node `parallelize_sorts` is 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 an `is_root` flag on a private `_impl` function; the public entry point keeps its signature. ## Are these changes tested? Yes, at two levels: - An end-to-end sqllogictest in `datafusion/sqllogictest/test_files/joins.slt` reproducing it from SQL (the reproducer above, with the data written by `COPY` inside the test). On `main` it fails with exactly the distribution error above. - Two tests in `datafusion/core/tests/physical_optimizer/ensure_requirements.rs` covering both shapes of the build child (`UnknownPartitioning(n)` and `Hash([k], n)`), running the full `EnsureRequirements` rule and then `SanityCheckPlan` via the existing `optimize_and_sanity_check` helper, plus the idempotency check. `cargo test -p datafusion-physical-optimizer`, `cargo test -p datafusion --test core_integration -- physical_optimizer` (530 tests) and the full `sqllogictest` suite (498 files) pass. ## Are there any user-facing changes? No API changes. Plans that were previously rejected by `SanityCheckPlan` now plan and execute; a coalesce that is genuinely required is retained where it was previously (incorrectly) removed. --------- Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
apache#23948) - None filed; happy to open one if preferred. A valid query can be planned into a physical plan that `SanityCheckPlan` then rejects: ``` SanityCheckPlan caused by Error during planning: Plan: ["HashJoinExec: mode=CollectLeft, join_type=Left, on=[(id@0, id@0)], projection=[id@0]", " DataSourceExec: file_groups={4 groups: [...]}, projection=[id], file_type=parquet", " RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1", " CoalescePartitionsExec", " ProjectionExec: expr=[first_value(t.id) ORDER BY [...]@1 as id]", " AggregateExec: mode=FinalPartitioned, gby=[id@0 as id], aggr=[first_value(t.id) ORDER BY [...]]", " RepartitionExec: partitioning=Hash([id@0], 8), input_partitions=4", " AggregateExec: mode=Partial, gby=[id@1 as id], aggr=[first_value(t.id) ORDER BY [...]]", " DataSourceExec: file_groups={4 groups: [...]}, projection=[ts, id], file_type=parquet"] does not satisfy distribution requirements: SinglePartition. Child-0 output partitioning: UnknownPartitioning(4) ``` The `HashJoinExec` is in `CollectLeft` mode, which requires `Distribution::SinglePartition` on its build (left) child, but child 0 is a bare 4-partition `DataSourceExec` with no `CoalescePartitionsExec` above it. Self-contained reproducer with `datafusion-cli` (the four `COPY` statements are what make the scan multi-partition): ```sql set datafusion.execution.target_partitions = 8; set datafusion.optimizer.repartition_file_scans = false; create table src (id int, ts int) as values (1, 10), (2, 20), (3, 30); copy (select * from src) to 'data/0.parquet' stored as parquet; copy (select * from src) to 'data/1.parquet' stored as parquet; copy (select * from src) to 'data/2.parquet' stored as parquet; copy (select * from src) to 'data/3.parquet' stored as parquet; create external table t stored as parquet location 'data/'; select a.id from t a left join (select distinct on (id) id, ts from t order by id, ts) f on a.id = f.id order by a.id; ``` Setting `datafusion.optimizer.repartition_sorts = false` makes it plan fine, which points at the sort-parallelization phase. `EnsureRequirements` does insert the coalesce for the `SinglePartition` requirement (`enforce_distribution.rs`, `Distribution::SinglePartition => add_merge_on_top(...)`). Its own phase 3a (`parallelize_sorts`) then takes it back out: `remove_bottleneck_in_subplan` removes a `CoalescePartitionsExec` found at `children[0]` positionally, without consulting the parent's distribution requirement for that child. That parent is reached because `update_coalesce_ctx_children` marks a node as connected when *any* child qualifies. It correctly excludes a `SinglePartition`-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, so `SanityCheckPlan` is the first thing to notice. Note the surviving `CoalescePartitionsExec` on the probe side in the plan above: it is what propagated the flag, and it is untouched because the `if` returns 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_plan` re-add a merge using the per-child `child_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 a `JoinSelection` input swap leaves behind — a `CollectLeft` join reported as `join_type=Right` with an embedded projection. `remove_bottleneck_in_subplan` now checks the parent's per-child distribution requirement before removing a coalesce, both for `children[0]` and when recursing into the other children. The node `parallelize_sorts` is 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 an `is_root` flag on a private `_impl` function; the public entry point keeps its signature. Yes, at two levels: - An end-to-end sqllogictest in `datafusion/sqllogictest/test_files/joins.slt` reproducing it from SQL (the reproducer above, with the data written by `COPY` inside the test). On `main` it fails with exactly the distribution error above. - Two tests in `datafusion/core/tests/physical_optimizer/ensure_requirements.rs` covering both shapes of the build child (`UnknownPartitioning(n)` and `Hash([k], n)`), running the full `EnsureRequirements` rule and then `SanityCheckPlan` via the existing `optimize_and_sanity_check` helper, plus the idempotency check. `cargo test -p datafusion-physical-optimizer`, `cargo test -p datafusion --test core_integration -- physical_optimizer` (530 tests) and the full `sqllogictest` suite (498 files) pass. No API changes. Plans that were previously rejected by `SanityCheckPlan` now plan and execute; a coalesce that is genuinely required is retained where it was previously (incorrectly) removed. --------- Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
apache#23948) - None filed; happy to open one if preferred. A valid query can be planned into a physical plan that `SanityCheckPlan` then rejects: ``` SanityCheckPlan caused by Error during planning: Plan: ["HashJoinExec: mode=CollectLeft, join_type=Left, on=[(id@0, id@0)], projection=[id@0]", " DataSourceExec: file_groups={4 groups: [...]}, projection=[id], file_type=parquet", " RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1", " CoalescePartitionsExec", " ProjectionExec: expr=[first_value(t.id) ORDER BY [...]@1 as id]", " AggregateExec: mode=FinalPartitioned, gby=[id@0 as id], aggr=[first_value(t.id) ORDER BY [...]]", " RepartitionExec: partitioning=Hash([id@0], 8), input_partitions=4", " AggregateExec: mode=Partial, gby=[id@1 as id], aggr=[first_value(t.id) ORDER BY [...]]", " DataSourceExec: file_groups={4 groups: [...]}, projection=[ts, id], file_type=parquet"] does not satisfy distribution requirements: SinglePartition. Child-0 output partitioning: UnknownPartitioning(4) ``` The `HashJoinExec` is in `CollectLeft` mode, which requires `Distribution::SinglePartition` on its build (left) child, but child 0 is a bare 4-partition `DataSourceExec` with no `CoalescePartitionsExec` above it. Self-contained reproducer with `datafusion-cli` (the four `COPY` statements are what make the scan multi-partition): ```sql set datafusion.execution.target_partitions = 8; set datafusion.optimizer.repartition_file_scans = false; create table src (id int, ts int) as values (1, 10), (2, 20), (3, 30); copy (select * from src) to 'data/0.parquet' stored as parquet; copy (select * from src) to 'data/1.parquet' stored as parquet; copy (select * from src) to 'data/2.parquet' stored as parquet; copy (select * from src) to 'data/3.parquet' stored as parquet; create external table t stored as parquet location 'data/'; select a.id from t a left join (select distinct on (id) id, ts from t order by id, ts) f on a.id = f.id order by a.id; ``` Setting `datafusion.optimizer.repartition_sorts = false` makes it plan fine, which points at the sort-parallelization phase. `EnsureRequirements` does insert the coalesce for the `SinglePartition` requirement (`enforce_distribution.rs`, `Distribution::SinglePartition => add_merge_on_top(...)`). Its own phase 3a (`parallelize_sorts`) then takes it back out: `remove_bottleneck_in_subplan` removes a `CoalescePartitionsExec` found at `children[0]` positionally, without consulting the parent's distribution requirement for that child. That parent is reached because `update_coalesce_ctx_children` marks a node as connected when *any* child qualifies. It correctly excludes a `SinglePartition`-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, so `SanityCheckPlan` is the first thing to notice. Note the surviving `CoalescePartitionsExec` on the probe side in the plan above: it is what propagated the flag, and it is untouched because the `if` returns 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_plan` re-add a merge using the per-child `child_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 a `JoinSelection` input swap leaves behind — a `CollectLeft` join reported as `join_type=Right` with an embedded projection. `remove_bottleneck_in_subplan` now checks the parent's per-child distribution requirement before removing a coalesce, both for `children[0]` and when recursing into the other children. The node `parallelize_sorts` is 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 an `is_root` flag on a private `_impl` function; the public entry point keeps its signature. Yes, at two levels: - An end-to-end sqllogictest in `datafusion/sqllogictest/test_files/joins.slt` reproducing it from SQL (the reproducer above, with the data written by `COPY` inside the test). On `main` it fails with exactly the distribution error above. - Two tests in `datafusion/core/tests/physical_optimizer/ensure_requirements.rs` covering both shapes of the build child (`UnknownPartitioning(n)` and `Hash([k], n)`), running the full `EnsureRequirements` rule and then `SanityCheckPlan` via the existing `optimize_and_sanity_check` helper, plus the idempotency check. `cargo test -p datafusion-physical-optimizer`, `cargo test -p datafusion --test core_integration -- physical_optimizer` (530 tests) and the full `sqllogictest` suite (498 files) pass. No API changes. Plans that were previously rejected by `SanityCheckPlan` now plan and execute; a coalesce that is genuinely required is retained where it was previously (incorrectly) removed. --------- Co-authored-by: Claude Opus 5 <noreply@anthropic.com> (cherry picked from commit 6636d6b)
…Partition child (#169) * Revert "fix(lambda): only push referenced params into the merged batch (apache#24162) (#166)" This reverts commit 79de5e9. * fix: keep a CoalescePartitionsExec required by a SinglePartition child (apache#23948) - None filed; happy to open one if preferred. A valid query can be planned into a physical plan that `SanityCheckPlan` then rejects: ``` SanityCheckPlan caused by Error during planning: Plan: ["HashJoinExec: mode=CollectLeft, join_type=Left, on=[(id@0, id@0)], projection=[id@0]", " DataSourceExec: file_groups={4 groups: [...]}, projection=[id], file_type=parquet", " RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1", " CoalescePartitionsExec", " ProjectionExec: expr=[first_value(t.id) ORDER BY [...]@1 as id]", " AggregateExec: mode=FinalPartitioned, gby=[id@0 as id], aggr=[first_value(t.id) ORDER BY [...]]", " RepartitionExec: partitioning=Hash([id@0], 8), input_partitions=4", " AggregateExec: mode=Partial, gby=[id@1 as id], aggr=[first_value(t.id) ORDER BY [...]]", " DataSourceExec: file_groups={4 groups: [...]}, projection=[ts, id], file_type=parquet"] does not satisfy distribution requirements: SinglePartition. Child-0 output partitioning: UnknownPartitioning(4) ``` The `HashJoinExec` is in `CollectLeft` mode, which requires `Distribution::SinglePartition` on its build (left) child, but child 0 is a bare 4-partition `DataSourceExec` with no `CoalescePartitionsExec` above it. Self-contained reproducer with `datafusion-cli` (the four `COPY` statements are what make the scan multi-partition): ```sql set datafusion.execution.target_partitions = 8; set datafusion.optimizer.repartition_file_scans = false; create table src (id int, ts int) as values (1, 10), (2, 20), (3, 30); copy (select * from src) to 'data/0.parquet' stored as parquet; copy (select * from src) to 'data/1.parquet' stored as parquet; copy (select * from src) to 'data/2.parquet' stored as parquet; copy (select * from src) to 'data/3.parquet' stored as parquet; create external table t stored as parquet location 'data/'; select a.id from t a left join (select distinct on (id) id, ts from t order by id, ts) f on a.id = f.id order by a.id; ``` Setting `datafusion.optimizer.repartition_sorts = false` makes it plan fine, which points at the sort-parallelization phase. `EnsureRequirements` does insert the coalesce for the `SinglePartition` requirement (`enforce_distribution.rs`, `Distribution::SinglePartition => add_merge_on_top(...)`). Its own phase 3a (`parallelize_sorts`) then takes it back out: `remove_bottleneck_in_subplan` removes a `CoalescePartitionsExec` found at `children[0]` positionally, without consulting the parent's distribution requirement for that child. That parent is reached because `update_coalesce_ctx_children` marks a node as connected when *any* child qualifies. It correctly excludes a `SinglePartition`-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, so `SanityCheckPlan` is the first thing to notice. Note the surviving `CoalescePartitionsExec` on the probe side in the plan above: it is what propagated the flag, and it is untouched because the `if` returns 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_plan` re-add a merge using the per-child `child_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 a `JoinSelection` input swap leaves behind — a `CollectLeft` join reported as `join_type=Right` with an embedded projection. `remove_bottleneck_in_subplan` now checks the parent's per-child distribution requirement before removing a coalesce, both for `children[0]` and when recursing into the other children. The node `parallelize_sorts` is 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 an `is_root` flag on a private `_impl` function; the public entry point keeps its signature. Yes, at two levels: - An end-to-end sqllogictest in `datafusion/sqllogictest/test_files/joins.slt` reproducing it from SQL (the reproducer above, with the data written by `COPY` inside the test). On `main` it fails with exactly the distribution error above. - Two tests in `datafusion/core/tests/physical_optimizer/ensure_requirements.rs` covering both shapes of the build child (`UnknownPartitioning(n)` and `Hash([k], n)`), running the full `EnsureRequirements` rule and then `SanityCheckPlan` via the existing `optimize_and_sanity_check` helper, plus the idempotency check. `cargo test -p datafusion-physical-optimizer`, `cargo test -p datafusion --test core_integration -- physical_optimizer` (530 tests) and the full `sqllogictest` suite (498 files) pass. No API changes. Plans that were previously rejected by `SanityCheckPlan` now plan and execute; a coalesce that is genuinely required is retained where it was previously (incorrectly) removed. --------- Co-authored-by: Claude Opus 5 <noreply@anthropic.com> * fix(lambda): only push referenced params into the merged batch (apache#24162) (#166) * fix(lambda): only push referenced params into the merged batch (apache#24162) ## Which issue does this PR close? basically this PR apache#22853 + a few more tests ## Rationale for this change The current lambdas in DF only take a single parameter `(v -> ...)`, so nobody had noticed that `LambdaExpr` mishandles lambdas with more than one parameter. The bug surfaced while working on `transform_values` (apache#22689), which needs `(k, v) -> expr ` two parameters, one of which is very often unused (e.g. `(k, v) -> v * 2`, k never referenced). The bug is that when a higher order function with more than 1 param evaluates a lambda, it fills each parameter into a slot based on its declared position — for example for `(k, v) -> v` `k` always goes into slot 0, `v` always into slot 1. `LambdaExpr` separately scans the body and renumbers whatever it finds referenced into a dense `0..n` range, to avoid carrying around columns nothing uses (like `v` in this case). That renumbering is fine for outer captures, but applying it to the lambda's own parameters is wrong, because it changes where the body looks for a value without changing where the evaluator put it. ### Example: in `(k, v) -> v` `v` is declared second (slot 1), but since it's the only parameter the body references, the renumbering logic reassigns it to slot 0. The evaluator, unaware of this, writes `k`'s values into slot 0 and `v`'s into slot 1. So the body ends up reading slot 0 expecting `v` — and gets `k` instead. So the results end up being incorrect. ## What changes are included in this PR? - `LambdaExpr` now computes `used_params`: which is the subset of its own declared parameters that are actually referenced in the body. - `LambdaArgument::new` takes `used_params` and only pushes the referenced parameters in the body into the merged batch, in original declaration order — so the body's indices always line up with what's actually built. - `HigherOrderFunctionExpr::evaluate` forwards `lambda.used_params()` to `LambdaArgument::new` ## Are these changes tested? yes, added two new tests one for the unused-parameter case and nested-lambda for the shadowing case. ## Are there any user-facing changes? The only public api change is on `LambdaArgument::new ` which now requires a new argument: `used_params: &HashSet<String>`, however LambdaArgument::new is very unlikely to be called outside datafusion, see [this](apache#22853 (comment)) comment (cherry picked from commit 4e6acfe) * Adjust to API change --------- Co-authored-by: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Co-authored-by: Claude Opus 5 <noreply@anthropic.com> Co-authored-by: Lía Adriana <lia.castaneda@datadoghq.com>
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.