Implement nested recursive CTEs (#18956)

It gives a name (the table name) to each `WorkTable`. This way
`WorkTableExec` can recognize its own `WorkTable`

Note that it doesn't allow multiple occurrences of the same CTE name:
it's not possible to implement things like "join with itself" correctly
with only the work table.

## Which issue does this PR close?

- Closes #18955.

## Rationale for this change

Support nested recursive CTEs without co-recursion. This is useful to
e.g. implement SPARQL or other graph query languages.

## What changes are included in this PR?

## Are these changes tested?

Yes! There is a nested recursive query in the test file

## Are there any user-facing changes?

Nested recursive queries are now allowed instead of failing with a "not
implemented" error

Author: Tpt

datafusion/physical-plan/src/recursive_query.rs

@@ -86,7 +86,7 @@ impl RecursiveQueryExec {
         is_distinct: bool,
     ) -> Result<Self> {
         // Each recursive query needs its own work table
-        let work_table = Arc::new(WorkTable::new());
+        let work_table = Arc::new(WorkTable::new(name.clone()));
         // Use the same work table for both the WorkTableExec and the recursive term
         let recursive_term = assign_work_table(recursive_term, &work_table)?;
         let cache = Self::compute_properties(static_term.schema());
@@ -380,8 +380,6 @@ fn assign_work_table(
                 work_table_refs += 1;
                 Ok(Transformed::yes(new_plan))
             }
-        } else if plan.as_any().is::<RecursiveQueryExec>() {
-            not_impl_err!("Recursive queries cannot be nested")
         } else {
             Ok(Transformed::no(plan))
         }

datafusion/physical-plan/src/work_table.rs

@@ -58,13 +58,15 @@ impl ReservedBatches {
 #[derive(Debug)]
 pub struct WorkTable {
     batches: Mutex<Option<ReservedBatches>>,
+    name: String,
 }
 
 impl WorkTable {
     /// Create a new work table.
-    pub(super) fn new() -> Self {
+    pub(super) fn new(name: String) -> Self {
         Self {
             batches: Mutex::new(None),
+            name,
         }
     }
 
@@ -113,10 +115,10 @@ impl WorkTableExec {
     pub fn new(name: String, schema: SchemaRef) -> Self {
         let cache = Self::compute_properties(Arc::clone(&schema));
         Self {
-            name,
+            name: name.clone(),
             schema,
             metrics: ExecutionPlanMetricsSet::new(),
-            work_table: Arc::new(WorkTable::new()),
+            work_table: Arc::new(WorkTable::new(name)),
             cache,
         }
     }
@@ -230,6 +232,10 @@ impl ExecutionPlan for WorkTableExec {
         // Down-cast to the expected state type; propagate `None` on failure
         let work_table = state.downcast::<WorkTable>().ok()?;
 
+        if work_table.name != self.name {
+            return None; // Different table
+        }
+
         Some(Arc::new(Self {
             name: self.name.clone(),
             schema: Arc::clone(&self.schema),
@@ -248,7 +254,7 @@ mod tests {
 
     #[test]
     fn test_work_table() {
-        let work_table = WorkTable::new();
+        let work_table = WorkTable::new("test".into());
         // Can't take from empty work_table
         assert!(work_table.take().is_err());
 

datafusion/sqllogictest/test_files/cte.slt

@@ -125,6 +125,25 @@ SELECT * FROM nodes
 3
 4
 
+# deduplicating recursive CTE with two variables works
+query II
+WITH RECURSIVE ranges AS (
+    SELECT min, max from (VALUES (1, 1), (2, 2)) ranges(min, max)
+    UNION
+    SELECT min, max + 1 as max
+    FROM ranges
+    WHERE max < 4
+)
+SELECT * FROM ranges
+----
+1 1
+2 2
+1 2
+2 3
+1 3
+2 4
+1 4
+
 # setup
 statement ok
 CREATE EXTERNAL TABLE balance STORED as CSV LOCATION '../core/tests/data/recursive_cte/balance.csv' OPTIONS ('format.has_header' 'true');
@@ -647,21 +666,51 @@ ORDER BY
 3 1400 1
 1 2700 2
 
-#expect error from recursive CTE with nested recursive terms
-query error DataFusion error: This feature is not implemented: Recursive queries cannot be nested
+#nested recursive ctes
+query I
 WITH RECURSIVE outer_cte AS (
     SELECT 1 as a
     UNION ALL (
-        WITH  RECURSIVE nested_cte AS (
+        WITH RECURSIVE nested_cte AS (
            SELECT 1 as a
            UNION ALL
-           SELECT a+2 as a
-	   FROM nested_cte where a < 3
-         )
-    SELECT outer_cte.a +2
-    FROM outer_cte JOIN nested_cte USING(a)
-    WHERE nested_cte.a < 4
-   )
+           SELECT a + 2 as a
+           FROM nested_cte where a < 3
+        )
+        SELECT outer_cte.a + 2 as a
+        FROM outer_cte JOIN nested_cte USING(a)
+        WHERE nested_cte.a < 4
+    )
+)
+SELECT a FROM outer_cte;
+----
+1
+3
+5
+
+# Check that CTE name shadowing is returning an error
+query error DataFusion error: Error during planning: WITH query name "outer_cte" specified more than once
+WITH RECURSIVE outer_cte AS (
+    SELECT 1 as a
+    UNION ALL (
+        WITH RECURSIVE nested_cte AS (
+           SELECT 1 as a
+           UNION ALL (
+               WITH RECURSIVE outer_cte AS (
+                    SELECT 1 as a
+                    UNION ALL
+                    SELECT a + 2 as a
+                    FROM outer_cte where a < 3
+               )
+               SELECT nested_cte.a + outer_cte.a as a
+               FROM nested_cte JOIN outer_cte USING(a)
+               WHERE outer_cte_cte.a < 8
+           )
+        )
+        SELECT outer_cte.a + nested_cte.a as a
+        FROM outer_cte JOIN nested_cte USING(a)
+        WHERE nested_cte.a < 8
+    )
 )
 SELECT a FROM outer_cte;