Repository: flink
Updated Branches:
  refs/heads/release-1.5 81aa5acee -> 347ec3848


[FLINK-8821] [table] Fix non-terminating decimal error

This closes #5608.


Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/347ec384
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/347ec384
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/347ec384

Branch: refs/heads/release-1.5
Commit: 347ec3848ec603ac452e394a5211cf888db6663f
Parents: 81aa5ac
Author: Xpray <leonxp...@gmail.com>
Authored: Fri Mar 2 12:11:45 2018 +0800
Committer: Timo Walther <twal...@apache.org>
Committed: Fri Mar 2 09:34:24 2018 +0100

----------------------------------------------------------------------
 .../apache/flink/table/api/TableConfig.scala    | 21 ++++++++
 .../flink/table/codegen/CodeGenerator.scala     | 18 +++----
 .../table/codegen/calls/ScalarOperators.scala   | 29 ++++++++--
 .../functions/aggfunctions/AvgAggFunction.scala |  7 +--
 .../plan/nodes/dataset/DataSetAggregate.scala   |  3 +-
 .../nodes/dataset/DataSetWindowAggregate.scala  | 55 ++++++++++++-------
 .../datastream/DataStreamGroupAggregate.scala   |  1 +
 .../DataStreamGroupWindowAggregate.scala        |  6 ++-
 .../datastream/DataStreamOverAggregate.scala    | 12 +++--
 .../table/runtime/aggregate/AggregateUtil.scala | 57 ++++++++++++++------
 .../table/expressions/DecimalTypeTest.scala     | 11 +++-
 .../runtime/aggfunctions/AvgFunctionTest.scala  | 12 +++--
 .../table/runtime/batch/table/CalcITCase.scala  |  8 ++-
 13 files changed, 176 insertions(+), 64 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/347ec384/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/TableConfig.scala
----------------------------------------------------------------------
diff --git 
a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/TableConfig.scala
 
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/TableConfig.scala
index 6448657..c78a022 100644
--- 
a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/TableConfig.scala
+++ 
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/TableConfig.scala
@@ -18,6 +18,7 @@
 package org.apache.flink.table.api
 
 import _root_.java.util.TimeZone
+import _root_.java.math.MathContext
 
 import org.apache.flink.table.calcite.CalciteConfig
 
