This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-23834-13dadad9cab1f9a259e3833c91fc413ef4b75885 in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit dd1ebde2d312a81ab42cbc8f2f3e3ecc53e5c97c Author: Burak Şen <[email protected]> AuthorDate: Thu Jul 23 22:40:56 2026 +0300 refactor(proto): migrate CrossJoinExec and NestedLoopJoinExec serde (#23834) ## Which issue does this PR close? - Closes #23506. ## Rationale for this change Part of epic #23494. Moves `CrossJoinExec` and `NestedLoopJoinExec` protobuf serialization from central dispatch. ## What changes are included in this PR? Add new proto functions in physical plan and deprecate proto ones. ## Are these changes tested? Yes, existing round trip tests pass. ## Are there any user-facing changes? Existing methods are deprecated with no immediate API change. --- datafusion/physical-plan/src/joins/cross_join.rs | 50 +++++ datafusion/physical-plan/src/joins/mod.rs | 2 + .../physical-plan/src/joins/nested_loop_join.rs | 85 +++++++++ datafusion/physical-plan/src/joins/proto.rs | 161 ++++++++++++++++ .../src/joins/sort_merge_join/exec.rs | 129 ++----------- datafusion/proto/src/physical_plan/mod.rs | 207 +++++---------------- 6 files changed, 365 insertions(+), 269 deletions(-) diff --git a/datafusion/physical-plan/src/joins/cross_join.rs b/datafusion/physical-plan/src/joins/cross_join.rs index 2e60e53681..1a631aac98 100644 --- a/datafusion/physical-plan/src/joins/cross_join.rs +++ b/datafusion/physical-plan/src/joins/cross_join.rs @@ -457,6 +457,56 @@ impl ExecutionPlan for CrossJoinExec { Arc::new(new_right), )))) } + #[cfg(feature = "proto")] + fn try_to_proto( + &self, + ctx: &crate::proto::ExecutionPlanEncodeCtx<'_>, + ) -> Result<Option<datafusion_proto_models::protobuf::PhysicalPlanNode>> { + use datafusion_proto_models::protobuf; + + let left = ctx.encode_child(self.left())?; + let right = ctx.encode_child(self.right())?; + + Ok(Some(protobuf::PhysicalPlanNode { + physical_plan_type: Some( + protobuf::physical_plan_node::PhysicalPlanType::CrossJoin(Box::new( + protobuf::CrossJoinExecNode { + left: Some(Box::new(left)), + right: Some(Box::new(right)), + }, + )), + ), + })) + } +} + +#[cfg(feature = "proto")] +impl CrossJoinExec { + pub fn try_from_proto( + node: &datafusion_proto_models::protobuf::PhysicalPlanNode, + ctx: &crate::proto::ExecutionPlanDecodeCtx<'_>, + ) -> Result<Arc<dyn ExecutionPlan>> { + use datafusion_proto_models::protobuf; + + let crossjoin = crate::expect_plan_variant!( + node, + protobuf::physical_plan_node::PhysicalPlanType::CrossJoin, + "CrossJoinExec", + ); + + let left = ctx.decode_required_child( + crossjoin.left.as_deref(), + "CrossJoinExec", + "left", + )?; + let right = ctx.decode_required_child( + crossjoin.right.as_deref(), + "CrossJoinExec", + "right", + )?; + + Ok(Arc::new(CrossJoinExec::new(left, right))) + } } /// [left/right]_col_count are required in case the column statistics are None diff --git a/datafusion/physical-plan/src/joins/mod.rs b/datafusion/physical-plan/src/joins/mod.rs index bbb25dda65..e4f7e2e123 100644 --- a/datafusion/physical-plan/src/joins/mod.rs +++ b/datafusion/physical-plan/src/joins/mod.rs @@ -34,6 +34,8 @@ mod cross_join; mod hash_join; mod nested_loop_join; mod piecewise_merge_join; +#[cfg(feature = "proto")] +mod proto; mod sort_merge_join; mod stream_join_utils; mod symmetric_hash_join; diff --git a/datafusion/physical-plan/src/joins/nested_loop_join.rs b/datafusion/physical-plan/src/joins/nested_loop_join.rs index ac4b0d0ebc..515dcc2931 100644 --- a/datafusion/physical-plan/src/joins/nested_loop_join.rs +++ b/datafusion/physical-plan/src/joins/nested_loop_join.rs @@ -768,6 +768,91 @@ impl ExecutionPlan for NestedLoopJoinExec { try_embed_projection(projection, self) } } + #[cfg(feature = "proto")] + fn try_to_proto( + &self, + ctx: &crate::proto::ExecutionPlanEncodeCtx<'_>, + ) -> Result<Option<datafusion_proto_models::protobuf::PhysicalPlanNode>> { + use datafusion_proto_models::protobuf; + + let left = ctx.encode_child(self.left())?; + let right = ctx.encode_child(self.right())?; + + let join_type = crate::joins::proto::join_type_to_proto(*self.join_type()); + + let filter = self + .filter() + .map(|f| crate::joins::proto::join_filter_to_proto(f, ctx)) + .transpose()?; + + Ok(Some(protobuf::PhysicalPlanNode { + physical_plan_type: Some( + protobuf::physical_plan_node::PhysicalPlanType::NestedLoopJoin(Box::new( + protobuf::NestedLoopJoinExecNode { + left: Some(Box::new(left)), + right: Some(Box::new(right)), + join_type: join_type.into(), + filter, + projection: match self.projection.as_ref() { + None => Vec::new(), + Some(v) if v.is_empty() => vec![u32::MAX], + Some(v) => v.iter().map(|x| *x as u32).collect(), + }, + }, + )), + ), + })) + } +} + +#[cfg(feature = "proto")] +impl NestedLoopJoinExec { + pub fn try_from_proto( + node: &datafusion_proto_models::protobuf::PhysicalPlanNode, + ctx: &crate::proto::ExecutionPlanDecodeCtx<'_>, + ) -> Result<Arc<dyn ExecutionPlan>> { + use datafusion_proto_models::protobuf; + + let join = crate::expect_plan_variant!( + node, + protobuf::physical_plan_node::PhysicalPlanType::NestedLoopJoin, + "NestedLoopJoinExec", + ); + + let left = ctx.decode_required_child( + join.left.as_deref(), + "NestedLoopJoinExec", + "left", + )?; + let right = ctx.decode_required_child( + join.right.as_deref(), + "NestedLoopJoinExec", + "right", + )?; + + let join_type = crate::joins::proto::join_type_from_proto( + join.join_type, + "NestedLoopJoinExec", + )?; + + let filter = join + .filter + .as_ref() + .map(|f| { + crate::joins::proto::join_filter_from_proto(f, ctx, "NestedLoopJoinExec") + }) + .transpose()?; + + let projection = match join.projection.as_slice() { + [] => None, + [u32::MAX] => Some(Vec::new()), + indices => Some(indices.iter().map(|i| *i as usize).collect()), + }; + + Ok(Arc::new(NestedLoopJoinExec::try_new( + left, right, filter, &join_type, projection, + )?)) + } } impl EmbeddedProjection for NestedLoopJoinExec { diff --git a/datafusion/physical-plan/src/joins/proto.rs b/datafusion/physical-plan/src/joins/proto.rs new file mode 100644 index 0000000000..2272828b69 --- /dev/null +++ b/datafusion/physical-plan/src/joins/proto.rs @@ -0,0 +1,161 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Protobuf conversions shared by the join operators' `try_to_proto` / +//! `try_from_proto` implementations. +//! +//! The enum conversions are by-name exhaustive matches on purpose: the proto +//! enums and the `datafusion_common` enums are numbered differently, so a +//! numeric cast would silently corrupt them. + +use std::sync::Arc; + +use arrow::datatypes::Schema; +use datafusion_common::{ + JoinSide, JoinType, NullEquality, Result, internal_datafusion_err, +}; +use datafusion_proto_models::protobuf; + +use crate::joins::utils::{ColumnIndex, JoinFilter}; +use crate::proto::{ExecutionPlanDecodeCtx, ExecutionPlanEncodeCtx}; + +pub(crate) fn join_type_to_proto(join_type: JoinType) -> protobuf::JoinType { + match join_type { + JoinType::Inner => protobuf::JoinType::Inner, + JoinType::Left => protobuf::JoinType::Left, + JoinType::Right => protobuf::JoinType::Right, + JoinType::Full => protobuf::JoinType::Full, + JoinType::LeftSemi => protobuf::JoinType::Leftsemi, + JoinType::RightSemi => protobuf::JoinType::Rightsemi, + JoinType::LeftAnti => protobuf::JoinType::Leftanti, + JoinType::RightAnti => protobuf::JoinType::Rightanti, + JoinType::LeftMark => protobuf::JoinType::Leftmark, + JoinType::RightMark => protobuf::JoinType::Rightmark, + } +} + +pub(crate) fn join_type_from_proto(value: i32, plan_name: &str) -> Result<JoinType> { + let join_type = protobuf::JoinType::try_from(value) + .map_err(|_| internal_datafusion_err!("{plan_name}: unknown JoinType {value}"))?; + Ok(match join_type { + protobuf::JoinType::Inner => JoinType::Inner, + protobuf::JoinType::Left => JoinType::Left, + protobuf::JoinType::Right => JoinType::Right, + protobuf::JoinType::Full => JoinType::Full, + protobuf::JoinType::Leftsemi => JoinType::LeftSemi, + protobuf::JoinType::Rightsemi => JoinType::RightSemi, + protobuf::JoinType::Leftanti => JoinType::LeftAnti, + protobuf::JoinType::Rightanti => JoinType::RightAnti, + protobuf::JoinType::Leftmark => JoinType::LeftMark, + protobuf::JoinType::Rightmark => JoinType::RightMark, + }) +} + +pub(crate) fn join_side_to_proto(side: JoinSide) -> protobuf::JoinSide { + match side { + JoinSide::Left => protobuf::JoinSide::LeftSide, + JoinSide::Right => protobuf::JoinSide::RightSide, + JoinSide::None => protobuf::JoinSide::None, + } +} + +pub(crate) fn join_side_from_proto(value: i32, plan_name: &str) -> Result<JoinSide> { + let side = protobuf::JoinSide::try_from(value) + .map_err(|_| internal_datafusion_err!("{plan_name}: unknown JoinSide {value}"))?; + Ok(match side { + protobuf::JoinSide::LeftSide => JoinSide::Left, + protobuf::JoinSide::RightSide => JoinSide::Right, + protobuf::JoinSide::None => JoinSide::None, + }) +} + +pub(crate) fn null_equality_to_proto( + null_equality: NullEquality, +) -> protobuf::NullEquality { + match null_equality { + NullEquality::NullEqualsNothing => protobuf::NullEquality::NullEqualsNothing, + NullEquality::NullEqualsNull => protobuf::NullEquality::NullEqualsNull, + } +} + +pub(crate) fn null_equality_from_proto( + value: i32, + plan_name: &str, +) -> Result<NullEquality> { + let null_equality = protobuf::NullEquality::try_from(value).map_err(|_| { + internal_datafusion_err!("{plan_name}: unknown NullEquality {value}") + })?; + Ok(match null_equality { + protobuf::NullEquality::NullEqualsNothing => NullEquality::NullEqualsNothing, + protobuf::NullEquality::NullEqualsNull => NullEquality::NullEqualsNull, + }) +} + +pub(crate) fn join_filter_to_proto( + filter: &JoinFilter, + ctx: &ExecutionPlanEncodeCtx<'_>, +) -> Result<protobuf::JoinFilter> { + let expression = ctx.encode_expr(filter.expression())?; + let column_indices = filter + .column_indices() + .iter() + .map(|column_index| protobuf::ColumnIndex { + index: column_index.index as u32, + side: join_side_to_proto(column_index.side).into(), + }) + .collect(); + Ok(protobuf::JoinFilter { + expression: Some(expression), + column_indices, + schema: Some(filter.schema().as_ref().try_into()?), + }) +} + +pub(crate) fn join_filter_from_proto( + filter: &protobuf::JoinFilter, + ctx: &ExecutionPlanDecodeCtx<'_>, + plan_name: &str, +) -> Result<JoinFilter> { + let schema: Schema = filter + .schema + .as_ref() + .ok_or_else(|| { + internal_datafusion_err!("{plan_name}: JoinFilter missing schema") + })? + .try_into()?; + let expression = ctx.decode_required_expr( + filter.expression.as_ref(), + &schema, + plan_name, + "filter.expression", + )?; + let column_indices = filter + .column_indices + .iter() + .map(|column_index| { + Ok(ColumnIndex { + index: column_index.index as usize, + side: join_side_from_proto(column_index.side, plan_name)?, + }) + }) + .collect::<Result<Vec<_>>>()?; + Ok(JoinFilter::new( + expression, + column_indices, + Arc::new(schema), + )) +} diff --git a/datafusion/physical-plan/src/joins/sort_merge_join/exec.rs b/datafusion/physical-plan/src/joins/sort_merge_join/exec.rs index f15820e6da..00cac069aa 100644 --- a/datafusion/physical-plan/src/joins/sort_merge_join/exec.rs +++ b/datafusion/physical-plan/src/joins/sort_merge_join/exec.rs @@ -672,48 +672,13 @@ impl ExecutionPlan for SortMergeJoinExec { }) .collect::<Result<Vec<_>>>()?; - let join_type = match self.join_type() { - JoinType::Inner => protobuf::JoinType::Inner, - JoinType::Left => protobuf::JoinType::Left, - JoinType::Right => protobuf::JoinType::Right, - JoinType::Full => protobuf::JoinType::Full, - JoinType::LeftSemi => protobuf::JoinType::Leftsemi, - JoinType::RightSemi => protobuf::JoinType::Rightsemi, - JoinType::LeftAnti => protobuf::JoinType::Leftanti, - JoinType::RightAnti => protobuf::JoinType::Rightanti, - JoinType::LeftMark => protobuf::JoinType::Leftmark, - JoinType::RightMark => protobuf::JoinType::Rightmark, - }; - let null_equality = match self.null_equality() { - NullEquality::NullEqualsNothing => protobuf::NullEquality::NullEqualsNothing, - NullEquality::NullEqualsNull => protobuf::NullEquality::NullEqualsNull, - }; + let join_type = crate::joins::proto::join_type_to_proto(self.join_type()); + let null_equality = + crate::joins::proto::null_equality_to_proto(self.null_equality()); let filter = self .filter() .as_ref() - .map(|filter| -> Result<protobuf::JoinFilter> { - let expression = ctx.encode_expr(filter.expression())?; - let column_indices = filter - .column_indices() - .iter() - .map(|column_index| { - let side = match column_index.side { - JoinSide::Left => protobuf::JoinSide::LeftSide, - JoinSide::Right => protobuf::JoinSide::RightSide, - JoinSide::None => protobuf::JoinSide::None, - }; - protobuf::ColumnIndex { - index: column_index.index as u32, - side: side.into(), - } - }) - .collect(); - Ok(protobuf::JoinFilter { - expression: Some(expression), - column_indices, - schema: Some(filter.schema().as_ref().try_into()?), - }) - }) + .map(|filter| crate::joins::proto::join_filter_to_proto(filter, ctx)) .transpose()?; let sort_options = self .sort_options() @@ -754,9 +719,6 @@ impl SortMergeJoinExec { node: &datafusion_proto_models::protobuf::PhysicalPlanNode, ctx: &crate::proto::ExecutionPlanDecodeCtx<'_>, ) -> Result<Arc<dyn ExecutionPlan>> { - use crate::joins::utils::ColumnIndex; - use arrow::datatypes::Schema; - use datafusion_common::internal_datafusion_err; use datafusion_proto_models::protobuf; let sort_join = crate::expect_plan_variant!( @@ -796,82 +758,23 @@ impl SortMergeJoinExec { }) .collect::<Result<JoinOn>>()?; - let join_type = - match protobuf::JoinType::try_from(sort_join.join_type).map_err(|_| { - internal_datafusion_err!( - "SortMergeJoinExec: unknown JoinType {}", - sort_join.join_type - ) - })? { - protobuf::JoinType::Inner => JoinType::Inner, - protobuf::JoinType::Left => JoinType::Left, - protobuf::JoinType::Right => JoinType::Right, - protobuf::JoinType::Full => JoinType::Full, - protobuf::JoinType::Leftsemi => JoinType::LeftSemi, - protobuf::JoinType::Rightsemi => JoinType::RightSemi, - protobuf::JoinType::Leftanti => JoinType::LeftAnti, - protobuf::JoinType::Rightanti => JoinType::RightAnti, - protobuf::JoinType::Leftmark => JoinType::LeftMark, - protobuf::JoinType::Rightmark => JoinType::RightMark, - }; - let null_equality = match protobuf::NullEquality::try_from( + let join_type = crate::joins::proto::join_type_from_proto( + sort_join.join_type, + "SortMergeJoinExec", + )?; + let null_equality = crate::joins::proto::null_equality_from_proto( sort_join.null_equality, - ) - .map_err(|_| { - internal_datafusion_err!( - "SortMergeJoinExec: unknown NullEquality {}", - sort_join.null_equality - ) - })? { - protobuf::NullEquality::NullEqualsNothing => NullEquality::NullEqualsNothing, - protobuf::NullEquality::NullEqualsNull => NullEquality::NullEqualsNull, - }; + "SortMergeJoinExec", + )?; let filter = sort_join .filter .as_ref() - .map(|filter| -> Result<JoinFilter> { - let schema: Schema = filter - .schema - .as_ref() - .ok_or_else(|| { - internal_datafusion_err!( - "SortMergeJoinExec: JoinFilter missing schema" - ) - })? - .try_into()?; - let expression = ctx.decode_required_expr( - filter.expression.as_ref(), - &schema, + .map(|filter| { + crate::joins::proto::join_filter_from_proto( + filter, + ctx, "SortMergeJoinExec", - "filter.expression", - )?; - let column_indices = filter - .column_indices - .iter() - .map(|column_index| { - let side = protobuf::JoinSide::try_from(column_index.side) - .map_err(|_| { - internal_datafusion_err!( - "SortMergeJoinExec: unknown JoinSide {}", - column_index.side - ) - })?; - let side = match side { - protobuf::JoinSide::LeftSide => JoinSide::Left, - protobuf::JoinSide::RightSide => JoinSide::Right, - protobuf::JoinSide::None => JoinSide::None, - }; - Ok(ColumnIndex { - index: column_index.index as usize, - side, - }) - }) - .collect::<Result<Vec<_>>>()?; - Ok(JoinFilter::new( - expression, - column_indices, - Arc::new(schema), - )) + ) }) .transpose()?; let sort_options = sort_join diff --git a/datafusion/proto/src/physical_plan/mod.rs b/datafusion/proto/src/physical_plan/mod.rs index 9c4fd8e347..7139741dd3 100644 --- a/datafusion/proto/src/physical_plan/mod.rs +++ b/datafusion/proto/src/physical_plan/mod.rs @@ -814,8 +814,8 @@ pub trait PhysicalPlanNodeExt: Sized { PhysicalPlanType::Interleave(interleave) => { self.try_into_interleave_physical_plan(interleave, ctx, proto_converter) } - PhysicalPlanType::CrossJoin(crossjoin) => { - self.try_into_cross_join_physical_plan(crossjoin, ctx, proto_converter) + PhysicalPlanType::CrossJoin(_) => { + CrossJoinExec::try_from_proto(self.node(), &decode_ctx) } PhysicalPlanType::Empty(empty) => { self.try_into_empty_physical_plan(empty, ctx, proto_converter) @@ -832,8 +832,8 @@ pub trait PhysicalPlanNodeExt: Sized { PhysicalPlanType::Extension(extension) => { self.try_into_extension_physical_plan(extension, ctx, proto_converter) } - PhysicalPlanType::NestedLoopJoin(join) => { - self.try_into_nested_loop_join_physical_plan(join, ctx, proto_converter) + PhysicalPlanType::NestedLoopJoin(_) => { + NestedLoopJoinExec::try_from_proto(self.node(), &decode_ctx) } PhysicalPlanType::Analyze(analyze) => { self.try_into_analyze_physical_plan(analyze, ctx, proto_converter) @@ -928,14 +928,6 @@ pub trait PhysicalPlanNodeExt: Sized { ); } - if let Some(exec) = plan.downcast_ref::<CrossJoinExec>() { - return protobuf::PhysicalPlanNode::try_from_cross_join_exec( - exec, - codec, - proto_converter, - ); - } - if let Some(exec) = plan.downcast_ref::<AggregateExec>() { return protobuf::PhysicalPlanNode::try_from_aggregate_exec( exec, @@ -980,14 +972,6 @@ pub trait PhysicalPlanNodeExt: Sized { ); } - if let Some(exec) = plan.downcast_ref::<NestedLoopJoinExec>() { - return protobuf::PhysicalPlanNode::try_from_nested_loop_join_exec( - exec, - codec, - proto_converter, - ); - } - if let Some(exec) = plan.downcast_ref::<WindowAggExec>() { return protobuf::PhysicalPlanNode::try_from_window_agg_exec( exec, @@ -2040,17 +2024,27 @@ pub trait PhysicalPlanNodeExt: Sized { Ok(Arc::new(InterleaveExec::try_new(inputs)?)) } + #[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<Arc<dyn ExecutionPlan>> { - let left: Arc<dyn ExecutionPlan> = - into_physical_plan(&crossjoin.left, ctx, proto_converter)?; - let right: Arc<dyn ExecutionPlan> = - into_physical_plan(&crossjoin.right, ctx, proto_converter)?; - Ok(Arc::new(CrossJoinExec::new(left, right))) + 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) } fn try_into_empty_physical_plan( @@ -2146,76 +2140,27 @@ pub trait PhysicalPlanNodeExt: Sized { 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<Arc<dyn ExecutionPlan>> { - let left: Arc<dyn ExecutionPlan> = - into_physical_plan(&join.left, ctx, proto_converter)?; - let right: Arc<dyn ExecutionPlan> = - into_physical_plan(&join.right, ctx, proto_converter)?; - let join_type = protobuf::JoinType::try_from(join.join_type).map_err(|_| { - proto_error(format!( - "Received a NestedLoopJoinExecNode message with unknown JoinType {}", - join.join_type - )) - })?; - let filter = join - .filter - .as_ref() - .map(|f| { - let schema = f - .schema - .as_ref() - .ok_or_else(|| proto_error("Missing JoinFilter schema"))? - .try_into()?; - - let expression = proto_converter - .proto_to_physical_expr( - f.expression.as_ref().ok_or_else(|| { - proto_error("Unexpected empty filter expression") - })?, - &schema, - ctx, - )?; - let column_indices = f.column_indices - .iter() - .map(|i| { - let side = protobuf::JoinSide::try_from(i.side) - .map_err(|_| proto_error(format!( - "Received a NestedLoopJoinExecNode message with JoinSide in Filter {}", - i.side)) - )?; - - Ok(ColumnIndex { - index: i.index as usize, - side: side.into(), - }) - }) - .collect::<Result<Vec<_>>>()?; - - Ok(JoinFilter::new(expression, column_indices, Arc::new(schema))) - }) - .map_or(Ok(None), |v: Result<JoinFilter>| v.map(Some))?; - - // See `try_into_hash_join_physical_plan` for the rationale behind the - // `[u32::MAX]` sentinel; `NestedLoopJoinExec` has the same `Option<Vec<usize>>` - // projection field and shares the proto3 `repeated` ambiguity. - let projection = match join.projection.as_slice() { - [] => None, - [u32::MAX] => Some(Vec::new()), - indices => Some(indices.iter().map(|i| *i as usize).collect()), + let node = protobuf::PhysicalPlanNode { + physical_plan_type: Some(PhysicalPlanType::NestedLoopJoin(Box::new( + join.clone(), + ))), }; - - Ok(Arc::new(NestedLoopJoinExec::try_new( - left, - right, - filter, - &JoinType::from_proto(join_type), - projection, - )?)) + let decoder = ConverterPlanDecoder { + ctx, + proto_converter, + }; + let decode_ctx = ExecutionPlanDecodeCtx::new(&decoder); + NestedLoopJoinExec::try_from_proto(&node, &decode_ctx) } fn try_into_analyze_physical_plan( @@ -2981,29 +2926,22 @@ pub trait PhysicalPlanNodeExt: Sized { }) } + #[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<protobuf::PhysicalPlanNode> { - let left = protobuf::PhysicalPlanNode::try_from_physical_plan_with_converter( - exec.left().to_owned(), - codec, - proto_converter, - )?; - let right = protobuf::PhysicalPlanNode::try_from_physical_plan_with_converter( - exec.right().to_owned(), + let encoder = ConverterPlanEncoder { codec, proto_converter, - )?; - Ok(protobuf::PhysicalPlanNode { - physical_plan_type: Some(PhysicalPlanType::CrossJoin(Box::new( - protobuf::CrossJoinExecNode { - left: Some(Box::new(left)), - right: Some(Box::new(right)), - }, - ))), - }) + }; + let encode_ctx = ExecutionPlanEncodeCtx::new(&encoder); + exec.try_to_proto(&encode_ctx)? + .ok_or_else(|| internal_datafusion_err!("CrossJoinExec is not serializable")) } fn try_from_aggregate_exec( @@ -3454,65 +3392,22 @@ pub trait PhysicalPlanNodeExt: Sized { }) } + #[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<protobuf::PhysicalPlanNode> { - let left = protobuf::PhysicalPlanNode::try_from_physical_plan_with_converter( - exec.left().to_owned(), - codec, - proto_converter, - )?; - let right = protobuf::PhysicalPlanNode::try_from_physical_plan_with_converter( - exec.right().to_owned(), + let encoder = ConverterPlanEncoder { codec, proto_converter, - )?; - - let join_type = protobuf::JoinType::from_proto(exec.join_type().to_owned()); - let filter = exec - .filter() - .as_ref() - .map(|f| { - let expression = - proto_converter.physical_expr_to_proto(f.expression(), codec)?; - let column_indices = f - .column_indices() - .iter() - .map(|i| { - let side: protobuf::JoinSide = i.side.to_owned().into(); - protobuf::ColumnIndex { - index: i.index as u32, - side: side.into(), - } - }) - .collect(); - let schema = f.schema().as_ref().try_into()?; - Ok(protobuf::JoinFilter { - expression: Some(expression), - column_indices, - schema: Some(schema), - }) - }) - .map_or(Ok(None), |v: Result<protobuf::JoinFilter>| v.map(Some))?; - - Ok(protobuf::PhysicalPlanNode { - physical_plan_type: Some(PhysicalPlanType::NestedLoopJoin(Box::new( - protobuf::NestedLoopJoinExecNode { - left: Some(Box::new(left)), - right: Some(Box::new(right)), - join_type: join_type.into(), - filter, - // `[u32::MAX]` sentinel distinguishes `Some(vec![])` from `None`; - // see `try_from_hash_join_exec`. - projection: match exec.projection().as_ref() { - None => Vec::new(), - Some(v) if v.is_empty() => vec![u32::MAX], - Some(v) => v.iter().map(|x| *x as u32).collect(), - }, - }, - ))), + }; + let encode_ctx = ExecutionPlanEncodeCtx::new(&encoder); + exec.try_to_proto(&encode_ctx)?.ok_or_else(|| { + internal_datafusion_err!("NestedLoopJoinExec is not serializable") }) } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
