cloud-fan commented on code in PR #58419:
URL: https://github.com/apache/spark/pull/58419#discussion_r4000488270
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala:
##########
@@ -1035,13 +1039,79 @@ 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 `unionRDDs`.
- // A `KeyedPartitioning` union also uses `sparkContext.union(...)` in
`unionRDDs`, 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]
+ /**
+ * True when this union behaves as a plain concatenation, so
`unionedInputRDD` matches the
+ * semantics of `sparkContext.union(...)` in `unionRDDs`. It satisfies the
partitioning gate on
+ * the codegen path, not the whole of it: `supportCodegenFailureReason`
still applies its other
+ * checks. When this union merges its children's `KeyedPartitioning`
instead, it concatenates all
+ * the same, but codegen stays off, with the reason "partitioning-aware",
because a downstream
+ * `GroupPartitionsExec` consumes its key descriptor.
+ *
+ * Stamped, 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 where a child had
to be adapted, and a
+ * copy that re-derived here came back with empty `metrics` while
`doProduce` asked `metricTerm`
+ * for `numOutputRows`. A fresh copy inherits the answer instead, since
`withNewChildren` ends in
+ * `copyTagsFrom`.
+ *
+ * `UNION_OUTPUT_PARTITIONING` is read where the decision is stamped rather
than in
+ * `rawPartitioning`, so it too is fixed once the plan is prepared: `conf`
is live, and a plan
+ * must execute by the partitioning it was planned against.
+ *
+ * A read before `StampUnionDecisions` answers from the children as they are
then, and does not
+ * write, so observing an unprepared plan cannot decide anything for the
prepared one.
+ */
+ private[execution] def isPlainUnion: Boolean =
stampedDecisions.map(_.plainUnion).getOrElse {
+ !conf.getConf(SQLConf.UNION_OUTPUT_PARTITIONING) ||
Review Comment:
**Non-blocking (P2):** `EnsureRequirements` and this later stamp sample
`UNION_OUTPUT_PARTITIONING` separately. Because the shared `SQLConf` can be
updated concurrently, a true-to-false flip between them lets the parent omit an
exchange using a concrete union partitioning, then freezes plain concatenating
execution here. An aggregate above the union can consequently emit duplicate
groups. Please capture the configuration input before `EnsureRequirements` and
use that same value when deriving the post-requirements decision.
**Recommended change:** Introduce a preparation-only configuration-input
snapshot for UnionExec before each initial EnsureRequirements pass, then derive
and publish the final plain-layout decision after requirements using that same
captured input. Add deterministic concurrency coverage that changes the session
setting between those phases.
**Why this works:** A pre-requirements rule records only the relevant
configuration inputs on each unstamped UnionExec without deriving child
partitioning. isPlainUnion consults a completed decision first, then the
preparation input snapshot, and only uses the live session configuration before
preparation. EnsureRequirements therefore plans against the captured setting;
its rebuilt tagless copies inherit the input through normal tag propagation;
the following StampUnionDecisions combines that same setting with the
post-requirements child partitioning into the existing immutable decision.
Fresh unions created by later supported hooks continue to be stamped at their
post-hook barriers, and write-once completed decisions remain unchanged.
**Scope:** sql/core/src/main/scala/org/apache/spark/sql/execution,
sql/core/src/test/scala/org/apache/spark/sql/execution
**Compatibility:** Keep the preparation-time configuration lifecycle, the
partitioning decision derived after EnsureRequirements, the late-extension
barriers, and the accepted exchange-elision, bucket-scan fusion, and
skew-coalescing tradeoffs.
**Risks:** If the pre-requirements snapshot derives rawPartitioning as well
as configuration, it can freeze the pre-exchange answer and lose valid
partitioning-aware execution. If requirements rebuilds a UnionExec without
propagating the input snapshot, the two phases can diverge again. Late
extension-created unions must retain their existing post-hook stamping behavior
and must not overwrite an earlier completed decision.
**Constraints:** Reads of an unprepared plan outside physical preparation
remain non-mutating and reflect live state. Do not hold the shared SQLConf
settings monitor across physical rule execution. Preserve the existing
write-once decision, codegen-copy propagation, and documented result-preserving
optimization tradeoffs. Capture configuration inputs without capturing
child-derived partitioning before EnsureRequirements.
**Success:** EnsureRequirements and final UnionExec execution use the same
UNION_OUTPUT_PARTITIONING value even when another thread changes the session
setting between their rule invocations. A parent exchange elided from a
concrete union partitioning cannot be followed by plain concatenating execution
because of an intervening configuration update. Pre-preparation inspection does
not pin a decision for the prepared clone, and plans first prepared after an
ordinary non-concurrent configuration change still observe the new value. Late
extension barriers and rebuilt codegen-shell copies preserve the intended
completed decision without restamping 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]
