Serialize and deserialize NullTreatment (#261)

datafusion/proto/proto/datafusion.proto

@@ -515,6 +515,7 @@ message AggregateExprNode {
   bool distinct = 3;
   LogicalExprNode filter = 4;
   repeated LogicalExprNode order_by = 5;
+  bool ignore_nulls = 6;
 }
 
 message AggregateUDFExprNode {
@@ -524,6 +525,7 @@ message AggregateUDFExprNode {
   repeated LogicalExprNode order_by = 4;
   bool distinct = 5;
   optional bytes fun_definition = 6;
+  bool ignore_nulls = 7;
 }
 
 message ScalarUDFExprNode {

datafusion/proto/src/logical_plan/from_proto.rs

@@ -18,6 +18,7 @@
 use std::sync::Arc;
 
 use datafusion::execution::registry::FunctionRegistry;
+use datafusion::sql::sqlparser::ast::NullTreatment;
 use datafusion_common::{
     internal_err, plan_datafusion_err, DataFusionError, Result, ScalarValue,
     TableReference, UnnestOptions,
@@ -410,16 +411,14 @@ pub fn parse_expr(
             }
         }
         ExprType::AggregateExpr(expr) => {
-            let fun = parse_i32_to_aggregate_function(&expr.aggr_function)?;
-
             Ok(Expr::AggregateFunction(expr::AggregateFunction::new(
-                fun,
+                parse_i32_to_aggregate_function(&expr.aggr_function)?,
                 parse_exprs(&expr.expr, registry, codec)?,
                 expr.distinct,
                 parse_optional_expr(expr.filter.as_deref(), registry, codec)?
                     .map(Box::new),
                 parse_vec_expr(&expr.order_by, registry, codec)?,
-                None,
+                expr.ignore_nulls.then_some(NullTreatment::IgnoreNulls),
             )))
         }
         ExprType::Alias(alias) => Ok(Expr::Alias(Alias::new(
@@ -661,7 +660,7 @@ pub fn parse_expr(
                 pb.distinct,
                 parse_optional_expr(pb.filter.as_deref(), registry, codec)?.map(Box::new),
                 parse_vec_expr(&pb.order_by, registry, codec)?,
-                None,
+                pb.ignore_nulls.then_some(NullTreatment::IgnoreNulls),
             )))
         }
 

datafusion/proto/src/logical_plan/to_proto.rs

@@ -19,6 +19,7 @@
 //! DataFusion logical plans to be serialized and transmitted between
 //! processes.
 
+use datafusion::sql::sqlparser::ast::NullTreatment;
 use datafusion_common::{TableReference, UnnestOptions};
 use datafusion_expr::expr::{
     self, AggregateFunctionDefinition, Alias, Between, BinaryExpr, Cast, GroupingSet,
@@ -401,12 +402,12 @@ pub fn serialize_expr(
             }
         }
         Expr::AggregateFunction(expr::AggregateFunction {
-            ref func_def,
-            ref args,
-            ref distinct,
-            ref filter,
-            ref order_by,
-            null_treatment: _,
+            func_def,
+            args,
+            distinct,
+            filter,
+            order_by,
+            null_treatment,
         }) => match func_def {
             AggregateFunctionDefinition::BuiltIn(fun) => {
                 let aggr_function = match fun {
@@ -479,6 +480,7 @@ pub fn serialize_expr(
                         Some(e) => serialize_exprs(e, codec)?,
                         None => vec![],
                     },
+                    ignore_nulls: null_treatment == &Some(NullTreatment::IgnoreNulls),
                 };
                 protobuf::LogicalExprNode {
                     expr_type: Some(ExprType::AggregateExpr(Box::new(aggregate_expr))),
@@ -504,6 +506,8 @@ pub fn serialize_expr(
                             },
                             distinct: *distinct,
                             fun_definition: (!buf.is_empty()).then_some(buf),
+                            ignore_nulls: null_treatment
+                                == &Some(NullTreatment::IgnoreNulls),
                         },
                     ))),
                 }