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 2020/07/29 15:40:43 UTC
[flink] 01/06: [hotfix][table] Rename UserDefinedAggregateFunction
to ImperativeAggregateFunction
This is an automated email from the ASF dual-hosted git repository.
twalthr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 82f3094e19ea1db6ee94b129d9b21852b9c366dc
Author: Timo Walther <tw...@apache.org>
AuthorDate: Mon Jul 27 15:44:18 2020 +0200
[hotfix][table] Rename UserDefinedAggregateFunction to ImperativeAggregateFunction
Renames this base class because users can easily mix up AggregateFunction and
UserDefinedAggregateFunction. The new naming also prepares for the future
DeclarativeAggregateFunction. ImperativeAggregateFunction has already been introduced
as a concept in the Blink planner.
---
flink-python/pyflink/table/table_environment.py | 2 +-
.../flink/table/catalog/FunctionCatalog.java | 4 +-
.../table/api/ImplicitExpressionConversions.scala | 6 +--
.../api/ExpressionsConsistencyCheckTest.scala | 2 +-
.../flink/table/functions/AggregateFunction.java | 4 +-
...ction.java => ImperativeAggregateFunction.java} | 12 ++---
.../table/functions/TableAggregateFunction.java | 4 +-
.../table/functions/UserDefinedFunctionHelper.java | 12 ++---
.../codegen/agg/AggsHandlerCodeGenerator.scala | 4 +-
.../planner/codegen/agg/ImperativeAggCodeGen.scala | 4 +-
.../table/planner/dataview/DataViewUtils.scala | 7 +--
.../table/planner/expressions/aggregations.scala | 4 +-
.../planner/functions/utils/AggSqlFunction.scala | 21 ++++----
.../functions/utils/UserDefinedFunctionUtils.scala | 54 ++++++++++----------
.../table/planner/plan/utils/AggregateUtil.scala | 4 +-
.../flink/table/planner/utils/TableTestBase.scala | 8 +--
.../flink/table/api/internal/TableEnvImpl.scala | 4 +-
.../table/codegen/AggregationCodeGenerator.scala | 59 +++++++++++-----------
.../flink/table/codegen/MatchCodeGenerator.scala | 19 +++----
.../flink/table/expressions/aggregations.scala | 20 ++++----
.../table/functions/utils/AggSqlFunction.scala | 34 +++++++------
.../functions/utils/UserDefinedFunctionUtils.scala | 10 ++--
.../table/runtime/aggregate/AggregateUtil.scala | 40 ++++++++-------
23 files changed, 175 insertions(+), 163 deletions(-)
diff --git a/flink-python/pyflink/table/table_environment.py b/flink-python/pyflink/table/table_environment.py
index aa05b13..d080b93 100644
--- a/flink-python/pyflink/table/table_environment.py
+++ b/flink-python/pyflink/table/table_environment.py
@@ -1571,7 +1571,7 @@ class TableEnvironment(object):
def _is_aggregate_function(java_function):
java_function_class = java_function.getClass()
j_aggregate_function_class = get_java_class(
- get_gateway().jvm.org.apache.flink.table.functions.UserDefinedAggregateFunction)
+ get_gateway().jvm.org.apache.flink.table.functions.ImperativeAggregateFunction)
return j_aggregate_function_class.isAssignableFrom(java_function_class)
def _register_table_function(self, name, table_function):
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/FunctionCatalog.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/FunctionCatalog.java
index 38c4094..d0843d4 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/FunctionCatalog.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/FunctionCatalog.java
@@ -32,13 +32,13 @@ import org.apache.flink.table.functions.AggregateFunctionDefinition;
import org.apache.flink.table.functions.FunctionDefinition;
import org.apache.flink.table.functions.FunctionDefinitionUtil;
import org.apache.flink.table.functions.FunctionIdentifier;
+import org.apache.flink.table.functions.ImperativeAggregateFunction;
import org.apache.flink.table.functions.ScalarFunction;
import org.apache.flink.table.functions.ScalarFunctionDefinition;
import org.apache.flink.table.functions.TableAggregateFunction;
import org.apache.flink.table.functions.TableAggregateFunctionDefinition;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.table.functions.TableFunctionDefinition;
-import org.apache.flink.table.functions.UserDefinedAggregateFunction;
import org.apache.flink.table.functions.UserDefinedFunction;
import org.apache.flink.table.functions.UserDefinedFunctionHelper;
import org.apache.flink.table.module.ModuleManager;
@@ -409,7 +409,7 @@ public final class FunctionCatalog {
public <T, ACC> void registerTempSystemAggregateFunction(
String name,
- UserDefinedAggregateFunction<T, ACC> function,
+ ImperativeAggregateFunction<T, ACC> function,
TypeInformation<T> resultType,
TypeInformation<ACC> accType) {
UserDefinedFunctionHelper.prepareInstance(config, function);
diff --git a/flink-table/flink-table-api-scala/src/main/scala/org/apache/flink/table/api/ImplicitExpressionConversions.scala b/flink-table/flink-table-api-scala/src/main/scala/org/apache/flink/table/api/ImplicitExpressionConversions.scala
index 23e45a3..2e63b68 100644
--- a/flink-table/flink-table-api-scala/src/main/scala/org/apache/flink/table/api/ImplicitExpressionConversions.scala
+++ b/flink-table/flink-table-api-scala/src/main/scala/org/apache/flink/table/api/ImplicitExpressionConversions.scala
@@ -23,7 +23,7 @@ import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.table.expressions.ApiExpressionUtils.{unresolvedCall, unresolvedRef, valueLiteral}
import org.apache.flink.table.expressions.{ApiExpressionUtils, Expression, TableSymbol, TimePointUnit}
import org.apache.flink.table.functions.BuiltInFunctionDefinitions.{DISTINCT, RANGE_TO}
-import org.apache.flink.table.functions.{ScalarFunction, TableFunction, UserDefinedAggregateFunction, UserDefinedFunctionHelper, _}
+import org.apache.flink.table.functions.{ScalarFunction, TableFunction, ImperativeAggregateFunction, UserDefinedFunctionHelper, _}
import org.apache.flink.table.types.DataType
import org.apache.flink.types.Row
@@ -173,8 +173,8 @@ trait ImplicitExpressionConversions {
}
}
- implicit class UserDefinedAggregateFunctionCall[T: TypeInformation, ACC: TypeInformation]
- (val a: UserDefinedAggregateFunction[T, ACC]) {
+ implicit class ImperativeAggregateFunctionCall[T: TypeInformation, ACC: TypeInformation]
+ (val a: ImperativeAggregateFunction[T, ACC]) {
private def createFunctionDefinition(): FunctionDefinition = {
val resultTypeInfo: TypeInformation[T] = UserDefinedFunctionHelper
diff --git a/flink-table/flink-table-api-scala/src/test/scala/org/apache/flink/table/api/ExpressionsConsistencyCheckTest.scala b/flink-table/flink-table-api-scala/src/test/scala/org/apache/flink/table/api/ExpressionsConsistencyCheckTest.scala
index e9ef414..26c30d5 100644
--- a/flink-table/flink-table-api-scala/src/test/scala/org/apache/flink/table/api/ExpressionsConsistencyCheckTest.scala
+++ b/flink-table/flink-table-api-scala/src/test/scala/org/apache/flink/table/api/ExpressionsConsistencyCheckTest.scala
@@ -55,7 +55,7 @@ class ExpressionsConsistencyCheckTest {
val explicitScalaToJavaStaticMethodsMapping = Map(
"FieldExpression" -> "$",
"UnresolvedFieldExpression" -> "$",
- "UserDefinedAggregateFunctionCall" -> "call",
+ "ImperativeAggregateFunctionCall" -> "call",
"ScalarFunctionCall" -> "call",
"TableFunctionCall" -> "call",
"concat_ws" -> "concatWs"
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/AggregateFunction.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/AggregateFunction.java
index 6d0f30a..13c2745 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/AggregateFunction.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/AggregateFunction.java
@@ -47,7 +47,7 @@ import java.util.Set;
*
* <p>All these methods must be declared publicly, not static, and named exactly as the names
* mentioned above. The method {@link #createAccumulator()} is defined in the
- * {@link UserDefinedAggregateFunction} function, and method {@link #getValue} is defined in
+ * {@link ImperativeAggregateFunction} function, and method {@link #getValue} is defined in
* the {@link AggregateFunction} while other methods are explained below.
*
* <pre>
@@ -114,7 +114,7 @@ import java.util.Set;
* AggregateFunction must be put into the accumulator.
*/
@PublicEvolving
-public abstract class AggregateFunction<T, ACC> extends UserDefinedAggregateFunction<T, ACC> {
+public abstract class AggregateFunction<T, ACC> extends ImperativeAggregateFunction<T, ACC> {
/**
* Called every time when an aggregation result should be materialized.
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/UserDefinedAggregateFunction.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/ImperativeAggregateFunction.java
similarity index 73%
rename from flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/UserDefinedAggregateFunction.java
rename to flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/ImperativeAggregateFunction.java
index e03910d..e39a9d0 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/UserDefinedAggregateFunction.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/ImperativeAggregateFunction.java
@@ -25,10 +25,10 @@ import org.apache.flink.api.common.typeinfo.TypeInformation;
* Base class for user-defined aggregates and table aggregates.
*/
@PublicEvolving
-public abstract class UserDefinedAggregateFunction<T, ACC> extends UserDefinedFunction {
+public abstract class ImperativeAggregateFunction<T, ACC> extends UserDefinedFunction {
/**
- * Creates and initializes the accumulator for this {@link UserDefinedAggregateFunction}. The
+ * Creates and initializes the accumulator for this {@link ImperativeAggregateFunction}. The
* accumulator is used to keep the aggregated values which are needed to compute an aggregation
* result.
*
@@ -37,9 +37,9 @@ public abstract class UserDefinedAggregateFunction<T, ACC> extends UserDefinedFu
public abstract ACC createAccumulator();
/**
- * Returns the {@link TypeInformation} of the {@link UserDefinedAggregateFunction}'s result.
+ * Returns the {@link TypeInformation} of the {@link ImperativeAggregateFunction}'s result.
*
- * @return The {@link TypeInformation} of the {@link UserDefinedAggregateFunction}'s result or
+ * @return The {@link TypeInformation} of the {@link ImperativeAggregateFunction}'s result or
* <code>null</code> if the result type should be automatically inferred.
*/
public TypeInformation<T> getResultType() {
@@ -47,9 +47,9 @@ public abstract class UserDefinedAggregateFunction<T, ACC> extends UserDefinedFu
}
/**
- * Returns the {@link TypeInformation} of the {@link UserDefinedAggregateFunction}'s accumulator.
+ * Returns the {@link TypeInformation} of the {@link ImperativeAggregateFunction}'s accumulator.
*
- * @return The {@link TypeInformation} of the {@link UserDefinedAggregateFunction}'s accumulator
+ * @return The {@link TypeInformation} of the {@link ImperativeAggregateFunction}'s accumulator
* or <code>null</code> if the accumulator type should be automatically inferred.
*/
public TypeInformation<ACC> getAccumulatorType() {
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/TableAggregateFunction.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/TableAggregateFunction.java
index c8ee292..3941eaa 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/TableAggregateFunction.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/TableAggregateFunction.java
@@ -42,7 +42,7 @@ import org.apache.flink.util.Collector;
*
* <p>All these methods must be declared publicly, not static, and named exactly as the names
* mentioned above. The method {@link #createAccumulator()} is defined in
- * the {@link UserDefinedAggregateFunction} functions, while other methods are explained below.
+ * the {@link ImperativeAggregateFunction} functions, while other methods are explained below.
*
* <pre>
* {@code
@@ -112,7 +112,7 @@ import org.apache.flink.util.Collector;
* the TableAggregateFunction must be put into the accumulator.
*/
@PublicEvolving
-public abstract class TableAggregateFunction<T, ACC> extends UserDefinedAggregateFunction<T, ACC> {
+public abstract class TableAggregateFunction<T, ACC> extends ImperativeAggregateFunction<T, ACC> {
/**
* Collects a record and forwards it. The collector can output retract messages with the retract
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/UserDefinedFunctionHelper.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/UserDefinedFunctionHelper.java
index e22fd7c..17af912 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/UserDefinedFunctionHelper.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/UserDefinedFunctionHelper.java
@@ -81,7 +81,7 @@ public final class UserDefinedFunctionHelper {
* @return The inferred accumulator type of the AggregateFunction.
*/
public static <T, ACC> TypeInformation<T> getReturnTypeOfAggregateFunction(
- UserDefinedAggregateFunction<T, ACC> aggregateFunction) {
+ ImperativeAggregateFunction<T, ACC> aggregateFunction) {
return getReturnTypeOfAggregateFunction(aggregateFunction, null);
}
@@ -94,7 +94,7 @@ public final class UserDefinedFunctionHelper {
* @return The inferred accumulator type of the AggregateFunction.
*/
public static <T, ACC> TypeInformation<T> getReturnTypeOfAggregateFunction(
- UserDefinedAggregateFunction<T, ACC> aggregateFunction,
+ ImperativeAggregateFunction<T, ACC> aggregateFunction,
TypeInformation<T> scalaType) {
TypeInformation<T> userProvidedType = aggregateFunction.getResultType();
@@ -105,7 +105,7 @@ public final class UserDefinedFunctionHelper {
} else {
return TypeExtractor.createTypeInfo(
aggregateFunction,
- UserDefinedAggregateFunction.class,
+ ImperativeAggregateFunction.class,
aggregateFunction.getClass(),
0);
}
@@ -118,7 +118,7 @@ public final class UserDefinedFunctionHelper {
* @return The inferred accumulator type of the AggregateFunction.
*/
public static <T, ACC> TypeInformation<ACC> getAccumulatorTypeOfAggregateFunction(
- UserDefinedAggregateFunction<T, ACC> aggregateFunction) {
+ ImperativeAggregateFunction<T, ACC> aggregateFunction) {
return getAccumulatorTypeOfAggregateFunction(aggregateFunction, null);
}
@@ -131,7 +131,7 @@ public final class UserDefinedFunctionHelper {
* @return The inferred accumulator type of the AggregateFunction.
*/
public static <T, ACC> TypeInformation<ACC> getAccumulatorTypeOfAggregateFunction(
- UserDefinedAggregateFunction<T, ACC> aggregateFunction,
+ ImperativeAggregateFunction<T, ACC> aggregateFunction,
TypeInformation<ACC> scalaType) {
TypeInformation<ACC> userProvidedType = aggregateFunction.getAccumulatorType();
@@ -142,7 +142,7 @@ public final class UserDefinedFunctionHelper {
} else {
return TypeExtractor.createTypeInfo(
aggregateFunction,
- UserDefinedAggregateFunction.class,
+ ImperativeAggregateFunction.class,
aggregateFunction.getClass(),
1);
}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/agg/AggsHandlerCodeGenerator.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/agg/AggsHandlerCodeGenerator.scala
index dfccc46..39f4420 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/agg/AggsHandlerCodeGenerator.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/agg/AggsHandlerCodeGenerator.scala
@@ -21,7 +21,7 @@ import org.apache.flink.api.common.functions.RuntimeContext
import org.apache.flink.table.api.TableException
import org.apache.flink.table.data.GenericRowData
import org.apache.flink.table.expressions._
-import org.apache.flink.table.functions.UserDefinedAggregateFunction
+import org.apache.flink.table.functions.ImperativeAggregateFunction
import org.apache.flink.table.planner.codegen.CodeGenUtils.{ROW_DATA, _}
import org.apache.flink.table.planner.codegen.Indenter.toISC
import org.apache.flink.table.planner.codegen._
@@ -227,7 +227,7 @@ class AggsHandlerCodeGenerator(
inputFieldTypes,
constants,
relBuilder)
- case _: UserDefinedAggregateFunction[_, _] =>
+ case _: ImperativeAggregateFunction[_, _] =>
new ImperativeAggCodeGen(
ctx,
aggInfo,
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/agg/ImperativeAggCodeGen.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/agg/ImperativeAggCodeGen.scala
index 36935f9..c59acf7 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/agg/ImperativeAggCodeGen.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/agg/ImperativeAggCodeGen.scala
@@ -19,7 +19,7 @@ package org.apache.flink.table.planner.codegen.agg
import org.apache.flink.table.data.{GenericRowData, RowData, UpdatableRowData}
import org.apache.flink.table.expressions.Expression
-import org.apache.flink.table.functions.{UserDefinedAggregateFunction, UserDefinedFunctionHelper}
+import org.apache.flink.table.functions.{ImperativeAggregateFunction, UserDefinedFunctionHelper}
import org.apache.flink.table.planner.codegen.CodeGenUtils._
import org.apache.flink.table.planner.codegen.GenerateUtils.generateFieldAccess
import org.apache.flink.table.planner.codegen.agg.AggsHandlerCodeGenerator._
@@ -79,7 +79,7 @@ class ImperativeAggCodeGen(
private val SINGLE_ITERABLE = className[SingleElementIterator[_]]
private val UPDATABLE_ROW = className[UpdatableRowData]
- val function = aggInfo.function.asInstanceOf[UserDefinedAggregateFunction[_, _]]
+ val function = aggInfo.function.asInstanceOf[ImperativeAggregateFunction[_, _]]
val functionTerm: String = ctx.addReusableFunction(
function,
contextTerm = s"$STORE_TERM.getRuntimeContext()")
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/dataview/DataViewUtils.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/dataview/DataViewUtils.scala
index 6cb96e3..c028d9a 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/dataview/DataViewUtils.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/dataview/DataViewUtils.scala
@@ -25,7 +25,7 @@ import org.apache.flink.table.api.dataview._
import org.apache.flink.table.data.binary.BinaryRawValueData
import org.apache.flink.table.data.{GenericRowData, RowData}
import org.apache.flink.table.dataview.{ListViewTypeInfo, MapViewTypeInfo}
-import org.apache.flink.table.functions.UserDefinedAggregateFunction
+import org.apache.flink.table.functions.ImperativeAggregateFunction
import org.apache.flink.table.runtime.types.TypeInfoLogicalTypeConverter.{fromLogicalTypeToTypeInfo, fromTypeInfoToLogicalType}
import org.apache.flink.table.runtime.typeutils.InternalTypeInfo
import org.apache.flink.table.types.DataType
@@ -50,9 +50,10 @@ object DataViewUtils {
*/
def useNullSerializerForStateViewFieldsFromAccType(
index: Int,
- aggFun: UserDefinedAggregateFunction[_, _],
+ aggFun: ImperativeAggregateFunction[_, _],
externalAccType: DataType,
- isStateBackedDataViews: Boolean): (DataType, Array[DataViewSpec]) = {
+ isStateBackedDataViews: Boolean)
+ : (DataType, Array[DataViewSpec]) = {
val acc = aggFun.createAccumulator()
val accumulatorSpecs = new mutable.ArrayBuffer[DataViewSpec]
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/expressions/aggregations.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/expressions/aggregations.scala
index 7249c34..7e903c5 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/expressions/aggregations.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/expressions/aggregations.scala
@@ -20,7 +20,7 @@ package org.apache.flink.table.planner.expressions
import org.apache.flink.api.common.typeinfo.{BasicTypeInfo, TypeInformation}
import org.apache.flink.api.java.typeutils.MultisetTypeInfo
import org.apache.flink.table.expressions.CallExpression
-import org.apache.flink.table.functions.{AggregateFunction, TableAggregateFunction, UserDefinedAggregateFunction}
+import org.apache.flink.table.functions.ImperativeAggregateFunction
import org.apache.flink.table.planner.calcite.FlinkTypeSystem
import org.apache.flink.table.planner.functions.utils.UserDefinedFunctionUtils._
import org.apache.flink.table.planner.typeutils.TypeInfoCheckUtils
@@ -208,7 +208,7 @@ case class VarSamp(child: PlannerExpression) extends Aggregation {
* Expression for calling a user-defined (table)aggregate function.
*/
case class AggFunctionCall(
- aggregateFunction: UserDefinedAggregateFunction[_, _],
+ aggregateFunction: ImperativeAggregateFunction[_, _],
resultTypeInfo: TypeInformation[_],
accTypeInfo: TypeInformation[_],
args: Seq[PlannerExpression])
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/functions/utils/AggSqlFunction.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/functions/utils/AggSqlFunction.scala
index 0b467be..301f4d4 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/functions/utils/AggSqlFunction.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/functions/utils/AggSqlFunction.scala
@@ -19,7 +19,7 @@
package org.apache.flink.table.planner.functions.utils
import org.apache.flink.table.api.ValidationException
-import org.apache.flink.table.functions.{AggregateFunction, FunctionIdentifier, TableAggregateFunction, UserDefinedAggregateFunction}
+import org.apache.flink.table.functions.{AggregateFunction, FunctionIdentifier, TableAggregateFunction, ImperativeAggregateFunction}
import org.apache.flink.table.planner.calcite.FlinkTypeFactory
import org.apache.flink.table.planner.functions.utils.AggSqlFunction.{createOperandTypeChecker, createOperandTypeInference, createReturnTypeInference}
import org.apache.flink.table.planner.functions.utils.UserDefinedFunctionUtils._
@@ -50,7 +50,7 @@ import java.util
class AggSqlFunction(
identifier: FunctionIdentifier,
displayName: String,
- val aggregateFunction: UserDefinedAggregateFunction[_, _],
+ val aggregateFunction: ImperativeAggregateFunction[_, _],
val externalResultType: DataType,
val externalAccType: DataType,
typeFactory: FlinkTypeFactory,
@@ -77,7 +77,7 @@ class AggSqlFunction(
*/
def makeFunction(
constants: Array[AnyRef],
- argTypes: Array[LogicalType]): UserDefinedAggregateFunction[_, _] = aggregateFunction
+ argTypes: Array[LogicalType]): ImperativeAggregateFunction[_, _] = aggregateFunction
override def isDeterministic: Boolean = aggregateFunction.isDeterministic
@@ -91,11 +91,12 @@ object AggSqlFunction {
def apply(
identifier: FunctionIdentifier,
displayName: String,
- aggregateFunction: UserDefinedAggregateFunction[_, _],
+ aggregateFunction: ImperativeAggregateFunction[_, _],
externalResultType: DataType,
externalAccType: DataType,
typeFactory: FlinkTypeFactory,
- requiresOver: Boolean): AggSqlFunction = {
+ requiresOver: Boolean)
+ : AggSqlFunction = {
new AggSqlFunction(
identifier,
@@ -109,9 +110,10 @@ object AggSqlFunction {
private[flink] def createOperandTypeInference(
name: String,
- aggregateFunction: UserDefinedAggregateFunction[_, _],
+ aggregateFunction: ImperativeAggregateFunction[_, _],
typeFactory: FlinkTypeFactory,
- externalAccType: DataType): SqlOperandTypeInference = {
+ externalAccType: DataType)
+ : SqlOperandTypeInference = {
/**
* Operand type inference based on [[AggregateFunction]] given information.
*/
@@ -161,8 +163,9 @@ object AggSqlFunction {
private[flink] def createOperandTypeChecker(
name: String,
- aggregateFunction: UserDefinedAggregateFunction[_, _],
- externalAccType: DataType): SqlOperandTypeChecker = {
+ aggregateFunction: ImperativeAggregateFunction[_, _],
+ externalAccType: DataType)
+ : SqlOperandTypeChecker = {
val methods = checkAndExtractMethods(aggregateFunction, "accumulate")
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/functions/utils/UserDefinedFunctionUtils.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/functions/utils/UserDefinedFunctionUtils.scala
index 5f099a8..dc02dc1 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/functions/utils/UserDefinedFunctionUtils.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/functions/utils/UserDefinedFunctionUtils.scala
@@ -151,9 +151,10 @@ object UserDefinedFunctionUtils {
}
def getAggUserDefinedInputTypes(
- func: UserDefinedAggregateFunction[_, _],
+ func: ImperativeAggregateFunction[_, _],
externalAccType: DataType,
- expectedTypes: Array[LogicalType]): Array[DataType] = {
+ expectedTypes: Array[LogicalType])
+ : Array[DataType] = {
val accMethod = getAggFunctionUDIMethod(
func, "accumulate", externalAccType, expectedTypes).getOrElse(
throwValidationException(func.getClass.getCanonicalName, func, expectedTypes)
@@ -191,9 +192,9 @@ object UserDefinedFunctionUtils {
* Elements of the signature can be null (act as a wildcard).
*/
def getAccumulateMethodSignature(
- function: UserDefinedAggregateFunction[_, _],
+ function: ImperativeAggregateFunction[_, _],
expectedTypes: Seq[LogicalType])
- : Option[Array[Class[_]]] = {
+ : Option[Array[Class[_]]] = {
getAggFunctionUDIMethod(
function,
"accumulate",
@@ -242,11 +243,11 @@ object UserDefinedFunctionUtils {
}
def getAggFunctionUDIMethod(
- function: UserDefinedAggregateFunction[_, _],
+ function: ImperativeAggregateFunction[_, _],
methodName: String,
accType: DataType,
expectedTypes: Seq[LogicalType])
- : Option[Method] = {
+ : Option[Method] = {
val input = (Array(fromDataTypeToLogicalType(accType)) ++ expectedTypes).toSeq
val possibleSignatures = Seq(
logicalTypesToInternalClasses(input),
@@ -539,16 +540,17 @@ object UserDefinedFunctionUtils {
// ----------------------------------------------------------------------------------------------
/**
- * Tries to infer the DataType of a [[UserDefinedAggregateFunction]]'s return type.
+ * Tries to infer the DataType of a [[ImperativeAggregateFunction]]'s return type.
*
- * @param userDefinedAggregateFunction The [[UserDefinedAggregateFunction]] for which the return
+ * @param userDefinedAggregateFunction The [[ImperativeAggregateFunction]] for which the return
* type is inferred.
* @param extractedType The implicitly inferred type of the result type.
- * @return The inferred result type of the [[UserDefinedAggregateFunction]].
+ * @return The inferred result type of the [[ImperativeAggregateFunction]].
*/
def getResultTypeOfAggregateFunction(
- userDefinedAggregateFunction: UserDefinedAggregateFunction[_, _],
- extractedType: DataType = null): DataType = {
+ userDefinedAggregateFunction: ImperativeAggregateFunction[_, _],
+ extractedType: DataType = null)
+ : DataType = {
val resultType = userDefinedAggregateFunction.getResultType
if (resultType != null) {
@@ -562,7 +564,7 @@ object UserDefinedFunctionUtils {
case ite: InvalidTypesException =>
throw new TableException(
"Cannot infer generic type of ${aggregateFunction.getClass}. " +
- "You can override UserDefinedAggregateFunction.getResultType() to " +
+ "You can override ImperativeAggregateFunction.getResultType() to " +
"specify the type.",
ite
)
@@ -571,17 +573,17 @@ object UserDefinedFunctionUtils {
}
/**
- * Tries to infer the Type of a [[UserDefinedAggregateFunction]]'s accumulator type.
+ * Tries to infer the Type of a [[ImperativeAggregateFunction]]'s accumulator type.
*
- * @param userDefinedAggregateFunction The [[UserDefinedAggregateFunction]] for which the
+ * @param userDefinedAggregateFunction The [[ImperativeAggregateFunction]] for which the
* accumulator type is inferred.
* @param extractedType The implicitly inferred type of the accumulator type.
- * @return The inferred accumulator type of the [[UserDefinedAggregateFunction]].
+ * @return The inferred accumulator type of the [[ImperativeAggregateFunction]].
*/
def getAccumulatorTypeOfAggregateFunction(
- userDefinedAggregateFunction: UserDefinedAggregateFunction[_, _],
- extractedType: DataType = null): DataType = {
-
+ userDefinedAggregateFunction: ImperativeAggregateFunction[_, _],
+ extractedType: DataType = null)
+ : DataType = {
val accType = userDefinedAggregateFunction.getAccumulatorType
if (accType != null) {
fromLegacyInfoToDataType(accType)
@@ -594,7 +596,7 @@ object UserDefinedFunctionUtils {
case ite: InvalidTypesException =>
throw new TableException(
"Cannot infer generic type of ${aggregateFunction.getClass}. " +
- "You can override UserDefinedAggregateFunction.getAccumulatorType() to specify " +
+ "You can override ImperativeAggregateFunction.getAccumulatorType() to specify " +
"the type.",
ite
)
@@ -604,21 +606,21 @@ object UserDefinedFunctionUtils {
}
/**
- * Internal method to extract a type from a [[UserDefinedAggregateFunction]]'s type parameters.
- *
- * @param aggregateFunction The [[UserDefinedAggregateFunction]] for which the type is extracted.
- * @param parameterTypePos The position of the type parameter for which the type is extracted.
+ * Internal method to extract a type from a [[ImperativeAggregateFunction]]'s type parameters.
*
+ * @param aggregateFunction The [[ImperativeAggregateFunction]] for which the type is extracted.
+ * @param parameterTypePos The position of the type parameter for which the type is extracted.
* @return The extracted type.
*/
@throws(classOf[InvalidTypesException])
private def extractTypeFromAggregateFunction(
- aggregateFunction: UserDefinedAggregateFunction[_, _],
- parameterTypePos: Int): DataType = {
+ aggregateFunction: ImperativeAggregateFunction[_, _],
+ parameterTypePos: Int)
+ : DataType = {
fromLegacyInfoToDataType(TypeExtractor.createTypeInfo(
aggregateFunction,
- classOf[UserDefinedAggregateFunction[_, _]],
+ classOf[ImperativeAggregateFunction[_, _]],
aggregateFunction.getClass,
parameterTypePos))
}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/utils/AggregateUtil.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/utils/AggregateUtil.scala
index 9145f4d..f24706d 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/utils/AggregateUtil.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/utils/AggregateUtil.scala
@@ -24,7 +24,7 @@ import org.apache.flink.table.data.{DecimalData, RowData, StringData, TimestampD
import org.apache.flink.table.dataview.MapViewTypeInfo
import org.apache.flink.table.expressions.ExpressionUtils.extractValue
import org.apache.flink.table.expressions._
-import org.apache.flink.table.functions.{AggregateFunction, TableAggregateFunction, UserDefinedAggregateFunction, UserDefinedFunction}
+import org.apache.flink.table.functions.{AggregateFunction, TableAggregateFunction, ImperativeAggregateFunction, UserDefinedFunction}
import org.apache.flink.table.planner.JLong
import org.apache.flink.table.planner.calcite.FlinkRelBuilder.PlannerNamedWindowProperty
import org.apache.flink.table.planner.calcite.{FlinkTypeFactory, FlinkTypeSystem}
@@ -314,7 +314,7 @@ object AggregateUtil extends Enumeration {
Array.empty[DataViewSpec],
fromLogicalTypeToDataType(a.getResultType.getLogicalType)
)
- case a: UserDefinedAggregateFunction[_, _] =>
+ case a: ImperativeAggregateFunction[_, _] =>
val (implicitAccType, implicitResultType) = call.getAggregation match {
case aggSqlFun: AggSqlFunction =>
(aggSqlFun.externalAccType, aggSqlFun.externalResultType)
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/utils/TableTestBase.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/utils/TableTestBase.scala
index c942911..9a8dd36 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/utils/TableTestBase.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/utils/TableTestBase.scala
@@ -990,7 +990,7 @@ class TestingTableEnvironment private(
def registerFunction[T: TypeInformation, ACC: TypeInformation](
name: String,
f: AggregateFunction[T, ACC]): Unit = {
- registerUserDefinedAggregateFunction(name, f)
+ registerImperativeAggregateFunction(name, f)
}
// just for testing, remove this method while
@@ -999,12 +999,12 @@ class TestingTableEnvironment private(
def registerFunction[T: TypeInformation, ACC: TypeInformation](
name: String,
f: TableAggregateFunction[T, ACC]): Unit = {
- registerUserDefinedAggregateFunction(name, f)
+ registerImperativeAggregateFunction(name, f)
}
- private def registerUserDefinedAggregateFunction[T: TypeInformation, ACC: TypeInformation](
+ private def registerImperativeAggregateFunction[T: TypeInformation, ACC: TypeInformation](
name: String,
- f: UserDefinedAggregateFunction[T, ACC]): Unit = {
+ f: ImperativeAggregateFunction[T, ACC]): Unit = {
val typeInfo = UserDefinedFunctionHelper
.getReturnTypeOfAggregateFunction(f, implicitly[TypeInformation[T]])
val accTypeInfo = UserDefinedFunctionHelper
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/internal/TableEnvImpl.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/internal/TableEnvImpl.scala
index 226b620..2ac2749 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/internal/TableEnvImpl.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/internal/TableEnvImpl.scala
@@ -32,7 +32,7 @@ import org.apache.flink.table.delegation.Parser
import org.apache.flink.table.expressions._
import org.apache.flink.table.expressions.resolver.lookups.TableReferenceLookup
import org.apache.flink.table.factories.{TableFactoryUtil, TableSinkFactoryContextImpl}
-import org.apache.flink.table.functions.{AggregateFunction, ScalarFunction, TableFunction, UserDefinedAggregateFunction, _}
+import org.apache.flink.table.functions.{AggregateFunction, ScalarFunction, TableFunction, ImperativeAggregateFunction, _}
import org.apache.flink.table.module.{Module, ModuleManager}
import org.apache.flink.table.operations.ddl._
import org.apache.flink.table.operations.utils.OperationTreeBuilder
@@ -244,7 +244,7 @@ abstract class TableEnvImpl(
*/
private[flink] def registerAggregateFunctionInternal[T: TypeInformation, ACC: TypeInformation](
name: String,
- function: UserDefinedAggregateFunction[T, ACC])
+ function: ImperativeAggregateFunction[T, ACC])
: Unit = {
val resultTypeInfo: TypeInformation[T] = UserDefinedFunctionHelper
.getReturnTypeOfAggregateFunction(
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/codegen/AggregationCodeGenerator.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/codegen/AggregationCodeGenerator.scala
index 2de3f0f..426cdc5 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/codegen/AggregationCodeGenerator.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/codegen/AggregationCodeGenerator.scala
@@ -17,27 +17,20 @@
*/
package org.apache.flink.table.codegen
-import java.lang.reflect.Modifier
-import java.lang.{Iterable => JIterable}
-import java.util.{List => JList}
-
-import org.apache.calcite.rel.`type`.RelDataType
-import org.apache.calcite.rel.core.AggregateCall
-import org.apache.calcite.rex.RexLiteral
import org.apache.flink.api.common.state.{ListStateDescriptor, MapStateDescriptor, State, StateDescriptor}
import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.api.java.typeutils.RowTypeInfo
import org.apache.flink.api.java.typeutils.TypeExtractionUtils.{extractTypeArgument, getRawClass}
-import org.apache.flink.table.api.{TableConfig, ValidationException}
import org.apache.flink.table.api.dataview._
+import org.apache.flink.table.api.{TableConfig, ValidationException}
import org.apache.flink.table.calcite.FlinkTypeFactory
import org.apache.flink.table.codegen.CodeGenUtils.{newName, reflectiveFieldWriteAccess}
import org.apache.flink.table.codegen.Indenter.toISC
import org.apache.flink.table.dataview.{StateListView, StateMapView}
-import org.apache.flink.table.functions.{TableAggregateFunction, UserDefinedAggregateFunction, UserDefinedFunction}
import org.apache.flink.table.functions.aggfunctions.DistinctAccumulator
-import org.apache.flink.table.functions.utils.{AggSqlFunction, UserDefinedFunctionUtils}
import org.apache.flink.table.functions.utils.UserDefinedFunctionUtils.{getUserDefinedMethod, signatureToString}
+import org.apache.flink.table.functions.utils.{AggSqlFunction, UserDefinedFunctionUtils}
+import org.apache.flink.table.functions.{ImperativeAggregateFunction, TableAggregateFunction, UserDefinedFunction}
import org.apache.flink.table.runtime.CRowWrappingCollector
import org.apache.flink.table.runtime.aggregate.AggregateUtil.CalcitePair
import org.apache.flink.table.runtime.aggregate.{AggregateUtil, GeneratedAggregations, GeneratedTableAggregations, SingleElementIterable}
@@ -45,6 +38,14 @@ import org.apache.flink.table.utils.EncodingUtils
import org.apache.flink.types.Row
import org.apache.flink.util.Collector
+import org.apache.calcite.rel.`type`.RelDataType
+import org.apache.calcite.rel.core.AggregateCall
+import org.apache.calcite.rex.RexLiteral
+
+import java.lang.reflect.Modifier
+import java.lang.{Iterable => JIterable}
+import java.util.{List => JList}
+
import scala.collection.JavaConversions._
import scala.collection.mutable
@@ -82,25 +83,25 @@ import scala.collection.mutable
* @param accConfig Data view specification for accumulators
*/
class AggregationCodeGenerator(
- config: TableConfig,
- nullableInput: Boolean,
- inputTypeInfo: TypeInformation[_ <: Any],
- constants: Option[Seq[RexLiteral]],
- classNamePrefix: String,
- physicalInputTypes: Seq[TypeInformation[_]],
- aggregates: Array[UserDefinedAggregateFunction[_ <: Any, _ <: Any]],
- aggFields: Array[Array[Int]],
- aggMapping: Array[Int],
- distinctAccMapping: Array[(Integer, JList[Integer])],
- isStateBackedDataViews: Boolean,
- partialResults: Boolean,
- fwdMapping: Array[Int],
- mergeMapping: Option[Array[Int]],
- outputArity: Int,
- needRetract: Boolean,
- needMerge: Boolean,
- needReset: Boolean,
- accConfig: Option[Array[Seq[DataViewSpec[_]]]])
+ config: TableConfig,
+ nullableInput: Boolean,
+ inputTypeInfo: TypeInformation[_ <: Any],
+ constants: Option[Seq[RexLiteral]],
+ classNamePrefix: String,
+ physicalInputTypes: Seq[TypeInformation[_]],
+ aggregates: Array[ImperativeAggregateFunction[_ <: Any, _ <: Any]],
+ aggFields: Array[Array[Int]],
+ aggMapping: Array[Int],
+ distinctAccMapping: Array[(Integer, JList[Integer])],
+ isStateBackedDataViews: Boolean,
+ partialResults: Boolean,
+ fwdMapping: Array[Int],
+ mergeMapping: Option[Array[Int]],
+ outputArity: Int,
+ needRetract: Boolean,
+ needMerge: Boolean,
+ needReset: Boolean,
+ accConfig: Option[Array[Seq[DataViewSpec[_]]]])
extends CodeGenerator(config, nullableInput, inputTypeInfo) {
// set of statements for cleanup dataview that will be added only once
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/codegen/MatchCodeGenerator.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/codegen/MatchCodeGenerator.scala
index f248188..569a7e7 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/codegen/MatchCodeGenerator.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/codegen/MatchCodeGenerator.scala
@@ -18,12 +18,6 @@
package org.apache.flink.table.codegen
-import java.lang.{Long => JLong}
-import java.util
-import org.apache.calcite.rel.`type`.RelDataType
-import org.apache.calcite.rex._
-import org.apache.calcite.sql.SqlAggFunction
-import org.apache.calcite.sql.fun.SqlStdOperatorTable._
import org.apache.flink.api.common.functions._
import org.apache.flink.api.common.typeinfo.{SqlTimeTypeInfo, TypeInformation}
import org.apache.flink.cep.functions.PatternProcessFunction
@@ -32,23 +26,30 @@ import org.apache.flink.configuration.Configuration
import org.apache.flink.table.api.dataview.DataViewSpec
import org.apache.flink.table.api.{TableConfig, TableException}
import org.apache.flink.table.calcite.FlinkTypeFactory
+import org.apache.flink.table.catalog.BasicOperatorTable.{MATCH_PROCTIME, MATCH_ROWTIME}
import org.apache.flink.table.codegen.CodeGenUtils.{boxedTypeTermForTypeInfo, newName, primitiveDefaultValue, primitiveTypeTermForTypeInfo}
import org.apache.flink.table.codegen.GeneratedExpression.{NEVER_NULL, NO_CODE}
import org.apache.flink.table.codegen.Indenter.toISC
-import org.apache.flink.table.functions.UserDefinedAggregateFunction
+import org.apache.flink.table.functions.ImperativeAggregateFunction
import org.apache.flink.table.plan.schema.RowSchema
import org.apache.flink.table.runtime.`match`.{IterativeConditionRunner, PatternProcessFunctionRunner}
import org.apache.flink.table.runtime.aggregate.AggregateUtil
import org.apache.flink.table.typeutils.TimeIndicatorTypeInfo
import org.apache.flink.table.util.MatchUtil.{ALL_PATTERN_VARIABLE, AggregationPatternVariableFinder}
import org.apache.flink.table.utils.EncodingUtils
-import org.apache.flink.table.catalog.BasicOperatorTable.{MATCH_PROCTIME, MATCH_ROWTIME}
import org.apache.flink.types.Row
import org.apache.flink.util.Collector
import org.apache.flink.util.MathUtils.checkedDownCast
+import org.apache.calcite.rel.`type`.RelDataType
+import org.apache.calcite.rex._
+import org.apache.calcite.sql.SqlAggFunction
+import org.apache.calcite.sql.fun.SqlStdOperatorTable._
import org.apache.calcite.util.ImmutableBitSet
+import java.lang.{Long => JLong}
+import java.util
+
import scala.collection.JavaConverters._
import scala.collection.mutable
@@ -890,7 +891,7 @@ class MatchCodeGenerator(
)
private case class SingleAggCall(
- aggFunction: UserDefinedAggregateFunction[_, _],
+ aggFunction: ImperativeAggregateFunction[_, _],
inputIndices: Array[Int],
dataViews: Seq[DataViewSpec[_]],
distinctAccIndex: Int
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/aggregations.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/aggregations.scala
index f6d0725..af02d6f 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/aggregations.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/aggregations.scala
@@ -17,20 +17,20 @@
*/
package org.apache.flink.table.expressions
+import org.apache.flink.api.common.typeinfo.{BasicTypeInfo, TypeInformation}
+import org.apache.flink.api.java.typeutils.MultisetTypeInfo
+import org.apache.flink.table.calcite.FlinkTypeFactory
+import org.apache.flink.table.functions.ImperativeAggregateFunction
+import org.apache.flink.table.functions.utils.AggSqlFunction
+import org.apache.flink.table.functions.utils.UserDefinedFunctionUtils._
+import org.apache.flink.table.typeutils.TypeCheckUtils
+import org.apache.flink.table.validate.{ValidationFailure, ValidationResult, ValidationSuccess}
+
import org.apache.calcite.rex.RexNode
import org.apache.calcite.sql.SqlAggFunction
import org.apache.calcite.sql.fun._
import org.apache.calcite.tools.RelBuilder
import org.apache.calcite.tools.RelBuilder.AggCall
-import org.apache.flink.api.common.typeinfo.TypeInformation
-import org.apache.flink.table.functions.UserDefinedAggregateFunction
-import org.apache.flink.table.functions.utils.AggSqlFunction
-import org.apache.flink.table.typeutils.TypeCheckUtils
-import org.apache.flink.api.common.typeinfo.BasicTypeInfo
-import org.apache.flink.api.java.typeutils.MultisetTypeInfo
-import org.apache.flink.table.calcite.FlinkTypeFactory
-import org.apache.flink.table.functions.utils.UserDefinedFunctionUtils._
-import org.apache.flink.table.validate.{ValidationFailure, ValidationResult, ValidationSuccess}
abstract sealed class Aggregation extends PlannerExpression {
@@ -361,7 +361,7 @@ case class VarSamp(child: PlannerExpression) extends Aggregation {
* Expression for calling a user-defined (table)aggregate function.
*/
case class AggFunctionCall(
- aggregateFunction: UserDefinedAggregateFunction[_, _],
+ aggregateFunction: ImperativeAggregateFunction[_, _],
resultTypeInfo: TypeInformation[_],
accTypeInfo: TypeInformation[_],
args: Seq[PlannerExpression])
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/functions/utils/AggSqlFunction.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/functions/utils/AggSqlFunction.scala
index de71825..ce80a4f 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/functions/utils/AggSqlFunction.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/functions/utils/AggSqlFunction.scala
@@ -18,21 +18,22 @@
package org.apache.flink.table.functions.utils
-import java.util
+import org.apache.flink.api.common.typeinfo._
+import org.apache.flink.table.api.ValidationException
+import org.apache.flink.table.calcite.FlinkTypeFactory
+import org.apache.flink.table.functions.utils.AggSqlFunction.{createOperandTypeChecker, createOperandTypeInference, createReturnTypeInference}
+import org.apache.flink.table.functions.utils.UserDefinedFunctionUtils._
+import org.apache.flink.table.functions.{AggregateFunction, FunctionRequirement, ImperativeAggregateFunction, TableAggregateFunction}
import org.apache.calcite.rel.`type`.RelDataType
import org.apache.calcite.sql._
-import org.apache.calcite.sql.`type`._
import org.apache.calcite.sql.`type`.SqlOperandTypeChecker.Consistency
+import org.apache.calcite.sql.`type`._
import org.apache.calcite.sql.parser.SqlParserPos
import org.apache.calcite.sql.validate.SqlUserDefinedAggFunction
import org.apache.calcite.util.Optionality
-import org.apache.flink.api.common.typeinfo._
-import org.apache.flink.table.api.ValidationException
-import org.apache.flink.table.calcite.FlinkTypeFactory
-import org.apache.flink.table.functions.{AggregateFunction, FunctionRequirement, TableAggregateFunction, UserDefinedAggregateFunction}
-import org.apache.flink.table.functions.utils.AggSqlFunction.{createOperandTypeChecker, createOperandTypeInference, createReturnTypeInference}
-import org.apache.flink.table.functions.utils.UserDefinedFunctionUtils._
+
+import java.util
/**
* Calcite wrapper for user-defined aggregate functions. Currently, the aggregate function can be
@@ -48,7 +49,7 @@ import org.apache.flink.table.functions.utils.UserDefinedFunctionUtils._
class AggSqlFunction(
name: String,
displayName: String,
- aggregateFunction: UserDefinedAggregateFunction[_, _],
+ aggregateFunction: ImperativeAggregateFunction[_, _],
val returnType: TypeInformation[_],
val accType: TypeInformation[_],
typeFactory: FlinkTypeFactory,
@@ -67,7 +68,7 @@ class AggSqlFunction(
typeFactory
) {
- def getFunction: UserDefinedAggregateFunction[_, _] = aggregateFunction
+ def getFunction: ImperativeAggregateFunction[_, _] = aggregateFunction
override def isDeterministic: Boolean = aggregateFunction.isDeterministic
@@ -81,10 +82,11 @@ object AggSqlFunction {
def apply(
name: String,
displayName: String,
- aggregateFunction: UserDefinedAggregateFunction[_, _],
+ aggregateFunction: ImperativeAggregateFunction[_, _],
returnType: TypeInformation[_],
accType: TypeInformation[_],
- typeFactory: FlinkTypeFactory): AggSqlFunction = {
+ typeFactory: FlinkTypeFactory)
+ : AggSqlFunction = {
val requiresOver = aggregateFunction match {
case a: AggregateFunction[_, _] =>
@@ -103,9 +105,9 @@ object AggSqlFunction {
}
private[flink] def createOperandTypeInference(
- aggregateFunction: UserDefinedAggregateFunction[_, _],
- typeFactory: FlinkTypeFactory,
- accType: TypeInformation[_])
+ aggregateFunction: ImperativeAggregateFunction[_, _],
+ typeFactory: FlinkTypeFactory,
+ accType: TypeInformation[_])
: SqlOperandTypeInference = {
/**
* Operand type inference based on [[AggregateFunction]] given information.
@@ -157,7 +159,7 @@ object AggSqlFunction {
}
private[flink] def createOperandTypeChecker(
- aggregateFunction: UserDefinedAggregateFunction[_, _],
+ aggregateFunction: ImperativeAggregateFunction[_, _],
accType: TypeInformation[_])
: SqlOperandTypeChecker = {
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/functions/utils/UserDefinedFunctionUtils.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/functions/utils/UserDefinedFunctionUtils.scala
index cdb2479..20137db 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/functions/utils/UserDefinedFunctionUtils.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/functions/utils/UserDefinedFunctionUtils.scala
@@ -67,11 +67,11 @@ object UserDefinedFunctionUtils {
* of [[TypeInformation]]. Elements of the signature can be null (act as a wildcard).
*/
def getAccumulateMethodSignature(
- function: UserDefinedAggregateFunction[_, _],
- signature: Seq[TypeInformation[_]])
+ function: ImperativeAggregateFunction[_, _],
+ signature: Seq[TypeInformation[_]])
: Option[Array[Class[_]]] = {
val accType = TypeExtractor.createTypeInfo(
- function, classOf[UserDefinedAggregateFunction[_, _]], function.getClass, 1)
+ function, classOf[ImperativeAggregateFunction[_, _]], function.getClass, 1)
val input = (Array(accType) ++ signature).toSeq
getUserDefinedMethod(
function,
@@ -292,11 +292,11 @@ object UserDefinedFunctionUtils {
def createAggregateSqlFunction(
name: String,
displayName: String,
- aggFunction: UserDefinedAggregateFunction[_, _],
+ aggFunction: ImperativeAggregateFunction[_, _],
resultType: TypeInformation[_],
accTypeInfo: TypeInformation[_],
typeFactory: FlinkTypeFactory)
- : SqlFunction = {
+ : SqlFunction = {
//check if a qualified accumulate method exists before create Sql function
checkAndExtractMethods(aggFunction, "accumulate")
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/runtime/aggregate/AggregateUtil.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/runtime/aggregate/AggregateUtil.scala
index 21625cb..0d47ca9 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/runtime/aggregate/AggregateUtil.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/runtime/aggregate/AggregateUtil.scala
@@ -17,16 +17,6 @@
*/
package org.apache.flink.table.runtime.aggregate
-import java.util
-import java.util.{ArrayList => JArrayList, List => JList}
-
-import org.apache.calcite.rel.`type`._
-import org.apache.calcite.rel.core.AggregateCall
-import org.apache.calcite.rex.RexLiteral
-import org.apache.calcite.sql.`type`.SqlTypeName
-import org.apache.calcite.sql.`type`.SqlTypeName._
-import org.apache.calcite.sql.fun._
-import org.apache.calcite.sql.{SqlAggFunction, SqlKind}
import org.apache.flink.api.common.functions.{MapFunction, RichGroupReduceFunction, AggregateFunction => DataStreamAggFunction, _}
import org.apache.flink.api.common.typeinfo.{BasicTypeInfo, TypeInformation, Types}
import org.apache.flink.api.java.typeutils.RowTypeInfo
@@ -43,13 +33,24 @@ import org.apache.flink.table.expressions._
import org.apache.flink.table.functions.aggfunctions._
import org.apache.flink.table.functions.utils.AggSqlFunction
import org.apache.flink.table.functions.utils.UserDefinedFunctionUtils._
-import org.apache.flink.table.functions.{AggregateFunction, TableAggregateFunction, UserDefinedAggregateFunction, UserDefinedFunctionHelper}
+import org.apache.flink.table.functions.{AggregateFunction, ImperativeAggregateFunction, TableAggregateFunction, UserDefinedFunctionHelper}
import org.apache.flink.table.plan.logical._
import org.apache.flink.table.runtime.types.{CRow, CRowTypeInfo}
import org.apache.flink.table.typeutils.TimeIntervalTypeInfo
import org.apache.flink.table.typeutils.TypeCheckUtils._
import org.apache.flink.types.Row
+import org.apache.calcite.rel.`type`._
+import org.apache.calcite.rel.core.AggregateCall
+import org.apache.calcite.rex.RexLiteral
+import org.apache.calcite.sql.`type`.SqlTypeName
+import org.apache.calcite.sql.`type`.SqlTypeName._
+import org.apache.calcite.sql.fun._
+import org.apache.calcite.sql.{SqlAggFunction, SqlKind}
+
+import java.util
+import java.util.{ArrayList => JArrayList, List => JList}
+
import scala.collection.JavaConversions._
import scala.collection.JavaConverters._
import scala.collection.mutable
@@ -1246,7 +1247,7 @@ object AggregateUtil {
* Return true if all aggregates can be partially merged. False otherwise.
*/
private[flink] def doAllSupportPartialMerge(
- aggregateList: Array[UserDefinedAggregateFunction[_ <: Any, _ <: Any]]): Boolean = {
+ aggregateList: Array[ImperativeAggregateFunction[_ <: Any, _ <: Any]]): Boolean = {
aggregateList.forall(ifMethodExistInFunction("merge", _))
}
@@ -1324,7 +1325,7 @@ object AggregateUtil {
private val aggregates: Seq[(AggregateCallMetadata, Array[Int])],
private val distinctAccTypesWithSpecs: Seq[(TypeInformation[_], Seq[DataViewSpec[_]])]) {
- def getAggregateFunctions: Array[UserDefinedAggregateFunction[_, _]] = {
+ def getAggregateFunctions: Array[ImperativeAggregateFunction[_, _]] = {
aggregates.map(_._1.aggregateFunction).toArray
}
@@ -1370,7 +1371,7 @@ object AggregateUtil {
* function.
*/
private[flink] case class AggregateCallMetadata(
- aggregateFunction: UserDefinedAggregateFunction[_, _],
+ aggregateFunction: ImperativeAggregateFunction[_, _],
accumulatorType: TypeInformation[_],
accumulatorSpecs: Seq[DataViewSpec[_]],
distinctAccIndex: Int
@@ -1414,7 +1415,7 @@ object AggregateUtil {
// store the aggregate fields of each aggregate function, by the same order of aggregates.
// create aggregate function instances by function type and aggregate field data type.
- val aggregate: UserDefinedAggregateFunction[_, _] = createFlinkAggFunction(
+ val aggregate: ImperativeAggregateFunction[_, _] = createFlinkAggFunction(
aggregateFunction,
needRetraction,
aggregateInputTypes,
@@ -1542,7 +1543,7 @@ object AggregateUtil {
needRetraction: Boolean,
inputDataType: Seq[RelDataType],
tableConfig: TableConfig)
- : UserDefinedAggregateFunction[_ <: Any, _ <: Any] = {
+ : ImperativeAggregateFunction[_ <: Any, _ <: Any] = {
lazy val outputType = inputDataType.get(0)
lazy val outputTypeName = if (inputDataType.isEmpty) {
@@ -1796,10 +1797,11 @@ object AggregateUtil {
private def createRowTypeForKeysAndAggregates(
groupings: Array[Int],
- aggregates: Array[UserDefinedAggregateFunction[_, _]],
+ aggregates: Array[ImperativeAggregateFunction[_, _]],
aggTypes: Array[TypeInformation[_]],
inputType: RelDataType,
- windowKeyTypes: Option[Array[TypeInformation[_]]] = None): RowTypeInfo = {
+ windowKeyTypes: Option[Array[TypeInformation[_]]] = None)
+ : RowTypeInfo = {
// get the field data types of group keys.
val groupingTypes: Seq[TypeInformation[_]] =
@@ -1864,7 +1866,7 @@ object AggregateUtil {
}
private[flink] def containsTableAggregateFunction(
- aggregates: Seq[UserDefinedAggregateFunction[_, _]])
+ aggregates: Seq[ImperativeAggregateFunction[_, _]])
: Boolean = {
aggregates.exists(_.isInstanceOf[TableAggregateFunction[_, _]])
}