[SPARK-52549][SQL] Disable Recursive CTE self-references from window functions and inside sorts

### What changes were proposed in this pull request?

Throw error when a UnionLoopRef is found inside a window function.
### Why are the changes needed?

The context of a Recursive CTE is only the result of the previous iteration, so any type of cumulative operations should not be allowed. This is also the reason why self references inside aggregates aren't allowed.

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

No.

### How was this patch tested?

New golden file test.

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

No.

Closes apache#51178 from Pajaraja/pavle-martinovic_data/noWindowFunctionsrCTE.

Authored-by: pavle-martinovic_data <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>

Author: pavle-martinovic_data

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

@@ -3429,7 +3429,7 @@
     "subClass" : {
       "PLACE" : {
         "message" : [
-          "Recursive references cannot be used on the right side of left outer/semi/anti joins, on the left side of right outer joins, in full outer joins and in aggregates"
+          "Recursive references cannot be used on the right side of left outer/semi/anti joins, on the left side of right outer joins, in full outer joins, in aggregates, window functions or sorts"
         ]
       }
     },

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

@@ -347,6 +347,10 @@ object ResolveWithCTE extends Rule[LogicalPlan] {
         checkIfSelfReferenceIsPlacedCorrectly(right, cteId, allowRecursiveRef = false)
       case Aggregate(_, _, child, _) =>
         checkIfSelfReferenceIsPlacedCorrectly(child, cteId, allowRecursiveRef = false)
+      case Window(_, _, _, child, _) =>
+        checkIfSelfReferenceIsPlacedCorrectly(child, cteId, allowRecursiveRef = false)
+      case Sort(_, _, child, _) =>
+        checkIfSelfReferenceIsPlacedCorrectly(child, cteId, allowRecursiveRef = false)
       case r: UnionLoopRef if !allowRecursiveRef && r.loopId == cteId =>
         throw new AnalysisException(
           errorClass = "INVALID_RECURSIVE_REFERENCE.PLACE",

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

@@ -2011,3 +2011,34 @@ WithCTE
          +- Project [n#x]
             +- SubqueryAlias t1
                +- CTERelationRef xxxx, true, [n#x], false, false
+
+
+-- !query
+WITH RECURSIVE win(id, val) AS (
+    SELECT 1, CAST(10 AS BIGINT)
+    UNION ALL
+    SELECT id + 1, SUM(val) OVER (ORDER BY id ROWS BETWEEN 1 PRECEDING AND CURRENT ROW)
+    FROM win WHERE id < 3
+)
+SELECT * FROM win
+-- !query analysis
+org.apache.spark.sql.catalyst.ExtendedAnalysisException
+{
+  "errorClass" : "INVALID_RECURSIVE_REFERENCE.PLACE",
+  "sqlState" : "42836"
+}
+
+
+-- !query
+WITH RECURSIVE t1(n) AS (
+    SELECT 1
+    UNION ALL
+    (SELECT n + 1 FROM t1 WHERE n < 5 ORDER BY n)
+)
+SELECT * FROM t1
+-- !query analysis
+org.apache.spark.sql.catalyst.ExtendedAnalysisException
+{
+  "errorClass" : "INVALID_RECURSIVE_REFERENCE.PLACE",
+  "sqlState" : "42836"
+}

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

@@ -725,4 +725,20 @@ WITH RECURSIVE t1(n) AS (
     UNION ALL
     SELECT n + 1 FROM t1
 )
-    ((SELECT n FROM t1) UNION ALL (SELECT n FROM t1)) LIMIT 20
+    ((SELECT n FROM t1) UNION ALL (SELECT n FROM t1)) LIMIT 20;
+
+-- Recursive CTE with self reference inside Window function
+WITH RECURSIVE win(id, val) AS (
+    SELECT 1, CAST(10 AS BIGINT)
+    UNION ALL
+    SELECT id + 1, SUM(val) OVER (ORDER BY id ROWS BETWEEN 1 PRECEDING AND CURRENT ROW)
+    FROM win WHERE id < 3
+)
+SELECT * FROM win;
+
+WITH RECURSIVE t1(n) AS (
+    SELECT 1
+    UNION ALL
+    (SELECT n + 1 FROM t1 WHERE n < 5 ORDER BY n)
+)
+SELECT * FROM t1;

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

@@ -1804,3 +1804,38 @@ struct<n:int>
 7
 8
 9
+
+
+-- !query
+WITH RECURSIVE win(id, val) AS (
+    SELECT 1, CAST(10 AS BIGINT)
+    UNION ALL
+    SELECT id + 1, SUM(val) OVER (ORDER BY id ROWS BETWEEN 1 PRECEDING AND CURRENT ROW)
+    FROM win WHERE id < 3
+)
+SELECT * FROM win
+-- !query schema
+struct<>
+-- !query output
+org.apache.spark.sql.catalyst.ExtendedAnalysisException
+{
+  "errorClass" : "INVALID_RECURSIVE_REFERENCE.PLACE",
+  "sqlState" : "42836"
+}
+
+
+-- !query
+WITH RECURSIVE t1(n) AS (
+    SELECT 1
+    UNION ALL
+    (SELECT n + 1 FROM t1 WHERE n < 5 ORDER BY n)
+)
+SELECT * FROM t1
+-- !query schema
+struct<>
+-- !query output
+org.apache.spark.sql.catalyst.ExtendedAnalysisException
+{
+  "errorClass" : "INVALID_RECURSIVE_REFERENCE.PLACE",
+  "sqlState" : "42836"
+}