ulysses-you commented on code in PR #58339:
URL: https://github.com/apache/spark/pull/58339#discussion_r3939155993
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala:
##########
@@ -78,6 +78,18 @@ case class GroupPartitionsExec(
// data types match the reduced partition keys for the
identity-vs-transform and
// single-side-transform reducers; for the both-sides-reduce shape no
single transform
// describes the keys (see `KeyedShuffleSpec.reducersBothWays`).
+ //
+ // A marked claim pins undeclared rows to hash(key) % numPartitions
(see
+ // `KeyedPartitioning.mayContainUnknownPartitionKeys`). Only an
identity grouping keeps
+ // that relationship: any other grouping -- a reorder, a coalesce, a
resize, or the
+ // collapse a reduction applies -- moves those rows. Clearing only the
marker would
+ // misreport the undeclared rows that remain, so give up the keyed
partitioning at the
+ // physical output count (one per group, padding included) that a
parent's
+ // `PartitioningCollection` requires for uniformity.
`identityGrouping` is a lazy val, so
+ // repeated `outputPartitioning` calls scan it at most once.
+ if (PartitioningCollection.keyedMarkerOf(p).contains(true) &&
!identityGrouping) {
Review Comment:
Thank you for the second look. The pairing-commit gap is real: a marked side
whose regrouping is not the identity gives up its keyed claim, and the
regrouping test exercises that on this head (the committed pair then carries a
child `ValidateRequirements` rejects, at the cost of AQE rules on that stage;
results stay correct because the give-up forces downstream re-shuffles).
Closing it at the pairing site is not a per-side `satisfies` though. The
soundness of the join scenario rests on `alignToExpectedKeys`: the pairing
aligns both rebuilt children onto the same expected key sequence - same per-key
partition spread, padding and partial clustering included - so they co-locate
as a pair even when neither satisfies its `ClusteredDistribution` individually
(partial clustering deliberately leaves them ungrouped; measured, a per-side
`satisfies` gate regresses the partial-clustering suites). What the commit
should be re-checked against is that joint alignment: whether both rebuilt
children still declare the same aligned key sequence and neither gave up its
keyed layout. A pairwise check capturing that, reconciled with
`ValidateRequirements`' per-side semantics, is the scope of SPARK-59272.
##########
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:
Fixed in 3c6b0143308: `a marked member alongside an unmarked one is
spurious`.
##########
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:
Fixed in 3c6b0143308: the identity check now streams
`grouping.partitions.iterator.zipWithIndex` instead of materializing one tuple
per physical partition.
--
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]
