This is an automated email from the ASF dual-hosted git repository.
ulysses-you 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 185b0f3420d5 [SPARK-58099][SQL] Remove the local sort dangling below
the shuffle added by skew join optimization
185b0f3420d5 is described below
commit 185b0f3420d5174fa18c16a77bed53bebb6df218
Author: Xiduo You <[email protected]>
AuthorDate: Wed Jul 15 14:56:03 2026 +0800
[SPARK-58099][SQL] Remove the local sort dangling below the shuffle added
by skew join optimization
### What changes were proposed in this pull request?
This PR enhances `RemoveRedundantSorts` to remove a *dangling* local sort
-- a local `SortExec` sitting directly below a shuffle that neither consumes
nor exposes an ordering (both its `requiredChildOrdering` and `outputOrdering`
are empty). Such a shuffle destroys the child ordering, so the sort has no
effect and is dead.
To make this effective for the skew-join case, `RemoveRedundantSorts` is
moved after `OptimizeSkewedJoin` in the AQE `queryStagePreparationRules`
(`AdaptiveSparkPlanExec`), and the same reorder is applied to the non-AQE
`QueryExecution.preparations` to keep the two rule chains consistent. In both
chains `RemoveRedundantSorts` now runs after `DisableUnnecessaryBucketedScan`
(see the correctness note below).
### Why are the changes needed?
Consider a stage with a `ShuffledHashJoin` feeding a `SortMergeJoin` on the
same key, where the SHJ output partitioning already satisfies the SMJ
requirement (no exchange in between, only the SMJ's local sort). When the SHJ
is skewed and skew-join optimization is applied, an extra shuffle is inserted
between the two joins. `EnsureRequirements` wraps that shuffle on top of the
existing `Sort -> SHJ`, and adds a new sort above the shuffle for the SMJ. The
original local sort is then lef [...]
The current `RemoveRedundantSorts` cannot remove it: it only strips a sort
whose child already satisfies the ordering, whereas this sort is dead because
it sits directly under an ordering-destroying shuffle -- a different kind of
redundancy. Running before `OptimizeSkewedJoin` also means it never sees the
added shuffle.
### Does this PR introduce _any_ user-facing change?
No. It only removes a redundant sort operator from the physical plan (and
fixes the latent ordering-loss hazard described above).
### How was this patch tested?
New UT `SPARK-58099` in `AdaptiveQueryExecSuite` reproduces the SHJ->SMJ
skew-join scenario: with the rule disabled it asserts a local sort dangles
below the added shuffle, and with the rule enabled it asserts the sort is
removed while both joins remain skew joins and results are unchanged.
New UTs in `RemoveRedundantSortsSuite` (both AE / non-AE variants):
- `remove local sort dangling below a shuffle that does not require
ordering` exercises the new removal branch directly on a physical plan (the
dangling-sort shape cannot be produced from SQL/DataFrame because the logical
`EliminateSorts` strips such a local sort before physical planning), covering
local-sort removal, global-sort preservation, and the rule-disabled case.
- `keep local sort satisfied only by a bucketed scan output ordering` is a
regression test for the reorder-vs-`DisableUnnecessaryBucketedScan` correctness
hazard; it fails with the old rule order and passes with the new one.
Also verified `RemoveRedundantSortsSuite` (both AE / non-AE variants),
`DisableUnnecessaryBucketedScanSuite`, and the full `AdaptiveQueryExecSuite`
pass.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 4.8)
Closes #57221 from ulysses-you/SPARK-58099.
Authored-by: Xiduo You <[email protected]>
Signed-off-by: Xiduo You <[email protected]>
(cherry picked from commit bc6ca483b224e14c7bf7f6c7879ffc726381703c)
Signed-off-by: Xiduo You <[email protected]>
---
.../spark/sql/execution/QueryExecution.scala | 12 ++--
.../spark/sql/execution/RemoveRedundantSorts.scala | 28 +++++++--
.../execution/adaptive/AdaptiveSparkPlanExec.scala | 7 ++-
.../sql/execution/RemoveRedundantSortsSuite.scala | 53 ++++++++++++++++-
.../adaptive/AdaptiveQueryExecSuite.scala | 69 ++++++++++++++++++++++
5 files changed, 157 insertions(+), 12 deletions(-)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala
index e37588faf8fb..2174c5899286 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala
@@ -767,12 +767,16 @@ object QueryExecution {
// `ReplaceHashWithSortAgg` needs to be added after `EnsureRequirements`
to guarantee the
// sort order of each node is checked to be valid.
ReplaceHashWithSortAgg,
- // `RemoveRedundantSorts` and `RemoveRedundantWindowGroupLimits` needs
to be added after
- // `EnsureRequirements` to guarantee the same number of partitions when
instantiating
- // PartitioningCollection.
- RemoveRedundantSorts,
+ // `RemoveRedundantWindowGroupLimits` needs to be added after
`EnsureRequirements` to
+ // guarantee the same number of partitions when instantiating
PartitioningCollection.
RemoveRedundantWindowGroupLimits,
DisableUnnecessaryBucketedScan,
+ // `RemoveRedundantSorts` also needs to run after `EnsureRequirements`
for the same reason.
+ // It must run after `DisableUnnecessaryBucketedScan`: disabling a
bucketed scan drops its
+ // output ordering, so running sort-removal first could strip a sort
that the scan appeared
+ // to satisfy and then silently lose that ordering. (This also matches
the AQE rule order,
+ // see `AdaptiveSparkPlanExec.queryStagePreparationRules`.)
+ RemoveRedundantSorts,
ApplyColumnarRulesAndInsertTransitions(
sparkSession.sessionState.columnarRules, outputsColumnar = false),
CollapseCodegenStages()) ++
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/RemoveRedundantSorts.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/RemoveRedundantSorts.scala
index 87c08ec865fe..85c29b230daf 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/RemoveRedundantSorts.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/RemoveRedundantSorts.scala
@@ -19,14 +19,25 @@ package org.apache.spark.sql.execution
import org.apache.spark.sql.catalyst.expressions.SortOrder
import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.execution.exchange.ShuffleExchangeLike
import org.apache.spark.sql.internal.SQLConf
/**
- * Remove redundant SortExec node from the spark plan. A sort node is
redundant when
- * its child satisfies both its sort orders and its required child
distribution. Note
- * this rule differs from the Optimizer rule EliminateSorts in that this rule
also checks
- * if the child satisfies the required distribution so that it is safe to
remove not only a
- * local sort but also a global sort when its child already satisfies required
sort orders.
+ * Remove redundant SortExec node from the spark plan. A sort node is
redundant when either:
+ * - its child satisfies both its sort orders and its required child
distribution. Note this
+ * rule differs from the Optimizer rule EliminateSorts in that this rule
also checks if the
+ * child satisfies the required distribution so that it is safe to remove
not only a local
+ * sort but also a global sort when its child already satisfies required
sort orders; or
+ * - it is a local sort that is the direct child of a shuffle which neither
requires its child
+ * to be ordered (empty `requiredChildOrdering`) nor exposes an ordering
itself (empty
+ * `outputOrdering`). A regular shuffle does not preserve the child
ordering, so such a local
+ * sort has no effect on the query result and is dead. The
`outputOrdering.isEmpty` guard is
+ * what keeps this safe for a custom `ShuffleExchangeLike` that does
preserve the child
+ * ordering: such a shuffle reports a non-empty `outputOrdering`, its local
sort is not dead
+ * and must be kept. This commonly happens in AQE after
`OptimizeSkewedJoin` inserts an extra
+ * shuffle between two joins: the local sort that used to feed the upper
join is left dangling
+ * right below the newly added shuffle and ends up being computed in the
wrong stage for
+ * nothing.
*/
object RemoveRedundantSorts extends Rule[SparkPlan] {
def apply(plan: SparkPlan): SparkPlan = {
@@ -42,5 +53,12 @@ object RemoveRedundantSorts extends Rule[SparkPlan] {
if SortOrder.orderingSatisfies(child.outputOrdering, orders) &&
child.outputPartitioning.satisfies(s.requiredChildDistribution.head)
=>
child
+
+ case shuffle: ShuffleExchangeLike
+ if shuffle.requiredChildOrdering.head.isEmpty &&
shuffle.outputOrdering.isEmpty =>
+ shuffle.child match {
+ case SortExec(_, false, sortChild, _) =>
shuffle.withNewChildren(Seq(sortChild))
+ case _ => shuffle
+ }
}
}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AdaptiveSparkPlanExec.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AdaptiveSparkPlanExec.scala
index 7040ab51cf51..f74115d98369 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AdaptiveSparkPlanExec.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AdaptiveSparkPlanExec.scala
@@ -132,10 +132,13 @@ case class AdaptiveSparkPlanExec(
// turn a hash aggregate into a sort aggregate.
ConvertSortMergeJoinToShuffledHashJoin(ensureRequirements),
ReplaceHashWithSortAgg,
- RemoveRedundantSorts,
RemoveRedundantWindowGroupLimits,
DisableUnnecessaryBucketedScan,
- OptimizeSkewedJoin(ensureRequirements)
+ OptimizeSkewedJoin(ensureRequirements),
+ // `RemoveRedundantSorts` runs after `OptimizeSkewedJoin` so that it can
also clean up the
+ // local sort left dangling right below the extra shuffle that skew join
optimization may
+ // insert between two joins.
+ RemoveRedundantSorts
) ++ context.session.sessionState.adaptiveRulesHolder.queryStagePrepRules
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/RemoveRedundantSortsSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/RemoveRedundantSortsSuite.scala
index 3cba30079cdb..9892a1c46242 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/RemoveRedundantSortsSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/RemoveRedundantSortsSuite.scala
@@ -18,11 +18,14 @@
package org.apache.spark.sql.execution
import org.apache.spark.sql.DataFrame
-import org.apache.spark.sql.catalyst.plans.physical.{RangePartitioning,
UnknownPartitioning}
+import org.apache.spark.sql.catalyst.expressions.{Ascending,
AttributeReference, SortOrder}
+import org.apache.spark.sql.catalyst.plans.physical.{HashPartitioning,
RangePartitioning, UnknownPartitioning}
import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanHelper,
DisableAdaptiveExecutionSuite, EnableAdaptiveExecutionSuite}
+import org.apache.spark.sql.execution.exchange.ShuffleExchangeExec
import org.apache.spark.sql.execution.joins.ShuffledJoin
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.IntegerType
abstract class RemoveRedundantSortsSuiteBase
@@ -139,6 +142,54 @@ abstract class RemoveRedundantSortsSuiteBase
}
}
+ test("SPARK-58099: remove local sort dangling below a shuffle that does not
require ordering") {
+ // A local sort directly below a round-robin repartition survives logical
optimization
+ // (`EliminateSorts` only recurses through
Sort/Join/order-irrelevant-Aggregate parents, and
+ // `CollapseRepartition` removes only a global sort under
`RepartitionByExpression` or any sort
+ // under `RebalancePartitions` -- never a sort under a round-robin
`Repartition`), so it reaches
+ // the physical plan as `Shuffle(RoundRobin) <- Sort(local)` and is
removed by the new branch.
+ // The source must be unsorted so the local sort is not itself elided as
redundant.
+ withTempView("t") {
+ Seq(3, 1, 2, 5, 4).toDF("key").createOrReplaceTempView("t")
+ val query = "SELECT /*+ REPARTITION(3) */ * FROM (SELECT key FROM t SORT
BY key)"
+ checkSorts(query, 0, 1)
+ }
+
+ // A global sort below such a shuffle must be kept: it carries its own
distribution requirement
+ // and is not the dead within-partition sort this rule targets. This shape
is not reachable via
+ // SQL/DataFrame, so the rule is applied directly to a physical plan.
+ val attr = AttributeReference("key", IntegerType)()
+ val scan = LocalTableScanExec(Seq(attr), Nil, None)
+ val globalSort = SortExec(SortOrder(attr, Ascending) :: Nil, global =
true, scan)
+ withSQLConf(SQLConf.REMOVE_REDUNDANT_SORTS_ENABLED.key -> "true") {
+ val danglingGlobal = ShuffleExchangeExec(HashPartitioning(Seq(attr), 5),
globalSort)
+
assert(RemoveRedundantSorts(danglingGlobal).find(_.isInstanceOf[SortExec]).isDefined)
+ }
+ }
+
+ test("SPARK-58099: keep local sort satisfied only by a bucketed scan output
ordering") {
+ // With `RemoveRedundantSorts` running before
`DisableUnnecessaryBucketedScan`, the local sort
+ // could be stripped based on the bucketed scan's output ordering and then
the ordering would
+ // be silently lost once `DisableUnnecessaryBucketedScan` disables the
scan. Running
+ // `RemoveRedundantSorts` after `DisableUnnecessaryBucketedScan` closes
that hole.
+ withSQLConf(
+ SQLConf.REMOVE_REDUNDANT_SORTS_ENABLED.key -> "true",
+ SQLConf.LEGACY_BUCKETED_TABLE_SCAN_OUTPUT_ORDERING.key -> "true",
+ SQLConf.AUTO_BUCKETED_SCAN_ENABLED.key -> "true") {
+ withTable("t") {
+ // A single file per bucket so the bucketed scan reports an output
ordering.
+ spark.range(100).selectExpr("id as i").repartition(1)
+ .write.format("parquet").bucketBy(8,
"i").sortBy("i").saveAsTable("t")
+ val df = sql("SELECT * FROM t SORT BY i")
+ val plan = df.queryExecution.executedPlan
+ // The bucketed scan is disabled since there is no
interesting-partition operator above it.
+ assert(collect(plan) { case s: FileSourceScanExec if s.bucketedScan =>
s }.isEmpty)
+ // The local sort must be kept, otherwise the within-partition
ordering would be lost.
+ assert(collect(plan) { case s: SortExec => s }.nonEmpty)
+ }
+ }
+ }
+
test("SPARK-33472: shuffled join with different left and right side
partition numbers") {
withTempView("t1", "t2") {
spark.range(0, 100, 1, 2).select($"id" as
"key").createOrReplaceTempView("t1")
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala
index 4b57546475ac..7446168c84fc 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala
@@ -831,6 +831,75 @@ class AdaptiveQueryExecSuite
}
}
+ test("SPARK-58099: Remove local sort dangling below the shuffle added by
skew join") {
+ // A ShuffledHashJoin feeding a SortMergeJoin in the same stage. When the
SHJ is skewed and
+ // skew-join optimization is force-applied, an extra shuffle is inserted
between the two joins.
+ // The local sort that used to feed the SMJ is then left dangling right
below that new shuffle,
+ // computed in the wrong stage for nothing. `RemoveRedundantSorts` should
strip it.
+
+ // Collect local sorts sitting directly below a shuffle. In the final plan
the sort may be
+ // wrapped in a WholeStageCodegenExec, so unwrap it before matching.
+ def collectDanglingSorts(plan: SparkPlan): Seq[SortExec] = {
+ def unwrap(p: SparkPlan): SparkPlan = p match {
+ case w: WholeStageCodegenExec => unwrap(w.child)
+ case other => other
+ }
+ collect(plan) {
+ case sh: ShuffleExchangeLike if
unwrap(sh.child).isInstanceOf[SortExec] &&
+ !unwrap(sh.child).asInstanceOf[SortExec].global =>
+ unwrap(sh.child).asInstanceOf[SortExec]
+ }
+ }
+
+ def runQuery(): SparkPlan = {
+ val q =
+ """
+ |SELECT /*+ SHUFFLE_HASH(skewData2), MERGE(data3) */ skewData1.key1
+ |FROM skewData1 JOIN skewData2 ON skewData1.key1 = skewData2.key2 -
100
+ |JOIN data3 ON skewData1.key1 = data3.key3 - 200
+ |""".stripMargin
+ val (_, adaptivePlan) = runAdaptiveAndVerifyResult(q)
+ adaptivePlan
+ }
+
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.ADAPTIVE_FORCE_OPTIMIZE_SKEWED_JOIN.key -> "true",
+ SQLConf.SKEW_JOIN_SKEWED_PARTITION_THRESHOLD.key -> "100",
+ SQLConf.ADVISORY_PARTITION_SIZE_IN_BYTES.key -> "100",
+ SQLConf.SHUFFLE_PARTITIONS.key -> "10",
+ SQLConf.COALESCE_PARTITIONS_ENABLED.key -> "false") {
+ withTempView("skewData1", "skewData2", "data3") {
+ spark.range(0, 300, 1, 10)
+ .selectExpr("id % 3 as key1", "id as
value1").createOrReplaceTempView("skewData1")
+ spark.range(0, 300, 1, 10)
+ .selectExpr("(id % 3) + 100 as key2", "id as value2")
+ .createOrReplaceTempView("skewData2")
+ spark.range(0, 300, 1, 10)
+ .selectExpr("(id % 3) + 200 as key3", "id as value3")
+ .createOrReplaceTempView("data3")
+
+ // Both joins should be optimized as skew joins so that the extra
shuffle is introduced.
+ val disabledPlan =
withSQLConf(SQLConf.REMOVE_REDUNDANT_SORTS_ENABLED.key -> "false") {
+ runQuery()
+ }
+ assert(collect(disabledPlan) { case j: ShuffledHashJoinExec => j
}.exists(_.isSkewJoin))
+ assert(collect(disabledPlan) { case j: SortMergeJoinExec => j
}.exists(_.isSkewJoin))
+ // Without the rule, a local sort dangles right below the shuffle
added on top of the SHJ.
+ assert(collectDanglingSorts(disabledPlan).nonEmpty)
+
+ val enabledPlan =
withSQLConf(SQLConf.REMOVE_REDUNDANT_SORTS_ENABLED.key -> "true") {
+ runQuery()
+ }
+ assert(collect(enabledPlan) { case j: ShuffledHashJoinExec => j
}.exists(_.isSkewJoin))
+ assert(collect(enabledPlan) { case j: SortMergeJoinExec => j
}.exists(_.isSkewJoin))
+ // With the rule, the dangling local sort below the shuffle is removed.
+ assert(collectDanglingSorts(enabledPlan).isEmpty)
+ }
+ }
+ }
+
test("SPARK-29544: adaptive skew join with different join types") {
Seq("SHUFFLE_MERGE", "SHUFFLE_HASH").foreach { joinHint =>
def getJoinNode(plan: SparkPlan): Seq[ShuffledJoin] = if (joinHint ==
"SHUFFLE_MERGE") {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
