peter-toth commented on code in PR #57762:
URL: https://github.com/apache/spark/pull/57762#discussion_r3714437363
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/exchange/EnsureRequirementsSuite.scala:
##########
@@ -1016,6 +1001,32 @@ class EnsureRequirementsSuite extends SharedSparkSession
{
}
}
+ test("KeyedPartitioning: duplicated join keys do not block SPJ") {
+ // The coverage check of requireAllClusterKeysForCoPartition ignores key
order and
+ // duplicated cluster keys: join keys [a, b, b] are fully covered by
partition keys
+ // on [a, b], so SPJ is allowed with either config value.
+ val plan1 = new DummySparkPlanWithBatchScanChild(
+ outputPartitioning =
+ KeyedPartitioning(bucket(4, exprA) :: bucket(4, exprB) :: Nil,
Seq.empty))
+ val plan2 = new DummySparkPlanWithBatchScanChild(
+ outputPartitioning =
+ KeyedPartitioning(bucket(4, exprA) :: bucket(4, exprC) :: Nil,
Seq.empty))
+ val smjExec = SortMergeJoinExec(
+ exprA :: exprB :: exprB :: Nil, exprA :: exprC :: exprC :: Nil, Inner,
None, plan1, plan2)
Review Comment:
**Finding 8.** This key list can't come from a query:
`BooleanSimplification` dedups the conjunction before `ExtractEquiJoinKeys`
builds the join keys, so the pair `(t1.b, t2.c)` cannot appear twice (details
in finding 6 on `SQLConf.scala:1110`).
The `KeyGroupedPartitioningSuite` removals do cover the other win properly.
Going through all 13: 8 of them (`:1829`, `:2305`, `:2426`, `:2488`, `:2699`,
`:3200`, `:3277`, `:3470`) have more partition expressions than join keys —
`(identity(id), identity(data))` joined on `data` alone, and so on — so base
failed `attributes.length == clustering.length` and they genuinely needed the
`=false` override; 4 (`:1199`, `:1970`, `:2129`, `:2369`) already matched
positionally on base, so those overrides were dead and dropping them changes
nothing either way (finding 9).
But all 8 also need `allowKeysSubsetOfPartitionKeys=true`, so nothing covers
the shape that newly works with no config at all: every partition attribute is
a join key, yet the partition expressions outnumber the join keys because a
column is partitioned twice. That case is one edit away in this file — the
block at `:909-925` already partitions on `(years(a), bucket(4, b), days(a))`:
```scala
smjExec = SortMergeJoinExec(
exprA :: exprB :: Nil, exprA :: exprC :: Nil, Inner, None, plan1,
plan2)
```
attrs `[a, b, a]` against cluster keys `[a, b]`: base fails the length
check, coverage passes, the existing `left.expressions`/`right.expressions`
assertions still hold, and `reorderJoinPredicates` doesn't interfere (`reorder`
bails on the 3-vs-2 size mismatch). An end-to-end case is worth having too,
since that is where reachability actually gets exercised: both sides
`PARTITIONED BY (bucket(8, id), identity(id))` joined `ON t1.id = t2.id` gives
cluster keys `[id]` against partition attrs `[id, id]` — shuffle on base, SPJ
with this patch, and both transforms are supported by `InMemoryBaseTable`.
--
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]
