LuciferYang commented on code in PR #58419:
URL: https://github.com/apache/spark/pull/58419#discussion_r4091552525
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AdaptiveSparkPlanExec.scala:
##########
@@ -119,10 +119,16 @@ case class AdaptiveSparkPlanExec(
conf.costEvaluatorCountLocalSortEnabled)
}
+ // Read once for this execution so that the union barriers in the lists
below, which run per
+ // re-planning round and per stage created, cannot answer from different
values. Taken from the
+ // same session conf this query's `QueryExecution.preparations` reads.
+ @transient private val unionConf =
UnionConfSnapshot(context.session.sessionState.conf)
Review Comment:
Added two cases in `SparkSessionExtensionSuite`, both flipping a conf after
the `AdaptiveSparkPlanExec` wrapper has read its snapshot and before the stage
that carries the injected union is created.
- `the barrier in AQE post stage creation stamps from the adaptive
snapshot`: an injected columnar rule adds the union,
`spark.sql.unionOutputPartitioning` goes off once `executedPlan` is built, and
the union still reports its children's `HashPartitioning`.
- `a late AQE barrier records the codegen confs the wrapper was built with`:
an injected columnar rule rebuilds a three-child union, with
`spark.sql.codegen.wholeStage.union.enabled` off in one iteration and
`maxChildren` at 2 in the other. It asserts the union still fuses, its children
inside the shell are `InputAdapter`s, and the shell copy's `numOutputRows`
reads 60.
Replacing that barrier's `unionConf` with `UnionConfSnapshot(conf)` turns
both red: `UnknownPartitioning(0)` for the first, no fused union for the second.
The query-stage-preparation barrier has no case of its own. It cannot show
the difference: `queryStagePreparationRules` is a `val`, so the barrier in it
holds the construction snapshot whichever round applies it, and the union an
injected prep rule adds is stamped before a flip can land. With that barrier
changed to read `SQLConf.get` per application, a case of the same shape stays
green. `postStageCreationRules` is the list rebuilt per stage, so it is the one
whose value could come from a later read.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -8178,7 +8182,8 @@ object SQLConf {
.internal()
.doc("When set to true, the output partitioning of UnionExec will be the
same as the " +
"input partitioning if its children have same partitioning. Otherwise,
it will be a " +
- "default partitioning.")
+ "default partitioning. The value is read once per physical
preparation, so the exchanges " +
Review Comment:
Added to the description: "One decided to concatenate keeps reporting the
default partitioning if its children come to share one afterwards."
--
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]
