LuciferYang commented on code in PR #58419:
URL: https://github.com/apache/spark/pull/58419#discussion_r4038314034
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala:
##########
@@ -1284,6 +1391,35 @@ case class UnionExec(children: Seq[SparkPlan]) extends
SparkPlan with CodegenSup
}
object UnionExec {
+ /**
+ * What `StampUnionDecisions` fixes on a `UnionExec`: whether it is a plain
concatenation, and the
+ * two confs the codegen gate reads. Everything else the gate asks is
derived per instance, so a
+ * rule replacing a child cannot inherit an answer taken from the topology
it replaced.
+ */
+ private case class Decisions(
+ plainUnion: Boolean,
+ unionCodegenEnabled: Boolean,
+ maxChildren: Int)
+
+ /**
+ * The stamped decisions. See `isPlainUnion` and `stampDecisions`.
+ *
+ * `withNewChildren` copies the tag onto a rebuilt node, and so does a
transform rule's
+ * replacement, but only where the target carries no tags of its own:
`copyTagsFrom` leaves a node
+ * that already has some untouched. A `UnionExec` reaching execution
unstamped therefore answers
+ * from the state it sees then, and can leave `metrics` empty, so
`doProduce` fails asking
+ * `metricTerm` for `numOutputRows`.
+ */
+ private val DECISIONS = TreeNodeTag[Decisions]("unionDecisions")
Review Comment:
Done for the configuration half. Both barriers in one preparation now take
the same `UnionConfSnapshot`, read once in `QueryExecution.preparations` and
once per `AdaptiveSparkPlanExec`, and `stampDecisions` records that value on a
node that reached the barrier without one. So the replacement you describe is
stamped from what `EnsureRequirements` read rather than from the conf as it is
by then. A new case pins it: an unrelated tag on a replacement, then a late
barrier run while the conf says the opposite. It fails with
`UnknownPartitioning(0)` without the change.
I would rather not move the state off tags, though. What loses it is
`TreeNode.transformUp` calling `afterRule.copyTagsFrom(this)` on a rule's
replacement node, and that is the only hook such a node goes through. A field
outside the product is not copied there at all, so it would be lost in this
case and also in every case where tags do survive today: `withNewChildren`,
`mapChildren`, and the copy `insertInputAdapter` puts inside the codegen shell,
which is the one that has to agree with the gate. Overriding
`withNewChildrenInternal` covers the copies we make; nothing covers a node a
rule constructed.
Failing closed on an unprepared union has the same problem from the other
side: `outputPartitioning` and `supportCodegen` are read on plans that never
went through `prepareForExecution`, which is what the `reading the unprepared
plan does not decide the prepared one` case covers, and answering those from
the live conf is what every union does on master today, so it cannot regress
anything. The stamp only has to win where a decision has to hold still.
--
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]
