LuciferYang commented on code in PR #58419:
URL: https://github.com/apache/spark/pull/58419#discussion_r4105767975
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AdaptiveSparkPlanExec.scala:
##########
@@ -200,6 +227,14 @@ case class AdaptiveSparkPlanExec(
}
val optimized = rules.foldLeft(plan) { case (latestPlan, rule) =>
val applied = rule.apply(latestPlan)
+ if (applied ne latestPlan) {
+ // A `UnionExec` this rule just created carries no record of the confs
this execution
+ // answers from, and the next rule reads the plan before the barrier in
+ // `postStageCreationRules` stamps it. So does the
`ValidateRequirements` check below, for a
Review Comment:
Added `an injected shuffle read rule's union is recorded before validation`.
The injected rule mixes in `AQEShuffleReadRule` and puts a fresh one-child
union under the final aggregate, so `ValidateRequirements` judges the
aggregate's clustering requirement against what that union reports;
`spark.sql.unionOutputPartitioning` goes off after the wrapper read its
snapshot, so the live value says concatenate while the record says pass
through. Moving `recordUnionConf` below the validate call drops the rewrite and
the case fails with no union in the plan, and it is the only case that moves:
the other ten stay green.
--
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]
