You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ch...@apache.org on 2026/09/09 06:50:58 UTC
(spark) branch branch-4.x updated: [SPARK-59294][SQL] Support partially clustered distribution and skew join split for LeftSingle and ExistenceJoin
This is an automated email from the ASF dual-hosted git repository.
pan3793 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 556459a314d5 [SPARK-59294][SQL] Support partially clustered distribution and skew join split for LeftSingle and ExistenceJoin
556459a314d5 is described below
commit 556459a314d5c99a8f69d87274c81bf148e661b9
Author: Cheng Pan <pa...@gmail.com>
AuthorDate: Wed Sep 9 14:50:33 2026 +0800
[SPARK-59294][SQL] Support partially clustered distribution and skew join split for LeftSingle and ExistenceJoin
### What changes were proposed in this pull request?
Add `LeftSingle` and `ExistenceJoin` to the join types whose right side may be duplicated. Two rules ask that question and each kept its own copy of the answer:
- `EnsureRequirements`, which gates partially clustered distribution for storage-partitioned joins (`spark.sql.sources.v2.bucketing.partiallyClusteredDistribution.enabled`).
- `OptimizeSkewedJoin`, which gates splitting a skewed left partition under AQE.
Both rules act on shuffled joins only, so the two predicates now live once on `ShuffledJoin`: `canDuplicateRightSide` (`InnerLike`, `LeftOuter`, `LeftSingle`, `LeftExistence`) and `canDuplicateLeftSide` (`InnerLike`, `RightOuter`). Their docs state the invariant: duplicating a side replicates it over the other side's splits, so it is safe only when no output row comes from a row of that side alone. Duplicating the right qualifies every join that drops unmatched right rows; duplicating [...]
Raised in the review of #58514, which taught SPJ partition filtering about these join types: https://github.com/apache/spark/pull/58514#discussion_r3933601278
### Why are the changes needed?
`LeftSingle` and `ExistenceJoin` emit each left row exactly once against every right row for its key, the same shape as `LeftOuter`. Both optimizations were skipped outright for them, so a storage-partitioned existence join or scalar-subquery join with a partially clustered side fell back to one task per key group, and a skewed left partition of such a join was never split.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
- `KeyGroupedPartitioningSuite`: two new tests for an existence join and a left single join. With three splits for one key on the larger left side, partially clustered distribution keeps the left splits apart, replicates the right side and runs over five partitions instead of three key groups. With the right side larger, the left is picked for replication, which these join types do not allow, so both sides stay grouped by key. Both assert no shuffle and the query result.
- `AdaptiveQueryExecSuite`: the SPARK-29544 skew join test gains an existence join case for both shuffle hints and a left single join case for the hash hint (against a one-row-per-key right side, as that join errors on a second match), each asserting the planned join type and that the skewed left side is split with the right side untouched.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Fable 5.1)
Closes #58569 from pan3793/SPARK-59294.
Authored-by: Cheng Pan <pa...@gmail.com>
Signed-off-by: Cheng Pan <ch...@apache.org>
(cherry picked from commit b7d01b36d8ea2764f583caae503f042861a837f7)
Signed-off-by: Cheng Pan <ch...@apache.org>
---
.../execution/adaptive/OptimizeSkewedJoin.scala | 16 +--
.../execution/exchange/EnsureRequirements.scala | 19 +---
.../spark/sql/execution/joins/ShuffledJoin.scala | 26 ++++-
.../connector/KeyGroupedPartitioningSuite.scala | 111 +++++++++++++++++++++
.../adaptive/AdaptiveQueryExecSuite.scala | 30 +++++-
5 files changed, 171 insertions(+), 31 deletions(-)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/OptimizeSkewedJoin.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/OptimizeSkewedJoin.scala
index 7f55487407a7..cea464fa6b84 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/OptimizeSkewedJoin.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/OptimizeSkewedJoin.scala
@@ -29,7 +29,7 @@ import org.apache.spark.sql.catalyst.plans.physical.Partitioning
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.execution._
import org.apache.spark.sql.execution.exchange.{ENSURE_REQUIREMENTS, EnsureRequirements, ValidateRequirements}
-import org.apache.spark.sql.execution.joins.{BroadcastHashJoinExec, ShuffledHashJoinExec, SortMergeJoinExec}
+import org.apache.spark.sql.execution.joins.{BroadcastHashJoinExec, ShuffledHashJoinExec, ShuffledJoin, SortMergeJoinExec}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.util.Utils
@@ -82,15 +82,6 @@ case class OptimizeSkewedJoin(ensureRequirements: EnsureRequirements)
}
}
- private def canSplitLeftSide(joinType: JoinType) = {
- joinType == Inner || joinType == Cross || joinType == LeftSemi ||
- joinType == LeftAnti || joinType == LeftOuter
- }
-
- private def canSplitRightSide(joinType: JoinType) = {
- joinType == Inner || joinType == Cross || joinType == RightOuter
- }
-
private def getSizeInfo(medianSize: Long, sizes: Array[Long]): String = {
s"median size: $medianSize, max size: ${sizes.max}, min size: ${sizes.min}, avg size: " +
sizes.sum / sizes.length
@@ -112,8 +103,9 @@ case class OptimizeSkewedJoin(ensureRequirements: EnsureRequirements)
left: ShuffleQueryStageExec,
right: ShuffleQueryStageExec,
joinType: JoinType): Option[(SparkPlan, SparkPlan)] = {
- val canSplitLeft = canSplitLeftSide(joinType)
- val canSplitRight = canSplitRightSide(joinType)
+ // Splitting a side replicates the other side's partition to every split.
+ val canSplitLeft = ShuffledJoin.canDuplicateRightSide(joinType)
+ val canSplitRight = ShuffledJoin.canDuplicateLeftSide(joinType)
if (!canSplitLeft && !canSplitRight) return None
val leftSizes = left.mapStats.get.bytesByPartitionId
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala
index 271654f51534..36265f38561d 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala
@@ -30,7 +30,7 @@ import org.apache.spark.sql.catalyst.util.InternalRowComparableWrapper
import org.apache.spark.sql.errors.QueryExecutionErrors
import org.apache.spark.sql.execution._
import org.apache.spark.sql.execution.datasources.v2.GroupPartitionsExec
-import org.apache.spark.sql.execution.joins.{ShuffledHashJoinExec, SortMergeJoinExec}
+import org.apache.spark.sql.execution.joins.{ShuffledHashJoinExec, ShuffledJoin, SortMergeJoinExec}
import org.apache.spark.sql.internal.SQLConf
/**
@@ -642,8 +642,8 @@ case class EnsureRequirements(
// optimization cannot be applied to a left outer join, where the left hand
// side is chosen as the side to replicate partitions according to stats.
// Otherwise, query result could be incorrect.
- val canReplicateLeft = canReplicateLeftSide(joinType)
- val canReplicateRight = canReplicateRightSide(joinType)
+ val canReplicateLeft = ShuffledJoin.canDuplicateLeftSide(joinType)
+ val canReplicateRight = ShuffledJoin.canDuplicateRightSide(joinType)
if (!canReplicateLeft && !canReplicateRight) {
logInfo(log"Skipping partially clustered distribution as it cannot be applied for " +
@@ -695,7 +695,7 @@ case class EnsureRequirements(
// Similar to skewed join, we need to check the join type to see whether replication
// of partitions can be applied. For instance, replication should not be allowed for
- // the left-hand side of a right outer join.
+ // the left-hand side of a left outer join.
if (replicateLeftSide && !canReplicateLeft) {
logInfo(log"Left-hand side is picked but cannot be applied to join type " +
log"'${MDC(LogKeys.JOIN_TYPE, joinType)}'. Skipping partially clustered " +
@@ -777,17 +777,6 @@ case class EnsureRequirements(
}
}
- // Similar to `OptimizeSkewedJoin.canSplitRightSide`
- private def canReplicateLeftSide(joinType: JoinType): Boolean = {
- joinType == Inner || joinType == Cross || joinType == RightOuter
- }
-
- // Similar to `OptimizeSkewedJoin.canSplitLeftSide`
- private def canReplicateRightSide(joinType: JoinType): Boolean = {
- joinType == Inner || joinType == Cross || joinType == LeftSemi ||
- joinType == LeftAnti || joinType == LeftOuter
- }
-
/**
* The innermost `GroupPartitionsExec` reachable from `plan` by descending only through nodes
* this rule itself inserted above it, together with a function rebuilding the traversed local
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/ShuffledJoin.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/ShuffledJoin.scala
index 0318fe88a64b..9b6bbeeba66c 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/ShuffledJoin.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins/ShuffledJoin.scala
@@ -18,7 +18,7 @@
package org.apache.spark.sql.execution.joins
import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression}
-import org.apache.spark.sql.catalyst.plans.{ExistenceJoin, FullOuter, InnerLike, LeftAnti, LeftExistence, LeftOuter, LeftSingle, RightOuter}
+import org.apache.spark.sql.catalyst.plans.{ExistenceJoin, FullOuter, InnerLike, JoinType, LeftAnti, LeftExistence, LeftOuter, LeftSingle, RightOuter}
import org.apache.spark.sql.catalyst.plans.physical.{ClusteredDistribution, Distribution, KeyedPartitioning, Partitioning, PartitioningCollection, UnknownPartitioning, UnspecifiedDistribution}
import org.apache.spark.sql.internal.SQLConf
@@ -135,3 +135,27 @@ trait ShuffledJoin extends JoinCodegenSupport {
}
}
}
+
+object ShuffledJoin {
+ /**
+ * Whether replicating the right side over splits of the left cannot change the result. The
+ * right partition then reaches every left split, so no output row may come from a right row
+ * alone. Every join that drops unmatched right rows qualifies: its output is one row per left
+ * row or one per matching pair, and each left row still lands in exactly one split.
+ */
+ def canDuplicateRightSide(joinType: JoinType): Boolean = joinType match {
+ case _: InnerLike | LeftOuter | LeftSingle | LeftExistence(_) => true
+ case _ => false
+ }
+
+ /**
+ * Whether replicating the left side over splits of the right cannot change the result. Every
+ * output row must then be tied to one right row. The left-preserving joins are out for that
+ * reason, and so is LeftSemi: it drops unmatched left rows yet emits one row per left row, so
+ * a left row matching in two right splits would come out twice.
+ */
+ def canDuplicateLeftSide(joinType: JoinType): Boolean = joinType match {
+ case _: InnerLike | RightOuter => true
+ case _ => false
+ }
+}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala
index 3cdd8d7181cb..3aa2c9422446 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala
@@ -654,6 +654,66 @@ class KeyGroupedPartitioningSuite
s"expected $expectedNumGroups key groups, got ${actualNumGroups.mkString(", ")}")
}
+ /** An `items` to `purchases` existence join, planned by the `EXISTS` in a disjunction. */
+ private val existenceJoinQuery =
+ s"""
+ |SELECT id, name FROM testcat.ns.$items i
+ |WHERE EXISTS (SELECT /*+ MERGE(p) */ 1 FROM testcat.ns.$purchases p
+ | WHERE i.id = p.item_id)
+ | OR i.name = 'bb'
+ |""".stripMargin
+
+ /**
+ * An `items` to `purchases` left single join, planned by a correlated scalar subquery not proven
+ * to return one row. It cannot sort-merge, hence the shuffled hash join hint.
+ */
+ private val leftSingleJoinQuery =
+ s"""
+ |SELECT id, name,
+ | (SELECT /*+ SHUFFLE_HASH(p) */ p.price FROM testcat.ns.$purchases p
+ | WHERE i.id = p.item_id) AS sale_price
+ |FROM testcat.ns.$items i
+ |""".stripMargin
+
+ /**
+ * Runs `query` with partially clustered distribution on and off, asserting `expectedRows`, that
+ * its one shuffled join accepted by `isJoinType` has no shuffle under it, and per setting what
+ * `expected` gives: whether the left grouping distributes its splits (the right never does) and
+ * both sides' partition count.
+ */
+ private def checkPartiallyClusteredJoin(
+ query: String,
+ isJoinType: JoinType => Boolean,
+ expectedRows: Seq[Row],
+ expected: Boolean => (Boolean, Int)): Unit = {
+ Seq(true, false).foreach { partiallyClustered =>
+ val (expectedLeftDistributed, expectedNumPartitions) = expected(partiallyClustered)
+ withSQLConf(
+ SQLConf.V2_BUCKETING_PUSH_PART_VALUES_ENABLED.key -> "true",
+ SQLConf.V2_BUCKETING_PARTITION_FILTER_ENABLED.key -> "true",
+ SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key ->
+ partiallyClustered.toString,
+ SQLConf.SCALAR_SUBQUERY_USE_SINGLE_JOIN.key -> "true") {
+ val df = sql(query)
+ checkAnswer(df, expectedRows)
+
+ val plan = df.queryExecution.executedPlan
+ val joins = collect(plan) { case j: ShuffledJoin if isJoinType(j.joinType) => j }
+ assert(joins.size == 1, s"expected one matching join in\n$plan")
+ assert(collectAllShuffles(joins.head).isEmpty,
+ "should not add shuffle for both sides of the join")
+ val sides = Seq(joins.head.left, joins.head.right).map(collectAllGroupPartitions)
+ val distributed = sides.map(_.map(_.distributePartitions))
+ assert(distributed == Seq(Seq(expectedLeftDistributed), Seq(false)),
+ s"left/right distributePartitions with partially clustered $partiallyClustered: " +
+ distributed)
+ val numPartitions = sides.flatten.map(_.outputPartitioning.numPartitions)
+ assert(numPartitions == Seq(expectedNumPartitions, expectedNumPartitions),
+ s"left/right partitions with partially clustered $partiallyClustered: $numPartitions")
+ }
+ }
+ }
+
/**
* Creates a table partitioned by `bucket(numBuckets, id)` and holding the ids 0 until `numIds`.
* Joining two such tables reduces both sides onto the greatest common divisor of their bucket
@@ -4382,6 +4442,57 @@ class KeyGroupedPartitioningSuite
}
}
+ test("SPARK-59294: partially clustered distribution with existence join and left single join") {
+ createTable(items, itemsColumns, Array(bucket(8, "id")))
+ sql(s"INSERT INTO testcat.ns.$items VALUES " +
+ s"(0, 'aa', 38.0, cast('2020-01-01' as timestamp)), " +
+ s"(1, 'bb', 39.0, cast('2020-01-02' as timestamp)), " +
+ s"(4, 'cc', 40.0, cast('2020-01-02' as timestamp)), " +
+ s"(4, 'dd', 41.0, cast('2020-01-03' as timestamp)), " +
+ s"(4, 'ee', 42.0, cast('2020-01-04' as timestamp))")
+
+ createTable(purchases, purchasesColumns, Array(bucket(8, "item_id")))
+ sql(s"INSERT INTO testcat.ns.$purchases VALUES " +
+ s"(4, 42.0, cast('2020-01-01' as timestamp)), " +
+ s"(5, 44.0, cast('2020-01-15' as timestamp))")
+
+ // `items` holds three splits for key 4 and is the larger side, so partially clustered
+ // distribution keeps its splits apart and replicates `purchases` to each of them. Each left
+ // row still lands in one task that sees every right row for its key, which is all these join
+ // types need, so the join runs over five partitions instead of three key groups.
+ checkPartiallyClusteredJoin(existenceJoinQuery, _.isInstanceOf[ExistenceJoin],
+ Seq(Row(1, "bb"), Row(4, "cc"), Row(4, "dd"), Row(4, "ee")),
+ expected = on => if (on) (true, 5) else (false, 3))
+ checkPartiallyClusteredJoin(leftSingleJoinQuery, _ == LeftSingle,
+ Seq(Row(0, "aa", null), Row(1, "bb", null), Row(4, "cc", 42.0), Row(4, "dd", 42.0),
+ Row(4, "ee", 42.0)),
+ expected = on => if (on) (true, 5) else (false, 3))
+ }
+
+ test("SPARK-59294: partially clustered distribution is skipped when the smaller side is the " +
+ "left of an existence join or a left single join") {
+ createTable(items, itemsColumns, Array(bucket(8, "id")))
+ sql(s"INSERT INTO testcat.ns.$items VALUES " +
+ s"(0, 'aa', 38.0, cast('2020-01-01' as timestamp)), " +
+ s"(1, 'bb', 39.0, cast('2020-01-02' as timestamp)), " +
+ s"(4, 'cc', 40.0, cast('2020-01-02' as timestamp))")
+
+ // One row per item id, as a left single join errors on a second match. Eight rows outweigh
+ // the three `items` rows, so `items` is the side picked for replication.
+ createTable(purchases, purchasesColumns, Array(bucket(8, "item_id")))
+ sql(s"INSERT INTO testcat.ns.$purchases VALUES " +
+ (4 to 11).map(id => s"($id, ${id + 38}.0, cast('2020-01-01' as timestamp))")
+ .mkString(", "))
+
+ // Only the right side may be replicated for these join types, so picking the left leaves
+ // both sides grouped by key whether or not the distribution is enabled.
+ checkPartiallyClusteredJoin(existenceJoinQuery, _.isInstanceOf[ExistenceJoin],
+ Seq(Row(1, "bb"), Row(4, "cc")), expected = _ => (false, 3))
+ checkPartiallyClusteredJoin(leftSingleJoinQuery, _ == LeftSingle,
+ Seq(Row(0, "aa", null), Row(1, "bb", null), Row(4, "cc", 42.0)),
+ expected = _ => (false, 3))
+ }
+
test("SPARK-53322: checkpointed scans avoid shuffles for aggregates") {
withTempDir { dir =>
spark.sparkContext.setCheckpointDir(dir.getPath)
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 2715bee1e4e1..3fad780013fa 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
@@ -36,7 +36,7 @@ import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.{Ascending, Attribute, AttributeReference, EqualTo, IsNull, Literal, Or, SortOrder}
import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight, EliminateLimits}
-import org.apache.spark.sql.catalyst.plans.{Inner, LeftAnti, LeftSemi}
+import org.apache.spark.sql.catalyst.plans.{ExistenceJoin, Inner, LeftAnti, LeftSemi, LeftSingle}
import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, EmptyRelation, GlobalLimit, Join, JoinHint, LeafNode, LocalRelation, LogicalPlan, Statistics}
import org.apache.spark.sql.catalyst.plans.physical.{CoalescedNullAwareHashPartitioning, SinglePartition}
import org.apache.spark.sql.classic.Strategy
@@ -1983,8 +1983,9 @@ class AdaptiveQueryExecSuite
SQLConf.COALESCE_PARTITIONS_MIN_PARTITION_NUM.key -> "1",
SQLConf.SHUFFLE_PARTITIONS.key -> "100",
SQLConf.SKEW_JOIN_SKEWED_PARTITION_THRESHOLD.key -> "800",
- SQLConf.ADVISORY_PARTITION_SIZE_IN_BYTES.key -> "800") {
- withTempView("skewData1", "skewData2") {
+ SQLConf.ADVISORY_PARTITION_SIZE_IN_BYTES.key -> "800",
+ SQLConf.SCALAR_SUBQUERY_USE_SINGLE_JOIN.key -> "true") {
+ withTempView("skewData1", "skewData2", "skewData3") {
spark
.range(0, 1000, 1, 10)
.select(
@@ -2000,6 +2001,11 @@ class AdaptiveQueryExecSuite
.otherwise($"id").as("key2"),
$"id" as "value2")
.createOrReplaceTempView("skewData2")
+ // one row per key, as a left single join errors on a second match
+ spark
+ .range(0, 1000, 1, 10)
+ .select($"id" as "key3", $"id" as "value3")
+ .createOrReplaceTempView("skewData3")
def checkSkewJoin(
joins: Seq[ShuffledJoin],
@@ -2038,6 +2044,24 @@ class AdaptiveQueryExecSuite
"RIGHT OUTER JOIN skewData2 ON key1 = key2")
val rightJoin = getJoinNode(rightAdaptivePlan)
checkSkewJoin(rightJoin, 0, 1)
+
+ // skewed existence join optimization, splitting the left side like a left outer join
+ val (_, existenceAdaptivePlan) = runAdaptiveAndVerifyResult(
+ s"SELECT * FROM skewData1 WHERE EXISTS (SELECT /*+ $joinHint(skewData2) */ 1 " +
+ "FROM skewData2 WHERE key1 = key2) OR value1 < 0")
+ val existenceJoin = getJoinNode(existenceAdaptivePlan)
+ assert(existenceJoin.head.joinType.isInstanceOf[ExistenceJoin])
+ checkSkewJoin(existenceJoin, 2, 0)
+
+ // skewed left single join optimization, hash join only as it cannot sort-merge
+ if (joinHint == "SHUFFLE_HASH") {
+ val (_, singleAdaptivePlan) = runAdaptiveAndVerifyResult(
+ s"SELECT key1, value1, (SELECT /*+ $joinHint(skewData3) */ value3 " +
+ "FROM skewData3 WHERE key3 = key1) FROM skewData1")
+ val singleJoin = getJoinNode(singleAdaptivePlan)
+ assert(singleJoin.head.joinType == LeftSingle)
+ checkSkewJoin(singleJoin, 2, 0)
+ }
}
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org