refactor(proto): delegate deprecated ProjectionExec serde shims to new hooks

datafusion/proto/src/physical_plan/mod.rs

@@ -85,7 +85,7 @@ use datafusion_physical_plan::limit::{GlobalLimitExec, LocalLimitExec};
 use datafusion_physical_plan::memory::LazyMemoryExec;
 use datafusion_physical_plan::metrics::MetricCategory;
 use datafusion_physical_plan::placeholder_row::PlaceholderRowExec;
-use datafusion_physical_plan::projection::{ProjectionExec, ProjectionExpr};
+use datafusion_physical_plan::projection::ProjectionExec;
 use datafusion_physical_plan::proto::{
     ExecutionPlanDecode, ExecutionPlanDecodeCtx, ExecutionPlanEncode,
     ExecutionPlanEncodeCtx,
@@ -1188,28 +1188,21 @@ pub trait PhysicalPlanNodeExt: Sized {
         ctx: &PhysicalPlanDecodeContext<'_>,
         proto_converter: &dyn PhysicalProtoConverterExtension,
     ) -> Result<Arc<dyn ExecutionPlan>> {
-        let input: Arc<dyn ExecutionPlan> =
-            into_physical_plan(&projection.input, ctx, proto_converter)?;
-        let exprs = projection
-            .expr
-            .iter()
-            .zip(projection.expr_name.iter())
-            .map(|(expr, name)| {
-                Ok((
-                    proto_converter.proto_to_physical_expr(
-                        expr,
-                        input.schema().as_ref(),
-                        ctx,
-                    )?,
-                    name.to_string(),
-                ))
-            })
-            .collect::<Result<Vec<(Arc<dyn PhysicalExpr>, String)>>>()?;
-        let proj_exprs: Vec<ProjectionExpr> = exprs
-            .into_iter()
-            .map(|(expr, alias)| ProjectionExpr { expr, alias })
-            .collect();
-        Ok(Arc::new(ProjectionExec::try_new(proj_exprs, input)?))
+        let decoder = ConverterPlanDecoder {
+            ctx,
+            proto_converter,
+        };
+        let decode_ctx = ExecutionPlanDecodeCtx::new(&decoder);
+        // `try_from_proto` takes the enclosing `PhysicalPlanNode`, while this
+        // deprecated method is driven by the `ProjectionExecNode` argument.
+        // Re-wrap the argument so the decoded plan keeps depending on it rather
+        // than on `self`, which a caller may not have kept in sync.
+        let node = protobuf::PhysicalPlanNode {
+            physical_plan_type: Some(PhysicalPlanType::Projection(Box::new(
+                projection.clone(),
+            ))),
+        };
+        ProjectionExec::try_from_proto(&node, &decode_ctx)
     }
 
     fn try_into_filter_physical_plan(
@@ -2898,31 +2891,13 @@ pub trait PhysicalPlanNodeExt: Sized {
         codec: &dyn PhysicalExtensionCodec,
         proto_converter: &dyn PhysicalProtoConverterExtension,
     ) -> Result<protobuf::PhysicalPlanNode> {
-        let input = protobuf::PhysicalPlanNode::try_from_physical_plan_with_converter(
-            exec.input().to_owned(),
+        let encoder = ConverterPlanEncoder {
             codec,
             proto_converter,
-        )?;
-        let expr = exec
-            .expr()
-            .iter()
-            .map(|proj_expr| {
-                proto_converter.physical_expr_to_proto(&proj_expr.expr, codec)
-            })
-            .collect::<Result<Vec<_>>>()?;
-        let expr_name = exec
-            .expr()
-            .iter()
-            .map(|proj_expr| proj_expr.alias.clone())
-            .collect();
-        Ok(protobuf::PhysicalPlanNode {
-            physical_plan_type: Some(PhysicalPlanType::Projection(Box::new(
-                protobuf::ProjectionExecNode {
-                    input: Some(Box::new(input)),
-                    expr,
-                    expr_name,
-                },
-            ))),
+        };
+        let ctx = ExecutionPlanEncodeCtx::new(&encoder);
+        exec.try_to_proto(&ctx)?.ok_or_else(|| {
+            internal_datafusion_err!("ProjectionExec::try_to_proto returned None")
         })
     }
 

datafusion/proto/tests/cases/roundtrip_physical_plan.rs

@@ -2392,6 +2392,64 @@ async fn roundtrip_physical_plan_node() {
     let _ = plan.execute(0, ctx.task_ctx()).unwrap();
 }
 
+/// The deprecated `try_into_projection_physical_plan` shim now delegates to
+/// [`ProjectionExec::try_from_proto`], which reads the enclosing
+/// `PhysicalPlanNode` rather than a `ProjectionExecNode`. Assert the shim still
+/// decodes the node passed as an argument, not `self`, so an out-of-tree caller
+/// that passes a projection unrelated to `self` keeps the old behaviour.
+#[test]
+fn deprecated_projection_shim_decodes_argument_not_self() -> Result<()> {
+    use datafusion_proto::protobuf::PhysicalPlanNode;
+    use datafusion_proto::protobuf::physical_plan_node::PhysicalPlanType;
+
+    let schema = Schema::new(vec![Field::new("a", DataType::Int32, false)]);
+    let input = Arc::new(EmptyExec::new(Arc::new(schema.clone())));
+    let projection = Arc::new(ProjectionExec::try_new(
+        vec![ProjectionExpr::new(
+            col("a", &schema)?,
+            "renamed".to_string(),
+        )],
+        input,
+    )?);
+
+    let codec = DefaultPhysicalExtensionCodec {};
+    let proto_converter = DefaultPhysicalProtoConverter {};
+    let projection_node = PhysicalPlanNode::try_from_physical_plan_with_converter(
+        projection,
+        &codec,
+        &proto_converter,
+    )?;
+    let Some(PhysicalPlanType::Projection(projection_exec_node)) =
+        &projection_node.physical_plan_type
+    else {
+        panic!("expected a Projection node, got {projection_node:?}");
+    };
+
+    // `self` is deliberately a different plan variant than the argument.
+    let unrelated_node = PhysicalPlanNode::try_from_physical_plan_with_converter(
+        Arc::new(EmptyExec::new(Arc::new(schema))),
+        &codec,
+        &proto_converter,
+    )?;
+
+    let session_ctx = SessionContext::new();
+    let task_ctx = session_ctx.task_ctx();
+    let decode_ctx = PhysicalPlanDecodeContext::new(task_ctx.as_ref(), &codec);
+    #[allow(deprecated)]
+    let decoded = unrelated_node.try_into_projection_physical_plan(
+        projection_exec_node,
+        &decode_ctx,
+        &proto_converter,
+    )?;
+
+    let decoded = decoded
+        .downcast_ref::<ProjectionExec>()
+        .expect("decoded plan should be a ProjectionExec");
+    assert_eq!(decoded.expr().len(), 1);
+    assert_eq!(decoded.expr()[0].alias, "renamed");
+    Ok(())
+}
+
 /// Helper function to create a SessionContext with all TPC-H tables registered as external tables
 async fn tpch_context() -> Result<SessionContext> {
     use datafusion_common::test_util::datafusion_test_data;