From df27c5210e0df8a4afc0656195b222ccdb2def33 Mon Sep 17 00:00:00 2001 From: Jiawei Zhao Date: Tue, 21 Jul 2026 20:01:53 +0800 Subject: [PATCH] refactor(proto): migrate aggregate exec serde Aggregate protobuf conversion lived in the central proto crate, which prevented AggregateExec from owning its function-codec serialization. Move encoding and decoding into AggregateExec and keep the deprecated helpers as delegates so existing callers retain wire-compatible behavior. CLOSES #23512 Signed-off-by: Jiawei Zhao --- .../physical-plan/src/aggregates/mod.rs | 367 +++++++++++++++++ datafusion/proto/src/physical_plan/mod.rs | 383 ++---------------- 2 files changed, 398 insertions(+), 352 deletions(-) diff --git a/datafusion/physical-plan/src/aggregates/mod.rs b/datafusion/physical-plan/src/aggregates/mod.rs index 732da32ab0391..ea6ea5c5734e5 100644 --- a/datafusion/physical-plan/src/aggregates/mod.rs +++ b/datafusion/physical-plan/src/aggregates/mod.rs @@ -2183,6 +2183,373 @@ impl ExecutionPlan for AggregateExec { Ok(result) } + + #[cfg(feature = "proto")] + fn try_to_proto( + &self, + ctx: &crate::proto::ExecutionPlanEncodeCtx<'_>, + ) -> Result> { + use datafusion_proto_models::protobuf; + + let input = ctx.encode_child(self.input())?; + let group_by = self.group_expr(); + let group_expr = + ctx.encode_expressions(group_by.expr().iter().map(|(expr, _)| expr))?; + let group_expr_name = group_by + .expr() + .iter() + .map(|(_, name)| name.to_owned()) + .collect(); + let null_expr = + ctx.encode_expressions(group_by.null_expr().iter().map(|(expr, _)| expr))?; + let groups = group_by.groups().iter().flatten().copied().collect(); + let aggr_expr = self + .aggr_expr() + .iter() + .map(|expr| encode_aggregate_expr(expr, ctx)) + .collect::>>()?; + let aggr_expr_name = self + .aggr_expr() + .iter() + .map(|expr| expr.name().to_string()) + .collect(); + let filter_expr = self + .filter_expr() + .iter() + .map(|filter| { + Ok(protobuf::MaybeFilter { + expr: filter + .as_ref() + .map(|expr| ctx.encode_expr(expr)) + .transpose()?, + }) + }) + .collect::>>()?; + let mode = match self.mode() { + AggregateMode::Partial => protobuf::AggregateMode::Partial, + AggregateMode::Final => protobuf::AggregateMode::Final, + AggregateMode::FinalPartitioned => protobuf::AggregateMode::FinalPartitioned, + AggregateMode::Single => protobuf::AggregateMode::Single, + AggregateMode::SinglePartitioned => { + protobuf::AggregateMode::SinglePartitioned + } + AggregateMode::PartialReduce => protobuf::AggregateMode::PartialReduce, + }; + let limit = self.limit_options().map(|options| protobuf::AggLimit { + limit: options.limit() as u64, + descending: options.descending(), + }); + let dynamic_filter = match self.dynamic_filter_expr() { + Some(filter) => { + let expr: Arc = + Arc::clone(filter) as Arc; + Some(ctx.encode_expr(&expr)?) + } + None => None, + }; + + Ok(Some(protobuf::PhysicalPlanNode { + physical_plan_type: Some( + protobuf::physical_plan_node::PhysicalPlanType::Aggregate(Box::new( + protobuf::AggregateExecNode { + group_expr, + group_expr_name, + aggr_expr, + filter_expr, + aggr_expr_name, + mode: mode as i32, + input: Some(Box::new(input)), + input_schema: Some(self.input_schema().as_ref().try_into()?), + null_expr, + groups, + limit, + has_grouping_set: group_by.has_grouping_set(), + dynamic_filter, + }, + )), + ), + })) + } +} + +#[cfg(feature = "proto")] +const HUMAN_DISPLAY_ALIAS_PREFIX: &str = "\u{1f}datafusion_human_display_alias_v1:"; + +#[cfg(feature = "proto")] +fn encode_human_display_alias(human_display: &str, alias: &str) -> String { + format!( + "{HUMAN_DISPLAY_ALIAS_PREFIX}{}:{alias}{human_display}", + alias.len() + ) +} + +#[cfg(feature = "proto")] +fn split_human_display_alias<'a>( + human_display: &'a str, + name: &'a str, +) -> (&'a str, Option<&'a str>) { + if let Some(encoded) = human_display.strip_prefix(HUMAN_DISPLAY_ALIAS_PREFIX) + && let Some((alias_len, encoded)) = encoded.split_once(':') + && let Ok(alias_len) = alias_len.parse::() + && let Some(alias) = encoded.get(..alias_len) + && let Some(human_display) = encoded.get(alias_len..) + && alias == name + && !human_display.is_empty() + { + return (human_display, Some(alias)); + } + + (human_display, None) +} + +#[cfg(feature = "proto")] +fn encode_aggregate_expr( + aggr_expr: &Arc, + ctx: &crate::proto::ExecutionPlanEncodeCtx<'_>, +) -> Result { + use datafusion_proto_models::protobuf; + + let expressions = aggr_expr.expressions(); + let expr = ctx.encode_expressions(expressions.iter())?; + let ordering_req = aggr_expr + .order_bys() + .iter() + .map(|sort_expr| { + Ok(protobuf::PhysicalSortExprNode { + expr: Some(Box::new(ctx.encode_expr(&sort_expr.expr)?)), + asc: !sort_expr.options.descending, + nulls_first: sort_expr.options.nulls_first, + }) + }) + .collect::>>()?; + let name = aggr_expr.fun().name().to_string(); + let fun_definition = ctx.encode_udaf(aggr_expr.fun())?; + let human_display = match (aggr_expr.human_display(), aggr_expr.human_display_alias()) + { + (Some(display), Some(alias)) => encode_human_display_alias(display, alias), + (Some(display), None) => display.to_string(), + (None, _) => String::new(), + }; + + Ok(protobuf::PhysicalExprNode { + expr_id: None, + expr_type: Some(protobuf::physical_expr_node::ExprType::AggregateExpr( + protobuf::PhysicalAggregateExprNode { + aggregate_function: Some( + protobuf::physical_aggregate_expr_node::AggregateFunction::UserDefinedAggrFunction(name), + ), + expr, + ordering_req, + distinct: aggr_expr.is_distinct(), + ignore_nulls: aggr_expr.ignore_nulls(), + fun_definition, + human_display, + }, + )), + }) +} + +#[cfg(feature = "proto")] +impl AggregateExec { + /// Reconstruct an [`AggregateExec`] from its protobuf representation. + /// + /// Grouping expressions are decoded against the child schema. Aggregate + /// arguments, ordering, filters, and the dynamic filter are decoded against + /// the aggregate input schema carried in the protobuf node. + pub fn try_from_proto( + node: &datafusion_proto_models::protobuf::PhysicalPlanNode, + ctx: &crate::proto::ExecutionPlanDecodeCtx<'_>, + ) -> Result> { + use datafusion_physical_expr::PhysicalSortExpr; + use datafusion_physical_expr::aggregate::AggregateExprBuilder; + use datafusion_proto_models::protobuf; + use protobuf::physical_aggregate_expr_node::AggregateFunction; + use protobuf::physical_expr_node::ExprType; + + let hash_agg = crate::expect_plan_variant!( + node, + protobuf::physical_plan_node::PhysicalPlanType::Aggregate, + "AggregateExec", + ); + let input = ctx.decode_required_child( + hash_agg.input.as_deref(), + "AggregateExec", + "input", + )?; + let mode = protobuf::AggregateMode::try_from(hash_agg.mode).map_err(|_| { + datafusion_common::internal_datafusion_err!( + "Received an AggregateNode message with unknown AggregateMode {}", + hash_agg.mode + ) + })?; + let mode = match mode { + protobuf::AggregateMode::Partial => AggregateMode::Partial, + protobuf::AggregateMode::Final => AggregateMode::Final, + protobuf::AggregateMode::FinalPartitioned => AggregateMode::FinalPartitioned, + protobuf::AggregateMode::Single => AggregateMode::Single, + protobuf::AggregateMode::SinglePartitioned => { + AggregateMode::SinglePartitioned + } + protobuf::AggregateMode::PartialReduce => AggregateMode::PartialReduce, + }; + let num_expr = hash_agg.group_expr.len(); + let child_schema = input.schema(); + let group_expr = hash_agg + .group_expr + .iter() + .zip(hash_agg.group_expr_name.iter()) + .map(|(expr, name)| { + Ok(( + ctx.decode_expr(expr, child_schema.as_ref())?, + name.to_string(), + )) + }) + .collect::>>()?; + let null_expr = hash_agg + .null_expr + .iter() + .zip(hash_agg.group_expr_name.iter()) + .map(|(expr, name)| { + Ok(( + ctx.decode_expr(expr, child_schema.as_ref())?, + name.to_string(), + )) + }) + .collect::>>()?; + let groups = if hash_agg.groups.is_empty() { + vec![] + } else { + hash_agg + .groups + .chunks(num_expr) + .map(|group| group.to_vec()) + .collect() + }; + let input_schema = hash_agg.input_schema.as_ref().ok_or_else(|| { + datafusion_common::internal_datafusion_err!( + "input_schema in AggregateNode is missing." + ) + })?; + let input_schema: SchemaRef = SchemaRef::new(input_schema.try_into()?); + let filter_expr = hash_agg + .filter_expr + .iter() + .map(|filter| { + filter + .expr + .as_ref() + .map(|expr| ctx.decode_expr(expr, input_schema.as_ref())) + .transpose() + }) + .collect::>>()?; + let aggr_expr = hash_agg + .aggr_expr + .iter() + .zip(hash_agg.aggr_expr_name.iter()) + .map(|(expr, name)| { + let expr_type = expr.expr_type.as_ref().ok_or_else(|| { + datafusion_common::internal_datafusion_err!( + "Unexpected empty aggregate physical expression" + ) + })?; + let ExprType::AggregateExpr(aggregate) = expr_type else { + return internal_err!( + "Invalid aggregate expression for AggregateExec" + ); + }; + let args = aggregate + .expr + .iter() + .map(|expr| ctx.decode_expr(expr, input_schema.as_ref())) + .collect::>>()?; + let order_by = aggregate + .ordering_req + .iter() + .map(|sort_expr| { + let expr = sort_expr.expr.as_deref().ok_or_else(|| { + datafusion_common::internal_datafusion_err!( + "AggregateExec ordering expression is missing its inner expr" + ) + })?; + Ok(PhysicalSortExpr { + expr: ctx.decode_expr(expr, input_schema.as_ref())?, + options: arrow::compute::SortOptions { + descending: !sort_expr.asc, + nulls_first: sort_expr.nulls_first, + }, + }) + }) + .collect::>>()?; + let Some(AggregateFunction::UserDefinedAggrFunction(udaf_name)) = + aggregate.aggregate_function.as_ref() + else { + return internal_err!( + "Invalid AggregateExpr, missing aggregate_function" + ); + }; + let udaf = ctx.decode_udaf( + udaf_name, + aggregate.fun_definition.as_deref(), + )?; + let (human_display, human_display_alias) = + split_human_display_alias(&aggregate.human_display, name); + let builder = AggregateExprBuilder::new(udaf, args) + .schema(Arc::clone(&input_schema)) + .alias(name) + .with_ignore_nulls(aggregate.ignore_nulls) + .with_distinct(aggregate.distinct) + .order_by(order_by) + .human_display(human_display); + let builder = if let Some(alias) = human_display_alias { + builder.human_display_alias(alias) + } else { + builder + }; + builder.build().map(Arc::new) + }) + .collect::>>()?; + let aggregate = AggregateExec::try_new( + mode, + PhysicalGroupBy::new( + group_expr, + null_expr, + groups, + hash_agg.has_grouping_set, + ), + aggr_expr, + filter_expr, + input, + Arc::clone(&input_schema), + )?; + let aggregate = if let Some(limit) = &hash_agg.limit { + let options = match limit.descending { + Some(descending) => { + LimitOptions::new_with_order(limit.limit as usize, descending) + } + None => LimitOptions::new(limit.limit as usize), + }; + aggregate.with_limit_options(Some(options)) + } else { + aggregate + }; + let aggregate = if let Some(dynamic_filter) = &hash_agg.dynamic_filter { + let dynamic_filter = + ctx.decode_expr(dynamic_filter, input_schema.as_ref())?; + let dynamic_filter = (dynamic_filter + as Arc) + .downcast::() + .map_err(|_| { + datafusion_common::internal_datafusion_err!( + "AggregateExec dynamic_filter did not decode to a DynamicFilterPhysicalExpr" + ) + })?; + aggregate.with_dynamic_filter_expr(dynamic_filter)? + } else { + aggregate + }; + + Ok(Arc::new(aggregate)) + } } /// Creates the output schema for an [`AggregateExec`] containing the group by columns followed diff --git a/datafusion/proto/src/physical_plan/mod.rs b/datafusion/proto/src/physical_plan/mod.rs index cea334e42aace..d577eb78fa2d7 100644 --- a/datafusion/proto/src/physical_plan/mod.rs +++ b/datafusion/proto/src/physical_plan/mod.rs @@ -58,15 +58,12 @@ use datafusion_expr::{AggregateUDF, HigherOrderUDF, ScalarUDF, WindowUDF}; use datafusion_functions_table::generate_series::{ Empty, GenSeriesArgs, GenerateSeriesTable, GenericSeriesState, TimestampValue, }; -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_common::physical_expr::proto_decode::PhysicalExprDecodeCtx; use datafusion_physical_expr_common::physical_expr::proto_encode::PhysicalExprEncodeCtx; -use datafusion_physical_plan::aggregates::{ - AggregateExec, AggregateMode, LimitOptions, PhysicalGroupBy, -}; +use datafusion_physical_plan::aggregates::AggregateExec; use datafusion_physical_plan::analyze::AnalyzeExec; use datafusion_physical_plan::async_func::AsyncFuncExec; use datafusion_physical_plan::buffer::BufferExec; @@ -112,16 +109,15 @@ use crate::common::{byte_to_string, str_to_byte}; use crate::convert::{FromProto, TryFromProto}; use crate::convert_required; use crate::physical_plan::from_proto::{ - parse_physical_expr_with_converter, parse_physical_sort_expr, - parse_physical_sort_exprs, parse_physical_window_expr, - parse_protobuf_file_scan_config, parse_record_batches, parse_table_schema_from_proto, + parse_physical_expr_with_converter, parse_physical_sort_exprs, + parse_physical_window_expr, parse_protobuf_file_scan_config, parse_record_batches, + parse_table_schema_from_proto, }; use crate::physical_plan::to_proto::{ - serialize_file_scan_config, serialize_maybe_filter, serialize_physical_aggr_expr, - serialize_physical_expr_with_converter, serialize_physical_sort_exprs, - serialize_physical_window_expr, serialize_record_batches, + serialize_file_scan_config, serialize_physical_expr_with_converter, + serialize_physical_sort_exprs, serialize_physical_window_expr, + serialize_record_batches, }; -use crate::protobuf::physical_aggregate_expr_node::AggregateFunction; use crate::protobuf::physical_expr_node::ExprType; use crate::protobuf::physical_plan_node::PhysicalPlanType; use crate::protobuf::{ @@ -141,48 +137,10 @@ fn encode_human_display_alias(human_display: &str, alias: &str) -> String { ) } -fn split_human_display_alias<'a>( - human_display: &'a str, - name: &'a str, -) -> (&'a str, Option<&'a str>) { - if let Some(encoded) = human_display.strip_prefix(HUMAN_DISPLAY_ALIAS_PREFIX) - && let Some((alias_len, encoded)) = encoded.split_once(':') - && let Ok(alias_len) = alias_len.parse::() - && let Some(alias) = encoded.get(..alias_len) - && let Some(human_display) = encoded.get(alias_len..) - && alias == name - && !human_display.is_empty() - { - return (human_display, Some(alias)); - } - - (human_display, None) -} - #[cfg(test)] mod tests { use super::*; - #[test] - fn split_human_display_alias_ignores_mismatched_alias() { - let encoded = encode_human_display_alias("sum(value)", "revenue"); - - assert_eq!( - split_human_display_alias(&encoded, "other"), - (encoded.as_str(), None) - ); - } - - #[test] - fn split_human_display_alias_keeps_malformed_prefix_literal() { - let display = format!("{HUMAN_DISPLAY_ALIAS_PREFIX}not-an-encoding"); - - assert_eq!( - split_human_display_alias(&display, "agg"), - (display.as_str(), None) - ); - } - /// Unit tests for the bytes-only function serde exposed on /// [`ExecutionPlanEncodeCtx`] / [`ExecutionPlanDecodeCtx`] and backed by /// [`ConverterPlanEncoder`] / [`ConverterPlanDecoder`]. Function-carrying @@ -802,8 +760,8 @@ pub trait PhysicalPlanNodeExt: Sized { PhysicalPlanType::Window(window_agg) => { self.try_into_window_physical_plan(window_agg, ctx, proto_converter) } - PhysicalPlanType::Aggregate(hash_agg) => { - self.try_into_aggregate_physical_plan(hash_agg, ctx, proto_converter) + PhysicalPlanType::Aggregate(_) => { + AggregateExec::try_from_proto(self.node(), &decode_ctx) } PhysicalPlanType::HashJoin(hashjoin) => { self.try_into_hash_join_physical_plan(hashjoin, ctx, proto_converter) @@ -957,14 +915,6 @@ pub trait PhysicalPlanNodeExt: Sized { ); } - if let Some(exec) = plan.downcast_ref::() { - return protobuf::PhysicalPlanNode::try_from_aggregate_exec( - exec, - codec, - proto_converter, - ); - } - if let Some(empty) = plan.downcast_ref::() { return protobuf::PhysicalPlanNode::try_from_empty_exec(empty, codec); } @@ -1591,212 +1541,27 @@ pub trait PhysicalPlanNodeExt: Sized { } } + #[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 input: Arc = - into_physical_plan(&hash_agg.input, ctx, proto_converter)?; - let mode = protobuf::AggregateMode::try_from(hash_agg.mode).map_err(|_| { - proto_error(format!( - "Received a AggregateNode message with unknown AggregateMode {}", - hash_agg.mode - )) - })?; - let agg_mode: AggregateMode = match mode { - protobuf::AggregateMode::Partial => AggregateMode::Partial, - protobuf::AggregateMode::Final => AggregateMode::Final, - protobuf::AggregateMode::FinalPartitioned => AggregateMode::FinalPartitioned, - protobuf::AggregateMode::Single => AggregateMode::Single, - protobuf::AggregateMode::SinglePartitioned => { - AggregateMode::SinglePartitioned - } - protobuf::AggregateMode::PartialReduce => AggregateMode::PartialReduce, - }; - - let num_expr = hash_agg.group_expr.len(); - - let group_expr = hash_agg - .group_expr - .iter() - .zip(hash_agg.group_expr_name.iter()) - .map(|(expr, name)| { - proto_converter - .proto_to_physical_expr(expr, input.schema().as_ref(), ctx) - .map(|expr| (expr, name.to_string())) - }) - .collect::, _>>()?; - - let null_expr = hash_agg - .null_expr - .iter() - .zip(hash_agg.group_expr_name.iter()) - .map(|(expr, name)| { - proto_converter - .proto_to_physical_expr(expr, input.schema().as_ref(), ctx) - .map(|expr| (expr, name.to_string())) - }) - .collect::, _>>()?; - - let groups: Vec> = if !hash_agg.groups.is_empty() { - hash_agg - .groups - .chunks(num_expr) - .map(|g| g.to_vec()) - .collect::>>() - } else { - vec![] - }; - - let has_grouping_set = hash_agg.has_grouping_set; - - let input_schema = hash_agg.input_schema.as_ref().ok_or_else(|| { - internal_datafusion_err!("input_schema in AggregateNode is missing.") - })?; - let physical_schema: SchemaRef = SchemaRef::new(input_schema.try_into()?); - - let physical_filter_expr = hash_agg - .filter_expr - .iter() - .map(|expr| { - expr.expr - .as_ref() - .map(|e| { - proto_converter.proto_to_physical_expr(e, &physical_schema, ctx) - }) - .transpose() - }) - .collect::, _>>()?; - - let physical_aggr_expr: Vec> = hash_agg - .aggr_expr - .iter() - .zip(hash_agg.aggr_expr_name.iter()) - .map(|(expr, name)| { - let expr_type = expr.expr_type.as_ref().ok_or_else(|| { - proto_error("Unexpected empty aggregate physical expression") - })?; - - match expr_type { - ExprType::AggregateExpr(agg_node) => { - let input_phy_expr: Vec> = agg_node - .expr - .iter() - .map(|e| { - proto_converter.proto_to_physical_expr( - e, - &physical_schema, - ctx, - ) - }) - .collect::>>()?; - let order_bys = agg_node - .ordering_req - .iter() - .map(|e| { - parse_physical_sort_expr( - e, - ctx, - &physical_schema, - proto_converter, - ) - }) - .collect::>()?; - agg_node - .aggregate_function - .as_ref() - .map(|func| match func { - AggregateFunction::UserDefinedAggrFunction(udaf_name) => { - let agg_udf = match &agg_node.fun_definition { - Some(buf) => { - ctx.codec().try_decode_udaf(udaf_name, buf)? - } - None => ctx.task_ctx().udaf(udaf_name).or_else( - |_| { - ctx.codec() - .try_decode_udaf(udaf_name, &[]) - }, - )?, - }; - - let (human_display, human_display_alias) = - split_human_display_alias( - &agg_node.human_display, - name, - ); - let builder = AggregateExprBuilder::new( - agg_udf, - input_phy_expr, - ) - .schema(Arc::clone(&physical_schema)) - .alias(name) - .with_ignore_nulls(agg_node.ignore_nulls) - .with_distinct(agg_node.distinct) - .order_by(order_bys) - .human_display(human_display); - let builder = if let Some(alias) = human_display_alias - { - builder.human_display_alias(alias) - } else { - builder - }; - builder.build().map(Arc::new) - } - }) - .transpose()? - .ok_or_else(|| { - proto_error( - "Invalid AggregateExpr, missing aggregate_function", - ) - }) - } - _ => internal_err!("Invalid aggregate expression for AggregateExec"), - } - }) - .collect::, _>>()?; - - let physical_schema_ref = Arc::clone(&physical_schema); - let agg = AggregateExec::try_new( - agg_mode, - PhysicalGroupBy::new(group_expr, null_expr, groups, has_grouping_set), - physical_aggr_expr, - physical_filter_expr, - input, - physical_schema, - )?; - - let agg = if let Some(limit_proto) = &hash_agg.limit { - let limit = limit_proto.limit as usize; - let limit_options = match limit_proto.descending { - Some(descending) => LimitOptions::new_with_order(limit, descending), - None => LimitOptions::new(limit), - }; - agg.with_limit_options(Some(limit_options)) - } else { - agg + let node = protobuf::PhysicalPlanNode { + physical_plan_type: Some(PhysicalPlanType::Aggregate(Box::new( + hash_agg.clone(), + ))), }; - - let agg = if let Some(dynamic_filter_proto) = &hash_agg.dynamic_filter { - let dynamic_filter_expr = proto_converter.proto_to_physical_expr( - dynamic_filter_proto, - physical_schema_ref.as_ref(), - ctx, - )?; - let df = (dynamic_filter_expr as Arc) - .downcast::() - .map_err(|_| { - internal_datafusion_err!( - "AggregateExec dynamic_filter did not decode to a DynamicFilterPhysicalExpr" - ) - })?; - agg.with_dynamic_filter_expr(df)? - } else { - agg + let decoder = ConverterPlanDecoder { + ctx, + proto_converter, }; - - Ok(Arc::new(agg)) + let decode_ctx = ExecutionPlanDecodeCtx::new(&decoder); + AggregateExec::try_from_proto(&node, &decode_ctx) } fn try_into_hash_join_physical_plan( @@ -3163,108 +2928,22 @@ pub trait PhysicalPlanNodeExt: Sized { }) } + #[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 groups: Vec = exec - .group_expr() - .groups() - .iter() - .flatten() - .copied() - .collect(); - - let group_names = exec - .group_expr() - .expr() - .iter() - .map(|expr| expr.1.to_owned()) - .collect(); - - let filter = exec - .filter_expr() - .iter() - .map(|expr| serialize_maybe_filter(expr.to_owned(), codec, proto_converter)) - .collect::>>()?; - - let agg = exec - .aggr_expr() - .iter() - .map(|expr| { - serialize_physical_aggr_expr(expr.to_owned(), codec, proto_converter) - }) - .collect::>>()?; - - let agg_names = exec - .aggr_expr() - .iter() - .map(|expr| expr.name().to_string()) - .collect::>(); - - let agg_mode = match exec.mode() { - AggregateMode::Partial => protobuf::AggregateMode::Partial, - AggregateMode::Final => protobuf::AggregateMode::Final, - AggregateMode::FinalPartitioned => protobuf::AggregateMode::FinalPartitioned, - AggregateMode::Single => protobuf::AggregateMode::Single, - AggregateMode::SinglePartitioned => { - protobuf::AggregateMode::SinglePartitioned - } - AggregateMode::PartialReduce => protobuf::AggregateMode::PartialReduce, - }; - let input_schema = exec.input_schema(); - let input = protobuf::PhysicalPlanNode::try_from_physical_plan_with_converter( - exec.input().to_owned(), + let encoder = ConverterPlanEncoder { codec, proto_converter, - )?; - - let null_expr = exec - .group_expr() - .null_expr() - .iter() - .map(|expr| proto_converter.physical_expr_to_proto(&expr.0, codec)) - .collect::>>()?; - - let group_expr = exec - .group_expr() - .expr() - .iter() - .map(|expr| proto_converter.physical_expr_to_proto(&expr.0, codec)) - .collect::>>()?; - - let limit = exec.limit_options().map(|config| protobuf::AggLimit { - limit: config.limit() as u64, - descending: config.descending(), - }); - - Ok(protobuf::PhysicalPlanNode { - physical_plan_type: Some(PhysicalPlanType::Aggregate(Box::new( - protobuf::AggregateExecNode { - group_expr, - group_expr_name: group_names, - aggr_expr: agg, - filter_expr: filter, - aggr_expr_name: agg_names, - mode: agg_mode as i32, - input: Some(Box::new(input)), - input_schema: Some(input_schema.as_ref().try_into()?), - null_expr, - groups, - limit, - has_grouping_set: exec.group_expr().has_grouping_set(), - dynamic_filter: exec - .dynamic_filter_expr() - .map(|df| { - let df_expr: Arc = - Arc::clone(df) as Arc; - proto_converter.physical_expr_to_proto(&df_expr, codec) - }) - .transpose()?, - }, - ))), - }) + }; + let encode_ctx = ExecutionPlanEncodeCtx::new(&encoder); + exec.try_to_proto(&encode_ctx)? + .ok_or_else(|| internal_datafusion_err!("AggregateExec is not serializable")) } fn try_from_empty_exec(