cloud-fan commented on code in PR #58279:
URL: https://github.com/apache/spark/pull/58279#discussion_r3933715965
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -634,11 +642,18 @@ case class EnsureRequirements(
|""".stripMargin)
leftLink.get.stats.sizeInBytes < rightLink.get.stats.sizeInBytes
} else {
- // As a simple heuristic, we pick the side with fewer number of
partitions
- // to apply the grouping & replication of partitions
+ // As a simple heuristic, we pick the side with fewer number of
partitions to
Review Comment:
**Nit (P3):** `fewer number of partitions` is ungrammatical. Please use
`fewer partitions` (or `a smaller number of partitions`).
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -611,8 +614,13 @@ case class EnsureRequirements(
logInfo(log"Skipping partially clustered distribution as it cannot
be applied for " +
log"join type '${MDC(LogKeys.JOIN_TYPE, joinType)}'")
} else {
- val unwrappedLeft = unwrapGroupPartitions(left)
- val unwrappedRight = unwrapGroupPartitions(right)
+ // The pre-alignment plan of each side and the grouping this rule
inserted over it,
Review Comment:
**Nit (P3):** This sentence is missing a finite verb. Please change `read
once` to `are read once` so the subject (`The pre-alignment plan ... and the
grouping ...`) has a predicate.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -731,38 +754,97 @@ case class EnsureRequirements(
}
/**
- * Unwraps a GroupPartitionsExec to get the underlying child plan.
+ * The innermost `GroupPartitionsExec` reachable from `plan` by descending
only through nodes
+ * this rule itself inserted above it, together with a function rebuilding
the traversed local
+ * sorts over a replacement node. `None` when no `GroupPartitionsExec` is
reachable.
+ *
+ * The descent only traverses a `GroupPartitionsExec` and a *local*
`SortExec`. That bound is a
+ * decision, not an omission: a `GroupPartitionsExec` hidden behind any
other node belongs to a
+ * different operator, and reusing it would move that operator's alignment.
Instrumenting the
+ * descent over `KeyGroupedPartitioningSuite`, the non-`SortExec` shapes
hiding a node are
+ * `Project > SortMergeJoin > Sort > GroupPartitions` and `Project > Filter
> Window >
+ * WindowGroupLimit > GroupPartitions`, where refusing to descend is right
every time. A global
+ * `SortExec` also stops the descent: it requires `OrderedDistribution`,
which a
+ * `KeyedPartitioning` can satisfy (behind
`spark.sql.sources.v2.bucketing.sorting.enabled`)
+ * through a `GroupPartitionsExec` built to emit the partition keys in
sorted order, and
+ * reusing that node for a join would destroy the ordering it exists to
provide.
*/
- private def unwrapGroupPartitions(plan: SparkPlan): SparkPlan = plan match {
- case g: GroupPartitionsExec => g.child
- case other => other
+ private def innermostGroupPartition(
+ plan: SparkPlan): Option[(GroupPartitionsExec, SparkPlan => SparkPlan)]
= plan match {
+ case g: GroupPartitionsExec =>
+ // A grouping over another grouping is one this rule added in an earlier
pass: keep the
Review Comment:
**Nit (P3):** This comment reverses the ownership of the two layers. On a
rerun, this invocation's distribution step has just added the outer grouping;
the inner GroupPartitionsExec is the one inherited from the earlier pass and
owns the alignment we need to preserve. Please describe the outer node as
current-pass and the inner node as earlier-pass.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -731,38 +754,97 @@ case class EnsureRequirements(
}
/**
- * Unwraps a GroupPartitionsExec to get the underlying child plan.
+ * The innermost `GroupPartitionsExec` reachable from `plan` by descending
only through nodes
+ * this rule itself inserted above it, together with a function rebuilding
the traversed local
+ * sorts over a replacement node. `None` when no `GroupPartitionsExec` is
reachable.
+ *
+ * The descent only traverses a `GroupPartitionsExec` and a *local*
`SortExec`. That bound is a
+ * decision, not an omission: a `GroupPartitionsExec` hidden behind any
other node belongs to a
+ * different operator, and reusing it would move that operator's alignment.
Instrumenting the
Review Comment:
**Nit (P3):** The opening modifier dangles because the following subject is
`the non-SortExec shapes`. Please use a construction such as `Instrumentation
of the descent over KeyGroupedPartitioningSuite found these non-SortExec shapes
...`.
--
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]
