ulysses-you commented on code in PR #58279:
URL: https://github.com/apache/spark/pull/58279#discussion_r3939078041
##########
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:
Done -- the comment now attributes the outer grouping to this invocation's
distribution step and the one below to the earlier pass.
3d07619d4e0
##########
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:
Done -- `are read once`.
3d07619d4e0
##########
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:
Done -- `fewer partitions`.
3d07619d4e0
##########
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:
Done -- reworded as suggested: "Instrumentation of the descent over
`KeyGroupedPartitioningSuite` found these non-`SortExec` shapes hiding a node:
...".
3d07619d4e0
--
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]
