Support encoding and decoding UnnestExec

datafusion/physical-expr/Cargo.toml

@@ -81,4 +81,4 @@ name = "in_list"
 
 [[bench]]
 harness = false
-name = "is_null"
+name = "is_null"

datafusion/physical-plan/src/unnest.rs

@@ -109,6 +109,25 @@ impl UnnestExec {
             input.execution_mode(),
         )
     }
+
+    /// Input execution plan
+    pub fn input(&self) -> &Arc<dyn ExecutionPlan> {
+        &self.input
+    }
+
+    /// indices of the list-typed columns in the input schema
+    pub fn list_column_indices(&self) -> &[usize] {
+        &self.list_column_indices
+    }
+
+    /// indices of the struct-typed columns in the input schema
+    pub fn struct_column_indices(&self) -> &[usize] {
+        &self.struct_column_indices
+    }
+
+    pub fn options(&self) -> &UnnestOptions {
+        &self.options
+    }
 }
 
 impl DisplayAs for UnnestExec {

datafusion/proto/proto/datafusion.proto

@@ -742,10 +742,11 @@ message PhysicalPlanNode {
     AnalyzeExecNode analyze = 23;
     JsonSinkExecNode json_sink = 24;
     SymmetricHashJoinExecNode symmetric_hash_join = 25;
-    InterleaveExecNode  interleave = 26;
+    InterleaveExecNode interleave = 26;
     PlaceholderRowExecNode placeholder_row = 27;
     CsvSinkExecNode csv_sink = 28;
     ParquetSinkExecNode parquet_sink = 29;
+    UnnestExecNode unnest = 30;
   }
 }
 
@@ -802,6 +803,14 @@ message ParquetSinkExecNode {
   PhysicalSortExprNodeCollection sort_order = 4;
 }
 
+message UnnestExecNode {
+  PhysicalPlanNode input = 1;
+  datafusion_common.Schema schema = 2;
+  repeated uint64 list_type_columns = 3;
+  repeated uint64 struct_type_columns = 4;
+  UnnestOptions options = 5;
+}
+
 message PhysicalExtensionNode {
   bytes node = 1;
   repeated PhysicalPlanNode inputs = 2;