@@ -615,7 +615,7 @@ case class EnsureRequirements(
615615 log " join type ' ${MDC (LogKeys .JOIN_TYPE , joinType)}' " )
616616 } else {
617617 // The pre-alignment plan of each side and the grouping this rule inserted over it,
618- // read once: the statistics and the original partition keys below come from the
618+ // are read once: the statistics and the original partition keys below come from the
619619 // plan, the positions projecting them from the grouping.
620620 val leftGrouping = innermostGroupPartition(left)
621621 val rightGrouping = innermostGroupPartition(right)
@@ -642,8 +642,8 @@ case class EnsureRequirements(
642642 | """ .stripMargin)
643643 leftLink.get.stats.sizeInBytes < rightLink.get.stats.sizeInBytes
644644 } else {
645- // As a simple heuristic, we pick the side with fewer number of partitions to
646- // apply the grouping & replication of partitions. The counts read the
645+ // As a simple heuristic, we pick the side with fewer partitions to apply the
646+ // grouping & replication of partitions. The counts read the
647647 // pre-alignment plans, for the same reason the statistics do: on a re-run both
648648 // aligned reports hold the same number of keys, so comparing them decides nothing.
649649 // This also changes a first pass, which compared the aligned report's distinct
@@ -760,9 +760,9 @@ case class EnsureRequirements(
760760 *
761761 * The descent only traverses a `GroupPartitionsExec` and a *local* `SortExec`. That bound is a
762762 * decision, not an omission: a `GroupPartitionsExec` hidden behind any other node belongs to a
763- * different operator, and reusing it would move that operator's alignment. Instrumenting the
764- * descent over `KeyGroupedPartitioningSuite`, the non-`SortExec` shapes hiding a node are
765- * `Project > SortMergeJoin > Sort > GroupPartitions` and `Project > Filter > Window >
763+ * different operator, and reusing it would move that operator's alignment. Instrumentation of
764+ * the descent over `KeyGroupedPartitioningSuite` found these non-`SortExec` shapes hiding a
765+ * node: `Project > SortMergeJoin > Sort > GroupPartitions` and `Project > Filter > Window >
766766 * WindowGroupLimit > GroupPartitions`, where refusing to descend is right every time. A global
767767 * `SortExec` also stops the descent: it requires `OrderedDistribution`, which a
768768 * `KeyedPartitioning` can satisfy (behind `spark.sql.sources.v2.bucketing.sorting.enabled`)
@@ -772,8 +772,9 @@ case class EnsureRequirements(
772772 private def innermostGroupPartition (
773773 plan : SparkPlan ): Option [(GroupPartitionsExec , SparkPlan => SparkPlan )] = plan match {
774774 case g : GroupPartitionsExec =>
775- // A grouping over another grouping is one this rule added in an earlier pass: keep the
776- // descent below it and drop this one.
775+ // When groupings stack, the outer one is the wrap this invocation's distribution step
776+ // just added; the one below is inherited from an earlier pass and owns the alignment to
777+ // preserve. Keep the descent below the outer node and drop it.
777778 innermostGroupPartition(g.child).orElse(Some ((g, identity[SparkPlan ])))
778779 case s : SortExec if ! s.global =>
779780 innermostGroupPartition(s.child).map { case (g, rebuild) =>
0 commit comments