You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tw...@apache.org on 2017/04/03 16:25:06 UTC
flink git commit: [FLINK-6124] [table] Add min/max string aggregation
with retracion
Repository: flink
Updated Branches:
refs/heads/master 05ceec0ac -> c5282cbcf
[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/master
Commit: c5282cbcf898b99593ca753fce7557bae1ae09aa
Parents: 05ceec0
Author: twalthr <tw...@apache.org>
Authored: Wed Mar 22 10:03:56 2017 +0100
Committer: twalthr <tw...@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()
+}