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 17634176a4 refactor(proto): migrate UnionExec and InterleaveExec serde
(#23782)
17634176a4 is described below
commit 17634176a4f406a50b31a3711514831c9d3ecfa0
Author: Burak Şen <[email protected]>
AuthorDate: Thu Jul 23 23:34:08 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 7139741dd3..1b122fcde6 100644
--- a/datafusion/proto/src/physical_plan/mod.rs
+++ b/datafusion/proto/src/physical_plan/mod.rs
@@ -808,11 +808,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(_) => {
CrossJoinExec::try_from_proto(self.node(), &decode_ctx)
@@ -956,22 +956,6 @@ pub trait PhysicalPlanNodeExt: Sized {
return Ok(node);
}
- 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::<WindowAggExec>() {
return protobuf::PhysicalPlanNode::try_from_window_agg_exec(
exec,
@@ -1998,30 +1982,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)
}
#[deprecated(
@@ -3329,48 +3329,42 @@ pub trait PhysicalPlanNodeExt: Sized {
.ok_or_else(|| internal_datafusion_err!("SortExec is not
serializable"))
}
+ #[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"))
}
#[deprecated(
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
