[SPARK-59279][SQL] Don't re-read the sorted-merge config after GroupPartitionsExec is planned by peter-toth · Pull Request #58543 · apache/spark

peter-toth · 2026-09-05T10:40:44Z

What changes were proposed in this pull request?

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.

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.

Measured on branch-4.3 and branch-4.2 as well, same 3 rows of 5 in both, 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. Snapshotting the config into a private val at construction, the way SortExec does with enableRadixSort, fixes the same case, but SparkPlan.conf is the live conf and a val freezes at construction rather than at the decision, so the window reopens for any node copied after EnsureRequirements ran. Dropping the && canUseSortedMerge re-check outright would take childIsSafeForKWayMerge with it.

Reading conf instead of SQLConf.get is not what fixes this, and that was measured rather than assumed. SparkPlan.conf is session.sessionState.conf, the session's live mutable SQLConf, which is the same object spark.conf.set and withSQLConf mutate. The read does switch to conf here, on separate grounds. It now happens only on the driver during planning, so the node's own session conf is the right one to ask, and SparkPlan.conf falls back to SQLConf.get when the node has no session. It also makes the file consistent, since outputOrdering already read conf.

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.

How was this patch tested?

A new test in KeyGroupedPartitioningSuite, "SPARK-59279: a planned k-way merge survives a later config change". It 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. AQE builds it in AdaptiveSparkPlanExec.initialPlan, and without AQE prepareForExecution does. Without the fix both modes return 3 of the 5 rows, dropping [1,aa] and [2,cc].

One existing test changed, plus a new unit test beside it. GroupPartitionsExecSuite's "SPARK-55715: coalescing with enableSortedMerge = true returns full child ordering" wrapped its flag assertion in withSQLConf(preserveOrderingOnCoalesce -> true). That wrapper is inert now, because outputOrdering no longer reads that config, so it is dropped and 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. The sibling test's name lost the words "sorted merge config enabled", for the same reason. All three fail without the fix.

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. The third copy, under "SPARK-55715", is left alone to keep this diff small.

"SPARK-55992: GroupPartitions string in simple and extended explain" now expects the SortedMerge field.

KeyGroupedPartitioningSuite, GroupPartitionsExecSuite, EnsureRequirementsSuite, SortedMergeCoalescedRDDSuite, PlannerSuite and ProjectedOrderingAndPartitioningSuite, 313 tests.

Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude Code (Opus 5)

peter-toth · 2026-09-07T06:41:56Z

cc @dongjoon-hyun , @ulysses-you

…artitionsExec is planned ### What changes were proposed in this pull request? `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. ### 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. Measured on `branch-4.3` and `branch-4.2` as well, same 3 rows of 5 in both, 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. Snapshotting the config into a `private val` at construction, the way `SortExec` does with `enableRadixSort`, fixes the same case, but `SparkPlan.conf` is the live conf and a `val` freezes at construction rather than at the decision, so the window reopens for any node copied after `EnsureRequirements` ran. Dropping the `&& canUseSortedMerge` re-check outright would take `childIsSafeForKWayMerge` with it. Reading `conf` instead of `SQLConf.get` is not what fixes this, and that was measured rather than assumed. `SparkPlan.conf` is `session.sessionState.conf`, the session's live mutable `SQLConf`, which is the same object `spark.conf.set` and `withSQLConf` mutate. The read does switch to `conf` here, on separate grounds. It now happens only on the driver during planning, so the node's own session conf is the right one to ask, and `SparkPlan.conf` falls back to `SQLConf.get` when the node has no session. It also makes the file consistent, since `outputOrdering` already read `conf`. `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. ### How was this patch tested? A new test in `KeyGroupedPartitioningSuite`, "SPARK-59279: a planned k-way merge survives a later config change". It 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. AQE builds it in `AdaptiveSparkPlanExec.initialPlan`, and without AQE `prepareForExecution` does. Without the fix both modes return 3 of the 5 rows, dropping `[1,aa]` and `[2,cc]`. One existing test changed, plus a new unit test beside it. `GroupPartitionsExecSuite`'s "SPARK-55715: coalescing with enableSortedMerge = true returns full child ordering" wrapped its flag assertion in `withSQLConf(preserveOrderingOnCoalesce -> true)`. That wrapper is inert now, because `outputOrdering` no longer reads that config, so it is dropped and 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. The sibling test's name lost the words "sorted merge config enabled", for the same reason. All three fail without the fix. 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. The third copy, under "SPARK-55715", is left alone to keep this diff small. "SPARK-55992: GroupPartitions string in simple and extended explain" now expects the `SortedMerge` field. `KeyGroupedPartitioningSuite`, `GroupPartitionsExecSuite`, `EnsureRequirementsSuite`, `SortedMergeCoalescedRDDSuite`, `PlannerSuite` and `ProjectedOrderingAndPartitioningSuite`, 313 tests. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (Opus 5)

