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-23853-0d8482c76984e615990b594a44fbc596c6b24a34 in repository https://gitbox.apache.org/repos/asf/datafusion.git
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]
