peter-toth opened a new pull request, #58633:
URL: https://github.com/apache/spark/pull/58633
### What changes were proposed in this pull request?
This backports #58543 to `branch-4.2`. The merge script carried the fix to
`master`, `branch-4.x` and `branch-4.3`, and stopped there.
`GroupPartitionsExec.canUseSortedMerge` is replaced by two members, and the
config read moves into the planner's own method.
- `kWayMergeIsFeasible` holds the two live terms, the child having an
ordering and the child subtree being `SafeForKWayMerge`.
- `usesSortedMerge` is what `doExecute`, `supportsColumnar` and
`outputOrdering` ask, and it carries no config term.
- `tryEnableSortedMerge` reads the config itself, so it appears once in the
file, at the only place that decides anything with it. It reads it through
`conf` rather than `SQLConf.get`.
The `enableSortedMerge` scaladoc now states the contract, that the flag is
the decision rather than a hint. EXPLAIN shows the flag too, since it is now
the only thing that says whether a node k-way merges.
### What was tailored for this branch?
Nothing in the change itself. The diff is byte-identical to the merged
commit, verified hunk by hunk against `6a23cb61f69`.
The cherry-pick still conflicted, on `KeyGroupedPartitioningSuite` alone.
That file has diverged a long way from `master` on this branch, so git could
not place the three test hunks and produced one tail-of-file conflict. They
were re-applied by hand at the matching places, and nothing was dropped or
adapted: the fixture helper, the `SPARK-56549` refactor, the new test and the
two EXPLAIN keyword strings are all as they merged. `GroupPartitionsExec.scala`
and `GroupPartitionsExecSuite.scala` applied cleanly.
### Why are the changes needed?
A sort-merge join silently drops rows when
`spark.sql.sources.v2.bucketing.preserveOrderingOnCoalesce.enabled` is turned
off between planning and execution.
Both tables identity-partitioned on the join key, both reporting a
two-column ordering, two splits per key so `GroupPartitionsExec` coalesces:
val df = sql("SELECT /*+ MERGE(i, p) */ i.id, i.name FROM items i JOIN
purchases p " +
"ON p.item_id = i.id AND p.time = i.arrive_time")
df.queryExecution.executedPlan // planned with the config on, no
SortExec below the join
spark.conf.set("spark.sql.sources.v2.bucketing.preserveOrderingOnCoalesce.enabled",
"false")
df.collect() // 3 rows instead of 5
At planning, `tryEnableSortedMerge()` finds the config on and returns
`copy(enableSortedMerge = true)`. That copy reports the child's full ordering,
so `EnsureRequirements` adds no `SortExec` under the join. The copy is a new
instance, so its own `canUseSortedMerge` is still unevaluated. At execution
`doExecute` forces it, now under the new config value, and builds a plain
`CoalescedRDD` instead. The concatenated partitions are no longer sorted, and
the sort-merge join walks them as if they were.
`outputOrdering` read the same member, so the node and the plan above it
ended up disagreeing about what the node delivers.
`supportsColumnar` read it too, which is a second route to the same lost
rows. Under AQE, `ApplyColumnarRulesAndInsertTransitions` runs when the result
stage is created, so it sees the flipped config. A columnar child would then
make `supportsColumnar` true and route to `doExecuteColumnar`, which only ever
builds a plain `CoalescedRDD`, while the join above had already been planned
against the merged ordering. That route is closed by the same change and is not
covered by a test, because the suite has no columnar V2 source.
**This branch was one of the branches the bug was measured on**, before the
fix was written: the same 3 rows of 5, with the same two rows lost.
`branch-4.1` has no `GroupPartitionsExec`, so the path does not exist there.
`enableSortedMerge` is already the record of the planner's decision, and
only `tryEnableSortedMerge` sets it, after checking the config. So the
execution side does not need the config at all.
`childIsSafeForKWayMerge` does have to stay live.
`ApplyColumnarRulesAndInsertTransitions` and `CollapseCodegenStages` insert
nodes under the child after `EnsureRequirements` ran. Those wrappers are all
whitelisted, which is what keeps the answer stable, and the live check is the
fallback if something outside the whitelist ever appears. Splitting the terms
apart is what lets the config term go while that guard stays.
Two alternatives were rejected, a config snapshot in a `private val` at
construction and dropping the re-check outright, and the `conf` versus
`SQLConf.get` question was measured rather than assumed. The reasoning is on
#58543 and is unchanged here.
`outputOrdering`'s other config, `preserveKeyOrderingOnCoalesce`, is
deliberately still a live read, and the file now says why. It gates whether to
*report* an ordering that holds either way, so a late read only makes the node
claim less than it delivers. The sorted-merge config gates whether the merge
*happens*.
### Does this PR introduce _any_ user-facing change?
Yes, it fixes wrong results. A query whose plan was built with the config on
keeps the k-way merge and returns the right rows when the config is turned off
before it runs.
One consequence worth stating. The config is documented as a cost knob, and
turning it off no longer stops a merge in a plan that is already built. That
includes an `InMemoryRelation`'s cached plan, which can outlive many config
changes in a session. Re-planning is what picks the new value up.
The EXPLAIN string of `GroupPartitions` gains a `SortedMerge` field. On this
branch that is a new node with no golden file behind it, and the line already
carries an always-printed `DistributePartitions`, so this adds one more field
to a shape that is itself new in 4.2. dongjoon-hyun asked for the field to stay
always-printed in the backports too, for exactly that reason.
### How was this patch tested?
The same tests as the original, run on this branch.
The new integration test in `KeyGroupedPartitioningSuite`, "SPARK-59279: a
planned k-way merge survives a later config change", forces the plan with the
config on, asserts there is no `SortExec` below the sort-merge join, then turns
the config off and executes the same `DataFrame`. It covers both AQE modes,
because they freeze the plan at different points.
In `GroupPartitionsExecSuite`, "SPARK-55715: coalescing with
enableSortedMerge = true returns full child ordering" drops a
`withSQLConf(preserveOrderingOnCoalesce -> true)` wrapper that is inert now, so
the assertion runs at the config's default of `false`. A new "SPARK-59279:
enableSortedMerge decides the k-way merge, not the config" states the invariant
directly, as a grid over the config with and without the flag.
"SPARK-55992: GroupPartitions string in simple and extended explain" now
expects the `SortedMerge` field.
The new integration test's fixture was byte-identical to the two tests above
it, so it is extracted into `createOrderedIdTables` plus `orderedIdJoinRows`,
next to the suite's other fixture helpers. "SPARK-56549: k-way merge enabled
only when parent requires ordering" now uses it too.
`KeyGroupedPartitioningSuite`, `GroupPartitionsExecSuite`,
`EnsureRequirementsSuite`, `SortedMergeCoalescedRDDSuite`, `PlannerSuite` and
`ProjectedOrderingAndPartitioningSuite` are green on this branch, and the three
new or changed assertions were run against the unfixed branch to confirm they
fail there.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 5)
--
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]
