LuciferYang commented on code in PR #58419:
URL: https://github.com/apache/spark/pull/58419#discussion_r3995829265
##########
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:
Done in `3ed87a98166`, with a trim in `773c28999eb`. `StampUnionDecisions`
runs right after `EnsureRequirements` in both `QueryExecution.preparations` and
AQE's `queryStagePreparationRules`, and nothing else writes the decision, so a
read before that point answers from the state it sees and decides nothing.
Both halves reproduce as test failures before the change. Reading
`queryExecution.sparkPlan`'s union partitioning left the prepared plan
reporting `UnknownPartitioning(0)`, the pre-`EnsureRequirements` answer riding
in through `clone` and `copyTagsFrom`. And with whole-stage codegen off over a
root union, where nothing consults the node during preparation, a conf flip
afterwards moved the executed layout from four partitions to eight.
I put the stamp immediately after `EnsureRequirements` rather than at the
end of preparation. Late stamping has a worse failure: `EnsureRequirements` can
elide a parent's exchange on the strength of a concrete pass-through, and a
stamp taken afterwards could freeze plain, leaving the union to concatenate
under a parent that no longer shuffles. Stamping early can only freeze an
answer a later rule would have sharpened, which costs an exchange elision
rather than a result, and every rule below the stamp then reads the frozen
value, including the `EnsureRequirements` re-runs inside `OptimizeSkewedJoin`.
Your constraint about pre-preparation reads sent me back to the codegen
confs, which were still pinned on first read. They are stamp-only now too, so
the whole decision is one immutable value written in one place. That also left
`decisionLock` with nothing to serialize, so it and its lock-order comment are
gone, and the two tags became one `Decisions(plainUnion, unionCodegenEnabled,
maxChildren)`, written by a single `setTagValue`, so a concurrent reader can no
longer see half a decision. The cost of the read-no-write change is that a
`UnionExec` built after the stamp point by an injected rule no longer gets the
conf pinned by the gate's own read, so for that node a conf flip can still
reach the copy in the codegen shell. It seemed the right way round, since such
a node has no protection on the partitioning half either.
On the ordering question: reordering the gate does not buy anything once the
decision is stamped, because every `UnionExec` is asked during preparation
whether or not the gate would reject it, and `isPlainUnion` no longer writes. I
tried it and reverted it in `dc85f998986`; all it changed was the reason
reported for a union failing several gates.
I cannot confirm the other half of your question, and I think the answer is
the opposite of hopeful. The stamp lands after `EnsureRequirements`, which is
still before a cached child's inner AQE plan finalises, so such a union is
stamped plain and keeps reporting `UnknownPartitioning` for the rest of the
query. `UnionCodegenSuite`'s first `SPARK-59122` case shows it: after
`collect()` the fused union reports `UnknownPartitioning` while its children
have both reached the same concrete layout, which the test now asserts as the
premise of that check. So the elimination is not merely unavailable already, it
is given up. Deferring past that point means deciding at execution, which is
the crash this PR fixes; `outputPartitioning`'s scaladoc records the trade
rather than hiding it.
--
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]
