diff --git a/datafusion/physical-plan/src/windows/bounded_window_agg_exec.rs b/datafusion/physical-plan/src/windows/bounded_window_agg_exec.rs index 97cafd24c0a0d..defa6182d174a 100644 --- a/datafusion/physical-plan/src/windows/bounded_window_agg_exec.rs +++ b/datafusion/physical-plan/src/windows/bounded_window_agg_exec.rs @@ -402,6 +402,53 @@ impl ExecutionPlan for BoundedWindowAggExec { fn cardinality_effect(&self) -> CardinalityEffect { CardinalityEffect::Equal } + + #[cfg(feature = "proto")] + fn try_to_proto( + &self, + ctx: &crate::proto::ExecutionPlanEncodeCtx<'_>, + ) -> Result> { + use super::window_agg_exec::encode_physical_window_expr; + use datafusion_proto_common::protobuf_common::EmptyMessage; + use datafusion_proto_models::protobuf; + use protobuf::window_agg_exec_node::InputOrderMode as ProtoInputOrderMode; + + let input = ctx.encode_child(self.input())?; + let window_expr = self + .window_expr() + .iter() + .map(|expr| encode_physical_window_expr(expr, ctx)) + .collect::>>()?; + let partition_keys = self + .partition_keys() + .iter() + .map(|expr| ctx.encode_expr(expr)) + .collect::>>()?; + let input_order_mode = match &self.input_order_mode { + InputOrderMode::Linear => ProtoInputOrderMode::Linear(EmptyMessage {}), + InputOrderMode::PartiallySorted(columns) => { + ProtoInputOrderMode::PartiallySorted( + protobuf::PartiallySortedInputOrderMode { + columns: columns.iter().map(|column| *column as u64).collect(), + }, + ) + } + InputOrderMode::Sorted => ProtoInputOrderMode::Sorted(EmptyMessage {}), + }; + + Ok(Some(protobuf::PhysicalPlanNode { + physical_plan_type: Some( + protobuf::physical_plan_node::PhysicalPlanType::Window(Box::new( + protobuf::WindowAggExecNode { + input: Some(Box::new(input)), + window_expr, + partition_keys, + input_order_mode: Some(input_order_mode), + }, + )), + ), + })) + } } /// Trait that specifies how we search for (or calculate) partitions. It has two diff --git a/datafusion/physical-plan/src/windows/window_agg_exec.rs b/datafusion/physical-plan/src/windows/window_agg_exec.rs index 4e8dbc06f09a9..f76a4eaddbc16 100644 --- a/datafusion/physical-plan/src/windows/window_agg_exec.rs +++ b/datafusion/physical-plan/src/windows/window_agg_exec.rs @@ -318,6 +318,389 @@ impl ExecutionPlan for WindowAggExec { fn cardinality_effect(&self) -> CardinalityEffect { CardinalityEffect::Equal } + + #[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 window_expr = self + .window_expr() + .iter() + .map(|expr| encode_physical_window_expr(expr, ctx)) + .collect::>>()?; + let partition_keys = self + .partition_keys() + .iter() + .map(|expr| ctx.encode_expr(expr)) + .collect::>>()?; + + Ok(Some(protobuf::PhysicalPlanNode { + physical_plan_type: Some( + protobuf::physical_plan_node::PhysicalPlanType::Window(Box::new( + protobuf::WindowAggExecNode { + input: Some(Box::new(input)), + window_expr, + partition_keys, + input_order_mode: None, + }, + )), + ), + })) + } +} + +#[cfg(feature = "proto")] +impl WindowAggExec { + /// Reconstruct a window plan from its protobuf representation. + /// + /// This returns a [`WindowAggExec`] when `input_order_mode` is absent and a + /// [`BoundedWindowAggExec`] when it is present. + /// + /// [`BoundedWindowAggExec`]: crate::windows::BoundedWindowAggExec + pub fn try_from_proto( + node: &datafusion_proto_models::protobuf::PhysicalPlanNode, + ctx: &crate::proto::ExecutionPlanDecodeCtx<'_>, + ) -> Result> { + use super::BoundedWindowAggExec; + use crate::InputOrderMode; + use datafusion_proto_models::protobuf; + use protobuf::window_agg_exec_node::InputOrderMode as ProtoInputOrderMode; + + let window_agg = crate::expect_plan_variant!( + node, + protobuf::physical_plan_node::PhysicalPlanType::Window, + "WindowAggExec", + ); + let input = ctx.decode_required_child( + window_agg.input.as_deref(), + "WindowAggExec", + "input", + )?; + let input_schema = input.schema(); + let window_expr = window_agg + .window_expr + .iter() + .map(|expr| decode_physical_window_expr(expr, ctx, input_schema.as_ref())) + .collect::>>()?; + let partition_keys = window_agg + .partition_keys + .iter() + .map(|expr| ctx.decode_expr(expr, input_schema.as_ref())) + .collect::>>()?; + + if let Some(input_order_mode) = window_agg.input_order_mode.as_ref() { + let input_order_mode = match input_order_mode { + ProtoInputOrderMode::Linear(_) => InputOrderMode::Linear, + ProtoInputOrderMode::PartiallySorted( + protobuf::PartiallySortedInputOrderMode { columns }, + ) => InputOrderMode::PartiallySorted( + columns.iter().map(|column| *column as usize).collect(), + ), + ProtoInputOrderMode::Sorted(_) => InputOrderMode::Sorted, + }; + Ok(Arc::new(BoundedWindowAggExec::try_new( + window_expr, + input, + input_order_mode, + !partition_keys.is_empty(), + )?)) + } else { + Ok(Arc::new(WindowAggExec::try_new( + window_expr, + input, + !partition_keys.is_empty(), + )?)) + } + } +} + +#[cfg(feature = "proto")] +pub(crate) fn encode_physical_window_expr( + window_expr: &Arc, + ctx: &crate::proto::ExecutionPlanEncodeCtx<'_>, +) -> Result { + use super::{PlainAggregateWindowExpr, StandardWindowExpr, WindowUDFExpr}; + use datafusion_common::not_impl_err; + use datafusion_physical_expr::window::SlidingAggregateWindowExpr; + use datafusion_proto_models::protobuf::{self, physical_window_expr_node}; + + let expr = window_expr.as_any(); + let mut args = window_expr.expressions().to_vec(); + let window_frame = window_expr.get_window_frame(); + let (window_function, fun_definition, ignore_nulls, distinct) = + if let Some(plain) = expr.downcast_ref::() { + let aggregate_expr = plain.get_aggregate_expr(); + ( + physical_window_expr_node::WindowFunction::UserDefinedAggrFunction( + aggregate_expr.fun().name().to_string(), + ), + ctx.encode_udaf(aggregate_expr.fun())?, + aggregate_expr.ignore_nulls(), + aggregate_expr.is_distinct(), + ) + } else if let Some(sliding) = expr.downcast_ref::() { + let aggregate_expr = sliding.get_aggregate_expr(); + ( + physical_window_expr_node::WindowFunction::UserDefinedAggrFunction( + aggregate_expr.fun().name().to_string(), + ), + ctx.encode_udaf(aggregate_expr.fun())?, + aggregate_expr.ignore_nulls(), + aggregate_expr.is_distinct(), + ) + } else if let Some(standard) = expr.downcast_ref::() { + if let Some(window_udf) = standard + .get_standard_func_expr() + .as_any() + .downcast_ref::() + { + args = window_udf.args().to_vec(); + ( + physical_window_expr_node::WindowFunction::UserDefinedWindowFunction( + window_udf.fun().name().to_string(), + ), + ctx.encode_udwf(window_udf.fun().as_ref())?, + false, + false, + ) + } else { + return not_impl_err!( + "User-defined window function not supported: {window_expr:?}" + ); + } + } else { + return not_impl_err!("WindowExpr not supported: {window_expr:?}"); + }; + + let args = ctx.encode_expressions(&args)?; + let partition_by = ctx.encode_expressions(window_expr.partition_by())?; + let order_by = window_expr + .order_by() + .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::>>()?; + + Ok(protobuf::PhysicalWindowExprNode { + args, + partition_by, + order_by, + window_frame: Some(encode_window_frame(window_frame.as_ref())?), + window_function: Some(window_function), + name: window_expr.name().to_string(), + fun_definition, + ignore_nulls, + distinct, + }) +} + +#[cfg(feature = "proto")] +fn decode_physical_window_expr( + proto: &datafusion_proto_models::protobuf::PhysicalWindowExprNode, + ctx: &crate::proto::ExecutionPlanDecodeCtx<'_>, + input_schema: &arrow::datatypes::Schema, +) -> Result> { + use super::{create_window_expr, schema_add_window_field}; + use arrow::compute::SortOptions; + use datafusion_common::{internal_datafusion_err, internal_err}; + use datafusion_expr::WindowFunctionDefinition; + use datafusion_proto_models::protobuf::physical_window_expr_node; + + let args = proto + .args + .iter() + .map(|expr| ctx.decode_expr(expr, input_schema)) + .collect::>>()?; + let partition_by = proto + .partition_by + .iter() + .map(|expr| ctx.decode_expr(expr, input_schema)) + .collect::>>()?; + let order_by = proto + .order_by + .iter() + .map(|sort_expr| { + let expr = sort_expr.expr.as_ref().ok_or_else(|| { + internal_datafusion_err!( + "Missing expr in window order_by sort expression" + ) + })?; + Ok(PhysicalSortExpr { + expr: ctx.decode_expr(expr, input_schema)?, + options: SortOptions { + descending: !sort_expr.asc, + nulls_first: sort_expr.nulls_first, + }, + }) + }) + .collect::>>()?; + let window_frame = proto + .window_frame + .as_ref() + .map(decode_window_frame) + .transpose()? + .ok_or_else(|| { + internal_datafusion_err!("Missing required field 'window_frame' in protobuf") + })?; + let function = match proto.window_function.as_ref() { + Some(physical_window_expr_node::WindowFunction::UserDefinedAggrFunction( + name, + )) => WindowFunctionDefinition::AggregateUDF( + ctx.decode_udaf(name, proto.fun_definition.as_deref())?, + ), + Some(physical_window_expr_node::WindowFunction::UserDefinedWindowFunction( + name, + )) => WindowFunctionDefinition::WindowUDF( + ctx.decode_udwf(name, proto.fun_definition.as_deref())?, + ), + None => { + return internal_err!("Missing required field 'window_function' in protobuf"); + } + }; + + let name = proto.name.clone(); + let extended_schema = schema_add_window_field(&args, input_schema, &function, &name)?; + create_window_expr( + &function, + name, + &args, + &partition_by, + &order_by, + Arc::new(window_frame), + extended_schema, + proto.ignore_nulls, + proto.distinct, + None, + ) +} + +#[cfg(feature = "proto")] +fn encode_window_frame( + window_frame: &datafusion_expr::WindowFrame, +) -> Result { + use datafusion_expr::WindowFrameUnits; + use datafusion_proto_models::protobuf; + + let units = match window_frame.units { + WindowFrameUnits::Rows => protobuf::WindowFrameUnits::Rows, + WindowFrameUnits::Range => protobuf::WindowFrameUnits::Range, + WindowFrameUnits::Groups => protobuf::WindowFrameUnits::Groups, + }; + Ok(protobuf::WindowFrame { + window_frame_units: units.into(), + start_bound: Some(encode_window_frame_bound(&window_frame.start_bound)?), + end_bound: Some(protobuf::window_frame::EndBound::Bound( + encode_window_frame_bound(&window_frame.end_bound)?, + )), + }) +} + +#[cfg(feature = "proto")] +fn encode_window_frame_bound( + bound: &datafusion_expr::WindowFrameBound, +) -> Result { + use datafusion_expr::WindowFrameBound; + use datafusion_proto_common::protobuf_common; + use datafusion_proto_models::protobuf; + + let encode_value = + |value: &datafusion_common::ScalarValue| -> Result { + Ok(value.try_into()?) + }; + Ok(match bound { + WindowFrameBound::CurrentRow => protobuf::WindowFrameBound { + window_frame_bound_type: protobuf::WindowFrameBoundType::CurrentRow.into(), + bound_value: None, + }, + WindowFrameBound::Preceding(value) => protobuf::WindowFrameBound { + window_frame_bound_type: protobuf::WindowFrameBoundType::Preceding.into(), + bound_value: Some(encode_value(value)?), + }, + WindowFrameBound::Following(value) => protobuf::WindowFrameBound { + window_frame_bound_type: protobuf::WindowFrameBoundType::Following.into(), + bound_value: Some(encode_value(value)?), + }, + }) +} + +#[cfg(feature = "proto")] +fn decode_window_frame( + window_frame: &datafusion_proto_models::protobuf::WindowFrame, +) -> Result { + use datafusion_common::internal_datafusion_err; + use datafusion_expr::{WindowFrame, WindowFrameBound, WindowFrameUnits}; + use datafusion_proto_models::protobuf; + + let units = protobuf::WindowFrameUnits::try_from(window_frame.window_frame_units) + .map_err(|_| { + internal_datafusion_err!( + "Received a WindowFrame message with unknown WindowFrameUnits {}", + window_frame.window_frame_units + ) + })?; + let units = match units { + protobuf::WindowFrameUnits::Rows => WindowFrameUnits::Rows, + protobuf::WindowFrameUnits::Range => WindowFrameUnits::Range, + protobuf::WindowFrameUnits::Groups => WindowFrameUnits::Groups, + }; + let start_bound = + decode_window_frame_bound(window_frame.start_bound.as_ref().ok_or_else( + || internal_datafusion_err!("Missing start_bound in WindowFrame"), + )?)?; + let end_bound = window_frame + .end_bound + .as_ref() + .map(|end_bound| match end_bound { + protobuf::window_frame::EndBound::Bound(bound) => { + decode_window_frame_bound(bound) + } + }) + .transpose()? + .unwrap_or(WindowFrameBound::CurrentRow); + Ok(WindowFrame::new_bounds(units, start_bound, end_bound)) +} + +#[cfg(feature = "proto")] +fn decode_window_frame_bound( + bound: &datafusion_proto_models::protobuf::WindowFrameBound, +) -> Result { + use datafusion_common::{ScalarValue, internal_datafusion_err}; + use datafusion_expr::WindowFrameBound; + use datafusion_proto_common::protobuf_common; + use datafusion_proto_models::protobuf; + + let decode_value = |value: &protobuf_common::ScalarValue| -> Result { + Ok(ScalarValue::try_from(value)?) + }; + let bound_type = protobuf::WindowFrameBoundType::try_from( + bound.window_frame_bound_type, + ) + .map_err(|_| { + internal_datafusion_err!( + "Received a WindowFrameBound message with unknown WindowFrameBoundType {}", + bound.window_frame_bound_type + ) + })?; + match bound_type { + protobuf::WindowFrameBoundType::CurrentRow => Ok(WindowFrameBound::CurrentRow), + protobuf::WindowFrameBoundType::Preceding => match &bound.bound_value { + Some(value) => Ok(WindowFrameBound::Preceding(decode_value(value)?)), + None => Ok(WindowFrameBound::Preceding(ScalarValue::UInt64(None))), + }, + protobuf::WindowFrameBoundType::Following => match &bound.bound_value { + Some(value) => Ok(WindowFrameBound::Following(decode_value(value)?)), + None => Ok(WindowFrameBound::Following(ScalarValue::UInt64(None))), + }, + } } /// Compute the window aggregate columns diff --git a/datafusion/proto/src/physical_plan/mod.rs b/datafusion/proto/src/physical_plan/mod.rs index cea334e42aace..595b36b79b082 100644 --- a/datafusion/proto/src/physical_plan/mod.rs +++ b/datafusion/proto/src/physical_plan/mod.rs @@ -102,7 +102,7 @@ use datafusion_physical_plan::sorts::sort_preserving_merge::SortPreservingMergeE use datafusion_physical_plan::union::{InterleaveExec, UnionExec}; use datafusion_physical_plan::unnest::{ListUnnest, UnnestExec}; use datafusion_physical_plan::windows::{BoundedWindowAggExec, WindowAggExec}; -use datafusion_physical_plan::{ExecutionPlan, InputOrderMode, PhysicalExpr, WindowExpr}; +use datafusion_physical_plan::{ExecutionPlan, PhysicalExpr}; use prost::Message; use prost::bytes::BufMut; @@ -113,20 +113,19 @@ 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_sort_exprs, 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_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::{ self, ListUnnest as ProtoListUnnest, SortMergeJoinExecNode, proto_error, - window_agg_exec_node, }; pub mod from_proto; @@ -799,8 +798,8 @@ pub trait PhysicalPlanNodeExt: Sized { PhysicalPlanType::LocalLimit(limit) => { self.try_into_local_limit_physical_plan(limit, ctx, proto_converter) } - PhysicalPlanType::Window(window_agg) => { - self.try_into_window_physical_plan(window_agg, ctx, proto_converter) + PhysicalPlanType::Window(_) => { + WindowAggExec::try_from_proto(self.node(), &decode_ctx) } PhysicalPlanType::Aggregate(hash_agg) => { self.try_into_aggregate_physical_plan(hash_agg, ctx, proto_converter) @@ -1033,22 +1032,6 @@ pub trait PhysicalPlanNodeExt: Sized { ); } - if let Some(exec) = plan.downcast_ref::() { - return protobuf::PhysicalPlanNode::try_from_window_agg_exec( - exec, - codec, - proto_converter, - ); - } - - if let Some(exec) = plan.downcast_ref::() { - return protobuf::PhysicalPlanNode::try_from_bounded_window_agg_exec( - exec, - codec, - proto_converter, - ); - } - if let Some(exec) = plan.downcast_ref::() && let Some(node) = protobuf::PhysicalPlanNode::try_from_data_sink_exec( exec, @@ -1534,61 +1517,27 @@ pub trait PhysicalPlanNodeExt: Sized { Ok(Arc::new(LocalLimitExec::new(input, limit.fetch as usize))) } + #[deprecated( + since = "55.0.0", + note = "unused by DataFusion; window plans deserialize via `WindowAggExec::try_from_proto`" + )] fn try_into_window_physical_plan( &self, window_agg: &protobuf::WindowAggExecNode, ctx: &PhysicalPlanDecodeContext<'_>, proto_converter: &dyn PhysicalProtoConverterExtension, ) -> Result> { - let input: Arc = - into_physical_plan(&window_agg.input, ctx, proto_converter)?; - let input_schema = input.schema(); - - let physical_window_expr: Vec> = window_agg - .window_expr - .iter() - .map(|window_expr| { - parse_physical_window_expr( - window_expr, - ctx, - input_schema.as_ref(), - proto_converter, - ) - }) - .collect::, _>>()?; - - let partition_keys = window_agg - .partition_keys - .iter() - .map(|expr| { - proto_converter.proto_to_physical_expr(expr, input.schema().as_ref(), ctx) - }) - .collect::>>>()?; - - if let Some(input_order_mode) = window_agg.input_order_mode.as_ref() { - let input_order_mode = match input_order_mode { - window_agg_exec_node::InputOrderMode::Linear(_) => InputOrderMode::Linear, - window_agg_exec_node::InputOrderMode::PartiallySorted( - protobuf::PartiallySortedInputOrderMode { columns }, - ) => InputOrderMode::PartiallySorted( - columns.iter().map(|c| *c as usize).collect(), - ), - window_agg_exec_node::InputOrderMode::Sorted(_) => InputOrderMode::Sorted, - }; - - Ok(Arc::new(BoundedWindowAggExec::try_new( - physical_window_expr, - input, - input_order_mode, - !partition_keys.is_empty(), - )?)) - } else { - Ok(Arc::new(WindowAggExec::try_new( - physical_window_expr, - input, - !partition_keys.is_empty(), - )?)) - } + let node = protobuf::PhysicalPlanNode { + physical_plan_type: Some(PhysicalPlanType::Window(Box::new( + window_agg.clone(), + ))), + }; + let decoder = ConverterPlanDecoder { + ctx, + proto_converter, + }; + let decode_ctx = ExecutionPlanDecodeCtx::new(&decoder); + WindowAggExec::try_from_proto(&node, &decode_ctx) } fn try_into_aggregate_physical_plan( @@ -3727,89 +3676,40 @@ pub trait PhysicalPlanNodeExt: Sized { }) } + #[deprecated( + since = "55.0.0", + note = "unused by DataFusion; `WindowAggExec` serializes itself via `ExecutionPlan::try_to_proto`" + )] fn try_from_window_agg_exec( exec: &WindowAggExec, codec: &dyn PhysicalExtensionCodec, proto_converter: &dyn PhysicalProtoConverterExtension, ) -> Result { - let input = protobuf::PhysicalPlanNode::try_from_physical_plan_with_converter( - exec.input().to_owned(), + let encoder = ConverterPlanEncoder { codec, proto_converter, - )?; - - let window_expr = exec - .window_expr() - .iter() - .map(|e| serialize_physical_window_expr(e, codec, proto_converter)) - .collect::>>()?; - - let partition_keys = exec - .partition_keys() - .iter() - .map(|e| proto_converter.physical_expr_to_proto(e, codec)) - .collect::>>()?; - - Ok(protobuf::PhysicalPlanNode { - physical_plan_type: Some(PhysicalPlanType::Window(Box::new( - protobuf::WindowAggExecNode { - input: Some(Box::new(input)), - window_expr, - partition_keys, - input_order_mode: None, - }, - ))), - }) + }; + let encode_ctx = ExecutionPlanEncodeCtx::new(&encoder); + exec.try_to_proto(&encode_ctx)? + .ok_or_else(|| internal_datafusion_err!("WindowAggExec is not serializable")) } + #[deprecated( + since = "55.0.0", + note = "unused by DataFusion; `BoundedWindowAggExec` serializes itself via `ExecutionPlan::try_to_proto`" + )] fn try_from_bounded_window_agg_exec( exec: &BoundedWindowAggExec, codec: &dyn PhysicalExtensionCodec, proto_converter: &dyn PhysicalProtoConverterExtension, ) -> Result { - let input = protobuf::PhysicalPlanNode::try_from_physical_plan_with_converter( - exec.input().to_owned(), + let encoder = ConverterPlanEncoder { codec, proto_converter, - )?; - - let window_expr = exec - .window_expr() - .iter() - .map(|e| serialize_physical_window_expr(e, codec, proto_converter)) - .collect::>>()?; - - let partition_keys = exec - .partition_keys() - .iter() - .map(|e| proto_converter.physical_expr_to_proto(e, codec)) - .collect::>>()?; - - let input_order_mode = match &exec.input_order_mode { - InputOrderMode::Linear => { - window_agg_exec_node::InputOrderMode::Linear(protobuf::EmptyMessage {}) - } - InputOrderMode::PartiallySorted(columns) => { - window_agg_exec_node::InputOrderMode::PartiallySorted( - protobuf::PartiallySortedInputOrderMode { - columns: columns.iter().map(|c| *c as u64).collect(), - }, - ) - } - InputOrderMode::Sorted => { - window_agg_exec_node::InputOrderMode::Sorted(protobuf::EmptyMessage {}) - } }; - - Ok(protobuf::PhysicalPlanNode { - physical_plan_type: Some(PhysicalPlanType::Window(Box::new( - protobuf::WindowAggExecNode { - input: Some(Box::new(input)), - window_expr, - partition_keys, - input_order_mode: Some(input_order_mode), - }, - ))), + let encode_ctx = ExecutionPlanEncodeCtx::new(&encoder); + exec.try_to_proto(&encode_ctx)?.ok_or_else(|| { + internal_datafusion_err!("BoundedWindowAggExec is not serializable") }) }