dongjoon-hyun commented on code in PR #59323:
URL: https://github.com/apache/spark/pull/59323#discussion_r4235423830
##########
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:
The new check gates only the member that is picked as the layout
(`canCreatePartitioning`). The other child is still paired with it through
`isCompatibleWith`, which does not look at coverage, so that child can keep a
layout that covers only some of its own join keys. That contradicts the
migration guide ("This holds whether the other side keeps its partitioning or
is shuffled onto those keys"), and it is a pairing the storage-partitioned join
path has just refused for the same child.
It shows up with a duplicated join key. For `x.a = y.c AND x.a = y.d`, the
left clustering is `[a, a]` and the right one is `[c, d]`. `x` partitioned on
`a` covers `[a, a]` and wins. `y` partitioned on `c` does not cover `d`, so
`createKeyedShuffleSpecs` refuses it and SPJ declines. But `keyPositions` of
`c` (`{0}`) overlaps that of `a` (`{0, 1}`), so `y` is paired as it stands.
I reproduced it on 31740c4 with this test in `KeyGroupedPartitioningSuite`:
```scala
test("REPRO: duplicated join key lets the paired side skip the coverage
requirement") {
createTable("rx", Array(Column.create("a", IntegerType),
Column.create("b", IntegerType)),
Array(identity("a")))
sql("INSERT INTO testcat.ns.rx VALUES (1, 10), (2, 20), (3, 30)")
createTable("ry", Array(Column.create("c", IntegerType),
Column.create("d", IntegerType)),
Array(identity("c")))
sql("INSERT INTO testcat.ns.ry VALUES (1, 1), (2, 5), (3, 3)")
withSQLConf(
SQLConf.V2_BUCKETING_SHUFFLE_ENABLED.key -> "true",
SQLConf.REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION.key -> "true") {
val df = sql(
s"""${selectWithMergeJoinHint("x", "y")} x.a, x.b, y.c, y.d
|FROM testcat.ns.rx x JOIN testcat.ns.ry y ON x.a = y.c AND x.a =
y.d
|""".stripMargin)
checkAnswer(df, Seq(Row(1, 10, 1, 1), Row(3, 30, 3, 3)))
assert(collectShuffles(df.queryExecution.executedPlan).nonEmpty,
"y is partitioned on c alone, which does not cover the join key d")
}
}
```
`checkAnswer` passes and the last assertion fails (`List() was empty`). The
plan has no shuffle below the `SortMergeJoin` (runtime filters trimmed):
```
+- SortMergeJoin [a#12, a#12], [c#14, d#15], Inner
:- Sort [a#12 ASC NULLS FIRST, a#12 ASC NULLS FIRST], false, 0
: +- Project [a#12, b#13]
: +- Filter isnotnull(a#12)
: +- BatchScan testcat.ns.rx[a#12, b#13] ...
+- Sort [c#14 ASC NULLS FIRST, d#15 ASC NULLS FIRST], false, 0
+- Project [c#14, d#15]
+- Filter ((isnotnull(c#14) AND isnotnull(d#15)) AND (c#14 = d#15))
+- BatchScan testcat.ns.ry[c#14, d#15] ...
```
The same happens at unit level in `EnsureRequirementsSuite`:
`SortMergeJoinExec(Seq(xa, xa), Seq(ya, yb), Inner, None, left, right)` over
`KeyedPartitioning(Seq(xa), keys)` and `KeyedPartitioning(Seq(ya), keys)` (same
three keys) gets no shuffle on either side, and the right child keeps
`keyedpartitioning(ya, ...)`.
The answer is correct, since the matched rows have `c = d` and partitioning
on `c` is effectively partitioning on the whole key, so the impact is low. But
either the pairing in `pickCoPartitionTarget` should also require the paired
member to cover its own clustering when the conf is on, or the migration guide
should not claim the rule holds for the side that keeps its partitioning.
##########
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:
When the keyed side covers only part of the join keys, it now answers
`canCreatePartitioning = false`, so in `pickCoPartitionTarget`
`shouldConsiderMinParallelism` becomes true whenever the other child is a
`ShuffleExchangeLike`. That exchange then serves as the layout only if it has
at least `spark.sql.shuffle.partitions` partitions. Otherwise both sides go to
the default partition count.
For example, a keyed scan on `[id]` joined with an existing
`HashPartitioning([item_id, time], 10)` exchange under
`spark.sql.shuffle.partitions = 200`: previously the exchange was re-targeted
onto the keyed layout, and now both sides end up with 200 partitions.
This is what the hash path does too, so I think the behavior is fine. But
the note only says "both sides are shuffled on all the join keys". Would it be
worth mentioning that the partition count can change as well?
##########
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:
Only `spec(a)` is checked to pass with the conf off. Could we also assert
`spec(bucketA).canCreatePartitioning` in the conf-off block below?
Otherwise this negative case would still pass if a lone bucket key were
refused for an unrelated reason (for example a change to
`expressionsDescribeKeys` or to the transform check), so it does not pin that
the coverage clause is what rejects it.
##########
docs/sql-performance-tuning.md:
##########
@@ -684,7 +684,7 @@ The following SQL properties enable Storage Partition Join
in different join que
<td><code>spark.sql.sources.v2.bucketing.shuffle.enabled</code></td>
<td>false</td>
<td>
- When enabled, try to avoid shuffle on one side of the join, by
recognizing the partitioning reported by a V2 data source on the other side.
+ When enabled, try to avoid shuffle on one side of the join, by
recognizing the partitioning reported by a V2 data source on the other side.
With <code>spark.sql.requireAllClusterKeysForCoPartition</code> on, the
partition keys must cover every join key.
Review Comment:
`spark.sql.requireAllClusterKeysForCoPartition` is an `.internal()` conf, so
it is not in the generated SQL configuration reference. This page, and the
public `spark.sql.sources.v2.bucketing.shuffle.enabled` doc, now point users at
it, but there is no public place that explains what it is or that turning it
off affects every co-partitioned operator.
Should we describe it briefly here, or make it non-internal? The SPARK-58558
migration entry already mentions it, so the gap is pre-existing, but this PR
makes it user-visible for the one-side shuffle.
##########
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:
With the conf off, `keyedPurchases` is paired by `checkKeyGroupCompatible`,
which this PR does not touch, so this case does not exercise the changed
`canCreatePartitioning` path.
Could we add `(purchases, "false", 1)` here, so that the new test covers
both conf states of the one-side shuffle itself, instead of relying on the
SPARK-54439 test above?
##########
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:
This only counts shuffles. Could we also check that the two exchanges are
partitioned on both join keys, e.g. `HashPartitioning` over `[id, arrive_time]`
and `[item_id, time]`?
A regression that still shuffles both sides, but onto `[id]` alone, would
keep the count at 2 and pass, while that is exactly what the requirement is
meant to prevent.
##########
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:
Nit: `keyPositions.forall(_.nonEmpty)` above has already mapped each
expression's single reference onto the clustering positions (duplicated cluster
keys share one `BitSet`). So at this point coverage is just whether those
positions span every clustering index, e.g.
```scala
keyPositions.foldLeft(mutable.BitSet.empty)(_ | _).size ==
distribution.clustering.length
```
That would avoid deriving the same mapping a second time with
`semanticEquals` on every `canCreatePartitioning` call (`pickCoPartitionTarget`
makes at least two per member), and keep the two derivations from drifting
apart. Keeping `referencesAllClusterKeys` for `createKeyedShuffleSpecs` is
fine, since no `KeyedShuffleSpec` exists there yet.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -2591,7 +2591,9 @@ object SQLConf {
.doc("During a storage-partitioned join, whether to allow to shuffle
only one side. " +
"When only one side is KeyedPartitioning, if the conditions are met,
spark will " +
"only shuffle the other side. This optimization will reduce the amount
of data that " +
- s"needs to be shuffle. This config requires
${V2_BUCKETING_ENABLED.key} to be enabled")
+ s"needs to be shuffle. With
${REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION.key} on, the " +
Review Comment:
The doc of `REQUIRE_ALL_CLUSTER_KEYS_FOR_CO_PARTITION` itself (L1129) still
describes only the hash partition keys and storage-partitioned join.
Since it now also decides whether a `KeyedShuffleSpec` can be the layout
another child is shuffled onto, could we add a sentence there too, so that the
conf definition stays the authoritative description?
--
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]
