refactor(proto): migrate window serde

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<Option<datafusion_proto_models::protobuf::PhysicalPlanNode>> {
+        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::<Result<Vec<_>>>()?;
+        let partition_keys = self
+            .partition_keys()
+            .iter()
+            .map(|expr| ctx.encode_expr(expr))
+            .collect::<Result<Vec<_>>>()?;
+        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