This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new dd1ebde2d3 refactor(proto): migrate CrossJoinExec and
NestedLoopJoinExec serde (#23834)
dd1ebde2d3 is described below
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]
