ulysses-you commented on code in PR #58279:
URL: https://github.com/apache/spark/pull/58279#discussion_r3921909354
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -732,6 +732,47 @@ case class EnsureRequirements(
case other => other
}
+ /**
+ * Finds the innermost `GroupPartitionsExec` in `plan`, rewrites it with
`f`, and drops any
+ * redundant grouping stacked above it. Returns `None` when `plan` holds no
`GroupPartitionsExec`,
+ * leaving it to the caller to create one.
+ *
+ * This is what makes the rule idempotent for storage-partitioned joins.
`EnsureRequirements` is
+ * re-run on plans it already produced --
`ConvertSortMergeJoinToShuffledHashJoin` and
+ * `OptimizeSkewedJoin` hand the whole tree back to it after rewriting some
other join -- so a
+ * join child arrives as `SortExec(GroupPartitionsExec(...))` rather than a
bare scan. The
+ * distribution step then adds a plain `GroupPartitionsExec` on top, because
a partially clustered
+ * `KeyedPartitioning` reports `isGrouped = false` by design and so is only
satisfied "after
+ * grouping". Rewriting that fresh outer node instead of the one below it
re-derives the
+ * alignment from an already-aligned layout: the inner node replicates an
input partition across
+ * the expected partitions and the outer one concatenates those replicas
back together before
+ * replicating again, duplicating rows. Descending to the innermost node and
dropping what sits
+ * above it reproduces exactly the plan a single pass would have produced.
+ *
+ * Only a *local* `SortExec` is traversed. A global one requires
`OrderedDistribution`, which a
Review Comment:
Done -- the scaladoc states the rule the descent follows (only nodes this
rule itself inserted directly above the child, with the shapes you measured)
and now lives on the shared descent helper `innermostGroupPartition`, per your
suggestion of one descent serving both the statistics read and the rewrite.
5af8667631b
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -745,32 +786,29 @@ case class EnsureRequirements(
mergedPartitionKeys: Seq[(InternalRowComparableWrapper, Int)],
reducers: Option[Seq[Option[Reducer[_, _]]]],
distributePartitions: Boolean): SparkPlan = {
- plan match {
- case g: GroupPartitionsExec =>
- val newGroupPartitions = g.copy(
- joinKeyPositions = joinKeyPositions,
- expectedPartitionKeys = Some(mergedPartitionKeys),
- reducers = reducers,
- distributePartitions = distributePartitions)
- newGroupPartitions.copyTagsFrom(g)
- newGroupPartitions
- case _ =>
- GroupPartitionsExec(plan, joinKeyPositions, Some(mergedPartitionKeys),
reducers,
- distributePartitions)
+ rewriteGroupPartitions(plan) { g =>
Review Comment:
Noted -- the five-shape table and the decide-the-node-once direction go into
a separate follow-up ticket.
--
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]
