pan3793 commented on code in PR #57762:
URL: https://github.com/apache/spark/pull/57762#discussion_r3713600503
##########
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:
Adopted the suggested coverage check in 084ca5fe191. The
partition-keys-cover-part-of-join-keys case stays gated by the config;
duplicated join keys and `allowKeysSubsetOfPartitionKeys=true` no longer need
`requireAllClusterKeysForCoPartition=false`.
##########
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:
Rewrote the entry in 084ca5fe191: it states the coverage semantics and that
the partition-keys-cover-part-of-join-keys case still requires setting the
config to `false`.
##########
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:
Fixed in 084ca5fe191, switched to the current config name.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -2246,7 +2242,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",
Review Comment:
Restored the override in ee18670641d, so the bucket-count comparison is
exercised again. Audited the remaining removed overrides for the same pattern:
the only other negative test ("Compatible buckets does not support SPJ with
push-down values or partially-clustered") has both join keys covered by its
partition keys and a passing SPJ branch, so no other test lost its purpose.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -1105,7 +1105,9 @@ 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.")
Review Comment:
Adopted the suggested wording in 66ae049fc14 (keeping a short note that hash
partitioning deliberately keeps the positional match), and rewrote the
migration entry, the tuning-guide row, and the PR description around the two
reachable shapes -- the multi-transform column and the subset direction. The
duplicated-join-key story is demoted to a planner-robustness note: the
hand-built test is retitled, and its comment points out that
`BooleanSimplification` dedups the conjunction but is excludable via
`spark.sql.optimizer.excludedRules`, the one configuration where the shape is
query-reachable.
--
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]
