LuciferYang commented on code in PR #58419:
URL: https://github.com/apache/spark/pull/58419#discussion_r3981176170
##########
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:
Fixed in `ef44aa21596`, though not by the design you prescribed. The hole is
real, and I turned it into a test before changing anything: with the reason on
a tag, `rebuilt.supportCodegen` came back true after `withNewChildren`
installed a child the gate rejects.
Instead of moving both decisions into per-instance state with a handoff from
`CollapseCodegenStages`, I split the reason by what makes each term move. The
two confs stay latched in a tag, since they move only because the read happens
at a different time. Everything that reads the children (nested union,
multi-RDD child, partition-index dependence, child count, `supportsColumnar`,
type mismatch) is memoized per instance, so a copy starts cold and answers
against its own children.
The copy in the codegen shell still agrees with the gate, and by
construction rather than by routing: `InputAdapter` delegates `output` and
`supportsColumnar` to its child, the other terms walk the subtree through it,
and each of them is fixed for a given set of children
(`InMemoryTableScanExec.supportsColumnar` is an `override val`,
`AdaptiveSparkPlanExec`'s is a constructor val). So this needs no
`UnionExec`-shaped hook inside a generic rule, and the guarantee does not rest
on `insertInputAdapter` remaining the only rebuild path between the gate and
execution. The new case is a plain unit test with no injected extension: force
`metrics`, hand the node a nested union through `withNewChildren`, assert the
gate flips.
`isPlainUnion` stays latched on purpose. Stale-plain is the conservative
direction (the node reports `UnknownPartitioning`), per-plan stability is what
fixes the wrong answer this PR is about, and making it per-instance would let
`CoalesceShufflePartitions`' grouping decision and a later rebuilt node
disagree again. Glad to reconsider if you read that one differently.
--
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]
