Add ExecutionPlan try_to_proto / try_from_proto hooks + ProjectionExec reference

datafusion/physical-plan/src/execution_plan.rs

@@ -757,6 +757,27 @@ pub trait ExecutionPlan: Any + Debug + DisplayAs + Send + Sync {
     ) -> Option<Arc<dyn ExecutionPlan>> {
         None
     }
+
+    /// Serialize this plan to its protobuf representation, if it knows how.
+    ///
+    /// This is the `ExecutionPlan` analog of
+    /// [`PhysicalExpr::try_to_proto`](datafusion_physical_expr_common::physical_expr::PhysicalExpr::try_to_proto).
+    ///
+    /// * `Ok(None)` (the default) — "I don't serialize myself"; the caller
+    ///   (`datafusion-proto`) falls back to the central downcast chain. Every
+    ///   un-migrated plan keeps its existing behavior.
+    /// * `Ok(Some(node))` — fully serialized; the caller must not fall back.
+    /// * `Err(_)` — a real failure (e.g. a child failed to serialize).
+    ///
+    /// Only *self-contained* plans should override this — see [`crate::proto`]
+    /// for the session-dependency boundary.
+    #[cfg(feature = "proto")]
+    fn try_to_proto(
+        &self,
+        _ctx: &crate::proto::ExecutionPlanEncodeCtx<'_>,
+    ) -> Result<Option<datafusion_proto_models::protobuf::PhysicalPlanNode>> {
+        Ok(None)
+    }
 }
 
 impl dyn ExecutionPlan {

datafusion/physical-plan/src/lib.rs

@@ -84,6 +84,8 @@ pub mod metrics;
 pub mod operator_statistics;
 pub mod placeholder_row;
 pub mod projection;
+#[cfg(feature = "proto")]
+pub mod proto;
 pub mod recursive_query;
 pub mod repartition;
 pub mod scalar_subquery;

datafusion/physical-plan/src/projection.rs

@@ -494,6 +494,71 @@ impl ExecutionPlan for ProjectionExec {
                     .ok()
             })
     }
+
+    #[cfg(feature = "proto")]
+    fn try_to_proto(
+        &self,
+        ctx: &crate::proto::ExecutionPlanEncodeCtx<'_>,
+    ) -> Result<Option<datafusion_proto_models::protobuf::PhysicalPlanNode>> {
+        use datafusion_proto_models::protobuf;
+        let input = ctx.encode_child(self.input())?;
+        let expr = ctx.encode_expressions(self.expr().iter().map(|p| &p.expr))?;
+        let expr_name = self.expr().iter().map(|p| p.alias.clone()).collect();
+        Ok(Some(protobuf::PhysicalPlanNode {
+            physical_plan_type: Some(
+                protobuf::physical_plan_node::PhysicalPlanType::Projection(Box::new(
+                    protobuf::ProjectionExecNode {
+                        input: Some(Box::new(input)),
+                        expr,
+                        expr_name,
+                    },
+                )),
+            ),
+        }))
+    }
+}
+
+#[cfg(feature = "proto")]
+impl ProjectionExec {
+    /// Reconstruct a [`ProjectionExec`] from its protobuf representation.
+    ///
+    /// The exact inverse of [`ExecutionPlan::try_to_proto`]: it takes the whole
+    /// [`PhysicalPlanNode`] so every plan's `try_from_proto` shares one
+    /// signature. Child plans and expressions are decoded recursively via the
+    /// [`ExecutionPlanDecodeCtx`].
+    ///
+    /// [`PhysicalPlanNode`]: datafusion_proto_models::protobuf::PhysicalPlanNode
+    /// [`ExecutionPlan::try_to_proto`]: crate::ExecutionPlan::try_to_proto
+    /// [`ExecutionPlanDecodeCtx`]: crate::proto::ExecutionPlanDecodeCtx
+    pub fn try_from_proto(
+        node: &datafusion_proto_models::protobuf::PhysicalPlanNode,
+        ctx: &crate::proto::ExecutionPlanDecodeCtx<'_>,
+    ) -> Result<Arc<dyn ExecutionPlan>> {
+        use datafusion_proto_models::protobuf;
+        let projection = crate::expect_plan_variant!(
+            node,
+            protobuf::physical_plan_node::PhysicalPlanType::Projection,
+            "ProjectionExec",
+        );
+        let input = ctx.decode_required_child(
+            projection.input.as_deref(),
+            "ProjectionExec",
+            "input",
+        )?;
+        let input_schema = input.schema();
+        let exprs = projection
+            .expr
+            .iter()
+            .zip(projection.expr_name.iter())
+            .map(|(expr, name)| {
+                Ok(ProjectionExpr {
+                    expr: ctx.decode_expr(expr, input_schema.as_ref())?,
+                    alias: name.to_string(),
+                })
+            })
+            .collect::<Result<Vec<ProjectionExpr>>>()?;
+        Ok(Arc::new(ProjectionExec::try_new(exprs, input)?))
+    }
 }
 
 impl ProjectionStream {