You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2026/08/20 13:56:33 UTC
(spark) branch master updated: [SPARK-58779][SQL] Make InlineCTE tolerate a CTE reference without its definition during analysis
This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new bb61ee6980be [SPARK-58779][SQL] Make InlineCTE tolerate a CTE reference without its definition during analysis
bb61ee6980be is described below
commit bb61ee6980be3e0b8ce4d6657f89e36c70feef11
Author: Avery <av...@gmail.com>
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 <av...@gmail.com>
Co-authored-by: Avery Qi <av...@databricks.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../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: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org