[SPARK-50739][SQL] Recursive CTE. Analyzer changes to unravel and resolve the recursion components

### What changes were proposed in this pull request?
Instruction for reviewers https://docs.google.com/document/d/1qcEJxqoXcr5cSt6HgIQjWQSqhfkSaVYkoDHsg5oxXp4/edit
Introduction of UnionLoop and UnionLoopRef logical plan classes. Changes in ResolveWithCTE.scala to have the analyzer grok recursive anchors. Specifically we substitute CTERelationRef with UnionLoopRef, and Union with UnionLoop. We untangle the dead loop in resolving where recursive CTE reference is referring to an unresolved CTE definition, which itself cannot be resolved as one of its descendants is an unresolved CTE reference.

### Why are the changes needed?
Support for the recursive CTE.

### Does this PR introduce _any_ user-facing change?
No.

### How was this patch tested?
Proposed changes in this PR are no-op. Tested
./build/sbt "test:testOnly org.apache.spark.sql.SQLQueryTestSuite"
./build/sbt "test:testOnly *PlanParserSuite"

### Was this patch authored or co-authored using generative AI tooling?
No

Closes #49351 from nemanjapetr-db/nemanjapetr-db/rcte3.

Authored-by: Nemanja Petrovic <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>

Author: nemanjapetr-db

common/utils/src/main/resources/error/error-conditions.json

@@ -3105,6 +3105,12 @@
     ],
     "sqlState" : "42613"
   },
