This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 6ec97a445899 [SPARK-58779][SQL] Make InlineCTE tolerate a CTE
reference without its definition during analysis
6ec97a445899 is described below
commit 6ec97a4458994c5ff8c1c8722dd682b68a4d9e2a
Author: Avery <[email protected]>
AuthorDate: Thu Aug 20 21:56:16 2026 +0800
[SPARK-58779][SQL] Make InlineCTE tolerate a CTE reference without its
definition during analysis
### What changes were proposed in this pull request?
`InlineCTE` now guards its two `cteMap(ref.cteId)` lookups (in
`buildCTEMap` and `inlineCTE`) so that a `CTERelationRef` whose
`CTERelationDef` is not present in the plan being processed is handled instead
of throwing a bare `NoSuchElementException`. A new `isAnalysis` flag (default
`true`) distinguishes the two kinds of caller:
- During analysis (`CheckAnalysis.checkAnalysis`, which runs
`InlineCTE(alwaysInline = true, keepDanglingRelations = true)`), such an
out-of-scope reference is tolerated and left for the scope that owns the
definition to resolve.
- In the optimizer (`isAnalysis = false`), the plan is complete, so a
reference with no definition indicates a corrupt plan and raises
`SparkException.internalError`.
The optimizer's `InlineCTE()` call in `Optimizer.scala` is updated to
`InlineCTE(isAnalysis = false)`; the analysis-time callers keep the default.
### Why are the changes needed?
`buildCTEMap` and `inlineCTE` did unguarded `cteMap(ref.cteId)` lookups
that threw `NoSuchElementException` when a `CTERelationRef`'s definition is not
in the plan being processed. `CheckAnalysis.checkAnalysis` runs `InlineCTE` and
only catches `AnalysisException`, so the `NoSuchElementException` propagated
uncaught.
This happens when `checkAnalysis` runs on a subplan that references an
outer CTE - e.g. `ResolveSQLTableFunctions` calls `checkAnalysis` on a resolved
SQL table function whose argument is a scalar subquery referencing an outer CTE:
```sql
CREATE FUNCTION my_tvf(a INT, b INT) RETURNS TABLE(x INT) RETURN SELECT a +
b AS x;
WITH cte AS (SELECT 1 AS c1, 2 AS c2)
SELECT * FROM my_tvf((SELECT c1 FROM cte), (SELECT c2 FROM cte));
```
The resolved function plan contains CTERelationRef(cte), while its
CTERelationDef lives in the outer WithCTE (outside the subplan that
checkAnalysis is validating), so the lookup throws.
### Does this PR introduce any user-facing change?
Yes. A query such as the one above previously failed analysis with an
internal NoSuchElementException (both on released versions and on master); with
this change it analyzes and executes correctly. There is no API or
result-schema change.
### How was this patch tested?
New tests:
- SQLFunctionSuite - end-to-end: a SQL table function called with
scalar-subquery arguments that reference a CTE now succeeds (previously threw
NoSuchElementException during analysis).
- InlineCTESuite - unit: the optimizer path (isAnalysis = false) raises
INTERNAL_ERROR on a CTERelationRef with no definition, while the analysis path
(isAnalysis = true) tolerates it.
### Was this patch authored or co-authored using generative AI tooling?
Yes. Generated-by: Claude Code (Anthropic).
Closes #58012 from AveryQi115/SPARK-58779.
Lead-authored-by: Avery <[email protected]>
Co-authored-by: Avery Qi <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
(cherry picked from commit bb61ee6980be3e0b8ce4d6657f89e36c70feef11)
Signed-off-by: Wenchen Fan <[email protected]>
---
.../sql/catalyst/analysis/CheckAnalysis.scala | 2 +-
.../spark/sql/catalyst/optimizer/InlineCTE.scala | 118 ++++++++++++++-------
.../sql/catalyst/optimizer/InlineCTESuite.scala | 18 ++++
.../spark/sql/execution/SQLFunctionSuite.scala | 21 ++++
4 files changed, 118 insertions(+), 41 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala
index 36eeedaf30ed..53b5780847f2 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala
@@ -307,7 +307,7 @@ trait CheckAnalysis extends LookupCatalog with
QueryErrorsBase with PlanToString
// We should inline all CTE relations to restore the original plan shape,
as the analysis check
// may need to match certain plan shapes. For dangling CTE relations, they
will still be kept
// in the original `WithCTE` node, as we need to perform analysis check
for them as well.
- val inlineCTE = InlineCTE(alwaysInline = true, keepDanglingRelations =
true)
+ val inlineCTE = InlineCTE(alwaysInline = true, keepDanglingRelations =
true, isAnalysis = true)
val inlinedPlan: LogicalPlan = try {
inlineCTE(plan)
} catch {
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InlineCTE.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InlineCTE.scala
index c8bd01ed8849..1b36ba04cfd7 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InlineCTE.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InlineCTE.scala
@@ -40,10 +40,19 @@ import
org.apache.spark.sql.catalyst.trees.TreePattern.{CTE, PLAN_EXPRESSION}
* @param alwaysInline if true, inline all CTEs in the query plan.
* @param keepDanglingRelations if true, dangling CTE relations will be kept
in the original
* `WithCTE` node.
+ * @param isAnalysis if true, this rule runs during analysis (e.g. from
`CheckAnalysis`), where
+ * the plan may be a subplan that references
`CTERelationDef`s owned by a
+ * surrounding scope; such out-of-scope references are
tolerated and left for
+ * the owning scope to resolve. Defaults to false: every
other caller (the
+ * optimizer, `ProgressReporter`) operates on a complete
plan, so a
+ * `CTERelationRef` with no definition in the plan indicates
corruption and
+ * raises an error rather than being silently dropped. Only
the scoped
+ * `CheckAnalysis` call opts into the tolerant behavior.
*/
case class InlineCTE(
alwaysInline: Boolean = false,
- keepDanglingRelations: Boolean = false) extends Rule[LogicalPlan] {
+ keepDanglingRelations: Boolean = false,
+ isAnalysis: Boolean = false) extends Rule[LogicalPlan] {
override def apply(plan: LogicalPlan): LogicalPlan = {
if (!plan.isInstanceOf[Subquery] && plan.containsPattern(CTE)) {
@@ -148,18 +157,36 @@ case class InlineCTE(
buildCTEMap(child, cteMap, outerCTEId)
}
- case ref: CTERelationRef =>
- cteMap(ref.cteId) = cteMap(ref.cteId).withRefCountIncreased(1)
+ case ref: CTERelationRef => cteMap.get(ref.cteId) match {
+ case Some(refInfo) =>
+ cteMap(ref.cteId) = refInfo.withRefCountIncreased(1)
- // The `outerCTEId` CTE definition can either reference `cteId`
definition if `cteId` is in
- // the same or in an outer `WithCTE` node, or `outerCTEId` can contain
`cteId` definition if
- // `cteId` is an inner `WithCTE` node inside `outerCTEId`.
- // In both cases we can track the relations in `outgoingRefs` when we
see a definition the
- // first time. But if we encounter a conflicting duplicated contains
relation later, then we
- // will remove the references of the first contains relation.
- outerCTEId.foreach { cteId =>
- cteMap(cteId).increaseOutgoingRefCount(ref.cteId, 1)
- }
+ // The `outerCTEId` CTE definition can either reference `cteId`
definition if `cteId` is
+ // in the same or in an outer `WithCTE` node, or `outerCTEId` can
contain `cteId`
+ // definition if `cteId` is an inner `WithCTE` node inside
`outerCTEId`.
+ // In both cases we can track the relations in `outgoingRefs` when
we see a definition the
+ // first time. But if we encounter a conflicting duplicated contains
relation later, then
+ // we will remove the references of the first contains relation.
+ outerCTEId.foreach { cteId =>
+ cteMap(cteId).increaseOutgoingRefCount(ref.cteId, 1)
+ }
+
+ case None =>
+ // The referenced CTE definition is not present in this plan. During
analysis
+ // (`isAnalysis` = true) this legitimately happens when a check runs
on a subplan that
+ // contains the reference but not its enclosing `WithCTE` -- e.g.
+ // `ResolveSQLTableFunctions` calls `checkAnalysis` (which runs
`InlineCTE`) on a resolved
+ // SQL table function whose argument is a scalar subquery
referencing an outer CTE. The
+ // reference is resolved by the scope that owns the definition, so
there is nothing to
+ // count here. Otherwise (`isAnalysis` = false) the plan is
complete, so a missing
+ // definition indicates corruption -- fail loudly rather than
silently dropping the
+ // reference.
+ if (!isAnalysis) {
+ throw SparkException.internalError(
+ "No CTERelationDef found for CTERelationRef with id " +
+ s"${ref.cteId} while building the CTE map.")
+ }
+ }
case _ =>
if (plan.containsPattern(CTE)) {
@@ -229,39 +256,50 @@ case class InlineCTE(
WithCTE(inlined, notInlined)
}
- case ref: CTERelationRef =>
- val refInfo = cteMap(ref.cteId)
+ case ref: CTERelationRef => cteMap.get(ref.cteId) match {
+ case None =>
+ // Out-of-scope reference whose definition is not in this plan
(mirrors the guard in
+ // `buildCTEMap`). During analysis it is left unchanged for the
scope that owns the
+ // definition; otherwise a missing definition is corruption.
+ if (!isAnalysis) {
+ throw SparkException.internalError(
+ "No CTERelationDef found for CTERelationRef with id " +
+ s"${ref.cteId} while inlining CTEs.")
+ }
+ ref
- val cteBody = if (ref.isUnlimitedRecursion) {
- setUnlimitedRecursion(refInfo.cteDef.child, ref.cteId)
- } else {
- refInfo.cteDef.child
- }
- if (refInfo.shouldInline) {
- if (ref.outputSet == refInfo.cteDef.outputSet) {
- cteBody
+ case Some(refInfo) =>
+ val cteBody = if (ref.isUnlimitedRecursion) {
+ setUnlimitedRecursion(refInfo.cteDef.child, ref.cteId)
} else {
- val ctePlan = DeduplicateRelations(
- Join(
- cteBody,
- cteBody,
- Inner,
- None,
- JoinHint(None, None)
- )
- ).children(1)
- val projectList = ref.output.zip(ctePlan.output).map { case
(tgtAttr, srcAttr) =>
- if (srcAttr.semanticEquals(tgtAttr)) {
- tgtAttr
- } else {
- Alias(srcAttr, tgtAttr.name)(exprId = tgtAttr.exprId)
+ refInfo.cteDef.child
+ }
+ if (refInfo.shouldInline) {
+ if (ref.outputSet == refInfo.cteDef.outputSet) {
+ cteBody
+ } else {
+ val ctePlan = DeduplicateRelations(
+ Join(
+ cteBody,
+ cteBody,
+ Inner,
+ None,
+ JoinHint(None, None)
+ )
+ ).children(1)
+ val projectList = ref.output.zip(ctePlan.output).map { case
(tgtAttr, srcAttr) =>
+ if (srcAttr.semanticEquals(tgtAttr)) {
+ tgtAttr
+ } else {
+ Alias(srcAttr, tgtAttr.name)(exprId = tgtAttr.exprId)
+ }
}
+ Project(projectList, ctePlan)
}
- Project(projectList, ctePlan)
+ } else {
+ ref
}
- } else {
- ref
- }
+ }
case _ if plan.containsPattern(CTE) =>
plan
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/InlineCTESuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/InlineCTESuite.scala
index 0d515c482401..2daad7dfb64b 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/InlineCTESuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/InlineCTESuite.scala
@@ -108,4 +108,22 @@ class InlineCTESuite extends PlanTest {
assert(e.getMessage.contains(
"found a subquery with outer-scope reference"))
}
+
+ test("SPARK-58779: optimizer InlineCTE (isAnalysis = false) fails on a ref
with no definition") {
+ // During analysis a CTERelationRef whose definition is not in the plan is
tolerated -- it is
+ // owned by a surrounding scope (e.g. when `ResolveSQLTableFunctions` runs
`checkAnalysis` on a
+ // table-function subplan whose argument references an outer CTE). In the
optimizer the plan is
+ // complete, so a missing definition indicates corruption and must fail
loudly rather than be
+ // dropped.
+ val defX = CTERelationDef(TestRelation(Seq($"a".int)).select($"a"))
+ val refX = CTERelationRef(defX.id, defX.resolved, defX.output,
defX.isStreaming)
+ val danglingRef = CTERelationRef(defX.id + 1000, true, Seq($"a".int),
false)
+ val plan = WithCTE(refX.union(danglingRef), Seq(defX))
+ val e = intercept[SparkException] {
+ InlineCTE(isAnalysis = false).apply(plan)
+ }
+ assert(e.getCondition == "INTERNAL_ERROR")
+ // The analysis path tolerates the out-of-scope reference (no throw).
+ InlineCTE(isAnalysis = true).apply(plan)
+ }
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/SQLFunctionSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/SQLFunctionSuite.scala
index 7406f953d063..d7956a9c1217 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/SQLFunctionSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/SQLFunctionSuite.scala
@@ -79,6 +79,27 @@ class SQLFunctionSuite extends SharedSparkSession {
}
}
+ test("SPARK-58779: SQL table function argument referencing a CTE") {
+ withUserDefinedFunction("my_tvf" -> false) {
+ sql(
+ """
+ |CREATE FUNCTION my_tvf(a INT, b INT)
+ |RETURNS TABLE(x INT)
+ |RETURN SELECT a + b AS x
+ |""".stripMargin)
+ // The table function is called with scalar-subquery arguments that
reference a CTE.
+ // ResolveSQLTableFunctions runs checkAnalysis (which runs InlineCTE) on
the resolved
+ // function plan, which contains a CTERelationRef whose CTERelationDef
lives in the outer
+ // WithCTE scope. Previously the unguarded cteMap(ref.cteId) lookup threw
+ // NoSuchElementException.
+ checkAnswer(sql(
+ """
+ |WITH cte AS (SELECT 1 AS c1, 2 AS c2)
+ |SELECT * FROM my_tvf((SELECT c1 FROM cte), (SELECT c2 FROM cte))
+ |""".stripMargin), Row(3))
+ }
+ }
+
test("SQL scalar function with default value") {
withUserDefinedFunction("bar" -> false) {
sql(
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
