peter-toth commented on code in PR #58339:
URL: https://github.com/apache/spark/pull/58339#discussion_r3931955382
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala:
##########
@@ -197,7 +214,9 @@ case class GroupPartitionsExec(
// Both cheap terms come first, so the scan below runs only where a merge
is possible. A
// grouping that left the keys as they are groups the child's own key
values, and one of those
- // groups can only ever cover the one key it was built from.
+ // groups can only ever cover the one key it was built from. `keysChanged`
also feeds
+ // `identityGrouping`: a projection or reduction re-labels the groups into
a different key
+ // space, which no index alignment can undo.
val keysChanged =
joinKeyPositions.exists(_.length < childKp.expressions.length) ||
reducers.isDefined
Review Comment:
**Finding 27.** `keysChanged` only catches a projection that *drops* a
position. One that keeps every position but *reorders* them reads as
key-unchanged, and `projectKeys` still rewrites every key row's columns. When
the reordered keys happen to sort into the child's own order, each group keeps
its index and the count is unchanged, so `identityGrouping` is `true` and the
marked claim rides onto a key space the child's undeclared rows were never
hashed in.
Measured, marked child keys `[(1, 8), (2, 9)]` over `[a, b]` with
`joinKeyPositions = Some(Seq(1, 0))`:
```
groupedPartitions = [((8,1), [0]), ((9,2), [1])]
outputPartitioning = KeyedPartitioning(exprs=[b, a], keys=[(8,1), (9,2)],
mayContainUnknownPartitionKeys=true)
```
The child's out-of-set rows sit at `nonNegativeMod(hash((k_a, k_b)), 2)`,
while the reported claim pins them to `nonNegativeMod(hash((k_b, k_a)), 2)`.
That is the hole [finding
26](https://github.com/apache/spark/pull/58339#discussion_r3924348492) closed
for a narrowing projection and a reduction, on the other axis.
It is unreachable today. All three producers of `joinKeyPositions` emit
ascending positions: `createShuffleSpec`'s `zipWithIndex.filter.map`,
`clusterKeyPositions`' `BitSet.toSeq`, and `applyGroupPartitions` /
`withJoinKeyPositions`, which pass a spec's own. My finding 26 is also what put
the predicate on `length`, so the gap is as much mine as the code's. What makes
it worth closing is the doc right above, which now claims the stronger
property: "no projection or reduction rewrote the keys ... `keysChanged`
rejects it up front".
`keysChanged` cannot just be tightened, because `isCollapsed` wants the
current meaning — a permutation merges no key, so widening it there only buys a
scan that finds nothing. A second flag keeps both readings:
```scala
val keysChanged =
joinKeyPositions.exists(_.length < childKp.expressions.length) ||
reducers.isDefined
// `identityGrouping` asks the stronger question: a projection that only
reorders the key
// columns merges no key, but it does re-label the groups into a
different key space.
val keysRewritten =
joinKeyPositions.exists(_ != childKp.expressions.indices) ||
reducers.isDefined
```
with `!grouping.keysRewritten` in `identityGrouping`. Saying in the comment
that a reordering projection is out of reach rather than rejected would do as
well.
--
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]
