cleanup

Signed-off-by: jayzhan211 <[email protected]>

Author: jayzhan211

datafusion/common/src/scalar.rs

@@ -758,40 +758,45 @@ impl ScalarValue {
             Some(v) => {
                 let array = PrimitiveArray::<T>::new(vec![v].into(), None)
                     .with_data_type(d.clone());
-                let res = Self::try_from_array(&array, 0).unwrap();
-                // let res2 = ScalarValue::List(Arc::new(array));
-                // thread 'tokio-runtime-worker' panicked at 'assertion failed: `(left == right)`
-                // left: `Int64(0)`,
-                // right: `List([PrimitiveArray<Int64>
-                // [
-                // 0,
-                // ]])`', /Users/jayzhan/arrow-datafusion/datafusion/common/src/scalar.rs:763:17
-                // assert_eq!(res, res2);
-                res
+                Self::try_from_array(&array, 0).unwrap()
             }
         }
     }
 
+    // ListArray compatible version of new_primitive
     pub fn new_primitives<T: ArrowPrimitiveType>(
         values: Vec<Option<T::Native>>,
         d: &DataType,
     ) -> Self {
-        if values.len() == 0 {
+        if values.is_empty() {
             return d.try_into().unwrap();
         }
+
+        // We need to convert it to ScalarValue::Primitive (Int64(0)) instead of ScalarValue::List (List([PrimitiveArray<Int64> [0]]))
         if values.len() == 1 {
             return Self::new_primitive::<T>(values[0], d);
         }
-        let mut vals = vec![];
+
+        let mut array = Vec::with_capacity(values.len());
+        let mut nulls = Vec::with_capacity(values.len());
+
         for a in values {
             match a {
-                None => return d.try_into().unwrap(),
-                Some(v) => vals.push(v),
+                Some(v) => {
+                    array.push(v);
+                    nulls.push(true);
+                }
+                None => {
+                    array.push(T::Native::default());
+                    nulls.push(false);
+                }
             }
         }
-        let array = PrimitiveArray::<T>::new(vals.into(), None).with_data_type(d.clone());
 
-        ScalarValue::List(Arc::new(array))
+        let arr = PrimitiveArray::<T>::new(array.into(), Some(NullBuffer::from(nulls)))
+            .with_data_type(d.clone());
+
+        ScalarValue::List(Arc::new(arr))
     }
 
     /// Create a decimal Scalar from value/precision and scale.

datafusion/expr/src/built_in_function.rs

@@ -532,10 +532,7 @@ impl BuiltinScalarFunction {
                 Ok(data_type)
             }
             BuiltinScalarFunction::ArrayAppend => Ok(input_expr_types[0].clone()),
-            BuiltinScalarFunction::ArrayAggregate => {
-                // TODO: Fix this
-                Ok(Int64)
-            }
+            BuiltinScalarFunction::ArrayAggregate => unimplemented!("ArrayAggregate is based on Aggreation function, so no return value for it."),
             BuiltinScalarFunction::ArrayConcat => {
                 let mut expr_type = Null;
                 let mut max_dims = 0;

datafusion/physical-expr/src/aggregate/sum.rs

@@ -233,12 +233,10 @@ impl<T: ArrowNumericType> Accumulator for SumAccumulator<T> {
     }
 
     fn evaluate(&self) -> Result<ScalarValue> {
-        if !self.sum.is_empty() {
-            let arr = ScalarValue::new_primitives::<T>(self.sum.clone(), &self.data_type);
-            Ok(arr)
-        } else {
-            Ok(ScalarValue::new_primitives::<T>(vec![], &self.data_type))
-        }
+        Ok(ScalarValue::new_primitives::<T>(
+            self.sum.clone(),
+            &self.data_type,
+        ))
     }
 
     fn size(&self) -> usize {

datafusion/proto/proto/datafusion.proto

@@ -620,8 +620,7 @@ enum ScalarFunction {
   ArrayEmpty = 115;
   ArrayPopBack = 116;
   StringToArray = 117;
-  ArraySum = 118;
-  ArrayAggregate = 119;
+  ArrayAggregate = 118;
 }
 
 message ScalarFunctionNode {

datafusion/proto/src/generated/pbjson.rs

@@ -16,7 +16,9 @@
 // under the License.
 
 use crate::planner::{ContextProvider, PlannerContext, SqlToRel};
-use datafusion_common::{not_impl_err, plan_err, DFSchema, DataFusionError, Result};
+use datafusion_common::{
+    internal_err, not_impl_err, plan_err, DFSchema, DataFusionError, Result, ScalarValue,
+};
 use datafusion_expr;
 use datafusion_expr::expr::{ScalarFunction, ScalarUDF};
 use datafusion_expr::function::suggest_valid_function;
@@ -62,33 +64,20 @@ impl<'a, S: ContextProvider> SqlToRel<'a, S> {
             // Translate array_aggregate to aggregate function with array argument.
             if fun == BuiltinScalarFunction::ArrayAggregate {
                 let fun = match &args[1] {
-                    Expr::Literal(datafusion_common::ScalarValue::Utf8(Some(name))) => {
-                        match name.as_str() {
-                            "sum" => AggregateFunction::Sum,
-                            "count" => AggregateFunction::Count,
-                            "min" => AggregateFunction::Min,
-                            "max" => AggregateFunction::Max,
-                            "avg" => AggregateFunction::Avg,
-                            _ => {
-                                return Err(DataFusionError::NotImplemented(format!(
-                                    "Aggregate function {name} is not implemented"
-                                )))
-                            }
+                    Expr::Literal(ScalarValue::Utf8(Some(name))) => match name.as_str() {
+                        "sum" => AggregateFunction::Sum,
+                        _ => {
+                            return not_impl_err!(
+                                "Aggregate function {name} is not implemented"
+                            )
                         }
-                    }
-                    _ => {
-                        return Err(DataFusionError::Internal(format!(
-                            "Aggregate function name is not a string"
-                        )))
-                    }
+                    },
+                    _ => return internal_err!("Aggregate function name is not a string"),
                 };
-                println!("args: {:?}", args);
                 let args = vec![args[0].to_owned()];
-                let e = Expr::AggregateFunction(expr::AggregateFunction::new(
+                return Ok(Expr::AggregateFunction(expr::AggregateFunction::new(
                     fun, args, false, None, None,
-                ));
-                println!("e: {:?}", e);
-                return Ok(e);
+                )));
             }
 
             return Ok(Expr::ScalarFunction(ScalarFunction::new(fun, args)));