You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/07/23 20:34:30 UTC
(datafusion) branch gh-readonly-queue/main/pr-23782-e084034f7c45c5e8e38d7a2a619369a43e6cf347 created (now 17634176a4)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-23782-e084034f7c45c5e8e38d7a2a619369a43e6cf347
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at 17634176a4 refactor(proto): migrate UnionExec and InterleaveExec serde (#23782)
This branch includes the following new commits:
new 17634176a4 refactor(proto): migrate UnionExec and InterleaveExec serde (#23782)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion) 01/01: refactor(proto): migrate UnionExec and InterleaveExec serde (#23782)
Posted by gi...@apache.org.
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-e084034f7c45c5e8e38d7a2a619369a43e6cf347
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 17634176a4f406a50b31a3711514831c9d3ecfa0
Author: Burak Şen <bu...@gmail.com>
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: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org