peter-toth commented on code in PR #58552:
URL: https://github.com/apache/spark/pull/58552#discussion_r3979225183
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -1229,41 +1275,57 @@ object PartitioningCollection {
representativeOf(p).map(_.mayContainUnknownPartitionKeys)
/**
- * Builds a [[PartitioningCollection]], unifying the `partitionKeys`
reference across all
- * [[KeyedPartitioning]]s (including those in nested collections). Use this
when combining
- * independently-computed partitionings (e.g. join `outputPartitioning`)
where
- * `KeyedPartitioning.partitionKeys` are structurally equal but may not be
reference-equal.
+ * Builds a [[PartitioningCollection]], unifying the [[KeyLayout]] reference
across all
+ * [[KeyedPartitioning]]s, including those in nested collections. Use this
when combining
+ * independently-computed partitionings, such as a join's
`outputPartitioning`, whose layouts
+ * describe the same partitions but are not the same object.
*
* Note: this can't be implemented with `TreeNode.transform`.
*/
def fromPartitionings(partitionings: Seq[Partitioning]):
PartitioningCollection = {
// See the class doc for why the flags are normalized by OR rather than
required to agree. One
- // representative per member is enough, because every collection agrees on
the flags
- // internally by this same construction, and only a member that disagrees
is rebuilt.
+ // representative per member is enough, because every collection agrees
internally by this same
+ // construction, and only a member that disagrees is rebuilt.
val anyCollapsed =
partitionings.exists(representativeOf(_).exists(_.isCollapsed))
val anyUnknownKeys =
partitionings.exists(representativeOf(_).exists(_.mayContainUnknownPartitionKeys))
- var canonicalKeys: Seq[InternalRowComparableWrapper] = null
+ var canonicalLayout: KeyLayout = null
// A partitioning with no `KeyedPartitioning` in it has nothing to
normalize, and one that
- // already agrees on the keys and both flags is returned as it is. That is
what keeps
- // repeated `outputPartitioning` computations over deeply nested
collections (e.g. chains of
- // same-key joins) O(1) per level.
+ // already holds the canonical layout is returned as it is. That is what
keeps repeated
+ // `outputPartitioning` computations over deeply nested collections (e.g.
chains of same-key
+ // joins) O(1) per level.
def intern(p: Partitioning): Partitioning = representativeOf(p) match {
case None => p
case Some(representative) =>
- if (canonicalKeys == null) canonicalKeys = representative.partitionKeys
- if ((representative.partitionKeys eq canonicalKeys) &&
- representative.isCollapsed == anyCollapsed &&
- representative.mayContainUnknownPartitionKeys == anyUnknownKeys) {
+ if (canonicalLayout == null) {
+ val layout = representative.layout
+ canonicalLayout =
+ if (layout.isCollapsed == anyCollapsed &&
+ layout.mayContainUnknownPartitionKeys == anyUnknownKeys) {
+ layout
+ } else {
+ layout.copy(
+ isCollapsed = anyCollapsed, mayContainUnknownPartitionKeys =
anyUnknownKeys)
+ }
+ }
+ if (representative.layout eq canonicalLayout) {
p
} else {
- require(representative.partitionKeys == canonicalKeys,
+ require(representative.partitionKeys ==
canonicalLayout.partitionKeys,
"All KeyedPartitionings in a PartitioningCollection must have
equal partitionKeys")
+ // Whether the keys are unique is a property of the keys, so two
layouts over equal keys
+ // that disagree on it cannot both be right.
+ require(representative.isGrouped == canonicalLayout.isGrouped,
+ "All KeyedPartitionings in a PartitioningCollection must agree on
isGrouped")
+ // Interning replaces a member's layout whole, so a member that
describes another key
+ // space would be silently retyped. Two empty key lists compare
equal whatever they
+ // describe, which is the case the clause above them cannot see.
+ require(representative.keyDataTypes == canonicalLayout.dataTypes,
Review Comment:
Taken in
[`72919b8`](https://github.com/apache/spark/commit/72919b80ccaca955b024658cda277d70a1a12648)
and finished in
[`63ee2d6`](https://github.com/apache/spark/commit/63ee2d6c53e324ef0a585362638669b3431f1476).
`KeyLayout.describesSameKeys` holds the predicate, and both `intern` and
`KeyedShuffleSpec.isCompatibleWith` call it.
To be accurate about the two commits: the first added the method and moved
`intern` onto it, but left `isCompatibleWith` comparing the types and the rows
inline with the rationale duplicated, so your "the next site that compares keys
cannot forget the type clause" was only half met. The second moves that site
too and deletes the duplicated comment.
The rationale now lives once: why the types are asked as well as the rows,
and the residual you and ulysses-you both landed on, that two empty sides
sharing a type pair still merge and why that is inert.
`isGrouped` stays a separate `require`, and the note on it in the first
commit was wrong: it said two layouts can describe one key set and disagree on
whether it is grouped, which contradicts the invariant asserted two hundred
lines below. `isGrouped` follows from the keys, so equal keys cannot answer it
differently. The second commit says that instead, and frames the `require` as
what it is, a consistency check on layouts built independently.
The `keyDataTypes` versus `dataTypes` mixture is gone from the requires too.
--
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]
