dongjoon-hyun commented on code in PR #59323:
URL: https://github.com/apache/spark/pull/59323#discussion_r4283222345
##########
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:
The migration guide says "A cogroup changes the same way", but this test and
the changed ones only run joins. The existing cogroup tests in
`EnsureRequirementsSuite`, e.g. SPARK-59080 "pushed-down positions index into
the child's own partition expressions", use keyed sides that cover the cogroup
key.
Could we add a cogroup case where the keyed side covers only some of the
cogroup keys? For example, a `FlatMapCoGroupsInPandasExec` over a keyed child
and an unpartitioned child, checking that both sides are shuffled with the conf
on and only the unpartitioned one with it off.
##########
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:
This checks coverage against this side's own clustering only. When that
clustering repeats a key, the child shuffled onto this layout can still end up
on only some of its own join keys.
For `x.a = y.c AND x.a = y.d`, with `x` partitioned on `a` and `y`
unpartitioned, the left clustering is `[a, a]`. `[a]` covers it, so `x` is
picked as the layout. `y` is then shuffled with `createPartitioning([c, d])`.
`keyPositions` of `a` is `{0, 1}`, and `clustering(positionSet.head)` at L2516
takes `c`, so `y` is partitioned on `c` alone and `d` is not covered.
This is a different path from my earlier comment on L2490, which is about
the side paired through `isCompatibleWith`. I found it by reading the code and
have not run it. The answer is still correct, since the matched rows have `c =
d`. But the migration guide says the rule holds when the other side "is
shuffled onto those keys". Either the guide should leave out a repeated join
key, or `pickCoPartitionTarget` should also check the coverage of the
partitioning it builds for the re-shuffled child.
##########
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:
With the conf now off here and in the test at L1615, no test covers the
default case for a collection: the coarse member `[id, t2]` cannot serve, and
the re-shuffled side has to land on `[id, t1]`.
Could we also run this test with the conf on, and assert the same placement?
Otherwise, under the default conf, a regression in `pickCoPartitionTarget` that
picks the coarse member, or no member at all, would not fail any test.
##########
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:
A correction to this comment: this page does document
`spark.sql.requireAllClusterKeysForCoPartition`, in the row at L652-656 of the
same table, so the pointer here is fine.
That row still describes the conf only for storage-partitioned join, though.
Could we update it as well, to say that the conf now also decides whether a
one-side shuffle can run on partition keys that cover only some of the join
keys?
--
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]
