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 d1c47bb318ee [SPARK-58558][SQL] Make
requireAllClusterKeysForCoPartition check key coverage instead of exact match
for SPJ
d1c47bb318ee is described below
commit d1c47bb318eef2abc4070ea2976ac114f8cee040
Author: Cheng Pan <[email protected]>
AuthorDate: Thu Aug 6 11:45:14 2026 +0800
[SPARK-58558][SQL] Make requireAllClusterKeysForCoPartition check key
coverage instead of exact match for SPJ
### What changes were proposed in this pull request?
For storage-partitioned joins, change the
`spark.sql.requireAllClusterKeysForCoPartition` check in
`EnsureRequirements.createKeyedShuffleSpec` from a positional exact match to a
coverage check: every join key must be covered by some partition key, rather
than matching the partition keys positionally.
The config still gates `HashPartitioning` shuffle elimination (e.g., V1
bucketing) with the exact-match semantics, where the positional match has no
reachable downside (hash partitioning cannot repeat a column, and key order is
normalized by `reorderJoinPredicates`). Its `SQLConf` doc, the SPJ tuning
guide, and the migration guide are updated accordingly; the tuning guide also
switches to the current name of
`spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled` (the
[...]
### Why are the changes needed?
This is a pure relaxation of the config-`true` path: everything it enables
was already reachable with `requireAllClusterKeysForCoPartition=false`, but
flipping that guard off is global and also admits the genuinely risky case
(partition keys covering only part of the join keys). Two skew-free shapes no
longer require it:
1. A join-key column partitioned by more than one transform, e.g. both
tables `PARTITIONED BY (bucket(8, id), identity(id))` joined `ON t1.id =
t2.id`: the partition expressions outnumber the join keys, so the positional
check failed, although the partition tuple is a function of the join key and
carries no skew risk.
2. Join keys that are a strict subset of the partition keys
(`v2BucketingAllowKeysSubsetOfPartitionKeys=true`): the extra keys are
partition keys, which coverage does not restrict, so the config no longer needs
`requireAllClusterKeysForCoPartition=false` alongside it. When the join keys
only partially overlap the partition keys, `false` is still required, as before.
The genuinely risky case -- partition keys covering only part of the join
keys, where eliminating the shuffle drops the join's parallelism to the storage
layout's partition count -- remains gated by the config, as before.
### Does this PR introduce _any_ user-facing change?
Yes. SPJ now works by default when a join-key column is partitioned by more
than one transform, and when the join keys are a subset of the partition keys,
`spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled=true` no
longer needs `spark.sql.requireAllClusterKeysForCoPartition=false` alongside
it. As before, when the partition keys cover only part of the join keys, SPJ
still requires `spark.sql.requireAllClusterKeysForCoPartition=false`.
### How was this patch tested?
- `EnsureRequirementsSuite`: added a multi-transform-column case (partition
expressions outnumber the join keys) that works with default configs; the
partition-keys-cover-part-of-join-keys cases assert shuffle by default and SPJ
with `requireAllClusterKeysForCoPartition=false`; a duplicated-join-keys plan
(producible only with `BooleanSimplification` excluded via
`spark.sql.optimizer.excludedRules`, or by hand-built plans) keeps
planner-robustness coverage under both config values.
- `KeyGroupedPartitioningSuite`: added an end-to-end SPJ test joining on a
column partitioned by two transforms with all-default configs. Of the 20
`requireAllClusterKeysForCoPartition=false` overrides on the base branch, 12
are removed -- 7 as load-bearing demonstrations of the subset-key win (each
also sets `allowKeysSubsetOfPartitionKeys=true`) and 5 that were already no-ops
(4 where the join keys matched the partition keys positionally, 1 where the
one-side-shuffle path is not gat [...]
- `ShuffleSpecSuite`: unchanged (covers `HashShuffleSpec`, which keeps the
exact-match semantics).
- All three suites pass.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Fable 5)
Closes #57762 from pan3793/SPARK-58558.
Authored-by: Cheng Pan <[email protected]>
Signed-off-by: Cheng Pan <[email protected]>
(cherry picked from commit fbe9fbc1cb12374b939b3ecc851f687a05303103)
Signed-off-by: Cheng Pan <[email protected]>
---
docs/sql-migration-guide.md | 4 ++
docs/sql-performance-tuning.md | 9 ++-
.../org/apache/spark/sql/internal/SQLConf.scala | 6 +-
.../execution/exchange/EnsureRequirements.scala | 23 +++----
.../connector/KeyGroupedPartitioningSuite.scala | 34 +++++----
.../exchange/EnsureRequirementsSuite.scala | 80 ++++++++++++++--------
6 files changed, 99 insertions(+), 57 deletions(-)
diff --git a/docs/sql-migration-guide.md b/docs/sql-migration-guide.md
index 40fd895d9983..b267a5d6fa45 100644
--- a/docs/sql-migration-guide.md
+++ b/docs/sql-migration-guide.md
@@ -22,6 +22,10 @@ license: |
* Table of contents
{:toc}
+## Upgrading from Spark SQL 4.3 to 4.4
+
+- Since Spark 4.4, for storage-partitioned joins,
`spark.sql.requireAllClusterKeysForCoPartition` requires every join key to be
covered by some partition key instead of matching the partition keys
positionally. As a result, a join-key column partitioned by more than one
transform no longer prevents shuffle elimination, and
`spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled` no
longer additionally requires `spark.sql.requireAllClusterKeysForCoPartition` to
be `false` w [...]
+
## Upgrading from Spark SQL 4.2 to 4.3
- Since Spark 4.3, [ASOF JOIN](sql-ref-syntax-qry-select-asof-join.html) is
available as an opt-in SQL feature gated by `spark.sql.join.asofJoin.enabled`
(default `false`). When disabled, `ASOF JOIN` fails at parse time with
`UNSUPPORTED_FEATURE.ASOF_JOIN`.
diff --git a/docs/sql-performance-tuning.md b/docs/sql-performance-tuning.md
index e0cba9362736..3822c56bc3b9 100644
--- a/docs/sql-performance-tuning.md
+++ b/docs/sql-performance-tuning.md
@@ -582,9 +582,9 @@ The following SQL properties enable Storage Partition Join
in different join que
<td><code>spark.sql.requireAllClusterKeysForCoPartition</code></td>
<td>true</td>
<td>
- When true, require the join or MERGE keys to be same and in the same
order as the partition keys to eliminate shuffle. Hence, set to <b>false</b> in
this situation to eliminate shuffle.
+ When true, storage-partitioned join requires every join or MERGE key
to be covered by some partition key (rather than matching the partition keys
positionally) to eliminate shuffle. When the partition keys cover only part of
the join or MERGE keys, set to <b>false</b> to eliminate shuffle, at the risk
of data skew and reduced parallelism from the coarser storage partitioning.
</td>
- <td>3.4.0</td>
+ <td>3.3.0</td>
</tr>
<tr>
<td><code>spark.sql.sources.v2.bucketing.partiallyClusteredDistribution.enabled</code></td>
@@ -595,10 +595,10 @@ The following SQL properties enable Storage Partition
Join in different join que
<td>3.4.0</td>
</tr>
<tr>
-
<td><code>spark.sql.sources.v2.bucketing.allowJoinKeysSubsetOfPartitionKeys.enabled</code></td>
+
<td><code>spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled</code></td>
<td>false</td>
<td>
- When enabled, try to avoid shuffle if join or MERGE condition does not
include all partition columns. This config requires both
<code>spark.sql.sources.v2.bucketing.enabled</code> and
<code>spark.sql.sources.v2.bucketing.pushPartValues.enabled</code> to be true,
and <code>spark.sql.requireAllClusterKeysForCoPartition</code> to be false.
+ When enabled, try to avoid shuffle if join or MERGE condition does not
include all partition columns. This config requires both
<code>spark.sql.sources.v2.bucketing.enabled</code> and
<code>spark.sql.sources.v2.bucketing.pushPartValues.enabled</code> to be true.
</td>
<td>4.0.0</td>
</tr>
@@ -654,7 +654,6 @@ ON t.dep = s.dep AND t.id = s.id
SET 'spark.sql.sources.v2.bucketing.enabled' 'true'
SET 'spark.sql.iceberg.planning.preserve-data-grouping' 'true'
SET 'spark.sql.sources.v2.bucketing.pushPartValues.enabled' 'true'
-SET 'spark.sql.requireAllClusterKeysForCoPartition' 'false'
SET 'spark.sql.sources.v2.bucketing.partiallyClusteredDistribution.enabled'
'true'
-- Plan with Storage Partition Join
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
index 8a0d88346b63..536b3af02611 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
@@ -1105,7 +1105,11 @@ object SQLConf {
.doc("When true, the planner requires all the clustering keys as the
hash partition keys " +
"of the children, to eliminate the shuffles for the operator that
needs its children to " +
"be co-partitioned, such as JOIN node. This is to avoid data skews
which can lead to " +
- "significant performance regression if shuffles are eliminated.")
+ "significant performance regression if shuffles are eliminated. For
storage-partitioned " +
+ "join, every clustering key must be covered by some partition key,
rather than matching " +
+ "the partition keys positionally, so a column partitioned by more than
one transform " +
+ "does not prevent shuffle elimination; hash partitioning deliberately
keeps the " +
+ "positional match.")
.version("3.3.0")
.booleanConf
.createWithDefault(true)
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 8ca344e51194..52a761090b0b 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
@@ -782,20 +782,19 @@ case class EnsureRequirements(
partitioning: Partitioning,
distribution: ClusteredDistribution): Option[KeyedShuffleSpec] = {
def tryCreate(partitioning: KeyedPartitioning): Option[KeyedShuffleSpec] =
{
- // The single-column invariant in KeyedPartitioning.supportsExpressions
guarantees one
- // attribute per partition expression.
- val attributes = partitioning.expressions.flatMap(_.references)
- val clustering = distribution.clustering
-
- val satisfies = if
(SQLConf.get.getConf(SQLConf.REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION)) {
- attributes.length == clustering.length &&
attributes.zip(clustering).forall {
- case (l, r) => l.semanticEquals(r)
- }
- } else {
- partitioning.satisfies(distribution)
+ // The config requires all the cluster keys to be covered by the
partition keys, to avoid
+ // the skew of joining on keys that are coarser than the join keys. Key
order and duplicated
+ // cluster keys don't matter.
+ def allClusterKeysCovered: Boolean = {
+ // The single-column invariant in
KeyedPartitioning.supportsExpressions guarantees one
+ // attribute per partition expression.
+ val attributes = partitioning.expressions.flatMap(_.references)
+ distribution.clustering.forall(c =>
attributes.exists(_.semanticEquals(c)))
}
- if (satisfies) {
+ if (partitioning.satisfies(distribution) &&
+
(!SQLConf.get.getConf(SQLConf.REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION) ||
+ allClusterKeysCovered)) {
Some(partitioning.createShuffleSpec(distribution).asInstanceOf[KeyedShuffleSpec])
} else {
None
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 ceaac9729ddd..a732294786b2 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
@@ -1199,7 +1199,6 @@ class KeyGroupedPartitioningSuite extends
DistributionAndOrderingSuiteBase with
enable <- Seq("true", "false")
} yield {
withSQLConf(
- SQLConf.REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION.key ->
false.toString,
SQLConf.V2_BUCKETING_PUSH_PART_VALUES_ENABLED.key ->
pushDownValues.toString,
SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key ->
enable) {
// The left side uses a key-grouped partitioning to satisfy the WINDOW
function's
@@ -1830,7 +1829,6 @@ class KeyGroupedPartitioningSuite extends
DistributionAndOrderingSuiteBase with
Seq(true, false).foreach { partiallyClustered =>
Seq(true, false).foreach { allowKeysSubsetOfPartitionKeys =>
withSQLConf(
- SQLConf.REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION.key -> "false",
SQLConf.V2_BUCKETING_PUSH_PART_VALUES_ENABLED.key ->
pushDownValues.toString,
SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key ->
partiallyClustered.toString,
@@ -1972,7 +1970,6 @@ class KeyGroupedPartitioningSuite extends
DistributionAndOrderingSuiteBase with
Seq(true, false).foreach { allowKeysSubsetOfPartitionKeys =>
withSQLConf(
- SQLConf.REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION.key -> "false",
SQLConf.V2_BUCKETING_PUSH_PART_VALUES_ENABLED.key -> "true",
SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key
-> "false",
SQLConf.V2_BUCKETING_ALLOW_KEYS_SUBSET_OF_PARTITION_KEYS.key ->
@@ -2132,7 +2129,6 @@ class KeyGroupedPartitioningSuite extends
DistributionAndOrderingSuiteBase with
Seq(true, false).foreach { allowKeysSubsetOfPartitionKeys =>
withSQLConf(
- SQLConf.REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION.key -> "false",
SQLConf.V2_BUCKETING_PUSH_PART_VALUES_ENABLED.key -> "true",
SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key
-> "false",
SQLConf.V2_BUCKETING_ALLOW_KEYS_SUBSET_OF_PARTITION_KEYS.key ->
@@ -2310,7 +2306,6 @@ class KeyGroupedPartitioningSuite extends
DistributionAndOrderingSuiteBase with
withSQLConf(
- SQLConf.REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION.key -> "false",
SQLConf.V2_BUCKETING_PUSH_PART_VALUES_ENABLED.key -> "true",
SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key ->
"false",
SQLConf.V2_BUCKETING_ALLOW_KEYS_SUBSET_OF_PARTITION_KEYS.key ->
"true",
@@ -2375,7 +2370,6 @@ class KeyGroupedPartitioningSuite extends
DistributionAndOrderingSuiteBase with
Seq(true, false).foreach{ allowPushDown =>
Seq(true, false).foreach{ partiallyClustered =>
withSQLConf(
- SQLConf.REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION.key -> "false",
SQLConf.V2_BUCKETING_PUSH_PART_VALUES_ENABLED.key ->
allowPushDown.toString,
SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key ->
partiallyClustered.toString,
@@ -2433,7 +2427,6 @@ class KeyGroupedPartitioningSuite extends
DistributionAndOrderingSuiteBase with
Seq(true, false).foreach { partiallyClustered =>
Seq(true, false).foreach { allowKeysSubsetOfPartitionKeys =>
withSQLConf(
- SQLConf.REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION.key -> "false",
SQLConf.V2_BUCKETING_PUSH_PART_VALUES_ENABLED.key ->
pushDownValues.toString,
SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key
->
@@ -2496,7 +2489,6 @@ class KeyGroupedPartitioningSuite extends
DistributionAndOrderingSuiteBase with
Seq(true, false).foreach { allowKeysSubsetOfPartitionKeys =>
withSQLConf(
- SQLConf.REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION.key -> "false",
SQLConf.V2_BUCKETING_PUSH_PART_VALUES_ENABLED.key ->
pushDownValues.toString,
SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key
->
partiallyClustered.toString,
@@ -2708,7 +2700,6 @@ class KeyGroupedPartitioningSuite extends
DistributionAndOrderingSuiteBase with
"(6, 50.0, cast('2023-02-01' as timestamp))")
withSQLConf(
- SQLConf.REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION.key -> "false",
SQLConf.V2_BUCKETING_SHUFFLE_ENABLED.key -> "true",
SQLConf.V2_BUCKETING_PUSH_PART_VALUES_ENABLED.key -> "true",
SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key ->
"false",
@@ -3210,7 +3201,6 @@ class KeyGroupedPartitioningSuite extends
DistributionAndOrderingSuiteBase with
test("SPARK-55411: Fix ArrayIndexOutOfBoundsException when join keys " +
"are less than cluster keys") {
withSQLConf(
- SQLConf.REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION.key -> "false",
SQLConf.V2_BUCKETING_SHUFFLE_ENABLED.key -> "true",
SQLConf.V2_BUCKETING_PUSH_PART_VALUES_ENABLED.key -> "true",
SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key ->
"false",
@@ -3286,7 +3276,6 @@ class KeyGroupedPartitioningSuite extends
DistributionAndOrderingSuiteBase with
test("SPARK-55535: Multi table join granular partition grouping") {
withSQLConf(
- SQLConf.REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION.key -> "false",
SQLConf.V2_BUCKETING_ALLOW_KEYS_SUBSET_OF_PARTITION_KEYS.key -> "true",
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
val items_partitions = Array(identity("id"), years("arrive_time"))
@@ -3482,7 +3471,6 @@ class KeyGroupedPartitioningSuite extends
DistributionAndOrderingSuiteBase with
createTable(purchases, purchasesColumns, purchases_partitions)
sql(s"INSERT INTO testcat.ns.$purchases VALUES (2, 10.0, cast('2021-01-01'
as timestamp))")
withSQLConf(
- SQLConf.REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION.key -> "false",
SQLConf.V2_BUCKETING_ALLOW_KEYS_SUBSET_OF_PARTITION_KEYS.key -> "true",
SQLConf.V2_BUCKETING_ALLOW_COMPATIBLE_TRANSFORMS.key -> "true") {
val df = sql(
@@ -4495,4 +4483,26 @@ class KeyGroupedPartitioningSuite extends
DistributionAndOrderingSuiteBase with
}
}
}
+
+ test("SPARK-58558: SPJ on a join key column partitioned by multiple
transforms") {
+ // Partition expressions outnumber the join keys because `id` is
partitioned twice, but
+ // every join key is covered, so SPJ works with default configs.
+ val items_partitions = Array(bucket(8, "id"), identity("id"))
+ createTable(items, itemsColumns, items_partitions)
+ sql(s"INSERT INTO testcat.ns.$items VALUES " +
+ "(1, 'aa', 40.0, cast('2020-01-01' as timestamp)), " +
+ "(2, 'bb', 10.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 'cc', 15.5, cast('2020-02-01' as timestamp))")
+
+ val purchases_partitions = Array(bucket(8, "item_id"), identity("item_id"))
+ createTable(purchases, purchasesColumns, purchases_partitions)
+ sql(s"INSERT INTO testcat.ns.$purchases VALUES " +
+ "(1, 42.0, cast('2020-01-01' as timestamp)), " +
+ "(2, 19.5, cast('2020-02-01' as timestamp))")
+
+ val df = createJoinTestDF(Seq("id" -> "item_id"))
+ val shuffles = collectShuffles(df.queryExecution.executedPlan)
+ assert(shuffles.isEmpty, "should not contain any shuffle")
+ checkAnswer(df, Seq(Row(1, "aa", 40.0, 42.0), Row(2, "bb", 10.0, 19.5)))
+ }
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/exchange/EnsureRequirementsSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/exchange/EnsureRequirementsSuite.scala
index 95085b71bc25..7a0664b46a38 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/exchange/EnsureRequirementsSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/exchange/EnsureRequirementsSuite.scala
@@ -35,7 +35,6 @@ import
org.apache.spark.sql.execution.joins.{ShuffledHashJoinExec, SortMergeJoin
import org.apache.spark.sql.execution.python.FlatMapCoGroupsInPandasExec
import org.apache.spark.sql.execution.window.WindowExec
import org.apache.spark.sql.internal.SQLConf
-import
org.apache.spark.sql.internal.SQLConf.REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.types.{IntegerType, StructField, StructType}
@@ -862,28 +861,9 @@ class EnsureRequirementsSuite extends SharedSparkSession {
assert(right.expressions === Seq(bucket(4, exprA), years(exprC)))
case other => fail(other.toString)
}
-
- // by default spark.sql.requireAllClusterKeysForCoPartition is true, so
when there isn't
- // exact match on all partition keys, Spark will fallback to shuffle.
- plan1 = new DummySparkPlanWithBatchScanChild(
- outputPartitioning = KeyedPartitioning(bucket(4, exprA) :: bucket(4,
exprB) :: Nil, Seq.empty)
- )
- plan2 = new DummySparkPlanWithBatchScanChild(
- outputPartitioning = KeyedPartitioning(bucket(4, exprA) :: bucket(4,
exprC) :: Nil, Seq.empty)
- )
- smjExec = SortMergeJoinExec(
- exprA :: exprB :: exprB :: Nil, exprA :: exprC :: exprC :: Nil, Inner,
None, plan1, plan2)
- EnsureRequirements.apply(smjExec) match {
- case SortMergeJoinExec(_, _, _, _,
- SortExec(_, _, ShuffleExchangeExec(left: HashPartitioning, _, _, _,
_), _),
- SortExec(_, _, ShuffleExchangeExec(right: HashPartitioning, _, _, _,
_), _), _) =>
- assert(left.expressions === Seq(exprA, exprB, exprB))
- assert(right.expressions === Seq(exprA, exprC, exprC))
- case other => fail(other.toString)
- }
}
- test(s"KeyedPartitioning with
${REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION.key} = false") {
+ test("KeyedPartitioning with subset of join keys") {
var plan1 = new DummySparkPlanWithBatchScanChild(
outputPartitioning = KeyedPartitioning(bucket(4, exprB) :: years(exprC)
:: Nil, Seq.empty)
)
@@ -891,9 +871,14 @@ class EnsureRequirementsSuite extends SharedSparkSession {
outputPartitioning = KeyedPartitioning(bucket(4, exprC) :: years(exprB)
:: Nil, Seq.empty)
)
- // simple case
+ // simple case: join key exprA is not covered by either side's partition
keys, so by default
+ // the coverage check of requireAllClusterKeysForCoPartition falls back to
shuffle to avoid
+ // joining on a partitioning coarser than the join keys
var smjExec = SortMergeJoinExec(
exprA :: exprB :: exprC :: Nil, exprA :: exprC :: exprB :: Nil, Inner,
None, plan1, plan2)
+ assert(EnsureRequirements.apply(smjExec)
+ .collect { case s: ShuffleExchangeLike => s }.length == 2)
+ // with requireAllClusterKeysForCoPartition=false, SPJ is allowed
applyEnsureRequirementsWithSubsetKeys(smjExec) match {
case SortMergeJoinExec(_, _, _, _,
SortExec(_, _, DummySparkPlan(_, _, left: KeyedPartitioning, _, _), _),
@@ -912,7 +897,7 @@ class EnsureRequirementsSuite extends SharedSparkSession {
)
smjExec = SortMergeJoinExec(
exprA :: exprB :: exprB :: Nil, exprA :: exprC :: exprC :: Nil, Inner,
None, plan1, plan2)
- applyEnsureRequirementsWithSubsetKeys(smjExec) match {
+ EnsureRequirements.apply(smjExec) match {
case SortMergeJoinExec(_, _, _, _,
SortExec(_, _, DummySparkPlan(_, _, left: KeyedPartitioning, _, _), _),
SortExec(_, _, DummySparkPlan(_, _, right: KeyedPartitioning, _, _), _),
_) =>
@@ -930,7 +915,20 @@ class EnsureRequirementsSuite extends SharedSparkSession {
KeyedPartitioning(years(exprA) :: bucket(4, exprC) :: days(exprA) ::
Nil, Seq.empty))
smjExec = SortMergeJoinExec(
exprA :: exprB :: exprB :: Nil, exprA :: exprC :: exprC :: Nil, Inner,
None, plan1, plan2)
- applyEnsureRequirementsWithSubsetKeys(smjExec) match {
+ EnsureRequirements.apply(smjExec) match {
+ case SortMergeJoinExec(_, _, _, _,
+ SortExec(_, _, DummySparkPlan(_, _, left: KeyedPartitioning, _, _), _),
+ SortExec(_, _, DummySparkPlan(_, _, right: KeyedPartitioning, _, _), _),
_) =>
+ assert(left.expressions === Seq(years(exprA), bucket(4, exprB),
days(exprA)))
+ assert(right.expressions === Seq(years(exprA), bucket(4, exprC),
days(exprA)))
+ case other => fail(other.toString)
+ }
+
+ // a column partitioned by more than one transform: partition expressions
outnumber the
+ // join keys, but every join key is covered, so SPJ is allowed with
default configs
+ smjExec = SortMergeJoinExec(
+ exprA :: exprB :: Nil, exprA :: exprC :: Nil, Inner, None, plan1, plan2)
+ EnsureRequirements.apply(smjExec) match {
case SortMergeJoinExec(_, _, _, _,
SortExec(_, _, DummySparkPlan(_, _, left: KeyedPartitioning, _, _), _),
SortExec(_, _, DummySparkPlan(_, _, right: KeyedPartitioning, _, _), _),
_) =>
@@ -967,7 +965,7 @@ class EnsureRequirementsSuite extends SharedSparkSession {
)
smjExec = SortMergeJoinExec(
exprA :: exprB :: exprB :: Nil, exprA :: exprC :: exprC :: Nil, Inner,
None, plan1, plan2)
- applyEnsureRequirementsWithSubsetKeys(smjExec) match {
+ EnsureRequirements.apply(smjExec) match {
case SortMergeJoinExec(_, _, _, _,
SortExec(_, _, ShuffleExchangeExec(left: HashPartitioning, _, _, _, _),
_),
SortExec(_, _, ShuffleExchangeExec(right: HashPartitioning, _, _, _, _),
_), _) =>
@@ -985,7 +983,7 @@ class EnsureRequirementsSuite extends SharedSparkSession {
)
smjExec = SortMergeJoinExec(
exprA :: exprB :: exprB :: Nil, exprA :: exprC :: exprC :: Nil, Inner,
None, plan1, plan2)
- applyEnsureRequirementsWithSubsetKeys(smjExec) match {
+ EnsureRequirements.apply(smjExec) match {
case SortMergeJoinExec(_, _, _, _,
SortExec(_, _, ShuffleExchangeExec(left: HashPartitioning, _, _, _, _),
_),
SortExec(_, _, ShuffleExchangeExec(right: HashPartitioning, _, _, _, _),
_), _) =>
@@ -1006,7 +1004,7 @@ class EnsureRequirementsSuite extends SharedSparkSession {
)
smjExec = SortMergeJoinExec(
exprA :: exprB :: exprB :: Nil, exprA :: exprC :: exprC :: Nil, Inner,
None, plan1, plan2)
- applyEnsureRequirementsWithSubsetKeys(smjExec) match {
+ EnsureRequirements.apply(smjExec) match {
case SortMergeJoinExec(_, _, _, _,
SortExec(_, _, ShuffleExchangeExec(left: HashPartitioning, _, _, _, _),
_),
SortExec(_, _, ShuffleExchangeExec(right: HashPartitioning, _, _, _, _),
_), _) =>
@@ -1016,6 +1014,34 @@ class EnsureRequirementsSuite extends SharedSparkSession
{
}
}
+ test("KeyedPartitioning: duplicated join keys in hand-built plans do not
block SPJ") {
+ // Queries produce this key list only in unusual configurations:
BooleanSimplification
+ // normally dedups the conjunction, but it is an excludable rule
+ // (spark.sql.optimizer.excludedRules), and EnsureRequirements must also
stay robust
+ // for hand-built or rewritten plans. The coverage check treats duplicated
cluster
+ // keys as covered, so SPJ is allowed with either config value.
+ val plan1 = new DummySparkPlanWithBatchScanChild(
+ outputPartitioning =
+ KeyedPartitioning(bucket(4, exprA) :: bucket(4, exprB) :: Nil,
Seq.empty))
+ val plan2 = new DummySparkPlanWithBatchScanChild(
+ outputPartitioning =
+ KeyedPartitioning(bucket(4, exprA) :: bucket(4, exprC) :: Nil,
Seq.empty))
+ val smjExec = SortMergeJoinExec(
+ exprA :: exprB :: exprB :: Nil, exprA :: exprC :: exprC :: Nil, Inner,
None, plan1, plan2)
+ Seq("true", "false").foreach { requireAllKeys =>
+ withSQLConf(SQLConf.REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION.key ->
requireAllKeys) {
+ EnsureRequirements.apply(smjExec) match {
+ case SortMergeJoinExec(_, _, _, _,
+ SortExec(_, _, DummySparkPlan(_, _, left: KeyedPartitioning, _,
_), _),
+ SortExec(_, _, DummySparkPlan(_, _, right: KeyedPartitioning, _,
_), _), _) =>
+ assert(left.expressions === Seq(bucket(4, exprA), bucket(4,
exprB)))
+ assert(right.expressions === Seq(bucket(4, exprA), bucket(4,
exprC)))
+ case other => fail(s"Expected no shuffle, but got: $other")
+ }
+ }
+ }
+ }
+
test("SPARK-41413: check compatibility when partition values mismatch") {
withSQLConf(SQLConf.V2_BUCKETING_PUSH_PART_VALUES_ENABLED.key -> "true") {
val leftPartValues = Seq(Array[Any](1, 1), Array[Any](2, 2)).map(new
GenericInternalRow(_))
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
