Skip to content

Commit 4a70dc0

Browse files
committed
[SPARK-58324][SQL] Drop unused sameOrderExpressions from GroupPartitionsExec k-way merge ordering
### What changes were proposed in this pull request? `GroupPartitionsExec` builds a `SortedMergeCoalescedRDD` for the k-way merge and hands it a `LazyCodeGenOrdering` built from `child.outputOrdering`. The generated comparator (`GenerateOrdering`) only needs each `SortOrder`'s sort key (child, direction, null ordering), so this drops `sameOrderExpressions` -- planner-only metadata -- via a small `kWayMergeOrdering` helper before constructing the ordering, so it is not serialized with the RDD in every task. ### Why are the changes needed? `sameOrderExpressions` is unused by the merge comparator and is unnecessary payload serialized with every task. It was also the vector for the `StackOverflowError` fixed in SPARK-58323 (an unforced, deeply-nested `LazyList`); not carrying it here removes this operator's exposure to any such ordering entirely (defense-in-depth), independent of that fix. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? New unit test in `GroupPartitionsExecSuite` asserting `kWayMergeOrdering` keeps the sort key but drops `sameOrderExpressions`. Existing SPARK-55715 sorted-merge tests cover comparator correctness. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Opus 4.8 Closes #57503 from peter-toth/SPARK-58324-drop-sameorderexpressions-kway-merge. Authored-by: Peter Toth <peter.toth@gmail.com> Signed-off-by: Peter Toth <peter.toth@gmail.com> (cherry picked from commit cce4355) Signed-off-by: Peter Toth <peter.toth@gmail.com>
1 parent aae1628 commit 4a70dc0

2 files changed

Lines changed: 26 additions & 1 deletion

File tree

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -233,12 +233,21 @@ case class GroupPartitionsExec(
233233
}
234234
}
235235

236+
/**
237+
* The ordering used by the k-way merge in [[SortedMergeCoalescedRDD]]. The generated comparator
238+
* ([[GenerateOrdering]]) only needs each [[SortOrder]]'s sort key (child, direction, null
239+
* ordering), so `sameOrderExpressions` -- planner-only metadata that would otherwise be
240+
* serialized with the RDD in every task -- is dropped.
241+
*/
242+
private[v2] def kWayMergeOrdering: Seq[SortOrder] =
243+
child.outputOrdering.map(_.copy(sameOrderExpressions = Seq.empty))
244+
236245
override protected def doExecute(): RDD[InternalRow] = {
237246
if (groupedPartitions.isEmpty) {
238247
sparkContext.emptyRDD
239248
} else if (hasCoalescing && enableSortedMerge && canUseSortedMerge) {
240249
val partitionCoalescer = new GroupedPartitionCoalescer(groupedPartitions.map(_._2))
241-
val rowOrdering = new LazyCodeGenOrdering(child.outputOrdering, child.output)
250+
val rowOrdering = new LazyCodeGenOrdering(kWayMergeOrdering, child.output)
242251
new SortedMergeCoalescedRDD[InternalRow](
243252
child.execute(),
244253
groupedPartitions.size,

sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExecSuite.scala

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -48,6 +48,22 @@ class GroupPartitionsExecSuite extends SharedSparkSession {
4848
assert(gpe.outputOrdering === childOrdering)
4949
}
5050

51+
test("SPARK-58324: k-way merge ordering drops sameOrderExpressions") {
52+
// The child ordering carries sameOrderExpressions (planner metadata). The k-way merge
53+
// comparator only needs the sort key, so kWayMergeOrdering keeps child/direction/nullOrdering
54+
// but drops sameOrderExpressions, so LazyCodeGenOrdering does not serialize them with the RDD.
55+
val childOrdering = Seq(SortOrder(exprA, Ascending, Seq(exprB, exprC)))
56+
val child = DummySparkPlan(
57+
outputPartitioning = KeyedPartitioning(Seq(exprA), Seq(row(1), row(2), row(1))),
58+
outputOrdering = childOrdering)
59+
val gpe = GroupPartitionsExec(child)
60+
61+
assert(child.outputOrdering.head.sameOrderExpressions.nonEmpty, "test setup")
62+
val merged = gpe.kWayMergeOrdering
63+
assert(merged.map(so => (so.child, so.direction)) === Seq((exprA, Ascending)))
64+
assert(merged.forall(_.sameOrderExpressions.isEmpty))
65+
}
66+
5167
test("SPARK-56241: coalescing without reducers keeps key-expression orders from child") {
5268
// Key 1 appears on partitions 0 and 2, causing coalescing.
5369
val partitionKeys = Seq(row(1), row(2), row(1))

0 commit comments

Comments
 (0)