peter-toth commented on code in PR #59323:
URL: https://github.com/apache/spark/pull/59323#discussion_r4313987652
##########
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:
Done in
[d089dbeef1e](https://github.com/apache/spark/pull/59323/commits/d089dbeef1e2d7ad6fb34866b4928e5a214338f6).
The `requireAllClusterKeysForCoPartition` row now says that it also applies to
the one-side shuffle.
[b886e54cc18](https://github.com/apache/spark/pull/59323/commits/b886e54cc1881d67f988a75660e95167251d0f65)
adds a row for the new V2 config. The `shuffle.enabled` row and doc no longer
point at the requirement, since those two rows cover it.
##########
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:
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)
||
Review Comment:
Same reasoning as in the L2490 thread. With `y` unpartitioned, this PR
doesn't change the plan. `y` was shuffled onto `c` alone before too. For an
inner join, the rows of `y` that can match have `c = d`, and Spark filters on
that. So `c` alone is not coarser there. A right or full outer join also keeps
the other rows of `y`, and those are spread on `c` alone.
When `y` is partitioned on `c` with a different partition count, the plan
does change. `y` used to be the layout, and now it is shuffled onto `x`'s
layout on `c`. Either way the join runs on `c` for `y`. The migration note no
longer claims this case, in
[d089dbeef1e](https://github.com/apache/spark/pull/59323/commits/d089dbeef1e2d7ad6fb34866b4928e5a214338f6).
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/exchange/EnsureRequirementsSuite.scala:
##########
@@ -1467,9 +1467,11 @@ class EnsureRequirementsSuite extends SharedSparkSession
{
KeyedPartitioning(Seq(id, t1), keys))))
val unpartitioned = DummySparkPlan(outputPartitioning =
UnknownPartitioning(0))
+ // The coarse member covers only `id`, so the co-partition key requirement
is turned off.
Review Comment:
Done in
[d089dbeef1e](https://github.com/apache/spark/pull/59323/commits/d089dbeef1e2d7ad6fb34866b4928e5a214338f6).
Both tests now run with the conf on and off, and assert the same placement.
##########
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") {
Review Comment:
Done in
[d089dbeef1e](https://github.com/apache/spark/pull/59323/commits/d089dbeef1e2d7ad6fb34866b4928e5a214338f6).
The new test "SPARK-59971: a cogroup runs on a keyed side only if it covers
every cogroup key" in `EnsureRequirementsSuite` runs your example. It fails
without the fix.
--
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]
