(spark) branch master updated: [SPARK-58324][SQL] Drop unused sameOrderExpressions from GroupPartitionsExec k-way merge ordering

[pt...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

peter-toth pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git


The following commit(s) were added to refs/heads/master by this push:
     new cce435599ed3 [SPARK-58324][SQL] Drop unused sameOrderExpressions from GroupPartitionsExec k-way merge ordering
cce435599ed3 is described below

commit cce435599ed332cd5b785d49b26d4e7dde631d8c
Author: Peter Toth <pe...@gmail.com>
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 <pe...@gmail.com>
    Signed-off-by: Peter Toth <pe...@gmail.com>
---
 .../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: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org