This is an automated email from the ASF dual-hosted git repository.
peter-toth pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 4a70dc07df39 [SPARK-58324][SQL] Drop unused sameOrderExpressions from
GroupPartitionsExec k-way merge ordering
4a70dc07df39 is described below
commit 4a70dc07df39fd7d561e34750bd2d478b5665a17
Author: Peter Toth <[email protected]>
AuthorDate: Sat Jul 25 11:49:53 2026 +0200
[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 <[email protected]>
Signed-off-by: Peter Toth <[email protected]>
(cherry picked from commit cce435599ed332cd5b785d49b26d4e7dde631d8c)
Signed-off-by: Peter Toth <[email protected]>
---
.../execution/datasources/v2/GroupPartitionsExec.scala | 11 ++++++++++-
.../datasources/v2/GroupPartitionsExecSuite.scala | 16 ++++++++++++++++
2 files changed, 26 insertions(+), 1 deletion(-)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala
index 4d87be662293..ccff1f5b3311 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala
@@ -233,12 +233,21 @@ case class GroupPartitionsExec(
}
}
+ /**
+ * The ordering used by the k-way merge in [[SortedMergeCoalescedRDD]]. The
generated comparator
+ * ([[GenerateOrdering]]) only needs each [[SortOrder]]'s sort key (child,
direction, null
+ * ordering), so `sameOrderExpressions` -- planner-only metadata that would
otherwise be
+ * serialized with the RDD in every task -- is dropped.
+ */
+ private[v2] def kWayMergeOrdering: Seq[SortOrder] =
+ child.outputOrdering.map(_.copy(sameOrderExpressions = Seq.empty))
+
override protected def doExecute(): RDD[InternalRow] = {
if (groupedPartitions.isEmpty) {
sparkContext.emptyRDD
} else if (hasCoalescing && enableSortedMerge && canUseSortedMerge) {
val partitionCoalescer = new
GroupedPartitionCoalescer(groupedPartitions.map(_._2))
- val rowOrdering = new LazyCodeGenOrdering(child.outputOrdering,
child.output)
+ val rowOrdering = new LazyCodeGenOrdering(kWayMergeOrdering,
child.output)
new SortedMergeCoalescedRDD[InternalRow](
child.execute(),
groupedPartitions.size,
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExecSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExecSuite.scala
index 51951d68cc60..cd09fca7c702 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExecSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExecSuite.scala
@@ -48,6 +48,22 @@ class GroupPartitionsExecSuite extends SharedSparkSession {
assert(gpe.outputOrdering === childOrdering)
}
+ test("SPARK-58324: k-way merge ordering drops sameOrderExpressions") {
+ // The child ordering carries sameOrderExpressions (planner metadata). The
k-way merge
+ // comparator only needs the sort key, so kWayMergeOrdering keeps
child/direction/nullOrdering
+ // but drops sameOrderExpressions, so LazyCodeGenOrdering does not
serialize them with the RDD.
+ val childOrdering = Seq(SortOrder(exprA, Ascending, Seq(exprB, exprC)))
+ val child = DummySparkPlan(
+ outputPartitioning = KeyedPartitioning(Seq(exprA), Seq(row(1), row(2),
row(1))),
+ outputOrdering = childOrdering)
+ val gpe = GroupPartitionsExec(child)
+
+ assert(child.outputOrdering.head.sameOrderExpressions.nonEmpty, "test
setup")
+ val merged = gpe.kWayMergeOrdering
+ assert(merged.map(so => (so.child, so.direction)) === Seq((exprA,
Ascending)))
+ assert(merged.forall(_.sameOrderExpressions.isEmpty))
+ }
+
test("SPARK-56241: coalescing without reducers keeps key-expression orders
from child") {
// Key 1 appears on partitions 0 and 2, causing coalescing.
val partitionKeys = Seq(row(1), row(2), row(1))
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
