fix test

Author: shuo.cs

flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/aggfunctions/AvgAggFunction.java

@@ -24,6 +24,7 @@
 import org.apache.flink.table.expressions.UnresolvedReferenceExpression;
 import org.apache.flink.table.types.DataType;
 import org.apache.flink.table.types.logical.DecimalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
 import org.apache.flink.table.types.logical.utils.LogicalTypeMerging;
 
 import java.math.BigDecimal;
@@ -95,7 +96,11 @@ public Expression[] mergeExpressions() {
     }
 
     private UnresolvedCallExpression adjustSumType(UnresolvedCallExpression sumExpr) {
-        return cast(sumExpr, typeLiteral(getSumType()));
+        if (getResultType().getLogicalType().getTypeRoot() == LogicalTypeRoot.DECIMAL) {
+            return cast(sumExpr, typeLiteral(getResultType()));
+        } else {
+            return sumExpr;
+        }
     }
 
     /** If all input are nulls, count will be 0 and we will get null after the division. */

flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/aggfunctions/Sum0AggFunction.java

@@ -20,19 +20,23 @@
 
 import org.apache.flink.table.api.DataTypes;
 import org.apache.flink.table.expressions.Expression;
+import org.apache.flink.table.expressions.UnresolvedCallExpression;
 import org.apache.flink.table.expressions.UnresolvedReferenceExpression;
 import org.apache.flink.table.types.DataType;
 import org.apache.flink.table.types.logical.DecimalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
 import org.apache.flink.table.types.logical.utils.LogicalTypeMerging;
 
 import java.math.BigDecimal;
 
 import static org.apache.flink.table.expressions.ApiExpressionUtils.unresolvedRef;
+import static org.apache.flink.table.planner.expressions.ExpressionBuilder.cast;
 import static org.apache.flink.table.planner.expressions.ExpressionBuilder.ifThenElse;
 import static org.apache.flink.table.planner.expressions.ExpressionBuilder.isNull;
 import static org.apache.flink.table.planner.expressions.ExpressionBuilder.literal;
 import static org.apache.flink.table.planner.expressions.ExpressionBuilder.minus;
 import static org.apache.flink.table.planner.expressions.ExpressionBuilder.plus;
+import static org.apache.flink.table.planner.expressions.ExpressionBuilder.typeLiteral;
 
 /** built-in sum0 aggregate function. */
 public abstract class Sum0AggFunction extends DeclarativeAggregateFunction {
@@ -56,20 +60,29 @@ public DataType[] getAggBufferTypes() {
     @Override
     public Expression[] accumulateExpressions() {
         return new Expression[] {
-            /* sum0 = */ ifThenElse(isNull(operand(0)), sum0, plus(sum0, operand(0)))
+            /* sum0 = */ adjustSumType(ifThenElse(isNull(operand(0)), sum0, plus(sum0, operand(0))))
         };
     }
 
     @Override
     public Expression[] retractExpressions() {
         return new Expression[] {
-            /* sum0 = */ ifThenElse(isNull(operand(0)), sum0, minus(sum0, operand(0)))
+            /* sum0 = */ adjustSumType(
+                    ifThenElse(isNull(operand(0)), sum0, minus(sum0, operand(0))))
         };
     }
 
     @Override
     public Expression[] mergeExpressions() {
-        return new Expression[] {/* sum0 = */ plus(sum0, mergeOperand(sum0))};
+        return new Expression[] {/* sum0 = */ adjustSumType(plus(sum0, mergeOperand(sum0)))};
+    }
+
+    private UnresolvedCallExpression adjustSumType(UnresolvedCallExpression sumExpr) {
+        if (getResultType().getLogicalType().getTypeRoot() == LogicalTypeRoot.DECIMAL) {
+            return cast(sumExpr, typeLiteral(getResultType()));
+        } else {
+            return sumExpr;
+        }
     }
 
     @Override

flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/aggfunctions/SumAggFunction.java

@@ -25,6 +25,7 @@
 import org.apache.flink.table.expressions.UnresolvedReferenceExpression;
 import org.apache.flink.table.types.DataType;
 import org.apache.flink.table.types.logical.DecimalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
 import org.apache.flink.table.types.logical.utils.LogicalTypeMerging;
 
 import static org.apache.flink.table.expressions.ApiExpressionUtils.unresolvedRef;
@@ -89,7 +90,11 @@ public Expression[] mergeExpressions() {
     }
 
     private UnresolvedCallExpression adjustSumType(UnresolvedCallExpression sumExpr) {
-        return cast(sumExpr, typeLiteral(getResultType()));
+        if (getResultType().getLogicalType().getTypeRoot() == LogicalTypeRoot.DECIMAL) {
+            return cast(sumExpr, typeLiteral(getResultType()));
+        } else {
+            return sumExpr;
+        }
     }
 
     @Override

flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/functions/aggfunctions/SumWithRetractAggFunction.java

@@ -24,6 +24,7 @@
 import org.apache.flink.table.expressions.UnresolvedReferenceExpression;
 import org.apache.flink.table.types.DataType;
 import org.apache.flink.table.types.logical.DecimalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
 import org.apache.flink.table.types.logical.utils.LogicalTypeMerging;
 
 import static org.apache.flink.table.expressions.ApiExpressionUtils.unresolvedRef;
@@ -103,7 +104,11 @@ public Expression[] mergeExpressions() {
     }
 
     private UnresolvedCallExpression adjustSumType(UnresolvedCallExpression sumExpr) {
-        return cast(sumExpr, typeLiteral(getResultType()));
+        if (getResultType().getLogicalType().getTypeRoot() == LogicalTypeRoot.DECIMAL) {
+            return cast(sumExpr, typeLiteral(getResultType()));
+        } else {
+            return sumExpr;
+        }
     }
 
     @Override