AveryQi115 commented on code in PR #58924:
URL: https://github.com/apache/spark/pull/58924#discussion_r4397544768
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -6869,6 +6869,33 @@ object SQLConf {
.booleanConf
.createWithDefault(false)
+ val REPLACE_CTE_REF_WITH_CTE_REUSE =
+ buildConf("spark.sql.optimizer.replaceCTERefWithCTEReuse.enabled")
+ .internal()
+ .doc("When true, plan-reuse repartitions (those carrying a nonzero
repartition id, " +
+ "currently produced only for CTE definitions marked forceSkipInline)
are converted to " +
+ "CTEReuse nodes instead of plain Repartition nodes; ordinary retained
or user " +
+ "MATERIALIZED CTEs keep plain repartitions and are unaffected.
CTEReuse enables " +
+ "guaranteed exchange reuse in AQE by sharing a single inner
AdaptiveSparkPlanExec " +
+ "across all references.")
+ .version("4.4.0")
+ .withBindingPolicy(ConfigBindingPolicy.SESSION)
+ .booleanConf
+ .createWithDefault(true)
+
+ val FAIL_ON_CTE_REUSE_WITHOUT_AQE =
+ buildConf("spark.sql.optimizer.failOnCTEReuseWithoutAQE.enabled")
+ .internal()
+ .doc("When true, throw an error if guaranteed CTE/subplan shuffle reuse
fails with AQE " +
+ "off, i.e. two or more shuffles tagged with the same cteId survive
after " +
+ "ReuseExchangeAndSubquery (they were not deduplicated into a
ReusedExchangeExec). " +
+ "When false, only log the plan, error, and cteIds. Enabled by default
in tests to " +
+ "catch reuse regressions.")
+ .version("4.4.0")
+ .withBindingPolicy(ConfigBindingPolicy.SESSION)
+ .booleanConf
+ .createWithDefault(false)
Review Comment:
Thanks for the detailed suggestion — I'd like to keep the current verifier
for this PR; I think it's actually the safer choice here. A rule that
force-rewrites every reference onto one owner by `cteId` is risky exactly when
the references' canonical bodies aren't identical: it would wire
structurally-different bodies onto a single materialization and could silently
return wrong rows. The current path only seals when the bodies are canonically
equal (via the normal `ReuseExchangeAndSubquery` dedup); if duplicates survive
it surfaces that rather than forcing a merge — throwing when
`failOnCTEReuseWithoutAQE` is on (so tests catch regressions) and logging the
plan + cteIds and continuing when it's off, which preserves the existing
behavior. Today the only producer of `forceSkipInline` CTEs is the DSv2 MERGE
path, and this matches that use case. I'd suggest revisiting a stronger
cteId-keyed owner (with canonical-body validation) once more producers rely on
guaranteed reuse.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala:
##########
@@ -1196,6 +1196,34 @@ abstract class SparkStrategies extends
QueryPlanner[SparkPlan] {
f, key, lObj, rObj, lGroup, rGroup, lAttr, rAttr, lOrder, rOrder,
oAttr,
planLater(left), planLater(right)) :: Nil
+ case r: logical.CTEReuseRelation =>
+ // Plan the shared subplan eagerly rather than using planLater, because
+ // CTEReuseExchange is a LeafExecNode whose subplan is not visible via
+ // children. QueryPlanner.collectPlaceholders would never find a
PlanLater
+ // inside the subplan, leaving it unresolved.
+ //
+ // sharedSubplan is a RepartitionByExpression wrapping the CTE body.
Planning it produces a
+ // plain ShuffleExchangeExec; we re-stamp its origin as
LOCAL_SHUFFLE_FOR_CTE(cteId) so the
+ // shuffle is recognized as a reuse boundary
(`isCreatedForSubplanReuse`) by
+ // EnsureRequirements, the AQE-off unwrap path, and VerifyCTEReuse. We
use this shuffle
+ // directly instead of adding a separate exchange -- the Repartition
serves as the logical
+ // plan for the inner AQE, enabling correct re-optimization (broadcast
stages are findable
+ // as descendants, and re-planning preserves the root exchange).
+ val physicalSubplan = SparkStrategies.this.plan(r.sharedSubplan).next()
+ val shuffle = physicalSubplan match {
+ case s: ShuffleExchangeExec => s.copy(shuffleOrigin =
LOCAL_SHUFFLE_FOR_CTE(r.cteId))
Review Comment:
Good point on the repeated planning cost — I'd prefer to leave this as a
follow-up rather than fold it into this PR. The current behavior matches how
exchange/subquery reuse already works: each reference is planned independently
and shared afterward, not memoized during planning. A per-cteId planning memo
would also have to account for the AQE-off unwrap path — once
`CTEReuseExchange` is unwrapped, the shared subplan can need its exprIds
re-assigned per reference, so a single immutable physical template isn't a
drop-in. It's a worthwhile optimization, but orthogonal to the correctness of
this change, so I'd rather do it separately.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -6869,6 +6869,33 @@ object SQLConf {
.booleanConf
.createWithDefault(false)
+ val REPLACE_CTE_REF_WITH_CTE_REUSE =
+ buildConf("spark.sql.optimizer.replaceCTERefWithCTEReuse.enabled")
+ .internal()
+ .doc("When true, plan-reuse repartitions (those carrying a nonzero
repartition id, " +
+ "currently produced only for CTE definitions marked forceSkipInline)
are converted to " +
+ "CTEReuse nodes instead of plain Repartition nodes; ordinary retained
or user " +
+ "MATERIALIZED CTEs keep plain repartitions and are unaffected.
CTEReuse enables " +
+ "guaranteed exchange reuse in AQE by sharing a single inner
AdaptiveSparkPlanExec " +
+ "across all references.")
+ .version("4.4.0")
+ .withBindingPolicy(ConfigBindingPolicy.SESSION)
+ .booleanConf
+ .createWithDefault(true)
Review Comment:
The default-on is intentional — Julek asked for it so the new lowering gets
broad coverage. It only changes plans for `forceSkipInline` CTEs (today just
the DSv2 MERGE path), so other plans are unaffected even with it on. You're
right the description is inconsistent, though — I'll update the PR text so the
"defaults off" wording matches the actual default, and drop the
`LocalPartition`/Photon references that aren't in Apache Spark.
--
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]
