[GitHub] flink pull request #3354: [FLINK-5767] [Table] New aggregate function interf...
Github user asfgit closed the pull request at: https://github.com/apache/flink/pull/3354 --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request #3354: [FLINK-5767] [Table] New aggregate function interf...
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/3354#discussion_r102826611
--- Diff:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/aggfunctions/MaxAggFunction.scala
---
@@ -109,8 +119,11 @@ class DecimalMaxAggFunction extends
MaxAggFunction[BigDecimal] {
if (value != null) {
val v = value.asInstanceOf[BigDecimal]
val accum = accumulator.asInstanceOf[MaxAccumulator[BigDecimal]]
- if (accum.max == null || accum.max.compareTo(v) < 0) {
-accum.max = v
+ if (!accum.f1 || accum.f0.compareTo(v) < 0) {
+accum.f0 = v
+if (!accum.f1) {
--- End diff --
remove condition.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3354: [FLINK-5767] [Table] New aggregate function interf...
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/3354#discussion_r102835585
--- Diff:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/aggfunctions/SumAggFunction.scala
---
@@ -32,7 +31,8 @@ abstract class SumAggFunction[T: Numeric] extends
AggregateFunction[T] {
/** The initial accumulator for Sum aggregate function */
class SumAccumulator[T] extends JTuple2[T, Boolean] with Accumulator {
--- End diff --
We can remove the type `T`. It will then be passed down from the parent
class.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3354: [FLINK-5767] [Table] New aggregate function interf...
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/3354#discussion_r102828705
--- Diff:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/aggfunctions/SumAggFunction.scala
---
@@ -127,30 +132,31 @@ class DecimalSumAggFunction extends
AggregateFunction[BigDecimal] {
if (value != null) {
val v = value.asInstanceOf[BigDecimal]
val accum = accumulator.asInstanceOf[DecimalSumAccumulator]
- if (accum.sum == null) {
-accum.sum = v
+ if (accum.f1 == false) {
--- End diff --
remove condition.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3354: [FLINK-5767] [Table] New aggregate function interf...
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/3354#discussion_r102828543
--- Diff:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/aggfunctions/SumAggFunction.scala
---
@@ -44,21 +44,22 @@ abstract class SumAggFunction[T: Numeric] extends
AggregateFunction[T] {
override def accumulate(accumulator: Accumulator, value: Any) = {
if (value != null) {
val v = value.asInstanceOf[T]
- val accum = accumulator.asInstanceOf[SumAccumulator[T]]
- if (accum.sum == null.asInstanceOf[T]) {
-accum.sum = v
+ val a = accumulator.asInstanceOf[SumAccumulator[T]]
+ if (!a.f1) {
--- End diff --
since we start with `sum = 0` we don't need this condition.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3354: [FLINK-5767] [Table] New aggregate function interf...
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/3354#discussion_r102835657
--- Diff:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/aggfunctions/MinAggFunction.scala
---
@@ -31,7 +31,8 @@ abstract class MinAggFunction[T](implicit ord:
Ordering[T]) extends AggregateFun
/** The initial accumulator for Min aggregate function */
class MinAccumulator[T] extends JTuple2[T, Boolean] with Accumulator {
--- End diff --
We can remove the type `T`
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3354: [FLINK-5767] [Table] New aggregate function interf...
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/3354#discussion_r102835712
--- Diff:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/aggfunctions/MaxAggFunction.scala
---
@@ -31,7 +31,8 @@ abstract class MaxAggFunction[T](implicit ord:
Ordering[T]) extends AggregateFun
/** The initial accumulator for Max aggregate function */
class MaxAccumulator[T] extends JTuple2[T, Boolean] with Accumulator {
--- End diff --
We can remove the type `T`
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3354: [FLINK-5767] [Table] New aggregate function interf...
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/3354#discussion_r102828658
--- Diff:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/aggfunctions/SumAggFunction.scala
---
@@ -67,10 +68,13 @@ abstract class SumAggFunction[T: Numeric] extends
AggregateFunction[T] {
var i: Int = 0
while (i < accumulators.size()) {
val a = accumulators.get(i).asInstanceOf[SumAccumulator[T]]
- if (ret.sum == null.asInstanceOf[T]) {
-ret.sum = a.sum
- } else if (a.sum != null.asInstanceOf[T]) {
-ret.sum = numeric.plus(ret.sum, a.sum)
+ if (a.f1) {
+if (!ret.f1) {
--- End diff --
Since we start with `sum = 0` we don't need this condition.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3354: [FLINK-5767] [Table] New aggregate function interf...
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/3354#discussion_r102835493
--- Diff:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/aggfunctions/SumAggFunction.scala
---
@@ -32,7 +31,8 @@ abstract class SumAggFunction[T: Numeric] extends
AggregateFunction[T] {
/** The initial accumulator for Sum aggregate function */
class SumAccumulator[T] extends JTuple2[T, Boolean] with Accumulator {
-var sum: T = null.asInstanceOf[T]
+f0 = 0.asInstanceOf[T] //sum
--- End diff --
change this to `f0 = numeric.zero`
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3354: [FLINK-5767] [Table] New aggregate function interf...
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/3354#discussion_r102826382
--- Diff:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/aggfunctions/MaxAggFunction.scala
---
@@ -41,24 +42,33 @@ abstract class MaxAggFunction[T](implicit ord:
Ordering[T]) extends AggregateFun
override def accumulate(accumulator: Accumulator, value: Any) = {
if (value != null) {
val v = value.asInstanceOf[T]
- val accum = accumulator.asInstanceOf[MaxAccumulator[T]]
- if (accum.max == null || ord.compare(accum.max, v) < 0) {
-accum.max = v
+ val a = accumulator.asInstanceOf[MaxAccumulator[T]]
+ if (!a.f1 || ord.compare(a.f0, v) < 0) {
+a.f0 = v
+if (!a.f1) {
--- End diff --
the condition can be removed. We can simply reassign `true`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3354: [FLINK-5767] [Table] New aggregate function interf...
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/3354#discussion_r102826652
--- Diff:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/aggfunctions/MinAggFunction.scala
---
@@ -41,24 +42,33 @@ abstract class MinAggFunction[T](implicit ord:
Ordering[T]) extends AggregateFun
override def accumulate(accumulator: Accumulator, value: Any) = {
if (value != null) {
val v = value.asInstanceOf[T]
- val accum = accumulator.asInstanceOf[MinAccumulator[T]]
- if (accum.max == null || ord.compare(accum.max, v) > 0) {
-accum.max = v
+ val a = accumulator.asInstanceOf[MinAccumulator[T]]
+ if (!a.f1 || ord.compare(a.f0, v) > 0) {
+a.f0 = v
+if (!a.f1) {
--- End diff --
Remove condition
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3354: [FLINK-5767] [Table] New aggregate function interf...
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/3354#discussion_r102693258
--- Diff:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/aggfunctions/MaxAggFunction.scala
---
@@ -28,8 +28,9 @@ import org.apache.flink.table.functions.{Accumulator,
AggregateFunction}
* @tparam T the type for the aggregation result
*/
abstract class MaxAggFunction[T](implicit ord: Ordering[T]) extends
AggregateFunction[T] {
+
/** The initial accumulator for Max aggregate function */
- class MaxAccumulator[T] extends Accumulator {
+ class MaxAccumulator[T] extends JTuple2[T, Boolean] with Accumulator {
var max: T = null.asInstanceOf[T]
--- End diff --
Instead of adding a field here, you need to provide default values in the
`createAccumulator` method. Note, `f0` and `f1` may not be null. Otherwise,
the tuple cannot be serialized.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3354: [FLINK-5767] [Table] New aggregate function interf...
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/3354#discussion_r102693561
--- Diff:
flink-libraries/flink-table/src/test/scala/org/apache/flink/table/functions/aggfunctions/AggFunctionTestBase.scala
---
@@ -56,11 +53,15 @@ abstract class AggFunctionTestBase[T] {
if (ifMethodExitInFunction("merge", aggregator)) {
// iterate over input sets
- for((vals, expected) <- inputValueSets.zip(expectedResults)) {
+ for ((vals, expected) <- inputValueSets.zip(expectedResults)) {
val (firstVals, secondVals) = vals.splitAt(vals.length / 2)
--- End diff --
In addition to splitting the `vals` data set, we should also test
correctness, if we merge with an empty accumulator, i.e., aggregate all `vals`
into one accumulator and merge it with an empty one.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3354: [FLINK-5767] [Table] New aggregate function interf...
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/3354#discussion_r102692834
--- Diff:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/aggfunctions/AvgAggFunction.scala
---
@@ -28,8 +28,9 @@ import org.apache.flink.table.functions.{Accumulator,
AggregateFunction}
* @tparam T the type for the aggregation result
*/
abstract class IntegralAvgAggFunction[T] extends AggregateFunction[T] {
+
/** The initial accumulator for Integral Avg aggregate function */
- class IntegralAvgAccumulator extends Accumulator {
+ class IntegralAvgAccumulator extends JTuple2[Long, Long] with
Accumulator {
--- End diff --
If you are using a `JTuple` you may not add additional fields but should
use the Tuple2 fields (`f0` and `f1`).
Since the `TupleSerializer` is not aware of the added fields,
`IntegralAvgAccumulator` won't be treated as Tuple2 by Flink.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3354: [FLINK-5767] [Table] New aggregate function interf...
Github user shaoxuan-wang commented on a diff in the pull request:
https://github.com/apache/flink/pull/3354#discussion_r102606570
--- Diff:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/builtInAggFuncs/SumAggFunction.scala
---
@@ -0,0 +1,146 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.table.functions.builtInAggFuncs
+
+import java.math.BigDecimal
+import org.apache.flink.table.functions.{Accumulator, AggregateFunction}
+
+/**
+ * Base class for built-in Sum aggregate function
+ *
+ * @tparam T the type for the aggregation result
+ */
+abstract class SumAggFunction[T: Numeric] extends AggregateFunction[T] {
+ /** The initial accumulator for Sum aggregate function */
+ class SumAccumulator[T] extends Accumulator {
+var sum: Option[T] = None
--- End diff --
This was cloning from current existing built-in functions. Did not think it
more thoroughly. Yes, you are right. I agree with you that we do not need the
Option here.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3354: [FLINK-5767] [Table] New aggregate function interf...
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/3354#discussion_r102220328
--- Diff:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/AggregateFunction.scala
---
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.table.functions
+
+/**
+ * Base class for User-Defined Aggregates.
+ *
+ * @tparam T the type of the aggregation result
+ */
+abstract class AggregateFunction[T] extends UserDefinedFunction {
+ /**
+* Create and init the Accumulator for this [[AggregateFunction]].
+*
+* @return the accumulator with the initial value
+*/
+ def createAccumulator(): Accumulator
+
+ /**
+* Called every time when an aggregation result should be materialized.
+* The returned value could be either a speculative result (periodically
--- End diff --
`a speculative` -> `an early and incomplete`
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3354: [FLINK-5767] [Table] New aggregate function interf...
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/3354#discussion_r102431482
--- Diff:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/builtInAggFuncs/AvgAggFunction.scala
---
@@ -0,0 +1,268 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.table.functions.builtInAggFuncs
+
+import java.math.{BigDecimal, BigInteger}
+import org.apache.flink.table.functions.{Accumulator, AggregateFunction}
+
+/**
+ * Base class for built-in Integral Avg aggregate function
+ *
+ * @tparam T the type for the aggregation result
+ */
+abstract class IntegralAvgAggFunction[T] extends AggregateFunction[T] {
+ /** The initial accumulator for Integral Avg aggregate function */
+ class IntegralAvgAccumulator extends Accumulator {
+var sum: Long = 0
+var count: Long = 0
+ }
+
+ override def createAccumulator(): Accumulator = {
+new IntegralAvgAccumulator
+ }
+
+ override def accumulate(accumulator: Accumulator, value: Any) = {
+if (value != null) {
+ val v = value.asInstanceOf[Number].longValue()
+ val accum = accumulator.asInstanceOf[IntegralAvgAccumulator]
+ accum.sum += v
+ accum.count += 1
+}
+ }
+
+ override def getValue(accumulator: Accumulator): T = {
+val accum = accumulator.asInstanceOf[IntegralAvgAccumulator]
+val sum = accum.sum
+if (accum.count == 0) {
+ null.asInstanceOf[T]
+} else {
+ resultTypeConvert(accum.sum / accum.count)
+}
+ }
+
+ override def merge(a: Accumulator, b: Accumulator): Accumulator = {
+val aAccum = a.asInstanceOf[IntegralAvgAccumulator]
+val bAccum = b.asInstanceOf[IntegralAvgAccumulator]
+aAccum.count += bAccum.count
+aAccum.sum += bAccum.sum
+a
+ }
+ /**
+* Convert the intermediate result to the expected aggregation result
type
+*
+* @param value the intermediate result. We use a Long container to save
+* the intermediate result to avoid the overflow by sum
operation.
+* @return the result value with the expected aggregation result type
+*/
+ def resultTypeConvert(value: Long): T
+}
+
+/**
+ * Built-in Byte Avg aggregate function
+ */
+class ByteAvgAggFunction extends IntegralAvgAggFunction[Byte] {
+ override def resultTypeConvert(value: Long): Byte = value.toByte
+}
+
+/**
+ * Built-in Short Avg aggregate function
+ */
+class ShortAvgAggFunction extends IntegralAvgAggFunction[Short] {
+ override def resultTypeConvert(value: Long): Short = value.toShort
+}
+
+/**
+ * Built-in Int Avg aggregate function
+ */
+class IntAvgAggFunction extends IntegralAvgAggFunction[Int] {
+ override def resultTypeConvert(value: Long): Int = value.toInt
+}
+
+/**
+ * Base Class for Built-in Big Integral Avg aggregate function
+ *
+ * @tparam T the type for the aggregation result
+ */
+abstract class BigIntegralAvgAggFunction[T] extends AggregateFunction[T] {
+ /** The initial accumulator for Big Integral Avg aggregate function */
+ class BigIntegralAvgAccumulator extends Accumulator {
+var sum: BigInteger = BigInteger.ZERO
+var count: Long = 0
+ }
+
+ override def createAccumulator(): Accumulator = {
+new BigIntegralAvgAccumulator
+ }
+
+ override def accumulate(accumulator: Accumulator, value: Any) = {
+if (value != null) {
+ val v = value.asInstanceOf[Long]
+ val accum = accumulator.asInstanceOf[BigIntegralAvgAccumulator]
+ accum.sum = accum.sum.add(BigInteger.valueOf(v))
+ accum.count += 1
+}
+ }
+
+ override def getValue(accumulator: Accumulator): T = {
+val accum = accumulator.asInstanceOf[BigIntegralAvgAccumulator]
+val sum = accum.sum
+
[GitHub] flink pull request #3354: [FLINK-5767] [Table] New aggregate function interf...
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/3354#discussion_r102431548
--- Diff:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/builtInAggFuncs/CountAggFunction.scala
---
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.table.functions.builtInAggFuncs
+
+import org.apache.flink.table.functions.{Accumulator, AggregateFunction}
+
+/**
+ * built-in count aggregate function
+ */
+class CountAggFunction extends AggregateFunction[Long] {
+ /** The initial accumulator for count aggregate function */
+ class CountAccumulator extends Accumulator {
--- End diff --
Convert to `JTuple1`
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3354: [FLINK-5767] [Table] New aggregate function interf...
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/3354#discussion_r102455635
--- Diff:
flink-libraries/flink-table/src/test/scala/org/apache/flink/table/functions/builtInAggFuncs/AggFunctionTestBase.scala
---
@@ -0,0 +1,104 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.table.functions.builtInAggFuncs
+
+import org.apache.flink.table.functions.{Accumulator, AggregateFunction}
+import java.math.BigDecimal
+
+import org.apache.flink.table.functions.utils.UserDefinedFunctionUtils._
+import org.apache.flink.types.Row
+import org.junit.Test
+import org.junit.Assert.assertEquals
+
+/**
+ * Base class for aggregate function test
+ *
+ * @tparam T the type for the aggregation result
+ */
+abstract class AggFunctionTestBase[T] {
+ private val offset = 2
--- End diff --
None of the tests uses `offset`, `numOfAggregates`, and `rowArity` or `Row`.
I think the tests can be simplified by removing these and the methods
`getResult()` (directly call `getValue()`) and `mergeRows()` (directly call
`merge()`).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3354: [FLINK-5767] [Table] New aggregate function interf...
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/3354#discussion_r102433532
--- Diff:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/builtInAggFuncs/SumAggFunction.scala
---
@@ -0,0 +1,146 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.table.functions.builtInAggFuncs
+
+import java.math.BigDecimal
+import org.apache.flink.table.functions.{Accumulator, AggregateFunction}
+
+/**
+ * Base class for built-in Sum aggregate function
+ *
+ * @tparam T the type for the aggregation result
+ */
+abstract class SumAggFunction[T: Numeric] extends AggregateFunction[T] {
+ /** The initial accumulator for Sum aggregate function */
+ class SumAccumulator[T] extends Accumulator {
--- End diff --
Same as for the accumulator for `MaxAggFunction`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3354: [FLINK-5767] [Table] New aggregate function interf...
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/3354#discussion_r102265663
--- Diff:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/builtInAggFuncs/SumAggFunction.scala
---
@@ -0,0 +1,146 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.table.functions.builtInAggFuncs
+
+import java.math.BigDecimal
+import org.apache.flink.table.functions.{Accumulator, AggregateFunction}
+
+/**
+ * Base class for built-in Sum aggregate function
+ *
+ * @tparam T the type for the aggregation result
+ */
+abstract class SumAggFunction[T: Numeric] extends AggregateFunction[T] {
+ /** The initial accumulator for Sum aggregate function */
+ class SumAccumulator[T] extends Accumulator {
+var sum: Option[T] = None
--- End diff --
why working with `Option` here? The other functions where initialized with
`null`. We should keep that consistent, IMO.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3354: [FLINK-5767] [Table] New aggregate function interf...
Github user fhueske commented on a diff in the pull request: https://github.com/apache/flink/pull/3354#discussion_r102220960 --- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/AggregateFunction.scala --- @@ -0,0 +1,73 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.flink.table.functions + +/** + * Base class for User-Defined Aggregates. --- End diff -- Mention that all state of the function must be put into the accumulator? --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request #3354: [FLINK-5767] [Table] New aggregate function interf...
Github user fhueske commented on a diff in the pull request: https://github.com/apache/flink/pull/3354#discussion_r102225539 --- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/builtInAggFuncs/AvgAggFunction.scala --- @@ -0,0 +1,268 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.flink.table.functions.builtInAggFuncs --- End diff -- Packages are all lowercase by convention. Rename to `aggfunctions`? --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] flink pull request #3354: [FLINK-5767] [Table] New aggregate function interf...
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/3354#discussion_r102220814
--- Diff:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/AggregateFunction.scala
---
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.table.functions
+
+/**
+ * Base class for User-Defined Aggregates.
+ *
+ * @tparam T the type of the aggregation result
+ */
+abstract class AggregateFunction[T] extends UserDefinedFunction {
+ /**
+* Create and init the Accumulator for this [[AggregateFunction]].
+*
+* @return the accumulator with the initial value
+*/
+ def createAccumulator(): Accumulator
+
+ /**
+* Called every time when an aggregation result should be materialized.
+* The returned value could be either a speculative result (periodically
+* emitted as data arrive) or the final result of the aggregation (when
+* the state of the aggregation is completely removed).
--- End diff --
How would the function distinguish these cases?
I think we should not mention that the state is removed because it is not
the responsibility of the method.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3354: [FLINK-5767] [Table] New aggregate function interf...
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/3354#discussion_r102241426
--- Diff:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/builtInAggFuncs/AvgAggFunction.scala
---
@@ -0,0 +1,268 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.table.functions.builtInAggFuncs
+
+import java.math.{BigDecimal, BigInteger}
+import org.apache.flink.table.functions.{Accumulator, AggregateFunction}
+
+/**
+ * Base class for built-in Integral Avg aggregate function
+ *
+ * @tparam T the type for the aggregation result
+ */
+abstract class IntegralAvgAggFunction[T] extends AggregateFunction[T] {
+ /** The initial accumulator for Integral Avg aggregate function */
+ class IntegralAvgAccumulator extends Accumulator {
+var sum: Long = 0
+var count: Long = 0
+ }
+
+ override def createAccumulator(): Accumulator = {
+new IntegralAvgAccumulator
+ }
+
+ override def accumulate(accumulator: Accumulator, value: Any) = {
+if (value != null) {
+ val v = value.asInstanceOf[Number].longValue()
+ val accum = accumulator.asInstanceOf[IntegralAvgAccumulator]
+ accum.sum += v
+ accum.count += 1
+}
+ }
+
+ override def getValue(accumulator: Accumulator): T = {
+val accum = accumulator.asInstanceOf[IntegralAvgAccumulator]
+val sum = accum.sum
+if (accum.count == 0) {
+ null.asInstanceOf[T]
+} else {
+ resultTypeConvert(accum.sum / accum.count)
+}
+ }
+
+ override def merge(a: Accumulator, b: Accumulator): Accumulator = {
+val aAccum = a.asInstanceOf[IntegralAvgAccumulator]
+val bAccum = b.asInstanceOf[IntegralAvgAccumulator]
+aAccum.count += bAccum.count
+aAccum.sum += bAccum.sum
+a
+ }
+ /**
+* Convert the intermediate result to the expected aggregation result
type
+*
+* @param value the intermediate result. We use a Long container to save
+* the intermediate result to avoid the overflow by sum
operation.
+* @return the result value with the expected aggregation result type
+*/
+ def resultTypeConvert(value: Long): T
+}
+
+/**
+ * Built-in Byte Avg aggregate function
+ */
+class ByteAvgAggFunction extends IntegralAvgAggFunction[Byte] {
+ override def resultTypeConvert(value: Long): Byte = value.toByte
+}
+
+/**
+ * Built-in Short Avg aggregate function
+ */
+class ShortAvgAggFunction extends IntegralAvgAggFunction[Short] {
+ override def resultTypeConvert(value: Long): Short = value.toShort
+}
+
+/**
+ * Built-in Int Avg aggregate function
+ */
+class IntAvgAggFunction extends IntegralAvgAggFunction[Int] {
+ override def resultTypeConvert(value: Long): Int = value.toInt
+}
+
+/**
+ * Base Class for Built-in Big Integral Avg aggregate function
+ *
+ * @tparam T the type for the aggregation result
+ */
+abstract class BigIntegralAvgAggFunction[T] extends AggregateFunction[T] {
+ /** The initial accumulator for Big Integral Avg aggregate function */
+ class BigIntegralAvgAccumulator extends Accumulator {
+var sum: BigInteger = BigInteger.ZERO
+var count: Long = 0
+ }
+
+ override def createAccumulator(): Accumulator = {
+new BigIntegralAvgAccumulator
+ }
+
+ override def accumulate(accumulator: Accumulator, value: Any) = {
+if (value != null) {
+ val v = value.asInstanceOf[Long]
+ val accum = accumulator.asInstanceOf[BigIntegralAvgAccumulator]
+ accum.sum = accum.sum.add(BigInteger.valueOf(v))
+ accum.count += 1
+}
+ }
+
+ override def getValue(accumulator: Accumulator): T = {
+val accum = accumulator.asInstanceOf[BigIntegralAvgAccumulator]
+val sum = accum.sum
+
[GitHub] flink pull request #3354: [FLINK-5767] [Table] New aggregate function interf...
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/3354#discussion_r102432444
--- Diff:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/builtInAggFuncs/MaxAggFunction.scala
---
@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.table.functions.builtInAggFuncs
+
+import java.math.BigDecimal
+import org.apache.flink.table.functions.{Accumulator, AggregateFunction}
+
+/**
+ * Base class for built-in Max aggregate function
+ *
+ * @tparam T the type for the aggregation result
+ */
+abstract class MaxAggFunction[T](implicit ord: Ordering[T]) extends
AggregateFunction[T] {
+ /** The initial accumulator for Max aggregate function */
+ class MaxAccumulator[T] extends Accumulator {
--- End diff --
Make this a `class MaxAccumulator[T] extends JTuple2[T, Boolean] with
Accumulator`?
We need the additional boolean field as a flag to indicate null values.
Java tuples do not support null values, so the first field must be initialized
but should be considered null.
This does not add space overhead, since a PojoSerializer would also
serialize a `byte` to indicate null valued fields.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3354: [FLINK-5767] [Table] New aggregate function interf...
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/3354#discussion_r102226285
--- Diff:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/builtInAggFuncs/AvgAggFunction.scala
---
@@ -0,0 +1,268 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.table.functions.builtInAggFuncs
+
+import java.math.{BigDecimal, BigInteger}
+import org.apache.flink.table.functions.{Accumulator, AggregateFunction}
+
+/**
+ * Base class for built-in Integral Avg aggregate function
+ *
+ * @tparam T the type for the aggregation result
+ */
+abstract class IntegralAvgAggFunction[T] extends AggregateFunction[T] {
+ /** The initial accumulator for Integral Avg aggregate function */
+ class IntegralAvgAccumulator extends Accumulator {
--- End diff --
This will be probably handled as a Pojo type which is less efficient than a
`Tuple2`
We can define this as
```
import org.apache.flink.api.java.tuple.{Tuple2 => JTuple2} // use Java
tuple because it's mutable.
class IntegralAvgAccumulator extends JTuple2[Long, Long] with Accumulator
```
which means that Flink can handle it as a Tuple which is more efficient
than a POJO which accesses fields (for de/serialization) via reflection.
I think this implementation will also be closer to the code-gen'd version.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3354: [FLINK-5767] [Table] New aggregate function interf...
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/3354#discussion_r102433475
--- Diff:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/functions/builtInAggFuncs/MinAggFunction.scala
---
@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.table.functions.builtInAggFuncs
+
+import java.math.BigDecimal
+import org.apache.flink.table.functions.{Accumulator, AggregateFunction}
+
+/**
+ * Base class for built-in Min aggregate function
+ *
+ * @tparam T the type for the aggregation result
+ */
+abstract class MinAggFunction[T](implicit ord: Ordering[T]) extends
AggregateFunction[T] {
+ /** The initial accumulator for Min aggregate function */
+ class MinAccumulator[T] extends Accumulator {
--- End diff --
Same as for the accumulator for `MaxAggFunction`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3354: [FLINK-5767] [Table] New aggregate function interf...
GitHub user shaoxuan-wang opened a pull request:
https://github.com/apache/flink/pull/3354
[FLINK-5767] [Table] New aggregate function interface and built-in
aggregate functions
Thanks for contributing to Apache Flink. Before you open your pull request,
please take the following check list into consideration.
If your changes take all of the items into account, feel free to open your
pull request. For more information and/or questions please refer to the [How To
Contribute guide](http://flink.apache.org/how-to-contribute.html).
In addition to going through the list, please provide a meaningful
description of your changes.
- [X] General
- The pull request references the related JIRA issue ("[FLINK-XXX] Jira
title text")
- The pull request addresses only one issue
- Each commit in the PR has a meaningful commit message (including the
JIRA id)
- [ ] Documentation
- Documentation has been added for new functionality
- Old documentation affected by the pull request has been updated
- JavaDoc for public methods has been added
- [X] Tests & Build
- Functionality added by the pull request is covered by tests
- `mvn clean verify` has been executed successfully locally or a Travis
build has passed
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/shaoxuan-wang/flink F5767-submit
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/3354.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #3354
commit 6a2937a94fb2a00345fb5792a8c00c604fbee2c7
Author: Shaoxuan Wang
Date: 2017-02-18T09:18:38Z
[FLINK-5767] [Table] New aggregate function interface and built-in
aggregate functions
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
