feat: support null_treatment for aggegrate udf

Author: dqkqd

datafusion/proto/proto/datafusion.proto

@@ -518,6 +518,7 @@ message AggregateUDFExprNode {
   LogicalExprNode filter = 3;
   repeated SortExprNode order_by = 4;
   optional bytes fun_definition = 6;
+  NullTreatment null_treatment = 7;
 }
 
 message ScalarUDFExprNode {

datafusion/proto/src/generated/pbjson.rs

@@ -587,14 +587,23 @@ pub fn parse_expr(
                     .udaf(&pb.fun_name)
                     .or_else(|_| codec.try_decode_udaf(&pb.fun_name, &[]))?,
             };
+            let null_treatment: Option<NullTreatment> =
+                protobuf::NullTreatment::try_from(pb.null_treatment)
+                    .map_err(|_| {
+                        proto_error(format!(
+                        "Received a AggregateUdfExprNode message with unknown NullTreatment {}",
+                        pb.null_treatment
+                    ))
+                    })?
+                    .into();
 
             Ok(Expr::AggregateFunction(expr::AggregateFunction::new_udf(
                 agg_fn,
                 parse_exprs(&pb.args, registry, codec)?,
                 pb.distinct,
                 parse_optional_expr(pb.filter.as_deref(), registry, codec)?.map(Box::new),
                 parse_sorts(&pb.order_by, registry, codec)?,
-                None,
+                null_treatment,
             )))
         }
 

datafusion/proto/src/generated/prost.rs

@@ -368,7 +368,7 @@ pub fn serialize_expr(
                     ref distinct,
                     ref filter,
                     ref order_by,
-                    null_treatment: _,
+                    ref null_treatment,
                 },
         }) => {
             let mut buf = Vec::new();
@@ -385,6 +385,8 @@ pub fn serialize_expr(
                         },
                         order_by: serialize_sorts(order_by, codec)?,
                         fun_definition: (!buf.is_empty()).then_some(buf),
+                        null_treatment: protobuf::NullTreatment::from(null_treatment)
+                            .into(),
                     },
                 ))),
             }