Implement protobuf serialization for AnalyzeExec

datafusion/physical-plan/src/analyze.rs

@@ -62,6 +62,21 @@ impl AnalyzeExec {
             schema,
         }
     }
+
+    /// access to verbose
+    pub fn verbose(&self) -> bool {
+        self.verbose
+    }
+
+    /// access to show_statistics
+    pub fn show_statistics(&self) -> bool {
+        self.show_statistics
+    }
+
+    /// The input plan
+    pub fn input(&self) -> &Arc<dyn ExecutionPlan> {
+        &self.input
+    }
 }
 
 impl DisplayAs for AnalyzeExec {

datafusion/proto/proto/datafusion.proto

@@ -1110,6 +1110,7 @@ message PhysicalPlanNode {
     ExplainExecNode explain = 20;
     SortPreservingMergeExecNode sort_preserving_merge = 21;
     NestedLoopJoinExecNode nested_loop_join = 22;
+    AnalyzeExecNode analyze = 23;
   }
 }
 
@@ -1345,6 +1346,13 @@ message ExplainExecNode {
   bool verbose = 3;
 }
 
+message AnalyzeExecNode {
+  bool verbose = 1;
+  bool show_statistics = 2;
+  PhysicalPlanNode input = 3;
+  Schema schema = 4;
+}
+
 message CrossJoinExecNode {
   PhysicalPlanNode left = 1;
   PhysicalPlanNode right = 2;

datafusion/proto/src/physical_plan/mod.rs

@@ -27,6 +27,7 @@ use datafusion::execution::FunctionRegistry;
 use datafusion::logical_expr::WindowFrame;
 use datafusion::physical_plan::aggregates::{create_aggregate_expr, AggregateMode};
 use datafusion::physical_plan::aggregates::{AggregateExec, PhysicalGroupBy};
+use datafusion::physical_plan::analyze::AnalyzeExec;
 use datafusion::physical_plan::coalesce_batches::CoalesceBatchesExec;
 use datafusion::physical_plan::coalesce_partitions::CoalescePartitionsExec;
 use datafusion::physical_plan::empty::EmptyExec;
@@ -778,6 +779,20 @@ impl AsExecutionPlan for PhysicalPlanNode {
                     &join_type.into(),
                 )?))
             }
+            PhysicalPlanType::Analyze(analyze) => {
+                let input: Arc<dyn ExecutionPlan> = into_physical_plan(
+                    &analyze.input,
+                    registry,
+                    runtime,
+                    extension_codec,
+                )?;
+                Ok(Arc::new(AnalyzeExec::new(
+                    analyze.verbose,
+                    analyze.show_statistics,
+                    input,
+                    Arc::new(analyze.schema.as_ref().unwrap().try_into()?),
+                )))
+            }
         }
     }
 
@@ -825,6 +840,21 @@ impl AsExecutionPlan for PhysicalPlanNode {
                     },
                 ))),
             })
+        } else if let Some(exec) = plan.downcast_ref::<AnalyzeExec>() {
+            let input = protobuf::PhysicalPlanNode::try_from_physical_plan(
+                exec.input().to_owned(),
+                extension_codec,
+            )?;
+            Ok(protobuf::PhysicalPlanNode {
+                physical_plan_type: Some(PhysicalPlanType::Analyze(Box::new(
+                    protobuf::AnalyzeExecNode {
+                        verbose: exec.verbose(),
+                        show_statistics: exec.show_statistics(),
+                        input: Some(Box::new(input)),
+                        schema: Some(exec.schema().as_ref().try_into()?),
+                    },
+                ))),
+            })
         } else if let Some(exec) = plan.downcast_ref::<FilterExec>() {
             let input = protobuf::PhysicalPlanNode::try_from_physical_plan(
                 exec.input().to_owned(),
@@ -1391,6 +1421,7 @@ mod roundtrip_tests {
     use datafusion::physical_expr::expressions::{cast, in_list};
     use datafusion::physical_expr::ScalarFunctionExpr;
     use datafusion::physical_plan::aggregates::PhysicalGroupBy;
+    use datafusion::physical_plan::analyze::AnalyzeExec;
     use datafusion::physical_plan::expressions::{like, BinaryExpr, GetIndexedFieldExpr};
     use datafusion::physical_plan::functions::make_scalar_function;
     use datafusion::physical_plan::projection::ProjectionExec;
@@ -1946,7 +1977,7 @@ mod roundtrip_tests {
         ];
 
         let schema = Schema::new(fields);
-        let input = Arc::new(EmptyExec::new(false, Arc::new(schema.clone())));
+        let input = Arc::new(EmptyExec::new(true, Arc::new(schema.clone())));
 
         let col_arg = col("arg", &schema)?;
         let col_key = col("key", &schema)?;
@@ -1993,4 +2024,19 @@ mod roundtrip_tests {
 
         roundtrip_test(plan)
     }
+
+    #[test]
+    fn rountrip_analyze() -> Result<()> {
+        let field_a = Field::new("plan_type", DataType::Utf8, false);
+        let field_b = Field::new("plan", DataType::Utf8, false);
+        let schema = Schema::new(vec![field_a, field_b]);
+        let input = Arc::new(EmptyExec::new(true, Arc::new(schema.clone())));
+
+        roundtrip_test(Arc::new(AnalyzeExec::new(
+            false,
+            false,
+            input,
+            Arc::new(schema),
+        )))
+    }
 }