support recursive CTEs logical plans in datafusion-proto (apache#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 <jonahgao@msn.com>

Author: 2 people

datafusion/core/src/datasource/cte_worktable.rs

@@ -38,8 +38,6 @@ use crate::datasource::{TableProvider, TableType};
 /// See here for more details: www.postgresql.org/docs/11/queries-with.html#id-1.5.6.12.5.4
 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,
@@ -55,6 +53,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-common/src/generated/pbjson.rs

@@ -45,10 +45,10 @@ pub struct ParquetFormat {
     pub options: ::core::option::Option<TableParquetOptions>,
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
-#[derive(Clone, PartialEq, ::prost::Message)]
+#[derive(Clone, Copy, PartialEq, ::prost::Message)]
 pub struct AvroFormat {}
 #[allow(clippy::derive_partial_eq_without_eq)]
-#[derive(Clone, PartialEq, ::prost::Message)]
+#[derive(Clone, Copy, PartialEq, ::prost::Message)]
 pub struct NdJsonFormat {
     #[prost(message, optional, tag = "1")]
     pub options: ::core::option::Option<JsonOptions>,
@@ -89,10 +89,10 @@ pub struct Constraints {
     pub constraints: ::prost::alloc::vec::Vec<Constraint>,
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
-#[derive(Clone, PartialEq, ::prost::Message)]
+#[derive(Clone, Copy, PartialEq, ::prost::Message)]
 pub struct AvroOptions {}
 #[allow(clippy::derive_partial_eq_without_eq)]
-#[derive(Clone, PartialEq, ::prost::Message)]
+#[derive(Clone, Copy, PartialEq, ::prost::Message)]
 pub struct ArrowOptions {}
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
@@ -137,15 +137,15 @@ pub struct Timestamp {
     pub timezone: ::prost::alloc::string::String,
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
-#[derive(Clone, PartialEq, ::prost::Message)]
+#[derive(Clone, Copy, PartialEq, ::prost::Message)]
 pub struct Decimal {
     #[prost(uint32, tag = "3")]
     pub precision: u32,
     #[prost(int32, tag = "4")]
     pub scale: i32,
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
-#[derive(Clone, PartialEq, ::prost::Message)]
+#[derive(Clone, Copy, PartialEq, ::prost::Message)]
 pub struct Decimal256Type {
     #[prost(uint32, tag = "3")]
     pub precision: u32,
@@ -223,15 +223,15 @@ pub mod scalar_nested_value {
     }
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
-#[derive(Clone, PartialEq, ::prost::Message)]
+#[derive(Clone, Copy, PartialEq, ::prost::Message)]
 pub struct ScalarTime32Value {
     #[prost(oneof = "scalar_time32_value::Value", tags = "1, 2")]
     pub value: ::core::option::Option<scalar_time32_value::Value>,
 }
 /// Nested message and enum types in `ScalarTime32Value`.
 pub mod scalar_time32_value {
     #[allow(clippy::derive_partial_eq_without_eq)]
-    #[derive(Clone, PartialEq, ::prost::Oneof)]
+    #[derive(Clone, Copy, PartialEq, ::prost::Oneof)]
     pub enum Value {
         #[prost(int32, tag = "1")]
         Time32SecondValue(i32),
@@ -240,15 +240,15 @@ pub mod scalar_time32_value {
     }
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
-#[derive(Clone, PartialEq, ::prost::Message)]
+#[derive(Clone, Copy, PartialEq, ::prost::Message)]
 pub struct ScalarTime64Value {
     #[prost(oneof = "scalar_time64_value::Value", tags = "1, 2")]
     pub value: ::core::option::Option<scalar_time64_value::Value>,
 }
 /// Nested message and enum types in `ScalarTime64Value`.
 pub mod scalar_time64_value {
     #[allow(clippy::derive_partial_eq_without_eq)]
-    #[derive(Clone, PartialEq, ::prost::Oneof)]
+    #[derive(Clone, Copy, PartialEq, ::prost::Oneof)]
     pub enum Value {
         #[prost(int64, tag = "1")]
         Time64MicrosecondValue(i64),
@@ -267,7 +267,7 @@ pub struct ScalarTimestampValue {
 /// Nested message and enum types in `ScalarTimestampValue`.
 pub mod scalar_timestamp_value {
     #[allow(clippy::derive_partial_eq_without_eq)]
-    #[derive(Clone, PartialEq, ::prost::Oneof)]
+    #[derive(Clone, Copy, PartialEq, ::prost::Oneof)]
     pub enum Value {
         #[prost(int64, tag = "1")]
         TimeMicrosecondValue(i64),
@@ -288,15 +288,15 @@ pub struct ScalarDictionaryValue {
     pub value: ::core::option::Option<::prost::alloc::boxed::Box<ScalarValue>>,
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
-#[derive(Clone, PartialEq, ::prost::Message)]
+#[derive(Clone, Copy, PartialEq, ::prost::Message)]
 pub struct IntervalDayTimeValue {
     #[prost(int32, tag = "1")]
     pub days: i32,
     #[prost(int32, tag = "2")]
     pub milliseconds: i32,
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
-#[derive(Clone, PartialEq, ::prost::Message)]
+#[derive(Clone, Copy, PartialEq, ::prost::Message)]
 pub struct IntervalMonthDayNanoValue {
     #[prost(int32, tag = "1")]
     pub months: i32,
@@ -558,10 +558,10 @@ pub mod arrow_type {
 ///    }
 /// }
 #[allow(clippy::derive_partial_eq_without_eq)]
-#[derive(Clone, PartialEq, ::prost::Message)]
+#[derive(Clone, Copy, PartialEq, ::prost::Message)]
 pub struct EmptyMessage {}
 #[allow(clippy::derive_partial_eq_without_eq)]
-#[derive(Clone, PartialEq, ::prost::Message)]
+#[derive(Clone, Copy, PartialEq, ::prost::Message)]
 pub struct JsonWriterOptions {
     #[prost(enumeration = "CompressionTypeVariant", tag = "1")]
     pub compression: i32,
@@ -658,7 +658,7 @@ pub struct CsvOptions {
 }
 /// Options controlling CSV format
 #[allow(clippy::derive_partial_eq_without_eq)]
-#[derive(Clone, PartialEq, ::prost::Message)]
+#[derive(Clone, Copy, PartialEq, ::prost::Message)]
 pub struct JsonOptions {
     /// Compression type
     #[prost(enumeration = "CompressionTypeVariant", tag = "1")]
@@ -723,7 +723,7 @@ pub struct ParquetColumnOptions {
 /// Nested message and enum types in `ParquetColumnOptions`.
 pub mod parquet_column_options {
     #[allow(clippy::derive_partial_eq_without_eq)]
-    #[derive(Clone, PartialEq, ::prost::Oneof)]
+    #[derive(Clone, Copy, PartialEq, ::prost::Oneof)]
     pub enum BloomFilterEnabledOpt {
         #[prost(bool, tag = "1")]
         BloomFilterEnabled(bool),
@@ -735,7 +735,7 @@ pub mod parquet_column_options {
         Encoding(::prost::alloc::string::String),
     }
     #[allow(clippy::derive_partial_eq_without_eq)]
-    #[derive(Clone, PartialEq, ::prost::Oneof)]
+    #[derive(Clone, Copy, PartialEq, ::prost::Oneof)]
     pub enum DictionaryEnabledOpt {
         #[prost(bool, tag = "3")]
         DictionaryEnabled(bool),
@@ -753,19 +753,19 @@ pub mod parquet_column_options {
         StatisticsEnabled(::prost::alloc::string::String),
     }
     #[allow(clippy::derive_partial_eq_without_eq)]
-    #[derive(Clone, PartialEq, ::prost::Oneof)]
+    #[derive(Clone, Copy, PartialEq, ::prost::Oneof)]
     pub enum BloomFilterFppOpt {
         #[prost(double, tag = "6")]
         BloomFilterFpp(f64),
     }
     #[allow(clippy::derive_partial_eq_without_eq)]
-    #[derive(Clone, PartialEq, ::prost::Oneof)]
+    #[derive(Clone, Copy, PartialEq, ::prost::Oneof)]
     pub enum BloomFilterNdvOpt {
         #[prost(uint64, tag = "7")]
         BloomFilterNdv(u64),
     }
     #[allow(clippy::derive_partial_eq_without_eq)]
-    #[derive(Clone, PartialEq, ::prost::Oneof)]
+    #[derive(Clone, Copy, PartialEq, ::prost::Oneof)]
     pub enum MaxStatisticsSizeOpt {
         #[prost(uint32, tag = "8")]
         MaxStatisticsSize(u32),
@@ -860,7 +860,7 @@ pub struct ParquetOptions {
 /// Nested message and enum types in `ParquetOptions`.
 pub mod parquet_options {
     #[allow(clippy::derive_partial_eq_without_eq)]
-    #[derive(Clone, PartialEq, ::prost::Oneof)]
+    #[derive(Clone, Copy, PartialEq, ::prost::Oneof)]
     pub enum MetadataSizeHintOpt {
         #[prost(uint64, tag = "4")]
         MetadataSizeHint(u64),
@@ -872,7 +872,7 @@ pub mod parquet_options {
         Compression(::prost::alloc::string::String),
     }
     #[allow(clippy::derive_partial_eq_without_eq)]
-    #[derive(Clone, PartialEq, ::prost::Oneof)]
+    #[derive(Clone, Copy, PartialEq, ::prost::Oneof)]
     pub enum DictionaryEnabledOpt {
         #[prost(bool, tag = "11")]
         DictionaryEnabled(bool),
@@ -884,13 +884,13 @@ pub mod parquet_options {
         StatisticsEnabled(::prost::alloc::string::String),
     }
     #[allow(clippy::derive_partial_eq_without_eq)]
-    #[derive(Clone, PartialEq, ::prost::Oneof)]
+    #[derive(Clone, Copy, PartialEq, ::prost::Oneof)]
     pub enum MaxStatisticsSizeOpt {
         #[prost(uint64, tag = "14")]
         MaxStatisticsSize(u64),
     }
     #[allow(clippy::derive_partial_eq_without_eq)]
-    #[derive(Clone, PartialEq, ::prost::Oneof)]
+    #[derive(Clone, Copy, PartialEq, ::prost::Oneof)]
     pub enum ColumnIndexTruncateLengthOpt {
         #[prost(uint64, tag = "17")]
         ColumnIndexTruncateLength(u64),
@@ -902,13 +902,13 @@ pub mod parquet_options {
         Encoding(::prost::alloc::string::String),
     }
     #[allow(clippy::derive_partial_eq_without_eq)]
-    #[derive(Clone, PartialEq, ::prost::Oneof)]
+    #[derive(Clone, Copy, PartialEq, ::prost::Oneof)]
     pub enum BloomFilterFppOpt {
         #[prost(double, tag = "21")]
         BloomFilterFpp(f64),
     }
     #[allow(clippy::derive_partial_eq_without_eq)]
-    #[derive(Clone, PartialEq, ::prost::Oneof)]
+    #[derive(Clone, Copy, PartialEq, ::prost::Oneof)]
     pub enum BloomFilterNdvOpt {
         #[prost(uint64, tag = "22")]
         BloomFilterNdv(u64),

datafusion/proto-common/src/generated/prost.rs

@@ -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;
   }
 }
 
@@ -1214,3 +1216,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;
+}