LuciferYang commented on code in PR #58419:
URL: https://github.com/apache/spark/pull/58419#discussion_r3981177751
##########
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:
You are right that it was stronger than the code. Narrowed it: what must not
be reachable under `decisionLock` is the AQE final-plan lock, since
`CoalesceShufflePartitions` reads `isPlainUnion` while holding it. The comment
now also says that child `lazy val`s such as
`AQEShuffleReadExec.outputPartitioning` do take their own instance monitor, and
why that is harmless here.
##########
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:
Fixed. It now says `isPlainUnion` satisfies the partitioning gate on the
codegen path rather than the whole of it, and that
`supportCodegenFailureReason` still applies its other 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]
