sum_ansi_mode_checks_fix_tests

Author: coderfender

native/spark-expr/src/agg_funcs/sum_int.rs

@@ -27,7 +27,7 @@ use datafusion::common::{DataFusionError, Result as DFResult, ScalarValue};
 use datafusion::logical_expr::function::{AccumulatorArgs, StateFieldsArgs};
 use datafusion::logical_expr::Volatility::Immutable;
 use datafusion::logical_expr::{
-    Accumulator, AggregateUDFImpl, EmitTo, GroupsAccumulator, Signature,
+    Accumulator, AggregateUDFImpl, EmitTo, GroupsAccumulator, ReversedUDAF, Signature,
 };
 use std::{any::Any, sync::Arc};
 
@@ -93,6 +93,10 @@ impl AggregateUDFImpl for SumInteger {
     ) -> DFResult<Box<dyn GroupsAccumulator>> {
         Ok(Box::new(SumIntGroupsAccumulator::new(self.eval_mode)))
     }
+
+    fn reverse_expr(&self) -> ReversedUDAF {
+        ReversedUDAF::Identical
+    }
 }
 
 #[derive(Debug)]

spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala

@@ -62,6 +62,8 @@ import org.apache.comet.shims.CometExprShim
  */
 object QueryPlanSerde extends Logging with CometExprShim {
 
+  val integerTypes = Seq(ByteType, ShortType, IntegerType, LongType)
+
   /**
    * Mapping of Spark operator class to Comet operator handler.
    */
@@ -414,7 +416,7 @@ object QueryPlanSerde extends Logging with CometExprShim {
               }
             case s: Sum =>
               if (AggSerde.sumDataTypeSupported(s.dataType) && !s.dataType
-                  .isInstanceOf[DecimalType]) {
+                  .isInstanceOf[DecimalType] && !integerTypes.contains(s.dataType)) {
                 Some(agg)
               } else {
                 withInfo(windowExpr, s"datatype ${s.dataType} is not supported", expr)