peter-toth commented on code in PR #59323:
URL: https://github.com/apache/spark/pull/59323#discussion_r4313985984
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -2470,6 +2487,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
Review Comment:
This shape plans the same as before this PR. I checked it with the new
clause taken out of `canCreatePartitioning`. `x` covers its clustering `[a,
a]`, so it still serves as the layout, and `y` pairs with it as before.
For an inner join this is harmless. Every row of `y` that can match has `c =
d`, and Spark infers that filter. So `c` alone spreads those rows just as `[c,
d]` would. A right or full outer join also keeps the rows of `y` that can't
match, and no filter is inferred there. Those rows stay grouped on `c` alone.
That comes from `isCompatibleWith`, which this PR doesn't touch, so I'd leave
it out of this PR. The migration note claimed more than that, so I dropped that
sentence in
[d089dbeef1e](https://github.com/apache/spark/pull/59323/commits/d089dbeef1e2d7ad6fb34866b4928e5a214338f6).
##########
docs/sql-migration-guide.md:
##########
@@ -38,6 +38,7 @@ license: |
- Since Spark 4.4, a nondeterministic common expression that a conditional
branch reads more than once is evaluated once, and every read sees that one
value. `BETWEEN` and `NULLIF` read their input twice, so an expression such as
`CASE WHEN c THEN monotonically_increasing_id() BETWEEN 3 AND 5 END` previously
gave each read its own value and could return a different result. Queries whose
results depended on the earlier behaviour change, and an optimized or physical
plan may now show a `with` node for such a branch. A definition that is cheap
to evaluate and reports itself deterministic is still substituted at every
read, as before.
- Since Spark 4.4, `format_string` and `printf` pass a decimal argument to
`java.util.Formatter` as a `java.math.BigDecimal`, so the floating-point
conversions `%f`, `%e`, and `%g` format decimals instead of failing with
`IllegalFormatConversionException`. `%a` still rejects a decimal, because
`Formatter` supports it only for `float` and `double`. As a side effect, `%h`
on a decimal now hashes the `java.math.BigDecimal` (unscaled value and scale)
rather than Spark's internal `Decimal` (which hashes consistently with
`Double`), so it can print a different hash than earlier releases; the two
coincide for some values, such as zero at scale zero. `%s` output and `NULL`
handling are unchanged.
- Since Spark 4.4, for storage-partitioned joins,
`spark.sql.requireAllClusterKeysForCoPartition` requires every join key to be
covered by some partition key instead of matching the partition keys
positionally. As a result, a join-key column partitioned by more than one
transform no longer prevents shuffle elimination, and
`spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled` no
longer additionally requires `spark.sql.requireAllClusterKeysForCoPartition` to
be `false` when the join keys are a subset of the partition keys. As before,
when the partition keys cover only part of the join keys, eliminating the
shuffle still requires `spark.sql.requireAllClusterKeysForCoPartition` to be
`false`.
+- Since Spark 4.4, `spark.sql.requireAllClusterKeysForCoPartition` also
applies to the one-side shuffle that
`spark.sql.sources.v2.bucketing.shuffle.enabled` turns on. A join runs on one
side's partition keys only if they cover every join key. This holds whether the
other side keeps its partitioning or is shuffled onto those keys. Previously,
such a join could run on partition keys that covered only some of the join
keys. Now it runs on the layout of a side that covers every join key, if there
is one. Otherwise both sides are shuffled on all the join keys. A cogroup
changes the same way. To restore the previous plans, set
`spark.sql.requireAllClusterKeysForCoPartition` to `false`. That lifts the
requirement for every co-partitioned operator, not only for the one-side
shuffle.
Review Comment:
Done in
[d089dbeef1e](https://github.com/apache/spark/pull/59323/commits/d089dbeef1e2d7ad6fb34866b4928e5a214338f6).
The note now says that the number of partitions can change too.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/ShuffleSpecSuite.scala:
##########
@@ -705,6 +705,31 @@ class ShuffleSpecSuite extends SparkFunSuite with
SQLHelper {
}
}
+ test("SPARK-59971: canCreatePartitioning: KeyedShuffleSpec requires all the
cluster keys") {
+ val a = $"a".int
+ val b = $"b".int
+ val distribution = ClusteredDistribution(Seq(a, b))
+ def spec(expressions: Expression*): KeyedShuffleSpec = KeyedShuffleSpec(
+ KeyedPartitioning(expressions, Seq(InternalRow.fromSeq(expressions.map(_
=> 1)))),
+ distribution)
+ val bucketA = TransformExpression(TestBucketFunction, Seq(a), Some(4))
+
+ withSQLConf(
+ SQLConf.V2_BUCKETING_SHUFFLE_ENABLED.key -> "true",
+ SQLConf.REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION.key -> "true") {
+ // Key order does not matter, and a transform covers the column it is
over.
+ Seq(spec(a, b), spec(b, a), spec(bucketA, b)).foreach { s =>
+ assert(s.canCreatePartitioning, s"${s.partitioning.expressions} covers
[a, b]")
+ }
+ Seq(spec(a), spec(bucketA)).foreach { s =>
Review Comment:
Done in
[d089dbeef1e](https://github.com/apache/spark/pull/59323/commits/d089dbeef1e2d7ad6fb34866b4928e5a214338f6).
The conf-off block now checks both `spec(a)` and `spec(bucketA)`.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -5645,6 +5664,46 @@ 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")
+
+ // `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, "true", 2),
+ (purchases, "true", 2),
+ // With it off, the storage-partitioned join pairs the keyed sides as
they stand. The
+ // SPARK-54439 test above covers the unpartitioned side.
+ (keyedPurchases, "false", 0)).foreach { case (other, requireAll,
expectedShuffles) =>
+ withSQLConf(
+ SQLConf.V2_BUCKETING_SHUFFLE_ENABLED.key -> "true",
+ SQLConf.REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION.key -> requireAll)
{
+ val df = sql(
+ s"""${selectWithMergeJoinHint("i", "p")} id, name, i.price, p.price
+ |FROM testcat.ns.$items i
+ |JOIN testcat.ns.$other p ON i.id = p.item_id AND i.arrive_time =
p.time
+ |""".stripMargin)
+ assert(collectShuffles(df.queryExecution.executedPlan).size ==
expectedShuffles,
Review Comment:
Done in
[d089dbeef1e](https://github.com/apache/spark/pull/59323/commits/d089dbeef1e2d7ad6fb34866b4928e5a214338f6).
The test now checks that both shuffles are on all the join keys.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -5645,6 +5664,46 @@ 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")
+
+ // `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, "true", 2),
+ (purchases, "true", 2),
+ // With it off, the storage-partitioned join pairs the keyed sides as
they stand. The
+ // SPARK-54439 test above covers the unpartitioned side.
+ (keyedPurchases, "false", 0)).foreach { case (other, requireAll,
expectedShuffles) =>
Review Comment:
Done in
[d089dbeef1e](https://github.com/apache/spark/pull/59323/commits/d089dbeef1e2d7ad6fb34866b4928e5a214338f6).
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -2470,6 +2487,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.getConf(SQLConf.REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION)
||
+ partitioning.referencesAllClusterKeys(distribution)) &&
Review Comment:
I'd keep the shared helper, so this check and `createKeyedShuffleSpecs` stay
on one rule. The cost is a pass over a few keys at planning time.
--
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]
