cloud-fan commented on code in PR #58339:
URL: https://github.com/apache/spark/pull/58339#discussion_r3933745511
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/ShuffledJoin.scala:
##########
@@ -69,8 +69,15 @@ trait ShuffledJoin extends JoinCodegenSupport {
override def outputPartitioning: Partitioning = joinType match {
case _: InnerLike =>
+ // Every `KeyedPartitioning` in the joined collection speaks the same
declared key set (the
+ // `PartitioningCollection` invariant), and `keysSatisfy` admits a
marked side into it only
+ // for full-key join keys, where the join equality ties every column of
its claim: a row of
+ // an undeclared key matches nothing on the accurate side and is
filtered, so a marked
+ // member among an unmarked one is spurious. Clear it at the one site
that can mix them, so
Review Comment:
**Nit (P3):** Nit: `among` cannot grammatically take the singular `an
unmarked one` here. Could this say `a marked member alongside an unmarked one`?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala:
##########
@@ -216,9 +239,31 @@ case class GroupPartitionsExec(
group.tail.exists(childKeys(_) != first)
}
}
- PartitionGrouping(partitions, isGrouped, isCollapsed)
+ PartitionGrouping(partitions, isGrouped, isCollapsed, keysRewritten)
}
+ /**
+ * Whether this node's grouping leaves the declared keys and every partition
where they were:
+ * no projection or reduction rewrote the keys, 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
projection or reduction
+ * re-labels the groups into a different key space, so even a grouping whose
indices line up
+ * would pin the claim to keys it no longer declares; `keysRewritten`
rejects it up front --
+ * it covers a narrowing projection, a reordering one, and any reducer slot.
A reducer slot
+ * is treated as key-changing: a conforming self-reducer cannot rewrite a
reachable key
+ * value, so the give-up there loses at most an optimization. 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 =
+ !grouping.keysRewritten &&
+ grouping.partitions.size == child.outputPartitioning.numPartitions &&
+ grouping.partitions.zipWithIndex.forall {
Review Comment:
**Nit (P3):** Nit: this strict `zipWithIndex` materializes one tuple per
physical partition before `forall`. Please use
`grouping.partitions.iterator.zipWithIndex.forall` to keep the same order and
short-circuiting without the intermediate collection.
--
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]
