This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-23782-60cdf8ad74845ff04edbdd10f9e739751106fbc5 in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit aa6bee575b4b3c686b917b66a34ff87beb0c7a35 Author: Burak Şen <[email protected]> AuthorDate: Thu Jul 23 20:07:50 2026 +0300 refactor(proto): migrate UnionExec and InterleaveExec serde (#23782) ## Which issue does this PR close? - Closes #23503. ## Rationale for this change Part of epic #23494. Moves `UnionExec` and `InterleaveExec` protobuf serialization from central dispatch ## What changes are included in this PR? add new proto functions in physical plan and deprecate proto ones ## Are these changes tested? yes existing round trip tests pass ## Are there any user-facing changes? existing methods are deprecated with no immediate api change --- datafusion/physical-plan/src/union.rs | 72 +++++++++++++++++++ datafusion/proto/src/physical_plan/mod.rs | 114 ++++++++++++++---------------- 2 files changed, 126 insertions(+), 60 deletions(-) diff --git a/datafusion/physical-plan/src/union.rs b/datafusion/physical-plan/src/union.rs index 3511609e2e..4722329ea5 100644 --- a/datafusion/physical-plan/src/union.rs +++ b/datafusion/physical-plan/src/union.rs @@ -473,6 +473,42 @@ impl ExecutionPlan for UnionExec { // on all children (either pushed down or via FilterExec) Ok(propagation) } + #[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 inputs = ctx.encode_children(self.inputs())?; + Ok(Some(protobuf::PhysicalPlanNode { + physical_plan_type: Some( + protobuf::physical_plan_node::PhysicalPlanType::Union( + protobuf::UnionExecNode { inputs }, + ), + ), + })) + } +} + +#[cfg(feature = "proto")] +impl UnionExec { + 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 union = crate::expect_plan_variant!( + node, + protobuf::physical_plan_node::PhysicalPlanType::Union, + "UnionExec", + ); + let inputs = union + .inputs + .iter() + .map(|input| ctx.decode_child(input)) + .collect::<Result<Vec<_>>>()?; + UnionExec::try_new(inputs) + } } /// Combines multiple input streams by interleaving them. @@ -683,6 +719,42 @@ impl ExecutionPlan for InterleaveExec { fn benefits_from_input_partitioning(&self) -> Vec<bool> { vec![false; self.children().len()] } + #[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 inputs = ctx.encode_children(self.inputs())?; + Ok(Some(protobuf::PhysicalPlanNode { + physical_plan_type: Some( + protobuf::physical_plan_node::PhysicalPlanType::Interleave( + protobuf::InterleaveExecNode { inputs }, + ), + ), + })) + } +} + +#[cfg(feature = "proto")] +impl InterleaveExec { + 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 interleave = crate::expect_plan_variant!( + node, + protobuf::physical_plan_node::PhysicalPlanType::Interleave, + "InterleaveExec", + ); + let inputs = interleave + .inputs + .iter() + .map(|input| ctx.decode_child(input)) + .collect::<Result<Vec<_>>>()?; + Ok(Arc::new(InterleaveExec::try_new(inputs)?)) + } } /// Returns true if all inputs have the same [`Partitioning::Hash`] or [`Partitioning::Range`] diff --git a/datafusion/proto/src/physical_plan/mod.rs b/datafusion/proto/src/physical_plan/mod.rs index 518d7f01b1..13ed27104b 100644 --- a/datafusion/proto/src/physical_plan/mod.rs +++ b/datafusion/proto/src/physical_plan/mod.rs @@ -814,11 +814,11 @@ pub trait PhysicalPlanNodeExt: Sized { ctx, proto_converter, ), - PhysicalPlanType::Union(union) => { - self.try_into_union_physical_plan(union, ctx, proto_converter) + PhysicalPlanType::Union(_) => { + UnionExec::try_from_proto(self.node(), &decode_ctx) } - PhysicalPlanType::Interleave(interleave) => { - self.try_into_interleave_physical_plan(interleave, ctx, proto_converter) + PhysicalPlanType::Interleave(_) => { + InterleaveExec::try_from_proto(self.node(), &decode_ctx) } PhysicalPlanType::CrossJoin(crossjoin) => { self.try_into_cross_join_physical_plan(crossjoin, ctx, proto_converter) @@ -1001,22 +1001,6 @@ pub trait PhysicalPlanNodeExt: Sized { ); } - if let Some(union) = plan.downcast_ref::<UnionExec>() { - return protobuf::PhysicalPlanNode::try_from_union_exec( - union, - codec, - proto_converter, - ); - } - - if let Some(interleave) = plan.downcast_ref::<InterleaveExec>() { - return protobuf::PhysicalPlanNode::try_from_interleave_exec( - interleave, - codec, - proto_converter, - ); - } - if let Some(exec) = plan.downcast_ref::<SortPreservingMergeExec>() { return protobuf::PhysicalPlanNode::try_from_sort_preserving_merge_exec( exec, @@ -2050,30 +2034,46 @@ pub trait PhysicalPlanNodeExt: Sized { .map(|e| Arc::new(e) as _) } + #[deprecated( + since = "55.0.0", + note = "unused by DataFusion; `UnionExec` deserializes itself via `UnionExec::try_from_proto`" + )] fn try_into_union_physical_plan( &self, union: &protobuf::UnionExecNode, ctx: &PhysicalPlanDecodeContext<'_>, proto_converter: &dyn PhysicalProtoConverterExtension, ) -> Result<Arc<dyn ExecutionPlan>> { - let mut inputs: Vec<Arc<dyn ExecutionPlan>> = vec![]; - for input in &union.inputs { - inputs.push(proto_converter.proto_to_execution_plan(input, ctx)?); - } - UnionExec::try_new(inputs) + let node = protobuf::PhysicalPlanNode { + physical_plan_type: Some(PhysicalPlanType::Union(union.clone())), + }; + let decoder = ConverterPlanDecoder { + ctx, + proto_converter, + }; + let decode_ctx = ExecutionPlanDecodeCtx::new(&decoder); + UnionExec::try_from_proto(&node, &decode_ctx) } + #[deprecated( + since = "55.0.0", + note = "unused by DataFusion; `InterleaveExec` deserializes itself via `InterleaveExec::try_from_proto`" + )] fn try_into_interleave_physical_plan( &self, interleave: &protobuf::InterleaveExecNode, ctx: &PhysicalPlanDecodeContext<'_>, proto_converter: &dyn PhysicalProtoConverterExtension, ) -> Result<Arc<dyn ExecutionPlan>> { - let mut inputs: Vec<Arc<dyn ExecutionPlan>> = vec![]; - for input in &interleave.inputs { - inputs.push(proto_converter.proto_to_execution_plan(input, ctx)?); - } - Ok(Arc::new(InterleaveExec::try_new(inputs)?)) + let node = protobuf::PhysicalPlanNode { + physical_plan_type: Some(PhysicalPlanType::Interleave(interleave.clone())), + }; + let decoder = ConverterPlanDecoder { + ctx, + proto_converter, + }; + let decode_ctx = ExecutionPlanDecodeCtx::new(&decoder); + InterleaveExec::try_from_proto(&node, &decode_ctx) } fn try_into_cross_join_physical_plan( @@ -3558,48 +3558,42 @@ pub trait PhysicalPlanNodeExt: Sized { }) } + #[deprecated( + since = "55.0.0", + note = "unused by DataFusion; `UnionExec` serializes itself via `ExecutionPlan::try_to_proto`" + )] fn try_from_union_exec( union: &UnionExec, codec: &dyn PhysicalExtensionCodec, proto_converter: &dyn PhysicalProtoConverterExtension, ) -> Result<protobuf::PhysicalPlanNode> { - let mut inputs: Vec<protobuf::PhysicalPlanNode> = vec![]; - for input in union.inputs() { - inputs.push( - protobuf::PhysicalPlanNode::try_from_physical_plan_with_converter( - input.to_owned(), - codec, - proto_converter, - )?, - ); - } - Ok(protobuf::PhysicalPlanNode { - physical_plan_type: Some(PhysicalPlanType::Union(protobuf::UnionExecNode { - inputs, - })), - }) + let encoder = ConverterPlanEncoder { + codec, + proto_converter, + }; + let encode_ctx = ExecutionPlanEncodeCtx::new(&encoder); + union + .try_to_proto(&encode_ctx)? + .ok_or_else(|| internal_datafusion_err!("UnionExec is not serializable")) } + #[deprecated( + since = "55.0.0", + note = "unused by DataFusion; `InterleaveExec` serializes itself via `ExecutionPlan::try_to_proto`" + )] fn try_from_interleave_exec( interleave: &InterleaveExec, codec: &dyn PhysicalExtensionCodec, proto_converter: &dyn PhysicalProtoConverterExtension, ) -> Result<protobuf::PhysicalPlanNode> { - let mut inputs: Vec<protobuf::PhysicalPlanNode> = vec![]; - for input in interleave.inputs() { - inputs.push( - protobuf::PhysicalPlanNode::try_from_physical_plan_with_converter( - input.to_owned(), - codec, - proto_converter, - )?, - ); - } - Ok(protobuf::PhysicalPlanNode { - physical_plan_type: Some(PhysicalPlanType::Interleave( - protobuf::InterleaveExecNode { inputs }, - )), - }) + let encoder = ConverterPlanEncoder { + codec, + proto_converter, + }; + let encode_ctx = ExecutionPlanEncodeCtx::new(&encoder); + interleave + .try_to_proto(&encode_ctx)? + .ok_or_else(|| internal_datafusion_err!("InterleaveExec is not serializable")) } fn try_from_sort_preserving_merge_exec( --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
