You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "peter-toth (via GitHub)" <gi...@apache.org> on 2026/04/21 13:14:27 UTC
[PR] [SPARK-56549][SQL] Dynamically enable k-way merge in `GroupPartitionsExec` only when parent requires ordering [spark]
peter-toth opened a new pull request, #55448:
URL: https://github.com/apache/spark/pull/55448
### What changes were proposed in this pull request?
`GroupPartitionsExec` can coalesce multiple input partitions sharing the same partition key into one output partition. When `spark.sql.sources.v2.bucketing. preserveOrderingOnCoalesce.enabled` is set, it uses `SortedMergeCoalescedRDD` (k-way merge) to preserve the child's output ordering through coalescing.
Previously the k-way merge path was taken unconditionally whenever the config was enabled and the child was safe, even when the parent operator did not require any ordering — incurring unnecessary merge overhead for e.g. hash joins.
This PR introduces a dynamic gate:
- A new `enableSortedMerge: Boolean` parameter on `GroupPartitionsExec` (default `false`) controls whether k-way merge is used.
- A new `canUseSortedMerge` lazy val captures the pure feasibility check (config enabled + child has ordering + child is `SafeForKWayMerge`).
- A new `tryEnableSortedMerge()` method returns `Some(copy(enableSortedMerge = true))` when feasible, `None` otherwise.
- `EnsureRequirements`, when it would otherwise insert a `SortExec` above a subtree, first walks down through `OrderPreservingUnaryExecNode`s to find a `GroupPartitionsExec` and calls `tryEnableSortedMerge()` on it. If the resulting `outputOrdering` satisfies the parent's requirement, the updated subtree is used instead of adding a `SortExec`.
- A `hasCoalescing` lazy val is extracted to avoid repeating `groupedPartitions.exists(_._2.size > 1)`.
### Why are the changes needed?
K-way merge is expensive than simple sequential concatenation. When the parent operator does not require ordering (e.g. a hash join), this overhead is wasted. The dynamic gate ensures k-way merge is activated only when it actually satisfies a parent ordering requirement.
### Does this PR introduce _any_ user-facing change?
No. But on the unreleased `master` branch when
`spark.sql.sources.v2.bucketing.preserveOrderingOnCoalesce.enabled=true`, k-way merge is now applied selectively (only when a parent requires the ordering) rather than eagerly for all coalescing `GroupPartitionsExec` nodes. Operators that do not require ordering (e.g. hash joins) will use the cheaper `CoalescedRDD` path even with the config enabled.
### How was this patch tested?
- `GroupPartitionsExecSuite`: updated existing SPARK-55715 tests to set `enableSortedMerge = true` explicitly; added new SPARK-56549 tests for `tryEnableSortedMerge()` covering all feasibility conditions.
- `KeyGroupedPartitioningSuite`: added end-to-end test asserting that a hash join keeps `enableSortedMerge = false` / uses `CoalescedRDD`, while a sort-merge join gets `enableSortedMerge = true` / uses `SortedMergeCoalescedRDD`.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Sonnet 4.6
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56549][SQL] Dynamically enable k-way merge in `GroupPartitionsExec` only when parent requires ordering [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #55448:
URL: https://github.com/apache/spark/pull/55448#discussion_r3139438389
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -286,13 +286,42 @@ case class EnsureRequirements(
if (SortOrder.orderingSatisfies(child.outputOrdering, requiredOrdering)) {
child
} else {
- SortExec(requiredOrdering, global = false, child = child)
+ // Before adding a SortExec, check whether a GroupPartitionsExec anywhere in the child
+ // subtree can self-satisfy via sorted merge.
+ tryEnableSortedMerge(child) match {
+ case Some(newChild)
+ if SortOrder.orderingSatisfies(newChild.outputOrdering, requiredOrdering) =>
+ newChild
+ case _ => SortExec(requiredOrdering, global = false, child = child)
+ }
}
}
children
}
+ private def tryEnableSortedMerge(plan: SparkPlan): Option[SparkPlan] = {
+ var sortedMergeEnabled = false
+ var stop = false
+ val updated = plan.transformDownWithPruning(_ => !stop) {
+ case gpe: GroupPartitionsExec =>
+ stop = true
+ gpe.tryEnableSortedMerge() match {
+ case Some(u) =>
+ sortedMergeEnabled = true
+ u
+ case None => gpe
+ }
+ case node if !node.isInstanceOf[OrderPreservingUnaryExecNode] =>
Review Comment:
~Yeah, you are right. I changed the logic in https://github.com/apache/spark/pull/55448/commits/8f3e08d775fcd233038ba7c2c6f447e24748d6ab to walk down on `UnaryExec` nodes till `SortExec` or `EchangeExec`.~
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56549][SQL] Dynamically enable k-way merge in `GroupPartitionsExec` only when parent requires ordering [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #55448:
URL: https://github.com/apache/spark/pull/55448#discussion_r3152375422
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala:
##########
@@ -223,10 +230,21 @@ case class GroupPartitionsExec(
child.outputOrdering.nonEmpty &&
childIsSafeForKWayMerge
+ /**
+ * Returns a copy of this node with k-way merge enabled if the config permits it.
+ */
Review Comment:
Honest late catch — I missed this in the prior round. "if the config permits it" is too narrow: the actual gate is `hasCoalescing && canUseSortedMerge`, which is config + child ordering non-empty + child subtree `SafeForKWayMerge` + the operator actually coalesces. Worth tightening so a reader doesn't assume just the config flag governs the result:
```suggestion
/**
* Returns a copy of this node with k-way merge enabled if it is feasible: the config is on,
* the child has an ordering, the child subtree is `SafeForKWayMerge`, and this node actually
* coalesces partitions.
*/
```
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/exchange/EnsureRequirementsSuite.scala:
##########
@@ -1397,4 +1398,157 @@ class EnsureRequirementsSuite extends SharedSparkSession {
requiredChildDistribution = Seq(UnspecifiedDistribution),
requiredChildOrdering = Seq(Seq.empty)
)
+
+ test("SPARK-56549: tryEnableSortedMerge traversal continues through plain unary nodes") {
+ withSQLConf(SQLConf.V2_BUCKETING_PRESERVE_ORDERING_ON_COALESCE_ENABLED.key -> "true") {
+ val exprKey = AttributeReference("k", IntegerType)()
+ val partitionKeys = Seq(InternalRow(1), InternalRow(2), InternalRow(1))
+ val ordering = Seq(SortOrder(exprKey, Ascending))
+ val leaf = DummyLeafSafeForKWayMerge(
+ outputPartitioning = KeyedPartitioning(Seq(exprKey), partitionKeys),
+ outputOrdering = ordering)
+ val gpe = GroupPartitionsExec(leaf)
+
+ // Baseline: GPE at root -- at least one alternative has sorted merge enabled.
+ assert(EnsureRequirements.tryEnableSortedMerge(gpe).exists(anyGpeEnabled))
+ // Plain unary wrapper (e.g. FilterExec): traversal continues and sorted merge is enabled.
+ assert(EnsureRequirements.tryEnableSortedMerge(DummyPassthroughUnaryExec(gpe))
+ .exists(anyGpeEnabled))
+ // Two levels of plain unary wrappers: still enabled.
+ assert(EnsureRequirements.tryEnableSortedMerge(
+ DummyPassthroughUnaryExec(DummyPassthroughUnaryExec(gpe)))
+ .exists(anyGpeEnabled))
+ }
+ }
+
+ test("SPARK-56549: tryEnableSortedMerge traversal continues through binary nodes that " +
+ "propagate ordering from one child (e.g. ShuffledHashJoinExec stream side)") {
+ withSQLConf(SQLConf.V2_BUCKETING_PRESERVE_ORDERING_ON_COALESCE_ENABLED.key -> "true") {
+ val exprKey = AttributeReference("k", IntegerType)()
+ val partitionKeys = Seq(InternalRow(1), InternalRow(2), InternalRow(1))
+ val ordering = Seq(SortOrder(exprKey, Ascending))
+ val leaf = DummyLeafSafeForKWayMerge(
+ outputPartitioning = KeyedPartitioning(Seq(exprKey), partitionKeys),
+ outputOrdering = ordering)
+ val gpe = GroupPartitionsExec(leaf)
+ val otherChild = DummyLeafSafeForKWayMerge()
+
+ // Binary node whose ordering comes from left child (GPE side): sorted merge enabled.
+ assert(EnsureRequirements.tryEnableSortedMerge(DummyOrderFromLeftBinaryExec(gpe, otherChild))
+ .exists(anyGpeEnabled))
+ // Binary node with GPE only on the non-ordering (right) side: the binary node's
+ // outputPartitioning = left.outputPartitioning carries no KeyedPartitioning, so the pruning
+ // condition stops traversal at the root; no GPE is enabled.
+ assert(!EnsureRequirements.tryEnableSortedMerge(DummyOrderFromLeftBinaryExec(otherChild, gpe))
+ .exists(anyGpeEnabled))
+ }
+ }
+
+ test("SPARK-56549: tryEnableSortedMerge traversal through binary nodes with " +
+ "PartitioningCollection (KP from both children, e.g. SHJ InnerLike)") {
+ withSQLConf(SQLConf.V2_BUCKETING_PRESERVE_ORDERING_ON_COALESCE_ENABLED.key -> "true") {
+ val exprKey = AttributeReference("k", IntegerType)()
+ val partitionKeys = Seq(InternalRow(1), InternalRow(2), InternalRow(1))
+ val ordering = Seq(SortOrder(exprKey, Ascending))
+ val leaf = DummyLeafSafeForKWayMerge(
+ outputPartitioning = KeyedPartitioning(Seq(exprKey), partitionKeys),
+ outputOrdering = ordering)
+ val gpe = GroupPartitionsExec(leaf)
+ val otherChild = DummyLeafSafeForKWayMerge(
+ outputPartitioning = UnknownPartitioning(gpe.outputPartitioning.numPartitions))
+
+ // GPE on ordering (left) side: sorted merge is enabled and the binary's outputOrdering
+ // becomes non-empty.
+ assert(EnsureRequirements.tryEnableSortedMerge(DummyBothKPBinaryExec(gpe, otherChild))
+ .exists(p => anyGpeEnabled(p) && p.outputOrdering.nonEmpty))
+
+ // GPE on non-ordering (right) side: the PartitioningCollection on the binary node includes
+ // KP from the right child, so traversal enters the binary and sorted merge IS enabled on the
+ // GPE. However, the binary's outputOrdering remains empty: it comes from the left (non-GPE)
+ // child. The call site's find correctly rejects all such alternatives.
+ assert(!EnsureRequirements.tryEnableSortedMerge(DummyBothKPBinaryExec(otherChild, gpe))
+ .exists(_.outputOrdering.nonEmpty))
Review Comment:
The comment above this assertion claims the GPE *is* enabled in some alternative ("the PartitioningCollection ... includes KP from the right child, so traversal enters the binary and sorted merge IS enabled on the GPE"), but the assertion only verifies the *outcome* (no alternative has non-empty `outputOrdering`). The interesting invariant — that `hasKeyedPartitioning` correctly recognizes the right-child KP via `PartitioningCollection` and traversal therefore enters the right side — isn't pinned. Worth adding a positive check so a future change to `hasKeyedPartitioning` that accidentally drops `PartitioningCollection` recursion would fail this test:
```suggestion
assert(EnsureRequirements.tryEnableSortedMerge(DummyBothKPBinaryExec(otherChild, gpe))
.exists(anyGpeEnabled))
assert(!EnsureRequirements.tryEnableSortedMerge(DummyBothKPBinaryExec(otherChild, gpe))
.exists(_.outputOrdering.nonEmpty))
```
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56549][SQL] Dynamically enable k-way merge in `GroupPartitionsExec` only when parent requires ordering [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #55448:
URL: https://github.com/apache/spark/pull/55448#discussion_r3119650652
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala:
##########
@@ -223,10 +230,16 @@ case class GroupPartitionsExec(
child.outputOrdering.nonEmpty &&
childIsSafeForKWayMerge
+ /**
+ * Returns a copy of this node with k-way merge enabled if the config permits it.
+ * */
Review Comment:
nit. Typo? `* */`?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56549][SQL] Dynamically enable k-way merge in `GroupPartitionsExec` only when parent requires ordering [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #55448:
URL: https://github.com/apache/spark/pull/55448#discussion_r3144033883
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -286,13 +286,42 @@ case class EnsureRequirements(
if (SortOrder.orderingSatisfies(child.outputOrdering, requiredOrdering)) {
child
} else {
- SortExec(requiredOrdering, global = false, child = child)
+ // Before adding a SortExec, check whether a GroupPartitionsExec anywhere in the child
+ // subtree can self-satisfy via sorted merge.
+ tryEnableSortedMerge(child) match {
+ case Some(newChild)
+ if SortOrder.orderingSatisfies(newChild.outputOrdering, requiredOrdering) =>
+ newChild
+ case _ => SortExec(requiredOrdering, global = false, child = child)
+ }
}
}
children
}
+ private def tryEnableSortedMerge(plan: SparkPlan): Option[SparkPlan] = {
Review Comment:
I refined the logic in https://github.com/apache/spark/pull/55448/commits/10fe8d4fe321f01312b39acb8ebfbd76d926037b.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56549][SQL] Dynamically enable k-way merge in `GroupPartitionsExec` only when parent requires ordering [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #55448:
URL: https://github.com/apache/spark/pull/55448#issuecomment-4333892734
Thanks @cloud-fan. Accepted your suggestions and added a comment in https://github.com/apache/spark/pull/55448/commits/cb4dca671806074cc204334eed2ae5997f2ccd0a related to "Pruning vs. ordering propagation" and updated the PR description.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56549][SQL] Dynamically enable k-way merge in `GroupPartitionsExec` only when parent requires ordering [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #55448:
URL: https://github.com/apache/spark/pull/55448#issuecomment-4338786060
Merged to master for Apache Spark 4.2.0. Thank you, @peter-toth and @cloud-fan .
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56549][SQL] Dynamically enable k-way merge in `GroupPartitionsExec` only when parent requires ordering [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #55448:
URL: https://github.com/apache/spark/pull/55448#discussion_r3144033331
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -286,13 +286,42 @@ case class EnsureRequirements(
if (SortOrder.orderingSatisfies(child.outputOrdering, requiredOrdering)) {
child
} else {
- SortExec(requiredOrdering, global = false, child = child)
+ // Before adding a SortExec, check whether a GroupPartitionsExec anywhere in the child
+ // subtree can self-satisfy via sorted merge.
+ tryEnableSortedMerge(child) match {
+ case Some(newChild)
+ if SortOrder.orderingSatisfies(newChild.outputOrdering, requiredOrdering) =>
+ newChild
+ case _ => SortExec(requiredOrdering, global = false, child = child)
+ }
}
}
children
}
+ private def tryEnableSortedMerge(plan: SparkPlan): Option[SparkPlan] = {
+ var sortedMergeEnabled = false
+ var stop = false
+ val updated = plan.transformDownWithPruning(_ => !stop) {
+ case gpe: GroupPartitionsExec =>
+ stop = true
+ gpe.tryEnableSortedMerge() match {
+ case Some(u) =>
+ sortedMergeEnabled = true
+ u
+ case None => gpe
+ }
+ case node if !node.isInstanceOf[OrderPreservingUnaryExecNode] =>
Review Comment:
I refined the logic in https://github.com/apache/spark/pull/55448/commits/10fe8d4fe321f01312b39acb8ebfbd76d926037b.
Actually it is pretty niche scenario when the output order requiring node is not the one that required inserting `GroupPartitionsExec` to satisfy its partitioning requirement, but it is possible and it can be non-unary. E.g. `SMJ(SHJ(a, b, LeftOuter, BuildRight), c)` can now become `SMJ(SHJ(GroupPartitionsExec(a, enableSortedMerge = true), GroupPartitionsExec(b), LeftOuter, Buildright), c)`.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56549][SQL] Dynamically enable k-way merge in `GroupPartitionsExec` only when parent requires ordering [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #55448:
URL: https://github.com/apache/spark/pull/55448#discussion_r3144033331
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -286,13 +286,42 @@ case class EnsureRequirements(
if (SortOrder.orderingSatisfies(child.outputOrdering, requiredOrdering)) {
child
} else {
- SortExec(requiredOrdering, global = false, child = child)
+ // Before adding a SortExec, check whether a GroupPartitionsExec anywhere in the child
+ // subtree can self-satisfy via sorted merge.
+ tryEnableSortedMerge(child) match {
+ case Some(newChild)
+ if SortOrder.orderingSatisfies(newChild.outputOrdering, requiredOrdering) =>
+ newChild
+ case _ => SortExec(requiredOrdering, global = false, child = child)
+ }
}
}
children
}
+ private def tryEnableSortedMerge(plan: SparkPlan): Option[SparkPlan] = {
+ var sortedMergeEnabled = false
+ var stop = false
+ val updated = plan.transformDownWithPruning(_ => !stop) {
+ case gpe: GroupPartitionsExec =>
+ stop = true
+ gpe.tryEnableSortedMerge() match {
+ case Some(u) =>
+ sortedMergeEnabled = true
+ u
+ case None => gpe
+ }
+ case node if !node.isInstanceOf[OrderPreservingUnaryExecNode] =>
Review Comment:
I refined the logic in https://github.com/apache/spark/pull/55448/commits/10fe8d4fe321f01312b39acb8ebfbd76d926037b.
Actually it is pretty niche scenario when the output order requiring node is not the one that required inserting `GroupPartitionsExec` to satisfy its partitioning requirement, but it is possible. E.g. `SMJ(SHJ(a, b, LeftOuter, BuildRight), c)` can now become `SMJ(SHJ(GroupPartitionsExec(a, enableSortedMerge = true), GroupPartitionsExec(b), LeftOuter, Buildright), c)`.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56549][SQL] Dynamically enable k-way merge in `GroupPartitionsExec` only when parent requires ordering [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #55448:
URL: https://github.com/apache/spark/pull/55448#discussion_r3139433179
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -3913,4 +3913,79 @@ class KeyGroupedPartitioningSuite extends DistributionAndOrderingSuiteBase with
}
}
}
+
+ test("SPARK-56549: k-way merge enabled only when parent requires ordering") {
+ // Both tables are partitioned by id/item_id and report a two-column ordering.
+ // Key 1 appears on two splits on each side, so GroupPartitionsExec must coalesce.
+ //
+ // Dynamic gate: with the config enabled, k-way merge must be activated only when the parent
+ // actually requires ordering (SMJ), and must stay off when the parent does not (hash join).
+ val itemOrdering = Array(
+ sort(FieldReference("id"), SortDirection.ASCENDING, NullOrdering.NULLS_FIRST),
+ sort(FieldReference("arrive_time"), SortDirection.ASCENDING, NullOrdering.NULLS_FIRST))
+ createTable(items, itemsColumns, Array(identity("id")), itemOrdering)
+ sql(s"INSERT INTO testcat.ns.$items VALUES " +
+ "(2, 'cc', 30.0, cast('2023-06-15' as timestamp)), " +
+ "(1, 'bb', 20.0, cast('2022-03-10' as timestamp)), " +
+ "(3, 'dd', 40.0, cast('2024-01-01' as timestamp)), " +
+ "(1, 'aa', 10.0, cast('2021-05-20' as timestamp)), " +
+ "(2, 'ee', 50.0, cast('2025-09-01' as timestamp))")
+
+ val purchaseOrdering = Array(
+ sort(FieldReference("item_id"), SortDirection.ASCENDING, NullOrdering.NULLS_FIRST),
+ sort(FieldReference("time"), SortDirection.ASCENDING, NullOrdering.NULLS_FIRST))
+ createTable(purchases, purchasesColumns, Array(identity("item_id")), purchaseOrdering)
+ sql(s"INSERT INTO testcat.ns.$purchases VALUES " +
+ "(2, 50.0, cast('2025-09-01' as timestamp)), " +
+ "(1, 10.0, cast('2021-05-20' as timestamp)), " +
+ "(3, 40.0, cast('2024-01-01' as timestamp)), " +
+ "(2, 30.0, cast('2023-06-15' as timestamp)), " +
+ "(1, 20.0, cast('2022-03-10' as timestamp))")
+
+ withSQLConf(
+ SQLConf.REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION.key -> "false",
+ SQLConf.V2_BUCKETING_PRESERVE_ORDERING_ON_COALESCE_ENABLED.key -> "true"
+ ) {
+ val hashDf = sql(
+ s"""
+ |SELECT /*+ SHUFFLE_HASH(i, p) */ i.id, i.name
+ |FROM testcat.ns.$items i
+ |JOIN testcat.ns.$purchases p ON p.item_id = i.id AND p.time = i.arrive_time
+ |""".stripMargin)
+ val hashPlan = hashDf.queryExecution.executedPlan
Review Comment:
Added in https://github.com/apache/spark/pull/55448/commits/10fe8d4fe321f01312b39acb8ebfbd76d926037b.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56549][SQL] Dynamically enable k-way merge in `GroupPartitionsExec` only when parent requires ordering [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun closed pull request #55448: [SPARK-56549][SQL] Dynamically enable k-way merge in `GroupPartitionsExec` only when parent requires ordering
URL: https://github.com/apache/spark/pull/55448
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56549][SQL] Dynamically enable k-way merge in `GroupPartitionsExec` only when parent requires ordering [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #55448:
URL: https://github.com/apache/spark/pull/55448#discussion_r3139433179
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -3913,4 +3913,79 @@ class KeyGroupedPartitioningSuite extends DistributionAndOrderingSuiteBase with
}
}
}
+
+ test("SPARK-56549: k-way merge enabled only when parent requires ordering") {
+ // Both tables are partitioned by id/item_id and report a two-column ordering.
+ // Key 1 appears on two splits on each side, so GroupPartitionsExec must coalesce.
+ //
+ // Dynamic gate: with the config enabled, k-way merge must be activated only when the parent
+ // actually requires ordering (SMJ), and must stay off when the parent does not (hash join).
+ val itemOrdering = Array(
+ sort(FieldReference("id"), SortDirection.ASCENDING, NullOrdering.NULLS_FIRST),
+ sort(FieldReference("arrive_time"), SortDirection.ASCENDING, NullOrdering.NULLS_FIRST))
+ createTable(items, itemsColumns, Array(identity("id")), itemOrdering)
+ sql(s"INSERT INTO testcat.ns.$items VALUES " +
+ "(2, 'cc', 30.0, cast('2023-06-15' as timestamp)), " +
+ "(1, 'bb', 20.0, cast('2022-03-10' as timestamp)), " +
+ "(3, 'dd', 40.0, cast('2024-01-01' as timestamp)), " +
+ "(1, 'aa', 10.0, cast('2021-05-20' as timestamp)), " +
+ "(2, 'ee', 50.0, cast('2025-09-01' as timestamp))")
+
+ val purchaseOrdering = Array(
+ sort(FieldReference("item_id"), SortDirection.ASCENDING, NullOrdering.NULLS_FIRST),
+ sort(FieldReference("time"), SortDirection.ASCENDING, NullOrdering.NULLS_FIRST))
+ createTable(purchases, purchasesColumns, Array(identity("item_id")), purchaseOrdering)
+ sql(s"INSERT INTO testcat.ns.$purchases VALUES " +
+ "(2, 50.0, cast('2025-09-01' as timestamp)), " +
+ "(1, 10.0, cast('2021-05-20' as timestamp)), " +
+ "(3, 40.0, cast('2024-01-01' as timestamp)), " +
+ "(2, 30.0, cast('2023-06-15' as timestamp)), " +
+ "(1, 20.0, cast('2022-03-10' as timestamp))")
+
+ withSQLConf(
+ SQLConf.REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION.key -> "false",
+ SQLConf.V2_BUCKETING_PRESERVE_ORDERING_ON_COALESCE_ENABLED.key -> "true"
+ ) {
+ val hashDf = sql(
+ s"""
+ |SELECT /*+ SHUFFLE_HASH(i, p) */ i.id, i.name
+ |FROM testcat.ns.$items i
+ |JOIN testcat.ns.$purchases p ON p.item_id = i.id AND p.time = i.arrive_time
+ |""".stripMargin)
+ val hashPlan = hashDf.queryExecution.executedPlan
Review Comment:
Added in https://github.com/apache/spark/pull/55448/commits/8f3e08d775fcd233038ba7c2c6f447e24748d6ab.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56549][SQL] Dynamically enable k-way merge in `GroupPartitionsExec` only when parent requires ordering [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #55448:
URL: https://github.com/apache/spark/pull/55448#discussion_r3119644310
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -286,7 +286,32 @@ case class EnsureRequirements(
if (SortOrder.orderingSatisfies(child.outputOrdering, requiredOrdering)) {
child
} else {
- SortExec(requiredOrdering, global = false, child = child)
+ // Before adding a SortExec, check whether a GroupPartitionsExec anywhere in the child
+ // subtree can self-satisfy via k-way merge.
Review Comment:
Do you think we can make this logic as a private helper function, @peter-toth ?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56549][SQL] Dynamically enable k-way merge in `GroupPartitionsExec` only when parent requires ordering [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #55448:
URL: https://github.com/apache/spark/pull/55448#discussion_r3122468632
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -286,7 +286,32 @@ case class EnsureRequirements(
if (SortOrder.orderingSatisfies(child.outputOrdering, requiredOrdering)) {
child
} else {
- SortExec(requiredOrdering, global = false, child = child)
+ // Before adding a SortExec, check whether a GroupPartitionsExec anywhere in the child
+ // subtree can self-satisfy via k-way merge.
Review Comment:
Good idea, extarcted in https://github.com/apache/spark/pull/55448/commits/33b0a75e494edf902e81a9d6be2ca13d1ad25f2c.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala:
##########
@@ -223,10 +230,16 @@ case class GroupPartitionsExec(
child.outputOrdering.nonEmpty &&
childIsSafeForKWayMerge
+ /**
+ * Returns a copy of this node with k-way merge enabled if the config permits it.
+ * */
Review Comment:
Fixed in https://github.com/apache/spark/pull/55448/commits/33b0a75e494edf902e81a9d6be2ca13d1ad25f2c.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56549][SQL] Dynamically enable k-way merge in `GroupPartitionsExec` only when parent requires ordering [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #55448:
URL: https://github.com/apache/spark/pull/55448#discussion_r3139442245
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -286,13 +286,42 @@ case class EnsureRequirements(
if (SortOrder.orderingSatisfies(child.outputOrdering, requiredOrdering)) {
child
} else {
- SortExec(requiredOrdering, global = false, child = child)
+ // Before adding a SortExec, check whether a GroupPartitionsExec anywhere in the child
+ // subtree can self-satisfy via sorted merge.
+ tryEnableSortedMerge(child) match {
+ case Some(newChild)
+ if SortOrder.orderingSatisfies(newChild.outputOrdering, requiredOrdering) =>
+ newChild
+ case _ => SortExec(requiredOrdering, global = false, child = child)
+ }
}
}
children
}
+ private def tryEnableSortedMerge(plan: SparkPlan): Option[SparkPlan] = {
Review Comment:
~Added tests for the new logic in https://github.com/apache/spark/pull/55448/commits/8f3e08d775fcd233038ba7c2c6f447e24748d6ab.~
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56549][SQL] Dynamically enable k-way merge in `GroupPartitionsExec` only when parent requires ordering [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #55448:
URL: https://github.com/apache/spark/pull/55448#discussion_r3139450291
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -286,13 +286,42 @@ case class EnsureRequirements(
if (SortOrder.orderingSatisfies(child.outputOrdering, requiredOrdering)) {
child
} else {
- SortExec(requiredOrdering, global = false, child = child)
+ // Before adding a SortExec, check whether a GroupPartitionsExec anywhere in the child
+ // subtree can self-satisfy via sorted merge.
+ tryEnableSortedMerge(child) match {
+ case Some(newChild)
+ if SortOrder.orderingSatisfies(newChild.outputOrdering, requiredOrdering) =>
+ newChild
+ case _ => SortExec(requiredOrdering, global = false, child = child)
+ }
}
}
children
}
+ private def tryEnableSortedMerge(plan: SparkPlan): Option[SparkPlan] = {
+ var sortedMergeEnabled = false
Review Comment:
Honestly, I like the current shape, it is concise, one-pass and and it uses public `TreeNode` APIs.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56549][SQL] Dynamically enable k-way merge in `GroupPartitionsExec` only when parent requires ordering [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #55448:
URL: https://github.com/apache/spark/pull/55448#issuecomment-4304391581
cc @cloud-fan
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56549][SQL] Dynamically enable k-way merge in `GroupPartitionsExec` only when parent requires ordering [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #55448:
URL: https://github.com/apache/spark/pull/55448#discussion_r3141514201
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -286,13 +286,42 @@ case class EnsureRequirements(
if (SortOrder.orderingSatisfies(child.outputOrdering, requiredOrdering)) {
child
} else {
- SortExec(requiredOrdering, global = false, child = child)
+ // Before adding a SortExec, check whether a GroupPartitionsExec anywhere in the child
+ // subtree can self-satisfy via sorted merge.
+ tryEnableSortedMerge(child) match {
+ case Some(newChild)
+ if SortOrder.orderingSatisfies(newChild.outputOrdering, requiredOrdering) =>
+ newChild
+ case _ => SortExec(requiredOrdering, global = false, child = child)
+ }
}
}
children
}
+ private def tryEnableSortedMerge(plan: SparkPlan): Option[SparkPlan] = {
+ var sortedMergeEnabled = false
+ var stop = false
+ val updated = plan.transformDownWithPruning(_ => !stop) {
+ case gpe: GroupPartitionsExec =>
+ stop = true
+ gpe.tryEnableSortedMerge() match {
+ case Some(u) =>
+ sortedMergeEnabled = true
+ u
+ case None => gpe
+ }
+ case node if !node.isInstanceOf[OrderPreservingUnaryExecNode] =>
Review Comment:
Actually, let me refine this further.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56549][SQL] Dynamically enable k-way merge in `GroupPartitionsExec` only when parent requires ordering [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #55448:
URL: https://github.com/apache/spark/pull/55448#discussion_r3139442245
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -286,13 +286,42 @@ case class EnsureRequirements(
if (SortOrder.orderingSatisfies(child.outputOrdering, requiredOrdering)) {
child
} else {
- SortExec(requiredOrdering, global = false, child = child)
+ // Before adding a SortExec, check whether a GroupPartitionsExec anywhere in the child
+ // subtree can self-satisfy via sorted merge.
+ tryEnableSortedMerge(child) match {
+ case Some(newChild)
+ if SortOrder.orderingSatisfies(newChild.outputOrdering, requiredOrdering) =>
+ newChild
+ case _ => SortExec(requiredOrdering, global = false, child = child)
+ }
}
}
children
}
+ private def tryEnableSortedMerge(plan: SparkPlan): Option[SparkPlan] = {
Review Comment:
Added tests for the new logic in https://github.com/apache/spark/pull/55448/commits/8f3e08d775fcd233038ba7c2c6f447e24748d6ab.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56549][SQL] Dynamically enable k-way merge in `GroupPartitionsExec` only when parent requires ordering [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #55448:
URL: https://github.com/apache/spark/pull/55448#discussion_r3139433179
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -3913,4 +3913,79 @@ class KeyGroupedPartitioningSuite extends DistributionAndOrderingSuiteBase with
}
}
}
+
+ test("SPARK-56549: k-way merge enabled only when parent requires ordering") {
+ // Both tables are partitioned by id/item_id and report a two-column ordering.
+ // Key 1 appears on two splits on each side, so GroupPartitionsExec must coalesce.
+ //
+ // Dynamic gate: with the config enabled, k-way merge must be activated only when the parent
+ // actually requires ordering (SMJ), and must stay off when the parent does not (hash join).
+ val itemOrdering = Array(
+ sort(FieldReference("id"), SortDirection.ASCENDING, NullOrdering.NULLS_FIRST),
+ sort(FieldReference("arrive_time"), SortDirection.ASCENDING, NullOrdering.NULLS_FIRST))
+ createTable(items, itemsColumns, Array(identity("id")), itemOrdering)
+ sql(s"INSERT INTO testcat.ns.$items VALUES " +
+ "(2, 'cc', 30.0, cast('2023-06-15' as timestamp)), " +
+ "(1, 'bb', 20.0, cast('2022-03-10' as timestamp)), " +
+ "(3, 'dd', 40.0, cast('2024-01-01' as timestamp)), " +
+ "(1, 'aa', 10.0, cast('2021-05-20' as timestamp)), " +
+ "(2, 'ee', 50.0, cast('2025-09-01' as timestamp))")
+
+ val purchaseOrdering = Array(
+ sort(FieldReference("item_id"), SortDirection.ASCENDING, NullOrdering.NULLS_FIRST),
+ sort(FieldReference("time"), SortDirection.ASCENDING, NullOrdering.NULLS_FIRST))
+ createTable(purchases, purchasesColumns, Array(identity("item_id")), purchaseOrdering)
+ sql(s"INSERT INTO testcat.ns.$purchases VALUES " +
+ "(2, 50.0, cast('2025-09-01' as timestamp)), " +
+ "(1, 10.0, cast('2021-05-20' as timestamp)), " +
+ "(3, 40.0, cast('2024-01-01' as timestamp)), " +
+ "(2, 30.0, cast('2023-06-15' as timestamp)), " +
+ "(1, 20.0, cast('2022-03-10' as timestamp))")
+
+ withSQLConf(
+ SQLConf.REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION.key -> "false",
+ SQLConf.V2_BUCKETING_PRESERVE_ORDERING_ON_COALESCE_ENABLED.key -> "true"
+ ) {
+ val hashDf = sql(
+ s"""
+ |SELECT /*+ SHUFFLE_HASH(i, p) */ i.id, i.name
+ |FROM testcat.ns.$items i
+ |JOIN testcat.ns.$purchases p ON p.item_id = i.id AND p.time = i.arrive_time
+ |""".stripMargin)
+ val hashPlan = hashDf.queryExecution.executedPlan
Review Comment:
Added in https://github.com/apache/spark/pull/55448/commits/fe6be74f84614c0eba1f4170f142aafe015a4762.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56549][SQL] Dynamically enable k-way merge in `GroupPartitionsExec` only when parent requires ordering [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #55448:
URL: https://github.com/apache/spark/pull/55448#issuecomment-4290965401
cc @szehon-ho, @cloud-fan, @dongjoon-hyun
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56549][SQL] Dynamically enable k-way merge in `GroupPartitionsExec` only when parent requires ordering [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #55448:
URL: https://github.com/apache/spark/pull/55448#discussion_r3137244395
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -286,13 +286,42 @@ case class EnsureRequirements(
if (SortOrder.orderingSatisfies(child.outputOrdering, requiredOrdering)) {
child
} else {
- SortExec(requiredOrdering, global = false, child = child)
+ // Before adding a SortExec, check whether a GroupPartitionsExec anywhere in the child
+ // subtree can self-satisfy via sorted merge.
+ tryEnableSortedMerge(child) match {
+ case Some(newChild)
+ if SortOrder.orderingSatisfies(newChild.outputOrdering, requiredOrdering) =>
+ newChild
+ case _ => SortExec(requiredOrdering, global = false, child = child)
+ }
}
}
children
}
+ private def tryEnableSortedMerge(plan: SparkPlan): Option[SparkPlan] = {
Review Comment:
The `KeyGroupedPartitioningSuite` test exercises this helper end-to-end through real SQL plans, but there's no direct test in `EnsureRequirementsSuite` that (a) `Project(GroupPartitionsExec(...))` still enables sorted merge and (b) a non-OPE wrapper between the parent and the GP correctly blocks the walk. Given the traversal rule is subtle and easy to break silently (e.g., someone adding a new node to `OrderPreservingUnaryExecNode`, or changing the pattern order), a couple of focused unit tests would pin the contract.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -286,13 +286,42 @@ case class EnsureRequirements(
if (SortOrder.orderingSatisfies(child.outputOrdering, requiredOrdering)) {
child
} else {
- SortExec(requiredOrdering, global = false, child = child)
+ // Before adding a SortExec, check whether a GroupPartitionsExec anywhere in the child
+ // subtree can self-satisfy via sorted merge.
+ tryEnableSortedMerge(child) match {
+ case Some(newChild)
+ if SortOrder.orderingSatisfies(newChild.outputOrdering, requiredOrdering) =>
+ newChild
+ case _ => SortExec(requiredOrdering, global = false, child = child)
+ }
}
}
children
}
+ private def tryEnableSortedMerge(plan: SparkPlan): Option[SparkPlan] = {
+ var sortedMergeEnabled = false
Review Comment:
The two `var`s plus the `_ => !stop` pruning callback work, but mix mutable state with a declarative traversal and only make sense if the reader knows the exact order `transformDownWithPruning` visits/prunes nodes. Given the target is a linear OPE chain, a direct recursive helper would read more clearly and avoid relying on visit order:
```scala
@tailrec
private def findGroupPartitions(plan: SparkPlan): Option[GroupPartitionsExec] = plan match {
case gpe: GroupPartitionsExec => Some(gpe)
case o: OrderPreservingUnaryExecNode => findGroupPartitions(o.child)
case _ => None
}
```
Then the caller does the GP swap via a small `transform`. Feel free to keep the current shape if you prefer — not blocking.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -286,13 +286,42 @@ case class EnsureRequirements(
if (SortOrder.orderingSatisfies(child.outputOrdering, requiredOrdering)) {
child
} else {
- SortExec(requiredOrdering, global = false, child = child)
+ // Before adding a SortExec, check whether a GroupPartitionsExec anywhere in the child
+ // subtree can self-satisfy via sorted merge.
+ tryEnableSortedMerge(child) match {
+ case Some(newChild)
+ if SortOrder.orderingSatisfies(newChild.outputOrdering, requiredOrdering) =>
+ newChild
+ case _ => SortExec(requiredOrdering, global = false, child = child)
+ }
}
}
children
}
+ private def tryEnableSortedMerge(plan: SparkPlan): Option[SparkPlan] = {
+ var sortedMergeEnabled = false
+ var stop = false
+ val updated = plan.transformDownWithPruning(_ => !stop) {
+ case gpe: GroupPartitionsExec =>
+ stop = true
+ gpe.tryEnableSortedMerge() match {
+ case Some(u) =>
+ sortedMergeEnabled = true
+ u
+ case None => gpe
+ }
+ case node if !node.isInstanceOf[OrderPreservingUnaryExecNode] =>
Review Comment:
The walk stops at any node that isn't an `OrderPreservingUnaryExecNode`, but that trait is about alias-aware ordering projection — plain unary nodes like `FilterExec` also preserve within-partition ordering by overriding `outputOrdering` to return `child.outputOrdering` and are outside the trait. If a `Filter(GroupPartitionsExec(...))` sits under an SMJ (uncommon in SPJ, since filters typically push into the v2 scan), the optimization is skipped and `SortExec` is inserted instead. Is that intentional (conservative whitelist) — and if so, worth a one-line comment calling it out so a future reader doesn't expand the trait by accident?
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -3913,4 +3913,79 @@ class KeyGroupedPartitioningSuite extends DistributionAndOrderingSuiteBase with
}
}
}
+
+ test("SPARK-56549: k-way merge enabled only when parent requires ordering") {
+ // Both tables are partitioned by id/item_id and report a two-column ordering.
+ // Key 1 appears on two splits on each side, so GroupPartitionsExec must coalesce.
+ //
+ // Dynamic gate: with the config enabled, k-way merge must be activated only when the parent
+ // actually requires ordering (SMJ), and must stay off when the parent does not (hash join).
+ val itemOrdering = Array(
+ sort(FieldReference("id"), SortDirection.ASCENDING, NullOrdering.NULLS_FIRST),
+ sort(FieldReference("arrive_time"), SortDirection.ASCENDING, NullOrdering.NULLS_FIRST))
+ createTable(items, itemsColumns, Array(identity("id")), itemOrdering)
+ sql(s"INSERT INTO testcat.ns.$items VALUES " +
+ "(2, 'cc', 30.0, cast('2023-06-15' as timestamp)), " +
+ "(1, 'bb', 20.0, cast('2022-03-10' as timestamp)), " +
+ "(3, 'dd', 40.0, cast('2024-01-01' as timestamp)), " +
+ "(1, 'aa', 10.0, cast('2021-05-20' as timestamp)), " +
+ "(2, 'ee', 50.0, cast('2025-09-01' as timestamp))")
+
+ val purchaseOrdering = Array(
+ sort(FieldReference("item_id"), SortDirection.ASCENDING, NullOrdering.NULLS_FIRST),
+ sort(FieldReference("time"), SortDirection.ASCENDING, NullOrdering.NULLS_FIRST))
+ createTable(purchases, purchasesColumns, Array(identity("item_id")), purchaseOrdering)
+ sql(s"INSERT INTO testcat.ns.$purchases VALUES " +
+ "(2, 50.0, cast('2025-09-01' as timestamp)), " +
+ "(1, 10.0, cast('2021-05-20' as timestamp)), " +
+ "(3, 40.0, cast('2024-01-01' as timestamp)), " +
+ "(2, 30.0, cast('2023-06-15' as timestamp)), " +
+ "(1, 20.0, cast('2022-03-10' as timestamp))")
+
+ withSQLConf(
+ SQLConf.REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION.key -> "false",
+ SQLConf.V2_BUCKETING_PRESERVE_ORDERING_ON_COALESCE_ENABLED.key -> "true"
+ ) {
+ val hashDf = sql(
+ s"""
+ |SELECT /*+ SHUFFLE_HASH(i, p) */ i.id, i.name
+ |FROM testcat.ns.$items i
+ |JOIN testcat.ns.$purchases p ON p.item_id = i.id AND p.time = i.arrive_time
+ |""".stripMargin)
+ val hashPlan = hashDf.queryExecution.executedPlan
Review Comment:
The SMJ branch calls `checkAnswer`, but the hash-join branch only asserts plan shape / RDD type and never runs the query against expected rows. Given this is a new test specifically about keeping the non-merged coalescing path on, adding a `checkAnswer(hashDf, ...)` would make it catch result-level regressions if the `CoalescedRDD` path ever breaks.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56549][SQL] Dynamically enable k-way merge in `GroupPartitionsExec` only when parent requires ordering [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #55448:
URL: https://github.com/apache/spark/pull/55448#discussion_r3139438389
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -286,13 +286,42 @@ case class EnsureRequirements(
if (SortOrder.orderingSatisfies(child.outputOrdering, requiredOrdering)) {
child
} else {
- SortExec(requiredOrdering, global = false, child = child)
+ // Before adding a SortExec, check whether a GroupPartitionsExec anywhere in the child
+ // subtree can self-satisfy via sorted merge.
+ tryEnableSortedMerge(child) match {
+ case Some(newChild)
+ if SortOrder.orderingSatisfies(newChild.outputOrdering, requiredOrdering) =>
+ newChild
+ case _ => SortExec(requiredOrdering, global = false, child = child)
+ }
}
}
children
}
+ private def tryEnableSortedMerge(plan: SparkPlan): Option[SparkPlan] = {
+ var sortedMergeEnabled = false
+ var stop = false
+ val updated = plan.transformDownWithPruning(_ => !stop) {
+ case gpe: GroupPartitionsExec =>
+ stop = true
+ gpe.tryEnableSortedMerge() match {
+ case Some(u) =>
+ sortedMergeEnabled = true
+ u
+ case None => gpe
+ }
+ case node if !node.isInstanceOf[OrderPreservingUnaryExecNode] =>
Review Comment:
Yeah, you are right. I changed the logic in https://github.com/apache/spark/pull/55448/commits/8f3e08d775fcd233038ba7c2c6f447e24748d6ab to walk down on `UnaryExec` nodes till `SortExec` or `EchangeExec`.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org