@@ -42,6 +43,12 @@ class TableConfig {
   private var calciteConfig = CalciteConfig.DEFAULT
 
   /**
+    * Defines the default context for decimal division calculation.
+    * We use Scala's default MathContext.DECIMAL128.
+    */
+  private var decimalContext = MathContext.DECIMAL128
+
+  /**
    * Sets the timezone for date/time/timestamp conversions.
    */
   def setTimeZone(timeZone: TimeZone): Unit = {
@@ -78,6 +85,20 @@ class TableConfig {
   def setCalciteConfig(calciteConfig: CalciteConfig): Unit = {
     this.calciteConfig = calciteConfig
   }
+
+  /**
+    * Returns the default context for decimal division calculation.
+    * [[_root_.java.math.MathContext#DECIMAL128]] by default.
+    */
+  def getDecimalContext: MathContext = decimalContext
+
+  /**
+    * Sets the default context for decimal division calculation.
+    * [[_root_.java.math.MathContext#DECIMAL128]] by default.
+    */
+  def setDecimalContext(mathContext: MathContext): Unit = {
+    this.decimalContext = mathContext
+  }
 }
 
 object TableConfig {

http://git-wip-us.apache.org/repos/asf/flink/blob/347ec384/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/CodeGenerator.scala
----------------------------------------------------------------------
diff --git 
a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/CodeGenerator.scala
 
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/CodeGenerator.scala
index 756a828..e4064d6 100644
--- 
a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/CodeGenerator.scala
+++ 
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/CodeGenerator.scala
@@ -742,56 +742,56 @@ abstract class CodeGenerator(
         val right = operands(1)
         requireNumeric(left)
         requireNumeric(right)
-        generateArithmeticOperator("+", nullCheck, resultType, left, right)
+        generateArithmeticOperator("+", nullCheck, resultType, left, right, 
config)
 
       case PLUS | DATETIME_PLUS if isTemporal(resultType) =>
         val left = operands.head
         val right = operands(1)
         requireTemporal(left)
         requireTemporal(right)
-        generateTemporalPlusMinus(plus = true, nullCheck, left, right)
+        generateTemporalPlusMinus(plus = true, nullCheck, left, right, config)
 
       case MINUS if isNumeric(resultType) =>
         val left = operands.head
         val right = operands(1)
         requireNumeric(left)
         requireNumeric(right)
-        generateArithmeticOperator("-", nullCheck, resultType, left, right)
+        generateArithmeticOperator("-", nullCheck, resultType, left, right, 
config)
 
       case MINUS | MINUS_DATE if isTemporal(resultType) =>
         val left = operands.head
         val right = operands(1)
         requireTemporal(left)
         requireTemporal(right)
-        generateTemporalPlusMinus(plus = false, nullCheck, left, right)
+        generateTemporalPlusMinus(plus = false, nullCheck, left, right, config)
 
       case MULTIPLY if isNumeric(resultType) =>
         val left = operands.head
         val right = operands(1)
         requireNumeric(left)
         requireNumeric(right)
-        generateArithmeticOperator("*", nullCheck, resultType, left, right)
+        generateArithmeticOperator("*", nullCheck, resultType, left, right, 
config)
 
       case MULTIPLY if isTimeInterval(resultType) =>
         val left = operands.head
         val right = operands(1)
         requireTimeInterval(left)
         requireNumeric(right)
-        generateArithmeticOperator("*", nullCheck, resultType, left, right)
+        generateArithmeticOperator("*", nullCheck, resultType, left, right, 
config)
 
       case DIVIDE | DIVIDE_INTEGER if isNumeric(resultType) =>
         val left = operands.head
         val right = operands(1)
         requireNumeric(left)
         requireNumeric(right)
-        generateArithmeticOperator("/", nullCheck, resultType, left, right)
+        generateArithmeticOperator("/", nullCheck, resultType, left, right, 
config)
 
       case MOD if isNumeric(resultType) =>
         val left = operands.head
         val right = operands(1)
         requireNumeric(left)
         requireNumeric(right)
-        generateArithmeticOperator("%", nullCheck, resultType, left, right)
+        generateArithmeticOperator("%", nullCheck, resultType, left, right, 
config)
 
       case UNARY_MINUS if isNumeric(resultType) =>
         val operand = operands.head
@@ -922,7 +922,7 @@ abstract class CodeGenerator(
         val left = operands.head
         val right = operands(1)
         requireString(left)
-        generateArithmeticOperator("+", nullCheck, resultType, left, right)
+        generateArithmeticOperator("+", nullCheck, resultType, left, right, 
config)
 
       // rows
       case ROW =>

http://git-wip-us.apache.org/repos/asf/flink/blob/347ec384/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/calls/ScalarOperators.scala
----------------------------------------------------------------------
diff --git 
a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/calls/ScalarOperators.scala
 
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/calls/ScalarOperators.scala
index a261b3d..57f1618 100644
--- 
a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/calls/ScalarOperators.scala
+++ 
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/calls/ScalarOperators.scala
@@ -17,12 +17,15 @@
  */
 package org.apache.flink.table.codegen.calls
 
+import java.math.MathContext
+
 import org.apache.calcite.avatica.util.DateTimeUtils.MILLIS_PER_DAY
 import org.apache.calcite.avatica.util.{DateTimeUtils, TimeUnitRange}
 import org.apache.calcite.util.BuiltInMethod
 import org.apache.flink.api.common.typeinfo.BasicTypeInfo._
 import org.apache.flink.api.common.typeinfo._
 import org.apache.flink.api.java.typeutils.{MapTypeInfo, ObjectArrayTypeInfo, 
RowTypeInfo}
+import org.apache.flink.table.api.TableConfig
 import org.apache.flink.table.codegen.CodeGenUtils._
 import 
org.apache.flink.table.codegen.calls.CallGenerator.generateCallIfArgsNotNull
 import org.apache.flink.table.codegen.{CodeGenException, CodeGenerator, 
GeneratedExpression}
@@ -46,7 +49,8 @@ object ScalarOperators {
       nullCheck: Boolean,
       resultType: TypeInformation[_],
       left: GeneratedExpression,
-      right: GeneratedExpression): GeneratedExpression = {
+      right: GeneratedExpression,
+      config: TableConfig): GeneratedExpression = {
 
     val leftCasting = operator match {
       case "%" =>
@@ -68,7 +72,15 @@ object ScalarOperators {
     generateOperatorIfNotNull(nullCheck, resultType, left, right) {
       (leftTerm, rightTerm) =>
         if (isDecimal(resultType)) {
-          
s"${leftCasting(leftTerm)}.${arithOpToDecMethod(operator)}(${rightCasting(rightTerm)})"
+          val decMethod = arithOpToDecMethod(operator)
+          operator match {
+            // include math context for decimal division
+            case "/" =>
+              val mathContext = mathContextToString(config.getDecimalContext)
+              
s"${leftCasting(leftTerm)}.$decMethod(${rightCasting(rightTerm)}, $mathContext)"
+            case _ =>
+              
s"${leftCasting(leftTerm)}.$decMethod(${rightCasting(rightTerm)})"
+          }
         } else {
           s"($resultTypeTerm) (${leftCasting(leftTerm)} $operator 
${rightCasting(rightTerm)})"
         }
@@ -814,14 +826,15 @@ object ScalarOperators {
       plus: Boolean,
       nullCheck: Boolean,
       left: GeneratedExpression,
-      right: GeneratedExpression)
+      right: GeneratedExpression,
+      config: TableConfig)
     : GeneratedExpression = {
 
     val op = if (plus) "+" else "-"
 
     (left.resultType, right.resultType) match {
       case (l: TimeIntervalTypeInfo[_], r: TimeIntervalTypeInfo[_]) if l == r 
=>
-        generateArithmeticOperator(op, nullCheck, l, left, right)
+        generateArithmeticOperator(op, nullCheck, l, left, right, config)
 
       case (SqlTimeTypeInfo.DATE, TimeIntervalTypeInfo.INTERVAL_MILLIS) =>
         generateOperatorIfNotNull(nullCheck, SqlTimeTypeInfo.DATE, left, 
right) {
@@ -1290,6 +1303,14 @@ object ScalarOperators {
     case _ => throw new CodeGenException(s"Unsupported decimal arithmetic 
operator: '$operator'")
   }
 
+  private def mathContextToString(mathContext: MathContext): String = 
mathContext match {
+    case MathContext.DECIMAL32 => "java.math.MathContext.DECIMAL32"
+    case MathContext.DECIMAL64 => "java.math.MathContext.DECIMAL64"
+    case MathContext.DECIMAL128 => "java.math.MathContext.DECIMAL128"
+    case MathContext.UNLIMITED => "java.math.MathContext.UNLIMITED"
+    case _ => s"""new java.math.MathContext("$mathContext")"""
+  }
+
   private def numericCasting(
       operandType: TypeInformation[_],
       resultType: TypeInformation[_])

http://git-wip-us.apache.org/repos/asf/flink/blob/347ec384/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/aggfunctions/AvgAggFunction.scala
----------------------------------------------------------------------
diff --git 
a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/aggfunctions/AvgAggFunction.scala
 
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/aggfunctions/AvgAggFunction.scala
index b651c42..26621b7 100644
--- 
a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/aggfunctions/AvgAggFunction.scala
+++ 
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/aggfunctions/AvgAggFunction.scala
@@ -17,7 +17,7 @@
  */
 package org.apache.flink.table.functions.aggfunctions
 
-import java.math.{BigDecimal, BigInteger}
+import java.math.{BigDecimal, BigInteger, MathContext}
 import java.lang.{Iterable => JIterable}
 
 import org.apache.flink.api.common.typeinfo.{BasicTypeInfo, TypeInformation}
@@ -295,7 +295,8 @@ class DecimalAvgAccumulator extends JTuple2[BigDecimal, 
Long] {
 /**
   * Base class for built-in Big Decimal Avg aggregate function
   */
-class DecimalAvgAggFunction extends AggregateFunction[BigDecimal, 
DecimalAvgAccumulator] {
+class DecimalAvgAggFunction(context: MathContext)
+  extends AggregateFunction[BigDecimal, DecimalAvgAccumulator] {
 
   override def createAccumulator(): DecimalAvgAccumulator = {
     new DecimalAvgAccumulator
@@ -321,7 +322,7 @@ class DecimalAvgAggFunction extends 
AggregateFunction[BigDecimal, DecimalAvgAccu
     if (acc.f1 == 0) {
       null.asInstanceOf[BigDecimal]
     } else {
-      acc.f0.divide(BigDecimal.valueOf(acc.f1))
+      acc.f0.divide(BigDecimal.valueOf(acc.f1), context)
     }
   }
 

http://git-wip-us.apache.org/repos/asf/flink/blob/347ec384/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/dataset/DataSetAggregate.scala
----------------------------------------------------------------------
diff --git 
a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/dataset/DataSetAggregate.scala
 
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/dataset/DataSetAggregate.scala
index 7dd307b..07dcf79 100644
--- 
a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/dataset/DataSetAggregate.scala
+++ 
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/dataset/DataSetAggregate.scala
@@ -110,7 +110,8 @@ class DataSetAggregate(
         input.getRowType,
         inputDS.getType.asInstanceOf[RowTypeInfo].getFieldTypes,
         rowRelDataType,
-        grouping)
+        grouping,
+        tableEnv.getConfig)
 
     val aggString = aggregationToString(inputType, grouping, getRowType, 
namedAggregates, Nil)
 

http://git-wip-us.apache.org/repos/asf/flink/blob/347ec384/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/dataset/DataSetWindowAggregate.scala
----------------------------------------------------------------------
diff --git 
a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/dataset/DataSetWindowAggregate.scala
 
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/dataset/DataSetWindowAggregate.scala
index 745c4ed..53748f5 100644
--- 
a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/dataset/DataSetWindowAggregate.scala
+++ 
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/dataset/DataSetWindowAggregate.scala
@@ -25,7 +25,7 @@ import org.apache.calcite.rel.{RelNode, RelWriter, SingleRel}
 import org.apache.flink.api.common.operators.Order
 import org.apache.flink.api.java.DataSet
 import org.apache.flink.api.java.typeutils.{ResultTypeQueryable, RowTypeInfo}
-import org.apache.flink.table.api.{BatchQueryConfig, BatchTableEnvironment}
+import org.apache.flink.table.api.{BatchQueryConfig, BatchTableEnvironment, 
TableConfig}
 import org.apache.flink.table.calcite.FlinkRelBuilder.NamedWindowProperty
 import org.apache.flink.table.calcite.FlinkTypeFactory
 import org.apache.flink.table.codegen.AggregationCodeGenerator
@@ -127,11 +127,12 @@ class DataSetWindowAggregate(
           generator,
           inputDS,
           isTimeIntervalLiteral(size),
-          caseSensitive)
+          caseSensitive,
+          tableEnv.getConfig)
 
       case SessionGroupWindow(_, timeField, gap)
           if isTimePoint(timeField.resultType) || isLong(timeField.resultType) 
=>
-        createEventTimeSessionWindowDataSet(generator, inputDS, caseSensitive)
+        createEventTimeSessionWindowDataSet(generator, inputDS, caseSensitive, 
tableEnv.getConfig)
 
       case SlidingGroupWindow(_, timeField, size, slide)
           if isTimePoint(timeField.resultType) || isLong(timeField.resultType) 
=>
@@ -141,7 +142,8 @@ class DataSetWindowAggregate(
           isTimeIntervalLiteral(size),
           asLong(size),
           asLong(slide),
-          caseSensitive)
+          caseSensitive,
+          tableEnv.getConfig)
 
       case _ =>
         throw new UnsupportedOperationException(
@@ -153,7 +155,8 @@ class DataSetWindowAggregate(
       generator: AggregationCodeGenerator,
       inputDS: DataSet[Row],
       isTimeWindow: Boolean,
-      isParserCaseSensitive: Boolean): DataSet[Row] = {
+      isParserCaseSensitive: Boolean,
+      tableConfig: TableConfig): DataSet[Row] = {
 
     val input = inputNode.asInstanceOf[DataSetRel]
 
@@ -164,7 +167,8 @@ class DataSetWindowAggregate(
       grouping,
       input.getRowType,
       inputDS.getType.asInstanceOf[RowTypeInfo].getFieldTypes,
-      isParserCaseSensitive)
+      isParserCaseSensitive,
+      tableConfig)
     val groupReduceFunction = 
createDataSetWindowAggregationGroupReduceFunction(
       generator,
       window,
@@ -173,7 +177,8 @@ class DataSetWindowAggregate(
       inputDS.getType.asInstanceOf[RowTypeInfo].getFieldTypes,
       getRowType,
       grouping,
-      namedProperties)
+      namedProperties,
+      tableConfig)
 
     val mappedInput = inputDS
       .map(mapFunction)
@@ -215,7 +220,8 @@ class DataSetWindowAggregate(
   private[this] def createEventTimeSessionWindowDataSet(
       generator: AggregationCodeGenerator,
       inputDS: DataSet[Row],
-      isParserCaseSensitive: Boolean): DataSet[Row] = {
+      isParserCaseSensitive: Boolean,
+      tableConfig: TableConfig): DataSet[Row] = {
 
     val input = inputNode.asInstanceOf[DataSetRel]
 
@@ -230,7 +236,8 @@ class DataSetWindowAggregate(
       grouping,
       input.getRowType,
       inputDS.getType.asInstanceOf[RowTypeInfo].getFieldTypes,
-      isParserCaseSensitive)
+      isParserCaseSensitive,
+      tableConfig)
 
     val mappedInput = inputDS.map(mapFunction).name(prepareOperatorName)
 
@@ -243,7 +250,8 @@ class DataSetWindowAggregate(
     if (doAllSupportPartialMerge(
       namedAggregates.map(_.getKey),
       inputType,
-      grouping.length)) {
+      grouping.length,
+      tableConfig)) {
 
       // gets the window-start and window-end position  in the intermediate 
result.
       val windowStartPos = rowTimeFieldPos
@@ -257,7 +265,8 @@ class DataSetWindowAggregate(
           namedAggregates,
           input.getRowType,
           inputDS.getType.asInstanceOf[RowTypeInfo].getFieldTypes,
-          grouping)
+          grouping,
+          tableConfig)
 
         // create groupReduceFunction for calculating the aggregations
         val groupReduceFunction = 
createDataSetWindowAggregationGroupReduceFunction(
@@ -269,6 +278,7 @@ class DataSetWindowAggregate(
           rowRelDataType,
           grouping,
           namedProperties,
+          tableConfig,
           isInputCombined = true)
 
         mappedInput
@@ -289,7 +299,8 @@ class DataSetWindowAggregate(
           namedAggregates,
           input.getRowType,
           inputDS.getType.asInstanceOf[RowTypeInfo].getFieldTypes,
-          grouping)
+          grouping,
+          tableConfig)
 
         // create groupReduceFunction for calculating the aggregations
         val groupReduceFunction = 
createDataSetWindowAggregationGroupReduceFunction(
@@ -301,6 +312,7 @@ class DataSetWindowAggregate(
           rowRelDataType,
           grouping,
           namedProperties,
+          tableConfig,
           isInputCombined = true)
 
         mappedInput.sortPartition(rowTimeFieldPos, Order.ASCENDING)
@@ -326,7 +338,8 @@ class DataSetWindowAggregate(
           inputDS.getType.asInstanceOf[RowTypeInfo].getFieldTypes,
           rowRelDataType,
           grouping,
-          namedProperties)
+          namedProperties,
+          tableConfig)
 
         mappedInput.groupBy(groupingKeys: _*)
           .sortGroup(rowTimeFieldPos, Order.ASCENDING)
@@ -343,7 +356,8 @@ class DataSetWindowAggregate(
           inputDS.getType.asInstanceOf[RowTypeInfo].getFieldTypes,
           rowRelDataType,
           grouping,
-          namedProperties)
+          namedProperties,
+          tableConfig)
 
         mappedInput.sortPartition(rowTimeFieldPos, 
Order.ASCENDING).setParallelism(1)
           .reduceGroup(groupReduceFunction)
@@ -360,7 +374,8 @@ class DataSetWindowAggregate(
       isTimeWindow: Boolean,
       size: Long,
       slide: Long,
-      isParserCaseSensitive: Boolean)
+      isParserCaseSensitive: Boolean,
+      tableConfig: TableConfig)
     : DataSet[Row] = {
 
     val input = inputNode.asInstanceOf[DataSetRel]
@@ -374,7 +389,8 @@ class DataSetWindowAggregate(
       grouping,
       input.getRowType,
       inputDS.getType.asInstanceOf[RowTypeInfo].getFieldTypes,
-      isParserCaseSensitive)
+      isParserCaseSensitive,
+      tableConfig)
 
     val mappedDataSet = inputDS
       .map(mapFunction)
@@ -389,7 +405,8 @@ class DataSetWindowAggregate(
     val isPartial = doAllSupportPartialMerge(
       namedAggregates.map(_.getKey),
       inputType,
-      grouping.length)
+      grouping.length,
+      tableConfig)
 
     // only pre-tumble if it is worth it
     val isLittleTumblingSize = determineLargestTumblingSize(size, slide) <= 1
@@ -411,7 +428,8 @@ class DataSetWindowAggregate(
           grouping,
           input.getRowType,
           inputDS.getType.asInstanceOf[RowTypeInfo].getFieldTypes,
-          isParserCaseSensitive)
+          isParserCaseSensitive,
+          tableConfig)
 
         mappedDataSet.asInstanceOf[DataSet[Row]]
           .groupBy(groupingKeysAndAlignedRowtime: _*)
@@ -451,6 +469,7 @@ class DataSetWindowAggregate(
       rowRelDataType,
       grouping,
       namedProperties,
+      tableConfig,
       isInputCombined = false)
 
     // gets the window-start position in the intermediate result.

http://git-wip-us.apache.org/repos/asf/flink/blob/347ec384/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamGroupAggregate.scala
----------------------------------------------------------------------
diff --git 
a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamGroupAggregate.scala
 
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamGroupAggregate.scala
index 71de57c..5f4b186 100644
--- 
a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamGroupAggregate.scala
+++ 
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamGroupAggregate.scala
@@ -138,6 +138,7 @@ class DataStreamGroupAggregate(
       inputSchema.fieldTypeInfos,
       groupings,
       queryConfig,
+      tableEnv.getConfig,
       DataStreamRetractionRules.isAccRetract(this),
       DataStreamRetractionRules.isAccRetract(getInput))
 

http://git-wip-us.apache.org/repos/asf/flink/blob/347ec384/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamGroupWindowAggregate.scala
----------------------------------------------------------------------
diff --git 
a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamGroupWindowAggregate.scala
 
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamGroupWindowAggregate.scala
index d527dc8..0a014b6 100644
--- 
a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamGroupWindowAggregate.scala
+++ 
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamGroupWindowAggregate.scala
@@ -207,7 +207,8 @@ class DataStreamGroupWindowAggregate(
           inputSchema.fieldTypeInfos,
           schema.relDataType,
           grouping,
-          needMerge)
+          needMerge,
+          tableEnv.getConfig)
 
       windowedStream
         .aggregate(aggFunction, windowFunction, accumulatorRowType, 
aggResultRowType, outRowType)
@@ -232,7 +233,8 @@ class DataStreamGroupWindowAggregate(
           inputSchema.fieldTypeInfos,
           schema.relDataType,
           Array[Int](),
-          needMerge)
+          needMerge,
+          tableEnv.getConfig)
 
       windowedStream
         .aggregate(aggFunction, windowFunction, accumulatorRowType, 
aggResultRowType, outRowType)

http://git-wip-us.apache.org/repos/asf/flink/blob/347ec384/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamOverAggregate.scala
----------------------------------------------------------------------
diff --git 
a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamOverAggregate.scala
 
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamOverAggregate.scala
index 635c7bc..c1693d9 100644
--- 
a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamOverAggregate.scala
+++ 
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamOverAggregate.scala
@@ -28,7 +28,7 @@ import org.apache.calcite.rel.{RelNode, RelWriter, SingleRel}
 import org.apache.calcite.rex.RexLiteral
 import org.apache.flink.api.java.functions.NullByteKeySelector
 import org.apache.flink.streaming.api.datastream.DataStream
-import org.apache.flink.table.api.{StreamQueryConfig, StreamTableEnvironment, 
TableException}
+import org.apache.flink.table.api.{StreamQueryConfig, StreamTableEnvironment, 
TableConfig, TableException}
 import org.apache.flink.table.calcite.FlinkTypeFactory
 import org.apache.flink.table.codegen.AggregationCodeGenerator
 import org.apache.flink.table.plan.nodes.OverAggregate
@@ -172,6 +172,7 @@ class DataStreamOverAggregate(
       // unbounded OVER window
       createUnboundedAndCurrentRowOverWindow(
         queryConfig,
+        tableEnv.getConfig,
         generator,
         inputDS,
         rowTimeIdx,
@@ -188,7 +189,8 @@ class DataStreamOverAggregate(
         inputDS,
         rowTimeIdx,
         aggregateInputType,
-        isRowsClause = overWindow.isRows)
+        isRowsClause = overWindow.isRows,
+        tableEnv.getConfig)
     } else {
       throw new TableException("OVER RANGE FOLLOWING windows are not supported 
yet.")
     }
@@ -196,6 +198,7 @@ class DataStreamOverAggregate(
 
   def createUnboundedAndCurrentRowOverWindow(
     queryConfig: StreamQueryConfig,
+    tableConfig: TableConfig,
     generator: AggregationCodeGenerator,
     inputDS: DataStream[CRow],
     rowTimeIdx: Option[Int],
@@ -219,6 +222,7 @@ class DataStreamOverAggregate(
       inputSchema.typeInfo,
       inputSchema.fieldTypeInfos,
       queryConfig,
+      tableConfig,
       rowTimeIdx,
       partitionKeys.nonEmpty,
       isRowsClause)
@@ -249,7 +253,8 @@ class DataStreamOverAggregate(
     inputDS: DataStream[CRow],
     rowTimeIdx: Option[Int],
     aggregateInputType: RelDataType,
-    isRowsClause: Boolean): DataStream[CRow] = {
+    isRowsClause: Boolean,
+    tableConfig: TableConfig): DataStream[CRow] = {
 
     val overWindow: Group = logicWindow.groups.get(0)
 
@@ -272,6 +277,7 @@ class DataStreamOverAggregate(
       inputSchema.fieldTypeInfos,
       precedingOffset,
       queryConfig,
+      tableConfig,
       isRowsClause,
       rowTimeIdx
     )

http://git-wip-us.apache.org/repos/asf/flink/blob/347ec384/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/AggregateUtil.scala
----------------------------------------------------------------------
diff --git 
a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/AggregateUtil.scala
 
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/AggregateUtil.scala
index 361a87e..df9b1c5 100644
--- 
a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/AggregateUtil.scala
+++ 
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/AggregateUtil.scala
@@ -32,7 +32,7 @@ import 
org.apache.flink.streaming.api.functions.ProcessFunction
 import org.apache.flink.streaming.api.functions.windowing.{AllWindowFunction, 
WindowFunction}
 import org.apache.flink.streaming.api.windowing.windows.{Window => 
DataStreamWindow}
 import org.apache.flink.table.api.dataview.DataViewSpec
-import org.apache.flink.table.api.{StreamQueryConfig, TableException}
+import org.apache.flink.table.api.{StreamQueryConfig, TableConfig, 
TableException}
 import org.apache.flink.table.calcite.FlinkRelBuilder.NamedWindowProperty
 import org.apache.flink.table.calcite.FlinkTypeFactory
 import org.apache.flink.table.codegen.AggregationCodeGenerator
@@ -78,6 +78,7 @@ object AggregateUtil {
       inputTypeInfo: TypeInformation[Row],
       inputFieldTypeInfo: Seq[TypeInformation[_]],
       queryConfig: StreamQueryConfig,
+      tableConfig: TableConfig,
       rowTimeIdx: Option[Int],
       isPartitioned: Boolean,
       isRowsClause: Boolean)
@@ -88,6 +89,7 @@ object AggregateUtil {
         namedAggregates.map(_.getKey),
         aggregateInputType,
         needRetraction = false,
+        tableConfig,
         isStateBackedDataViews = true)
 
     val aggregationStateType: RowTypeInfo = new RowTypeInfo(accTypes: _*)
@@ -159,6 +161,7 @@ object AggregateUtil {
       inputFieldTypes: Seq[TypeInformation[_]],
       groupings: Array[Int],
       queryConfig: StreamQueryConfig,
+      tableConfig: TableConfig,
       generateRetraction: Boolean,
       consumeRetraction: Boolean): ProcessFunction[CRow, CRow] = {
 
@@ -167,6 +170,7 @@ object AggregateUtil {
         namedAggregates.map(_.getKey),
         inputRowType,
         consumeRetraction,
+        tableConfig,
         isStateBackedDataViews = true)
 
     val aggMapping = aggregates.indices.map(_ + groupings.length).toArray
@@ -223,6 +227,7 @@ object AggregateUtil {
       inputFieldTypeInfo: Seq[TypeInformation[_]],
       precedingOffset: Long,
       queryConfig: StreamQueryConfig,
+      tableConfig: TableConfig,
       isRowsClause: Boolean,
       rowTimeIdx: Option[Int])
     : ProcessFunction[CRow, CRow] = {
@@ -233,6 +238,7 @@ object AggregateUtil {
         namedAggregates.map(_.getKey),
         aggregateInputType,
         needRetract,
+        tableConfig,
         isStateBackedDataViews = true)
 
     val aggregationStateType: RowTypeInfo = new RowTypeInfo(accTypes: _*)
@@ -325,14 +331,16 @@ object AggregateUtil {
     groupings: Array[Int],
     inputType: RelDataType,
     inputFieldTypeInfo: Seq[TypeInformation[_]],
-    isParserCaseSensitive: Boolean)
+    isParserCaseSensitive: Boolean,
+    tableConfig: TableConfig)
   : MapFunction[Row, Row] = {
 
     val needRetract = false
     val (aggFieldIndexes, aggregates, accTypes, _) = 
transformToAggregateFunctions(
       namedAggregates.map(_.getKey),
       inputType,
-      needRetract)
+      needRetract,
+      tableConfig)
 
     val mapReturnType: RowTypeInfo =
       createRowTypeForKeysAndAggregates(
@@ -430,14 +438,16 @@ object AggregateUtil {
       groupings: Array[Int],
       physicalInputRowType: RelDataType,
       physicalInputTypes: Seq[TypeInformation[_]],
-      isParserCaseSensitive: Boolean)
+      isParserCaseSensitive: Boolean,
+      tableConfig: TableConfig)
     : RichGroupReduceFunction[Row, Row] = {
 
     val needRetract = false
     val (aggFieldIndexes, aggregates, accTypes, _) = 
transformToAggregateFunctions(
       namedAggregates.map(_.getKey),
       physicalInputRowType,
-      needRetract)
+      needRetract,
+      tableConfig)
 
     val returnType: RowTypeInfo = createRowTypeForKeysAndAggregates(
       groupings,
@@ -543,6 +553,7 @@ object AggregateUtil {
       outputType: RelDataType,
       groupings: Array[Int],
       properties: Seq[NamedWindowProperty],
+      tableConfig: TableConfig,
       isInputCombined: Boolean = false)
     : RichGroupReduceFunction[Row, Row] = {
 
@@ -550,7 +561,8 @@ object AggregateUtil {
     val (aggFieldIndexes, aggregates, _, _) = transformToAggregateFunctions(
       namedAggregates.map(_.getKey),
       physicalInputRowType,
-      needRetract)
+      needRetract,
+      tableConfig)
 
     val aggMapping = aggregates.indices.toArray.map(_ + groupings.length)
 
@@ -695,13 +707,15 @@ object AggregateUtil {
     namedAggregates: Seq[CalcitePair[AggregateCall, String]],
     physicalInputRowType: RelDataType,
     physicalInputTypes: Seq[TypeInformation[_]],
-    groupings: Array[Int]): MapPartitionFunction[Row, Row] = {
+    groupings: Array[Int],
+    tableConfig: TableConfig): MapPartitionFunction[Row, Row] = {
 
     val needRetract = false
     val (aggFieldIndexes, aggregates, accTypes, _) = 
transformToAggregateFunctions(
       namedAggregates.map(_.getKey),
       physicalInputRowType,
-      needRetract)
+      needRetract,
+      tableConfig)
 
     val aggMapping = aggregates.indices.map(_ + groupings.length).toArray
 
@@ -767,14 +781,16 @@ object AggregateUtil {
       namedAggregates: Seq[CalcitePair[AggregateCall, String]],
       physicalInputRowType: RelDataType,
       physicalInputTypes: Seq[TypeInformation[_]],
-      groupings: Array[Int])
+      groupings: Array[Int],
+      tableConfig: TableConfig)
     : GroupCombineFunction[Row, Row] = {
 
     val needRetract = false
     val (aggFieldIndexes, aggregates, accTypes, _) = 
transformToAggregateFunctions(
       namedAggregates.map(_.getKey),
       physicalInputRowType,
-      needRetract)
+      needRetract,
+      tableConfig)
 
     val aggMapping = aggregates.indices.map(_ + groupings.length).toArray
 
@@ -831,7 +847,8 @@ object AggregateUtil {
       inputType: RelDataType,
       inputFieldTypeInfo: Seq[TypeInformation[_]],
       outputType: RelDataType,
-      groupings: Array[Int]): (
+      groupings: Array[Int],
+      tableConfig: TableConfig): (
         Option[DataSetPreAggFunction],
         Option[TypeInformation[Row]],
         Either[DataSetAggFunction, DataSetFinalAggFunction]) = {
@@ -840,7 +857,8 @@ object AggregateUtil {
     val (aggInFields, aggregates, accTypes, _) = transformToAggregateFunctions(
       namedAggregates.map(_.getKey),
       inputType,
-      needRetract)
+      needRetract,
+      tableConfig)
 
     val (gkeyOutMapping, aggOutMapping) = getOutputMappings(
       namedAggregates,
@@ -992,7 +1010,8 @@ object AggregateUtil {
       inputFieldTypeInfo: Seq[TypeInformation[_]],
       outputType: RelDataType,
       groupingKeys: Array[Int],
-      needMerge: Boolean)
+      needMerge: Boolean,
+      tableConfig: TableConfig)
     : (DataStreamAggFunction[CRow, Row, Row], RowTypeInfo, RowTypeInfo) = {
 
     val needRetract = false
@@ -1000,7 +1019,8 @@ object AggregateUtil {
       transformToAggregateFunctions(
         namedAggregates.map(_.getKey),
         inputType,
-        needRetract)
+        needRetract,
+        tableConfig)
 
     val aggMapping = aggregates.indices.toArray
     val outputArity = aggregates.length
@@ -1036,12 +1056,14 @@ object AggregateUtil {
   private[flink] def doAllSupportPartialMerge(
     aggregateCalls: Seq[AggregateCall],
     inputType: RelDataType,
-    groupKeysCount: Int): Boolean = {
+    groupKeysCount: Int,
+    tableConfig: TableConfig): Boolean = {
 
     val aggregateList = transformToAggregateFunctions(
       aggregateCalls,
       inputType,
-      needRetraction = false)._2
+      needRetraction = false,
+      tableConfig)._2
 
     doAllSupportPartialMerge(aggregateList)
   }
@@ -1121,6 +1143,7 @@ object AggregateUtil {
       aggregateCalls: Seq[AggregateCall],
       aggregateInputType: RelDataType,
       needRetraction: Boolean,
+      tableConfig: TableConfig,
       isStateBackedDataViews: Boolean = false)
   : (Array[Array[Int]],
     Array[TableAggregateFunction[_, _]],
@@ -1251,7 +1274,7 @@ object AggregateUtil {
               case DOUBLE =>
                 new DoubleAvgAggFunction
               case DECIMAL =>
-                new DecimalAvgAggFunction
+                new DecimalAvgAggFunction(tableConfig.getDecimalContext)
               case sqlType: SqlTypeName =>
                 throw new TableException(s"Avg aggregate does no support type: 
'$sqlType'")
             }

http://git-wip-us.apache.org/repos/asf/flink/blob/347ec384/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/DecimalTypeTest.scala
----------------------------------------------------------------------
diff --git 
a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/DecimalTypeTest.scala
 
b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/DecimalTypeTest.scala
index 42f8008..5de6f2c 100644
--- 
a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/DecimalTypeTest.scala
+++ 
b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/DecimalTypeTest.scala
@@ -233,6 +233,13 @@ class DecimalTypeTest extends ExpressionTestBase {
       "-f0",
       "-f0",
       "-123456789.123456789123456789")
+
+    testAllApis(
+      BigDecimal("1").toExpr / BigDecimal("3"),
+      "1p / 3p",
+      "CAST('1' AS DECIMAL) / CAST('3' AS DECIMAL)",
+      "0.3333333333333333333333333333333333"
+    )
   }
 
   @Test
@@ -287,7 +294,7 @@ class DecimalTypeTest extends ExpressionTestBase {
 
   // 
----------------------------------------------------------------------------------------------
 
-  def testData = {
+  def testData: Row = {
     val testData = new Row(6)
     testData.setField(0, BigDecimal("123456789.123456789123456789").bigDecimal)
     testData.setField(1, BigDecimal("123456789123456789123456789").bigDecimal)
@@ -298,7 +305,7 @@ class DecimalTypeTest extends ExpressionTestBase {
     testData
   }
 
-  def typeInfo = {
+  def typeInfo: TypeInformation[Any] = {
     new RowTypeInfo(
       Types.DECIMAL,
       Types.DECIMAL,

http://git-wip-us.apache.org/repos/asf/flink/blob/347ec384/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/aggfunctions/AvgFunctionTest.scala
----------------------------------------------------------------------
diff --git 
a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/aggfunctions/AvgFunctionTest.scala
 
b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/aggfunctions/AvgFunctionTest.scala
index 0671b40..d413c6c 100644
--- 
a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/aggfunctions/AvgFunctionTest.scala
+++ 
b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/aggfunctions/AvgFunctionTest.scala
@@ -18,7 +18,7 @@
 
 package org.apache.flink.table.runtime.aggfunctions
 
-import java.math.BigDecimal
+import java.math.{BigDecimal, MathContext}
 
 import org.apache.flink.table.functions.AggregateFunction
 import org.apache.flink.table.functions.aggfunctions._
@@ -178,17 +178,23 @@ class DecimalAvgAggFunctionTest extends 
AggFunctionTestBase[BigDecimal, DecimalA
       null,
       null,
       null
+    ),
+    Seq(
+      new BigDecimal("0.3"),
+      new BigDecimal("0.3"),
+      new BigDecimal("0.4")
     )
   )
 
   override def expectedResults: Seq[BigDecimal] = Seq(
     BigDecimal.ZERO,
     BigDecimal.ONE,
-    null
+    null,
+    BigDecimal.ONE.divide(new BigDecimal("3"), MathContext.DECIMAL128)
   )
 
   override def aggregator: AggregateFunction[BigDecimal, 
DecimalAvgAccumulator] =
-    new DecimalAvgAggFunction()
+    new DecimalAvgAggFunction(MathContext.DECIMAL128)
 
   override def retractFunc = aggregator.getClass.getMethod("retract", accType, 
classOf[Any])
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/347ec384/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/batch/table/CalcITCase.scala
----------------------------------------------------------------------
diff --git 
a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/batch/table/CalcITCase.scala
 
b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/batch/table/CalcITCase.scala
index 1b89229..aa37d1b 100644
--- 
a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/batch/table/CalcITCase.scala
+++ 
b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/batch/table/CalcITCase.scala
@@ -18,6 +18,7 @@
 
 package org.apache.flink.table.runtime.batch.table
 
+import java.math.MathContext
 import java.sql.{Date, Time, Timestamp}
 import java.util
 
@@ -41,6 +42,7 @@ import org.junit.runners.Parameterized
 
 import scala.collection.JavaConverters._
 import scala.collection.mutable
+import scala.math.BigDecimal.RoundingMode
 
 @RunWith(classOf[Parameterized])
 class CalcITCase(
@@ -330,6 +332,7 @@ class CalcITCase(
   def testAdvancedDataTypes(): Unit = {
     val env = ExecutionEnvironment.getExecutionEnvironment
     val tEnv = TableEnvironment.getTableEnvironment(env, config)
+    tEnv.getConfig.setDecimalContext(new MathContext(30))
 
     val t = env
       .fromElements((
@@ -341,10 +344,11 @@ class CalcITCase(
       .toTable(tEnv, 'a, 'b, 'c, 'd, 'e)
       .select('a, 'b, 'c, 'd, 'e, BigDecimal("11.2"), 
BigDecimal("11.2").bigDecimal,
         Date.valueOf("1984-07-12"), Time.valueOf("14:34:24"),
-        Timestamp.valueOf("1984-07-12 14:34:24"))
+        Timestamp.valueOf("1984-07-12 14:34:24"),
+        BigDecimal("1").toExpr / BigDecimal("3"))
 
     val expected = "78.454654654654654,4E+9999,1984-07-12,14:34:24,1984-07-12 
14:34:24.0," +
-      "11.2,11.2,1984-07-12,14:34:24,1984-07-12 14:34:24.0"
+      "11.2,11.2,1984-07-12,14:34:24,1984-07-12 
14:34:24.0,0.333333333333333333333333333333"
     val results = t.toDataSet[Row].collect()
     TestBaseUtils.compareResultAsText(results.asJava, expected)
   }

Reply via email to