You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "xumingming (via GitHub)" <gi...@apache.org> on 2026/08/25 07:26:14 UTC
[PR] [SPARK-58006][SQL][FOLLOWUP] Reject escaping outer refs in forceSkipInline defs [spark]
xumingming opened a new pull request, #58269:
URL: https://github.com/apache/spark/pull/58269
### What changes were proposed in this pull request?
`InlineCTE.validateNoOuterReferencesAcrossCTEBoundary` currently throws `INTERNAL_ERROR` whenever a `forceSkipInline` CTE definition's body contains any `OuterReference` or correlated subquery (with non-empty `outerScopeAttrs`), even when that reference resolves to an operator *inside* the definition body. Such an internal correlation is self-contained and safe to materialize, because after materialization the def body itself still provides the operator the reference resolves against.
This PR makes the validator reject a reference only when it genuinely escapes the CTE boundary: it collects every attribute bound anywhere in the definition (including inside nested subquery plans, skipping unresolved outputs), and throws only when an `OuterReference` / `OuterScopeReference`'s exprId resolves to none of those bound exprIds.
### Why are the changes needed?
SPARK-58006 added `forceSkipInline` to guarantee a CTE is materialized rather than duplicated. The follow-up validation was too strict: it flagged *any* outer reference in the def, including self-contained ones. Concretely, a view whose correlated subquery references a column produced within its own body could not be materialized as a `forceSkipInline` CTE — analysis failed or the conversion was silently skipped — even though such a correlation has a valid resolution home inside the materialized def.
### Does this PR introduce _any_ user-facing change?
No. It affects only internal optimizer behavior for the newly added `forceSkipInline` path: instead of raising `INTERNAL_ERROR` for self-contained correlations, the CTE is now materialized correctly. Truly escaping references still raise the same `INTERNAL_ERROR`.
### How was this patch tested?
Updated `InlineCTESuite` under `sql/catalyst`:
- regression: an internal `OuterReference` (correlated `EXISTS`) inside a `forceSkipInline` def is materialized (no throw);
- regression: an internal `OuterScopeReference` (nested correlated scalar subquery, two scopes out) is materialized (no throw);
- negative: an outer reference inside a nested subquery plan that escapes the def still fails with `INTERNAL_ERROR`.
Ran:
```
build/sbt "catalyst/Test/compile"
build/sbt "catalyst/Test/scalastyle"
build/sbt "catalyst/testOnly org.apache.spark.sql.catalyst.optimizer.InlineCTESuite"
```
All 10 tests in the suite pass; scalastyle reports 0 errors.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Pi
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58006][SQL][FOLLOWUP] Reject escaping outer refs in forceSkipInline defs [spark]
Posted by "AveryQi115 (via GitHub)" <gi...@apache.org>.
AveryQi115 commented on code in PR #58269:
URL: https://github.com/apache/spark/pull/58269#discussion_r3867969389
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InlineCTE.scala:
##########
@@ -89,29 +89,45 @@ case class InlineCTE(
}
private def validateNoOuterReferencesAcrossCTEBoundary(cteDef: CTERelationDef): Unit = {
- val outerRefs = cteDef.child.flatMap(
- _.expressions.flatMap(_.collect { case o: OuterReference => o }))
- if (outerRefs.nonEmpty) {
+ // Only an outer reference that actually escapes the CTE definition is invalid. An outer
+ // reference that points to an attribute produced by an operator inside the definition
+ // body (e.g. a correlated subquery whose correlated column lives in the body) does not
+ // escape, so the definition is self-contained and safe to materialize. Walk the whole
+ // definition (main tree plus nested subquery plans) once, collect every attribute bound
+ // anywhere in the definition, and reject only references that resolve to none of them.
+ val allNodes = cteDef.child.collectWithSubqueries { case n: LogicalPlan => n }
+ val boundExprIds = allNodes.iterator
Review Comment:
Using exprId to match the outer references and outer scope references might not be very reliable, especially for dataframe cases, which can contain duplicate exprIds without running single pass analyzer.
I use
```
val outerRefs = cteDef.child.flatMap(
_.expressions.flatMap(_.collect { case o: OuterReference => o }))
```
because this is the previous behavior check before `forceSkipInline` is added. If you want this validation to skip self-contained correlations, we can change the validation to be:
1. check if there are direct outer references exists without traversing to the subquery.plan or subquery child. If a direct outer reference exists in cteDef body, then it must refer to attributes outside the cteDef.
```
val outerRefs = cteDef.child.exists(_.expressions.exists(SubExprUtils.containsOuter))
```
2. check if there are subquery expressions has `OuterScopeReference` in cteDef body. This means that the outer references within the subquery plan is not from the direct parent scope (which is the cteDef itself) of the subquery.
```
val outerScopeSubqueries = cteDef.child.flatMap(
_.expressions.flatMap(_.collect {
case s: SubqueryExpression if s.outerScopeAttrs.nonEmpty => s
}))
```
I'm also okay with the current approach. But we'd better add some dataframe testcases.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58006][SQL][FOLLOWUP] Reject escaping outer refs in forceSkipInline defs [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #58269:
URL: https://github.com/apache/spark/pull/58269#discussion_r3935971993
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InlineCTE.scala:
##########
@@ -89,29 +89,55 @@ case class InlineCTE(
}
private def validateNoOuterReferencesAcrossCTEBoundary(cteDef: CTERelationDef): Unit = {
- val outerRefs = cteDef.child.flatMap(
- _.expressions.flatMap(_.collect { case o: OuterReference => o }))
- if (outerRefs.nonEmpty) {
+ // Only an outer reference that actually escapes the CTE definition is invalid. An outer
+ // reference that points to an attribute produced by an operator inside the definition
+ // body (e.g. a correlated subquery whose correlated column lives in the body) does not
+ // escape, so the definition is self-contained and safe to materialize. Walk the whole
+ // definition (main tree plus nested subquery plans) once, collect every attribute bound
+ // anywhere in the definition, and reject only references that resolve to none of them.
+ //
+ // Matching by exprId is exact for plans produced by a single analysis pass: every
+ // `OuterReference` wraps the very attribute it binds to, exprIds are allocated fresh
+ // per resolution, and `DeduplicateRelations` rewrites duplicated relation instances
+ // (self-join, union, etc.) with fresh exprIds before the optimizer runs. The known
+ // residual gap is a plan stitched together from already-analyzed plans (e.g. a producer
+ // embedding the same analyzed dataset twice): such a tree can carry the same exprId in
+ // multiple places, so an escaping reference could match a duplicated exprId and slip
+ // through. Such a query is ill-formed and fails later during planning or execution
+ // anyway; this check is defense-in-depth that fails fast with a clear error otherwise.
+ val allNodes = cteDef.child.collectWithSubqueries { case n: LogicalPlan => n }
+ val boundExprIds = allNodes.iterator
+ .flatMap(_.output.filter(_.resolved).map(_.exprId))
Review Comment:
Thanks for tracing the analyzer contract here. I agree that both masking shapes require bypassing analysis: projected outer references are aliased, conflicting embedded plans are deduplicated before optimization, and the residual stitched-plan limitation is documented. This resolves my concern.
<!-- SPARK_DEV_REVIEW_REPLY {"feedback_id":"inline:3933711073","thread_id":"inline:3933711073","verdict_sha256":"e5c35005451a40f4ace86649d78bab420032ce4855b51821ff2f8bc0cfb5dbf2"} -->
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58006][SQL][FOLLOWUP] Reject escaping outer refs in forceSkipInline defs [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #58269:
URL: https://github.com/apache/spark/pull/58269#discussion_r3862860872
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InlineCTE.scala:
##########
@@ -89,30 +89,45 @@ case class InlineCTE(
}
private def validateNoOuterReferencesAcrossCTEBoundary(cteDef: CTERelationDef): Unit = {
- val outerRefs = cteDef.child.flatMap(
- _.expressions.flatMap(_.collect { case o: OuterReference => o }))
- if (outerRefs.nonEmpty) {
- throw SparkException.internalError(
- "A force-materialized CTE cannot carry an outer reference across its boundary, but " +
- s"found outer reference '${outerRefs.head.name}' in the CTE definition " +
- s"(cteId=${cteDef.id}).")
- }
+ // Only an outer reference that actually escapes the CTE definition is invalid. An outer
+ // reference that points to an attribute produced by an operator inside the definition
+ // body (e.g. a correlated subquery whose correlated column lives in the body) is
+ // self-contained and safe to materialize. So walk the whole definition (main tree plus
+ // nested subquery plans) once, collect every attribute bound anywhere in the definition,
+ // and reject only references that resolve to none of them.
+ val allNodes = cteDef.child.collectWithSubqueries { case n: LogicalPlan => n }
+ val boundExprIds = allNodes.flatMap(_.output.filter(_.resolved).map(_.exprId)).toSet
- val outerScopeSubqueries = cteDef.child.flatMap(
- _.expressions.flatMap(_.collect {
- case s: SubqueryExpression if s.outerScopeAttrs.nonEmpty => s
+ // Locate either a direct `OuterReference` or a correlated subquery whose outer-scope
+ // attributes escape the def -- i.e. resolve to none of `boundExprIds`.
+ val escapingOuterRef = allNodes.iterator
+ .flatMap(_.expressions.iterator.flatMap(_.collect {
+ case o: OuterReference if !boundExprIds.contains(o.exprId) => o
}))
- if (outerScopeSubqueries.nonEmpty) {
- // `outerScopeAttrs` are expressions that contain an `OuterScopeReference` and may be
- // compound (e.g. `Add(OuterScopeReference(a), Literal(1))`), so collect the reference node
- // rather than assuming the attribute itself is one.
- val outerScopeRef = outerScopeSubqueries.head.outerScopeAttrs
- .flatMap(_.collect { case r: OuterScopeReference => r }).head
- throw SparkException.internalError(
- "A force-materialized CTE cannot carry an outer reference across its boundary, but " +
- "found a subquery with outer-scope reference " +
- s"'${outerScopeRef.name}' in the CTE definition " +
- s"(cteId=${cteDef.id}).")
+ .toSeq
Review Comment:
**Non-blocking:**
Stop at the first escaping reference instead of materializing every match. Both this pipeline and the one ending at line 115 build a `Seq` only to take `headOption`, so call `nextOption()` directly on each iterator. Build `boundExprIds` from `allNodes.iterator` as well to avoid its intermediate `Seq`.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58006][SQL][FOLLOWUP] Reject escaping outer refs in forceSkipInline defs [spark]
Posted by "xumingming (via GitHub) via reviews" <re...@spark.apache.org>.
xumingming commented on PR #58269:
URL: https://github.com/apache/spark/pull/58269#issuecomment-6649165548
@cloud-fan Shall we merge this one?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58006][SQL][FOLLOWUP] Reject escaping outer refs in forceSkipInline defs [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #58269:
URL: https://github.com/apache/spark/pull/58269#discussion_r3933711073
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InlineCTE.scala:
##########
@@ -89,29 +89,55 @@ case class InlineCTE(
}
private def validateNoOuterReferencesAcrossCTEBoundary(cteDef: CTERelationDef): Unit = {
- val outerRefs = cteDef.child.flatMap(
- _.expressions.flatMap(_.collect { case o: OuterReference => o }))
- if (outerRefs.nonEmpty) {
+ // Only an outer reference that actually escapes the CTE definition is invalid. An outer
+ // reference that points to an attribute produced by an operator inside the definition
+ // body (e.g. a correlated subquery whose correlated column lives in the body) does not
+ // escape, so the definition is self-contained and safe to materialize. Walk the whole
+ // definition (main tree plus nested subquery plans) once, collect every attribute bound
+ // anywhere in the definition, and reject only references that resolve to none of them.
+ //
+ // Matching by exprId is exact for plans produced by a single analysis pass: every
+ // `OuterReference` wraps the very attribute it binds to, exprIds are allocated fresh
+ // per resolution, and `DeduplicateRelations` rewrites duplicated relation instances
+ // (self-join, union, etc.) with fresh exprIds before the optimizer runs. The known
+ // residual gap is a plan stitched together from already-analyzed plans (e.g. a producer
+ // embedding the same analyzed dataset twice): such a tree can carry the same exprId in
+ // multiple places, so an escaping reference could match a duplicated exprId and slip
+ // through. Such a query is ill-formed and fails later during planning or execution
+ // anyway; this check is defense-in-depth that fails fast with a clear error otherwise.
+ val allNodes = cteDef.child.collectWithSubqueries { case n: LogicalPlan => n }
+ val boundExprIds = allNodes.iterator
+ .flatMap(_.output.filter(_.resolved).map(_.exprId))
Review Comment:
**Blocking (P1):** `boundExprIds` includes every plan node's output, but `Project.output` is derived from `projectList.toAttribute`, and `OuterReference.toAttribute` preserves the wrapped attribute's `exprId`. A nested project can therefore insert an escaping reference's own ID into this set and make the predicate accept it; the same definition-wide union also lets an unrelated sibling with a duplicate `exprId` mask the escape. The CTE is then materialized despite depending on an operator outside its boundary, so the intended `INTERNAL_ERROR` is deferred to later optimization or planning.
**Recommended change:** Make this validation scope-aware: only genuine producers in an outer reference's enclosing scope inside the CTE may bind it, and add regressions for both a Project that projects the escaping OuterReference and an unrelated sibling that reuses the same exprId.
**Why this works:** Walk the CTE definition while carrying lexical binder scopes from genuine operator inputs and children, then validate each OuterReference or OuterScopeReference against the appropriate enclosing scope instead of a definition-wide union of plan outputs.
**Scope:** InlineCTE.validateNoOuterReferencesAcrossCTEBoundary and focused Catalyst and SQL CTE-inline regressions covering the two masking shapes.
**Compatibility:** Preserve acceptance of genuinely self-contained correlations and the existing INTERNAL_ERROR contract for references whose producer is outside the CTE, without changing unrelated InlineCTE behavior.
**Risks:** An incorrect scope walk could reject a valid internal correlation or mis-handle nested subquery scope. Continuing to use exprId membership without lexical ownership would remain vulnerable to reused analyzed fragments.
**Constraints:** Keep unresolved outputs excluded as the current implementation does. Validate both OuterReference and OuterScopeReference across nested subqueries. Cover one direct Project self-masking case and one real-analyzer sibling duplicate-exprId case.
**Success:** Both escaping-reference shapes fail at the CTE boundary with INTERNAL_ERROR, while the existing self-contained OuterReference and OuterScopeReference cases still materialize successfully.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58006][SQL][FOLLOWUP] Reject escaping outer refs in forceSkipInline defs [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #58269:
URL: https://github.com/apache/spark/pull/58269#discussion_r3853304493
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InlineCTE.scala:
##########
@@ -89,30 +89,48 @@ case class InlineCTE(
}
private def validateNoOuterReferencesAcrossCTEBoundary(cteDef: CTERelationDef): Unit = {
- val outerRefs = cteDef.child.flatMap(
- _.expressions.flatMap(_.collect { case o: OuterReference => o }))
- if (outerRefs.nonEmpty) {
- throw SparkException.internalError(
- "A force-materialized CTE cannot carry an outer reference across its boundary, but " +
- s"found outer reference '${outerRefs.head.name}' in the CTE definition " +
- s"(cteId=${cteDef.id}).")
- }
+ // Only an outer reference that actually escapes the CTE definition is invalid. An outer
+ // reference that resolves to an operator inside the definition body (e.g. a correlated
+ // subquery whose correlated column lives in the body) is self-contained and safe to
+ // materialize. So collect every attribute bound anywhere in the definition (including in
+ // nested subquery plans) and reject only references that resolve to none of them.
+ val boundExprIds = cteDef.child
+ .collectWithSubqueries { case n: LogicalPlan => n }
+ .flatMap(_.output.filter(_.resolved).map(_.exprId))
+ .toSet
- val outerScopeSubqueries = cteDef.child.flatMap(
- _.expressions.flatMap(_.collect {
- case s: SubqueryExpression if s.outerScopeAttrs.nonEmpty => s
+ // Walk the whole definition (main tree plus nested subquery plans). Locate either a direct
+ // `OuterReference` or a correlated subquery whose outer-scope attributes escape the def --
+ // i.e. resolve to none of `boundExprIds`.
+ val allNodes = cteDef.child.collectWithSubqueries { case n: LogicalPlan => n }
Review Comment:
**Non-blocking:**
Reuse the first `collectWithSubqueries` result here and derive `boundExprIds` from it. This helper materializes the full CTE and nested-subquery node sequence, so the second call repeats a plan-size traversal for every force-materialized definition.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InlineCTE.scala:
##########
@@ -89,30 +89,48 @@ case class InlineCTE(
}
private def validateNoOuterReferencesAcrossCTEBoundary(cteDef: CTERelationDef): Unit = {
- val outerRefs = cteDef.child.flatMap(
- _.expressions.flatMap(_.collect { case o: OuterReference => o }))
- if (outerRefs.nonEmpty) {
- throw SparkException.internalError(
- "A force-materialized CTE cannot carry an outer reference across its boundary, but " +
- s"found outer reference '${outerRefs.head.name}' in the CTE definition " +
- s"(cteId=${cteDef.id}).")
- }
+ // Only an outer reference that actually escapes the CTE definition is invalid. An outer
+ // reference that resolves to an operator inside the definition body (e.g. a correlated
Review Comment:
**Nit:**
Please say that the outer reference points to an attribute produced by an operator inside the definition. Catalyst outer references resolve to attributes, and this validator compares their ExprIds.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/InlineCTESuite.scala:
##########
@@ -109,6 +109,74 @@ class InlineCTESuite extends PlanTest {
"found a subquery with outer-scope reference"))
}
+ test("SPARK-58006: forceSkipInline CTE with an internal outer reference is materialized") {
+ // Corresponds to: WITH t AS (SELECT a FROM r WHERE EXISTS (SELECT 1 FROM s WHERE s.k = t.a))
Review Comment:
**Nit:**
Please change `t.a` to `r.a` here and at lines 134 and 136. `t` is not in scope inside its own CTE definition; both test plans correlate to `relation.output.head` from `r`.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58006][SQL][FOLLOWUP] Reject escaping outer refs in forceSkipInline defs [spark]
Posted by "uros-b (via GitHub)" <gi...@apache.org>.
uros-b commented on PR #58269:
URL: https://github.com/apache/spark/pull/58269#issuecomment-5554211550
Thank you @xumingming and @cloud-fan!!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58006][SQL][FOLLOWUP] Reject escaping outer refs in forceSkipInline defs [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #58269:
URL: https://github.com/apache/spark/pull/58269#discussion_r4103490490
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/InlineCTESuite.scala:
##########
@@ -109,6 +109,108 @@ class InlineCTESuite extends PlanTest {
"found a subquery with outer-scope reference"))
}
+ test("SPARK-58006: forceSkipInline CTE with an internal outer reference is materialized") {
+ // Corresponds to: WITH t AS (SELECT a FROM r WHERE EXISTS (SELECT 1 FROM s WHERE s.k = r.a))
+ // SELECT a FROM t
+ // The outer reference to `a` resolves inside the CTE def body, so it must be tolerated.
+ val relation = TestRelation(Seq($"a".int))
+ val subRel = TestRelation(Seq($"k".int))
+ val exists = Exists(
+ subRel.where(subRel.output.head === OuterReference(relation.output.head)),
+ outerAttrs = Seq(relation.output.head))
+ val cteChild = relation.where(exists)
+ val cteDef = CTERelationDef(cteChild, forceSkipInline = true)
+ val cteRef = CTERelationRef(cteDef.id, cteDef.resolved, cteDef.output, cteDef.isStreaming)
+ val plan = WithCTE(cteRef.select(relation.output.head), Seq(cteDef))
+ assert(plan.resolved)
+ val optimized = Optimize.execute(plan)
+ // The internal correlation is safe: the def stays materialized and validation passes.
+ assert(optimized.collectFirst { case _: WithCTE => true }.isDefined,
+ "CTE with an internal outer reference should be materialized, not rejected")
+ }
+
+ test("SPARK-58006: forceSkipInline CTE with an internal outer-scope subquery is materialized") {
+ // Corresponds to: WITH t AS (
+ // SELECT (SELECT v FROM s2 WHERE s2.v < (SELECT c FROM s1 WHERE s1.k = r.a)) AS s
+ // FROM r) SELECT s FROM t
+ // The innermost reference to `r.a` is two scopes out but resolves inside the def.
+ val relation = TestRelation(Seq($"a".int))
+ val s1 = TestRelation(Seq($"k".int, $"c".int))
+ val s2 = TestRelation(Seq($"v".int))
+ val deep = ScalarSubquery(
+ s1.where(s1.output.head === OuterReference(relation.output.head)).select(s1.output(1)),
+ outerAttrs = Seq(OuterScopeReference(relation.output.head)))
+ val shallow = ScalarSubquery(
+ s2.where(s2.output.head < deep),
+ outerAttrs = Seq(relation.output.head))
+ val cteChild = relation.select(shallow.as("s"))
+ val cteDef = CTERelationDef(cteChild, forceSkipInline = true)
+ val cteRef = CTERelationRef(cteDef.id, cteDef.resolved, cteDef.output, cteDef.isStreaming)
+ val plan = WithCTE(cteRef.select(cteDef.output.head), Seq(cteDef))
+ assert(plan.resolved)
+ val optimized = Optimize.execute(plan)
+ assert(optimized.collectFirst { case _: WithCTE => true }.isDefined,
+ "CTE with an internal outer-scope subquery should be materialized, not rejected")
+ }
+
+ test("SPARK-58006: forceSkipInline CTE with an outer reference inside a subquery plan fails") {
+ // Corresponds to: WITH t AS (SELECT (SELECT c FROM s WHERE s.k = m.a) AS x FROM r)
+ // SELECT x FROM t, m
+ // `m.a` is produced outside the CTE, so the reference escapes the boundary and must fail.
+ val relation = TestRelation(Seq($"c".int))
+ val outerA = $"m.a".int
+ val subquery = ScalarSubquery(
+ relation.where(relation.output.head === OuterReference(outerA)),
+ outerAttrs = Seq(outerA))
+ val cteChild = TestRelation(Seq($"b".int)).select(subquery.as("s"))
+ val cteDef = CTERelationDef(cteChild, forceSkipInline = true)
+ val cteRef = CTERelationRef(cteDef.id, cteDef.resolved, cteDef.output, cteDef.isStreaming)
+ val m = TestRelation(Seq(outerA))
+ val main = cteRef.select(cteDef.output.head).join(m, Cross)
+ val plan = WithCTE(main, Seq(cteDef))
+ assert(plan.resolved)
+ val e = intercept[SparkException] {
+ Optimize.execute(plan)
+ }
+ assert(e.getCondition == "INTERNAL_ERROR")
+ assert(e.getMessage.contains(
+ "A force-materialized CTE cannot carry an outer reference across its boundary"))
+ }
+
+ test("SPARK-58006: forceSkipInline CTE with an outer-scope reference " +
+ "inside a nested subquery fails") {
+ // Corresponds to: WITH t AS (
+ // SELECT (SELECT v FROM s2 WHERE v < (SELECT c FROM s1 WHERE s1.k = m.a)) AS s
+ // FROM r) SELECT s FROM t, m
+ // `m.a` is two scopes out of the deepest subquery and produced outside the def, so it
Review Comment:
**Nit (P3):** In this shape, `m` belongs to the main query outside the CTE body and both nested scalar-subquery queries, so `m.a` is three scopes out of the deepest query; the neighboring internal `r.a` case is the two-scope example. Please correct the count and replace `nested-wide` below with wording that plainly says the outer-scope scan covers nested subquery plans.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/cteOperators.scala:
##########
@@ -119,6 +119,12 @@ case class UnionLoopRef(
* regardless of determinism or reference count. This lets a producer
* force the CTE to be materialized instead of duplicated, e.g. when the
* CTE wraps a non-deterministic source that must be evaluated exactly once.
+ * A materialized definition must be self-contained: it cannot carry an
+ * outer reference across its boundary, because after materialization
+ * there is no surrounding operator to resolve the reference against.
+ * [[InlineCTE]] validates this and rejects escaping references with an
Review Comment:
**Nit (P3):** This states an unconditional rejection guarantee, but InlineCTE's validator is exact only for plans produced by a single analysis pass. Its own comment documents that duplicate ExprIds in a post-analysis stitched tree can let an escaping reference slip through and possibly bind to the wrong input. Please qualify this contract to the supported single-analysis-pass scope so producers do not rely on fail-fast behavior the validator cannot guarantee.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58006][SQL][FOLLOWUP] Reject escaping outer refs in forceSkipInline defs [spark]
Posted by "xumingming (via GitHub)" <gi...@apache.org>.
xumingming commented on code in PR #58269:
URL: https://github.com/apache/spark/pull/58269#discussion_r3935292649
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InlineCTE.scala:
##########
@@ -89,29 +89,55 @@ case class InlineCTE(
}
private def validateNoOuterReferencesAcrossCTEBoundary(cteDef: CTERelationDef): Unit = {
- val outerRefs = cteDef.child.flatMap(
- _.expressions.flatMap(_.collect { case o: OuterReference => o }))
- if (outerRefs.nonEmpty) {
+ // Only an outer reference that actually escapes the CTE definition is invalid. An outer
+ // reference that points to an attribute produced by an operator inside the definition
+ // body (e.g. a correlated subquery whose correlated column lives in the body) does not
+ // escape, so the definition is self-contained and safe to materialize. Walk the whole
+ // definition (main tree plus nested subquery plans) once, collect every attribute bound
+ // anywhere in the definition, and reject only references that resolve to none of them.
+ //
+ // Matching by exprId is exact for plans produced by a single analysis pass: every
+ // `OuterReference` wraps the very attribute it binds to, exprIds are allocated fresh
+ // per resolution, and `DeduplicateRelations` rewrites duplicated relation instances
+ // (self-join, union, etc.) with fresh exprIds before the optimizer runs. The known
+ // residual gap is a plan stitched together from already-analyzed plans (e.g. a producer
+ // embedding the same analyzed dataset twice): such a tree can carry the same exprId in
+ // multiple places, so an escaping reference could match a duplicated exprId and slip
+ // through. Such a query is ill-formed and fails later during planning or execution
+ // anyway; this check is defense-in-depth that fails fast with a clear error otherwise.
+ val allNodes = cteDef.child.collectWithSubqueries { case n: LogicalPlan => n }
+ val boundExprIds = allNodes.iterator
+ .flatMap(_.output.filter(_.resolved).map(_.exprId))
Review Comment:
@cloud-fan Thanks for the review.
I'd like to first pin down the threat model we're signing up for, because I believe the two masking shapes it describes cannot occur in the plans InlineCTE actually sees, and I want to make sure I'm not missing an
in-contract path.
1. Project self-masking. This requires a bare (un-aliased) OuterReference in a node's projectList, because OuterReference.toAttribute passes through the wrapped exprId. But the analyzer never leaves a bare OuterReference in a projectList or
aggregate expression: aliasIfOuterReference (see the doc comment at ColumnResolutionHelper.scala) wraps it in an Alias precisely because "an attribute's ExprId leaks through the operator's output", and Alias.toAttribute carries the
alias's own exprId, so nothing leaks. Ordering also helps: InlineCTE runs in the "Inline CTE" batch, before "Pullup Correlated Expressions", so no optimizer-created OuterReference-carrying Projects exist at validation time either. So a
Project cannot "mirror" an escaping reference's exprId in any plan produced by the analyzer.
2. Sibling duplicate-exprId masking. Within a single post-analysis plan, duplicate exprIds can't occur: every resolution allocates a fresh exprId, and whenever already-analyzed plans are embedded into a new query (temp views created from
DataFrames, cached plans re-injected by CacheManager, LogicalRDD from checkpoint(), DSv2 command rewrites), the embedding happens at the input of an analysis pass and DeduplicateRelations rewrites duplicated relation instances with fresh
exprIds before the optimizer runs. The PR's code comment documents exactly this residual case: plans stitched together after analysis, bypassing it.
Given that, both masking shapes require a plan that bypassed analysis — outside the contract the optimizer operates under. Notably, real analyzed SQL can't produce an escaping reference inside a CTE def either (a def body can't see the main
query's relations — the negative tests in this PR are hand-built or stitched for exactly that reason), so this validator is defense-in-depth against out-of-contract plans either way. The question is how airtight it must be against
plan-surgery shapes:
- If the standard is "fail-fast guard for well-formed (analyzed) plans", I'd argue the current design is sufficient, with the documented residual gap.
- If the standard is "must catch even hand-stitched trees that bypass analysis", I'm happy to implement the scope-aware validation (walk the def carrying the enclosing scope downward, check each OuterReference/OuterScopeReference against its
owning scope instead of a definition-wide union), which subsumes both masking vectors, plus regressions for both shapes.
Could you let me know which contract you'd prefer? If you'd rather go with the scope-aware version outright, I'll take that direction.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58006][SQL][FOLLOWUP] Reject escaping outer refs in forceSkipInline defs [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #58269:
URL: https://github.com/apache/spark/pull/58269#discussion_r4057127915
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InlineCTE.scala:
##########
@@ -101,29 +101,55 @@ case class InlineCTE(
}
private def validateNoOuterReferencesAcrossCTEBoundary(cteDef: CTERelationDef): Unit = {
- val outerRefs = cteDef.child.flatMap(
- _.expressions.flatMap(_.collect { case o: OuterReference => o }))
- if (outerRefs.nonEmpty) {
+ // Only an outer reference that actually escapes the CTE definition is invalid. An outer
+ // reference that points to an attribute produced by an operator inside the definition
+ // body (e.g. a correlated subquery whose correlated column lives in the body) does not
+ // escape, so the definition is self-contained and safe to materialize. Walk the whole
+ // definition (main tree plus nested subquery plans) once, collect every attribute bound
+ // anywhere in the definition, and reject only references that resolve to none of them.
+ //
+ // Matching by exprId is exact for plans produced by a single analysis pass: every
+ // `OuterReference` wraps the very attribute it binds to, exprIds are allocated fresh
+ // per resolution, and `DeduplicateRelations` rewrites duplicated relation instances
+ // (self-join, union, etc.) with fresh exprIds before the optimizer runs. The known
+ // residual gap is a plan stitched together from already-analyzed plans (e.g. a producer
+ // embedding the same analyzed dataset twice): such a tree can carry the same exprId in
+ // multiple places, so an escaping reference could match a duplicated exprId and slip
+ // through. Such a query is ill-formed and fails later during planning or execution
Review Comment:
**Nit (P3):** This does not necessarily fail later. `BindReferences.bindReference` uses the first input ordinal matching the attribute's exprId, so the duplicate described here can bind the escaping reference to an unintended input and continue. Please describe the residual stitched-plan risk as possible failure or wrong-input binding rather than guaranteeing failure.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InlineCTE.scala:
##########
@@ -101,29 +101,55 @@ case class InlineCTE(
}
private def validateNoOuterReferencesAcrossCTEBoundary(cteDef: CTERelationDef): Unit = {
- val outerRefs = cteDef.child.flatMap(
- _.expressions.flatMap(_.collect { case o: OuterReference => o }))
- if (outerRefs.nonEmpty) {
+ // Only an outer reference that actually escapes the CTE definition is invalid. An outer
+ // reference that points to an attribute produced by an operator inside the definition
+ // body (e.g. a correlated subquery whose correlated column lives in the body) does not
+ // escape, so the definition is self-contained and safe to materialize. Walk the whole
+ // definition (main tree plus nested subquery plans) once, collect every attribute bound
+ // anywhere in the definition, and reject only references that resolve to none of them.
+ //
+ // Matching by exprId is exact for plans produced by a single analysis pass: every
+ // `OuterReference` wraps the very attribute it binds to, exprIds are allocated fresh
+ // per resolution, and `DeduplicateRelations` rewrites duplicated relation instances
+ // (self-join, union, etc.) with fresh exprIds before the optimizer runs. The known
+ // residual gap is a plan stitched together from already-analyzed plans (e.g. a producer
+ // embedding the same analyzed dataset twice): such a tree can carry the same exprId in
+ // multiple places, so an escaping reference could match a duplicated exprId and slip
+ // through. Such a query is ill-formed and fails later during planning or execution
+ // anyway; this check is defense-in-depth that fails fast with a clear error otherwise.
+ val allNodes = cteDef.child.collectWithSubqueries { case n: LogicalPlan => n }
+ val boundExprIds = allNodes.iterator
+ .flatMap(_.output.filter(_.resolved).map(_.exprId))
+ .toSet
+
+ // Check the direct `OuterReference` first: if one escapes the def boundary, report it
+ // and stop; the outer-scope scan below only runs when no direct reference escapes.
+ val escapingOuterRef = allNodes.iterator
+ .flatMap(_.expressions.iterator.flatMap(_.collect {
+ case o: OuterReference if !boundExprIds.contains(o.exprId) => o
+ }))
+ .nextOption()
+ escapingOuterRef.foreach { o =>
throw SparkException.internalError(
"A force-materialized CTE cannot carry an outer reference across its boundary, but " +
- s"found outer reference '${outerRefs.head.name}' in the CTE definition " +
+ s"found outer reference '${o.name}' in the CTE definition " +
s"(cteId=${cteDef.id}).")
}
- val outerScopeSubqueries = cteDef.child.flatMap(
- _.expressions.flatMap(_.collect {
- case s: SubqueryExpression if s.outerScopeAttrs.nonEmpty => s
- }))
- if (outerScopeSubqueries.nonEmpty) {
- // `outerScopeAttrs` are expressions that contain an `OuterScopeReference` and may be
- // compound (e.g. `Add(OuterScopeReference(a), Literal(1))`), so collect the reference node
- // rather than assuming the attribute itself is one.
- val outerScopeRef = outerScopeSubqueries.head.outerScopeAttrs
- .flatMap(_.collect { case r: OuterScopeReference => r }).head
+ // Then check for a correlated subquery whose outer-scope attributes escape the def,
Review Comment:
**Non-blocking (P2):** Please add a negative case with an escaping `OuterScopeReference` inside a nested subquery plan. The current suite separately covers nesting through the `OuterReference` branch and direct outer-scope rejection, so restricting only this scan to the top-level CTE child would leave all existing assertions green while allowing that invalid definition through.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58006][SQL][FOLLOWUP] Reject escaping outer refs in forceSkipInline defs [spark]
Posted by "xumingming (via GitHub)" <gi...@apache.org>.
xumingming commented on PR #58269:
URL: https://github.com/apache/spark/pull/58269#issuecomment-5537847996
Gentle ping.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58006][SQL][FOLLOWUP] Reject escaping outer refs in forceSkipInline defs [spark]
Posted by "xumingming (via GitHub)" <gi...@apache.org>.
xumingming commented on PR #58269:
URL: https://github.com/apache/spark/pull/58269#issuecomment-5406984835
@cloud-fan @AveryQi115 Can you take a look at this one?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58006][SQL][FOLLOWUP] Reject escaping outer refs in forceSkipInline defs [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #58269:
URL: https://github.com/apache/spark/pull/58269#discussion_r3864364705
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InlineCTE.scala:
##########
@@ -89,30 +89,45 @@ case class InlineCTE(
}
private def validateNoOuterReferencesAcrossCTEBoundary(cteDef: CTERelationDef): Unit = {
- val outerRefs = cteDef.child.flatMap(
- _.expressions.flatMap(_.collect { case o: OuterReference => o }))
- if (outerRefs.nonEmpty) {
- throw SparkException.internalError(
- "A force-materialized CTE cannot carry an outer reference across its boundary, but " +
- s"found outer reference '${outerRefs.head.name}' in the CTE definition " +
- s"(cteId=${cteDef.id}).")
- }
+ // Only an outer reference that actually escapes the CTE definition is invalid. An outer
+ // reference that points to an attribute produced by an operator inside the definition
+ // body (e.g. a correlated subquery whose correlated column lives in the body) is
+ // self-contained and safe to materialize. So walk the whole definition (main tree plus
Review Comment:
**Nit:**
This sentence makes the outer reference, rather than the CTE definition, `self-contained and safe to materialize`. Say that the reference does not escape, so the definition is safe to materialize.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InlineCTE.scala:
##########
@@ -89,30 +89,45 @@ case class InlineCTE(
}
private def validateNoOuterReferencesAcrossCTEBoundary(cteDef: CTERelationDef): Unit = {
- val outerRefs = cteDef.child.flatMap(
- _.expressions.flatMap(_.collect { case o: OuterReference => o }))
- if (outerRefs.nonEmpty) {
- throw SparkException.internalError(
- "A force-materialized CTE cannot carry an outer reference across its boundary, but " +
- s"found outer reference '${outerRefs.head.name}' in the CTE definition " +
- s"(cteId=${cteDef.id}).")
- }
+ // Only an outer reference that actually escapes the CTE definition is invalid. An outer
+ // reference that points to an attribute produced by an operator inside the definition
+ // body (e.g. a correlated subquery whose correlated column lives in the body) is
+ // self-contained and safe to materialize. So walk the whole definition (main tree plus
+ // nested subquery plans) once, collect every attribute bound anywhere in the definition,
+ // and reject only references that resolve to none of them.
+ val allNodes = cteDef.child.collectWithSubqueries { case n: LogicalPlan => n }
+ val boundExprIds = allNodes.iterator
+ .flatMap(_.output.filter(_.resolved).map(_.exprId))
+ .toSet
- val outerScopeSubqueries = cteDef.child.flatMap(
- _.expressions.flatMap(_.collect {
- case s: SubqueryExpression if s.outerScopeAttrs.nonEmpty => s
+ // Locate either a direct `OuterReference` or a correlated subquery whose outer-scope
+ // attributes escape the def -- i.e. resolve to none of `boundExprIds`.
+ val escapingOuterRef = allNodes.iterator
+ .flatMap(_.expressions.iterator.flatMap(_.collect {
+ case o: OuterReference if !boundExprIds.contains(o.exprId) => o
}))
- if (outerScopeSubqueries.nonEmpty) {
- // `outerScopeAttrs` are expressions that contain an `OuterScopeReference` and may be
- // compound (e.g. `Add(OuterScopeReference(a), Literal(1))`), so collect the reference node
- // rather than assuming the attribute itself is one.
- val outerScopeRef = outerScopeSubqueries.head.outerScopeAttrs
- .flatMap(_.collect { case r: OuterScopeReference => r }).head
- throw SparkException.internalError(
- "A force-materialized CTE cannot carry an outer reference across its boundary, but " +
- "found a subquery with outer-scope reference " +
- s"'${outerScopeRef.name}' in the CTE definition " +
- s"(cteId=${cteDef.id}).")
+ .nextOption()
+ val escapingOuterScopeRef = allNodes.iterator
Review Comment:
**Non-blocking:**
Skip the outer-scope scan when `escapingOuterRef` is already defined. The following match always reports the direct reference, so this second full traversal cannot affect the result.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58006][SQL][FOLLOWUP] Reject escaping outer refs in forceSkipInline defs [spark]
Posted by "xumingming (via GitHub)" <gi...@apache.org>.
xumingming commented on PR #58269:
URL: https://github.com/apache/spark/pull/58269#issuecomment-6038845230
@cloud-fan Thanks for the reviews. Pushed c16bf7a531d addressing the review comments:
- `CTERelationDef`'s doc no longer promises fail-fast: it now states the escaping-reference validation matches by exprId and is exact only for plans produced by a single analysis pass, with the residual stitched-plan gap (duplicated exprIds can let an escape slip through and bind to an unintended input) leaning on the `InlineCTE` validation comment for details.
- The nested negative test comment now counts `m.a` as three scopes out of the deepest subquery and drops "nested-wide" for plain wording that the outer-scope scan also inspects nested subquery plans.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org