+  "INVALID_RECURSIVE_CTE" : {
+    "message" : [
+      "Invalid recursive definition found. Recursive queries must contain an UNION or an UNION ALL statement with 2 children. The first child needs to be the anchor term without any recursive references."
+    ],
+    "sqlState" : "42836"
+  },
   "INVALID_REGEXP_REPLACE" : {
     "message" : [
       "Could not perform regexp_replace for source = \"<source>\", pattern = \"<pattern>\", replacement = \"<replacement>\" and position = <position>."

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CTESubstitution.scala

@@ -316,7 +316,9 @@ object CTESubstitution extends Rule[LogicalPlan] {
       // CTE definition can reference a previous one or itself if recursion allowed.
       val substituted = substituteCTE(innerCTEResolved, alwaysInline,
         resolvedCTERelations, recursiveCTERelation)
-      val cteRelation = CTERelationDef(substituted)
+      val cteRelation = recursiveCTERelation
+        .map(_._2.copy(child = substituted))
+        .getOrElse(CTERelationDef(substituted))
       if (!alwaysInline) {
         cteDefs += cteRelation
       }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveWithCTE.scala

@@ -20,16 +20,22 @@ package org.apache.spark.sql.catalyst.analysis
 import scala.collection.mutable
 
 import org.apache.spark.sql.catalyst.expressions.SubqueryExpression
-import org.apache.spark.sql.catalyst.plans.logical.{CTERelationDef, CTERelationRef, LogicalPlan, WithCTE}
+import org.apache.spark.sql.catalyst.plans.logical._
 import org.apache.spark.sql.catalyst.rules.Rule
 import org.apache.spark.sql.catalyst.trees.TreePattern.{CTE, PLAN_EXPRESSION}
+import org.apache.spark.sql.errors.QueryCompilationErrors
 
 /**
  * Updates CTE references with the resolve output attributes of corresponding CTE definitions.
  */
 object ResolveWithCTE extends Rule[LogicalPlan] {
   override def apply(plan: LogicalPlan): LogicalPlan = {
     if (plan.containsAllPatterns(CTE)) {
+      // A helper map definitionID->Definition. Used for non-recursive CTE definitions only, either
+      // inherently non-recursive or that became non-recursive due to recursive CTERelationRef->
+      // UnionLoopRef substitution. Bridges the gap between two CTE resolutions (one for WithCTE,
+      // another for reference) by materializing the resolved definitions in first pass and
+      // consuming them in the second.
       val cteDefMap = mutable.HashMap.empty[Long, CTERelationDef]
       resolveWithCTE(plan, cteDefMap)
     } else {
@@ -41,16 +47,113 @@ object ResolveWithCTE extends Rule[LogicalPlan] {
       plan: LogicalPlan,
       cteDefMap: mutable.HashMap[Long, CTERelationDef]): LogicalPlan = {
     plan.resolveOperatorsDownWithPruning(_.containsAllPatterns(CTE)) {
-      case w @ WithCTE(_, cteDefs) =>
-        cteDefs.foreach { cteDef =>
-          if (cteDef.resolved) {
-            cteDefMap.put(cteDef.id, cteDef)
-          }
+      case withCTE @ WithCTE(_, cteDefs) =>
+        val newCTEDefs = cteDefs.map {
+          // `cteDef.recursive` means "presence of a recursive CTERelationRef under cteDef". The
+          // side effect of node substitution below is that after CTERelationRef substitution
+          // its cteDef is no more considered `recursive`. This code path is common for `cteDef`
+          // that were non-recursive from the get go, as well as those that are no more recursive
+          // due to node substitution.
+          case cteDef if !cteDef.recursive =>
+            if (cteDef.resolved) {
+              cteDefMap.put(cteDef.id, cteDef)
+            }
+            cteDef
+          case cteDef =>
+            cteDef.child match {
+              // If it is a supported recursive CTE query pattern (4 so far), extract the anchor and
+              // recursive plans from the Union and rewrite Union with UnionLoop. The recursive CTE
+              // references inside UnionLoop's recursive plan will be rewritten as UnionLoopRef,
+              // using the output of the resolved anchor plan. The side effect of recursive
+              // CTERelationRef->UnionLoopRef substitution is that `cteDef` that was originally
+              // considered `recursive` is no more in the context of `cteDef.recursive` method
+              // definition.
+              //
+              // Simple case of duplicating (UNION ALL) clause.
+              case alias @ SubqueryAlias(_, Union(Seq(anchor, recursion), false, false)) =>
+                if (!anchor.resolved) {
+                  cteDef
+                } else {
+                  val loop = UnionLoop(
+                    cteDef.id,
+                    anchor,
+                    rewriteRecursiveCTERefs(recursion, anchor, cteDef.id, None))
+                  cteDef.copy(child = alias.copy(child = loop))
+                }
+
+              // The case of CTE name followed by a parenthesized list of column name(s), eg.
+              // WITH RECURSIVE t(n).
+              case alias @ SubqueryAlias(_,
+                  columnAlias @ UnresolvedSubqueryColumnAliases(
+                  colNames,
+                  Union(Seq(anchor, recursion), false, false)
+                )) =>
+                if (!anchor.resolved) {
+                  cteDef
+                } else {
+                  val loop = UnionLoop(
+                    cteDef.id,
+                    anchor,
+                    rewriteRecursiveCTERefs(recursion, anchor, cteDef.id, Some(colNames)))
+                  cteDef.copy(child = alias.copy(child = columnAlias.copy(child = loop)))
+                }
+
+              // If the recursion is described with an UNION (deduplicating) clause then the
+              // recursive term should not return those rows that have been calculated previously,
+              // and we exclude those rows from the current iteration result.
+              case alias @ SubqueryAlias(_,
+                  Distinct(Union(Seq(anchor, recursion), false, false))) =>
+                if (!anchor.resolved) {
+                  cteDef
+                } else {
+                  val loop = UnionLoop(
+                    cteDef.id,
+                    Distinct(anchor),
+                    Except(
+                      rewriteRecursiveCTERefs(recursion, anchor, cteDef.id, None),
+                      UnionLoopRef(cteDef.id, anchor.output, true),
+                      isAll = false
+                    )
+                  )
+                  cteDef.copy(child = alias.copy(child = loop))
+                }
+
+              // The case of CTE name followed by a parenthesized list of column name(s).
+              case alias @ SubqueryAlias(_,
+                  columnAlias@UnresolvedSubqueryColumnAliases(
+                  colNames,
+                  Distinct(Union(Seq(anchor, recursion), false, false))
+                )) =>
+                if (!anchor.resolved) {
+                  cteDef
+                } else {
+                  val loop = UnionLoop(
+                    cteDef.id,
+                    Distinct(anchor),
+                    Except(
+                      rewriteRecursiveCTERefs(recursion, anchor, cteDef.id, Some(colNames)),
+                      UnionLoopRef(cteDef.id, anchor.output, true),
+                      isAll = false
+                    )
+                  )
+                  cteDef.copy(child = alias.copy(child = columnAlias.copy(child = loop)))
+                }
+
+              case other =>
+                // We do not support cases of sole Union (needs a SubqueryAlias above it), nor
+                // Project (as UnresolvedSubqueryColumnAliases have not been substituted with the
+                // Project yet), leaving us with cases of SubqueryAlias->Union and SubqueryAlias->
+                // UnresolvedSubqueryColumnAliases->Union. The same applies to Distinct Union.
+                throw QueryCompilationErrors.invalidRecursiveCteError(
+                  "Unsupported recursive CTE UNION placement.")
+            }
         }
-        w
+        withCTE.copy(cteDefs = newCTEDefs)
 
+      // This is a non-recursive reference to a definition.
       case ref: CTERelationRef if !ref.resolved =>
         cteDefMap.get(ref.cteId).map { cteDef =>
+          // cteDef is certainly resolved, otherwise it would not have been in the map.
           CTERelationRef(cteDef.id, cteDef.resolved, cteDef.output, cteDef.isStreaming)
         }.getOrElse {
           ref
@@ -62,4 +165,21 @@ object ResolveWithCTE extends Rule[LogicalPlan] {
         }
     }
   }
+
+  // Substitutes a recursive CTERelationRef with UnionLoopRef if CTERelationRef refers to cteDefId.
+  // Assumes that `anchor` is already resolved. If `columnNames` is set (which happens if an
+  // UnresolvedSubqueryColumnAliases is atop a Union) also places an UnresolvedSubqueryColumnAliases
+  // node above UnionLoopRef. At some later stage UnresolvedSubqueryColumnAliases gets resolved to
+  // a Project node.
+  private def rewriteRecursiveCTERefs(
+      recursion: LogicalPlan,
+      anchor: LogicalPlan,
+      cteDefId: Long,
+      columnNames: Option[Seq[String]]) = {
+    recursion.transformWithPruning(_.containsPattern(CTE)) {
+      case r: CTERelationRef if r.recursive && r.cteId == cteDefId =>
+        val ref = UnionLoopRef(r.cteId, anchor.output, false)
+        columnNames.map(UnresolvedSubqueryColumnAliases(_, ref)).getOrElse(ref)
+    }
+  }
 }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushdownPredicatesAndPruneColumnsForCTEDef.scala

@@ -122,7 +122,7 @@ object PushdownPredicatesAndPruneColumnsForCTEDef extends Rule[LogicalPlan] {
   private def pushdownPredicatesAndAttributes(
       plan: LogicalPlan,
       cteMap: CTEMap): LogicalPlan = plan.transformWithSubqueries {
-    case cteDef @ CTERelationDef(child, id, originalPlanWithPredicates, _, _) =>
+    case cteDef @ CTERelationDef(child, id, originalPlanWithPredicates, _) =>
       val (_, _, newPreds, newAttrSet) = cteMap(id)
       val originalPlan = originalPlanWithPredicates.map(_._1).getOrElse(child)
       val preds = originalPlanWithPredicates.map(_._2).getOrElse(Seq.empty)
@@ -170,7 +170,7 @@ object PushdownPredicatesAndPruneColumnsForCTEDef extends Rule[LogicalPlan] {
 object CleanUpTempCTEInfo extends Rule[LogicalPlan] {
   override def apply(plan: LogicalPlan): LogicalPlan =
     plan.transformWithPruning(_.containsPattern(CTE)) {
-      case cteDef @ CTERelationDef(_, _, Some(_), _, _) =>
+      case cteDef @ CTERelationDef(_, _, Some(_), _) =>
         cteDef.copy(originalPlanWithPredicates = None)
     }
 }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/QueryPlan.scala

@@ -774,4 +774,23 @@ object QueryPlan extends PredicateHelper {
       case e: AnalysisException => append(e.toString)
     }
   }
+
+  /**
+   * Generate detailed field string with different format based on type of input value. Supported
+   * input values are sequences and strings. An empty sequences converts to []. A non-empty
+   * sequences converts to square brackets-enclosed, comma-separated values, prefixed with a
+   * sequence length. An empty string converts to None, while a non-empty string is verbatim
+   * outputted. In all four cases, user-provided fieldName prefixes the output. Examples:
+   * List("Hello", "World") -> <fieldName>: [2]: [Hello, World]
+   * List()                 -> <fieldName>: []
+   * "hello_world"          -> <fieldName>: hello_world
+   * ""                     -> <fieldName>: None
+   */
+  def generateFieldString(fieldName: String, values: Any): String = values match {
+    case iter: Iterable[_] if (iter.size == 0) => s"${fieldName}: []"
+    case iter: Iterable[_] => s"${fieldName} [${iter.size}]: ${iter.mkString("[", ", ", "]")}"
+    case str: String if (str == null || str.isEmpty) => s"${fieldName}: None"
+    case str: String => s"${fieldName}: ${str}"
+    case _ => throw new IllegalArgumentException(s"Unsupported type for argument values: $values")
+  }
 }