From b398e8c507c8a4db9da8300cc7bf6c2422574aa1 Mon Sep 17 00:00:00 2001 From: Jiawei Zhao Date: Tue, 21 Jul 2026 17:30:44 +0800 Subject: [PATCH] refactor(proto): migrate window serde Window plans still relied on centralized protobuf dispatch, keeping serialization separate from the execution plans that own their state. Both window executors share one protobuf variant. Decoding must inspect `input_order_mode` before selecting the concrete plan. Add plan-local encoders for both executors and a decoder for the shared node. Keep the legacy helpers as deprecated delegates while preserving window frames and UDF payloads on the existing wire format. CLOSES #23513 Signed-off-by: Jiawei Zhao --- .../src/windows/bounded_window_agg_exec.rs | 47 +++ .../src/windows/window_agg_exec.rs | 383 ++++++++++++++++++ datafusion/proto/src/physical_plan/mod.rs | 176 ++------ 3 files changed, 468 insertions(+), 138 deletions(-) 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") }) }