[SPARK-50838][SQL] Add Cross Join as legal in recursion of Recursive CTE

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

@@ -21,7 +21,7 @@ import scala.collection.mutable
 
 import org.apache.spark.sql.AnalysisException
 import org.apache.spark.sql.catalyst.expressions.SubqueryExpression
-import org.apache.spark.sql.catalyst.plans.{Inner, LeftAnti, LeftOuter, LeftSemi, RightOuter}
+import org.apache.spark.sql.catalyst.plans.{Cross, Inner, LeftAnti, LeftOuter, LeftSemi, RightOuter}
 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}
@@ -220,6 +220,9 @@ object ResolveWithCTE extends Rule[LogicalPlan] {
     case Join(left, right, Inner, _, _) =>
       checkIfSelfReferenceIsPlacedCorrectly(left, cteId, allowRecursiveRef)
       checkIfSelfReferenceIsPlacedCorrectly(right, cteId, allowRecursiveRef)
+    case Join(left, right, Cross, _, _) =>
+      checkIfSelfReferenceIsPlacedCorrectly(left, cteId, allowRecursiveRef)
+      checkIfSelfReferenceIsPlacedCorrectly(right, cteId, allowRecursiveRef)
     case Join(left, right, LeftOuter, _, _) =>
       checkIfSelfReferenceIsPlacedCorrectly(left, cteId, allowRecursiveRef)
       checkIfSelfReferenceIsPlacedCorrectly(right, cteId, allowRecursiveRef = false)

sql/core/src/test/resources/sql-tests/analyzer-results/cte-recursion.sql.out

@@ -1167,3 +1167,55 @@ WithCTE
       +- Project [a#x]
          +- SubqueryAlias t1
             +- CTERelationRef xxxx, true, [a#x, b#x, c#x], false, false
+
+
+-- !query
+CREATE TABLE tb (
+                    next INT
+)
+-- !query analysis
+CreateDataSourceTableCommand `spark_catalog`.`default`.`tb`, false
+
+
+-- !query
+INSERT INTO tb VALUES (0), (1)
+-- !query analysis
+InsertIntoHadoopFsRelationCommand file:[not included in comparison]/{warehouse_dir}/tb, false, Parquet, [path=file:[not included in comparison]/{warehouse_dir}/tb], Append, `spark_catalog`.`default`.`tb`, org.apache.spark.sql.execution.datasources.InMemoryFileIndex(file:[not included in comparison]/{warehouse_dir}/tb), [next]
++- Project [cast(col1#x as int) AS next#x]
+   +- LocalRelation [col1#x]
+
+
+-- !query
+WITH RECURSIVE t(n) AS (
+    SELECT 1
+    UNION ALL
+    SELECT next FROM t CROSS JOIN tb
+    )
+SELECT * FROM t LIMIT 63
+-- !query analysis
+WithCTE
+:- CTERelationDef xxxx, false
+:  +- SubqueryAlias t
+:     +- Project [1#x AS n#x]
+:        +- UnionLoop xxxx
+:           :- Project [1 AS 1#x]
+:           :  +- OneRowRelation
+:           +- Project [next#x]
+:              +- Join Cross
+:                 :- SubqueryAlias t
+:                 :  +- Project [1#x AS n#x]
+:                 :     +- UnionLoopRef xxxx, [1#x], false
+:                 +- SubqueryAlias spark_catalog.default.tb
+:                    +- Relation spark_catalog.default.tb[next#x] parquet
++- GlobalLimit 63
+   +- LocalLimit 63
+      +- Project [n#x]
+         +- SubqueryAlias t
+            +- CTERelationRef xxxx, true, [n#x], false, false
+
+
+-- !query
+DROP TABLE tb
+-- !query analysis
+DropTable false, false
++- ResolvedIdentifier V2SessionCatalog(spark_catalog), default.tb

sql/core/src/test/resources/sql-tests/inputs/cte-recursion.sql

@@ -462,4 +462,21 @@ WITH RECURSIVE t1(a,b,c) AS (
     SELECT 1,1,1
     UNION ALL
     SELECT a+1,a+1,a+1 FROM t1)
-SELECT a FROM t1 LIMIT 5;
+SELECT a FROM t1 LIMIT 5;
+
+-- CROSS JOIN example
+CREATE TABLE tb (
+                    next INT
+);
+
+INSERT INTO tb VALUES (0), (1);
+
+-- create
+WITH RECURSIVE t(n) AS (
+    SELECT 1
+    UNION ALL
+    SELECT next FROM t CROSS JOIN tb
+    )
+SELECT * FROM t LIMIT 63;
+
+DROP TABLE tb;

sql/core/src/test/resources/sql-tests/results/cte-recursion.sql.out

@@ -1050,3 +1050,104 @@ struct<a:int>
 3
 4
 5
+
+
+-- !query
+CREATE TABLE tb (
+                    next INT
+)
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+INSERT INTO tb VALUES (0), (1)
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+WITH RECURSIVE t(n) AS (
+    SELECT 1
+    UNION ALL
+    SELECT next FROM t CROSS JOIN tb
+    )
+SELECT * FROM t LIMIT 63
+-- !query schema
+struct<n:int>
+-- !query output
+0
+0
+0
+0
+0
+0
+0
+0
+0
+0
+0
+0
+0
+0
+0
+0
+0
+0
+0
+0
+0
+0
+0
+0
+0
+0
+0
+0
+0
+0
+0
+1
+1
+1
+1
+1
+1
+1
+1
+1
+1
+1
+1
+1
+1
+1
+1
+1
+1
+1
+1
+1
+1
+1
+1
+1
+1
+1
+1
+1
+1
+1
+1
+
+
+-- !query
+DROP TABLE tb
+-- !query schema
+struct<>
+-- !query output
+