Skip to content

Commit 63ee2d6

Browse files
committed
Finish the describesSameKeys consolidation and fix its isGrouped note
`KeyedShuffleSpec.isCompatibleWith` now calls `describesSameKeys` too, which is what the previous commit claimed but did not do: it still compared the types and the rows inline, with the whole rationale duplicated. The scaladoc's note on `isGrouped` was wrong. It said two layouts can describe one key set and disagree on whether it is grouped, which contradicts the invariant `PartitioningCollection` asserts two hundred lines below. `isGrouped` follows from the keys, so equal keys cannot answer it differently, and the assert is a consistency check on independently built layouts.
1 parent 72919b8 commit 63ee2d6

1 file changed

Lines changed: 11 additions & 21 deletions

File tree

  • sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala

Lines changed: 11 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -510,16 +510,18 @@ case class KeyLayout(
510510
* The types are asked as well as the rows, because `InternalRowComparableWrapper.equals` compares
511511
* them first and **two empty key lists compare equal whatever they describe**. Without the type
512512
* clause a join between two sides whose partitions were all pruned would call two different key
513-
* spaces one layout. Where a key row exists the type clause is implied.
513+
* spaces one layout, and `ShuffledJoin.outputPartitioning` would then report both as alternative
514+
* descriptions of it. Where a key row exists the type clause is implied.
514515
*
515516
* A type list identifies a space only up to the type, so two empty sides whose spaces differ but
516517
* share a type pair anyway, e.g. `identity(id INT)` against `bucket(4, id INT)`. That residual is
517518
* a decision, not an oversight: an empty layout holds no row, so every claim over it is vacuous,
518519
* and a merge that later brings real rows under it rewrites each member's expressions through
519520
* `reducersBothWays` or a `GroupPartitionsExec` first.
520521
*
521-
* `isGrouped` is deliberately not part of this. Two layouts can describe one key set and disagree
522-
* on whether it is grouped, and the sites that care say so separately.
522+
* `isGrouped` is not part of this, because it follows from the keys: it says they are unique, so
523+
* two layouts over equal keys cannot answer it differently. `PartitioningCollection` still
524+
* asserts it, as a consistency check on layouts that were built independently.
523525
*/
524526
def describesSameKeys(other: KeyLayout): Boolean =
525527
dataTypes == other.dataTypes && partitionKeys == other.partitionKeys
@@ -1337,9 +1339,9 @@ object PartitioningCollection {
13371339
s"dataTypes ${representative.keyDataTypes} with partitionKeys " +
13381340
s"${representative.partitionKeys}, and dataTypes ${canonicalLayout.dataTypes} with " +
13391341
s"partitionKeys ${canonicalLayout.partitionKeys}")
1340-
// Whether the keys are unique is a property of the keys, so two layouts over equal keys
1341-
// that disagree on it cannot both be right. Kept separate, since two layouts can describe
1342-
// one key set and legitimately disagree on it elsewhere.
1342+
// Whether the keys are unique follows from the keys, so two layouts over equal keys that
1343+
// disagree on it cannot both be right. Asserted separately from `describesSameKeys`,
1344+
// which answers what the keys are rather than how they are laid out.
13431345
require(representative.isGrouped == canonicalLayout.isGrouped,
13441346
"All KeyedPartitionings in a PartitioningCollection must agree on isGrouped")
13451347
p match {
@@ -1809,21 +1811,9 @@ case class KeyedShuffleSpec(
18091811
case otherSpec @ KeyedShuffleSpec(otherPartitioning, otherDistribution, _) =>
18101812
distribution.clustering.length == otherDistribution.clustering.length &&
18111813
numPartitions == otherSpec.numPartitions && areKeysCompatible(otherSpec) &&
1812-
// The key rows are compared at their types, since `InternalRowComparableWrapper.equals`
1813-
// compares those first. Two empty key lists compare equal whatever they describe, so the
1814-
// key space is asked separately: without that, a join between two sides whose partitions
1815-
// were all pruned would call two different spaces one layout, and
1816-
// `ShuffledJoin.outputPartitioning` would then report both as alternative descriptions of
1817-
// it. Where a key row exists this clause is implied.
1818-
//
1819-
// A type list still identifies a space only up to the type, so two empty sides whose
1820-
// spaces differ but share a type pair anyway, e.g. `identity(id INT)` against
1821-
// `bucket(4, id INT)`. That residual is a decision, not an oversight: an empty layout
1822-
// holds no row, so every claim over it is vacuous, and a merge that later brings real
1823-
// rows under it rewrites each member's expressions through `reducersBothWays` or a
1824-
// `GroupPartitionsExec` first.
1825-
partitioning.keyDataTypes == otherPartitioning.keyDataTypes &&
1826-
partitioning.partitionKeys == otherPartitioning.partitionKeys
1814+
// The reason the types are asked as well as the rows is on `describesSameKeys`, so the
1815+
// next site comparing keys cannot forget the type clause.
1816+
partitioning.layout.describesSameKeys(otherPartitioning.layout)
18271817
case ShuffleSpecCollection(specs) =>
18281818
specs.exists(isCompatibleWith)
18291819
case _ => false

0 commit comments

Comments
 (0)