This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new f7aef23c6b refactor(proto): delegate deprecated ProjectionExec serde
shims to new hooks (#23731)
f7aef23c6b is described below
commit f7aef23c6b7ff1edb06dfcb08b6afa5a77114000
Author: Adrian Garcia Badaracco <[email protected]>
AuthorDate: Thu Jul 23 11:10:17 2026 -0500
refactor(proto): delegate deprecated ProjectionExec serde shims to new
hooks (#23731)
## Which issue does this PR close?
- Part of #23494.
## Rationale for this change
#23495 landed the deprecated `ProjectionExec` compatibility shims
(`try_from_projection_exec` / `try_into_projection_physical_plan`) with
their
original bodies fully duplicated in `datafusion-proto`, even though the
same
wire-format logic now lives in `ProjectionExec::try_to_proto` /
`try_from_proto`
(the #22419 hook pattern). This PR makes those deprecated shims
**delegate** to
the new hooks so the wire format is single-sourced in
`datafusion-physical-plan`
rather than duplicated. This matches the `FilterExec` delegate style in
#23708.
No wire-format change.
## What changes are included in this PR?
- Replace the body of `try_from_projection_exec` with a thin shim that
builds an
`ExecutionPlanEncodeCtx` from the `codec` + `proto_converter` (via
`ConverterPlanEncoder`) and delegates to `ProjectionExec::try_to_proto`.
- Replace the body of `try_into_projection_physical_plan` with a thin
shim that
builds an `ExecutionPlanDecodeCtx` (via `ConverterPlanDecoder`) and
delegates
to `ProjectionExec::try_from_proto`; the now-unused `projection` param
is
renamed to `_projection`.
- Drop the now-unused `ProjectionExpr` import.
The `#[deprecated(...)]` attributes and signatures are otherwise
unchanged, and
the live dispatch arms and the `ProjectionExec` hook impls are
untouched.
## Are these changes tested?
Yes — the existing `roundtrip_physical_plan` projection/filter roundtrip
tests
exercise this path and pass (`roundtrip_projection_source`,
`roundtrip_filter_with_fetch`, `roundtrip_filter_with_not_and_in_list`,
and the
full 203-test `roundtrip` suite). `cargo clippy -p datafusion-proto
--all-features -- -D warnings` is clean.
## Are there any user-facing changes?
No. The methods remain `#[deprecated]` with identical signatures and
wire
format; only their internals change.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
https://claude.ai/code/session_01U9qX6kekbJGpmTSxvrU8e5
---------
Co-authored-by: Claude Opus 4.8 (1M context) <[email protected]>
---
datafusion/proto/src/physical_plan/mod.rs | 67 +++++++---------------
.../proto/tests/cases/roundtrip_physical_plan.rs | 58 +++++++++++++++++++
2 files changed, 79 insertions(+), 46 deletions(-)
diff --git a/datafusion/proto/src/physical_plan/mod.rs
b/datafusion/proto/src/physical_plan/mod.rs
index cea334e42a..518d7f01b1 100644
--- a/datafusion/proto/src/physical_plan/mod.rs
+++ b/datafusion/proto/src/physical_plan/mod.rs
@@ -90,7 +90,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,
@@ -1145,28 +1145,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)
}
#[deprecated(
@@ -2777,31 +2770,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")
})
}
diff --git a/datafusion/proto/tests/cases/roundtrip_physical_plan.rs
b/datafusion/proto/tests/cases/roundtrip_physical_plan.rs
index 2671f4d015..89647d5da3 100644
--- a/datafusion/proto/tests/cases/roundtrip_physical_plan.rs
+++ b/datafusion/proto/tests/cases/roundtrip_physical_plan.rs
@@ -2456,6 +2456,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;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
