Handle serde for ScalarUDF (#9395)

Author: yyy1000

datafusion/proto/proto/datafusion.proto

@@ -751,6 +751,7 @@ message AggregateUDFExprNode {
 message ScalarUDFExprNode {
   string fun_name = 1;
   repeated LogicalExprNode args = 2;
+  optional bytes fun_definition = 3;
 }
 
 enum BuiltInWindowFunction {

datafusion/proto/src/bytes/mod.rs

@@ -16,6 +16,7 @@
 // under the License.
 
 //! Serialization / Deserialization to Bytes
+use crate::logical_plan::to_proto::serialize_expr;
 use crate::logical_plan::{
     self, AsLogicalPlan, DefaultLogicalExtensionCodec, LogicalExtensionCodec,
 };
@@ -87,8 +88,8 @@ pub trait Serializeable: Sized {
 impl Serializeable for Expr {
     fn to_bytes(&self) -> Result<Bytes> {
         let mut buffer = BytesMut::new();
-        let protobuf: protobuf::LogicalExprNode = self
-            .try_into()
+        let extension_codec = DefaultLogicalExtensionCodec {};
+        let protobuf: protobuf::LogicalExprNode = serialize_expr(self, &extension_codec)
             .map_err(|e| plan_datafusion_err!("Error encoding expr as protobuf: {e}"))?;
 
         protobuf
@@ -177,7 +178,8 @@ impl Serializeable for Expr {
         let protobuf = protobuf::LogicalExprNode::decode(bytes)
             .map_err(|e| plan_datafusion_err!("Error decoding expr as protobuf: {e}"))?;
 
-        logical_plan::from_proto::parse_expr(&protobuf, registry)
+        let extension_codec = DefaultLogicalExtensionCodec {};
+        logical_plan::from_proto::parse_expr(&protobuf, registry, &extension_codec)
             .map_err(|e| plan_datafusion_err!("Error parsing protobuf into Expr: {e}"))
     }
 }