peter-toth commented on code in PR #58339:
URL: https://github.com/apache/spark/pull/58339#discussion_r3880214651
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -648,7 +660,8 @@ case class KeyedPartitioning(
// and the other side is re-shuffled using this spec, the two
`KeyedPartitioning`s carry the
// same keys in a different order and
`PartitioningCollection.fromPartitionings` rejects them.
val projectedPartitioning =
- new KeyedPartitioning(projectedExpressions, projectedKeys, isGrouped =
false).toGrouped
+ new KeyedPartitioning(projectedExpressions, projectedKeys, isGrouped =
false,
+ mayContainUnknownPartitionKeys =
mayContainUnknownPartitionKeys).toGrouped
Review Comment:
**Finding 10.** This projects the declared keys and carries the marker onto
the projected set — the same coarsening finding 1 fixed in
`projectKeyedPartitionings`. Two full keys collapse onto one projected key, so
a key that was *outside* the declared set can be *inside* the projected one,
and `areKeysCompatible`'s subset test reads as a guarantee that no longer
holds. I missed this site in round 1; my list of producers named the
projection, the union and the reducers, and not this one.
Measured on `948df55` with
`spark.sql.sources.v2.bucketing.shuffle.enabled=true` and
`spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled=true`,
AQE off as in the new tests:
```sql
-- a keyed (id, k) = {(1,x),(2,x),(3,x),(4,x)}; u keyed id = {1,2,3,4}
-- t v1 parquet = (1,z),(2,z),(3,z),(4,z) -- every (id,k) is out-of-set, no
id is
SELECT r.id, r.k, u.data
FROM (SELECT t.id AS id, t.k AS k FROM testcat.ns.a a RIGHT OUTER JOIN t
ON a.id = t.id AND a.k = t.k) r
JOIN testcat.ns.u u ON r.id = u.id
```
`r.k` stays in the outer select list, so the projection keeps both positions
and your new guard in `projectKeyedPartitionings` does not fire. Expected 4
rows, **0 returned** — the same as base, so this shape is not fixed. The second
join storage-partitions on the coarsened `{1,2,3,4}`: `GroupPartitions
JoinKeyPositions: [0] ExpectedPartitionKeys: 4` on both sides, and no exchange
between them.
`GroupPartitionsExec.outputPartitioning` coarsens the same way one hop later
(`sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala:88-90`):
with `joinKeyPositions` defined it declares the *projected* keys and keeps the
marker, and only the `reducers` case drops the claim.
On the fix: please do not route it through `isNarrowed`. The hazard is about
keys that are not in the declared list at all — an out-of-set full key
projecting *into* the projected declared set — while the flag only describes
how distinct the declared keys are among themselves. Today's `isNarrowed` would
happen to be true here because it records provenance, but #58351 redefines it
as actual key collapse and renames it `isCollapsed`, and in the repro above the
four declared keys stay four distinct projected keys, so it reads `false`. A
direct test is both clearer and stable across that change: refuse the marker
path whenever `joinKeyPositions.length < expressions.length`.
`EnsureRequirements.createKeyedShuffleSpec.tryCreate` is the natural gate — it
already returns an `Option` and still holds the unprojected arity, so returning
`None` sends the child down the ordinary shuffle path. Doing it here by
returning the unprojected `result` instead would be shorter, but it risks `crea
tePartitioning`'s `clustering(positionSet.head)` on a position that maps to no
clustering key, which is why the projection exists.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala:
##########
@@ -71,10 +71,23 @@ case class GroupPartitionsExec(
// can only differ in `expressions`; their `partitionKeys` reference
is shared (enforced by
// `PartitioningCollection`), so `groupedPartitions` is computed only
once.
val partitionKeys = groupedPartitions.map(_._1)
+ // When `reducers` is defined, `partitionKeys` above are the *reduced*
keys: an
+ // out-of-set key of a partitioning that may contain unknown partition
keys can reduce
+ // into the declared set (e.g. identity keys {0, 1, 2, 3} holding an
out-of-set id=4,
+ // reduced by `bucket(4, id)`, declares {0, 1, 2, 3} again), so the
reduced keyed claim
+ // cannot be trusted and must be dropped entirely -- not just the
marker, because the
+ // regrouped data layout no longer matches the child's declared keys
either.
+ if (reducers.isDefined && p.exists {
+ case k: KeyedPartitioning => k.mayContainUnknownPartitionKeys
+ case _ => false
+ }) {
+ return UnknownPartitioning(0)
Review Comment:
**Finding 11.** Two problems on this line.
`UnknownPartitioning(0)` claims zero partitions while this node produces
`groupedPartitions.size` of them. Every other operator that gives up on
describing its partitioning reports the real count —
`PartitioningPreservingUnaryExecNode` uses
`UnknownPartitioning(child.outputPartitioning.numPartitions)`, `ShuffledJoin`'s
`FullOuter` arm the same. The `0` breaks `PartitioningCollection`'s
uniform-`numPartitions` requirement as soon as an inner join above this node
builds one from both sides.
Measured on `948df55` with
`spark.sql.sources.v2.bucketing.shuffle.enabled=true` and
`spark.sql.sources.v2.bucketing.allowCompatibleTransforms.enabled=true`:
```sql
-- a keyed bucket(4, id), ids 0..3; t v1 parquet, ids 0..3; u keyed
bucket(2, id), ids 0..3
SELECT * FROM testcat.ns.a a JOIN t ON a.id = t.id JOIN testcat.ns.u u ON
a.id = u.id
ORDER BY a.id
```
master returns the four rows; this commit throws:
```
java.lang.IllegalArgumentException: requirement failed:
PartitioningCollection requires all of its partitionings have the same
numPartitions.
at
org.apache.spark.sql.catalyst.plans.physical.PartitioningCollection.<init>(partitioning.scala:859)
at
org.apache.spark.sql.catalyst.plans.physical.PartitioningCollection$.fromPartitionings(partitioning.scala:978)
at
org.apache.spark.sql.execution.joins.ShuffledJoin.outputPartitioning(ShuffledJoin.scala:73)
at
...EnsureRequirements.ensureDistributionAndOrdering(EnsureRequirements.scala:69)
```
The left side is `GroupPartitions ... Reducers: [BucketReducer(2)]` over the
first join, reporting 0; the right side is a `GroupPartitionsExec` over `u`
reporting 2. The `ORDER BY` is only there to make something above the top join
ask for its `outputPartitioning`; a third join or an aggregate does it too.
```suggestion
return UnknownPartitioning(groupedPartitions.size)
```
The second problem is why the guard fires at all. It needs `reducers`
defined *and* a flagged KP in the child's partitioning, and finding 5's fix
makes those nearly exclusive: when the spec is flagged, `areKeysCompatible` now
requires `isSameFunction` per position, and a same-function pair has no reducer
(`BucketFunction.reducer` returns `null` when `gcd == thisNumBuckets`,
`DaysFunction.reducer(DaysFunction)` returns `null`). The only path I found is
the one above — the first join is `InnerLike`, so its `outputPartitioning` is a
`PartitioningCollection` holding `a`'s unflagged KP *and* the re-shuffled
side's flagged one, `createKeyedShuffleSpec`'s `collectFirst` picks the
unflagged one so `areKeysCompatible` never sees a marker and the reducer is
computed, while `p.exists` here still finds the flagged sibling. In that shape
the marker is spurious: the inner join has already dropped every row whose key
is outside the declared set, so the reduced claim is sound. Worth scoping the
predicate to the KP the reduction is actually about, or dropping the branch
and recording in the scaladoc why a flagged spec cannot carry reducers (a
third-party `ReducibleFunction` returning a self-reducer is the remaining hole).
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -538,12 +538,23 @@ case class CoalescedNullAwareHashPartitioning(
* partitioning can only satisfy `ClusteredDistribution` by
being grouped, and
* `groupedSatisfies` refuses that unless the config is
enabled, regardless of
* `requireAllClusterKeysForDistribution`.
+ * @param mayContainUnknownPartitionKeys Whether the data may contain rows
whose partition key is
+ * not among the declared `partitionKeys`.
This happens when a side
+ * is re-shuffled onto this partitioning (see
+ * `KeyedShuffleSpec.createPartitioning`):
`KeyGroupedPartitioner`
+ * silently routes keys outside the declared
set to arbitrary
+ * partitions, so only the declared keys are
guaranteed to be
+ * co-located. Such a partitioning is unsound
to storage-partition
+ * join against a side whose partition keys
are not a subset of the
+ * declared keys -- see
`KeyedShuffleSpec.areKeysCompatible`.
Review Comment:
**Finding 13.** This `@param` qualifies an invariant that the `== Partition
Keys ==` section states flatly a hundred lines above
(`sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:463`):
"partition `i` holds key `partitionKeys(i)`. A consumer must therefore treat
the given order as authoritative rather than re-derive one." That paragraph is
where a consumer looks for the layout contract, so one sentence there — a
partition may also hold rows whose key is not declared at all when
`mayContainUnknownPartitionKeys` is set — keeps the two in sync. #58351
restructures this class doc around key collapse, so land the sentence wherever
the layout paragraph ends up after the rebase.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/AliasAwareOutputExpression.scala:
##########
@@ -131,6 +131,16 @@ trait PartitioningPreservingUnaryExecNode extends
UnaryExecNode
if (projectablePositions.isEmpty) return LazyList.empty
+ // If any input KP may contain unknown partition keys, its keyed claim
only holds while the
+ // declared key set survives verbatim: the claim asserts that rows outside
the declared keys
+ // cannot exist, and dropping a key position coarsens the declared set so
that an out-of-set
+ // key can land inside it (see
`KeyedPartitioning.mayContainUnknownPartitionKeys`). Drop the
+ // keyed claim entirely in that case.
+ if (projectablePositions.length < numPositions &&
+ kps.exists(_.mayContainUnknownPartitionKeys)) {
Review Comment:
**Finding 12.** `kps` can hold a flagged and an unflagged
`KeyedPartitioning` at the same time: `PartitioningCollection`'s invariant only
requires a shared `partitionKeys` reference and equal arity, and
`ShuffledJoin.outputPartitioning` builds exactly that for `InnerLike` — the
keyed side's own KP plus the re-shuffled side's flagged one. In that collection
the marker is spurious: the inner join has already dropped every row whose key
is outside the declared set, so the surviving rows do obey the declared layout.
`exists` drops a keyed claim that is sound.
Measured on `948df55` with only
`spark.sql.sources.v2.bucketing.shuffle.enabled=true`:
```sql
-- a keyed (id, k) = {(1,x)..(4,x)}; t v1 parquet = (1,x)..(4,x); u keyed
id = {1..4}
SELECT r.id, u.data
FROM (SELECT t.id AS id FROM testcat.ns.a a JOIN t ON a.id = t.id AND a.k =
t.k) r
JOIN testcat.ns.u u ON r.id = u.id
```
Four correct rows on master and here, but master plans one shuffle (the
first join's one-side shuffle) and this commit plans two — the second join
loses its storage-partitioned join.
Filtering `kps` to the unflagged ones before this guard, and keeping the
drop when none remain, would preserve it. The all-flagged case genuinely has to
drop: with both join sides re-shuffled onto equal declared keys, an out-of-set
key hashes to the same index on both sides, so the join can emit matched rows
whose key is not declared. The `p.exists` in `GroupPartitionsExec` has the same
scoping question (finding 11).
One thing to watch when this meets #58351: that PR normalizes `isCollapsed`
across a collection by OR and adds a `require` that the members agree. Please
do not extend either to this marker. `isCollapsed` describes the shared
physical layout, so every member naming that layout is equally coarse; this
marker describes which *rows* the side that produced it may hold, and an inner
join filters those rows away. Requiring agreement would make the mixed
collection above impossible to represent and would lock in the lost SPJ.
--
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]
