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 f3f381cd93 refactor(proto): migrate single-child plans (#23710)
f3f381cd93 is described below
commit f3f381cd93baccff1657e14980d94c19db181179
Author: Phoenix <[email protected]>
AuthorDate: Tue Jul 21 14:10:33 2026 +0800
refactor(proto): migrate single-child plans (#23710)
## Which issue does this PR close?
<!--
We generally require a GitHub issue to be filed for all bug fixes and
enhancements and this helps us generate change logs for our releases.
You can link an issue to this PR using the GitHub syntax. For example
`Closes #123` indicates that this PR will close issue #123.
-->
- Closes #23500.
## Rationale for this change
<!--
Why are you proposing this change? If this is already explained clearly
in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand
your changes and offer better suggestions for fixes.
-->
Physical plan serialization currently relies on centralized type
dispatch in `datafusion-proto`. This separates serialization from the
plans that own the relevant state and requires every built-in plan to
be handled specially.
Moving serialization into each plan keeps that logic with its owner and
allows the central dispatch chain to be removed incrementally.
`CoalesceBatchesExec` is deprecated but remains a supported public type.
Migrating its existing serialization keeps the dispatch model
consistent without extending its lifetime or changing its behavior.
## What changes are included in this PR?
<!--
There is no need to duplicate the description in the issue here but it
is sometimes worth providing a summary of the individual changes in this
PR.
-->
- Add per-plan protobuf encoding and decoding for:
- `CoalesceBatchesExec`
- `CoalescePartitionsExec`
- `CooperativeExec`
- `BufferExec`
- Route decoding through each plan's `try_from_proto` implementation.
- Remove the corresponding centralized encoding branches.
- Retain the old public conversion helpers as deprecated compatibility
APIs.
- Preserve the existing protobuf wire representation.
- Add roundtrip coverage for `CooperativeExec` and `BufferExec`.
## Are these changes tested?
<!--
We typically require tests for all PRs in order to:
1. Prevent the code from being accidentally broken by subsequent changes
2. Serve as another way to document the expected behavior of the code
If tests are not included in your PR, please explain why (for example,
are they covered by existing tests)?
-->
Yes
## Are there any user-facing changes?
<!--
If there are user-facing changes then we may require documentation to be
updated before approving the PR.
-->
<!--
If there are any breaking changes to public APIs, please add the `api
change` label.
-->
There are no runtime behavior or protobuf wire-format changes.
The old `PhysicalPlanNodeExt` conversion helpers remain available but
are deprecated in favor of the per-plan serialization hooks.
---------
Signed-off-by: Jiawei Zhao <[email protected]>
---
datafusion/physical-plan/src/buffer.rs | 42 ++++
datafusion/physical-plan/src/coalesce_batches.rs | 55 +++++
.../physical-plan/src/coalesce_partitions.rs | 50 +++++
datafusion/physical-plan/src/coop.rs | 44 ++++
datafusion/proto/src/physical_plan/mod.rs | 228 +++++++++++----------
.../proto/tests/cases/roundtrip_physical_plan.rs | 27 +++
6 files changed, 336 insertions(+), 110 deletions(-)
diff --git a/datafusion/physical-plan/src/buffer.rs
b/datafusion/physical-plan/src/buffer.rs
index 6f83e37196..3be331a1ee 100644
--- a/datafusion/physical-plan/src/buffer.rs
+++ b/datafusion/physical-plan/src/buffer.rs
@@ -298,6 +298,48 @@ impl ExecutionPlan for BufferExec {
Ok(Arc::new(Self::new(new_input, self.capacity)) as Arc<dyn
ExecutionPlan>)
})
}
+
+ #[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 input = ctx.encode_child(self.input())?;
+ Ok(Some(protobuf::PhysicalPlanNode {
+ physical_plan_type: Some(
+
protobuf::physical_plan_node::PhysicalPlanType::Buffer(Box::new(
+ protobuf::BufferExecNode {
+ input: Some(Box::new(input)),
+ capacity: self.capacity() as u64,
+ },
+ )),
+ ),
+ }))
+ }
+}
+
+#[cfg(feature = "proto")]
+impl BufferExec {
+ /// Reconstruct a [`BufferExec`] from its protobuf representation.
+ ///
+ /// The exact inverse of [`ExecutionPlan::try_to_proto`].
+ ///
+ /// [`ExecutionPlan::try_to_proto`]: crate::ExecutionPlan::try_to_proto
+ 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 buffer = crate::expect_plan_variant!(
+ node,
+ protobuf::physical_plan_node::PhysicalPlanType::Buffer,
+ "BufferExec",
+ );
+ let input =
+ ctx.decode_required_child(buffer.input.as_deref(), "BufferExec",
"input")?;
+ Ok(Arc::new(BufferExec::new(input, buffer.capacity as usize)))
+ }
}
/// Represents anything that occupies a capacity in a [MemoryBufferedStream].
diff --git a/datafusion/physical-plan/src/coalesce_batches.rs
b/datafusion/physical-plan/src/coalesce_batches.rs
index 667a8a3ce0..c5b9176777 100644
--- a/datafusion/physical-plan/src/coalesce_batches.rs
+++ b/datafusion/physical-plan/src/coalesce_batches.rs
@@ -290,6 +290,61 @@ impl ExecutionPlan for CoalesceBatchesExec {
) as Arc<dyn ExecutionPlan>)
})
}
+
+ #[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 input = ctx.encode_child(self.input())?;
+ Ok(Some(protobuf::PhysicalPlanNode {
+ physical_plan_type: Some(
+
protobuf::physical_plan_node::PhysicalPlanType::CoalesceBatches(
+ Box::new(protobuf::CoalesceBatchesExecNode {
+ input: Some(Box::new(input)),
+ target_batch_size: self.target_batch_size() as u32,
+ fetch: self.fetch().map(|n| n as u32),
+ }),
+ ),
+ ),
+ }))
+ }
+}
+
+#[cfg(feature = "proto")]
+#[expect(deprecated)]
+impl CoalesceBatchesExec {
+ /// Reconstruct a [`CoalesceBatchesExec`] from its protobuf representation.
+ ///
+ /// The exact inverse of [`ExecutionPlan::try_to_proto`]: it takes the
whole
+ /// [`PhysicalPlanNode`] so every plan's `try_from_proto` shares one
+ /// signature. The child plan is decoded recursively via the
+ /// [`ExecutionPlanDecodeCtx`].
+ ///
+ /// [`PhysicalPlanNode`]:
datafusion_proto_models::protobuf::PhysicalPlanNode
+ /// [`ExecutionPlan::try_to_proto`]: crate::ExecutionPlan::try_to_proto
+ /// [`ExecutionPlanDecodeCtx`]: crate::proto::ExecutionPlanDecodeCtx
+ 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 coalesce_batches = crate::expect_plan_variant!(
+ node,
+ protobuf::physical_plan_node::PhysicalPlanType::CoalesceBatches,
+ "CoalesceBatchesExec",
+ );
+ let input = ctx.decode_required_child(
+ coalesce_batches.input.as_deref(),
+ "CoalesceBatchesExec",
+ "input",
+ )?;
+ Ok(Arc::new(
+ CoalesceBatchesExec::new(input, coalesce_batches.target_batch_size
as usize)
+ .with_fetch(coalesce_batches.fetch.map(|f| f as usize)),
+ ))
+ }
}
/// Stream for [`CoalesceBatchesExec`]. See [`CoalesceBatchesExec`] for more
details.
diff --git a/datafusion/physical-plan/src/coalesce_partitions.rs
b/datafusion/physical-plan/src/coalesce_partitions.rs
index 1c6f95c53b..f9694e0d16 100644
--- a/datafusion/physical-plan/src/coalesce_partitions.rs
+++ b/datafusion/physical-plan/src/coalesce_partitions.rs
@@ -346,6 +346,56 @@ impl ExecutionPlan for CoalescePartitionsExec {
}
})
}
+
+ #[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 input = ctx.encode_child(self.input())?;
+ Ok(Some(protobuf::PhysicalPlanNode {
+ physical_plan_type: Some(
+ protobuf::physical_plan_node::PhysicalPlanType::Merge(Box::new(
+ protobuf::CoalescePartitionsExecNode {
+ input: Some(Box::new(input)),
+ fetch: self.fetch().map(|f| f as u32),
+ },
+ )),
+ ),
+ }))
+ }
+}
+
+#[cfg(feature = "proto")]
+impl CoalescePartitionsExec {
+ /// Reconstruct a [`CoalescePartitionsExec`] from its protobuf
representation.
+ ///
+ /// The exact inverse of [`ExecutionPlan::try_to_proto`]. Note the protobuf
+ /// variant is named `Merge` (node [`CoalescePartitionsExecNode`]).
+ ///
+ /// [`CoalescePartitionsExecNode`]:
datafusion_proto_models::protobuf::CoalescePartitionsExecNode
+ /// [`ExecutionPlan::try_to_proto`]: crate::ExecutionPlan::try_to_proto
+ 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 merge = crate::expect_plan_variant!(
+ node,
+ protobuf::physical_plan_node::PhysicalPlanType::Merge,
+ "CoalescePartitionsExec",
+ );
+ let input = ctx.decode_required_child(
+ merge.input.as_deref(),
+ "CoalescePartitionsExec",
+ "input",
+ )?;
+ Ok(Arc::new(
+ CoalescePartitionsExec::new(input)
+ .with_fetch(merge.fetch.map(|f| f as usize)),
+ ))
+ }
}
#[cfg(test)]
diff --git a/datafusion/physical-plan/src/coop.rs
b/datafusion/physical-plan/src/coop.rs
index 94e4fdca2b..a5b57f546b 100644
--- a/datafusion/physical-plan/src/coop.rs
+++ b/datafusion/physical-plan/src/coop.rs
@@ -369,6 +369,50 @@ impl ExecutionPlan for CooperativeExec {
}
}
}
+
+ #[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 input = ctx.encode_child(self.input())?;
+ Ok(Some(protobuf::PhysicalPlanNode {
+ physical_plan_type: Some(
+
protobuf::physical_plan_node::PhysicalPlanType::Cooperative(Box::new(
+ protobuf::CooperativeExecNode {
+ input: Some(Box::new(input)),
+ },
+ )),
+ ),
+ }))
+ }
+}
+
+#[cfg(feature = "proto")]
+impl CooperativeExec {
+ /// Reconstruct a [`CooperativeExec`] from its protobuf representation.
+ ///
+ /// The exact inverse of [`ExecutionPlan::try_to_proto`].
+ ///
+ /// [`ExecutionPlan::try_to_proto`]: crate::ExecutionPlan::try_to_proto
+ 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 cooperative = crate::expect_plan_variant!(
+ node,
+ protobuf::physical_plan_node::PhysicalPlanType::Cooperative,
+ "CooperativeExec",
+ );
+ let input = ctx.decode_required_child(
+ cooperative.input.as_deref(),
+ "CooperativeExec",
+ "input",
+ )?;
+ Ok(Arc::new(CooperativeExec::new(input)))
+ }
}
/// Creates a [`CooperativeStream`] wrapper around the given
[`RecordBatchStream`].
diff --git a/datafusion/proto/src/physical_plan/mod.rs
b/datafusion/proto/src/physical_plan/mod.rs
index 8649630002..fa18b1ffc6 100644
--- a/datafusion/proto/src/physical_plan/mod.rs
+++ b/datafusion/proto/src/physical_plan/mod.rs
@@ -70,7 +70,10 @@ use datafusion_physical_plan::aggregates::{
use datafusion_physical_plan::analyze::AnalyzeExec;
use datafusion_physical_plan::async_func::AsyncFuncExec;
use datafusion_physical_plan::buffer::BufferExec;
-#[expect(deprecated)]
+#[expect(
+ deprecated,
+ reason = "`CoalesceBatchesExec` remains supported for protobuf
compatibility"
+)]
use datafusion_physical_plan::coalesce_batches::CoalesceBatchesExec;
use datafusion_physical_plan::coalesce_partitions::CoalescePartitionsExec;
use datafusion_physical_plan::coop::CooperativeExec;
@@ -777,14 +780,15 @@ pub trait PhysicalPlanNodeExt: Sized {
PhysicalPlanType::ArrowScan(scan) => {
self.try_into_arrow_scan_physical_plan(scan, ctx,
proto_converter)
}
- PhysicalPlanType::CoalesceBatches(coalesce_batches) => self
- .try_into_coalesce_batches_physical_plan(
- coalesce_batches,
- ctx,
- proto_converter,
- ),
- PhysicalPlanType::Merge(merge) => {
- self.try_into_merge_physical_plan(merge, ctx, proto_converter)
+ #[expect(
+ deprecated,
+ reason = "`CoalesceBatchesExec` remains supported for protobuf
compatibility"
+ )]
+ PhysicalPlanType::CoalesceBatches(_) => {
+ CoalesceBatchesExec::try_from_proto(self.node(), &decode_ctx)
+ }
+ PhysicalPlanType::Merge(_) => {
+ CoalescePartitionsExec::try_from_proto(self.node(),
&decode_ctx)
}
PhysicalPlanType::Repartition(repart) => {
self.try_into_repartition_physical_plan(repart, ctx,
proto_converter)
@@ -852,8 +856,8 @@ pub trait PhysicalPlanNodeExt: Sized {
PhysicalPlanType::Unnest(unnest) => {
self.try_into_unnest_physical_plan(unnest, ctx,
proto_converter)
}
- PhysicalPlanType::Cooperative(cooperative) => {
- self.try_into_cooperative_physical_plan(cooperative, ctx,
proto_converter)
+ PhysicalPlanType::Cooperative(_) => {
+ CooperativeExec::try_from_proto(self.node(), &decode_ctx)
}
PhysicalPlanType::GenerateSeries(generate_series) => {
self.try_into_generate_series_physical_plan(generate_series)
@@ -864,8 +868,8 @@ pub trait PhysicalPlanNodeExt: Sized {
PhysicalPlanType::AsyncFunc(async_func) => {
self.try_into_async_func_physical_plan(async_func, ctx,
proto_converter)
}
- PhysicalPlanType::Buffer(buffer) => {
- self.try_into_buffer_physical_plan(buffer, ctx,
proto_converter)
+ PhysicalPlanType::Buffer(_) => {
+ BufferExec::try_from_proto(self.node(), &decode_ctx)
}
PhysicalPlanType::ScalarSubquery(sq) => {
self.try_into_scalar_subquery_physical_plan(sq, ctx,
proto_converter)
@@ -979,15 +983,6 @@ pub trait PhysicalPlanNodeExt: Sized {
);
}
- #[expect(deprecated)]
- if let Some(coalesce_batches) =
plan.downcast_ref::<CoalesceBatchesExec>() {
- return protobuf::PhysicalPlanNode::try_from_coalesce_batches_exec(
- coalesce_batches,
- codec,
- proto_converter,
- );
- }
-
if let Some(data_source_exec) = plan.downcast_ref::<DataSourceExec>()
&& let Some(node) =
protobuf::PhysicalPlanNode::try_from_data_source_exec(
data_source_exec,
@@ -998,14 +993,6 @@ pub trait PhysicalPlanNodeExt: Sized {
return Ok(node);
}
- if let Some(exec) = plan.downcast_ref::<CoalescePartitionsExec>() {
- return
protobuf::PhysicalPlanNode::try_from_coalesce_partitions_exec(
- exec,
- codec,
- proto_converter,
- );
- }
-
if let Some(exec) = plan.downcast_ref::<RepartitionExec>() {
return protobuf::PhysicalPlanNode::try_from_repartition_exec(
exec,
@@ -1088,14 +1075,6 @@ pub trait PhysicalPlanNodeExt: Sized {
);
}
- if let Some(exec) = plan.downcast_ref::<CooperativeExec>() {
- return protobuf::PhysicalPlanNode::try_from_cooperative_exec(
- exec,
- codec,
- proto_converter,
- );
- }
-
if let Some(exec) = plan.downcast_ref::<LazyMemoryExec>()
&& let Some(node) =
protobuf::PhysicalPlanNode::try_from_lazy_memory_exec(exec)?
@@ -1111,14 +1090,6 @@ pub trait PhysicalPlanNodeExt: Sized {
);
}
- if let Some(exec) = plan.downcast_ref::<BufferExec>() {
- return protobuf::PhysicalPlanNode::try_from_buffer_exec(
- exec,
- codec,
- proto_converter,
- );
- }
-
if let Some(exec) = plan.downcast_ref::<ScalarSubqueryExec>() {
return protobuf::PhysicalPlanNode::try_from_scalar_subquery_exec(
exec,
@@ -1471,33 +1442,52 @@ pub trait PhysicalPlanNodeExt: Sized {
Ok(DataSourceExec::from_data_source(source))
}
+ #[deprecated(
+ since = "55.0.0",
+ note = "unused by DataFusion; `CoalesceBatchesExec` deserializes
itself via `CoalesceBatchesExec::try_from_proto`"
+ )]
fn try_into_coalesce_batches_physical_plan(
&self,
coalesce_batches: &protobuf::CoalesceBatchesExecNode,
ctx: &PhysicalPlanDecodeContext<'_>,
proto_converter: &dyn PhysicalProtoConverterExtension,
) -> Result<Arc<dyn ExecutionPlan>> {
- let input: Arc<dyn ExecutionPlan> =
- into_physical_plan(&coalesce_batches.input, ctx, proto_converter)?;
- Ok(Arc::new(
- #[expect(deprecated)]
- CoalesceBatchesExec::new(input, coalesce_batches.target_batch_size
as usize)
- .with_fetch(coalesce_batches.fetch.map(|f| f as usize)),
- ))
+ let node = protobuf::PhysicalPlanNode {
+ physical_plan_type:
Some(PhysicalPlanType::CoalesceBatches(Box::new(
+ coalesce_batches.clone(),
+ ))),
+ };
+ let decoder = ConverterPlanDecoder {
+ ctx,
+ proto_converter,
+ };
+ let decode_ctx = ExecutionPlanDecodeCtx::new(&decoder);
+ #[expect(
+ deprecated,
+ reason = "`CoalesceBatchesExec` remains supported for protobuf
compatibility"
+ )]
+ CoalesceBatchesExec::try_from_proto(&node, &decode_ctx)
}
+ #[deprecated(
+ since = "55.0.0",
+ note = "unused by DataFusion; `CoalescePartitionsExec` deserializes
itself via `CoalescePartitionsExec::try_from_proto`"
+ )]
fn try_into_merge_physical_plan(
&self,
merge: &protobuf::CoalescePartitionsExecNode,
ctx: &PhysicalPlanDecodeContext<'_>,
proto_converter: &dyn PhysicalProtoConverterExtension,
) -> Result<Arc<dyn ExecutionPlan>> {
- let input: Arc<dyn ExecutionPlan> =
- into_physical_plan(&merge.input, ctx, proto_converter)?;
- Ok(Arc::new(
- CoalescePartitionsExec::new(input)
- .with_fetch(merge.fetch.map(|f| f as usize)),
- ))
+ let node = protobuf::PhysicalPlanNode {
+ physical_plan_type:
Some(PhysicalPlanType::Merge(Box::new(merge.clone()))),
+ };
+ let decoder = ConverterPlanDecoder {
+ ctx,
+ proto_converter,
+ };
+ let decode_ctx = ExecutionPlanDecodeCtx::new(&decoder);
+ CoalescePartitionsExec::try_from_proto(&node, &decode_ctx)
}
fn try_into_repartition_physical_plan(
@@ -2737,14 +2727,27 @@ pub trait PhysicalPlanNodeExt: Sized {
Ok(Arc::new(LazyMemoryExec::try_new(schema, vec![generator])?))
}
+ #[deprecated(
+ since = "55.0.0",
+ note = "unused by DataFusion; `CooperativeExec` deserializes itself
via `CooperativeExec::try_from_proto`"
+ )]
fn try_into_cooperative_physical_plan(
&self,
field_stream: &protobuf::CooperativeExecNode,
ctx: &PhysicalPlanDecodeContext<'_>,
proto_converter: &dyn PhysicalProtoConverterExtension,
) -> Result<Arc<dyn ExecutionPlan>> {
- let input = into_physical_plan(&field_stream.input, ctx,
proto_converter)?;
- Ok(Arc::new(CooperativeExec::new(input)))
+ let node = protobuf::PhysicalPlanNode {
+ physical_plan_type: Some(PhysicalPlanType::Cooperative(Box::new(
+ field_stream.clone(),
+ ))),
+ };
+ let decoder = ConverterPlanDecoder {
+ ctx,
+ proto_converter,
+ };
+ let decode_ctx = ExecutionPlanDecodeCtx::new(&decoder);
+ CooperativeExec::try_from_proto(&node, &decode_ctx)
}
fn try_into_async_func_physical_plan(
@@ -2784,16 +2787,25 @@ pub trait PhysicalPlanNodeExt: Sized {
Ok(Arc::new(AsyncFuncExec::try_new(async_exprs, input)?))
}
+ #[deprecated(
+ since = "55.0.0",
+ note = "unused by DataFusion; `BufferExec` deserializes itself via
`BufferExec::try_from_proto`"
+ )]
fn try_into_buffer_physical_plan(
&self,
buffer: &protobuf::BufferExecNode,
ctx: &PhysicalPlanDecodeContext<'_>,
proto_converter: &dyn PhysicalProtoConverterExtension,
) -> Result<Arc<dyn ExecutionPlan>> {
- let input: Arc<dyn ExecutionPlan> =
- into_physical_plan(&buffer.input, ctx, proto_converter)?;
-
- Ok(Arc::new(BufferExec::new(input, buffer.capacity as usize)))
+ let node = protobuf::PhysicalPlanNode {
+ physical_plan_type:
Some(PhysicalPlanType::Buffer(Box::new(buffer.clone()))),
+ };
+ let decoder = ConverterPlanDecoder {
+ ctx,
+ proto_converter,
+ };
+ let decode_ctx = ExecutionPlanDecodeCtx::new(&decoder);
+ BufferExec::try_from_proto(&node, &decode_ctx)
}
fn try_into_scalar_subquery_physical_plan(
@@ -3446,25 +3458,26 @@ pub trait PhysicalPlanNodeExt: Sized {
})
}
- #[expect(deprecated)]
+ #[deprecated(
+ since = "55.0.0",
+ note = "unused by DataFusion; `CoalesceBatchesExec` serializes itself
via `ExecutionPlan::try_to_proto`"
+ )]
+ #[expect(
+ deprecated,
+ reason = "`CoalesceBatchesExec` remains supported for protobuf
compatibility"
+ )]
fn try_from_coalesce_batches_exec(
coalesce_batches: &CoalesceBatchesExec,
codec: &dyn PhysicalExtensionCodec,
proto_converter: &dyn PhysicalProtoConverterExtension,
) -> Result<protobuf::PhysicalPlanNode> {
- let input =
protobuf::PhysicalPlanNode::try_from_physical_plan_with_converter(
- coalesce_batches.input().to_owned(),
+ let encoder = ConverterPlanEncoder {
codec,
proto_converter,
- )?;
- Ok(protobuf::PhysicalPlanNode {
- physical_plan_type:
Some(PhysicalPlanType::CoalesceBatches(Box::new(
- protobuf::CoalesceBatchesExecNode {
- input: Some(Box::new(input)),
- target_batch_size: coalesce_batches.target_batch_size() as
u32,
- fetch: coalesce_batches.fetch().map(|n| n as u32),
- },
- ))),
+ };
+ let encode_ctx = ExecutionPlanEncodeCtx::new(&encoder);
+ coalesce_batches.try_to_proto(&encode_ctx)?.ok_or_else(|| {
+ internal_datafusion_err!("CoalesceBatchesExec is not serializable")
})
}
@@ -3640,23 +3653,22 @@ pub trait PhysicalPlanNodeExt: Sized {
Ok(None)
}
+ #[deprecated(
+ since = "55.0.0",
+ note = "unused by DataFusion; `CoalescePartitionsExec` serializes
itself via `ExecutionPlan::try_to_proto`"
+ )]
fn try_from_coalesce_partitions_exec(
exec: &CoalescePartitionsExec,
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,
- )?;
- Ok(protobuf::PhysicalPlanNode {
- physical_plan_type: Some(PhysicalPlanType::Merge(Box::new(
- protobuf::CoalescePartitionsExecNode {
- input: Some(Box::new(input)),
- fetch: exec.fetch().map(|f| f as u32),
- },
- ))),
+ };
+ let encode_ctx = ExecutionPlanEncodeCtx::new(&encoder);
+ exec.try_to_proto(&encode_ctx)?.ok_or_else(|| {
+ internal_datafusion_err!("CoalescePartitionsExec is not
serializable")
})
}
@@ -4076,23 +4088,22 @@ pub trait PhysicalPlanNodeExt: Sized {
})
}
+ #[deprecated(
+ since = "55.0.0",
+ note = "unused by DataFusion; `CooperativeExec` serializes itself via
`ExecutionPlan::try_to_proto`"
+ )]
fn try_from_cooperative_exec(
exec: &CooperativeExec,
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,
- )?;
-
- Ok(protobuf::PhysicalPlanNode {
- physical_plan_type: Some(PhysicalPlanType::Cooperative(Box::new(
- protobuf::CooperativeExecNode {
- input: Some(Box::new(input)),
- },
- ))),
+ };
+ let encode_ctx = ExecutionPlanEncodeCtx::new(&encoder);
+ exec.try_to_proto(&encode_ctx)?.ok_or_else(|| {
+ internal_datafusion_err!("CooperativeExec is not serializable")
})
}
@@ -4250,25 +4261,22 @@ pub trait PhysicalPlanNodeExt: Sized {
})
}
+ #[deprecated(
+ since = "55.0.0",
+ note = "unused by DataFusion; `BufferExec` serializes itself via
`ExecutionPlan::try_to_proto`"
+ )]
fn try_from_buffer_exec(
exec: &BufferExec,
extension_codec: &dyn PhysicalExtensionCodec,
proto_converter: &dyn PhysicalProtoConverterExtension,
) -> Result<protobuf::PhysicalPlanNode> {
- let input =
protobuf::PhysicalPlanNode::try_from_physical_plan_with_converter(
- Arc::clone(exec.input()),
- extension_codec,
+ let encoder = ConverterPlanEncoder {
+ codec: extension_codec,
proto_converter,
- )?;
-
- Ok(protobuf::PhysicalPlanNode {
- physical_plan_type: Some(PhysicalPlanType::Buffer(Box::new(
- protobuf::BufferExecNode {
- input: Some(Box::new(input)),
- capacity: exec.capacity() as u64,
- },
- ))),
- })
+ };
+ let encode_ctx = ExecutionPlanEncodeCtx::new(&encoder);
+ exec.try_to_proto(&encode_ctx)?
+ .ok_or_else(|| internal_datafusion_err!("BufferExec is not
serializable"))
}
fn try_from_scalar_subquery_exec(
diff --git a/datafusion/proto/tests/cases/roundtrip_physical_plan.rs
b/datafusion/proto/tests/cases/roundtrip_physical_plan.rs
index 8b7223ddd7..8027c0fa78 100644
--- a/datafusion/proto/tests/cases/roundtrip_physical_plan.rs
+++ b/datafusion/proto/tests/cases/roundtrip_physical_plan.rs
@@ -60,9 +60,11 @@ use datafusion::physical_plan::aggregates::{
AggregateExec, AggregateMode, LimitOptions, PhysicalGroupBy,
};
use datafusion::physical_plan::analyze::AnalyzeExec;
+use datafusion::physical_plan::buffer::BufferExec;
#[expect(deprecated)]
use datafusion::physical_plan::coalesce_batches::CoalesceBatchesExec;
use datafusion::physical_plan::coalesce_partitions::CoalescePartitionsExec;
+use datafusion::physical_plan::coop::CooperativeExec;
use datafusion::physical_plan::empty::EmptyExec;
use datafusion::physical_plan::expressions::{
BinaryExpr, Column, DynamicFilterPhysicalExpr, NotExpr, PhysicalSortExpr,
binary,
@@ -1158,6 +1160,31 @@ fn roundtrip_coalesce_partitions_with_fetch() ->
Result<()> {
))
}
+#[test]
+fn roundtrip_cooperative() -> Result<()> {
+ let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Boolean,
false)]));
+ roundtrip_test(Arc::new(CooperativeExec::new(Arc::new(EmptyExec::new(
+ schema,
+ )))))
+}
+
+#[test]
+fn roundtrip_buffer() -> Result<()> {
+ let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Boolean,
false)]));
+ let ctx = SessionContext::new();
+ let codec = DefaultPhysicalExtensionCodec {};
+ let proto_converter = DefaultPhysicalProtoConverter {};
+ let result = roundtrip_test_and_return(
+ Arc::new(BufferExec::new(Arc::new(EmptyExec::new(schema)), 4096)),
+ &ctx,
+ &codec,
+ &proto_converter,
+ )?;
+ let result = result.downcast_ref::<BufferExec>().unwrap();
+ assert_eq!(result.capacity(), 4096);
+ Ok(())
+}
+
#[test]
fn roundtrip_parquet_exec_with_pruning_predicate() -> Result<()> {
let file_schema =
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
