peter-toth commented on code in PR #59323:
URL: https://github.com/apache/spark/pull/59323#discussion_r4315825690
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -2470,6 +2489,11 @@ case class KeyedShuffleSpec(
// cannot rewrite it. This also keeps the unprojected spec returned by
`createShuffleSpec`
// for a marked narrowing projection from being chosen as the best spec.
keyPositions.forall(_.nonEmpty) &&
+ // A child lined up on these keys is co-partitioned only on the cluster
keys they cover. So
+ // the co-partition key requirement applies here too. It uses the
coverage rule of
+ // `referencesAllClusterKeys`, not the positional match of
`HashShuffleSpec`.
+ (!SQLConf.get.v2BucketingRequireAllClusterKeysForCoPartition ||
+ partitioning.referencesAllClusterKeys(distribution)) &&
Review Comment:
Done in
[7a3479f8fba](https://github.com/apache/spark/pull/59323/commits/7a3479f8fbae6fc3326a5a0cd6a72e21277777bd).
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -2101,10 +2118,12 @@ case class IdentityReducer(transform:
TransformExpression) extends Reducer[Any,
* all, and it reaches `None` two ways. An identity
projection over already
* grouped and sorted keys rebuilds the same
partitioning. A marked claim is
* refused outright, since neither narrowing nor
sorting its keys leaves the
- * routing it promises for its undeclared rows. The
first always leaves a
- * spec a consumer can be laid out on, and so does the
second where every
- * partition expression covers a clustering key. A
marked claim that narrows
- * does not: one `keyPositions` entry is empty there,
which
+ * routing it promises for its undeclared rows. The
first leaves a spec a
+ * consumer can be laid out on, and so does the second
where every
+ * partition expression covers a clustering key. With
the co-partition
+ * key requirement on (see
`KeyedPartitioning.referencesAllClusterKeys`),
+ * both also need the keys to cover every clustering
key. A marked claim
Review Comment:
Done in
[7a3479f8fba](https://github.com/apache/spark/pull/59323/commits/7a3479f8fbae6fc3326a5a0cd6a72e21277777bd).
The requirement sentence now comes after "... both turn away".
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -2596,6 +2600,18 @@ object SQLConf {
.booleanConf
.createWithDefault(false)
+ val V2_BUCKETING_REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION =
+
buildConf("spark.sql.sources.v2.bucketing.requireAllClusterKeysForCoPartition.enabled")
+ .doc("When true, a storage-partitioned join, or the one-side shuffle of
" +
+ s"${V2_BUCKETING_SHUFFLE_ENABLED.key}, runs on the partition keys only
if they cover " +
+ "every join or MERGE key. Set it to false to also allow partition keys
that cover only " +
+ "some of those keys, at the risk of data skew and reduced parallelism
from the coarser " +
+ "storage partitioning. Hash partitionings still follow " +
+ s"${REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION.key}. Defaults to that
config.")
+ .version("4.4.0")
+ .withBindingPolicy(ConfigBindingPolicy.SESSION)
Review Comment:
Done in
[7a3479f8fba](https://github.com/apache/spark/pull/59323/commits/7a3479f8fbae6fc3326a5a0cd6a72e21277777bd).
--
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]
