From f1de51049c58527c8c37d299b0dcc3a67754598a Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Mon, 20 Jul 2026 18:36:51 -0500 Subject: [PATCH 1/2] refactor(proto): delegate deprecated ProjectionExec serde shims to new hooks The deprecated `try_from_projection_exec` / `try_into_projection_physical_plan` compatibility methods (kept from #23495) duplicated the wire-format logic that now lives in `ProjectionExec::try_to_proto` / `try_from_proto`. Replace their bodies with thin shims that build an `ExecutionPlanEncodeCtx` / `DecodeCtx` from the codec + converter and delegate to the hook, so the format is single-sourced. No wire-format or behavior change. Matches the delegate style used by FilterExec in #23708. Part of #23494. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01U9qX6kekbJGpmTSxvrU8e5 --- datafusion/proto/src/physical_plan/mod.rs | 60 +++++------------------ 1 file changed, 13 insertions(+), 47 deletions(-) diff --git a/datafusion/proto/src/physical_plan/mod.rs b/datafusion/proto/src/physical_plan/mod.rs index 4f72668813243..5105c0be39ac2 100644 --- a/datafusion/proto/src/physical_plan/mod.rs +++ b/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, @@ -1184,32 +1184,16 @@ pub trait PhysicalPlanNodeExt: Sized { )] fn try_into_projection_physical_plan( &self, - projection: &protobuf::ProjectionExecNode, + _projection: &protobuf::ProjectionExecNode, ctx: &PhysicalPlanDecodeContext<'_>, proto_converter: &dyn PhysicalProtoConverterExtension, ) -> Result> { - let input: Arc = - 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::, String)>>>()?; - let proj_exprs: Vec = 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); + ProjectionExec::try_from_proto(self.node(), &decode_ctx) } fn try_into_filter_physical_plan( @@ -2898,31 +2882,13 @@ pub trait PhysicalPlanNodeExt: Sized { codec: &dyn PhysicalExtensionCodec, proto_converter: &dyn PhysicalProtoConverterExtension, ) -> Result { - 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::>>()?; - 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") }) } From 581e7c8ea7b3d24fed04b4f9d05afbdcd0f80422 Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Tue, 21 Jul 2026 00:54:49 -0500 Subject: [PATCH 2/2] fix: keep deprecated projection decode shim driven by its argument `try_from_proto` reads the enclosing `PhysicalPlanNode`, so delegating with `self.node()` changed the deprecated shim's contract: an out-of-tree caller passing a `ProjectionExecNode` unrelated to `self` would get a wrong-variant error instead of the decoded argument. Re-wrap the argument in a `PhysicalPlanNode` before delegating, and add a regression test that calls the shim with a `self` of a different plan variant. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01FiNDkYXSeuTbCxUPDnu6dB --- datafusion/proto/src/physical_plan/mod.rs | 13 ++++- .../tests/cases/roundtrip_physical_plan.rs | 58 +++++++++++++++++++ 2 files changed, 69 insertions(+), 2 deletions(-) diff --git a/datafusion/proto/src/physical_plan/mod.rs b/datafusion/proto/src/physical_plan/mod.rs index 5105c0be39ac2..122e4d3def3d8 100644 --- a/datafusion/proto/src/physical_plan/mod.rs +++ b/datafusion/proto/src/physical_plan/mod.rs @@ -1184,7 +1184,7 @@ pub trait PhysicalPlanNodeExt: Sized { )] fn try_into_projection_physical_plan( &self, - _projection: &protobuf::ProjectionExecNode, + projection: &protobuf::ProjectionExecNode, ctx: &PhysicalPlanDecodeContext<'_>, proto_converter: &dyn PhysicalProtoConverterExtension, ) -> Result> { @@ -1193,7 +1193,16 @@ pub trait PhysicalPlanNodeExt: Sized { proto_converter, }; let decode_ctx = ExecutionPlanDecodeCtx::new(&decoder); - ProjectionExec::try_from_proto(self.node(), &decode_ctx) + // `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( diff --git a/datafusion/proto/tests/cases/roundtrip_physical_plan.rs b/datafusion/proto/tests/cases/roundtrip_physical_plan.rs index d87efcf98665b..351046d062ed9 100644 --- a/datafusion/proto/tests/cases/roundtrip_physical_plan.rs +++ b/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::() + .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 { use datafusion_common::test_util::datafusion_test_data;