Skip to content

[SPARK-58558][SQL] Make requireAllClusterKeysForCoPartition check key coverage instead of exact match for SPJ #57762

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
pan3793 wants to merge 3 commits into
base: master
Choose a base branch
from
Open

[SPARK-58558][SQL] Make requireAllClusterKeysForCoPartition check key coverage instead of exact match for SPJ #57762

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
266a19c
[SPARK-58558][SQL] Remove requireAllClusterKeysForCoPartition as a ga…
pan3793 Aug 4, 2026
bdafe4d
Check key coverage instead of removing the gate
pan3793 Aug 4, 2026
25c15cb
Restore incompatible-buckets test override, state V2-only coverage scope
pan3793 Aug 4, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions docs/sql-migration-guide.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 (V2 data sources), `spark.sql.requireAllClusterKeysForCoPartition` no longer requires the join keys to exactly match the partition keys: duplicated join keys no longer prevent shuffle elimination, and `spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled` no longer additionally requires `spark.sql.requireAllClusterKeysForCoPartition` to be `false`. As before, when the partition keys cover only part of the join keys, eliminating the shuffle still requires `spark.sql.requireAllClusterKeysForCoPartition` to be `false`.

## 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`.
Expand Down
9 changes: 4 additions & 5 deletions docs/sql-performance-tuning.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -535,9 +535,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 the partition keys, ignoring key order and duplicated keys, 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>

@pan3793 pan3793 Aug 4, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

a small doc fix, the config spark.sql.requireAllClusterKeysForCoPartition is introduced in 3.3.0

</tr>
<tr>
<td><code>spark.sql.sources.v2.bucketing.partiallyClusteredDistribution.enabled</code></td>
Expand All @@ -548,10 +548,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>
Expand Down Expand Up @@ -607,7 +607,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
Expand Down
5 changes: 4 additions & 1 deletion sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1105,7 +1105,10 @@ 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 V2 data source " +
"partitioning (storage-partitioned join), the check ignores key order and duplicated " +
"clustering keys: it requires every clustering key to be covered by the partition keys; " +
"hash partitioning deliberately keeps the exact match (same keys in the same order).")
.version("3.3.0")
.booleanConf
.createWithDefault(true)
Expand Down
23 changes: 11 additions & 12 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
12 changes: 0 additions & 12 deletions sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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 ->
Expand Down Expand Up @@ -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 ->
Expand Down Expand Up @@ -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",
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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 ->
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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",
Expand Down Expand Up @@ -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",
Expand Down Expand Up @@ -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"))
Expand Down Expand Up @@ -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(
Expand Down
65 changes: 38 additions & 27 deletions ...core/src/test/scala/org/apache/spark/sql/execution/exchange/EnsureRequirementsSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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}

Expand Down Expand Up @@ -862,38 +861,24 @@ 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)
)
var plan2 = new DummySparkPlanWithBatchScanChild(
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, _, _), _),
Expand All @@ -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, _, _), _), _) =>
Expand All @@ -930,7 +915,7 @@ 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, _, _), _), _) =>
Expand Down Expand Up @@ -967,7 +952,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, _, _, _), _), _) =>
Expand All @@ -985,7 +970,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, _, _, _), _), _) =>
Expand All @@ -1006,7 +991,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, _, _, _), _), _) =>
Expand All @@ -1016,6 +1001,32 @@ class EnsureRequirementsSuite extends SharedSparkSession {
}
}

test("KeyedPartitioning: duplicated join keys do not block SPJ") {
// The coverage check of requireAllClusterKeysForCoPartition ignores key order and
// duplicated cluster keys: join keys [a, b, b] are fully covered by partition keys
// on [a, b], 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)

@peter-toth peter-toth Aug 4, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Finding 8. This key list can't come from a query: BooleanSimplification dedups the conjunction before ExtractEquiJoinKeys builds the join keys, so the pair (t1.b, t2.c) cannot appear twice (details in finding 6 on SQLConf.scala:1110).

The KeyGroupedPartitioningSuite removals do cover the other win properly. Going through all 13: 8 of them (:1829, :2305, :2426, :2488, :2699, :3200, :3277, :3470) have more partition expressions than join keys — (identity(id), identity(data)) joined on data alone, and so on — so base failed attributes.length == clustering.length and they genuinely needed the =false override; 4 (:1199, :1970, :2129, :2369) already matched positionally on base, so those overrides were dead and dropping them changes nothing either way (finding 9).

But all 8 also need allowKeysSubsetOfPartitionKeys=true, so nothing covers the shape that newly works with no config at all: every partition attribute is a join key, yet the partition expressions outnumber the join keys because a column is partitioned twice. That case is one edit away in this file — the block at :909-925 already partitions on (years(a), bucket(4, b), days(a)):

    smjExec = SortMergeJoinExec(
      exprA :: exprB :: Nil, exprA :: exprC :: Nil, Inner, None, plan1, plan2)

attrs [a, b, a] against cluster keys [a, b]: base fails the length check, coverage passes, the existing left.expressions/right.expressions assertions still hold, and reorderJoinPredicates doesn't interfere (reorder bails on the 3-vs-2 size mismatch). An end-to-end case is worth having too, since that is where reachability actually gets exercised: both sides PARTITIONED BY (bucket(8, id), identity(id)) joined ON t1.id = t2.id gives cluster keys [id] against partition attrs [id, id] — shuffle on base, SPJ with this patch, and both transforms are supported by InMemoryBaseTable.

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(_))
Expand Down