error if recursive cte is referenced multiple times within the recursive term

Author: matthewgapp

datafusion/physical-plan/src/recursive_query.rs

@@ -304,11 +304,19 @@ fn assign_work_table(
     plan: Arc<dyn ExecutionPlan>,
     work_table: Arc<WorkTable>,
 ) -> Result<Arc<dyn ExecutionPlan>> {
-    plan.transform_down(&|plan| {
+    let mut work_table_refs = 0;
+    plan.transform_down_mut(&mut |plan| {
         if let Some(exec) = plan.as_any().downcast_ref::<WorkTableExec>() {
-            Ok(Transformed::Yes(Arc::new(
-                exec.with_work_table(work_table.clone()),
-            )))
+            if work_table_refs > 0 {
+                not_impl_err!(
+                    "Multiple recursive references to the same CTE are not supported"
+                )
+            } else {
+                work_table_refs += 1;
+                Ok(Transformed::Yes(Arc::new(
+                    exec.with_work_table(work_table.clone()),
+                )))
+            }
         } else if plan.as_any().is::<RecursiveQueryExec>() {
             not_impl_err!("Recursive queries cannot be nested")
         } else {

datafusion/sqllogictest/test_files/cte.slt

@@ -619,3 +619,14 @@ WITH RECURSIVE outer_cte AS (
    )
 )
 SELECT a FROM outer_cte;
+
+# expect error when recursive CTE is referenced multiple times in the recursive term
+query error DataFusion error: This feature is not implemented: Multiple recursive references to the same CTE are not supported
+WITH RECURSIVE my_cte AS (
+    SELECT 1 as a
+    UNION ALL
+    SELECT my_cte.a+2 as a
+    FROM my_cte join my_cte c2 using(a)
+    WHERE my_cte.a<5
+)
+SELECT a FROM my_cte;