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 17:09:24 UTC
(datafusion) 01/01: refactor(proto): migrate GlobalLimitExec and LocalLimitExec serde (#23791)
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-23791-aa6bee575b4b3c686b917b66a34ff87beb0c7a35
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit f1486705443df3d088e217c6f8ec67f47ba4064e
Author: Burak Şen <bu...@gmail.com>
AuthorDate: Thu Jul 23 20:09:00 2026 +0300
refactor(proto): migrate GlobalLimitExec and LocalLimitExec serde (#23791)
## Which issue does this PR close?
- Closes #23502.
## Rationale for this change
Part of epic #23494. Moves `GlobalLimitExec` and `LocalLimitExec`
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/limit.rs | 90 ++++++++++++++++++++++++
datafusion/proto/src/physical_plan/mod.rs | 112 ++++++++++++++----------------
2 files changed, 142 insertions(+), 60 deletions(-)
diff --git a/datafusion/physical-plan/src/limit.rs b/datafusion/physical-plan/src/limit.rs
index 2f63ac05e8..a1f6074cb9 100644
--- a/datafusion/physical-plan/src/limit.rs
+++ b/datafusion/physical-plan/src/limit.rs
@@ -244,6 +244,59 @@ impl ExecutionPlan for GlobalLimitExec {
fn supports_limit_pushdown(&self) -> bool {
true
}
+
+ #[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::GlobalLimit(Box::new(
+ protobuf::GlobalLimitExecNode {
+ input: Some(Box::new(input)),
+ skip: self.skip() as u32,
+ fetch: match self.fetch() {
+ Some(n) => n as i64,
+ _ => -1, // no limit
+ },
+ },
+ )),
+ ),
+ }))
+ }
+}
+
+#[cfg(feature = "proto")]
+impl GlobalLimitExec {
+ 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 limit = crate::expect_plan_variant!(
+ node,
+ protobuf::physical_plan_node::PhysicalPlanType::GlobalLimit,
+ "GlobalLimitExec",
+ );
+ let input = ctx.decode_required_child(
+ limit.input.as_deref(),
+ "GlobalLimitExec",
+ "input",
+ )?;
+ let fetch = if limit.fetch >= 0 {
+ Some(limit.fetch as usize)
+ } else {
+ None
+ };
+ Ok(Arc::new(GlobalLimitExec::new(
+ input,
+ limit.skip as usize,
+ fetch,
+ )))
+ }
}
/// LocalLimitExec applies a limit to a single partition
@@ -419,6 +472,43 @@ impl ExecutionPlan for LocalLimitExec {
fn cardinality_effect(&self) -> CardinalityEffect {
CardinalityEffect::LowerEqual
}
+
+ #[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::LocalLimit(Box::new(
+ protobuf::LocalLimitExecNode {
+ input: Some(Box::new(input)),
+ fetch: self.fetch() as u32,
+ },
+ )),
+ ),
+ }))
+ }
+}
+
+#[cfg(feature = "proto")]
+impl LocalLimitExec {
+ 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 limit = crate::expect_plan_variant!(
+ node,
+ protobuf::physical_plan_node::PhysicalPlanType::LocalLimit,
+ "LocalLimitExec",
+ );
+ let input =
+ ctx.decode_required_child(limit.input.as_deref(), "LocalLimitExec", "input")?;
+ Ok(Arc::new(LocalLimitExec::new(input, limit.fetch as usize)))
+ }
}
/// A Limit stream skips `skip` rows, and then fetch up to `fetch` rows.
diff --git a/datafusion/proto/src/physical_plan/mod.rs b/datafusion/proto/src/physical_plan/mod.rs
index 13ed27104b..1d9e3ea3cd 100644
--- a/datafusion/proto/src/physical_plan/mod.rs
+++ b/datafusion/proto/src/physical_plan/mod.rs
@@ -793,11 +793,11 @@ pub trait PhysicalPlanNodeExt: Sized {
PhysicalPlanType::Repartition(repart) => {
self.try_into_repartition_physical_plan(repart, ctx, proto_converter)
}
- PhysicalPlanType::GlobalLimit(limit) => {
- self.try_into_global_limit_physical_plan(limit, ctx, proto_converter)
+ PhysicalPlanType::GlobalLimit(_) => {
+ GlobalLimitExec::try_from_proto(self.node(), &decode_ctx)
}
- PhysicalPlanType::LocalLimit(limit) => {
- self.try_into_local_limit_physical_plan(limit, ctx, proto_converter)
+ PhysicalPlanType::LocalLimit(_) => {
+ LocalLimitExec::try_from_proto(self.node(), &decode_ctx)
}
PhysicalPlanType::Window(window_agg) => {
self.try_into_window_physical_plan(window_agg, ctx, proto_converter)
@@ -917,22 +917,6 @@ pub trait PhysicalPlanNodeExt: Sized {
);
}
- if let Some(limit) = plan.downcast_ref::<GlobalLimitExec>() {
- return protobuf::PhysicalPlanNode::try_from_global_limit_exec(
- limit,
- codec,
- proto_converter,
- );
- }
-
- if let Some(limit) = plan.downcast_ref::<LocalLimitExec>() {
- return protobuf::PhysicalPlanNode::try_from_local_limit_exec(
- limit,
- codec,
- proto_converter,
- );
- }
-
if let Some(exec) = plan.downcast_ref::<HashJoinExec>() {
return protobuf::PhysicalPlanNode::try_from_hash_join_exec(
exec,
@@ -1480,35 +1464,50 @@ pub trait PhysicalPlanNodeExt: Sized {
Ok(Arc::new(repart_exec))
}
+ #[deprecated(
+ since = "55.0.0",
+ note = "unused by DataFusion; `GlobalLimitExec` deserializes itself via `GlobalLimitExec::try_from_proto`"
+ )]
fn try_into_global_limit_physical_plan(
&self,
limit: &protobuf::GlobalLimitExecNode,
ctx: &PhysicalPlanDecodeContext<'_>,
proto_converter: &dyn PhysicalProtoConverterExtension,
) -> Result<Arc<dyn ExecutionPlan>> {
- let input: Arc<dyn ExecutionPlan> =
- into_physical_plan(&limit.input, ctx, proto_converter)?;
- let fetch = if limit.fetch >= 0 {
- Some(limit.fetch as usize)
- } else {
- None
+ let node = protobuf::PhysicalPlanNode {
+ physical_plan_type: Some(PhysicalPlanType::GlobalLimit(Box::new(
+ limit.clone(),
+ ))),
};
- Ok(Arc::new(GlobalLimitExec::new(
- input,
- limit.skip as usize,
- fetch,
- )))
+ let decoder = ConverterPlanDecoder {
+ ctx,
+ proto_converter,
+ };
+ let decode_ctx = ExecutionPlanDecodeCtx::new(&decoder);
+ GlobalLimitExec::try_from_proto(&node, &decode_ctx)
}
+ #[deprecated(
+ since = "55.0.0",
+ note = "unused by DataFusion; `LocalLimitExec` deserializes itself via `LocalLimitExec::try_from_proto`"
+ )]
fn try_into_local_limit_physical_plan(
&self,
limit: &protobuf::LocalLimitExecNode,
ctx: &PhysicalPlanDecodeContext<'_>,
proto_converter: &dyn PhysicalProtoConverterExtension,
) -> Result<Arc<dyn ExecutionPlan>> {
- let input: Arc<dyn ExecutionPlan> =
- into_physical_plan(&limit.input, ctx, proto_converter)?;
- Ok(Arc::new(LocalLimitExec::new(input, limit.fetch as usize)))
+ let node = protobuf::PhysicalPlanNode {
+ physical_plan_type: Some(PhysicalPlanType::LocalLimit(Box::new(
+ limit.clone(),
+ ))),
+ };
+ let decoder = ConverterPlanDecoder {
+ ctx,
+ proto_converter,
+ };
+ let decode_ctx = ExecutionPlanDecodeCtx::new(&decoder);
+ LocalLimitExec::try_from_proto(&node, &decode_ctx)
}
fn try_into_window_physical_plan(
@@ -2833,49 +2832,42 @@ pub trait PhysicalPlanNodeExt: Sized {
.ok_or_else(|| internal_datafusion_err!("FilterExec is not serializable"))
}
+ #[deprecated(
+ since = "55.0.0",
+ note = "unused by DataFusion; `GlobalLimitExec` serializes itself via `ExecutionPlan::try_to_proto`"
+ )]
fn try_from_global_limit_exec(
limit: &GlobalLimitExec,
codec: &dyn PhysicalExtensionCodec,
proto_converter: &dyn PhysicalProtoConverterExtension,
) -> Result<protobuf::PhysicalPlanNode> {
- let input = protobuf::PhysicalPlanNode::try_from_physical_plan_with_converter(
- limit.input().to_owned(),
+ let encoder = ConverterPlanEncoder {
codec,
proto_converter,
- )?;
-
- Ok(protobuf::PhysicalPlanNode {
- physical_plan_type: Some(PhysicalPlanType::GlobalLimit(Box::new(
- protobuf::GlobalLimitExecNode {
- input: Some(Box::new(input)),
- skip: limit.skip() as u32,
- fetch: match limit.fetch() {
- Some(n) => n as i64,
- _ => -1, // no limit
- },
- },
- ))),
+ };
+ let encode_ctx = ExecutionPlanEncodeCtx::new(&encoder);
+ limit.try_to_proto(&encode_ctx)?.ok_or_else(|| {
+ internal_datafusion_err!("GlobalLimitExec is not serializable")
})
}
+ #[deprecated(
+ since = "55.0.0",
+ note = "unused by DataFusion; `LocalLimitExec` serializes itself via `ExecutionPlan::try_to_proto`"
+ )]
fn try_from_local_limit_exec(
limit: &LocalLimitExec,
codec: &dyn PhysicalExtensionCodec,
proto_converter: &dyn PhysicalProtoConverterExtension,
) -> Result<protobuf::PhysicalPlanNode> {
- let input = protobuf::PhysicalPlanNode::try_from_physical_plan_with_converter(
- limit.input().to_owned(),
+ let encoder = ConverterPlanEncoder {
codec,
proto_converter,
- )?;
- Ok(protobuf::PhysicalPlanNode {
- physical_plan_type: Some(PhysicalPlanType::LocalLimit(Box::new(
- protobuf::LocalLimitExecNode {
- input: Some(Box::new(input)),
- fetch: limit.fetch() as u32,
- },
- ))),
- })
+ };
+ let encode_ctx = ExecutionPlanEncodeCtx::new(&encoder);
+ limit
+ .try_to_proto(&encode_ctx)?
+ .ok_or_else(|| internal_datafusion_err!("LocalLimitExec is not serializable"))
}
fn try_from_hash_join_exec(
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org