[FLINK-12073][table-planner-blink] Support appropriate precision and scale processing of Decimal (apache#8088)

Author: JingsongLi

flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/api/TableImpl.scala

@@ -18,10 +18,10 @@
 
 package org.apache.flink.table.api
 
+import org.apache.flink.table.`type`.TypeConverters.createExternalTypeInfoFromInternalType
 import org.apache.flink.table.calcite.FlinkTypeFactory._
 import org.apache.flink.table.expressions.Expression
 import org.apache.flink.table.functions.TemporalTableFunction
-import org.apache.flink.table.`type`.TypeConverters.createInternalTypeInfoFromInternalType
 
 import org.apache.calcite.rel.RelNode
 
@@ -45,7 +45,7 @@ class TableImpl(val tableEnv: TableEnvironment, relNode: RelNode) extends Table
     val fieldNames = rowType.getFieldList.map(_.getName)
     val fieldTypes = rowType.getFieldList map { tp =>
       val internalType = toInternalType(tp.getType)
-      createInternalTypeInfoFromInternalType(internalType)
+      createExternalTypeInfoFromInternalType(internalType)
     }
     new TableSchema(fieldNames.toArray, fieldTypes.toArray)
   }

flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/calcite/FlinkTypeFactory.scala

@@ -27,7 +27,7 @@ import org.apache.calcite.jdbc.JavaTypeFactoryImpl
 import org.apache.calcite.rel.`type`._
 import org.apache.calcite.sql.SqlIntervalQualifier
 import org.apache.calcite.sql.`type`.SqlTypeName._
-import org.apache.calcite.sql.`type`.{BasicSqlType, SqlTypeName}
+import org.apache.calcite.sql.`type`.{BasicSqlType, SqlTypeName, SqlTypeUtil}
 import org.apache.calcite.sql.parser.SqlParserPos
 import org.apache.calcite.util.ConversionUtil
 
@@ -347,6 +347,26 @@ class FlinkTypeFactory(typeSystem: RelDataTypeSystem) extends JavaTypeFactoryImp
   override def getDefaultCharset: Charset = {
     Charset.forName(ConversionUtil.NATIVE_UTF16_CHARSET_NAME)
   }
+
+  /**
+    * Calcite's default impl for division is apparently borrowed from T-SQL,
+    * but the details are a little different, e.g. when Decimal(34,0)/Decimal(10,0)
+    * To avoid confusion, follow the exact T-SQL behavior.
+    * Note that for (+-*), Calcite is also different from T-SQL;
+    * however, Calcite conforms to SQL2003 while T-SQL does not.
+    * therefore we keep Calcite's behavior on (+-*).
+    */
+  override def createDecimalQuotient(type1: RelDataType, type2: RelDataType): RelDataType = {
+    if (SqlTypeUtil.isExactNumeric(type1) && SqlTypeUtil.isExactNumeric(type2) &&
+        (SqlTypeUtil.isDecimal(type1) || SqlTypeUtil.isDecimal(type2))) {
+      val result = DecimalType.inferDivisionType(
+        type1.getPrecision, type1.getScale,
+        type2.getPrecision, type2.getScale)
+      createSqlType(SqlTypeName.DECIMAL, result.precision, result.scale)
+    } else {
+      null
+    }
+  }
 }
 
 object FlinkTypeFactory {

flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/calcite/FlinkTypeSystem.scala

@@ -18,9 +18,11 @@
 
 package org.apache.flink.table.calcite
 
-import org.apache.calcite.rel.`type`.RelDataTypeSystemImpl
+import org.apache.flink.table.`type`.{DecimalType, InternalType, InternalTypes}
+import org.apache.flink.table.typeutils.TypeCheckUtils
+
+import org.apache.calcite.rel.`type`.{RelDataType, RelDataTypeFactory, RelDataTypeSystemImpl}
 import org.apache.calcite.sql.`type`.SqlTypeName
-import org.apache.flink.table.`type`.DecimalType
 
 /**
   * Custom type system for Flink.
@@ -51,4 +53,43 @@ class FlinkTypeSystem extends RelDataTypeSystemImpl {
   // this fixes the problem of CASE WHEN with different length string literals but get wrong
   // result with additional space suffix
   override def shouldConvertRaggedUnionTypesToVarying(): Boolean = true
+
+  override def deriveAvgAggType(
+      typeFactory: RelDataTypeFactory, argType: RelDataType): RelDataType = {
+    val argTypeInfo = FlinkTypeFactory.toInternalType(argType)
+    val avgType = FlinkTypeSystem.deriveAvgAggType(argTypeInfo)
+    typeFactory.asInstanceOf[FlinkTypeFactory].createTypeFromInternalType(
+      avgType, argType.isNullable)
+  }
+
+  override def deriveSumType(
+      typeFactory: RelDataTypeFactory, argType: RelDataType): RelDataType = {
+    val argTypeInfo = FlinkTypeFactory.toInternalType(argType)
+    val sumType = FlinkTypeSystem.deriveSumType(argTypeInfo)
+    typeFactory.asInstanceOf[FlinkTypeFactory].createTypeFromInternalType(
+      sumType, argType.isNullable)
+  }
+}
+
+object FlinkTypeSystem {
+
+  def deriveAvgAggType(argType: InternalType): InternalType = argType match {
+    case dt: DecimalType =>
+      val result = DecimalType.inferAggAvgType(dt.scale())
+      DecimalType.of(result.precision(), result.scale())
+    case nt if TypeCheckUtils.isNumeric(nt) => InternalTypes.DOUBLE
+    case _ =>
+      throw new RuntimeException("Unsupported argType for AVG(): " + argType)
+  }
+
+  def deriveSumType(argType: InternalType): InternalType = argType match {
+    case dt: DecimalType =>
+      val result = DecimalType.inferAggSumType(dt.scale())
+      DecimalType.of(result.precision(), result.scale())
+    case nt if TypeCheckUtils.isNumeric(nt) =>
+      argType
+    case _ =>
+      throw new RuntimeException("Unsupported argType for SUM(): " + argType)
+  }
+
 }