flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From fhue...@apache.org
Subject [15/37] flink git commit: [FLINK-6124] [table] Add min/max string aggregation with retracion
Date Thu, 06 Apr 2017 07:28:28 GMT
[FLINK-6124] [table] Add min/max string aggregation with retracion

This closes #3593.


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

Branch: refs/heads/table-retraction
Commit: c5282cbcf898b99593ca753fce7557bae1ae09aa
Parents: 05ceec0
Author: twalthr <twalthr@apache.org>
Authored: Wed Mar 22 10:03:56 2017 +0100
Committer: twalthr <twalthr@apache.org>
Committed: Mon Apr 3 18:21:53 2017 +0200

----------------------------------------------------------------------
 .../functions/aggfunctions/MaxAggFunction.scala |  2 +-
 .../MaxAggFunctionWithRetract.scala             | 12 ++++++--
 .../functions/aggfunctions/MinAggFunction.scala |  2 +-
 .../MinAggFunctionWithRetract.scala             |  8 +++++
 .../table/runtime/aggregate/AggregateUtil.scala |  4 +++
 .../aggfunctions/AggFunctionTestBase.scala      |  4 +--
 .../MaxWithRetractAggFunctionTest.scala         | 32 ++++++++++++++++++++
 .../MinWithRetractAggFunctionTest.scala         | 32 ++++++++++++++++++++
 8 files changed, 90 insertions(+), 6 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/c5282cbc/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/aggfunctions/MaxAggFunction.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/aggfunctions/MaxAggFunction.scala
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/aggfunctions/MaxAggFunction.scala
index 3793434..2e666fa 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/aggfunctions/MaxAggFunction.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/aggfunctions/MaxAggFunction.scala
@@ -160,6 +160,6 @@ class DecimalMaxAggFunction extends MaxAggFunction[BigDecimal] {
   * Built-in String Max aggregate function
   */
 class StringMaxAggFunction extends MaxAggFunction[String] {
-  override def getInitValue = "".toString
+  override def getInitValue = ""
   override def getValueTypeInfo = BasicTypeInfo.STRING_TYPE_INFO
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/c5282cbc/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/aggfunctions/MaxAggFunctionWithRetract.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/aggfunctions/MaxAggFunctionWithRetract.scala
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/aggfunctions/MaxAggFunctionWithRetract.scala
index eb6e7dc..14ceba2 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/aggfunctions/MaxAggFunctionWithRetract.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/aggfunctions/MaxAggFunctionWithRetract.scala
@@ -201,7 +201,7 @@ class DoubleMaxWithRetractAggFunction extends MaxWithRetractAggFunction[Double]
   * Built-in Boolean Max with retraction aggregate function
   */
 class BooleanMaxWithRetractAggFunction extends MaxWithRetractAggFunction[Boolean] {
-  override def getInitValue = false
+  override def getInitValue: Boolean = false
   override def getValueTypeInfo = BasicTypeInfo.BOOLEAN_TYPE_INFO
 }
 
@@ -209,6 +209,14 @@ class BooleanMaxWithRetractAggFunction extends MaxWithRetractAggFunction[Boolean
   * Built-in Big Decimal Max with retraction aggregate function
   */
 class DecimalMaxWithRetractAggFunction extends MaxWithRetractAggFunction[BigDecimal] {
-  override def getInitValue = BigDecimal.ZERO
+  override def getInitValue: BigDecimal = BigDecimal.ZERO
   override def getValueTypeInfo = BasicTypeInfo.BIG_DEC_TYPE_INFO
 }
+
+/**
+  * Built-in String Max with retraction aggregate function
+  */
+class StringMaxWithRetractAggFunction extends MaxWithRetractAggFunction[String] {
+  override def getInitValue: String = ""
+  override def getValueTypeInfo = BasicTypeInfo.STRING_TYPE_INFO
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/c5282cbc/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/aggfunctions/MinAggFunction.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/aggfunctions/MinAggFunction.scala
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/aggfunctions/MinAggFunction.scala
index 41361fd..75a8ebc 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/aggfunctions/MinAggFunction.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/aggfunctions/MinAggFunction.scala
@@ -160,6 +160,6 @@ class DecimalMinAggFunction extends MinAggFunction[BigDecimal] {
   * Built-in String Min aggregate function
   */
 class StringMinAggFunction extends MinAggFunction[String] {
-  override def getInitValue = "".toString
+  override def getInitValue = ""
   override def getValueTypeInfo = BasicTypeInfo.STRING_TYPE_INFO
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/c5282cbc/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/aggfunctions/MinAggFunctionWithRetract.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/aggfunctions/MinAggFunctionWithRetract.scala
b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/aggfunctions/MinAggFunctionWithRetract.scala
index c953286..6f2c3a1 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/aggfunctions/MinAggFunctionWithRetract.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/aggfunctions/MinAggFunctionWithRetract.scala
@@ -212,3 +212,11 @@ class DecimalMinWithRetractAggFunction extends MinWithRetractAggFunction[BigDeci
   override def getInitValue: BigDecimal = BigDecimal.ZERO
   override def getValueTypeInfo = BasicTypeInfo.BIG_DEC_TYPE_INFO
 }
+
+/**
+  * Built-in String Min with retraction aggregate function
+  */
+class StringMinWithRetractAggFunction extends MinWithRetractAggFunction[String] {
+  override def getInitValue: String = ""
+  override def getValueTypeInfo = BasicTypeInfo.STRING_TYPE_INFO
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/c5282cbc/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 caa2818..40a1f8a 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
@@ -1003,6 +1003,8 @@ object AggregateUtil {
                   new DecimalMinWithRetractAggFunction
                 case BOOLEAN =>
                   new BooleanMinWithRetractAggFunction
+                case VARCHAR | CHAR =>
+                  new StringMinWithRetractAggFunction
                 case sqlType: SqlTypeName =>
                   throw new TableException("Min with retract aggregate does no support type:"
+
                                              sqlType)
@@ -1050,6 +1052,8 @@ object AggregateUtil {
                   new DecimalMaxWithRetractAggFunction
                 case BOOLEAN =>
                   new BooleanMaxWithRetractAggFunction
+                case VARCHAR | CHAR =>
+                  new StringMaxWithRetractAggFunction
                 case sqlType: SqlTypeName =>
                   throw new TableException("Max with retract aggregate does no support type:"
+
                                              sqlType)

http://git-wip-us.apache.org/repos/asf/flink/blob/c5282cbc/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/functions/aggfunctions/AggFunctionTestBase.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/functions/aggfunctions/AggFunctionTestBase.scala
b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/functions/aggfunctions/AggFunctionTestBase.scala
index 80fc947..cb1137f 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/functions/aggfunctions/AggFunctionTestBase.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/functions/aggfunctions/AggFunctionTestBase.scala
@@ -44,7 +44,7 @@ abstract class AggFunctionTestBase[T] {
     // iterate over input sets
     for ((vals, expected) <- inputValueSets.zip(expectedResults)) {
       val accumulator = accumulateVals(vals)
-      var result = aggregator.getValue(accumulator)
+      val result = aggregator.getValue(accumulator)
       validateResult[T](expected, result)
 
       if (supportRetraction) {
@@ -71,7 +71,7 @@ abstract class AggFunctionTestBase[T] {
         accumulators.add(accumulateVals(firstVals))
         accumulators.add(accumulateVals(secondVals))
 
-        var accumulator = aggregator.merge(accumulators)
+        val accumulator = aggregator.merge(accumulators)
         val result = aggregator.getValue(accumulator)
         validateResult[T](expected, result)
 

http://git-wip-us.apache.org/repos/asf/flink/blob/c5282cbc/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/functions/aggfunctions/MaxWithRetractAggFunctionTest.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/functions/aggfunctions/MaxWithRetractAggFunctionTest.scala
b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/functions/aggfunctions/MaxWithRetractAggFunctionTest.scala
index c2329a4..8bd5c6f 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/functions/aggfunctions/MaxWithRetractAggFunctionTest.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/functions/aggfunctions/MaxWithRetractAggFunctionTest.scala
@@ -186,3 +186,35 @@ class DecimalMaxWithRetractAggFunctionTest extends AggFunctionTestBase[BigDecima
 
   override def aggregator: AggregateFunction[BigDecimal] = new DecimalMaxWithRetractAggFunction()
 }
+
+class StringMaxWithRetractAggFunctionTest extends AggFunctionTestBase[String] {
+
+  override def inputValueSets: Seq[Seq[_]] = Seq(
+    Seq(
+      "abc",
+      "def",
+      "ghi",
+      null,
+      "jkl",
+      null,
+      "zzz"
+    ),
+    Seq(
+      null,
+      null
+    ),
+    Seq(
+      "x",
+      null,
+      "e"
+    )
+  )
+
+  override def expectedResults: Seq[String] = Seq(
+    "zzz",
+    null,
+    "x"
+  )
+
+  override def aggregator: AggregateFunction[String] = new StringMaxWithRetractAggFunction()
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/c5282cbc/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/functions/aggfunctions/MinWithRetractAggFunctionTest.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/functions/aggfunctions/MinWithRetractAggFunctionTest.scala
b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/functions/aggfunctions/MinWithRetractAggFunctionTest.scala
index 2ef6b67..9da3828 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/functions/aggfunctions/MinWithRetractAggFunctionTest.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/functions/aggfunctions/MinWithRetractAggFunctionTest.scala
@@ -186,3 +186,35 @@ class DecimalMinWithRetractAggFunctionTest extends AggFunctionTestBase[BigDecima
 
   override def aggregator: AggregateFunction[BigDecimal] = new DecimalMinWithRetractAggFunction()
 }
+
+class StringMinWithRetractAggFunctionTest extends AggFunctionTestBase[String] {
+
+  override def inputValueSets: Seq[Seq[_]] = Seq(
+    Seq(
+      "abc",
+      "def",
+      "ghi",
+      null,
+      "jkl",
+      null,
+      "zzz"
+    ),
+    Seq(
+      null,
+      null
+    ),
+    Seq(
+      "x",
+      null,
+      "e"
+    )
+  )
+
+  override def expectedResults: Seq[String] = Seq(
+    "abc",
+    null,
+    "e"
+  )
+
+  override def aggregator: AggregateFunction[String] = new StringMinWithRetractAggFunction()
+}


Mime
View raw message