dongjoon-hyun

Thank you for the detailed analysis, @peter-toth. The root cause is right: canUseSortedMerge is a lazy val on a node that EnsureRequirements copies and CollapseCodegenStages re-mints, so it is only forced at execution under whatever the config is by then, while the join above was already planned against the merged ordering. Moving the config read into tryEnableSortedMerge and making enableSortedMerge the sole decision is the right fix, and keeping childIsSafeForKWayMerge live is the right call too.

The new test forces the plan first and flips the config afterwards in both AQE modes, and the fixture extraction is a nice cleanup. I left two minor comments inline. Since this is a wrong-results fix, I agree it should go to branch-4.3 and branch-4.2 as well.

dongjoon-hyun · 2026-09-07T19:41:58Z

+      outputPartitioning = KeyedPartitioning(Seq(exprA), partitionKeys),
+      outputOrdering = childOrdering)
+
+    Seq(true, false).foreach { configEnabled =>


The PR description says this is "a grid over the config with and without the flag", but the test only iterates the config with enableSortedMerge = true. The assertion this PR removed from the test above ("config alone should not enable k-way merge") is no longer checked for config = true. How about adding the enableSortedMerge = false case here, asserting that GroupPartitionsExec(child).outputOrdering !== childOrdering regardless of the config? Then the test states exactly what its name says.

You are right, the grid was half a grid. Fixed in 6a138ec.

The unflagged case now runs under both config values, so the assertion this PR removed from the test above is back and covers more than it did: it used to check config = true only.

val unflagged = GroupPartitionsExec(child) assert(unflagged.outputOrdering !== childOrdering, s"config=$configEnabled: without the flag there is no k-way merge to report, and the " + "config cannot supply one")

dongjoon-hyun · 2026-09-07T19:41:58Z

+    // Rendered from the constructor field, as `DistributePartitions` above is. Not from
+    // `usesSortedMerge`, because that forces `grouping`, which can throw, and this method feeds
+    // `simpleString`, which `treeString` calls on error paths.
+    val sortedMergeStr = Iterator(s"SortedMerge: $enableSortedMerge")


Just noting for the record: this adds SortedMerge: false to every GroupPartitions line in EXPLAIN, even for nodes that do not coalesce. It is consistent with DistributePartitions, and no golden file contains GroupPartitions, so nothing else breaks. But since this will be backported to branch-4.3 and branch-4.2, the EXPLAIN string changes in maintenance releases too. I am fine with it, just want it to be a conscious choice.

Conscious, and thank you for pinning it down rather than letting it pass.

Two reasons I would keep it. The flag is now the only thing in the plan that says whether the node k-way merges, since the config no longer decides at execution, so a reader of an EXPLAIN has nowhere else to look. And that is exactly the failure this PR is about: the plan above the node was built on an ordering the node then did not deliver, and the EXPLAIN gave no sign of it.

On the maintenance branches, GroupPartitions itself is recent, no golden file contains it, and the line already carries DistributePartitions, so the shape of the string is not new, only one more field on it.

If you would rather not change the string in 4.3 and 4.2 at all, the alternative I would take is printing SortedMerge only when it is true. That keeps every existing EXPLAIN byte-identical and still surfaces the case a reader needs to see. Say the word and I will do that instead, here or in the backports only.

…grid Review response to apache#58543 (comment). The test iterated the config with `enableSortedMerge = true` only, so the assertion this PR removed from the test above, that the config alone does not enable a k-way merge, was no longer checked. It now asserts both halves under both config values, which is what its name says.

dongjoon-hyun

+1, LGTM.

Thank you for the follow-up commit, @peter-toth. The unit test grid now covers the unflagged case under both config values, which restores the assertion the first commit removed and extends it.

On the SortedMerge field in EXPLAIN, I agree with keeping it as is. GroupPartitions itself is new in 4.2, no golden file contains it, and the line already carries an always-printed DistributePartitions, so this only adds a field rather than changing the shape. Since the flag is now the only thing in the plan that says whether the node k-way merges, having it always visible in EXPLAIN helps debugging exactly the kind of mismatch this PR fixes.

uros-b · 2026-09-08T08:29:21Z

+1, thank you @peter-toth and @dongjoon-hyun!

ulysses-you

lgtm, thank you @peter-toth

…artitionsExec is planned ### What changes were proposed in this pull request? `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. ### 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. Measured on `branch-4.3` and `branch-4.2` as well, same 3 rows of 5 in both, 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. Snapshotting the config into a `private val` at construction, the way `SortExec` does with `enableRadixSort`, fixes the same case, but `SparkPlan.conf` is the live conf and a `val` freezes at construction rather than at the decision, so the window reopens for any node copied after `EnsureRequirements` ran. Dropping the `&& canUseSortedMerge` re-check outright would take `childIsSafeForKWayMerge` with it. Reading `conf` instead of `SQLConf.get` is not what fixes this, and that was measured rather than assumed. `SparkPlan.conf` is `session.sessionState.conf`, the session's live mutable `SQLConf`, which is the same object `spark.conf.set` and `withSQLConf` mutate. The read does switch to `conf` here, on separate grounds. It now happens only on the driver during planning, so the node's own session conf is the right one to ask, and `SparkPlan.conf` falls back to `SQLConf.get` when the node has no session. It also makes the file consistent, since `outputOrdering` already read `conf`. `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. ### How was this patch tested? A new test in `KeyGroupedPartitioningSuite`, "SPARK-59279: a planned k-way merge survives a later config change". It 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. AQE builds it in `AdaptiveSparkPlanExec.initialPlan`, and without AQE `prepareForExecution` does. Without the fix both modes return 3 of the 5 rows, dropping `[1,aa]` and `[2,cc]`. One existing test changed, plus a new unit test beside it. `GroupPartitionsExecSuite`'s "SPARK-55715: coalescing with enableSortedMerge = true returns full child ordering" wrapped its flag assertion in `withSQLConf(preserveOrderingOnCoalesce -> true)`. That wrapper is inert now, because `outputOrdering` no longer reads that config, so it is dropped and 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. The sibling test's name lost the words "sorted merge config enabled", for the same reason. All three fail without the fix. 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. The third copy, under "SPARK-55715", is left alone to keep this diff small. "SPARK-55992: GroupPartitions string in simple and extended explain" now expects the `SortedMerge` field. `KeyGroupedPartitioningSuite`, `GroupPartitionsExecSuite`, `EnsureRequirementsSuite`, `SortedMergeCoalescedRDDSuite`, `PlannerSuite` and `ProjectedOrderingAndPartitioningSuite`, 313 tests. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (Opus 5) Closes #58543 from peter-toth/SPARK-59279-sorted-merge-conf-freeze. Authored-by: Peter Toth <peter.toth@gmail.com> Signed-off-by: Peter Toth <peter.toth@gmail.com> (cherry picked from commit 6a23cb6) Signed-off-by: Peter Toth <peter.toth@gmail.com>

peter-toth · 2026-09-08T16:04:51Z

Merge Summary:

merged into master 6a23cb6
merged into branch-4.x 0f84262
merged into branch-4.3 bb578b2

Posted by merge_spark_pr.py

peter-toth · 2026-09-08T16:05:36Z

Thank you @dongjoon-hyun , @uros-b and @ulysses-you for the review.

I will open 4.2 backport soon.

…roupPartitionsExec is planned ### 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) Closes #58633 from peter-toth/SPARK-59279-sorted-merge-conf-freeze-4.2. Authored-by: Peter Toth <peter.toth@gmail.com> Signed-off-by: Peter Toth <peter.toth@gmail.com>

peter-toth force-pushed the SPARK-59279-sorted-merge-conf-freeze branch from a3a1889 to e2dad05 Compare September 7, 2026 10:53

peter-toth force-pushed the SPARK-59279-sorted-merge-conf-freeze branch from e2dad05 to 80e1fca Compare September 7, 2026 16:15

dongjoon-hyun reviewed Sep 7, 2026

View reviewed changes

dongjoon-hyun approved these changes Sep 8, 2026

View reviewed changes

uros-b approved these changes Sep 8, 2026

View reviewed changes

ulysses-you approved these changes Sep 8, 2026

View reviewed changes

peter-toth closed this in 6a23cb6 Sep 8, 2026

peter-toth mentioned this pull request Sep 8, 2026

[SPARK-59279][SQL][4.2] Don't re-read the sorted-merge config after GroupPartitionsExec is planned #58633

Closed

Navigation Menu

Uh oh!

[SPARK-59279][SQL] Don't re-read the sorted-merge config after GroupPartitionsExec is planned - #58543

[SPARK-59279][SQL] Don't re-read the sorted-merge config after GroupPartitionsExec is planned#58543
peter-toth wants to merge 2 commits into
apache:masterfrom
peter-toth:SPARK-59279-sorted-merge-conf-freeze

peter-toth commented Sep 5, 2026 •

edited

Loading

Uh oh!

peter-toth commented Sep 7, 2026

Uh oh!

dongjoon-hyun left a comment

Uh oh!

dongjoon-hyun Sep 7, 2026

Uh oh!

peter-toth Sep 8, 2026

Uh oh!

dongjoon-hyun Sep 7, 2026

Uh oh!

peter-toth Sep 8, 2026

Uh oh!

dongjoon-hyun left a comment •

edited

Loading

Uh oh!

uros-b commented Sep 8, 2026

Uh oh!

ulysses-you left a comment

Uh oh!

peter-toth commented Sep 8, 2026

Uh oh!

peter-toth commented Sep 8, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

Uh oh!

Conversation

peter-toth commented Sep 5, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Was this patch authored or co-authored using generative AI tooling?

Uh oh!

peter-toth commented Sep 7, 2026

Uh oh!

dongjoon-hyun left a comment

Choose a reason for hiding this comment

Uh oh!

dongjoon-hyun Sep 7, 2026

Choose a reason for hiding this comment

Uh oh!

peter-toth Sep 8, 2026

Choose a reason for hiding this comment

Uh oh!

dongjoon-hyun Sep 7, 2026

Choose a reason for hiding this comment

Uh oh!

peter-toth Sep 8, 2026

Choose a reason for hiding this comment

Uh oh!

dongjoon-hyun left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

uros-b commented Sep 8, 2026

Uh oh!

ulysses-you left a comment

Choose a reason for hiding this comment

Uh oh!

peter-toth commented Sep 8, 2026

Uh oh!

peter-toth commented Sep 8, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

peter-toth commented Sep 5, 2026 •

edited

Loading

dongjoon-hyun left a comment •

edited

Loading