refactor(rust): Correctly support more types in new-streaming sum

crates/polars-core/src/datatypes/any_value.rs

@@ -341,17 +341,21 @@ impl<'a> Deserialize<'a> for AnyValue<'static> {
 }
 
 impl AnyValue<'static> {
-    pub fn zero(dtype: &DataType) -> Self {
+    pub fn zero_sum(dtype: &DataType) -> Self {
         match dtype {
             DataType::String => AnyValue::StringOwned(PlSmallStr::EMPTY),
-            DataType::Boolean => AnyValue::Boolean(false),
-            // SAFETY:
-            // Numeric values are static, inform the compiler of this.
+            DataType::Binary => AnyValue::BinaryOwned(Vec::new()),
+            DataType::Boolean => (0 as IdxSize).into(),
+            // SAFETY: numeric values are static, inform the compiler of this.
             d if d.is_numeric() => unsafe {
                 std::mem::transmute::<AnyValue<'_>, AnyValue<'static>>(
                     AnyValue::UInt8(0).cast(dtype),
                 )
             },
+            DataType::Duration(unit) => AnyValue::Duration(0, *unit),
+            DataType::Decimal(_p, s) => {
+                AnyValue::Decimal(0, s.expect("unknown scale during execution"))
+            },
             _ => AnyValue::Null,
         }
     }
@@ -836,7 +840,21 @@ impl<'a> AnyValue<'a> {
             (UInt64(l), UInt64(r)) => UInt64(l + r),
             (Float32(l), Float32(r)) => Float32(l + r),
             (Float64(l), Float64(r)) => Float64(l + r),
-            _ => todo!(),
+            (Duration(l, lu), Duration(r, ru)) => {
+                if lu != ru {
+                    unimplemented!("adding durations with different units is not supported here");
+                }
+
+                Duration(l + r, *lu)
+            },
+            (Decimal(l, ls), Decimal(r, rs)) => {
+                if ls != rs {
+                    unimplemented!("adding decimals with different scales is not supported here");
+                }
+
+                Decimal(l + r, *ls)
+            },
+            _ => unimplemented!(),
         }
     }
 

crates/polars-expr/src/reduce/sum.rs

@@ -13,6 +13,7 @@ impl SumReduce {
         // returning the empty sum to be consistent.
         use DataType::*;
         let dtype = match dtype {
+            Boolean => IDX_DTYPE,
             Int8 | UInt8 | Int16 | UInt16 => Int64,
             dt => dt,
         };
@@ -22,7 +23,7 @@ impl SumReduce {
 
 impl Reduction for SumReduce {
     fn new_reducer(&self) -> Box<dyn ReductionState> {
-        let value = Scalar::new(self.dtype.clone(), AnyValue::zero(&self.dtype));
+        let value = Scalar::new(self.dtype.clone(), AnyValue::zero_sum(&self.dtype));
         Box::new(SumReduceState { value })
     }
 }