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)]