peter-toth commented on code in PR #58339:
URL: https://github.com/apache/spark/pull/58339#discussion_r3924348492
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala:
##########
@@ -219,6 +236,21 @@ case class GroupPartitionsExec(
PartitionGrouping(partitions, isGrouped, isCollapsed)
}
+ /**
+ * Whether this node's grouping leaves every partition where it was and
keeps all of them, i.e.
+ * output partition i holds exactly input partition i and there is one
output per input. That is
+ * the only grouping that keeps a marked layout's undeclared rows at
hash(key) % numPartitions:
+ * a grouping that drops trailing declared keys still reads identity for
every group it keeps,
+ * but the partition count shrinks and the hash modulus with it. The
`forall` stops at the first
+ * moved partition, so a reorder or coalesce is rejected without a full scan.
+ */
+ @transient private lazy val identityGrouping: Boolean =
Review Comment:
**Finding 26.** `identityGrouping` compares partition indices and counts,
and never asks whether the grouping rewrote the keys. The keys the node goes on
to declare are `grouping.partitions.map(_._1)`, which are the projected and
reduced ones, so a reduction that is injective and order-preserving on the
declared keys passes the predicate.
Take a marked child with keys `[0, 1]` and `reducers = Some(Seq(Some(r)))`
where `r` maps `0 -> 0` and `1 -> 1`. That is two groups, one input partition
each, in the same order, count unchanged, so `identityGrouping` is `true`. The
node then reports a marked `KeyedPartitioning` whose declared keys live in the
reducer's key space, while its undeclared rows still sit at `hash(originalKey)
% 2`. The claim's routing is about a different key than the one it declares.
That contradicts two statements this PR adds. Rule (1) of the `@param` at
`sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:601`
lists `reducer` among the coarsening sites that must drop the keyed
partitioning. The comment at line 85 above says a reduction moves the rows.
Only a collapsing reduction is caught, and it is the count clause that catches
it, not the reduction. `GroupPartitionsExecSuite`'s `mod 2` over `[1, 2, 3]`
collapses, so it passes either way.
Not reachable today. I said the opposite at round 6, where I wrote that the
give-up catches a contract-violating self-reducer, and it does not.
`areKeysCompatible`'s marked path demands `isSameFunction`, and
`BucketFunction.reducer` returns null for equal bucket counts, so
`reducersBothWays` gives `(None, None)`. Only a connector whose `reducer` is
non-identity for its own function at its own parameters gets here. The
projection half is closed the same way, since none of the three producers of
`joinKeyPositions` can narrow a marked partitioning.
So this is defence in depth, and it is one clause. `grouping` already
computes exactly the predicate:
```scala
val keysChanged =
joinKeyPositions.exists(_.length < childKp.expressions.length) ||
reducers.isDefined
```
Carry it on `PartitionGrouping` and add `!grouping.keysChanged` to
`identityGrouping`. Tightening the two comments instead would close it too, but
the clause is cheaper than the prose.
--
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]
