peter-toth commented on code in PR #59323:
URL: https://github.com/apache/spark/pull/59323#discussion_r4315824019
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -5645,6 +5664,63 @@ class KeyGroupedPartitioningSuite
}
}
+ test("SPARK-59971: a side is laid out on another only if its keys cover
every join key") {
+ createTable(items, itemsColumns, Array(identity("id")))
+ sql(s"INSERT INTO testcat.ns.$items VALUES " +
+ "(1, 'aa', 40.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 'bb', 10.0, cast('2020-01-01' as timestamp)), " +
+ "(4, 'cc', 15.5, cast('2020-02-01' as timestamp))")
+ // The same ids as `items`, so `keyed_purchases` holds the same partition
keys and can pair
+ // with `items` as it stands.
+ val purchaseRows = "(1, 42.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 19.5, cast('2020-02-01' as timestamp)), " +
+ "(4, 26.0, cast('2020-03-01' as timestamp))"
+ createTable(purchases, purchasesColumns, Array.empty)
+ sql(s"INSERT INTO testcat.ns.$purchases VALUES $purchaseRows")
+ val keyedPurchases = "keyed_purchases"
+ createTable(keyedPurchases, purchasesColumns, Array(identity("item_id")))
+ sql(s"INSERT INTO testcat.ns.$keyedPurchases VALUES $purchaseRows")
+
+ val shuffle = SQLConf.V2_BUCKETING_SHUFFLE_ENABLED.key
+ val requireAll = SQLConf.REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION.key
+ val v2RequireAll =
SQLConf.V2_BUCKETING_REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION.key
+ // `items` is partitioned by `id` alone, and the join is on `id` and the
time.
+ Seq(
+ // With the requirement on, the join must not run on `id` alone, so both
sides are shuffled.
+ (keyedPurchases, Seq(shuffle -> "true", requireAll -> "true"), 2),
+ (purchases, Seq(shuffle -> "true", requireAll -> "true"), 2),
+ // With it off, the storage-partitioned join pairs the keyed sides as
they stand, and the
+ // unpartitioned side is shuffled onto `items`.
+ (keyedPurchases, Seq(shuffle -> "true", requireAll -> "false"), 0),
+ (purchases, Seq(shuffle -> "true", requireAll -> "false"), 1),
+ // The V2 config lifts it for storage-partitioned layouts alone, both in
the one-side
+ // shuffle and in the storage-partitioned join. Without the one-side
shuffle, only the
+ // storage-partitioned join can keep the keyed sides as they stand.
+ (purchases, Seq(shuffle -> "true", requireAll -> "true", v2RequireAll ->
"false"), 1),
+ (keyedPurchases, Seq(shuffle -> "false", requireAll -> "true",
v2RequireAll -> "false"), 0)
Review Comment:
Done in
[7a3479f8fba](https://github.com/apache/spark/pull/59323/commits/7a3479f8fbae6fc3326a5a0cd6a72e21277777bd).
The new row takes two shuffles on all the join keys, as you expected. A gate
that lifts the requirement when either conf is `false` fails it.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -1614,8 +1614,11 @@ class KeyGroupedPartitioningSuite
// layout next to `bucket(4, id)`. The second join must not shuffle
`xy` onto the first
// one. That builds `bucket(4, y)` for `xy`, not the bucket of the
same function of `y`. A
// row with `b = y` then lands in a different bucket on each side. The
join order decides
- // which of the two layouts comes first, so both orders run.
- withSQLConf(SQLConf.V2_BUCKETING_SHUFFLE_ENABLED.key -> "true") {
+ // which of the two layouts comes first, so both orders run. Each
layout covers one of the
+ // two join keys, so the co-partition key requirement is turned off.
+ withSQLConf(
+ SQLConf.V2_BUCKETING_SHUFFLE_ENABLED.key -> "true",
+ SQLConf.REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION.key -> "false") {
Review Comment:
Done in
[7a3479f8fba](https://github.com/apache/spark/pull/59323/commits/7a3479f8fbae6fc3326a5a0cd6a72e21277777bd).
The test now also runs with the conf on. Both sides of the second join are
then shuffled on both of its keys, three shuffles in all, as you read it. That
case fails without the coverage clause in `canCreatePartitioning`.
##########
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 " +
Review Comment:
Done in
[7a3479f8fba](https://github.com/apache/spark/pull/59323/commits/7a3479f8fbae6fc3326a5a0cd6a72e21277777bd),
in the conf doc and in the new row.
##########
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 ||
Review Comment:
Done in
[7a3479f8fba](https://github.com/apache/spark/pull/59323/commits/7a3479f8fbae6fc3326a5a0cd6a72e21277777bd).
`referencesAllClusterKeys` became `meetsCoPartitionKeyRequirement`, which also
reads the conf. Both callers ask it.
##########
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
Review Comment:
Done in
[7a3479f8fba](https://github.com/apache/spark/pull/59323/commits/7a3479f8fbae6fc3326a5a0cd6a72e21277777bd),
here and in the two places you listed.
--
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]
