support recursive CTEs logical plans in datafusion-proto (#13314)

* support LogicaPlan::RecursiveQuery in datafusion-proto

* fixed and failing test roundtrip_recursive_query

* fix rebase artifact

* add node for CteWorkTableScan in datafusion-proto

* Use Arc::clone

---------

Co-authored-by: jonahgao <[email protected]>

Author: leoyvens

datafusion/core/src/datasource/cte_worktable.rs

@@ -39,8 +39,6 @@ use crate::datasource::{TableProvider, TableType};
 #[derive(Debug)]
 pub struct CteWorkTable {
     /// The name of the CTE work table
-    // WIP, see https://github.com/apache/datafusion/issues/462
-    #[allow(dead_code)]
     name: String,
     /// This schema must be shared across both the static and recursive terms of a recursive query
     table_schema: SchemaRef,
@@ -56,6 +54,16 @@ impl CteWorkTable {
             table_schema,
         }
     }
+
+    /// The user-provided name of the CTE
+    pub fn name(&self) -> &str {
+        &self.name
+    }
+
+    /// The schema of the recursive term of the query
+    pub fn schema(&self) -> SchemaRef {
+        Arc::clone(&self.table_schema)
+    }
 }
 
 #[async_trait]

datafusion/proto/proto/datafusion.proto

@@ -59,6 +59,8 @@ message LogicalPlanNode {
     DistinctOnNode distinct_on = 28;
     CopyToNode copy_to = 29;
     UnnestNode unnest = 30;
+    RecursiveQueryNode recursive_query = 31;
+    CteWorkTableScanNode cte_work_table_scan = 32;
   }
 }
 
@@ -1249,3 +1251,15 @@ message PartitionStats {
   int64 num_bytes = 3;
   repeated datafusion_common.ColumnStats column_stats = 4;
 }
+
+message RecursiveQueryNode {
+  string name = 1;
+  LogicalPlanNode static_term = 2;
+  LogicalPlanNode recursive_term = 3;
+  bool is_distinct = 4;
+}
+
+message CteWorkTableScanNode {
+    string name = 1;
+    datafusion_common.Schema schema = 2;
+}