peter-toth commented on code in PR #57762:
URL: https://github.com/apache/spark/pull/57762#discussion_r3713176239
##########
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, `spark.sql.requireAllClusterKeysForCoPartition` no longer
affects storage-partitioned joins (V2 data sources). A shuffle is now avoided
whenever all partition keys appear in the join keys, regardless of order;
joining on a subset of partition keys remains controlled by
`spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled`. Users
who previously set `spark.sql.requireAllClusterKeysForCoPartition` to `false`
solely to enable storage-partitioned joins no longer need to do so. The config
still applies to hash-partitioned children (e.g., V1 bucketing).
Review Comment:
**Finding 3.** This entry only states the upside. Two things a reader whose
join got slower after upgrading would need:
- the trade-off: when the partition keys cover only *part* of the join keys,
the join now runs with the storage layout's partition count instead of
`spark.sql.shuffle.partitions` - a table partitioned on `days(ts)` and joined
on `(ts, id)` loses parallelism and can get badly skewed (finding 1);
- how to get the old plan back.
`spark.sql.requireAllClusterKeysForCoPartition` no longer does it, and the only
remaining switches are `spark.sql.sources.v2.bucketing.enabled=false` (turns
SPJ off completely) or `spark.sql.requireAllClusterKeysForDistribution=true`
(also changes aggregate and window planning).
If finding 1 is addressed by keeping the gate order-insensitive instead,
this entry should say that key order and duplicated join keys no longer matter,
and that `allowKeysSubsetOfPartitionKeys` no longer needs
`requireAllClusterKeysForCoPartition=false` alongside it.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -782,20 +782,7 @@ 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)
- }
-
- if (satisfies) {
+ if (partitioning.satisfies(distribution)) {
Review Comment:
**Finding 1.** For a grouped `KeyedPartitioning` and a join's
`ClusteredDistribution` (whose `requireAllClusterKeys` comes from
`requireAllClusterKeysForDistribution`, default `false`),
`partitioning.satisfies(distribution)` reduces to
```scala
attributes.forall(x => requiredClustering.exists(_.semanticEquals(x)))
```
(`sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:595`)
- every *partition* key must be a join key, but not the other way round. So it
is also true when the partition keys are a strict subset of the join keys, i.e.
when the storage layout groups rows more coarsely than the join key does. That
is exactly what the removed gate blocked, and what the config was added for:
"This is to avoid data skews which can lead to significant performance
regression if shuffles are eliminated" (`SQLConf.scala:1105`) - the same
wording as the sibling check you keep for hash partitioning: "To avoid
potential data skew, we don't allow `HashShuffleSpec` to create partitioning if
the hash partition keys are not the full join keys" (`partitioning.scala:1138`).
`v2BucketingAllowKeysSubsetOfPartitionKeys` does not cover this. It guards
the opposite direction - join keys being a subset of the *partition* keys
(`partitioning.scala:583-587`).
Concretely, with two Iceberg tables `PARTITIONED BY (days(ts))`:
```sql
SELECT * FROM t JOIN s ON t.ts = s.ts AND t.id = s.id
```
on master both sides shuffle to `spark.sql.shuffle.partitions`; with this
patch SPJ fires and the join runs with one task per day, each holding a whole
day of rows. Nothing else puts a floor under it: `shouldConsiderMinParallelism`
/ `defaultNumShufflePartitions` only apply to the `bestSpecOpt` branch
(`EnsureRequirements.scala:192-200`), which is skipped entirely once
`areChildrenCompatible` is true (`:246`). And there is no targeted way back
afterwards - only turning SPJ off completely
(`spark.sql.sources.v2.bucketing.enabled=false`) or
`requireAllClusterKeysForDistribution=true`, which also changes
aggregate/window planning.
Your own test says as much: `EnsureRequirementsSuite.scala:866` is now named
"KeyedPartitioning with subset of join keys", and its first case joins on `[a,
b, c]` while neither side is partitioned on `a`.
If the goal is to stop the check caring about key *order*, the smaller fix
is to keep the gate but make it order-insensitive, which is what the config's
name says anyway:
```scala
def tryCreate(partitioning: KeyedPartitioning): Option[KeyedShuffleSpec]
= {
// 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 = {
// 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 (partitioning.satisfies(distribution) &&
(!SQLConf.get.getConf(SQLConf.REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION) ||
allClusterKeysCovered)) {
Some(partitioning.createShuffleSpec(distribution).asInstanceOf[KeyedShuffleSpec])
} else {
None
}
}
```
That keeps both wins from the description:
- the duplicated-join-key case works at defaults - a duplicated cluster key
is still covered by a partition key - so your new test at `:999` and the two
"duplicated keys" cases in `:866` still pass;
- `allowKeysSubsetOfPartitionKeys=true` no longer needs
`requireAllClusterKeysForCoPartition=false` alongside it, because that config
only ever adds extra *partition* keys, which the coverage check doesn't look at.
Only the first case of `:866` (join key `a` present on neither side's
partitioning) goes back to needing the config set to `false`, which is the case
I'd argue should stay opt-in.
If you do want coarser-than-join-key SPJ on by default, that's a bigger call
than "removing a redundant gate" - worth saying so plainly in the description
and the migration guide, and I'd expect a config to switch it off.
##########
docs/sql-performance-tuning.md:
##########
@@ -551,7 +543,7 @@ The following SQL properties enable Storage Partition Join
in different join que
<td><code>spark.sql.sources.v2.bucketing.allowJoinKeysSubsetOfPartitionKeys.enabled</code></td>
Review Comment:
**Finding 4.** While you're editing this row - this is the deprecated alias.
`SQLConf.scala:2500` defines the config as
`spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled` with
`.withAlternative("spark.sql.sources.v2.bucketing.allowJoinKeysSubsetOfPartitionKeys.enabled")`,
and the migration-guide entry you add uses the current name, so the two docs
disagree.
```suggestion
<td><code>spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled</code></td>
```
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
