ulysses-you commented on code in PR #58339:
URL: https://github.com/apache/spark/pull/58339#discussion_r3914425247
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -578,12 +582,39 @@ case class CoalescedNullAwareHashPartitioning(
* partitioning this one was derived from onto the same
key, so one key here can
* stand for several of the original ones. Sticky. See "Key
Collapse" above for
* what it gates and how it travels.
+ * @param mayContainUnknownPartitionKeys Whether the data may contain rows
whose partition key is
+ * not among the declared `partitionKeys`.
`KeyGroupedPartitioner`
+ * routes such rows by a deterministic hash
when a side is
+ * re-shuffled onto this partitioning (see
+ * `KeyedShuffleSpec.createPartitioning`), so
co-location holds
+ * for whole keys only: two marked
partitionings declaring the
Review Comment:
Thanks, all fair points -- addressed in 912f3d8. The @param's pairing claim
now names the same-function-per-position requirement, both garbled sentences
take your suggested wording, the nested-test comment states the real
unreachability (every inner join clears the mixed collection it builds, so an
all-marked nested shape only appears beside marked siblings), and the GPE
branch checks `reducers.isDefined` before any marker lookup.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/ShuffledJoin.scala:
##########
@@ -80,6 +87,34 @@ trait ShuffledJoin extends JoinCodegenSupport {
s"ShuffledJoin should not take $x as the JoinType")
}
+ /**
+ * Clears the `mayContainUnknownPartitionKeys` marker of every
`KeyedPartitioning` in
+ * `partitionings` when at least one member is unmarked. Only
`ShuffledJoin`'s `InnerLike` arm
+ * can mix the two; see the call site for the argument. The interning
`fromPartitionings` does
+ * afterwards changes no marker.
+ */
+ private def clearUnknownPartitionKeys(
+ partitionings: Seq[Partitioning]): Seq[Partitioning] = {
+ // Only a marked member among an unmarked one needs the rebuild. On the
common all-unmarked
+ // path, even a no-op `copy` is not free: `transform`'s `fastEquals` would
fall through to
+ // comparing every partition key.
+ val marked = partitionings.view.flatMap(PartitioningCollection.flatten)
Review Comment:
Thanks -- done in 912f3d8, and one step further than suggested: the read is
`PartitioningCollection.mayContainUnknownPartitionKeys(p): Boolean` (one cached
first-keyed member, sound via the constructor's marker agreement), used from
both this rebuild decision and the `GroupPartitionsExec` give-up branch, so
subtree-marker reading has exactly one entry point. No `hasKeyedPartitioning`
presence filter: `EnsureRequirements` normalizes both children of a shuffled
join onto one template before the collection is built, so a keyed and a keyless
child never meet here for the fold to matter, and the helper's doc says so.
`markers.forall(_ == markers.head)` short-circuits the common all-unmarked path
before anything is rebuilt, and the per-input pattern guard keeps `transform`
off unmarked subtrees, so the left-deep chain is now O(inputs) per level with
zero allocation on the hot path.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/GroupPartitionsExec.scala:
##########
@@ -79,6 +79,23 @@ case class GroupPartitionsExec(
// single-side-transform reducers; for the both-sides-reduce shape no
single transform
// describes the keys (see `KeyedShuffleSpec.reducersBothWays`).
val partitionKeys = grouping.partitions.map(_._1)
+ // Members carry a uniform marker (mixed collections are cleared at
construction by
+ // `ShuffledJoin`), so any member answers for all of them.
+ val mayContainUnknownKeys = p.exists {
Review Comment:
Thanks, all fair points -- addressed in 912f3d8. The @param's pairing claim
now names the same-function-per-position requirement, both garbled sentences
take your suggested wording, the nested-test comment states the real
unreachability (every inner join clears the mixed collection it builds, so an
all-marked nested shape only appears beside marked siblings), and the GPE
branch checks `reducers.isDefined` before any marker lookup.
--
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]
