From 312a347134f1cc38fc7139140b5fe524019832ad Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Tue, 11 Aug 2026 09:18:58 -0400 Subject: [PATCH] chore(proto): remove never-released deprecated PhysicalPlanNodeExt scaffolding Deletes the 59 `#[deprecated(since = "55.0.0")]` methods from the `PhysicalPlanNodeExt` trait. These were promoted from private inherent methods to public trait methods by #21929 (merged after the 54.x line was cut), so no released version of DataFusion ever exposed them. 55.0.0 would have been the first release to ship them, already deprecated. Also drops the imports that became unused (`DataSinkExec`, `BoundedWindowAggExec`, `SortMergeJoinExecNode`) and the test that only exercised the `try_into_projection_physical_plan` shim; `ProjectionExec` encode/decode stays covered by the existing `roundtrip_test` cases. No wire-format or dispatch changes. Part of #23494. Co-Authored-By: Claude Opus 5 --- datafusion/proto/src/physical_plan/mod.rs | 1560 +---------------- .../proto/tests/cases/plans/dispatch.rs | 63 +- 2 files changed, 92 insertions(+), 1531 deletions(-) diff --git a/datafusion/proto/src/physical_plan/mod.rs b/datafusion/proto/src/physical_plan/mod.rs index de684857f4446..da8873a208a46 100644 --- a/datafusion/proto/src/physical_plan/mod.rs +++ b/datafusion/proto/src/physical_plan/mod.rs @@ -26,8 +26,6 @@ use datafusion_catalog::memory::MemorySourceConfig; use datafusion_common::{ DataFusionError, Result, internal_datafusion_err, internal_err, not_impl_err, }; -use datafusion_datasource::sink::DataSinkExec; -use datafusion_datasource::source::DataSourceExec; use datafusion_datasource_arrow::source::ArrowSource; #[cfg(feature = "avro")] use datafusion_datasource_avro::source::AvroSource; @@ -79,7 +77,7 @@ use datafusion_physical_plan::sorts::sort::SortExec; use datafusion_physical_plan::sorts::sort_preserving_merge::SortPreservingMergeExec; use datafusion_physical_plan::union::{InterleaveExec, UnionExec}; use datafusion_physical_plan::unnest::UnnestExec; -use datafusion_physical_plan::windows::{BoundedWindowAggExec, WindowAggExec}; +use datafusion_physical_plan::windows::WindowAggExec; use datafusion_physical_plan::{ExecutionPlan, PhysicalExpr}; use prost::Message; use prost::bytes::BufMut; @@ -88,7 +86,7 @@ use crate::convert_required; use crate::physical_plan::from_proto::parse_physical_expr_with_converter; use crate::physical_plan::to_proto::serialize_physical_expr_with_converter; use crate::protobuf::physical_plan_node::PhysicalPlanType; -use crate::protobuf::{self, SortMergeJoinExecNode, proto_error}; +use crate::protobuf::{self, proto_error}; pub mod from_proto; pub mod to_proto; @@ -1280,1429 +1278,106 @@ pub trait PhysicalPlanNodeExt: Sized { } } - #[deprecated( - since = "55.0.0", - note = "unused by DataFusion; `ExplainExec` deserializes itself via `ExplainExec::try_from_proto`" - )] - fn try_into_explain_physical_plan( - &self, - _explain: &protobuf::ExplainExecNode, - ctx: &PhysicalPlanDecodeContext<'_>, - proto_converter: &dyn PhysicalProtoConverterExtension, - ) -> Result> { - let plan_decoder = ConverterPlanDecoder { - ctx, - proto_converter, - }; - let decode_ctx = ExecutionPlanDecodeCtx::new(&plan_decoder); - ExplainExec::try_from_proto(self.node(), &decode_ctx) - } - - #[deprecated( - since = "55.0.0", - note = "unused by DataFusion; `ProjectionExec` deserializes itself via `ProjectionExec::try_from_proto`" - )] - fn try_into_projection_physical_plan( - &self, - projection: &protobuf::ProjectionExecNode, - ctx: &PhysicalPlanDecodeContext<'_>, - proto_converter: &dyn PhysicalProtoConverterExtension, - ) -> Result> { - let decoder = ConverterPlanDecoder { - ctx, - proto_converter, - }; - let decode_ctx = ExecutionPlanDecodeCtx::new(&decoder); - // `try_from_proto` takes the enclosing `PhysicalPlanNode`, while this - // deprecated method is driven by the `ProjectionExecNode` argument. - // Re-wrap the argument so the decoded plan keeps depending on it rather - // than on `self`, which a caller may not have kept in sync. - let node = protobuf::PhysicalPlanNode { - physical_plan_type: Some(PhysicalPlanType::Projection(Box::new( - projection.clone(), - ))), - }; - ProjectionExec::try_from_proto(&node, &decode_ctx) - } - - #[deprecated( - since = "55.0.0", - note = "unused by DataFusion; `FilterExec` deserializes itself via `FilterExec::try_from_proto`" - )] - fn try_into_filter_physical_plan( - &self, - filter: &protobuf::FilterExecNode, - ctx: &PhysicalPlanDecodeContext<'_>, - proto_converter: &dyn PhysicalProtoConverterExtension, - ) -> Result> { - let node = protobuf::PhysicalPlanNode { - physical_plan_type: Some(PhysicalPlanType::Filter(Box::new(filter.clone()))), - }; - let decoder = ConverterPlanDecoder { - ctx, - proto_converter, - }; - let decode_ctx = ExecutionPlanDecodeCtx::new(&decoder); - FilterExec::try_from_proto(&node, &decode_ctx) - } - - #[deprecated( - since = "55.0.0", - note = "unused by DataFusion; `CsvSource` deserializes itself via `CsvSource::try_from_proto`" - )] - fn try_into_csv_scan_physical_plan( - &self, - scan: &protobuf::CsvScanExecNode, - ctx: &PhysicalPlanDecodeContext<'_>, - proto_converter: &dyn PhysicalProtoConverterExtension, - ) -> Result> { - let node = protobuf::PhysicalPlanNode { - physical_plan_type: Some(PhysicalPlanType::CsvScan(scan.clone())), - }; - let decoder = ConverterPlanDecoder { - ctx, - proto_converter, - }; - let decode_ctx = ExecutionPlanDecodeCtx::new(&decoder); - CsvSource::try_from_proto(&node, &decode_ctx) - } - - #[deprecated( - since = "55.0.0", - note = "unused by DataFusion; `JsonSource` deserializes itself via `JsonSource::try_from_proto`" - )] - fn try_into_json_scan_physical_plan( - &self, - scan: &protobuf::JsonScanExecNode, - ctx: &PhysicalPlanDecodeContext<'_>, - proto_converter: &dyn PhysicalProtoConverterExtension, - ) -> Result> { - let node = protobuf::PhysicalPlanNode { - physical_plan_type: Some(PhysicalPlanType::JsonScan(scan.clone())), - }; - let decoder = ConverterPlanDecoder { - ctx, - proto_converter, - }; - let decode_ctx = ExecutionPlanDecodeCtx::new(&decoder); - JsonSource::try_from_proto(&node, &decode_ctx) - } - - #[deprecated( - since = "55.0.0", - note = "unused by DataFusion; `ArrowSource` deserializes itself via `ArrowSource::try_from_proto`" - )] - fn try_into_arrow_scan_physical_plan( - &self, - scan: &protobuf::ArrowScanExecNode, - ctx: &PhysicalPlanDecodeContext<'_>, - proto_converter: &dyn PhysicalProtoConverterExtension, - ) -> Result> { - let node = protobuf::PhysicalPlanNode { - physical_plan_type: Some(PhysicalPlanType::ArrowScan(scan.clone())), - }; - let decoder = ConverterPlanDecoder { - ctx, - proto_converter, - }; - let decode_ctx = ExecutionPlanDecodeCtx::new(&decoder); - ArrowSource::try_from_proto(&node, &decode_ctx) - } - - #[cfg_attr(not(feature = "parquet"), expect(unused_variables))] - #[deprecated( - since = "55.0.0", - note = "unused by DataFusion; `ParquetSource` deserializes itself via `ParquetSource::try_from_proto`" - )] - fn try_into_parquet_scan_physical_plan( - &self, - scan: &protobuf::ParquetScanExecNode, - ctx: &PhysicalPlanDecodeContext<'_>, - proto_converter: &dyn PhysicalProtoConverterExtension, - ) -> Result> { - #[cfg(feature = "parquet")] - { - let node = protobuf::PhysicalPlanNode { - physical_plan_type: Some(PhysicalPlanType::ParquetScan(scan.clone())), - }; - let decoder = ConverterPlanDecoder { - ctx, - proto_converter, - }; - let decode_ctx = ExecutionPlanDecodeCtx::new(&decoder); - ParquetSource::try_from_proto(&node, &decode_ctx) - } - - #[cfg(not(feature = "parquet"))] - not_impl_err!( - "Unable to process a Parquet PhysicalPlan when the `parquet` feature is not enabled" - ) - } - - #[cfg_attr(not(feature = "avro"), expect(unused_variables))] - #[deprecated( - since = "55.0.0", - note = "unused by DataFusion; `AvroSource` deserializes itself via `AvroSource::try_from_proto`" - )] - fn try_into_avro_scan_physical_plan( - &self, - scan: &protobuf::AvroScanExecNode, - ctx: &PhysicalPlanDecodeContext<'_>, - proto_converter: &dyn PhysicalProtoConverterExtension, - ) -> Result> { - #[cfg(feature = "avro")] - { - let node = protobuf::PhysicalPlanNode { - physical_plan_type: Some(PhysicalPlanType::AvroScan(scan.clone())), - }; - let decoder = ConverterPlanDecoder { - ctx, - proto_converter, - }; - let decode_ctx = ExecutionPlanDecodeCtx::new(&decoder); - AvroSource::try_from_proto(&node, &decode_ctx) - } - - #[cfg(not(feature = "avro"))] - not_impl_err!( - "Unable to process an Avro PhysicalPlan when the `avro` feature is not enabled" - ) - } - - #[deprecated( - since = "55.0.0", - note = "unused by DataFusion; `MemorySourceConfig` deserializes itself via `MemorySourceConfig::try_from_proto`" - )] - fn try_into_memory_scan_physical_plan( - &self, - scan: &protobuf::MemoryScanExecNode, - ctx: &PhysicalPlanDecodeContext<'_>, - proto_converter: &dyn PhysicalProtoConverterExtension, - ) -> Result> { - let node = protobuf::PhysicalPlanNode { - physical_plan_type: Some(PhysicalPlanType::MemoryScan(scan.clone())), - }; - let decoder = ConverterPlanDecoder { - ctx, - proto_converter, - }; - let decode_ctx = ExecutionPlanDecodeCtx::new(&decoder); - MemorySourceConfig::try_from_proto(&node, &decode_ctx) - } - - #[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> { - 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> { - 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) - } - - #[deprecated( - since = "55.0.0", - note = "unused by DataFusion; `RepartitionExec` deserializes itself via `RepartitionExec::try_from_proto`" - )] - fn try_into_repartition_physical_plan( - &self, - repart: &protobuf::RepartitionExecNode, - ctx: &PhysicalPlanDecodeContext<'_>, - proto_converter: &dyn PhysicalProtoConverterExtension, - ) -> Result> { - let node = protobuf::PhysicalPlanNode { - physical_plan_type: Some(PhysicalPlanType::Repartition(Box::new( - repart.clone(), - ))), - }; - let decoder = ConverterPlanDecoder { - ctx, - proto_converter, - }; - let decode_ctx = ExecutionPlanDecodeCtx::new(&decoder); - RepartitionExec::try_from_proto(&node, &decode_ctx) - } - - #[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> { - let node = protobuf::PhysicalPlanNode { - physical_plan_type: Some(PhysicalPlanType::GlobalLimit(Box::new( - limit.clone(), - ))), - }; - 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> { - 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) - } - - #[deprecated( - since = "55.0.0", - note = "unused by DataFusion; window plans deserialize via `WindowAggExec::try_from_proto`" - )] - fn try_into_window_physical_plan( - &self, - window_agg: &protobuf::WindowAggExecNode, - ctx: &PhysicalPlanDecodeContext<'_>, - proto_converter: &dyn PhysicalProtoConverterExtension, - ) -> Result> { - let node = protobuf::PhysicalPlanNode { - physical_plan_type: Some(PhysicalPlanType::Window(Box::new( - window_agg.clone(), - ))), - }; - let decoder = ConverterPlanDecoder { - ctx, - proto_converter, - }; - let decode_ctx = ExecutionPlanDecodeCtx::new(&decoder); - WindowAggExec::try_from_proto(&node, &decode_ctx) - } - - #[deprecated( - since = "55.0.0", - note = "unused by DataFusion; `AggregateExec` deserializes itself via `AggregateExec::try_from_proto`" - )] - fn try_into_aggregate_physical_plan( - &self, - hash_agg: &protobuf::AggregateExecNode, - ctx: &PhysicalPlanDecodeContext<'_>, - proto_converter: &dyn PhysicalProtoConverterExtension, - ) -> Result> { - let node = protobuf::PhysicalPlanNode { - physical_plan_type: Some(PhysicalPlanType::Aggregate(Box::new( - hash_agg.clone(), - ))), - }; - let decoder = ConverterPlanDecoder { - ctx, - proto_converter, - }; - let decode_ctx = ExecutionPlanDecodeCtx::new(&decoder); - AggregateExec::try_from_proto(&node, &decode_ctx) - } - - #[deprecated( - since = "55.0.0", - note = "unused by DataFusion; `HashJoinExec` deserializes itself via `HashJoinExec::try_from_proto`" - )] - fn try_into_hash_join_physical_plan( - &self, - hashjoin: &protobuf::HashJoinExecNode, - ctx: &PhysicalPlanDecodeContext<'_>, - proto_converter: &dyn PhysicalProtoConverterExtension, - ) -> Result> { - let node = protobuf::PhysicalPlanNode { - physical_plan_type: Some(PhysicalPlanType::HashJoin(Box::new( - hashjoin.clone(), - ))), - }; - let decoder = ConverterPlanDecoder { - ctx, - proto_converter, - }; - let decode_ctx = ExecutionPlanDecodeCtx::new(&decoder); - HashJoinExec::try_from_proto(&node, &decode_ctx) - } - - #[deprecated( - since = "55.0.0", - note = "unused by DataFusion; `SymmetricHashJoinExec` deserializes itself via `SymmetricHashJoinExec::try_from_proto`" - )] - fn try_into_symmetric_hash_join_physical_plan( - &self, - sym_join: &protobuf::SymmetricHashJoinExecNode, - ctx: &PhysicalPlanDecodeContext<'_>, - proto_converter: &dyn PhysicalProtoConverterExtension, - ) -> Result> { - let node = protobuf::PhysicalPlanNode { - physical_plan_type: Some(PhysicalPlanType::SymmetricHashJoin(Box::new( - sym_join.clone(), - ))), - }; - let decoder = ConverterPlanDecoder { - ctx, - proto_converter, - }; - let decode_ctx = ExecutionPlanDecodeCtx::new(&decoder); - SymmetricHashJoinExec::try_from_proto(&node, &decode_ctx) - } - - #[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> { - 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> { - 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( - since = "55.0.0", - note = "unused by DataFusion; `CrossJoinExec` deserializes itself via `CrossJoinExec::try_from_proto`" - )] - fn try_into_cross_join_physical_plan( - &self, - crossjoin: &protobuf::CrossJoinExecNode, - ctx: &PhysicalPlanDecodeContext<'_>, - proto_converter: &dyn PhysicalProtoConverterExtension, - ) -> Result> { - let node = protobuf::PhysicalPlanNode { - physical_plan_type: Some(PhysicalPlanType::CrossJoin(Box::new( - crossjoin.clone(), - ))), - }; - let decoder = ConverterPlanDecoder { - ctx, - proto_converter, - }; - let decode_ctx = ExecutionPlanDecodeCtx::new(&decoder); - CrossJoinExec::try_from_proto(&node, &decode_ctx) - } - - #[deprecated( - since = "55.0.0", - note = "unused by DataFusion; `EmptyExec` deserializes itself via `EmptyExec::try_from_proto`" - )] - fn try_into_empty_physical_plan( - &self, - empty: &protobuf::EmptyExecNode, - ctx: &PhysicalPlanDecodeContext<'_>, - proto_converter: &dyn PhysicalProtoConverterExtension, - ) -> Result> { - let node = protobuf::PhysicalPlanNode { - physical_plan_type: Some(PhysicalPlanType::Empty(empty.clone())), - }; - let decoder = ConverterPlanDecoder { - ctx, - proto_converter, - }; - let decode_ctx = ExecutionPlanDecodeCtx::new(&decoder); - EmptyExec::try_from_proto(&node, &decode_ctx) - } - - #[deprecated( - since = "55.0.0", - note = "unused by DataFusion; `PlaceholderRowExec` deserializes itself via `PlaceholderRowExec::try_from_proto`" - )] - fn try_into_placeholder_row_physical_plan( - &self, - placeholder: &protobuf::PlaceholderRowExecNode, - ctx: &PhysicalPlanDecodeContext<'_>, - ) -> Result> { - let node = protobuf::PhysicalPlanNode { - physical_plan_type: Some(PhysicalPlanType::PlaceholderRow( - placeholder.clone(), - )), - }; - let proto_converter = DefaultPhysicalProtoConverter {}; - let decoder = ConverterPlanDecoder { - ctx, - proto_converter: &proto_converter, - }; - let decode_ctx = ExecutionPlanDecodeCtx::new(&decoder); - PlaceholderRowExec::try_from_proto(&node, &decode_ctx) - } - - #[deprecated( - since = "55.0.0", - note = "unused by DataFusion; `SortExec` deserializes itself via `SortExec::try_from_proto`" - )] - fn try_into_sort_physical_plan( - &self, - sort: &protobuf::SortExecNode, - ctx: &PhysicalPlanDecodeContext<'_>, - proto_converter: &dyn PhysicalProtoConverterExtension, - ) -> Result> { - let node = protobuf::PhysicalPlanNode { - physical_plan_type: Some(PhysicalPlanType::Sort(Box::new(sort.clone()))), - }; - let decoder = ConverterPlanDecoder { - ctx, - proto_converter, - }; - let decode_ctx = ExecutionPlanDecodeCtx::new(&decoder); - SortExec::try_from_proto(&node, &decode_ctx) - } - - #[deprecated( - since = "55.0.0", - note = "unused by DataFusion; `SortPreservingMergeExec` deserializes itself via `SortPreservingMergeExec::try_from_proto`" - )] - fn try_into_sort_preserving_merge_physical_plan( - &self, - sort: &protobuf::SortPreservingMergeExecNode, - ctx: &PhysicalPlanDecodeContext<'_>, - proto_converter: &dyn PhysicalProtoConverterExtension, - ) -> Result> { - let node = protobuf::PhysicalPlanNode { - physical_plan_type: Some(PhysicalPlanType::SortPreservingMerge(Box::new( - sort.clone(), - ))), - }; - let decoder = ConverterPlanDecoder { - ctx, - proto_converter, - }; - let decode_ctx = ExecutionPlanDecodeCtx::new(&decoder); - SortPreservingMergeExec::try_from_proto(&node, &decode_ctx) - } - - fn try_into_extension_physical_plan( - &self, - extension: &protobuf::PhysicalExtensionNode, - ctx: &PhysicalPlanDecodeContext<'_>, - proto_converter: &dyn PhysicalProtoConverterExtension, - ) -> Result> { - let inputs: Vec> = extension - .inputs - .iter() - .map(|i| proto_converter.proto_to_execution_plan(i, ctx)) - .collect::>()?; - - let extension_node = ctx.codec().try_decode( - extension.node.as_slice(), - &inputs, - ctx.task_ctx(), - proto_converter, - )?; - - Ok(extension_node) - } - - #[deprecated( - since = "55.0.0", - note = "unused by DataFusion; `NestedLoopJoinExec` deserializes itself via `NestedLoopJoinExec::try_from_proto`" - )] - fn try_into_nested_loop_join_physical_plan( - &self, - join: &protobuf::NestedLoopJoinExecNode, - ctx: &PhysicalPlanDecodeContext<'_>, - proto_converter: &dyn PhysicalProtoConverterExtension, - ) -> Result> { - let node = protobuf::PhysicalPlanNode { - physical_plan_type: Some(PhysicalPlanType::NestedLoopJoin(Box::new( - join.clone(), - ))), - }; - let decoder = ConverterPlanDecoder { - ctx, - proto_converter, - }; - let decode_ctx = ExecutionPlanDecodeCtx::new(&decoder); - NestedLoopJoinExec::try_from_proto(&node, &decode_ctx) - } - - #[deprecated( - since = "55.0.0", - note = "unused by DataFusion; `AnalyzeExec` deserializes itself via `AnalyzeExec::try_from_proto`" - )] - fn try_into_analyze_physical_plan( - &self, - _analyze: &protobuf::AnalyzeExecNode, - ctx: &PhysicalPlanDecodeContext<'_>, - proto_converter: &dyn PhysicalProtoConverterExtension, - ) -> Result> { - let plan_decoder = ConverterPlanDecoder { - ctx, - proto_converter, - }; - let decode_ctx = ExecutionPlanDecodeCtx::new(&plan_decoder); - AnalyzeExec::try_from_proto(self.node(), &decode_ctx) - } - - #[deprecated( - since = "55.0.0", - note = "unused by DataFusion; `JsonSink` deserializes itself via `JsonSink::try_from_proto`" - )] - fn try_into_json_sink_physical_plan( - &self, - sink: &protobuf::JsonSinkExecNode, - ctx: &PhysicalPlanDecodeContext<'_>, - proto_converter: &dyn PhysicalProtoConverterExtension, - ) -> Result> { - let node = protobuf::PhysicalPlanNode { - physical_plan_type: Some(PhysicalPlanType::JsonSink(Box::new(sink.clone()))), - }; - let decoder = ConverterPlanDecoder { - ctx, - proto_converter, - }; - let decode_ctx = ExecutionPlanDecodeCtx::new(&decoder); - JsonSink::try_from_proto(&node, &decode_ctx) - } - - #[deprecated( - since = "55.0.0", - note = "unused by DataFusion; `CsvSink` deserializes itself via `CsvSink::try_from_proto`" - )] - fn try_into_csv_sink_physical_plan( - &self, - sink: &protobuf::CsvSinkExecNode, - ctx: &PhysicalPlanDecodeContext<'_>, - proto_converter: &dyn PhysicalProtoConverterExtension, - ) -> Result> { - let node = protobuf::PhysicalPlanNode { - physical_plan_type: Some(PhysicalPlanType::CsvSink(Box::new(sink.clone()))), - }; - let decoder = ConverterPlanDecoder { - ctx, - proto_converter, - }; - let decode_ctx = ExecutionPlanDecodeCtx::new(&decoder); - CsvSink::try_from_proto(&node, &decode_ctx) - } - - #[cfg_attr(not(feature = "parquet"), expect(unused_variables))] - #[deprecated( - since = "55.0.0", - note = "unused by DataFusion; `ParquetSink` deserializes itself via `ParquetSink::try_from_proto`" - )] - fn try_into_parquet_sink_physical_plan( - &self, - sink: &protobuf::ParquetSinkExecNode, - ctx: &PhysicalPlanDecodeContext<'_>, - proto_converter: &dyn PhysicalProtoConverterExtension, - ) -> Result> { - #[cfg(feature = "parquet")] - { - let node = protobuf::PhysicalPlanNode { - physical_plan_type: Some(PhysicalPlanType::ParquetSink(Box::new( - sink.clone(), - ))), - }; - let decoder = ConverterPlanDecoder { - ctx, - proto_converter, - }; - let decode_ctx = ExecutionPlanDecodeCtx::new(&decoder); - ParquetSink::try_from_proto(&node, &decode_ctx) - } - #[cfg(not(feature = "parquet"))] - not_impl_err!("ParquetSink requires the `parquet` feature") - } - - #[deprecated( - since = "55.0.0", - note = "unused by DataFusion; `UnnestExec` deserializes itself via `UnnestExec::try_from_proto`" - )] - fn try_into_unnest_physical_plan( - &self, - unnest: &protobuf::UnnestExecNode, - ctx: &PhysicalPlanDecodeContext<'_>, - proto_converter: &dyn PhysicalProtoConverterExtension, - ) -> Result> { - let node = protobuf::PhysicalPlanNode { - physical_plan_type: Some(PhysicalPlanType::Unnest(Box::new(unnest.clone()))), - }; - let decoder = ConverterPlanDecoder { - ctx, - proto_converter, - }; - let decode_ctx = ExecutionPlanDecodeCtx::new(&decoder); - UnnestExec::try_from_proto(&node, &decode_ctx) - } - - fn generate_series_name_to_str(name: protobuf::GenerateSeriesName) -> &'static str { - match name { - protobuf::GenerateSeriesName::GsGenerateSeries => "generate_series", - protobuf::GenerateSeriesName::GsRange => "range", - } - } - #[deprecated( - since = "55.0.0", - note = "unused by DataFusion; `SortMergeJoinExec` deserializes itself via `SortMergeJoinExec::try_from_proto`" - )] - fn try_into_sort_join( - &self, - sort_join: &SortMergeJoinExecNode, - ctx: &PhysicalPlanDecodeContext<'_>, - proto_converter: &dyn PhysicalProtoConverterExtension, - ) -> Result> { - let node = protobuf::PhysicalPlanNode { - physical_plan_type: Some(PhysicalPlanType::SortMergeJoin(Box::new( - sort_join.clone(), - ))), - }; - let decoder = ConverterPlanDecoder { - ctx, - proto_converter, - }; - let decode_ctx = ExecutionPlanDecodeCtx::new(&decoder); - SortMergeJoinExec::try_from_proto(&node, &decode_ctx) - } - - fn try_into_generate_series_physical_plan( - &self, - generate_series: &protobuf::GenerateSeriesNode, - ) -> Result> { - let schema: SchemaRef = Arc::new(convert_required!(generate_series.schema)?); - - let args = match &generate_series.args { - Some(protobuf::generate_series_node::Args::ContainsNull(args)) => { - GenSeriesArgs::ContainsNull { - name: protobuf::PhysicalPlanNode::generate_series_name_to_str( - args.name(), - ), - } - } - Some(protobuf::generate_series_node::Args::Int64Args(args)) => { - GenSeriesArgs::Int64Args { - start: args.start, - end: args.end, - step: args.step, - include_end: args.include_end, - name: protobuf::PhysicalPlanNode::generate_series_name_to_str( - args.name(), - ), - } - } - Some(protobuf::generate_series_node::Args::TimestampArgs(args)) => { - let step_proto = args.step.as_ref().ok_or_else(|| { - internal_datafusion_err!("Missing step in TimestampArgs") - })?; - let step = IntervalMonthDayNanoType::make_value( - step_proto.months, - step_proto.days, - step_proto.nanos, - ); - GenSeriesArgs::TimestampArgs { - start: args.start, - end: args.end, - step, - tz: args.tz.as_ref().map(|s| Arc::from(s.as_str())), - include_end: args.include_end, - name: protobuf::PhysicalPlanNode::generate_series_name_to_str( - args.name(), - ), - } - } - Some(protobuf::generate_series_node::Args::DateArgs(args)) => { - let step_proto = args.step.as_ref().ok_or_else(|| { - internal_datafusion_err!("Missing step in DateArgs") - })?; - let step = IntervalMonthDayNanoType::make_value( - step_proto.months, - step_proto.days, - step_proto.nanos, - ); - GenSeriesArgs::DateArgs { - start: args.start, - end: args.end, - step, - include_end: args.include_end, - name: protobuf::PhysicalPlanNode::generate_series_name_to_str( - args.name(), - ), - } - } - None => return internal_err!("Missing args in GenerateSeriesNode"), - }; - - let table = GenerateSeriesTable::new(Arc::clone(&schema), args); - let generator = table.as_generator(generate_series.target_batch_size as usize)?; - - 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> { - 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) - } - - #[deprecated( - since = "55.0.0", - note = "unused by DataFusion; `AsyncFuncExec` deserializes itself via `AsyncFuncExec::try_from_proto`" - )] - fn try_into_async_func_physical_plan( - &self, - async_func: &protobuf::AsyncFuncExecNode, - ctx: &PhysicalPlanDecodeContext<'_>, - proto_converter: &dyn PhysicalProtoConverterExtension, - ) -> Result> { - let node = protobuf::PhysicalPlanNode { - physical_plan_type: Some(PhysicalPlanType::AsyncFunc(Box::new( - async_func.clone(), - ))), - }; - let decoder = ConverterPlanDecoder { - ctx, - proto_converter, - }; - let decode_ctx = ExecutionPlanDecodeCtx::new(&decoder); - AsyncFuncExec::try_from_proto(&node, &decode_ctx) - } - - #[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> { - 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) - } - - #[deprecated( - since = "55.0.0", - note = "unused by DataFusion; `ScalarSubqueryExec` deserializes itself via `ScalarSubqueryExec::try_from_proto`" - )] - fn try_into_scalar_subquery_physical_plan( - &self, - sq: &protobuf::ScalarSubqueryExecNode, - ctx: &PhysicalPlanDecodeContext<'_>, - proto_converter: &dyn PhysicalProtoConverterExtension, - ) -> Result> { - let node = protobuf::PhysicalPlanNode { - physical_plan_type: Some(PhysicalPlanType::ScalarSubquery(Box::new( - sq.clone(), - ))), - }; - let decoder = ConverterPlanDecoder { - ctx, - proto_converter, - }; - let decode_ctx = ExecutionPlanDecodeCtx::new(&decoder); - ScalarSubqueryExec::try_from_proto(&node, &decode_ctx) - } - - #[deprecated( - since = "55.0.0", - note = "unused by DataFusion; `ExplainExec` serializes itself via `ExecutionPlan::try_to_proto`" - )] - fn try_from_explain_exec( - exec: &ExplainExec, - codec: &dyn PhysicalExtensionCodec, - ) -> Result { - let proto_converter = DefaultPhysicalProtoConverter {}; - let plan_encoder = ConverterPlanEncoder { - codec, - proto_converter: &proto_converter, - }; - let encode_ctx = ExecutionPlanEncodeCtx::new(&plan_encoder); - exec.try_to_proto(&encode_ctx)?.ok_or_else(|| { - internal_datafusion_err!("ExplainExec did not serialize itself") - }) - } - - #[deprecated( - since = "55.0.0", - note = "unused by DataFusion; `ProjectionExec` serializes itself via `ExecutionPlan::try_to_proto`" - )] - fn try_from_projection_exec( - exec: &ProjectionExec, - codec: &dyn PhysicalExtensionCodec, - proto_converter: &dyn PhysicalProtoConverterExtension, - ) -> Result { - let encoder = ConverterPlanEncoder { - codec, - proto_converter, - }; - let ctx = ExecutionPlanEncodeCtx::new(&encoder); - exec.try_to_proto(&ctx)?.ok_or_else(|| { - internal_datafusion_err!("ProjectionExec::try_to_proto returned None") - }) - } - - #[deprecated( - since = "55.0.0", - note = "unused by DataFusion; `AnalyzeExec` serializes itself via `ExecutionPlan::try_to_proto`" - )] - fn try_from_analyze_exec( - exec: &AnalyzeExec, - codec: &dyn PhysicalExtensionCodec, - proto_converter: &dyn PhysicalProtoConverterExtension, - ) -> Result { - let plan_encoder = ConverterPlanEncoder { - codec, - proto_converter, - }; - let encode_ctx = ExecutionPlanEncodeCtx::new(&plan_encoder); - exec.try_to_proto(&encode_ctx)?.ok_or_else(|| { - internal_datafusion_err!("AnalyzeExec did not serialize itself") - }) - } - - #[deprecated( - since = "55.0.0", - note = "unused by DataFusion; `FilterExec` serializes itself via `ExecutionPlan::try_to_proto`" - )] - fn try_from_filter_exec( - exec: &FilterExec, - codec: &dyn PhysicalExtensionCodec, - proto_converter: &dyn PhysicalProtoConverterExtension, - ) -> Result { - let encoder = ConverterPlanEncoder { - codec, - proto_converter, - }; - let encode_ctx = ExecutionPlanEncodeCtx::new(&encoder); - exec.try_to_proto(&encode_ctx)? - .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 { - let encoder = ConverterPlanEncoder { - codec, - proto_converter, - }; - 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 { - let encoder = ConverterPlanEncoder { - codec, - proto_converter, - }; - let encode_ctx = ExecutionPlanEncodeCtx::new(&encoder); - limit - .try_to_proto(&encode_ctx)? - .ok_or_else(|| internal_datafusion_err!("LocalLimitExec is not serializable")) - } - - #[deprecated( - since = "55.0.0", - note = "unused by DataFusion; `HashJoinExec` serializes itself via `ExecutionPlan::try_to_proto`" - )] - fn try_from_hash_join_exec( - exec: &HashJoinExec, - codec: &dyn PhysicalExtensionCodec, - proto_converter: &dyn PhysicalProtoConverterExtension, - ) -> Result { - let encoder = ConverterPlanEncoder { - codec, - proto_converter, - }; - let encode_ctx = ExecutionPlanEncodeCtx::new(&encoder); - exec.try_to_proto(&encode_ctx)? - .ok_or_else(|| internal_datafusion_err!("HashJoinExec is not serializable")) - } - - #[deprecated( - since = "55.0.0", - note = "unused by DataFusion; `SymmetricHashJoinExec` serializes itself via `ExecutionPlan::try_to_proto`" - )] - fn try_from_symmetric_hash_join_exec( - exec: &SymmetricHashJoinExec, - codec: &dyn PhysicalExtensionCodec, - proto_converter: &dyn PhysicalProtoConverterExtension, - ) -> Result { - let encoder = ConverterPlanEncoder { - codec, - proto_converter, - }; - let encode_ctx = ExecutionPlanEncodeCtx::new(&encoder); - exec.try_to_proto(&encode_ctx)?.ok_or_else(|| { - internal_datafusion_err!("SymmetricHashJoinExec is not serializable") - }) - } - - #[deprecated( - since = "55.0.0", - note = "unused by DataFusion; `SortMergeJoinExec` serializes itself via `ExecutionPlan::try_to_proto`" - )] - fn try_from_sort_merge_join_exec( - exec: &SortMergeJoinExec, - codec: &dyn PhysicalExtensionCodec, - proto_converter: &dyn PhysicalProtoConverterExtension, - ) -> Result { - let encoder = ConverterPlanEncoder { - codec, - proto_converter, - }; - let encode_ctx = ExecutionPlanEncodeCtx::new(&encoder); - exec.try_to_proto(&encode_ctx)?.ok_or_else(|| { - internal_datafusion_err!("SortMergeJoinExec is not serializable") - }) - } - - #[deprecated( - since = "55.0.0", - note = "unused by DataFusion; `CrossJoinExec` serializes itself via `ExecutionPlan::try_to_proto`" - )] - fn try_from_cross_join_exec( - exec: &CrossJoinExec, - codec: &dyn PhysicalExtensionCodec, - proto_converter: &dyn PhysicalProtoConverterExtension, - ) -> Result { - let encoder = ConverterPlanEncoder { - codec, - proto_converter, - }; - let encode_ctx = ExecutionPlanEncodeCtx::new(&encoder); - exec.try_to_proto(&encode_ctx)? - .ok_or_else(|| internal_datafusion_err!("CrossJoinExec is not serializable")) - } - - #[deprecated( - since = "55.0.0", - note = "unused by DataFusion; `AggregateExec` serializes itself via `ExecutionPlan::try_to_proto`" - )] - fn try_from_aggregate_exec( - exec: &AggregateExec, - codec: &dyn PhysicalExtensionCodec, - proto_converter: &dyn PhysicalProtoConverterExtension, - ) -> Result { - let encoder = ConverterPlanEncoder { - codec, - proto_converter, - }; - let encode_ctx = ExecutionPlanEncodeCtx::new(&encoder); - exec.try_to_proto(&encode_ctx)? - .ok_or_else(|| internal_datafusion_err!("AggregateExec is not serializable")) - } - - #[deprecated( - since = "55.0.0", - note = "unused by DataFusion; `EmptyExec` serializes itself via `ExecutionPlan::try_to_proto`" - )] - fn try_from_empty_exec( - empty: &EmptyExec, - codec: &dyn PhysicalExtensionCodec, - ) -> Result { - let proto_converter = DefaultPhysicalProtoConverter {}; - let encoder = ConverterPlanEncoder { - codec, - proto_converter: &proto_converter, - }; - let ctx = ExecutionPlanEncodeCtx::new(&encoder); - empty.try_to_proto(&ctx)?.ok_or_else(|| { - internal_datafusion_err!("EmptyExec::try_to_proto returned None") - }) - } - - #[deprecated( - since = "55.0.0", - note = "unused by DataFusion; `PlaceholderRowExec` serializes itself via `ExecutionPlan::try_to_proto`" - )] - fn try_from_placeholder_row_exec( - placeholder: &PlaceholderRowExec, - codec: &dyn PhysicalExtensionCodec, - ) -> Result { - let proto_converter = DefaultPhysicalProtoConverter {}; - let encoder = ConverterPlanEncoder { - codec, - proto_converter: &proto_converter, - }; - let ctx = ExecutionPlanEncodeCtx::new(&encoder); - placeholder.try_to_proto(&ctx)?.ok_or_else(|| { - internal_datafusion_err!("PlaceholderRowExec::try_to_proto returned None") - }) - } - - #[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 { - let encoder = ConverterPlanEncoder { - codec, - proto_converter, - }; - let encode_ctx = ExecutionPlanEncodeCtx::new(&encoder); - coalesce_batches.try_to_proto(&encode_ctx)?.ok_or_else(|| { - internal_datafusion_err!("CoalesceBatchesExec is not serializable") - }) - } - - #[deprecated( - since = "55.0.0", - note = "unused by DataFusion; `DataSourceExec` serializes itself via `ExecutionPlan::try_to_proto`" - )] - fn try_from_data_source_exec( - data_source_exec: &DataSourceExec, - codec: &dyn PhysicalExtensionCodec, - proto_converter: &dyn PhysicalProtoConverterExtension, - ) -> Result> { - let encoder = ConverterPlanEncoder { - codec, - proto_converter, - }; - let encode_ctx = ExecutionPlanEncodeCtx::new(&encoder); - data_source_exec.try_to_proto(&encode_ctx) - } - - #[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 { - let encoder = ConverterPlanEncoder { - codec, - proto_converter, - }; - let encode_ctx = ExecutionPlanEncodeCtx::new(&encoder); - exec.try_to_proto(&encode_ctx)?.ok_or_else(|| { - internal_datafusion_err!("CoalescePartitionsExec is not serializable") - }) - } - - #[deprecated( - since = "55.0.0", - note = "unused by DataFusion; `RepartitionExec` serializes itself via `ExecutionPlan::try_to_proto`" - )] - fn try_from_repartition_exec( - exec: &RepartitionExec, - codec: &dyn PhysicalExtensionCodec, - proto_converter: &dyn PhysicalProtoConverterExtension, - ) -> Result { - let encoder = ConverterPlanEncoder { - codec, - proto_converter, - }; - let encode_ctx = ExecutionPlanEncodeCtx::new(&encoder); - exec.try_to_proto(&encode_ctx)?.ok_or_else(|| { - internal_datafusion_err!("RepartitionExec is not serializable") - }) - } - - #[deprecated( - since = "55.0.0", - note = "unused by DataFusion; `SortExec` serializes itself via `ExecutionPlan::try_to_proto`" - )] - fn try_from_sort_exec( - exec: &SortExec, - codec: &dyn PhysicalExtensionCodec, - proto_converter: &dyn PhysicalProtoConverterExtension, - ) -> Result { - let encoder = ConverterPlanEncoder { - codec, - proto_converter, - }; - let encode_ctx = ExecutionPlanEncodeCtx::new(&encoder); - exec.try_to_proto(&encode_ctx)? - .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 { - 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, + fn try_into_extension_physical_plan( + &self, + extension: &protobuf::PhysicalExtensionNode, + ctx: &PhysicalPlanDecodeContext<'_>, proto_converter: &dyn PhysicalProtoConverterExtension, - ) -> Result { - 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")) - } + ) -> Result> { + let inputs: Vec> = extension + .inputs + .iter() + .map(|i| proto_converter.proto_to_execution_plan(i, ctx)) + .collect::>()?; - #[deprecated( - since = "55.0.0", - note = "unused by DataFusion; `SortPreservingMergeExec` serializes itself via `ExecutionPlan::try_to_proto`" - )] - fn try_from_sort_preserving_merge_exec( - exec: &SortPreservingMergeExec, - codec: &dyn PhysicalExtensionCodec, - proto_converter: &dyn PhysicalProtoConverterExtension, - ) -> Result { - let encoder = ConverterPlanEncoder { - codec, + let extension_node = ctx.codec().try_decode( + extension.node.as_slice(), + &inputs, + ctx.task_ctx(), proto_converter, - }; - let encode_ctx = ExecutionPlanEncodeCtx::new(&encoder); - exec.try_to_proto(&encode_ctx)?.ok_or_else(|| { - internal_datafusion_err!("SortPreservingMergeExec is not serializable") - }) - } + )?; - #[deprecated( - since = "55.0.0", - note = "unused by DataFusion; `NestedLoopJoinExec` serializes itself via `ExecutionPlan::try_to_proto`" - )] - fn try_from_nested_loop_join_exec( - exec: &NestedLoopJoinExec, - codec: &dyn PhysicalExtensionCodec, - proto_converter: &dyn PhysicalProtoConverterExtension, - ) -> Result { - let encoder = ConverterPlanEncoder { - codec, - proto_converter, - }; - let encode_ctx = ExecutionPlanEncodeCtx::new(&encoder); - exec.try_to_proto(&encode_ctx)?.ok_or_else(|| { - internal_datafusion_err!("NestedLoopJoinExec is not serializable") - }) + Ok(extension_node) } - #[deprecated( - since = "55.0.0", - note = "unused by DataFusion; `WindowAggExec` serializes itself via `ExecutionPlan::try_to_proto`" - )] - fn try_from_window_agg_exec( - exec: &WindowAggExec, - codec: &dyn PhysicalExtensionCodec, - proto_converter: &dyn PhysicalProtoConverterExtension, - ) -> Result { - let encoder = ConverterPlanEncoder { - codec, - proto_converter, - }; - let encode_ctx = ExecutionPlanEncodeCtx::new(&encoder); - exec.try_to_proto(&encode_ctx)? - .ok_or_else(|| internal_datafusion_err!("WindowAggExec is not serializable")) + fn generate_series_name_to_str(name: protobuf::GenerateSeriesName) -> &'static str { + match name { + protobuf::GenerateSeriesName::GsGenerateSeries => "generate_series", + protobuf::GenerateSeriesName::GsRange => "range", + } } - #[deprecated( - since = "55.0.0", - note = "unused by DataFusion; `BoundedWindowAggExec` serializes itself via `ExecutionPlan::try_to_proto`" - )] - fn try_from_bounded_window_agg_exec( - exec: &BoundedWindowAggExec, - codec: &dyn PhysicalExtensionCodec, - proto_converter: &dyn PhysicalProtoConverterExtension, - ) -> Result { - let encoder = ConverterPlanEncoder { - codec, - proto_converter, - }; - let encode_ctx = ExecutionPlanEncodeCtx::new(&encoder); - exec.try_to_proto(&encode_ctx)?.ok_or_else(|| { - internal_datafusion_err!("BoundedWindowAggExec is not serializable") - }) - } + fn try_into_generate_series_physical_plan( + &self, + generate_series: &protobuf::GenerateSeriesNode, + ) -> Result> { + let schema: SchemaRef = Arc::new(convert_required!(generate_series.schema)?); - #[deprecated( - since = "55.0.0", - note = "unused by DataFusion; `DataSinkExec` serializes itself via `ExecutionPlan::try_to_proto`" - )] - fn try_from_data_sink_exec( - exec: &DataSinkExec, - codec: &dyn PhysicalExtensionCodec, - proto_converter: &dyn PhysicalProtoConverterExtension, - ) -> Result> { - let encoder = ConverterPlanEncoder { - codec, - proto_converter, + let args = match &generate_series.args { + Some(protobuf::generate_series_node::Args::ContainsNull(args)) => { + GenSeriesArgs::ContainsNull { + name: protobuf::PhysicalPlanNode::generate_series_name_to_str( + args.name(), + ), + } + } + Some(protobuf::generate_series_node::Args::Int64Args(args)) => { + GenSeriesArgs::Int64Args { + start: args.start, + end: args.end, + step: args.step, + include_end: args.include_end, + name: protobuf::PhysicalPlanNode::generate_series_name_to_str( + args.name(), + ), + } + } + Some(protobuf::generate_series_node::Args::TimestampArgs(args)) => { + let step_proto = args.step.as_ref().ok_or_else(|| { + internal_datafusion_err!("Missing step in TimestampArgs") + })?; + let step = IntervalMonthDayNanoType::make_value( + step_proto.months, + step_proto.days, + step_proto.nanos, + ); + GenSeriesArgs::TimestampArgs { + start: args.start, + end: args.end, + step, + tz: args.tz.as_ref().map(|s| Arc::from(s.as_str())), + include_end: args.include_end, + name: protobuf::PhysicalPlanNode::generate_series_name_to_str( + args.name(), + ), + } + } + Some(protobuf::generate_series_node::Args::DateArgs(args)) => { + let step_proto = args.step.as_ref().ok_or_else(|| { + internal_datafusion_err!("Missing step in DateArgs") + })?; + let step = IntervalMonthDayNanoType::make_value( + step_proto.months, + step_proto.days, + step_proto.nanos, + ); + GenSeriesArgs::DateArgs { + start: args.start, + end: args.end, + step, + include_end: args.include_end, + name: protobuf::PhysicalPlanNode::generate_series_name_to_str( + args.name(), + ), + } + } + None => return internal_err!("Missing args in GenerateSeriesNode"), }; - let encode_ctx = ExecutionPlanEncodeCtx::new(&encoder); - exec.try_to_proto(&encode_ctx) - } - #[deprecated( - since = "55.0.0", - note = "unused by DataFusion; `UnnestExec` serializes itself via `ExecutionPlan::try_to_proto`" - )] - fn try_from_unnest_exec( - exec: &UnnestExec, - codec: &dyn PhysicalExtensionCodec, - proto_converter: &dyn PhysicalProtoConverterExtension, - ) -> Result { - let encoder = ConverterPlanEncoder { - codec, - proto_converter, - }; - let encode_ctx = ExecutionPlanEncodeCtx::new(&encoder); - exec.try_to_proto(&encode_ctx)? - .ok_or_else(|| internal_datafusion_err!("UnnestExec is not serializable")) - } + let table = GenerateSeriesTable::new(Arc::clone(&schema), args); + let generator = table.as_generator(generate_series.target_batch_size as usize)?; - #[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 { - let encoder = ConverterPlanEncoder { - codec, - proto_converter, - }; - let encode_ctx = ExecutionPlanEncodeCtx::new(&encoder); - exec.try_to_proto(&encode_ctx)?.ok_or_else(|| { - internal_datafusion_err!("CooperativeExec is not serializable") - }) + Ok(Arc::new(LazyMemoryExec::try_new(schema, vec![generator])?)) } fn str_to_generate_series_name(name: &str) -> Result { @@ -2827,61 +1502,6 @@ pub trait PhysicalPlanNodeExt: Sized { Ok(None) } - - #[deprecated( - since = "55.0.0", - note = "unused by DataFusion; `AsyncFuncExec` serializes itself via `ExecutionPlan::try_to_proto`" - )] - fn try_from_async_func_exec( - exec: &AsyncFuncExec, - extension_codec: &dyn PhysicalExtensionCodec, - proto_converter: &dyn PhysicalProtoConverterExtension, - ) -> Result { - let encoder = ConverterPlanEncoder { - codec: extension_codec, - proto_converter, - }; - let encode_ctx = ExecutionPlanEncodeCtx::new(&encoder); - exec.try_to_proto(&encode_ctx)? - .ok_or_else(|| internal_datafusion_err!("AsyncFuncExec is not serializable")) - } - - #[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 { - let encoder = ConverterPlanEncoder { - codec: extension_codec, - proto_converter, - }; - let encode_ctx = ExecutionPlanEncodeCtx::new(&encoder); - exec.try_to_proto(&encode_ctx)? - .ok_or_else(|| internal_datafusion_err!("BufferExec is not serializable")) - } - - #[deprecated( - since = "55.0.0", - note = "unused by DataFusion; `ScalarSubqueryExec` serializes itself via `ExecutionPlan::try_to_proto`" - )] - fn try_from_scalar_subquery_exec( - exec: &ScalarSubqueryExec, - codec: &dyn PhysicalExtensionCodec, - proto_converter: &dyn PhysicalProtoConverterExtension, - ) -> Result { - let encoder = ConverterPlanEncoder { - codec, - proto_converter, - }; - let encode_ctx = ExecutionPlanEncodeCtx::new(&encoder); - exec.try_to_proto(&encode_ctx)?.ok_or_else(|| { - internal_datafusion_err!("ScalarSubqueryExec is not serializable") - }) - } } impl PhysicalPlanNodeExt for protobuf::PhysicalPlanNode { diff --git a/datafusion/proto/tests/cases/plans/dispatch.rs b/datafusion/proto/tests/cases/plans/dispatch.rs index 5299733447b5d..af9d8a62d32f7 100644 --- a/datafusion/proto/tests/cases/plans/dispatch.rs +++ b/datafusion/proto/tests/cases/plans/dispatch.rs @@ -35,9 +35,8 @@ use datafusion_common::tree_node::TreeNodeRecursion; use datafusion_common::{Result, exec_datafusion_err}; use datafusion_proto::physical_plan::to_proto::serialize_physical_expr_with_converter; use datafusion_proto::physical_plan::{ - AsExecutionPlan, DefaultPhysicalExtensionCodec, DefaultPhysicalProtoConverter, - PhysicalExtensionCodec, PhysicalPlanDecodeContext, PhysicalPlanNodeExt, - PhysicalProtoConverterExtension, + AsExecutionPlan, DefaultPhysicalExtensionCodec, PhysicalExtensionCodec, + PhysicalPlanDecodeContext, PhysicalPlanNodeExt, PhysicalProtoConverterExtension, }; use datafusion_proto::protobuf; use datafusion_proto::protobuf::{PhysicalExprNode, PhysicalPlanNode}; @@ -189,64 +188,6 @@ async fn roundtrip_physical_plan_node() { let _ = plan.execute(0, ctx.task_ctx()).unwrap(); } -/// The deprecated `try_into_projection_physical_plan` shim now delegates to -/// [`ProjectionExec::try_from_proto`], which reads the enclosing -/// `PhysicalPlanNode` rather than a `ProjectionExecNode`. Assert the shim still -/// decodes the node passed as an argument, not `self`, so an out-of-tree caller -/// that passes a projection unrelated to `self` keeps the old behaviour. -#[test] -fn deprecated_projection_shim_decodes_argument_not_self() -> Result<()> { - use datafusion_proto::protobuf::PhysicalPlanNode; - use datafusion_proto::protobuf::physical_plan_node::PhysicalPlanType; - - let schema = Schema::new(vec![Field::new("a", DataType::Int32, false)]); - let input = Arc::new(EmptyExec::new(Arc::new(schema.clone()))); - let projection = Arc::new(ProjectionExec::try_new( - vec![ProjectionExpr::new( - col("a", &schema)?, - "renamed".to_string(), - )], - input, - )?); - - let codec = DefaultPhysicalExtensionCodec {}; - let proto_converter = DefaultPhysicalProtoConverter {}; - let projection_node = PhysicalPlanNode::try_from_physical_plan_with_converter( - projection, - &codec, - &proto_converter, - )?; - let Some(PhysicalPlanType::Projection(projection_exec_node)) = - &projection_node.physical_plan_type - else { - panic!("expected a Projection node, got {projection_node:?}"); - }; - - // `self` is deliberately a different plan variant than the argument. - let unrelated_node = PhysicalPlanNode::try_from_physical_plan_with_converter( - Arc::new(EmptyExec::new(Arc::new(schema))), - &codec, - &proto_converter, - )?; - - let session_ctx = SessionContext::new(); - let task_ctx = session_ctx.task_ctx(); - let decode_ctx = PhysicalPlanDecodeContext::new(task_ctx.as_ref(), &codec); - #[expect(deprecated)] - let decoded = unrelated_node.try_into_projection_physical_plan( - projection_exec_node, - &decode_ctx, - &proto_converter, - )?; - - let decoded = decoded - .downcast_ref::() - .expect("decoded plan should be a ProjectionExec"); - assert_eq!(decoded.expr().len(), 1); - assert_eq!(decoded.expr()[0].alias, "renamed"); - Ok(()) -} - #[test] fn custom_proto_converter_intercepts() -> Result<()> { #[derive(Default)]