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 562f87dff5 refactor(proto): migrate HashJoinExec serde (#23853)
562f87dff5 is described below
commit 562f87dff5e2ee381cb37e45ce6802ddefdebbc3
Author: Burak Şen <[email protected]>
AuthorDate: Fri Jul 24 23:45:35 2026 +0300
refactor(proto): migrate HashJoinExec serde (#23853)
## Which issue does this PR close?
- Closes #23507.
## Rationale for this change
Part of epic #23494. Moves `HashJoinExec` protobuf serialization from
central dispatch.
## What changes are included in this PR?
Add protobuf serialization and deserialization to the `HashJoinExec`
physical plan implementation and deprecate the corresponding central
proto methods. The wire format remains unchanged.
## Are these changes tested?
Yes, existing HashJoin round-trip tests cover this plan.
## Are there any user-facing changes?
Existing methods are deprecated with no immediate API change.
---
.../physical-plan/src/joins/hash_join/exec.rs | 191 ++++++++++++++++
datafusion/proto/src/physical_plan/mod.rs | 249 +++------------------
2 files changed, 217 insertions(+), 223 deletions(-)
diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs
b/datafusion/physical-plan/src/joins/hash_join/exec.rs
index c746aba028..ccdb050d16 100644
--- a/datafusion/physical-plan/src/joins/hash_join/exec.rs
+++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs
@@ -1761,6 +1761,197 @@ impl ExecutionPlan for HashJoinExec {
.ok()
.map(|exec| Arc::new(exec) as _)
}
+ #[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 on = self
+ .on()
+ .iter()
+ .map(|(l, r)| -> Result<protobuf::JoinOn> {
+ Ok(protobuf::JoinOn {
+ left: Some(ctx.encode_expr(l)?),
+ right: Some(ctx.encode_expr(r)?),
+ })
+ })
+ .collect::<Result<Vec<_>>>()?;
+
+ 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());
+ // `PartitionMode` is specific to `HashJoinExec`, so its conversion
stays
+ // inline (by-name on purpose: the enums are numbered differently).
+ let partition_mode = match self.partition_mode() {
+ PartitionMode::CollectLeft => protobuf::PartitionMode::CollectLeft,
+ PartitionMode::Partitioned => protobuf::PartitionMode::Partitioned,
+ PartitionMode::Auto => protobuf::PartitionMode::Auto,
+ };
+
+ let filter = self
+ .filter()
+ .map(|f| crate::joins::proto::join_filter_to_proto(f, ctx))
+ .transpose()?;
+
+ let dynamic_filter = self
+ .dynamic_filter_expr()
+ .map(|df| {
+ let df_expr: Arc<dyn PhysicalExpr> =
+ Arc::clone(df) as Arc<dyn PhysicalExpr>;
+ ctx.encode_expr(&df_expr)
+ })
+ .transpose()?;
+
+ Ok(Some(protobuf::PhysicalPlanNode {
+ physical_plan_type: Some(
+
protobuf::physical_plan_node::PhysicalPlanType::HashJoin(Box::new(
+ protobuf::HashJoinExecNode {
+ left: Some(Box::new(left)),
+ right: Some(Box::new(right)),
+ on,
+ join_type: join_type.into(),
+ partition_mode: partition_mode.into(),
+ null_equality: null_equality.into(),
+ filter,
+ // Proto3 `repeated` cannot distinguish `None` from
+ // `Some(vec![])`. `Some(vec![])` (reachable via
+ // `try_embed_projection` for e.g. `SELECT count(1) …
JOIN …`)
+ // changes the output schema, so it is encoded with the
+ // single-element sentinel `[u32::MAX]` (never a valid
column
+ // index); every other state is sent as-is. See
+ // `try_from_proto` for the matching decoder.
+ 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(),
+ },
+ null_aware: self.null_aware,
+ dynamic_filter,
+ },
+ )),
+ ),
+ }))
+ }
+}
+
+#[cfg(feature = "proto")]
+impl HashJoinExec {
+ /// Reconstruct a [`HashJoinExec`] from its protobuf representation.
+ pub fn try_from_proto(
+ node: &datafusion_proto_models::protobuf::PhysicalPlanNode,
+ ctx: &crate::proto::ExecutionPlanDecodeCtx<'_>,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ use datafusion_common::internal_datafusion_err;
+ use datafusion_proto_models::protobuf;
+ use std::any::Any;
+
+ let hashjoin = crate::expect_plan_variant!(
+ node,
+ protobuf::physical_plan_node::PhysicalPlanType::HashJoin,
+ "HashJoinExec",
+ );
+
+ let left =
+ ctx.decode_required_child(hashjoin.left.as_deref(),
"HashJoinExec", "left")?;
+ let right = ctx.decode_required_child(
+ hashjoin.right.as_deref(),
+ "HashJoinExec",
+ "right",
+ )?;
+ let left_schema = left.schema();
+ let right_schema = right.schema();
+
+ let on: Vec<(PhysicalExprRef, PhysicalExprRef)> = hashjoin
+ .on
+ .iter()
+ .map(|col| {
+ let l = ctx.decode_required_expr(
+ col.left.as_ref(),
+ left_schema.as_ref(),
+ "HashJoinExec",
+ "on.left",
+ )?;
+ let r = ctx.decode_required_expr(
+ col.right.as_ref(),
+ right_schema.as_ref(),
+ "HashJoinExec",
+ "on.right",
+ )?;
+ Ok((l, r))
+ })
+ .collect::<Result<_>>()?;
+
+ let join_type = crate::joins::proto::join_type_from_proto(
+ hashjoin.join_type,
+ "HashJoinExec",
+ )?;
+ let null_equality = crate::joins::proto::null_equality_from_proto(
+ hashjoin.null_equality,
+ "HashJoinExec",
+ )?;
+ // `PartitionMode` is specific to `HashJoinExec`, so its conversion
stays
+ // inline (by-name on purpose: the enums are numbered differently).
+ let partition_mode = match protobuf::PartitionMode::try_from(
+ hashjoin.partition_mode,
+ )
+ .map_err(|_| {
+ internal_datafusion_err!(
+ "HashJoinExec: unknown PartitionMode {}",
+ hashjoin.partition_mode
+ )
+ })? {
+ protobuf::PartitionMode::CollectLeft => PartitionMode::CollectLeft,
+ protobuf::PartitionMode::Partitioned => PartitionMode::Partitioned,
+ protobuf::PartitionMode::Auto => PartitionMode::Auto,
+ };
+
+ let filter = hashjoin
+ .filter
+ .as_ref()
+ .map(|f| crate::joins::proto::join_filter_from_proto(f, ctx,
"HashJoinExec"))
+ .transpose()?;
+
+ // Preserve the empty-projection sentinel written by `try_to_proto`.
+ let projection = match hashjoin.projection.as_slice() {
+ [] => None,
+ [u32::MAX] => Some(Vec::new()),
+ indices => Some(indices.iter().map(|i| *i as usize).collect()),
+ };
+
+ let mut hash_join = HashJoinExec::try_new(
+ left,
+ right,
+ on,
+ filter,
+ &join_type,
+ projection,
+ partition_mode,
+ null_equality,
+ hashjoin.null_aware,
+ )?;
+
+ if let Some(dynamic_filter_proto) = &hashjoin.dynamic_filter {
+ // The dynamic filter is a `DynamicFilterPhysicalExpr` over the
probe
+ // (right) side; decode against the right schema then downcast.
+ let dynamic_filter_expr =
+ ctx.decode_expr(dynamic_filter_proto, right_schema.as_ref())?;
+ let df = (dynamic_filter_expr as Arc<dyn Any + Send + Sync>)
+ .downcast::<DynamicFilterPhysicalExpr>()
+ .map_err(|_| {
+ internal_datafusion_err!(
+ "HashJoinExec dynamic_filter did not decode to a
DynamicFilterPhysicalExpr"
+ )
+ })?;
+ hash_join = hash_join.with_dynamic_filter_expr(df)?;
+ }
+
+ Ok(Arc::new(hash_join))
+ }
}
/// Determines which sides of a join are "preserved" for filter pushdown.
diff --git a/datafusion/proto/src/physical_plan/mod.rs
b/datafusion/proto/src/physical_plan/mod.rs
index b459368bcb..345ab87bf5 100644
--- a/datafusion/proto/src/physical_plan/mod.rs
+++ b/datafusion/proto/src/physical_plan/mod.rs
@@ -27,8 +27,7 @@ use datafusion_common::config::CsvOptions;
use datafusion_common::display::StringifiedPlan;
use datafusion_common::format::ExplainFormat;
use datafusion_common::{
- DataFusionError, JoinType, NullEquality, Result, internal_datafusion_err,
- internal_err, not_impl_err,
+ DataFusionError, Result, internal_datafusion_err, internal_err,
not_impl_err,
};
#[cfg(feature = "parquet")]
use datafusion_datasource::file::FileSource;
@@ -60,7 +59,7 @@ use datafusion_functions_table::generate_series::{
use datafusion_physical_expr::aggregate::{AggregateExprBuilder,
AggregateFunctionExpr};
use datafusion_physical_expr::async_scalar_function::AsyncFuncExpr;
use datafusion_physical_expr::expressions::DynamicFilterPhysicalExpr;
-use datafusion_physical_expr::{LexOrdering, LexRequirement, PhysicalExprRef};
+use datafusion_physical_expr::{LexOrdering, LexRequirement};
use
datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprDecodeCtx;
use
datafusion_physical_expr_common::physical_expr::proto_encode::PhysicalExprEncodeCtx;
use datafusion_physical_plan::aggregates::{
@@ -80,9 +79,8 @@ use datafusion_physical_plan::empty::EmptyExec;
use datafusion_physical_plan::explain::ExplainExec;
use datafusion_physical_plan::expressions::PhysicalSortExpr;
use datafusion_physical_plan::filter::FilterExec;
-use datafusion_physical_plan::joins::utils::{ColumnIndex, JoinFilter};
use datafusion_physical_plan::joins::{
- CrossJoinExec, HashJoinExec, NestedLoopJoinExec, PartitionMode,
SortMergeJoinExec,
+ CrossJoinExec, HashJoinExec, NestedLoopJoinExec, SortMergeJoinExec,
SymmetricHashJoinExec,
};
use datafusion_physical_plan::limit::{GlobalLimitExec, LocalLimitExec};
@@ -799,8 +797,8 @@ pub trait PhysicalPlanNodeExt: Sized {
PhysicalPlanType::Aggregate(hash_agg) => {
self.try_into_aggregate_physical_plan(hash_agg, ctx,
proto_converter)
}
- PhysicalPlanType::HashJoin(hashjoin) => {
- self.try_into_hash_join_physical_plan(hashjoin, ctx,
proto_converter)
+ PhysicalPlanType::HashJoin(_) => {
+ HashJoinExec::try_from_proto(self.node(), &decode_ctx)
}
PhysicalPlanType::SymmetricHashJoin(_) => {
SymmetricHashJoinExec::try_from_proto(self.node(), &decode_ctx)
@@ -909,14 +907,6 @@ pub trait PhysicalPlanNodeExt: Sized {
);
}
- if let Some(exec) = plan.downcast_ref::<HashJoinExec>() {
- return protobuf::PhysicalPlanNode::try_from_hash_join_exec(
- exec,
- codec,
- proto_converter,
- );
- }
-
if let Some(exec) = plan.downcast_ref::<AggregateExec>() {
return protobuf::PhysicalPlanNode::try_from_aggregate_exec(
exec,
@@ -1703,137 +1693,27 @@ pub trait PhysicalPlanNodeExt: Sized {
Ok(Arc::new(agg))
}
+ #[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<Arc<dyn ExecutionPlan>> {
- let left: Arc<dyn ExecutionPlan> =
- into_physical_plan(&hashjoin.left, ctx, proto_converter)?;
- let right: Arc<dyn ExecutionPlan> =
- into_physical_plan(&hashjoin.right, ctx, proto_converter)?;
- let left_schema = left.schema();
- let right_schema = right.schema();
- let on: Vec<(PhysicalExprRef, PhysicalExprRef)> = hashjoin
- .on
- .iter()
- .map(|col| {
- let left = proto_converter.proto_to_physical_expr(
- &col.left.clone().unwrap(),
- left_schema.as_ref(),
- ctx,
- )?;
- let right = proto_converter.proto_to_physical_expr(
- &col.right.clone().unwrap(),
- right_schema.as_ref(),
- ctx,
- )?;
- Ok((left, right))
- })
- .collect::<Result<_>>()?;
- let join_type =
- protobuf::JoinType::try_from(hashjoin.join_type).map_err(|_| {
- proto_error(format!(
- "Received a HashJoinNode message with unknown JoinType {}",
- hashjoin.join_type
- ))
- })?;
- let null_equality =
protobuf::NullEquality::try_from(hashjoin.null_equality)
- .map_err(|_| {
- proto_error(format!(
- "Received a HashJoinNode message with unknown NullEquality
{}",
- hashjoin.null_equality
- ))
- })?;
- let filter = hashjoin
- .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 HashJoinNode 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))?;
-
- let partition_mode =
protobuf::PartitionMode::try_from(hashjoin.partition_mode)
- .map_err(|_| {
- proto_error(format!(
- "Received a HashJoinNode message with unknown PartitionMode
{}",
- hashjoin.partition_mode
- ))
- })?;
- let partition_mode = match partition_mode {
- protobuf::PartitionMode::CollectLeft => PartitionMode::CollectLeft,
- protobuf::PartitionMode::Partitioned => PartitionMode::Partitioned,
- protobuf::PartitionMode::Auto => PartitionMode::Auto,
+ let node = protobuf::PhysicalPlanNode {
+ physical_plan_type: Some(PhysicalPlanType::HashJoin(Box::new(
+ hashjoin.clone(),
+ ))),
};
- // Proto3 `repeated` cannot distinguish `None` from `Some(vec![])`.
The latter
- // is reachable via `try_embed_projection` for `SELECT count(1) … JOIN
…` and
- // changes the join's output schema, so the encoder reserves the
single-element
- // sentinel `[u32::MAX]` (never a valid column index) to mean
"explicitly empty";
- // every other state is sent as-is. See `try_from_hash_join_exec`.
- let projection = match hashjoin.projection.as_slice() {
- [] => None,
- [u32::MAX] => Some(Vec::new()),
- indices => Some(indices.iter().map(|i| *i as usize).collect()),
+ let decoder = ConverterPlanDecoder {
+ ctx,
+ proto_converter,
};
- let mut hash_join = HashJoinExec::try_new(
- left,
- right,
- on,
- filter,
- &JoinType::from_proto(join_type),
- projection,
- partition_mode,
- NullEquality::from_proto(null_equality),
- hashjoin.null_aware,
- )?;
-
- if let Some(dynamic_filter_proto) = &hashjoin.dynamic_filter {
- let dynamic_filter_expr = proto_converter.proto_to_physical_expr(
- dynamic_filter_proto,
- right_schema.as_ref(),
- ctx,
- )?;
- let df = (dynamic_filter_expr as Arc<dyn Any + Send + Sync>)
- .downcast::<DynamicFilterPhysicalExpr>()
- .map_err(|_| {
- internal_datafusion_err!(
- "HashJoinExec dynamic_filter did not decode to a
DynamicFilterPhysicalExpr"
- )
- })?;
- hash_join = hash_join.with_dynamic_filter_expr(df)?;
- }
-
- Ok(Arc::new(hash_join))
+ let decode_ctx = ExecutionPlanDecodeCtx::new(&decoder);
+ HashJoinExec::try_from_proto(&node, &decode_ctx)
}
#[deprecated(
@@ -2585,99 +2465,22 @@ pub trait PhysicalPlanNodeExt: Sized {
.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<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 on: Vec<protobuf::JoinOn> = exec
- .on()
- .iter()
- .map(|tuple| {
- let l = proto_converter.physical_expr_to_proto(&tuple.0,
codec)?;
- let r = proto_converter.physical_expr_to_proto(&tuple.1,
codec)?;
- Ok::<_, DataFusionError>(protobuf::JoinOn {
- left: Some(l),
- right: Some(r),
- })
- })
- .collect::<Result<_>>()?;
- let join_type =
protobuf::JoinType::from_proto(exec.join_type().to_owned());
- let null_equality =
protobuf::NullEquality::from_proto(exec.null_equality());
- 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))?;
-
- let partition_mode = match exec.partition_mode() {
- PartitionMode::CollectLeft => protobuf::PartitionMode::CollectLeft,
- PartitionMode::Partitioned => protobuf::PartitionMode::Partitioned,
- PartitionMode::Auto => protobuf::PartitionMode::Auto,
};
-
- let dynamic_filter = exec
- .dynamic_filter_expr()
- .map(|df| {
- let df_expr: Arc<dyn PhysicalExpr> =
- Arc::clone(df) as Arc<dyn PhysicalExpr>;
- proto_converter.physical_expr_to_proto(&df_expr, codec)
- })
- .transpose()?;
-
- Ok(protobuf::PhysicalPlanNode {
- physical_plan_type: Some(PhysicalPlanType::HashJoin(Box::new(
- protobuf::HashJoinExecNode {
- left: Some(Box::new(left)),
- right: Some(Box::new(right)),
- on,
- join_type: join_type.into(),
- partition_mode: partition_mode.into(),
- null_equality: null_equality.into(),
- filter,
- // Send `Some(vec![])` as `[u32::MAX]` (never a valid
index) so the
- // wire format can distinguish it from `None` (which stays
empty).
- // See `try_into_hash_join_physical_plan` for the matching
decoder.
- 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(),
- },
- null_aware: exec.null_aware,
- dynamic_filter,
- },
- ))),
- })
+ let encode_ctx = ExecutionPlanEncodeCtx::new(&encoder);
+ exec.try_to_proto(&encode_ctx)?
+ .ok_or_else(|| internal_datafusion_err!("HashJoinExec is not
serializable"))
}
#[deprecated(
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
