dongjoon-hyun commented on PR #58339:
URL: https://github.com/apache/spark/pull/58339#issuecomment-5496114072
I reviewed the compatibility rules, the marker propagation paths, and the
interaction with partition filtering / merge semantics. The bug analysis and
the overall approach look correct to me. A few findings:
### 1. The join-to-join direct path misses the spurious-marker scoping,
losing a sound SPJ (plan regression)
The spurious-marker rule ("a marked member next to an unmarked sibling in a
mixed `PartitioningCollection` comes from an `InnerLike` join, so the marker is
spurious") is applied in `PartitioningPreservingUnaryExecNode`,
`GroupPartitionsExec` and `UnionExec`, but not when the next join consumes the
collection directly: `EnsureRequirements.createKeyedShuffleSpec` picks a member
via `collectFirst` and trusts its marker as-is.
Repro (no `Project` between the two joins; AQE off for shuffle counting):
```sql
-- a: keyed on id, keys {1,2} / t: v1 parquet, keys {1,2,3} / u: keyed on
id, keys {1,2,3}
SET spark.sql.sources.v2.bucketing.shuffle.enabled=true;
SELECT * FROM t JOIN testcat.ns.a a ON t.id = a.id JOIN testcat.ns.u u ON
t.id = u.id;
```
The first join's collection is `[tKP(marked, {1,2}), aKP(unmarked, {1,2})]`,
and the second join's clustering key `t.id` matches only the marked member
(attribute origin), so a sibling-order preference cannot rescue it. u's keys
`{1,2,3}` are not a subset of `{1,2}`, so `areKeysCompatible` refuses and a
shuffle is inserted. Verified empirically:
- master: 1 shuffle, the second join storage-partitions with
`GroupPartitions` on both sides
- this PR: 2 shuffles, no `GroupPartitions`
The inner join already filtered every out-of-set row, so the lost SPJ was
sound -- this is a plan-quality regression only (results stay correct). The
"inner join clears the spurious marker on both member orders" test does not
catch it because a and u there share the same key set, so the subset test
passes anyway.
Suggestion: the only producer of a mixed collection is `ShuffledJoin`'s
`InnerLike` arm (I checked every `PartitioningCollection` construction site:
`ShuffledJoin`, `BroadcastHashJoinExec`, `UnionExec` case B,
`PartitioningPreservingUnaryExecNode`, `EnsureRequirements.splitPartitioning`
-- only the first can mix a marked KP with an unmarked one). Normalizing the
marker once at construction -- clearing it on the members of a mixed collection
in that arm (or in `fromPartitionings`) -- would fix this path automatically,
and would also let you delete the `forall`-based scoping in
`PartitioningPreservingUnaryExecNode` and `GroupPartitionsExec` entirely.
### 2. Minor: the `createShuffleSpec` refusal comment is imprecise
> it disagrees on arity with every other spec, so co-location is refused
Two unknown-keyed specs that both narrow have equal arity. What actually
refuses the pair is the `keyPositions` intersection check in
`areKeysCompatible`: the extra partition expressions map to no clustering key,
so their position sets are empty and the intersection test fails. Worth
rewording -- the new `keyPositions.forall(_.nonEmpty)` clause in
`canCreatePartitioning` shows the mechanism you already rely on.
### 3. Nits
- `ShuffleSpecSuite`'s "incompatibility without unknown partition keys" test
covers pre-existing generic behavior (arity mismatch, non-overlapping key
positions) unrelated to this change.
- `areKeysCompatible` builds `partitionKeys.toSet` on every call on the
marked path; harmless at current call rates, but easy to hoist if you touch
this again.
--
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]
