LuciferYang commented on code in PR #58419:
URL: https://github.com/apache/spark/pull/58419#discussion_r3939232797
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala:
##########
@@ -1023,13 +1026,53 @@ case class UnionExec(children: Seq[SparkPlan]) extends
SparkPlan with CodegenSup
}
}
- // True when the codegen path applies: `outputPartitioning` is
`UnknownPartitioning`,
- // and `unionedInputRDD` matches the semantics of `sparkContext.union(...)`
in `doExecute`.
- // A `KeyedPartitioning` union also uses `sparkContext.union(...)` in
`doExecute`, but
- // codegen is disabled for it (`supportCodegenFailureReason` reports
"partitioning-aware"):
- // the per-partition key descriptor is consumed by a downstream
`GroupPartitionsExec`, and
- // keeping these unions out of whole-stage codegen matches the
`HashPartitioning` union case.
- private[sql] def isPlainUnion: Boolean =
outputPartitioning.isInstanceOf[UnknownPartitioning]
+ // Serializes the latch below so concurrent first readers agree on one
answer. Not this node's own
+ // monitor, which `unionedInputRDD`'s `lazy val` holds while it drives
`child.execute()`.
+ // Driver-only, hence `@transient`.
+ @transient private val decisionLock = new Object()
+
+ /**
+ * True when this union behaves as a plain concatenation, so
`unionedInputRDD` matches
+ * `sparkContext.union(...)` in `doExecute` and the codegen path applies. A
`KeyedPartitioning`
+ * union also concatenates, but codegen stays off for it:
`supportCodegenFailureReason` reports
+ * "partitioning-aware", because a downstream `GroupPartitionsExec` consumes
its key descriptor.
+ *
+ * Latched, because the answer moves under its consumers.
+ * `InMemoryTableScanExec.outputPartitioning` reports `UnknownPartitioning`
while its inner
+ * `AdaptiveSparkPlanExec` has no final plan, so a union can look plain when
+ * `CollapseCodegenStages` gates on it and partitioning-aware by the time
the stage runs. The
+ * shell that gate builds wraps a `withNewChildren` copy, and a copy that
re-derives comes back
+ * with empty `metrics` while `doProduce` asks `metricTerm` for
`numOutputRows`. A `TreeNodeTag`
+ * survives that rebuild where a field would not, since `withNewChildren`
ends in `copyTagsFrom`.
+ *
+ * `UNION_OUTPUT_PARTITIONING` is read here rather than in `rawPartitioning`
so it is latched too:
+ * `conf` is live, and re-reading it let a plan made with the conf on
execute with it off.
+ */
+ private[sql] def isPlainUnion: Boolean = {
+
decisionLock.synchronized(getTagValue(UnionExec.PLAIN_UNION_DECISION)).getOrElse
{
+ val plain = !conf.getConf(SQLConf.UNION_OUTPUT_PARTITIONING) ||
+ rawPartitioning.isInstanceOf[UnknownPartitioning]
+ decisionLock.synchronized {
+ getTagValue(UnionExec.PLAIN_UNION_DECISION).getOrElse {
+ setTagValue(UnionExec.PLAIN_UNION_DECISION, plain)
+ plain
+ }
+ }
+ }
+ }
+
+ /**
+ * A node latched plain reports `UnknownPartitioning` even once its children
agree on a concrete
+ * one: a fused union concatenates, and claiming their partitioning would
let a parent skip an
+ * exchange it needs. The cost is SPARK-52921's exchange elimination for
such a union.
Review Comment:
Your reproduction matched mine exactly, down to the ten rows of four. This
became SPARK-59141 (#58445) and landed on master, so it arrives here through
the merge in `32a2593d7ba`, with `doExecute` and `doExecuteColumnar` sharing
one `unionRDDs` helper. The bug is reachable on branch-4.2 and branch-4.1 as
well, so I have backport PRs open for both (#58511, #58512).
--
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]
