fix decimal precision error

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

Author: jayzhan211

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

@@ -47,7 +47,7 @@ use datafusion_expr::Accumulator;
 pub struct Sum {
     name: String,
     data_type: DataType,
-    /// `return_data_type` is different for ArrayAggregate cases, see `datafusion_expr::type_coercion::aggregates::sum_return_type`
+    // `return_data_type` is different for ArrayAggregate cases, see `datafusion_expr::type_coercion::aggregates::sum_return_type`
     return_data_type: DataType,
     expr: Arc<dyn PhysicalExpr>,
     nullable: bool,
@@ -90,6 +90,27 @@ macro_rules! downcast_sum {
 }
 pub(crate) use downcast_sum;
 
+// TODO: Replace with `downcast_sum` after most of the AggregateExpr differentiate `return_data_type` and `data_type`
+// The reason we have this is because using the name `return_data_type` makes more much sense to me,
+// instead of changing `data_type` to `return_data_type` all the AggregateExpr that have `downcast_sum`, introduce v2 is better.
+macro_rules! downcast_sum_v2 {
+    ($s:ident, $helper:ident) => {
+        match $s.return_data_type {
+            DataType::UInt64 => $helper!(UInt64Type, $s.return_data_type),
+            DataType::Int64 => $helper!(Int64Type, $s.return_data_type),
+            DataType::Float64 => $helper!(Float64Type, $s.return_data_type),
+            DataType::Decimal128(_, _) => $helper!(Decimal128Type, $s.return_data_type),
+            DataType::Decimal256(_, _) => $helper!(Decimal256Type, $s.return_data_type),
+            _ => not_impl_err!(
+                "Sum not supported for {}: {}",
+                $s.name,
+                $s.return_data_type
+            ),
+        }
+    };
+}
+pub(crate) use downcast_sum_v2;
+
 impl AggregateExpr for Sum {
     /// Return a reference to Any that can be used for downcasting
     fn as_any(&self) -> &dyn Any {
@@ -146,7 +167,7 @@ impl AggregateExpr for Sum {
                 Ok(Box::new(SumAccumulator::<$t>::new($dt.clone())))
             };
         }
-        downcast_sum!(self, helper)
+        downcast_sum_v2!(self, helper)
     }
 
     fn state_fields(&self) -> Result<Vec<Field>> {
@@ -178,7 +199,7 @@ impl AggregateExpr for Sum {
                 )))
             };
         }
-        downcast_sum!(self, helper)
+        downcast_sum_v2!(self, helper)
     }
 
     fn reverse_expr(&self) -> Option<Arc<dyn AggregateExpr>> {
@@ -191,7 +212,7 @@ impl AggregateExpr for Sum {
                 Ok(Box::new(SlidingSumAccumulator::<$t>::new($dt.clone())))
             };
         }
-        downcast_sum!(self, helper)
+        downcast_sum_v2!(self, helper)
     }
 }