peter-toth commented on code in PR #57762:
URL: https://github.com/apache/spark/pull/57762#discussion_r3713962570
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -1105,7 +1105,9 @@ object SQLConf {
.doc("When true, the planner requires all the clustering keys as the
hash partition keys " +
"of the children, to eliminate the shuffles for the operator that
needs its children to " +
"be co-partitioned, such as JOIN node. This is to avoid data skews
which can lead to " +
- "significant performance regression if shuffles are eliminated.")
+ "significant performance regression if shuffles are eliminated. For V2
data source " +
+ "partitioning (storage-partitioned join), the check ignores key order
and duplicated " +
+ "clustering keys: it requires every clustering key to be covered by
the partition keys.")
Review Comment:
**Finding 6.** (Replaces my earlier finding-7 comment on this line — that
one is withdrawn, see the review body.)
"the check ignores key order and duplicated clustering keys" names two
relaxations that no query reaches:
- **Key order** was already handled before this gate ran:
`reorderJoinPredicates` runs first in the same `transformUp` case, and
`reorderJoinKeysRecursively` has explicit `KeyedPartitioning` cases that
permute the join keys into partition-key order
(`EnsureRequirements.scala:412-426`). In the one situation it can't fix — the
two sides' partition orders disagreeing relative to the join pairing —
`areKeysCompatible` rejects the pair anyway, gate or no gate.
- **Duplicated clustering keys** don't survive the optimizer:
`BooleanSimplification` splits the whole conjunction and dedups it through an
`ExpressionSet` (`expressions.scala:458-493`, its own comment being `(a && b)
&& a && (a && c) => a && b && c`), so `ON t1.a = t2.a AND t1.b = t2.c AND t1.b
= t2.c` collapses to two key pairs before `ExtractEquiJoinKeys` sees it. An
*asymmetric* duplicate does survive the optimizer but dies in the planner:
coverage is applied per side, so `ON t1.a = t2.a AND t1.b = t2.b AND t1.b =
t2.c` needs `c` covered on t2, and `areKeysCompatible` then rejects 2 vs 3
partition expressions (`partitioning.scala:1304`).
What the check does relax is the *partition* side: a partitioning may now
have more expressions than the join has key positions. Both sides `PARTITIONED
BY (bucket(8, id), truncate(4, id))` joined `ON t1.id = t2.id` gives cluster
keys `[id]` against partition attrs `[id, id]` — the old `attributes.length ==
clustering.length` failed, coverage passes, and it is safe because the
partition tuple is still a function of the join key. That, plus the
`allowKeysSubsetOfPartitionKeys` direction, is the whole win.
One framing note while you're rewriting it. This is a pure relaxation of the
config-`true` path, so nothing it enables was previously impossible — all of it
was reachable by setting `requireAllClusterKeysForCoPartition=false`. What it
buys is that you no longer have to switch that guard off globally, which would
also admit the genuinely risky partition-keys-cover-part-of-the-join-keys case,
to get two shapes that carry no skew risk: join keys a subset of the partition
columns, and a join-key column partitioned by more than one transform. That
reads as a stronger motivation than the duplicated-join-key story, and it is
what the tests actually demonstrate.
The current wording also appears in `docs/sql-migration-guide.md:27`
("duplicated join keys no longer prevent shuffle elimination"),
`docs/sql-performance-tuning.md:538` ("ignoring key order and duplicated keys")
and case 1 of the PR description, so all four want the same correction. For
this one (as a plain block, since the anchor can't reach line 1109):
```scala
.doc("When true, the planner requires all the clustering keys as the
hash partition keys " +
"of the children, to eliminate the shuffles for the operator that
needs its children to " +
"be co-partitioned, such as JOIN node. This is to avoid data skews
which can lead to " +
"significant performance regression if shuffles are eliminated. For
V2 data source " +
"partitioning (storage-partitioned join), every clustering key must
be covered by some " +
"partition key, rather than matching the partition keys
positionally, so a column " +
"partitioned by more than one transform does not prevent shuffle
elimination.")
```
--
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]
