LuciferYang commented on code in PR #58419:
URL: https://github.com/apache/spark/pull/58419#discussion_r3999587333
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala:
##########
@@ -1044,11 +1091,21 @@ case class UnionExec(children: Seq[SparkPlan]) extends
SparkPlan with CodegenSup
}
}
- // Memoized: consulted by `supportCodegen` (called multiple times by
- // `CollapseCodegenStages`) and by `metrics`. Conf and children are stable
- // for a given UnionExec instance; cross-plan staleness is impossible since
- // UnionExec is a case class and `withNewChildren` produces a fresh instance.
- @transient private lazy val supportCodegenFailureReason: Option[String] = {
+ // Latched for the same reason `isPlainUnion` is: `supportCodegen` and
`metrics` must see one
+ // answer, and `conf` is live. When a child is not `CodegenSupport`,
`insertInputAdapter` wraps
+ // it, so `withNewChildren` returns a real copy whose first evaluation of
this would land at
+ // execution; re-deriving there left `metrics` empty while `doProduce` asked
`metricTerm` for
+ // `numOutputRows`. The first force is not always the gate: under AQE it is
a plan-update event
+ // on the pre-stage-creation tree, so a term added here sees more of the
plan than the gate does.
+ private def supportCodegenFailureReason: Option[String] =
decisionLock.synchronized {
+ getTagValue(UnionExec.CODEGEN_FAILURE_REASON).getOrElse {
Review Comment:
Three cases added in `SparkSessionExtensionSuite`, each building a
`UnionExec` through a real hook: an injected columnar rule with AQE off, an
injected query stage prep rule, and the columnar hook inside AQE post stage
creation. Each one prepares a union over a repartition, then turns
`spark.sql.unionOutputPartitioning` off and reads the node again. A stamped
node keeps answering from the decision it was prepared with; one no barrier
reached derives a decision at that read and comes back `UnknownPartitioning`.
I then removed the barriers one at a time to see what each case actually
pins:
| removed | columnar, AQE off | prep rule | post stage creation |
| --- | --- | --- | --- |
| `preparations`, after the columnar rules | **fails** | passes | passes |
| AQE, after the injected prep rules | passes | passes | passes |
| AQE `postStageCreationRules` | passes | passes | **fails** |
| both AQE listings | passes | **fails** | **fails** |
Two of the listings are pinned one for one. The AQE pair is pinned jointly,
because the barrier in `postStageCreationRules` stands behind the one after the
prep rules: nothing an injected prep rule creates can reach execution undecided
even with that listing gone, so a case that fails on it alone does not exist. I
would rather tell you that than write one that looks like it does. What the
earlier listing buys is upstream of execution:
`CoalesceShufflePartitions.childrenNeedCompatiblePartitioning` reads
`isPlainUnion` to decide whether a union's children coalesce as one group, and
it runs before the post-stage barrier, so stamping earlier keeps that reader
and the final decision in step. The test name and comment say exactly that much
and no more.
An injected stage-optimizer rule needs no case of its own: it runs ahead of
the same `postStageCreationRules` barrier, which the third case pins.
Your two nits are in as well. `rawPartitioning`'s scaladoc now describes it
as the child-derived candidate and gives the plain answer's two grounds
separately, and the `whereas`.
--
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]
