cloud-fan commented on code in PR #58419:
URL: https://github.com/apache/spark/pull/58419#discussion_r3979092650
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala:
##########
@@ -1020,13 +1023,57 @@ 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]
+ // Serializes the two latches below so concurrent first readers agree on one
answer. Private to
+ // this node, so the only lock taken under it is a nested union's own
`decisionLock`, always a
Review Comment:
**Nit (P3):** This lock-order claim is stronger than the implementation.
Both derivations call child `outputPartitioning`, `supportsColumnar`, or
`output` while holding `decisionLock`; ordinary children such as
`BroadcastHashJoinExec`, `AQEShuffleReadExec`, and `FileSourceScanExec`
implement some of those as lazy vals, so their first read synchronizes on the
child. Please state the narrower lock that must be avoided here, especially the
AQE final-plan lock, rather than claiming that only a descendant union lock can
be acquired.
##########
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:
**Blocking (P1):** `CODEGEN_FAILURE_REASON` is derived from the current
children, but storing it in a `TreeNodeTag` lets generic `withNewChildren`
copies inherit it after those children change. Under AQE, `SparkPlanInfo` can
force `metrics` before a supported query-stage or columnar rule replaces the
children, and the final `CollapseCodegenStages` pass then trusts the cached
`None`. A newly added `SparkPartitionID` can read the global UnionRDD index
instead of the child-local index; a dual-mode columnar replacement can instead
hit the row-codegen assertion. Please keep these decisions in private
per-instance `UnionExec` state and transfer an immutable snapshot explicitly
only through a dedicated copy path used for the `InputAdapter` rebuild.
Ordinary child rewrites should get fresh state, and `UnionCodegenSuite` should
cover a post-metrics rewrite.
**Recommended change:** Replace the UnionExec TreeNodeTag decisions with
private per-instance state, and add a UnionExec-specific codegen copy path that
transfers an immutable decision snapshot only for CollapseCodegenStages'
intended InputAdapter rebuild.
**Why this works:** An early metrics read stores an allow decision in
generic TreeNode metadata; withNewChildren can then copy it after an extension
changes the children, so the final codegen pass skips gates that the
replacement topology fails.
**Scope:** UnionExec decision state, CollapseCodegenStages' InputAdapter
rebuild, and focused AQE extension coverage in UnionCodegenSuite.
**Compatibility:** Keep the intended per-plan configuration stability and
preserve both decisions across the specific InputAdapter rebuild; ordinary
child rewrites must recompute from their actual topology.
**Risks:** Failing to transfer the snapshot on the intended InputAdapter
copy can reintroduce the missing numOutputRows crash this PR fixes. Reusing the
snapshot on any other child-changing path can leave another stale eligibility
or partitioning decision.
**Constraints:** Do not add decision state to UnionExec's case-class
parameters or reintroduce live-conf divergence between planning, metrics, and
execution. Preserve partitioning-aware, multi-RDD, partition-index-dependent,
and columnar fallbacks.
**Success:** Generic child rewrites receive fresh decision state, only the
intended codegen copy receives the frozen snapshot, the existing cache and
conf-flip regressions remain fixed, and a post-metrics query-stage or columnar
rewrite neither changes partition-index values nor hits the row-codegen
assertion.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala:
##########
@@ -1020,13 +1023,57 @@ 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]
+ // Serializes the two latches below so concurrent first readers agree on one
answer. Private to
+ // this node, so the only lock taken under it is a nested union's own
`decisionLock`, always a
+ // descendant's. It has to stay that way: nothing either derivation walks
may take a lock, or it
+ // would invert `CoalesceShufflePartitions`, which reads `isPlainUnion`
while holding the AQE
+ // lock. That surface is the children's `outputPartitioning`,
`supportsColumnar` and `output`;
+ // `InMemoryTableScanExec` qualifies only because it reads
`adaptive.executedPlan`, a volatile
+ // read, not `finalPhysicalPlan`, which is `lock.synchronized`. Driver-only,
hence `@transient`.
+ @transient private val decisionLock = new Object()
+
+ /**
+ * True when this union behaves as a plain concatenation, so
`unionedInputRDD` matches the
+ * semantics of `sparkContext.union(...)` in `unionRDDs`, and the codegen
path applies. A
Review Comment:
**Nit (P3):** `isPlainUnion` establishes only the partitioning prerequisite,
so saying that “the codegen path applies” is too strong.
`deriveCodegenFailureReason` can still reject a plain union for nested unions,
multi-RDD children, partition-index-dependent expressions, the child limit,
columnar support, or a type mismatch. Please describe this as satisfying the
partitioning gate while remaining subject to the other codegen checks.
--
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]
