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 2019/12/18 09:35:25 UTC
[flink] branch master updated (6241894 -> 524b443)
This is an automated email from the ASF dual-hosted git repository.
twalthr pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git.
from 6241894 [FLINK-15291][table-common] Rename WatermarkSepc#getWatermarkExpressionString to getWatermarkExpr (#10614)
new 09bcde2 [hotfix][table] Rename UserFunctionsTypeHelper
new 91629d6 [FLINK-12283][table] Relax UDF constraints by using the ClosureCleaner
new 7fb9f9b [hotfix][table] Update function tests to use testing infrastructure
new 524b443 [hotfix][table] Use testing infrastacture in tests
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../client/gateway/local/ExecutionContext.java | 27 +-
.../java/internal/StreamTableEnvironmentImpl.java | 14 +-
.../internal/StreamTableEnvironmentImplTest.java | 5 +-
.../table/api/internal/TableEnvironmentImpl.java | 20 +-
.../flink/table/catalog/FunctionCatalog.java | 38 +-
.../expressions/resolver/ExpressionResolver.java | 18 +-
.../resolver/rules/ResolveCallByArgumentsRule.java | 50 ++-
.../expressions/resolver/rules/ResolverRule.java | 6 +
...eHelper.java => UserDefinedFunctionHelper.java} | 50 ++-
.../operations/utils/OperationTreeBuilder.java | 36 +-
.../flink/table/catalog/FunctionCatalogTest.java | 2 +
.../flink/table/utils/TableEnvironmentMock.java | 12 +-
.../internal/StreamTableEnvironmentImpl.scala | 14 +-
.../internal/StreamTableEnvironmentImplTest.scala | 5 +-
.../org/apache/flink/table/api/expressionDsl.scala | 8 +-
.../functions/AggregateFunctionDefinition.java | 4 +
.../table/functions/ScalarFunctionDefinition.java | 4 +
.../TableAggregateFunctionDefinition.java | 4 +
.../table/functions/TableFunctionDefinition.java | 4 +
.../flink/table/functions/UserDefinedFunction.java | 2 +-
.../expressions/CallExpressionResolver.java | 1 +
.../planner/codegen/CodeGeneratorContext.scala | 2 +-
.../flink/table/planner/expressions/call.scala | 4 -
.../table/planner/functions/FunctionTestBase.java | 362 -----------------
.../operations/SqlToOperationConverterTest.java | 5 +-
.../planner/runtime/batch/sql/FunctionITCase.java | 39 --
.../planner/runtime/stream/sql/FunctionITCase.java | 390 ++++++++++++++++--
.../codegen/WatermarkGeneratorCodeGenTest.scala | 5 +-
.../metadata/AggCallSelectivityEstimatorTest.scala | 3 +-
.../plan/metadata/FlinkRelMdHandlerTestBase.scala | 8 +-
.../plan/metadata/SelectivityEstimatorTest.scala | 2 +-
.../planner/plan/utils/RexNodeExtractorTest.scala | 7 +-
.../planner/runtime/stream/table/CalcITCase.scala | 77 +++-
.../flink/table/planner/utils/TableTestBase.scala | 8 +-
.../flink/table/api/internal/TableEnvImpl.scala | 21 +-
.../apache/flink/table/codegen/CodeGenerator.scala | 2 +-
.../org/apache/flink/table/expressions/call.scala | 4 -
.../PushFilterIntoTableSourceScanRule.scala | 4 +-
.../table/runtime/aggregate/AggregateUtil.scala | 4 +-
.../flink/table/functions/FunctionTestBase.java | 436 ---------------------
.../table/runtime/batch/sql/FunctionITCase.java | 43 --
.../table/runtime/stream/sql/FunctionITCase.java | 431 +++++++++++++++++++-
.../table/sqlexec/SqlToOperationConverterTest.java | 5 +-
.../api/stream/StreamTableEnvironmentTest.scala | 2 +-
.../flink/table/api/stream/sql/AggregateTest.scala | 5 +-
.../flink/table/catalog/CatalogTableITCase.scala | 4 +-
.../flink/table/plan/RexProgramExtractorTest.scala | 2 +
.../batch/sql/PartitionableSinkITCase.scala | 11 +-
.../table/runtime/harness/HarnessTestBase.scala | 6 +-
.../table/runtime/stream/table/CalcITCase.scala | 88 +++++
.../apache/flink/table/utils/TableTestBase.scala | 2 +-
51 files changed, 1230 insertions(+), 1076 deletions(-)
rename flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/functions/{UserFunctionsTypeHelper.java => UserDefinedFunctionHelper.java} (77%)
delete mode 100644 flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/functions/FunctionTestBase.java
delete mode 100644 flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/FunctionITCase.java
delete mode 100644 flink-table/flink-table-planner/src/test/java/org/apache/flink/table/functions/FunctionTestBase.java
delete mode 100644 flink-table/flink-table-planner/src/test/java/org/apache/flink/table/runtime/batch/sql/FunctionITCase.java
[flink] 01/04: [hotfix][table] Rename UserFunctionsTypeHelper
Posted by tw...@apache.org.
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 09bcde276f25da11d5458b3f5ae5580143f2ead9
Author: Timo Walther <tw...@apache.org>
AuthorDate: Mon Dec 9 14:38:34 2019 +0100
[hotfix][table] Rename UserFunctionsTypeHelper
---
.../java/internal/StreamTableEnvironmentImpl.java | 12 ++++++------
.../table/api/internal/TableEnvironmentImpl.java | 14 +++++++-------
.../flink/table/catalog/FunctionCatalog.java | 22 +++++++++++-----------
...eHelper.java => UserDefinedFunctionHelper.java} | 10 +++++-----
.../internal/StreamTableEnvironmentImpl.scala | 12 ++++++------
.../org/apache/flink/table/api/expressionDsl.scala | 8 ++++----
.../flink/table/planner/expressions/call.scala | 4 ++--
.../plan/metadata/FlinkRelMdHandlerTestBase.scala | 6 +++---
.../flink/table/planner/utils/TableTestBase.scala | 6 +++---
.../flink/table/api/internal/TableEnvImpl.scala | 18 +++++++++---------
.../org/apache/flink/table/expressions/call.scala | 4 ++--
.../table/runtime/aggregate/AggregateUtil.scala | 4 ++--
.../table/runtime/harness/HarnessTestBase.scala | 6 +++---
13 files changed, 63 insertions(+), 63 deletions(-)
diff --git a/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/api/java/internal/StreamTableEnvironmentImpl.java b/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/api/java/internal/StreamTableEnvironmentImpl.java
index 1e42070..760a934 100644
--- a/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/api/java/internal/StreamTableEnvironmentImpl.java
+++ b/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/api/java/internal/StreamTableEnvironmentImpl.java
@@ -53,7 +53,7 @@ import org.apache.flink.table.factories.ComponentFactoryService;
import org.apache.flink.table.functions.AggregateFunction;
import org.apache.flink.table.functions.TableAggregateFunction;
import org.apache.flink.table.functions.TableFunction;
-import org.apache.flink.table.functions.UserFunctionsTypeHelper;
+import org.apache.flink.table.functions.UserDefinedFunctionHelper;
import org.apache.flink.table.module.ModuleManager;
import org.apache.flink.table.operations.JavaDataStreamQueryOperation;
import org.apache.flink.table.operations.OutputConversionModifyOperation;
@@ -152,7 +152,7 @@ public final class StreamTableEnvironmentImpl extends TableEnvironmentImpl imple
@Override
public <T> void registerFunction(String name, TableFunction<T> tableFunction) {
- TypeInformation<T> typeInfo = UserFunctionsTypeHelper.getReturnTypeOfTableFunction(tableFunction);
+ TypeInformation<T> typeInfo = UserDefinedFunctionHelper.getReturnTypeOfTableFunction(tableFunction);
functionCatalog.registerTempSystemTableFunction(
name,
@@ -163,8 +163,8 @@ public final class StreamTableEnvironmentImpl extends TableEnvironmentImpl imple
@Override
public <T, ACC> void registerFunction(String name, AggregateFunction<T, ACC> aggregateFunction) {
- TypeInformation<T> typeInfo = UserFunctionsTypeHelper.getReturnTypeOfAggregateFunction(aggregateFunction);
- TypeInformation<ACC> accTypeInfo = UserFunctionsTypeHelper
+ TypeInformation<T> typeInfo = UserDefinedFunctionHelper.getReturnTypeOfAggregateFunction(aggregateFunction);
+ TypeInformation<ACC> accTypeInfo = UserDefinedFunctionHelper
.getAccumulatorTypeOfAggregateFunction(aggregateFunction);
functionCatalog.registerTempSystemAggregateFunction(
@@ -177,9 +177,9 @@ public final class StreamTableEnvironmentImpl extends TableEnvironmentImpl imple
@Override
public <T, ACC> void registerFunction(String name, TableAggregateFunction<T, ACC> tableAggregateFunction) {
- TypeInformation<T> typeInfo = UserFunctionsTypeHelper.getReturnTypeOfAggregateFunction(
+ TypeInformation<T> typeInfo = UserDefinedFunctionHelper.getReturnTypeOfAggregateFunction(
tableAggregateFunction);
- TypeInformation<ACC> accTypeInfo = UserFunctionsTypeHelper
+ TypeInformation<ACC> accTypeInfo = UserDefinedFunctionHelper
.getAccumulatorTypeOfAggregateFunction(tableAggregateFunction);
functionCatalog.registerTempSystemAggregateFunction(
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
index 838edbb..920124e 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
@@ -65,7 +65,7 @@ import org.apache.flink.table.functions.ScalarFunction;
import org.apache.flink.table.functions.ScalarFunctionDefinition;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.table.functions.TableFunctionDefinition;
-import org.apache.flink.table.functions.UserFunctionsTypeHelper;
+import org.apache.flink.table.functions.UserDefinedFunctionHelper;
import org.apache.flink.table.module.Module;
import org.apache.flink.table.module.ModuleManager;
import org.apache.flink.table.operations.CatalogQueryOperation;
@@ -875,9 +875,9 @@ public class TableEnvironmentImpl implements TableEnvironment {
AggregateFunctionDefinition aggregateFunctionDefinition = (AggregateFunctionDefinition) functionDefinition;
AggregateFunction<T, ACC > aggregateFunction =
(AggregateFunction<T, ACC >) aggregateFunctionDefinition.getAggregateFunction();
- TypeInformation<T> typeInfo = UserFunctionsTypeHelper
+ TypeInformation<T> typeInfo = UserDefinedFunctionHelper
.getReturnTypeOfAggregateFunction(aggregateFunction);
- TypeInformation<ACC> accTypeInfo = UserFunctionsTypeHelper
+ TypeInformation<ACC> accTypeInfo = UserDefinedFunctionHelper
.getAccumulatorTypeOfAggregateFunction(aggregateFunction);
functionCatalog.registerTempCatalogAggregateFunction(
@@ -888,7 +888,7 @@ public class TableEnvironmentImpl implements TableEnvironment {
} else if (functionDefinition instanceof TableFunctionDefinition) {
TableFunctionDefinition tableFunctionDefinition = (TableFunctionDefinition) functionDefinition;
TableFunction<T> tableFunction = (TableFunction<T>) tableFunctionDefinition.getTableFunction();
- TypeInformation<T> typeInfo = UserFunctionsTypeHelper
+ TypeInformation<T> typeInfo = UserDefinedFunctionHelper
.getReturnTypeOfTableFunction(tableFunction);
functionCatalog.registerTempCatalogTableFunction(
functionIdentifier,
@@ -908,9 +908,9 @@ public class TableEnvironmentImpl implements TableEnvironment {
AggregateFunctionDefinition aggregateFunctionDefinition = (AggregateFunctionDefinition) functionDefinition;
AggregateFunction<T, ACC > aggregateFunction =
(AggregateFunction<T, ACC >) aggregateFunctionDefinition.getAggregateFunction();
- TypeInformation<T> typeInfo = UserFunctionsTypeHelper
+ TypeInformation<T> typeInfo = UserDefinedFunctionHelper
.getReturnTypeOfAggregateFunction(aggregateFunction);
- TypeInformation<ACC> accTypeInfo = UserFunctionsTypeHelper
+ TypeInformation<ACC> accTypeInfo = UserDefinedFunctionHelper
.getAccumulatorTypeOfAggregateFunction(aggregateFunction);
functionCatalog.registerTempSystemAggregateFunction(
functionName,
@@ -921,7 +921,7 @@ public class TableEnvironmentImpl implements TableEnvironment {
} else if (functionDefinition instanceof TableFunctionDefinition) {
TableFunctionDefinition tableFunctionDefinition = (TableFunctionDefinition) functionDefinition;
TableFunction<T> tableFunction = (TableFunction<T>) tableFunctionDefinition.getTableFunction();
- TypeInformation<T> typeInfo = UserFunctionsTypeHelper
+ TypeInformation<T> typeInfo = UserDefinedFunctionHelper
.getReturnTypeOfTableFunction(tableFunction);
functionCatalog.registerTempSystemTableFunction(
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 6658fb2..be94b0d 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
@@ -37,7 +37,7 @@ 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.UserFunctionsTypeHelper;
+import org.apache.flink.table.functions.UserDefinedFunctionHelper;
import org.apache.flink.table.module.ModuleManager;
import org.apache.flink.util.Preconditions;
@@ -76,7 +76,7 @@ public class FunctionCatalog implements FunctionLookup {
}
public void registerTempSystemScalarFunction(String name, ScalarFunction function) {
- UserFunctionsTypeHelper.validateInstantiation(function.getClass());
+ UserDefinedFunctionHelper.validateInstantiation(function.getClass());
registerTempSystemFunction(
name,
new ScalarFunctionDefinition(name, function)
@@ -88,9 +88,9 @@ public class FunctionCatalog implements FunctionLookup {
TableFunction<T> function,
TypeInformation<T> resultType) {
// check if class not Scala object
- UserFunctionsTypeHelper.validateNotSingleton(function.getClass());
+ UserDefinedFunctionHelper.validateNotSingleton(function.getClass());
// check if class could be instantiated
- UserFunctionsTypeHelper.validateInstantiation(function.getClass());
+ UserDefinedFunctionHelper.validateInstantiation(function.getClass());
registerTempSystemFunction(
name,
@@ -107,9 +107,9 @@ public class FunctionCatalog implements FunctionLookup {
TypeInformation<T> resultType,
TypeInformation<ACC> accType) {
// check if class not Scala object
- UserFunctionsTypeHelper.validateNotSingleton(function.getClass());
+ UserDefinedFunctionHelper.validateNotSingleton(function.getClass());
// check if class could be instantiated
- UserFunctionsTypeHelper.validateInstantiation(function.getClass());
+ UserDefinedFunctionHelper.validateInstantiation(function.getClass());
final FunctionDefinition definition;
if (function instanceof AggregateFunction) {
@@ -135,7 +135,7 @@ public class FunctionCatalog implements FunctionLookup {
}
public void registerTempCatalogScalarFunction(ObjectIdentifier oi, ScalarFunction function) {
- UserFunctionsTypeHelper.validateInstantiation(function.getClass());
+ UserDefinedFunctionHelper.validateInstantiation(function.getClass());
registerTempCatalogFunction(
oi,
new ScalarFunctionDefinition(oi.getObjectName(), function)
@@ -147,9 +147,9 @@ public class FunctionCatalog implements FunctionLookup {
TableFunction<T> function,
TypeInformation<T> resultType) {
// check if class not Scala object
- UserFunctionsTypeHelper.validateNotSingleton(function.getClass());
+ UserDefinedFunctionHelper.validateNotSingleton(function.getClass());
// check if class could be instantiated
- UserFunctionsTypeHelper.validateInstantiation(function.getClass());
+ UserDefinedFunctionHelper.validateInstantiation(function.getClass());
registerTempCatalogFunction(
oi,
@@ -166,9 +166,9 @@ public class FunctionCatalog implements FunctionLookup {
TypeInformation<T> resultType,
TypeInformation<ACC> accType) {
// check if class not Scala object
- UserFunctionsTypeHelper.validateNotSingleton(function.getClass());
+ UserDefinedFunctionHelper.validateNotSingleton(function.getClass());
// check if class could be instantiated
- UserFunctionsTypeHelper.validateInstantiation(function.getClass());
+ UserDefinedFunctionHelper.validateInstantiation(function.getClass());
final FunctionDefinition definition;
if (function instanceof AggregateFunction) {
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/functions/UserFunctionsTypeHelper.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/functions/UserDefinedFunctionHelper.java
similarity index 95%
rename from flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/functions/UserFunctionsTypeHelper.java
rename to flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/functions/UserDefinedFunctionHelper.java
index 5ab383f..6069cb9 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/functions/UserFunctionsTypeHelper.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/functions/UserDefinedFunctionHelper.java
@@ -19,8 +19,11 @@
package org.apache.flink.table.functions;
import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.ExecutionConfig.ClosureCleanerLevel;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.configuration.PipelineOptions;
+import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.api.ValidationException;
import org.apache.flink.util.InstantiationUtil;
@@ -31,7 +34,7 @@ import java.util.Arrays;
* {@link org.apache.flink.table.catalog.FunctionCatalog}.
*/
@Internal
-public class UserFunctionsTypeHelper {
+public class UserDefinedFunctionHelper {
/**
* Tries to infer the TypeInformation of an AggregateFunction's accumulator type.
@@ -152,9 +155,6 @@ public class UserFunctionsTypeHelper {
throw new ValidationException(String.format(
"Function class %s is no proper class," +
" it is either abstract, an interface, or a primitive type.", clazz.getCanonicalName()));
- } else if (InstantiationUtil.isNonStaticInnerClass(clazz)) {
- throw new ValidationException(String.format(
- "The class %s is an inner class, but not statically accessible.", clazz.getCanonicalName()));
}
}
@@ -171,6 +171,6 @@ public class UserFunctionsTypeHelper {
}
}
- private UserFunctionsTypeHelper() {
+ private UserDefinedFunctionHelper() {
}
}
diff --git a/flink-table/flink-table-api-scala-bridge/src/main/scala/org/apache/flink/table/api/scala/internal/StreamTableEnvironmentImpl.scala b/flink-table/flink-table-api-scala-bridge/src/main/scala/org/apache/flink/table/api/scala/internal/StreamTableEnvironmentImpl.scala
index 9253ecd..fcddda9 100644
--- a/flink-table/flink-table-api-scala-bridge/src/main/scala/org/apache/flink/table/api/scala/internal/StreamTableEnvironmentImpl.scala
+++ b/flink-table/flink-table-api-scala-bridge/src/main/scala/org/apache/flink/table/api/scala/internal/StreamTableEnvironmentImpl.scala
@@ -34,7 +34,7 @@ import org.apache.flink.table.delegation.{Executor, ExecutorFactory, Planner, Pl
import org.apache.flink.table.descriptors.{ConnectorDescriptor, StreamTableDescriptor}
import org.apache.flink.table.expressions.Expression
import org.apache.flink.table.factories.ComponentFactoryService
-import org.apache.flink.table.functions.{AggregateFunction, TableAggregateFunction, TableFunction, UserFunctionsTypeHelper}
+import org.apache.flink.table.functions.{AggregateFunction, TableAggregateFunction, TableFunction, UserDefinedFunctionHelper}
import org.apache.flink.table.module.ModuleManager
import org.apache.flink.table.operations.{OutputConversionModifyOperation, QueryOperation, ScalaDataStreamQueryOperation}
import org.apache.flink.table.sources.{TableSource, TableSourceValidation}
@@ -138,7 +138,7 @@ class StreamTableEnvironmentImpl (
}
override def registerFunction[T: TypeInformation](name: String, tf: TableFunction[T]): Unit = {
- val typeInfo = UserFunctionsTypeHelper
+ val typeInfo = UserDefinedFunctionHelper
.getReturnTypeOfTableFunction(tf, implicitly[TypeInformation[T]])
functionCatalog.registerTempSystemTableFunction(
name,
@@ -151,9 +151,9 @@ class StreamTableEnvironmentImpl (
name: String,
f: AggregateFunction[T, ACC])
: Unit = {
- val typeInfo = UserFunctionsTypeHelper
+ val typeInfo = UserDefinedFunctionHelper
.getReturnTypeOfAggregateFunction(f, implicitly[TypeInformation[T]])
- val accTypeInfo = UserFunctionsTypeHelper
+ val accTypeInfo = UserDefinedFunctionHelper
.getAccumulatorTypeOfAggregateFunction(f, implicitly[TypeInformation[ACC]])
functionCatalog.registerTempSystemAggregateFunction(
name,
@@ -167,9 +167,9 @@ class StreamTableEnvironmentImpl (
name: String,
f: TableAggregateFunction[T, ACC])
: Unit = {
- val typeInfo = UserFunctionsTypeHelper
+ val typeInfo = UserDefinedFunctionHelper
.getReturnTypeOfAggregateFunction(f, implicitly[TypeInformation[T]])
- val accTypeInfo = UserFunctionsTypeHelper
+ val accTypeInfo = UserDefinedFunctionHelper
.getAccumulatorTypeOfAggregateFunction(f, implicitly[TypeInformation[ACC]])
functionCatalog.registerTempSystemAggregateFunction(
name,
diff --git a/flink-table/flink-table-api-scala/src/main/scala/org/apache/flink/table/api/expressionDsl.scala b/flink-table/flink-table-api-scala/src/main/scala/org/apache/flink/table/api/expressionDsl.scala
index 2248f8e..0db50e8 100644
--- a/flink-table/flink-table-api-scala/src/main/scala/org/apache/flink/table/api/expressionDsl.scala
+++ b/flink-table/flink-table-api-scala/src/main/scala/org/apache/flink/table/api/expressionDsl.scala
@@ -27,7 +27,7 @@ import org.apache.flink.api.common.typeinfo.{SqlTimeTypeInfo, TypeInformation}
import org.apache.flink.table.expressions._
import org.apache.flink.table.expressions.utils.ApiExpressionUtils._
import org.apache.flink.table.functions.BuiltInFunctionDefinitions._
-import org.apache.flink.table.functions.{ScalarFunction, TableFunction, UserDefinedAggregateFunction, UserFunctionsTypeHelper, _}
+import org.apache.flink.table.functions.{ScalarFunction, TableFunction, UserDefinedAggregateFunction, UserDefinedFunctionHelper, _}
import org.apache.flink.table.types.DataType
import org.apache.flink.table.types.utils.TypeConversions
import org.apache.flink.table.types.utils.TypeConversions.fromLegacyInfoToDataType
@@ -1119,7 +1119,7 @@ trait ImplicitExpressionConversions {
* Calls a table function for the given parameters.
*/
def apply(params: Expression*): Expression = {
- val resultTypeInfo: TypeInformation[T] = UserFunctionsTypeHelper
+ val resultTypeInfo: TypeInformation[T] = UserDefinedFunctionHelper
.getReturnTypeOfTableFunction(t, implicitly[TypeInformation[T]])
unresolvedCall(new TableFunctionDefinition(t.getClass.getName, t, resultTypeInfo), params: _*)
}
@@ -1129,10 +1129,10 @@ trait ImplicitExpressionConversions {
(val a: UserDefinedAggregateFunction[T, ACC]) {
private def createFunctionDefinition(): FunctionDefinition = {
- val resultTypeInfo: TypeInformation[T] = UserFunctionsTypeHelper
+ val resultTypeInfo: TypeInformation[T] = UserDefinedFunctionHelper
.getReturnTypeOfAggregateFunction(a, implicitly[TypeInformation[T]])
- val accTypeInfo: TypeInformation[ACC] = UserFunctionsTypeHelper.
+ val accTypeInfo: TypeInformation[ACC] = UserDefinedFunctionHelper.
getAccumulatorTypeOfAggregateFunction(a, implicitly[TypeInformation[ACC]])
a match {
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/expressions/call.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/expressions/call.scala
index 7dacccf..b4aeb41 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/expressions/call.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/expressions/call.scala
@@ -201,9 +201,9 @@ case class PlannerTableFunctionCall(
override def validateInput(): ValidationResult = {
// check if not Scala object
- UserFunctionsTypeHelper.validateNotSingleton(tableFunction.getClass)
+ UserDefinedFunctionHelper.validateNotSingleton(tableFunction.getClass)
// check if class could be instantiated
- UserFunctionsTypeHelper.validateInstantiation(tableFunction.getClass)
+ UserDefinedFunctionHelper.validateInstantiation(tableFunction.getClass)
// look for a signature that matches the input types
val signature = parameters.map(_.resultType).map(fromLegacyInfoToDataType)
val foundMethod = getUserDefinedMethod(
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdHandlerTestBase.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdHandlerTestBase.scala
index 4e2e0d5..aa3d48f 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdHandlerTestBase.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdHandlerTestBase.scala
@@ -22,7 +22,7 @@ import org.apache.flink.table.api.{TableConfig, TableException}
import org.apache.flink.table.catalog.{CatalogManager, FunctionCatalog, GenericInMemoryCatalog}
import org.apache.flink.table.expressions._
import org.apache.flink.table.expressions.utils.ApiExpressionUtils.intervalOfMillis
-import org.apache.flink.table.functions.{FunctionIdentifier, UserFunctionsTypeHelper}
+import org.apache.flink.table.functions.{FunctionIdentifier, UserDefinedFunctionHelper}
import org.apache.flink.table.planner.calcite.FlinkRelBuilder.PlannerNamedWindowProperty
import org.apache.flink.table.planner.calcite.{FlinkRelBuilder, FlinkTypeFactory}
import org.apache.flink.table.planner.delegation.PlannerContext
@@ -713,8 +713,8 @@ class FlinkRelMdHandlerTestBase {
protected lazy val tableAggCall = {
val top3 = new Top3
- val resultTypeInfo = UserFunctionsTypeHelper.getReturnTypeOfAggregateFunction(top3)
- val accTypeInfo = UserFunctionsTypeHelper.getAccumulatorTypeOfAggregateFunction(top3)
+ val resultTypeInfo = UserDefinedFunctionHelper.getReturnTypeOfAggregateFunction(top3)
+ val accTypeInfo = UserDefinedFunctionHelper.getAccumulatorTypeOfAggregateFunction(top3)
val resultDataType = TypeConversions.fromLegacyInfoToDataType(resultTypeInfo)
val accDataType = TypeConversions.fromLegacyInfoToDataType(accTypeInfo)
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 2b59214..c65f5c7 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
@@ -963,7 +963,7 @@ class TestingTableEnvironment private(
// `<T, ACC> void registerFunction(String name, AggregateFunction<T, ACC> aggregateFunction);`
// is added into TableEnvironment
def registerFunction[T: TypeInformation](name: String, tf: TableFunction[T]): Unit = {
- val typeInfo = UserFunctionsTypeHelper
+ val typeInfo = UserDefinedFunctionHelper
.getReturnTypeOfTableFunction(tf, implicitly[TypeInformation[T]])
functionCatalog.registerTempSystemTableFunction(
name,
@@ -993,9 +993,9 @@ class TestingTableEnvironment private(
private def registerUserDefinedAggregateFunction[T: TypeInformation, ACC: TypeInformation](
name: String,
f: UserDefinedAggregateFunction[T, ACC]): Unit = {
- val typeInfo = UserFunctionsTypeHelper
+ val typeInfo = UserDefinedFunctionHelper
.getReturnTypeOfAggregateFunction(f, implicitly[TypeInformation[T]])
- val accTypeInfo = UserFunctionsTypeHelper
+ val accTypeInfo = UserDefinedFunctionHelper
.getAccumulatorTypeOfAggregateFunction(f, implicitly[TypeInformation[ACC]])
functionCatalog.registerTempSystemAggregateFunction(
name,
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 3f55bf3..d6eba73 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
@@ -142,7 +142,7 @@ abstract class TableEnvImpl(
name: String,
function: TableFunction[T])
: Unit = {
- val resultTypeInfo: TypeInformation[T] = UserFunctionsTypeHelper
+ val resultTypeInfo: TypeInformation[T] = UserDefinedFunctionHelper
.getReturnTypeOfTableFunction(
function,
implicitly[TypeInformation[T]])
@@ -161,12 +161,12 @@ abstract class TableEnvImpl(
name: String,
function: UserDefinedAggregateFunction[T, ACC])
: Unit = {
- val resultTypeInfo: TypeInformation[T] = UserFunctionsTypeHelper
+ val resultTypeInfo: TypeInformation[T] = UserDefinedFunctionHelper
.getReturnTypeOfAggregateFunction(
function,
implicitly[TypeInformation[T]])
- val accTypeInfo: TypeInformation[ACC] = UserFunctionsTypeHelper
+ val accTypeInfo: TypeInformation[ACC] = UserDefinedFunctionHelper
.getAccumulatorTypeOfAggregateFunction(
function,
implicitly[TypeInformation[ACC]])
@@ -824,8 +824,8 @@ abstract class TableEnvImpl(
val aggregateFunctionDefinition = functionDefinition.asInstanceOf[AggregateFunctionDefinition]
val aggregateFunction = aggregateFunctionDefinition
.getAggregateFunction.asInstanceOf[AggregateFunction[T, ACC]]
- val typeInfo = UserFunctionsTypeHelper.getReturnTypeOfAggregateFunction(aggregateFunction)
- val accTypeInfo = UserFunctionsTypeHelper
+ val typeInfo = UserDefinedFunctionHelper.getReturnTypeOfAggregateFunction(aggregateFunction)
+ val accTypeInfo = UserDefinedFunctionHelper
.getAccumulatorTypeOfAggregateFunction(aggregateFunction)
functionCatalog.registerTempCatalogAggregateFunction(
functionIdentifier,
@@ -836,7 +836,7 @@ abstract class TableEnvImpl(
} else if (functionDefinition.isInstanceOf[TableFunctionDefinition]) {
val tableFunctionDefinition = functionDefinition.asInstanceOf[TableFunctionDefinition]
val tableFunction = tableFunctionDefinition.getTableFunction.asInstanceOf[TableFunction[T]]
- val typeInfo = UserFunctionsTypeHelper.getReturnTypeOfTableFunction(tableFunction)
+ val typeInfo = UserDefinedFunctionHelper.getReturnTypeOfTableFunction(tableFunction)
functionCatalog.registerTempCatalogTableFunction(
functionIdentifier,
tableFunction,
@@ -857,8 +857,8 @@ abstract class TableEnvImpl(
val aggregateFunctionDefinition = functionDefinition.asInstanceOf[AggregateFunctionDefinition]
val aggregateFunction = aggregateFunctionDefinition
.getAggregateFunction.asInstanceOf[AggregateFunction[T, ACC]]
- val typeInfo = UserFunctionsTypeHelper.getReturnTypeOfAggregateFunction(aggregateFunction)
- val accTypeInfo = UserFunctionsTypeHelper
+ val typeInfo = UserDefinedFunctionHelper.getReturnTypeOfAggregateFunction(aggregateFunction)
+ val accTypeInfo = UserDefinedFunctionHelper
.getAccumulatorTypeOfAggregateFunction(aggregateFunction)
functionCatalog.registerTempSystemAggregateFunction(
functionName,
@@ -868,7 +868,7 @@ abstract class TableEnvImpl(
} else if (functionDefinition.isInstanceOf[TableFunctionDefinition]) {
val tableFunctionDefinition = functionDefinition.asInstanceOf[TableFunctionDefinition]
val tableFunction = tableFunctionDefinition.getTableFunction.asInstanceOf[TableFunction[T]]
- val typeInfo = UserFunctionsTypeHelper.getReturnTypeOfTableFunction(tableFunction)
+ val typeInfo = UserDefinedFunctionHelper.getReturnTypeOfTableFunction(tableFunction)
functionCatalog.registerTempSystemTableFunction(
functionName,
tableFunction,
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/call.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/call.scala
index f2da6fc..b40053a 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/call.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/call.scala
@@ -299,9 +299,9 @@ case class PlannerTableFunctionCall(
override def validateInput(): ValidationResult = {
// check if not Scala object
- UserFunctionsTypeHelper.validateNotSingleton(tableFunction.getClass)
+ UserDefinedFunctionHelper.validateNotSingleton(tableFunction.getClass)
// check if class could be instantiated
- UserFunctionsTypeHelper.validateInstantiation(tableFunction.getClass)
+ UserDefinedFunctionHelper.validateInstantiation(tableFunction.getClass)
// look for a signature that matches the input types
val signature = parameters.map(_.resultType)
val foundMethod = getUserDefinedMethod(tableFunction, "eval", typeInfoToClass(signature))
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 0d4e597..c9924db 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
@@ -43,7 +43,7 @@ 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, UserFunctionsTypeHelper}
+import org.apache.flink.table.functions.{AggregateFunction, TableAggregateFunction, UserDefinedAggregateFunction, UserDefinedFunctionHelper}
import org.apache.flink.table.plan.logical._
import org.apache.flink.table.runtime.types.{CRow, CRowTypeInfo}
import org.apache.flink.table.typeutils.TimeIntervalTypeInfo
@@ -1420,7 +1420,7 @@ object AggregateUtil {
val (accumulatorType, accSpecs) = {
val accType = aggregateFunction match {
case udagg: AggSqlFunction => udagg.accType
- case _ => UserFunctionsTypeHelper.getAccumulatorTypeOfAggregateFunction(aggregate)
+ case _ => UserDefinedFunctionHelper.getAccumulatorTypeOfAggregateFunction(aggregate)
}
removeStateViewFieldsFromAccTypeInfo(
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/runtime/harness/HarnessTestBase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/runtime/harness/HarnessTestBase.scala
index 94d2e2e..577ea7b 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/runtime/harness/HarnessTestBase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/runtime/harness/HarnessTestBase.scala
@@ -36,7 +36,7 @@ import org.apache.flink.table.api.StreamQueryConfig
import org.apache.flink.table.api.dataview.DataView
import org.apache.flink.table.codegen.GeneratedAggregationsFunction
import org.apache.flink.table.functions.aggfunctions.{CountAggFunction, IntSumWithRetractAggFunction, LongMaxWithRetractAggFunction, LongMinWithRetractAggFunction}
-import org.apache.flink.table.functions.{AggregateFunction, UserDefinedFunction, UserFunctionsTypeHelper}
+import org.apache.flink.table.functions.{AggregateFunction, UserDefinedFunction, UserDefinedFunctionHelper}
import org.apache.flink.table.runtime.aggregate.GeneratedAggregations
import org.apache.flink.table.runtime.harness.HarnessTestBase.{RowResultSortComparator, RowResultSortComparatorWithWatermarks}
import org.apache.flink.table.runtime.types.{CRow, CRowTypeInfo}
@@ -79,11 +79,11 @@ class HarnessTestBase extends StreamingWithStateTestBase {
protected val minMaxAggregationStateType: RowTypeInfo =
new RowTypeInfo(minMaxAggregates
- .map(UserFunctionsTypeHelper.getAccumulatorTypeOfAggregateFunction(_)): _*)
+ .map(UserDefinedFunctionHelper.getAccumulatorTypeOfAggregateFunction(_)): _*)
protected val sumAggregationStateType: RowTypeInfo =
new RowTypeInfo(sumAggregates
- .map(UserFunctionsTypeHelper.getAccumulatorTypeOfAggregateFunction(_)): _*)
+ .map(UserDefinedFunctionHelper.getAccumulatorTypeOfAggregateFunction(_)): _*)
protected val minMaxFuncName = "MinMaxAggregateHelper"
protected val sumFuncName = "SumAggregationHelper"
[flink] 03/04: [hotfix][table] Update function tests to use testing
infrastructure
Posted by tw...@apache.org.
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 7fb9f9bf40797833e13685b7445d91d902c65d7e
Author: Timo Walther <tw...@apache.org>
AuthorDate: Tue Dec 17 10:46:13 2019 +0100
[hotfix][table] Update function tests to use testing infrastructure
---
.../table/planner/functions/FunctionTestBase.java | 362 -----------------
.../planner/runtime/batch/sql/FunctionITCase.java | 39 --
.../planner/runtime/stream/sql/FunctionITCase.java | 390 ++++++++++++++++--
.../flink/table/functions/FunctionTestBase.java | 436 ---------------------
.../table/runtime/batch/sql/FunctionITCase.java | 43 --
.../table/runtime/stream/sql/FunctionITCase.java | 431 +++++++++++++++++++-
6 files changed, 767 insertions(+), 934 deletions(-)
diff --git a/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/functions/FunctionTestBase.java b/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/functions/FunctionTestBase.java
deleted file mode 100644
index e788df9..0000000
--- a/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/functions/FunctionTestBase.java
+++ /dev/null
@@ -1,362 +0,0 @@
-/*
- * 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.planner.functions;
-
-import org.apache.flink.table.api.TableEnvironment;
-import org.apache.flink.table.api.ValidationException;
-import org.apache.flink.table.catalog.Catalog;
-import org.apache.flink.table.catalog.CatalogFunction;
-import org.apache.flink.table.catalog.ObjectPath;
-import org.apache.flink.table.functions.ScalarFunction;
-import org.apache.flink.types.Row;
-
-import org.junit.Test;
-
-import java.util.Arrays;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
-
-/**
- * Tests for {@link CatalogFunction}.
- */
-public abstract class FunctionTestBase {
- protected static TableEnvironment tableEnv;
-
- @Test
- public void testCreateCatalogFunctionInDefaultCatalog() {
- String ddl1 = "create function f1 as 'org.apache.flink.function.TestFunction'";
- tableEnv.sqlUpdate(ddl1);
- assertTrue(Arrays.asList(tableEnv.listFunctions()).contains("f1"));
-
- tableEnv.sqlUpdate("DROP FUNCTION IF EXISTS default_catalog.default_database.f1");
- assertFalse(Arrays.asList(tableEnv.listFunctions()).contains("f1"));
- }
-
- @Test
- public void testCreateFunctionWithFullPath() {
- String ddl1 = "create function default_catalog.default_database.f2 as" +
- " 'org.apache.flink.function.TestFunction'";
- tableEnv.sqlUpdate(ddl1);
- assertTrue(Arrays.asList(tableEnv.listFunctions()).contains("f2"));
-
- tableEnv.sqlUpdate("DROP FUNCTION IF EXISTS default_catalog.default_database.f2");
- assertFalse(Arrays.asList(tableEnv.listFunctions()).contains("f2"));
- }
-
- @Test
- public void testCreateFunctionWithoutCatalogIdentifier() {
- String ddl1 = "create function default_database.f3 as" +
- " 'org.apache.flink.function.TestFunction'";
- tableEnv.sqlUpdate(ddl1);
- assertTrue(Arrays.asList(tableEnv.listFunctions()).contains("f3"));
-
- tableEnv.sqlUpdate("DROP FUNCTION IF EXISTS default_catalog.default_database.f3");
- assertFalse(Arrays.asList(tableEnv.listFunctions()).contains("f3"));
- }
-
- @Test
- public void testCreateFunctionCatalogNotExists() {
-
- String ddl1 = "create function catalog1.database1.f3 as 'org.apache.flink.function.TestFunction'";
-
- try {
- tableEnv.sqlUpdate(ddl1);
- } catch (Exception e){
- assertTrue(e.getMessage().equals("Catalog catalog1 does not exist"));
- }
- }
-
- @Test
- public void testCreateFunctionDBNotExists() {
- String ddl1 = "create function default_catalog.database1.f3 as 'org.apache.flink.function.TestFunction'";
-
- try {
- tableEnv.sqlUpdate(ddl1);
- } catch (Exception e){
- assertEquals(e.getMessage(), "Could not execute CREATE CATALOG FUNCTION:" +
- " (catalogFunction: [Optional[This is a user-defined function]], identifier:" +
- " [`default_catalog`.`database1`.`f3`], ignoreIfExists: [false])");
- }
- }
-
- @Test
- public void testCreateTemporaryCatalogFunction() {
- String ddl1 = "create temporary function default_catalog.default_database.f4" +
- " as 'org.apache.flink.table.planner.functions.FunctionTestBase$TestUDF'";
-
- String ddl2 = "create temporary function if not exists default_catalog.default_database.f4" +
- " as 'org.apache.flink.table.planner.functions.FunctionTestBase$TestUDF'";
-
- String ddl3 = "drop temporary function default_catalog.default_database.f4";
-
- String ddl4 = "drop temporary function if exists default_catalog.default_database.f4";
-
- tableEnv.sqlUpdate(ddl1);
- assertTrue(Arrays.asList(tableEnv.listFunctions()).contains("f4"));
-
- tableEnv.sqlUpdate(ddl2);
- assertTrue(Arrays.asList(tableEnv.listFunctions()).contains("f4"));
-
- tableEnv.sqlUpdate(ddl3);
- assertFalse(Arrays.asList(tableEnv.listFunctions()).contains("f4"));
-
- tableEnv.sqlUpdate(ddl1);
- try {
- tableEnv.sqlUpdate(ddl1);
- } catch (Exception e) {
- assertTrue(e instanceof ValidationException);
- assertEquals(e.getMessage(), "Temporary catalog function " +
- "`default_catalog`.`default_database`.`f4` is already defined");
- }
-
- tableEnv.sqlUpdate(ddl3);
- tableEnv.sqlUpdate(ddl4);
- try {
- tableEnv.sqlUpdate(ddl3);
- } catch (Exception e) {
- assertTrue(e instanceof ValidationException);
- assertEquals(e.getMessage(),
- "Temporary catalog function `default_catalog`.`default_database`.`f4` doesn't exist");
- }
- }
-
- @Test
- public void testCreateTemporarySystemFunction() {
- String ddl1 = "create temporary system function default_catalog.default_database.f5" +
- " as 'org.apache.flink.table.planner.functions.FunctionTestBase$TestUDF'";
-
- String ddl2 = "create temporary system function if not exists default_catalog.default_database.f5" +
- " as 'org.apache.flink.table.planner.functions.FunctionTestBase$TestUDF'";
-
- String ddl3 = "drop temporary system function default_catalog.default_database.f5";
-
- tableEnv.sqlUpdate(ddl1);
- tableEnv.sqlUpdate(ddl2);
- tableEnv.sqlUpdate(ddl3);
- }
-
- @Test
- public void testAlterFunction() throws Exception {
- String create = "create function f3 as 'org.apache.flink.function.TestFunction'";
- String alter = "alter function f3 as 'org.apache.flink.function.TestFunction2'";
-
- ObjectPath objectPath = new ObjectPath("default_database", "f3");
- Catalog catalog = tableEnv.getCatalog("default_catalog").get();
- tableEnv.sqlUpdate(create);
- CatalogFunction beforeUpdate = catalog.getFunction(objectPath);
- assertEquals("org.apache.flink.function.TestFunction", beforeUpdate.getClassName());
-
- tableEnv.sqlUpdate(alter);
- CatalogFunction afterUpdate = catalog.getFunction(objectPath);
- assertEquals("org.apache.flink.function.TestFunction2", afterUpdate.getClassName());
- }
-
- @Test
- public void testAlterFunctionNonExists() {
- String alterUndefinedFunction = "ALTER FUNCTION default_catalog.default_database.f4" +
- " as 'org.apache.flink.function.TestFunction'";
-
- String alterFunctionInWrongCatalog = "ALTER FUNCTION catalog1.default_database.f4 " +
- "as 'org.apache.flink.function.TestFunction'";
-
- String alterFunctionInWrongDB = "ALTER FUNCTION default_catalog.db1.f4 " +
- "as 'org.apache.flink.function.TestFunction'";
-
- try {
- tableEnv.sqlUpdate(alterUndefinedFunction);
- fail();
- } catch (Exception e){
- assertEquals(e.getMessage(),
- "Function default_database.f4 does not exist in Catalog default_catalog.");
- }
-
- try {
- tableEnv.sqlUpdate(alterFunctionInWrongCatalog);
- fail();
- } catch (Exception e) {
- assertTrue(e.getMessage().equals("Catalog catalog1 does not exist"));
- }
-
- try {
- tableEnv.sqlUpdate(alterFunctionInWrongDB);
- fail();
- } catch (Exception e) {
- assertEquals(e.getMessage(),
- "Function db1.f4 does not exist in Catalog default_catalog.");
- }
- }
-
- @Test
- public void testAlterTemporaryCatalogFunction() {
- String alterTemporary = "ALTER TEMPORARY FUNCTION default_catalog.default_database.f4" +
- " as 'org.apache.flink.function.TestFunction'";
-
- try {
- tableEnv.sqlUpdate(alterTemporary);
- fail();
- } catch (Exception e) {
- assertTrue(e.getMessage().equals("Alter temporary catalog function is not supported"));
- }
-
- }
-
- @Test
- public void testAlterTemporarySystemFunction() {
- String alterTemporary = "ALTER TEMPORARY SYSTEM FUNCTION default_catalog.default_database.f4" +
- " as 'org.apache.flink.function.TestFunction'";
-
- try {
- tableEnv.sqlUpdate(alterTemporary);
- fail();
- } catch (Exception e) {
- assertTrue(e.getMessage().equals("Alter temporary system function is not supported"));
- }
- }
-
- @Test
- public void testDropFunctionNonExists() {
- String dropUndefinedFunction = "DROP FUNCTION default_catalog.default_database.f4";
-
- String dropFunctionInWrongCatalog = "DROP FUNCTION catalog1.default_database.f4";
-
- String dropFunctionInWrongDB = "DROP FUNCTION default_catalog.db1.f4";
-
- try {
- tableEnv.sqlUpdate(dropUndefinedFunction);
- fail();
- } catch (Exception e){
- assertEquals(e.getMessage(),
- "Function default_database.f4 does not exist in Catalog default_catalog.");
- }
-
- try {
- tableEnv.sqlUpdate(dropFunctionInWrongCatalog);
- fail();
- } catch (Exception e) {
- assertTrue(e.getMessage().equals("Catalog catalog1 does not exist"));
- }
-
- try {
- tableEnv.sqlUpdate(dropFunctionInWrongDB);
- fail();
- } catch (Exception e) {
- assertEquals(e.getMessage(), "Function db1.f4 does not exist in Catalog default_catalog.");
- }
- }
-
- @Test
- public void testDropTemporaryFunctionNonExits() {
- String dropUndefinedFunction = "DROP TEMPORARY FUNCTION default_catalog.default_database.f4";
- String dropFunctionInWrongCatalog = "DROP TEMPORARY FUNCTION catalog1.default_database.f4";
- String dropFunctionInWrongDB = "DROP TEMPORARY FUNCTION default_catalog.db1.f4";
-
- try {
- tableEnv.sqlUpdate(dropUndefinedFunction);
- fail();
- } catch (Exception e){
- assertEquals(e.getMessage(),
- "Temporary catalog function `default_catalog`.`default_database`.`f4` doesn't exist");
- }
-
- try {
- tableEnv.sqlUpdate(dropFunctionInWrongCatalog);
- fail();
- } catch (Exception e) {
- assertEquals(e.getMessage(),
- "Temporary catalog function `catalog1`.`default_database`.`f4` doesn't exist");
- }
-
- try {
- tableEnv.sqlUpdate(dropFunctionInWrongDB);
- fail();
- } catch (Exception e) {
- assertEquals(e.getMessage(), "Temporary catalog function `default_catalog`.`db1`.`f4` doesn't exist");
- }
- }
-
- @Test
- public void testCreateAlterDropTemporaryCatalogFunctionsWithDifferentIdentifier() {
- String createNoCatalogDB = "create temporary function f4" +
- " as 'org.apache.flink.table.planner.functions.FunctionTestBase$TestUDF'";
-
- String dropNoCatalogDB = "drop temporary function f4";
-
- tableEnv.sqlUpdate(createNoCatalogDB);
- tableEnv.sqlUpdate(dropNoCatalogDB);
-
- String createNonExistsCatalog = "create temporary function catalog1.default_database.f4" +
- " as 'org.apache.flink.table.planner.functions.FunctionTestBase$TestUDF'";
-
- String dropNonExistsCatalog = "drop temporary function catalog1.default_database.f4";
-
- tableEnv.sqlUpdate(createNonExistsCatalog);
- tableEnv.sqlUpdate(dropNonExistsCatalog);
-
- String createNonExistsDB = "create temporary function default_catalog.db1.f4" +
- " as 'org.apache.flink.table.planner.functions.FunctionTestBase$TestUDF'";
-
- String dropNonExistsDB = "drop temporary function default_catalog.db1.f4";
-
- tableEnv.sqlUpdate(createNonExistsDB);
- tableEnv.sqlUpdate(dropNonExistsDB);
- }
-
- @Test
- public void testDropTemporarySystemFunction() {
- String ddl1 = "create temporary system function f5" +
- " as 'org.apache.flink.table.planner.functions.FunctionTestBase$TestUDF'";
-
- String ddl2 = "drop temporary system function f5";
-
- String ddl3 = "drop temporary system function if exists f5";
-
- tableEnv.sqlUpdate(ddl1);
- tableEnv.sqlUpdate(ddl2);
- tableEnv.sqlUpdate(ddl3);
-
- try {
- tableEnv.sqlUpdate(ddl2);
- } catch (Exception e) {
- assertEquals(e.getMessage(), "Temporary system function f5 doesn't exist");
- }
- }
-
- protected Row toRow(Object ... objects) {
- Row row = new Row(objects.length);
- for (int i = 0; i < objects.length; i++) {
- row.setField(i, objects[i]);
- }
-
- return row;
- }
-
- /**
- * Test udf class.
- */
- public static class TestUDF extends ScalarFunction {
-
- public Integer eval(Integer a, Integer b) {
- return a + b;
- }
- }
-}
-
diff --git a/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/FunctionITCase.java b/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/FunctionITCase.java
deleted file mode 100644
index a43c36a..0000000
--- a/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/runtime/batch/sql/FunctionITCase.java
+++ /dev/null
@@ -1,39 +0,0 @@
-/*
- * 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.planner.runtime.batch.sql;
-
-import org.apache.flink.table.api.EnvironmentSettings;
-import org.apache.flink.table.planner.functions.FunctionTestBase;
-import org.apache.flink.table.planner.utils.TestingTableEnvironment;
-
-import org.junit.BeforeClass;
-
-/**
- * Tests for catalog and system functions in batch table environment.
- */
-public class
-FunctionITCase extends FunctionTestBase {
-
- @BeforeClass
- public static void setup() {
- EnvironmentSettings environmentSettings =
- EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build();
- tableEnv = TestingTableEnvironment.create(environmentSettings);
- }
-}
diff --git a/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/FunctionITCase.java b/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/FunctionITCase.java
index 8aa8d68..310647a 100644
--- a/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/FunctionITCase.java
+++ b/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/FunctionITCase.java
@@ -18,14 +18,16 @@
package org.apache.flink.table.planner.runtime.stream.sql;
-import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.CatalogFunction;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.functions.ScalarFunction;
import org.apache.flink.table.planner.factories.utils.TestCollectionTableFactory;
-import org.apache.flink.table.planner.functions.FunctionTestBase;
-import org.apache.flink.table.planner.utils.TestingTableEnvironment;
+import org.apache.flink.table.planner.runtime.utils.StreamingTestBase;
import org.apache.flink.types.Row;
-import org.junit.BeforeClass;
import org.junit.Test;
import java.util.ArrayList;
@@ -33,82 +35,388 @@ import java.util.Arrays;
import java.util.List;
import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
/**
* Tests for catalog and system in stream table environment.
*/
-public class FunctionITCase extends FunctionTestBase {
+public class FunctionITCase extends StreamingTestBase {
- @BeforeClass
- public static void setup() {
- EnvironmentSettings environmentSettings =
- EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
- tableEnv = TestingTableEnvironment.create(environmentSettings);
+ private static final String TEST_FUNCTION = TestUDF.class.getName();
+
+ @Test
+ public void testCreateCatalogFunctionInDefaultCatalog() {
+ String ddl1 = "create function f1 as 'org.apache.flink.function.TestFunction'";
+ tEnv().sqlUpdate(ddl1);
+ assertTrue(Arrays.asList(tEnv().listFunctions()).contains("f1"));
+
+ tEnv().sqlUpdate("DROP FUNCTION IF EXISTS default_catalog.default_database.f1");
+ assertFalse(Arrays.asList(tEnv().listFunctions()).contains("f1"));
+ }
+
+ @Test
+ public void testCreateFunctionWithFullPath() {
+ String ddl1 = "create function default_catalog.default_database.f2 as" +
+ " 'org.apache.flink.function.TestFunction'";
+ tEnv().sqlUpdate(ddl1);
+ assertTrue(Arrays.asList(tEnv().listFunctions()).contains("f2"));
+
+ tEnv().sqlUpdate("DROP FUNCTION IF EXISTS default_catalog.default_database.f2");
+ assertFalse(Arrays.asList(tEnv().listFunctions()).contains("f2"));
+ }
+
+ @Test
+ public void testCreateFunctionWithoutCatalogIdentifier() {
+ String ddl1 = "create function default_database.f3 as" +
+ " 'org.apache.flink.function.TestFunction'";
+ tEnv().sqlUpdate(ddl1);
+ assertTrue(Arrays.asList(tEnv().listFunctions()).contains("f3"));
+
+ tEnv().sqlUpdate("DROP FUNCTION IF EXISTS default_catalog.default_database.f3");
+ assertFalse(Arrays.asList(tEnv().listFunctions()).contains("f3"));
+ }
+
+ @Test
+ public void testCreateFunctionCatalogNotExists() {
+ String ddl1 = "create function catalog1.database1.f3 as 'org.apache.flink.function.TestFunction'";
+
+ try {
+ tEnv().sqlUpdate(ddl1);
+ } catch (Exception e){
+ assertEquals("Catalog catalog1 does not exist", e.getMessage());
+ }
+ }
+
+ @Test
+ public void testCreateFunctionDBNotExists() {
+ String ddl1 = "create function default_catalog.database1.f3 as 'org.apache.flink.function.TestFunction'";
+
+ try {
+ tEnv().sqlUpdate(ddl1);
+ } catch (Exception e){
+ assertEquals(e.getMessage(), "Could not execute CREATE CATALOG FUNCTION:" +
+ " (catalogFunction: [Optional[This is a user-defined function]], identifier:" +
+ " [`default_catalog`.`database1`.`f3`], ignoreIfExists: [false])");
+ }
+ }
+
+ @Test
+ public void testCreateTemporaryCatalogFunction() {
+ String ddl1 = "create temporary function default_catalog.default_database.f4" +
+ " as '" + TEST_FUNCTION + "'";
+
+ String ddl2 = "create temporary function if not exists default_catalog.default_database.f4" +
+ " as '" + TEST_FUNCTION + "'";
+
+ String ddl3 = "drop temporary function default_catalog.default_database.f4";
+
+ String ddl4 = "drop temporary function if exists default_catalog.default_database.f4";
+
+ tEnv().sqlUpdate(ddl1);
+ assertTrue(Arrays.asList(tEnv().listFunctions()).contains("f4"));
+
+ tEnv().sqlUpdate(ddl2);
+ assertTrue(Arrays.asList(tEnv().listFunctions()).contains("f4"));
+
+ tEnv().sqlUpdate(ddl3);
+ assertFalse(Arrays.asList(tEnv().listFunctions()).contains("f4"));
+
+ tEnv().sqlUpdate(ddl1);
+ try {
+ tEnv().sqlUpdate(ddl1);
+ } catch (Exception e) {
+ assertTrue(e instanceof ValidationException);
+ assertEquals(e.getMessage(),
+ "Temporary catalog function `default_catalog`.`default_database`.`f4`" +
+ " is already defined");
+ }
+
+ tEnv().sqlUpdate(ddl3);
+ tEnv().sqlUpdate(ddl4);
+ try {
+ tEnv().sqlUpdate(ddl3);
+ } catch (Exception e) {
+ assertTrue(e instanceof ValidationException);
+ assertEquals(e.getMessage(),
+ "Temporary catalog function `default_catalog`.`default_database`.`f4`" +
+ " doesn't exist");
+ }
+ }
+
+ @Test
+ public void testCreateTemporarySystemFunction() {
+ String ddl1 = "create temporary system function default_catalog.default_database.f5" +
+ " as '" + TEST_FUNCTION + "'";
+
+ String ddl2 = "create temporary system function if not exists default_catalog.default_database.f5" +
+ " as '" + TEST_FUNCTION + "'";
+
+ String ddl3 = "drop temporary system function default_catalog.default_database.f5";
+
+ tEnv().sqlUpdate(ddl1);
+ tEnv().sqlUpdate(ddl2);
+ tEnv().sqlUpdate(ddl3);
+ }
+
+ @Test
+ public void testAlterFunction() throws Exception {
+ String create = "create function f3 as 'org.apache.flink.function.TestFunction'";
+ String alter = "alter function f3 as 'org.apache.flink.function.TestFunction2'";
+
+ ObjectPath objectPath = new ObjectPath("default_database", "f3");
+ assertTrue(tEnv().getCatalog("default_catalog").isPresent());
+ Catalog catalog = tEnv().getCatalog("default_catalog").get();
+ tEnv().sqlUpdate(create);
+ CatalogFunction beforeUpdate = catalog.getFunction(objectPath);
+ assertEquals("org.apache.flink.function.TestFunction", beforeUpdate.getClassName());
+
+ tEnv().sqlUpdate(alter);
+ CatalogFunction afterUpdate = catalog.getFunction(objectPath);
+ assertEquals("org.apache.flink.function.TestFunction2", afterUpdate.getClassName());
+ }
+
+ @Test
+ public void testAlterFunctionNonExists() {
+ String alterUndefinedFunction = "ALTER FUNCTION default_catalog.default_database.f4" +
+ " as 'org.apache.flink.function.TestFunction'";
+
+ String alterFunctionInWrongCatalog = "ALTER FUNCTION catalog1.default_database.f4 " +
+ "as 'org.apache.flink.function.TestFunction'";
+
+ String alterFunctionInWrongDB = "ALTER FUNCTION default_catalog.db1.f4 " +
+ "as 'org.apache.flink.function.TestFunction'";
+
+ try {
+ tEnv().sqlUpdate(alterUndefinedFunction);
+ fail();
+ } catch (Exception e){
+ assertEquals(e.getMessage(),
+ "Function default_database.f4 does not exist in Catalog default_catalog.");
+ }
+
+ try {
+ tEnv().sqlUpdate(alterFunctionInWrongCatalog);
+ fail();
+ } catch (Exception e) {
+ assertEquals("Catalog catalog1 does not exist", e.getMessage());
+ }
+
+ try {
+ tEnv().sqlUpdate(alterFunctionInWrongDB);
+ fail();
+ } catch (Exception e) {
+ assertEquals(e.getMessage(), "Function db1.f4 does not exist" +
+ " in Catalog default_catalog.");
+ }
+ }
+
+ @Test
+ public void testAlterTemporaryCatalogFunction() {
+ String alterTemporary = "ALTER TEMPORARY FUNCTION default_catalog.default_database.f4" +
+ " as 'org.apache.flink.function.TestFunction'";
+
+ try {
+ tEnv().sqlUpdate(alterTemporary);
+ fail();
+ } catch (Exception e) {
+ assertEquals("Alter temporary catalog function is not supported", e.getMessage());
+ }
+ }
+
+ @Test
+ public void testAlterTemporarySystemFunction() {
+ String alterTemporary = "ALTER TEMPORARY SYSTEM FUNCTION default_catalog.default_database.f4" +
+ " as 'org.apache.flink.function.TestFunction'";
+
+ try {
+ tEnv().sqlUpdate(alterTemporary);
+ fail();
+ } catch (Exception e) {
+ assertEquals("Alter temporary system function is not supported", e.getMessage());
+ }
+ }
+
+ @Test
+ public void testDropFunctionNonExists() {
+ String dropUndefinedFunction = "DROP FUNCTION default_catalog.default_database.f4";
+
+ String dropFunctionInWrongCatalog = "DROP FUNCTION catalog1.default_database.f4";
+
+ String dropFunctionInWrongDB = "DROP FUNCTION default_catalog.db1.f4";
+
+ try {
+ tEnv().sqlUpdate(dropUndefinedFunction);
+ fail();
+ } catch (Exception e){
+ assertEquals(e.getMessage(),
+ "Function default_database.f4 does not exist in Catalog default_catalog.");
+ }
+
+ try {
+ tEnv().sqlUpdate(dropFunctionInWrongCatalog);
+ fail();
+ } catch (Exception e) {
+ assertEquals("Catalog catalog1 does not exist", e.getMessage());
+ }
+
+ try {
+ tEnv().sqlUpdate(dropFunctionInWrongDB);
+ fail();
+ } catch (Exception e) {
+ assertEquals(e.getMessage(),
+ "Function db1.f4 does not exist in Catalog default_catalog.");
+ }
+ }
+
+ @Test
+ public void testDropTemporaryFunctionNonExits() {
+ String dropUndefinedFunction = "DROP TEMPORARY FUNCTION default_catalog.default_database.f4";
+ String dropFunctionInWrongCatalog = "DROP TEMPORARY FUNCTION catalog1.default_database.f4";
+ String dropFunctionInWrongDB = "DROP TEMPORARY FUNCTION default_catalog.db1.f4";
+
+ try {
+ tEnv().sqlUpdate(dropUndefinedFunction);
+ fail();
+ } catch (Exception e){
+ assertEquals(e.getMessage(), "Temporary catalog function" +
+ " `default_catalog`.`default_database`.`f4` doesn't exist");
+ }
+
+ try {
+ tEnv().sqlUpdate(dropFunctionInWrongCatalog);
+ fail();
+ } catch (Exception e) {
+ assertEquals(e.getMessage(), "Temporary catalog function " +
+ "`catalog1`.`default_database`.`f4` doesn't exist");
+ }
+
+ try {
+ tEnv().sqlUpdate(dropFunctionInWrongDB);
+ fail();
+ } catch (Exception e) {
+ assertEquals(e.getMessage(), "Temporary catalog function " +
+ "`default_catalog`.`db1`.`f4` doesn't exist");
+ }
}
@Test
- public void testUseDefinedRegularCatalogFunction() throws Exception {
- String functionDDL = "create function addOne as " +
- "'org.apache.flink.table.planner.functions.FunctionTestBase$TestUDF'";
+ public void testCreateDropTemporaryCatalogFunctionsWithDifferentIdentifier() {
+ String createNoCatalogDB = "create temporary function f4" +
+ " as '" + TEST_FUNCTION + "'";
+
+ String dropNoCatalogDB = "drop temporary function f4";
+
+ tEnv().sqlUpdate(createNoCatalogDB);
+ tEnv().sqlUpdate(dropNoCatalogDB);
+
+ String createNonExistsCatalog = "create temporary function catalog1.default_database.f4" +
+ " as '" + TEST_FUNCTION + "'";
+
+ String dropNonExistsCatalog = "drop temporary function catalog1.default_database.f4";
+
+ tEnv().sqlUpdate(createNonExistsCatalog);
+ tEnv().sqlUpdate(dropNonExistsCatalog);
+
+ String createNonExistsDB = "create temporary function default_catalog.db1.f4" +
+ " as '" + TEST_FUNCTION + "'";
+
+ String dropNonExistsDB = "drop temporary function default_catalog.db1.f4";
+
+ tEnv().sqlUpdate(createNonExistsDB);
+ tEnv().sqlUpdate(dropNonExistsDB);
+ }
+
+ @Test
+ public void testDropTemporarySystemFunction() {
+ String ddl1 = "create temporary system function f5 as '" + TEST_FUNCTION + "'";
+
+ String ddl2 = "drop temporary system function f5";
+
+ String ddl3 = "drop temporary system function if exists f5";
+
+ tEnv().sqlUpdate(ddl1);
+ tEnv().sqlUpdate(ddl2);
+ tEnv().sqlUpdate(ddl3);
+
+ try {
+ tEnv().sqlUpdate(ddl2);
+ } catch (Exception e) {
+ assertEquals(e.getMessage(), "Temporary system function f5 doesn't exist");
+ }
+ }
+
+ @Test
+ public void testUserDefinedRegularCatalogFunction() throws Exception {
+ String functionDDL = "create function addOne as '" + TEST_FUNCTION + "'";
String dropFunctionDDL = "drop function addOne";
- testUseDefinedCatalogFunction(functionDDL);
+ testUserDefinedCatalogFunction(functionDDL);
// delete the function
- tableEnv.sqlUpdate(dropFunctionDDL);
+ tEnv().sqlUpdate(dropFunctionDDL);
}
@Test
- public void testUseDefinedTemporaryCatalogFunction() throws Exception {
- String functionDDL = "create temporary function addOne as " +
- "'org.apache.flink.table.planner.functions.FunctionTestBase$TestUDF'";
+ public void testUserDefinedTemporaryCatalogFunction() throws Exception {
+ String functionDDL = "create temporary function addOne as '" + TEST_FUNCTION + "'";
String dropFunctionDDL = "drop temporary function addOne";
- testUseDefinedCatalogFunction(functionDDL);
+ testUserDefinedCatalogFunction(functionDDL);
// delete the function
- tableEnv.sqlUpdate(dropFunctionDDL);
+ tEnv().sqlUpdate(dropFunctionDDL);
}
@Test
- public void testUseDefinedTemporarySystemFunction() throws Exception {
- String functionDDL = "create temporary system function addOne as " +
- "'org.apache.flink.table.planner.functions.FunctionTestBase$TestUDF'";
+ public void testUserDefinedTemporarySystemFunction() throws Exception {
+ String functionDDL = "create temporary system function addOne as '" + TEST_FUNCTION + "'";
String dropFunctionDDL = "drop temporary system function addOne";
- testUseDefinedCatalogFunction(functionDDL);
+ testUserDefinedCatalogFunction(functionDDL);
// delete the function
- tableEnv.sqlUpdate(dropFunctionDDL);
+ tEnv().sqlUpdate(dropFunctionDDL);
+ }
+
+ /**
+ * Test udf class.
+ */
+ public static class TestUDF extends ScalarFunction {
+
+ public Integer eval(Integer a, Integer b) {
+ return a + b;
+ }
}
- // This test case only works for stream mode
- private void testUseDefinedCatalogFunction(String createFunctionDDL) throws Exception {
+ private void testUserDefinedCatalogFunction(String createFunctionDDL) throws Exception {
List<Row> sourceData = Arrays.asList(
- toRow(1, "1000", 2),
- toRow(2, "1", 3),
- toRow(3, "2000", 4),
- toRow(1, "2", 2),
- toRow(2, "3000", 3)
+ Row.of(1, "1000", 2),
+ Row.of(2, "1", 3),
+ Row.of(3, "2000", 4),
+ Row.of(1, "2", 2),
+ Row.of(2, "3000", 3)
);
TestCollectionTableFactory.reset();
- TestCollectionTableFactory.initData(sourceData, new ArrayList<Row>(), -1);
+ TestCollectionTableFactory.initData(sourceData, new ArrayList<>(), -1);
String sourceDDL = "create table t1(a int, b varchar, c int) with ('connector' = 'COLLECTION')";
String sinkDDL = "create table t2(a int, b varchar, c int) with ('connector' = 'COLLECTION')";
String query = "select t1.a, t1.b, addOne(t1.a, 1) as c from t1";
- tableEnv.sqlUpdate(sourceDDL);
- tableEnv.sqlUpdate(sinkDDL);
- tableEnv.sqlUpdate(createFunctionDDL);
- Table t2 = tableEnv.sqlQuery(query);
- tableEnv.insertInto("t2", t2);
- tableEnv.execute("job1");
+ tEnv().sqlUpdate(sourceDDL);
+ tEnv().sqlUpdate(sinkDDL);
+ tEnv().sqlUpdate(createFunctionDDL);
+ Table t2 = tEnv().sqlQuery(query);
+ tEnv().insertInto("t2", t2);
+ tEnv().execute("job1");
Row[] result = TestCollectionTableFactory.RESULT().toArray(new Row[0]);
Row[] expected = sourceData.toArray(new Row[0]);
assertArrayEquals(expected, result);
- tableEnv.sqlUpdate("drop table t1");
- tableEnv.sqlUpdate("drop table t2");
+ tEnv().sqlUpdate("drop table t1");
+ tEnv().sqlUpdate("drop table t2");
}
}
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/functions/FunctionTestBase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/functions/FunctionTestBase.java
deleted file mode 100644
index 5815da6..0000000
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/functions/FunctionTestBase.java
+++ /dev/null
@@ -1,436 +0,0 @@
-/*
- * 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;
-
-import org.apache.flink.table.api.TableEnvironment;
-import org.apache.flink.table.api.ValidationException;
-import org.apache.flink.table.catalog.Catalog;
-import org.apache.flink.table.catalog.CatalogFunction;
-import org.apache.flink.table.catalog.ObjectPath;
-import org.apache.flink.table.factories.utils.TestCollectionTableFactory;
-import org.apache.flink.types.Row;
-
-import org.junit.Test;
-
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
-
-import static org.junit.Assert.assertArrayEquals;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
-
-/**
- * Tests for both catalog and system function.
- */
-public abstract class FunctionTestBase {
- private static TableEnvironment tableEnv;
-
- public static void setTableEnv(TableEnvironment e) {
- tableEnv = e;
- }
-
- public abstract void execute() throws Exception;
-
- @Test
- public void testCreateCatalogFunctionInDefaultCatalog() {
- String ddl1 = "create function f1 as 'org.apache.flink.function.TestFunction'";
- tableEnv.sqlUpdate(ddl1);
- assertTrue(Arrays.asList(tableEnv.listFunctions()).contains("f1"));
-
- tableEnv.sqlUpdate("DROP FUNCTION IF EXISTS default_catalog.default_database.f1");
- assertFalse(Arrays.asList(tableEnv.listFunctions()).contains("f1"));
- }
-
- @Test
- public void testCreateFunctionWithFullPath() {
- String ddl1 = "create function default_catalog.default_database.f2 as" +
- " 'org.apache.flink.function.TestFunction'";
- tableEnv.sqlUpdate(ddl1);
- assertTrue(Arrays.asList(tableEnv.listFunctions()).contains("f2"));
-
- tableEnv.sqlUpdate("DROP FUNCTION IF EXISTS default_catalog.default_database.f2");
- assertFalse(Arrays.asList(tableEnv.listFunctions()).contains("f2"));
- }
-
- @Test
- public void testCreateFunctionWithoutCatalogIdentifier() {
- String ddl1 = "create function default_database.f3 as" +
- " 'org.apache.flink.function.TestFunction'";
- tableEnv.sqlUpdate(ddl1);
- assertTrue(Arrays.asList(tableEnv.listFunctions()).contains("f3"));
-
- tableEnv.sqlUpdate("DROP FUNCTION IF EXISTS default_catalog.default_database.f3");
- assertFalse(Arrays.asList(tableEnv.listFunctions()).contains("f3"));
- }
-
- @Test
- public void testCreateFunctionCatalogNotExists() {
-
- String ddl1 = "create function catalog1.database1.f3 as 'org.apache.flink.function.TestFunction'";
-
- try {
- tableEnv.sqlUpdate(ddl1);
- } catch (Exception e){
- assertTrue(e.getMessage().equals("Catalog catalog1 does not exist"));
- }
- }
-
- @Test
- public void testCreateFunctionDBNotExists() {
- String ddl1 = "create function default_catalog.database1.f3 as 'org.apache.flink.function.TestFunction'";
-
- try {
- tableEnv.sqlUpdate(ddl1);
- } catch (Exception e){
- assertEquals(e.getMessage(), "Could not execute CREATE CATALOG FUNCTION:" +
- " (catalogFunction: [Optional[This is a user-defined function]], identifier:" +
- " [`default_catalog`.`database1`.`f3`], ignoreIfExists: [false])");
- }
- }
-
- @Test
- public void testCreateTemporaryCatalogFunction() {
- String ddl1 = "create temporary function default_catalog.default_database.f4" +
- " as 'org.apache.flink.table.functions.FunctionTestBase$TestUDF'";
-
- String ddl2 = "create temporary function if not exists default_catalog.default_database.f4" +
- " as 'org.apache.flink.table.functions.FunctionTestBase$TestUDF'";
-
- String ddl3 = "drop temporary function default_catalog.default_database.f4";
-
- String ddl4 = "drop temporary function if exists default_catalog.default_database.f4";
-
- tableEnv.sqlUpdate(ddl1);
- assertTrue(Arrays.asList(tableEnv.listFunctions()).contains("f4"));
-
- tableEnv.sqlUpdate(ddl2);
- assertTrue(Arrays.asList(tableEnv.listFunctions()).contains("f4"));
-
- tableEnv.sqlUpdate(ddl3);
- assertFalse(Arrays.asList(tableEnv.listFunctions()).contains("f4"));
-
- tableEnv.sqlUpdate(ddl1);
- try {
- tableEnv.sqlUpdate(ddl1);
- } catch (Exception e) {
- assertTrue(e instanceof ValidationException);
- assertEquals(e.getMessage(),
- "Temporary catalog function `default_catalog`.`default_database`.`f4`" +
- " is already defined");
- }
-
- tableEnv.sqlUpdate(ddl3);
- tableEnv.sqlUpdate(ddl4);
- try {
- tableEnv.sqlUpdate(ddl3);
- } catch (Exception e) {
- assertTrue(e instanceof ValidationException);
- assertEquals(e.getMessage(),
- "Temporary catalog function `default_catalog`.`default_database`.`f4`" +
- " doesn't exist");
- }
- }
-
- @Test
- public void testCreateTemporarySystemFunction() {
- String ddl1 = "create temporary system function default_catalog.default_database.f5" +
- " as 'org.apache.flink.table.functions.FunctionTestBase$TestUDF'";
-
- String ddl2 = "create temporary system function if not exists default_catalog.default_database.f5" +
- " as 'org.apache.flink.table.functions.CatalogFunctionTestBase$TestUDF'";
-
- String ddl3 = "drop temporary system function default_catalog.default_database.f5";
-
- tableEnv.sqlUpdate(ddl1);
- tableEnv.sqlUpdate(ddl2);
- tableEnv.sqlUpdate(ddl3);
- }
-
- @Test
- public void testAlterFunction() throws Exception {
- String create = "create function f3 as 'org.apache.flink.function.TestFunction'";
- String alter = "alter function f3 as 'org.apache.flink.function.TestFunction2'";
-
- ObjectPath objectPath = new ObjectPath("default_database", "f3");
- Catalog catalog = tableEnv.getCatalog("default_catalog").get();
- tableEnv.sqlUpdate(create);
- CatalogFunction beforeUpdate = catalog.getFunction(objectPath);
- assertEquals("org.apache.flink.function.TestFunction", beforeUpdate.getClassName());
-
- tableEnv.sqlUpdate(alter);
- CatalogFunction afterUpdate = catalog.getFunction(objectPath);
- assertEquals("org.apache.flink.function.TestFunction2", afterUpdate.getClassName());
- }
-
- @Test
- public void testAlterFunctionNonExists() {
- String alterUndefinedFunction = "ALTER FUNCTION default_catalog.default_database.f4" +
- " as 'org.apache.flink.function.TestFunction'";
-
- String alterFunctionInWrongCatalog = "ALTER FUNCTION catalog1.default_database.f4 " +
- "as 'org.apache.flink.function.TestFunction'";
-
- String alterFunctionInWrongDB = "ALTER FUNCTION default_catalog.db1.f4 " +
- "as 'org.apache.flink.function.TestFunction'";
-
- try {
- tableEnv.sqlUpdate(alterUndefinedFunction);
- fail();
- } catch (Exception e){
- assertEquals(e.getMessage(),
- "Function default_database.f4 does not exist in Catalog default_catalog.");
- }
-
- try {
- tableEnv.sqlUpdate(alterFunctionInWrongCatalog);
- fail();
- } catch (Exception e) {
- assertTrue(e.getMessage().equals("Catalog catalog1 does not exist"));
- }
-
- try {
- tableEnv.sqlUpdate(alterFunctionInWrongDB);
- fail();
- } catch (Exception e) {
- assertEquals(e.getMessage(), "Function db1.f4 does not exist" +
- " in Catalog default_catalog.");
- }
- }
-
- @Test
- public void testAlterTemporaryCatalogFunction() {
- String alterTemporary = "ALTER TEMPORARY FUNCTION default_catalog.default_database.f4" +
- " as 'org.apache.flink.function.TestFunction'";
-
- try {
- tableEnv.sqlUpdate(alterTemporary);
- fail();
- } catch (Exception e) {
- assertTrue(e.getMessage().equals("Alter temporary catalog function is not supported"));
- }
- }
-
- @Test
- public void testAlterTemporarySystemFunction() {
- String alterTemporary = "ALTER TEMPORARY SYSTEM FUNCTION default_catalog.default_database.f4" +
- " as 'org.apache.flink.function.TestFunction'";
-
- try {
- tableEnv.sqlUpdate(alterTemporary);
- fail();
- } catch (Exception e) {
- assertTrue(e.getMessage().equals("Alter temporary system function is not supported"));
- }
- }
-
- @Test
- public void testDropFunctionNonExists() {
- String dropUndefinedFunction = "DROP FUNCTION default_catalog.default_database.f4";
-
- String dropFunctionInWrongCatalog = "DROP FUNCTION catalog1.default_database.f4";
-
- String dropFunctionInWrongDB = "DROP FUNCTION default_catalog.db1.f4";
-
- try {
- tableEnv.sqlUpdate(dropUndefinedFunction);
- fail();
- } catch (Exception e){
- assertEquals(e.getMessage(),
- "Function default_database.f4 does not exist in Catalog default_catalog.");
- }
-
- try {
- tableEnv.sqlUpdate(dropFunctionInWrongCatalog);
- fail();
- } catch (Exception e) {
- assertTrue(e.getMessage().equals("Catalog catalog1 does not exist"));
- }
-
- try {
- tableEnv.sqlUpdate(dropFunctionInWrongDB);
- fail();
- } catch (Exception e) {
- assertEquals(e.getMessage(),
- "Function db1.f4 does not exist in Catalog default_catalog.");
- }
- }
-
- @Test
- public void testDropTemporaryFunctionNonExits() {
- String dropUndefinedFunction = "DROP TEMPORARY FUNCTION default_catalog.default_database.f4";
- String dropFunctionInWrongCatalog = "DROP TEMPORARY FUNCTION catalog1.default_database.f4";
- String dropFunctionInWrongDB = "DROP TEMPORARY FUNCTION default_catalog.db1.f4";
-
- try {
- tableEnv.sqlUpdate(dropUndefinedFunction);
- fail();
- } catch (Exception e){
- assertEquals(e.getMessage(), "Temporary catalog function" +
- " `default_catalog`.`default_database`.`f4` doesn't exist");
- }
-
- try {
- tableEnv.sqlUpdate(dropFunctionInWrongCatalog);
- fail();
- } catch (Exception e) {
- assertEquals(e.getMessage(), "Temporary catalog function " +
- "`catalog1`.`default_database`.`f4` doesn't exist");
- }
-
- try {
- tableEnv.sqlUpdate(dropFunctionInWrongDB);
- fail();
- } catch (Exception e) {
- assertEquals(e.getMessage(), "Temporary catalog function " +
- "`default_catalog`.`db1`.`f4` doesn't exist");
- }
- }
-
- @Test
- public void testCreateDropTemporaryCatalogFunctionsWithDifferentIdentifier() {
- String createNoCatalogDB = "create temporary function f4" +
- " as 'org.apache.flink.table.functions.FunctionTestBase$TestUDF'";
-
- String dropNoCatalogDB = "drop temporary function f4";
-
- tableEnv.sqlUpdate(createNoCatalogDB);
- tableEnv.sqlUpdate(dropNoCatalogDB);
-
- String createNonExistsCatalog = "create temporary function catalog1.default_database.f4" +
- " as 'org.apache.flink.table.functions.FunctionTestBase$TestUDF'";
-
- String dropNonExistsCatalog = "drop temporary function catalog1.default_database.f4";
-
- tableEnv.sqlUpdate(createNonExistsCatalog);
- tableEnv.sqlUpdate(dropNonExistsCatalog);
-
- String createNonExistsDB = "create temporary function default_catalog.db1.f4" +
- " as 'org.apache.flink.table.functions.FunctionTestBase$TestUDF'";
-
- String dropNonExistsDB = "drop temporary function default_catalog.db1.f4";
-
- tableEnv.sqlUpdate(createNonExistsDB);
- tableEnv.sqlUpdate(dropNonExistsDB);
- }
-
- @Test
- public void testDropTemporarySystemFunction() {
- String ddl1 = "create temporary system function f5" +
- " as 'org.apache.flink.table.functions.FunctionTestBase$TestUDF'";
-
- String ddl2 = "drop temporary system function f5";
-
- String ddl3 = "drop temporary system function if exists f5";
-
- tableEnv.sqlUpdate(ddl1);
- tableEnv.sqlUpdate(ddl2);
- tableEnv.sqlUpdate(ddl3);
-
- try {
- tableEnv.sqlUpdate(ddl2);
- } catch (Exception e) {
- assertEquals(e.getMessage(), "Temporary system function f5 doesn't exist");
- }
- }
-
- @Test
- public void testUseDefinedRegularCatalogFunction() throws Exception {
- String functionDDL = "create function addOne as " +
- "'org.apache.flink.table.functions.FunctionTestBase$TestUDF'";
-
- String dropFunctionDDL = "drop function addOne";
- testUseDefinedCatalogFunction(functionDDL);
- // delete the function
- tableEnv.sqlUpdate(dropFunctionDDL);
- }
-
- @Test
- public void testUseDefinedTemporaryCatalogFunction() throws Exception {
- String functionDDL = "create temporary function addOne as " +
- "'org.apache.flink.table.functions.FunctionTestBase$TestUDF'";
-
- String dropFunctionDDL = "drop temporary function addOne";
- testUseDefinedCatalogFunction(functionDDL);
- // delete the function
- tableEnv.sqlUpdate(dropFunctionDDL);
- }
-
- @Test
- public void testUseDefinedTemporarySystemFunction() throws Exception {
- String functionDDL = "create temporary system function addOne as " +
- "'org.apache.flink.table.functions.FunctionTestBase$TestUDF'";
-
- String dropFunctionDDL = "drop temporary system function addOne";
- testUseDefinedCatalogFunction(functionDDL);
- // delete the function
- tableEnv.sqlUpdate(dropFunctionDDL);
- }
-
- private void testUseDefinedCatalogFunction(String createFunctionDDL) throws Exception {
- List<Row> sourceData = Arrays.asList(
- toRow(1, "1000", 2),
- toRow(2, "1", 3),
- toRow(3, "2000", 4),
- toRow(1, "2", 2),
- toRow(2, "3000", 3)
- );
-
- TestCollectionTableFactory.reset();
- TestCollectionTableFactory.initData(sourceData, new ArrayList<Row>(), -1);
-
- String sourceDDL = "create table t1(a int, b varchar, c int) with ('connector' = 'COLLECTION')";
- String sinkDDL = "create table t2(a int, b varchar, c int) with ('connector' = 'COLLECTION')";
- String query = " insert into t2 select t1.a, t1.b, addOne(t1.a, 1) as c from t1";
-
- tableEnv.sqlUpdate(sourceDDL);
- tableEnv.sqlUpdate(sinkDDL);
- tableEnv.sqlUpdate(createFunctionDDL);
- tableEnv.sqlUpdate(query);
- execute();
-
- Row[] result = TestCollectionTableFactory.RESULT().toArray(new Row[0]);
- Row[] expected = sourceData.toArray(new Row[0]);
- assertArrayEquals(expected, result);
-
- tableEnv.sqlUpdate("drop table t1");
- tableEnv.sqlUpdate("drop table t2");
- }
-
- private Row toRow(Object ... objects) {
- Row row = new Row(objects.length);
- for (int i = 0; i < objects.length; i++) {
- row.setField(i, objects[i]);
- }
-
- return row;
- }
-
- /**
- * Test udf class.
- */
- public static class TestUDF extends ScalarFunction {
-
- public Integer eval(Integer a, Integer b) {
- return a + b;
- }
- }
-}
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/runtime/batch/sql/FunctionITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/runtime/batch/sql/FunctionITCase.java
deleted file mode 100644
index 7e2a10f..0000000
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/runtime/batch/sql/FunctionITCase.java
+++ /dev/null
@@ -1,43 +0,0 @@
-/*
- * 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.runtime.batch.sql;
-
-import org.apache.flink.api.java.ExecutionEnvironment;
-import org.apache.flink.table.api.java.BatchTableEnvironment;
-import org.apache.flink.table.functions.FunctionTestBase;
-
-import org.junit.BeforeClass;
-
-/**
- * Tests for catalog and system function in batch table environment.
- */
-public class FunctionITCase extends FunctionTestBase {
- private static ExecutionEnvironment executionEnvironment = ExecutionEnvironment.getExecutionEnvironment();
-
- @BeforeClass
- public static void setup() {
- BatchTableEnvironment batchTableEnvironment = BatchTableEnvironment.create(executionEnvironment);
- setTableEnv(batchTableEnvironment);
- }
-
- @Override
- public void execute() throws Exception {
- executionEnvironment.execute();
- }
-}
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/runtime/stream/sql/FunctionITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/runtime/stream/sql/FunctionITCase.java
index 045c4b0..6ccfc02 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/runtime/stream/sql/FunctionITCase.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/runtime/stream/sql/FunctionITCase.java
@@ -19,26 +19,431 @@
package org.apache.flink.table.runtime.stream.sql;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.api.java.StreamTableEnvironment;
-import org.apache.flink.table.functions.FunctionTestBase;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.CatalogFunction;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.factories.utils.TestCollectionTableFactory;
+import org.apache.flink.table.functions.ScalarFunction;
+import org.apache.flink.test.util.AbstractTestBase;
+import org.apache.flink.types.Row;
-import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
/**
- * Tests for catalog and system function in stream table environment.
+ * Tests for both catalog and system function.
*/
-public class FunctionITCase extends FunctionTestBase {
- private static StreamExecutionEnvironment streamExecEnvironment;
+public class FunctionITCase extends AbstractTestBase {
+
+ private static final String TEST_FUNCTION = TestUDF.class.getName();
+
+ @Test
+ public void testCreateCatalogFunctionInDefaultCatalog() {
+ TableEnvironment tableEnv = getTableEnvironment();
+ String ddl1 = "create function f1 as 'org.apache.flink.function.TestFunction'";
+ tableEnv.sqlUpdate(ddl1);
+ assertTrue(Arrays.asList(tableEnv.listFunctions()).contains("f1"));
+
+ tableEnv.sqlUpdate("DROP FUNCTION IF EXISTS default_catalog.default_database.f1");
+ assertFalse(Arrays.asList(tableEnv.listFunctions()).contains("f1"));
+ }
+
+ @Test
+ public void testCreateFunctionWithFullPath() {
+ TableEnvironment tableEnv = getTableEnvironment();
+ String ddl1 = "create function default_catalog.default_database.f2 as" +
+ " 'org.apache.flink.function.TestFunction'";
+ tableEnv.sqlUpdate(ddl1);
+ assertTrue(Arrays.asList(tableEnv.listFunctions()).contains("f2"));
+
+ tableEnv.sqlUpdate("DROP FUNCTION IF EXISTS default_catalog.default_database.f2");
+ assertFalse(Arrays.asList(tableEnv.listFunctions()).contains("f2"));
+ }
+
+ @Test
+ public void testCreateFunctionWithoutCatalogIdentifier() {
+ TableEnvironment tableEnv = getTableEnvironment();
+ String ddl1 = "create function default_database.f3 as" +
+ " 'org.apache.flink.function.TestFunction'";
+ tableEnv.sqlUpdate(ddl1);
+ assertTrue(Arrays.asList(tableEnv.listFunctions()).contains("f3"));
+
+ tableEnv.sqlUpdate("DROP FUNCTION IF EXISTS default_catalog.default_database.f3");
+ assertFalse(Arrays.asList(tableEnv.listFunctions()).contains("f3"));
+ }
+
+ @Test
+ public void testCreateFunctionCatalogNotExists() {
+ TableEnvironment tableEnv = getTableEnvironment();
+ String ddl1 = "create function catalog1.database1.f3 as 'org.apache.flink.function.TestFunction'";
+
+ try {
+ tableEnv.sqlUpdate(ddl1);
+ } catch (Exception e){
+ assertEquals("Catalog catalog1 does not exist", e.getMessage());
+ }
+ }
+
+ @Test
+ public void testCreateFunctionDBNotExists() {
+ TableEnvironment tableEnv = getTableEnvironment();
+ String ddl1 = "create function default_catalog.database1.f3 as 'org.apache.flink.function.TestFunction'";
+
+ try {
+ tableEnv.sqlUpdate(ddl1);
+ } catch (Exception e){
+ assertEquals(e.getMessage(), "Could not execute CREATE CATALOG FUNCTION:" +
+ " (catalogFunction: [Optional[This is a user-defined function]], identifier:" +
+ " [`default_catalog`.`database1`.`f3`], ignoreIfExists: [false])");
+ }
+ }
+
+ @Test
+ public void testCreateTemporaryCatalogFunction() {
+ TableEnvironment tableEnv = getTableEnvironment();
+ String ddl1 = "create temporary function default_catalog.default_database.f4" +
+ " as '" + TEST_FUNCTION + "'";
+
+ String ddl2 = "create temporary function if not exists default_catalog.default_database.f4" +
+ " as '" + TEST_FUNCTION + "'";
+
+ String ddl3 = "drop temporary function default_catalog.default_database.f4";
+
+ String ddl4 = "drop temporary function if exists default_catalog.default_database.f4";
+
+ tableEnv.sqlUpdate(ddl1);
+ assertTrue(Arrays.asList(tableEnv.listFunctions()).contains("f4"));
+
+ tableEnv.sqlUpdate(ddl2);
+ assertTrue(Arrays.asList(tableEnv.listFunctions()).contains("f4"));
+
+ tableEnv.sqlUpdate(ddl3);
+ assertFalse(Arrays.asList(tableEnv.listFunctions()).contains("f4"));
+
+ tableEnv.sqlUpdate(ddl1);
+ try {
+ tableEnv.sqlUpdate(ddl1);
+ } catch (Exception e) {
+ assertTrue(e instanceof ValidationException);
+ assertEquals(e.getMessage(),
+ "Temporary catalog function `default_catalog`.`default_database`.`f4`" +
+ " is already defined");
+ }
+
+ tableEnv.sqlUpdate(ddl3);
+ tableEnv.sqlUpdate(ddl4);
+ try {
+ tableEnv.sqlUpdate(ddl3);
+ } catch (Exception e) {
+ assertTrue(e instanceof ValidationException);
+ assertEquals(e.getMessage(),
+ "Temporary catalog function `default_catalog`.`default_database`.`f4`" +
+ " doesn't exist");
+ }
+ }
+
+ @Test
+ public void testCreateTemporarySystemFunction() {
+ TableEnvironment tableEnv = getTableEnvironment();
+ String ddl1 = "create temporary system function default_catalog.default_database.f5" +
+ " as '" + TEST_FUNCTION + "'";
+
+ String ddl2 = "create temporary system function if not exists default_catalog.default_database.f5" +
+ " as 'org.apache.flink.table.functions.CatalogFunctionTestBase$TestUDF'";
+
+ String ddl3 = "drop temporary system function default_catalog.default_database.f5";
+
+ tableEnv.sqlUpdate(ddl1);
+ tableEnv.sqlUpdate(ddl2);
+ tableEnv.sqlUpdate(ddl3);
+ }
+
+ @Test
+ public void testAlterFunction() throws Exception {
+ TableEnvironment tableEnv = getTableEnvironment();
+ String create = "create function f3 as 'org.apache.flink.function.TestFunction'";
+ String alter = "alter function f3 as 'org.apache.flink.function.TestFunction2'";
+
+ ObjectPath objectPath = new ObjectPath("default_database", "f3");
+ assertTrue(tableEnv.getCatalog("default_catalog").isPresent());
+ Catalog catalog = tableEnv.getCatalog("default_catalog").get();
+ tableEnv.sqlUpdate(create);
+ CatalogFunction beforeUpdate = catalog.getFunction(objectPath);
+ assertEquals("org.apache.flink.function.TestFunction", beforeUpdate.getClassName());
+
+ tableEnv.sqlUpdate(alter);
+ CatalogFunction afterUpdate = catalog.getFunction(objectPath);
+ assertEquals("org.apache.flink.function.TestFunction2", afterUpdate.getClassName());
+ }
+
+ @Test
+ public void testAlterFunctionNonExists() {
+ TableEnvironment tableEnv = getTableEnvironment();
+ String alterUndefinedFunction = "ALTER FUNCTION default_catalog.default_database.f4" +
+ " as 'org.apache.flink.function.TestFunction'";
+
+ String alterFunctionInWrongCatalog = "ALTER FUNCTION catalog1.default_database.f4 " +
+ "as 'org.apache.flink.function.TestFunction'";
+
+ String alterFunctionInWrongDB = "ALTER FUNCTION default_catalog.db1.f4 " +
+ "as 'org.apache.flink.function.TestFunction'";
+
+ try {
+ tableEnv.sqlUpdate(alterUndefinedFunction);
+ fail();
+ } catch (Exception e){
+ assertEquals(e.getMessage(),
+ "Function default_database.f4 does not exist in Catalog default_catalog.");
+ }
+
+ try {
+ tableEnv.sqlUpdate(alterFunctionInWrongCatalog);
+ fail();
+ } catch (Exception e) {
+ assertEquals("Catalog catalog1 does not exist", e.getMessage());
+ }
+
+ try {
+ tableEnv.sqlUpdate(alterFunctionInWrongDB);
+ fail();
+ } catch (Exception e) {
+ assertEquals(e.getMessage(), "Function db1.f4 does not exist" +
+ " in Catalog default_catalog.");
+ }
+ }
+
+ @Test
+ public void testAlterTemporaryCatalogFunction() {
+ TableEnvironment tableEnv = getTableEnvironment();
+ String alterTemporary = "ALTER TEMPORARY FUNCTION default_catalog.default_database.f4" +
+ " as 'org.apache.flink.function.TestFunction'";
+
+ try {
+ tableEnv.sqlUpdate(alterTemporary);
+ fail();
+ } catch (Exception e) {
+ assertEquals("Alter temporary catalog function is not supported", e.getMessage());
+ }
+ }
+
+ @Test
+ public void testAlterTemporarySystemFunction() {
+ TableEnvironment tableEnv = getTableEnvironment();
+ String alterTemporary = "ALTER TEMPORARY SYSTEM FUNCTION default_catalog.default_database.f4" +
+ " as 'org.apache.flink.function.TestFunction'";
+
+ try {
+ tableEnv.sqlUpdate(alterTemporary);
+ fail();
+ } catch (Exception e) {
+ assertEquals("Alter temporary system function is not supported", e.getMessage());
+ }
+ }
+
+ @Test
+ public void testDropFunctionNonExists() {
+ TableEnvironment tableEnv = getTableEnvironment();
+ String dropUndefinedFunction = "DROP FUNCTION default_catalog.default_database.f4";
+
+ String dropFunctionInWrongCatalog = "DROP FUNCTION catalog1.default_database.f4";
+
+ String dropFunctionInWrongDB = "DROP FUNCTION default_catalog.db1.f4";
+
+ try {
+ tableEnv.sqlUpdate(dropUndefinedFunction);
+ fail();
+ } catch (Exception e){
+ assertEquals(e.getMessage(),
+ "Function default_database.f4 does not exist in Catalog default_catalog.");
+ }
+
+ try {
+ tableEnv.sqlUpdate(dropFunctionInWrongCatalog);
+ fail();
+ } catch (Exception e) {
+ assertEquals("Catalog catalog1 does not exist", e.getMessage());
+ }
+
+ try {
+ tableEnv.sqlUpdate(dropFunctionInWrongDB);
+ fail();
+ } catch (Exception e) {
+ assertEquals(e.getMessage(),
+ "Function db1.f4 does not exist in Catalog default_catalog.");
+ }
+ }
+
+ @Test
+ public void testDropTemporaryFunctionNonExits() {
+ TableEnvironment tableEnv = getTableEnvironment();
+ String dropUndefinedFunction = "DROP TEMPORARY FUNCTION default_catalog.default_database.f4";
+ String dropFunctionInWrongCatalog = "DROP TEMPORARY FUNCTION catalog1.default_database.f4";
+ String dropFunctionInWrongDB = "DROP TEMPORARY FUNCTION default_catalog.db1.f4";
+
+ try {
+ tableEnv.sqlUpdate(dropUndefinedFunction);
+ fail();
+ } catch (Exception e){
+ assertEquals(e.getMessage(), "Temporary catalog function" +
+ " `default_catalog`.`default_database`.`f4` doesn't exist");
+ }
+
+ try {
+ tableEnv.sqlUpdate(dropFunctionInWrongCatalog);
+ fail();
+ } catch (Exception e) {
+ assertEquals(e.getMessage(), "Temporary catalog function " +
+ "`catalog1`.`default_database`.`f4` doesn't exist");
+ }
+
+ try {
+ tableEnv.sqlUpdate(dropFunctionInWrongDB);
+ fail();
+ } catch (Exception e) {
+ assertEquals(e.getMessage(), "Temporary catalog function " +
+ "`default_catalog`.`db1`.`f4` doesn't exist");
+ }
+ }
+
+ @Test
+ public void testCreateDropTemporaryCatalogFunctionsWithDifferentIdentifier() {
+ TableEnvironment tableEnv = getTableEnvironment();
+ String createNoCatalogDB = "create temporary function f4" +
+ " as '" + TEST_FUNCTION + "'";
+
+ String dropNoCatalogDB = "drop temporary function f4";
+
+ tableEnv.sqlUpdate(createNoCatalogDB);
+ tableEnv.sqlUpdate(dropNoCatalogDB);
+
+ String createNonExistsCatalog = "create temporary function catalog1.default_database.f4" +
+ " as '" + TEST_FUNCTION + "'";
+
+ String dropNonExistsCatalog = "drop temporary function catalog1.default_database.f4";
+
+ tableEnv.sqlUpdate(createNonExistsCatalog);
+ tableEnv.sqlUpdate(dropNonExistsCatalog);
+
+ String createNonExistsDB = "create temporary function default_catalog.db1.f4" +
+ " as '" + TEST_FUNCTION + "'";
+
+ String dropNonExistsDB = "drop temporary function default_catalog.db1.f4";
+
+ tableEnv.sqlUpdate(createNonExistsDB);
+ tableEnv.sqlUpdate(dropNonExistsDB);
+ }
+
+ @Test
+ public void testDropTemporarySystemFunction() {
+ TableEnvironment tableEnv = getTableEnvironment();
+ String ddl1 = "create temporary system function f5" +
+ " as '" + TEST_FUNCTION + "'";
+
+ String ddl2 = "drop temporary system function f5";
+
+ String ddl3 = "drop temporary system function if exists f5";
+
+ tableEnv.sqlUpdate(ddl1);
+ tableEnv.sqlUpdate(ddl2);
+ tableEnv.sqlUpdate(ddl3);
+
+ try {
+ tableEnv.sqlUpdate(ddl2);
+ } catch (Exception e) {
+ assertEquals(e.getMessage(), "Temporary system function f5 doesn't exist");
+ }
+ }
+
+ @Test
+ public void testUserDefinedRegularCatalogFunction() throws Exception {
+ TableEnvironment tableEnv = getTableEnvironment();
+ String functionDDL = "create function addOne as " +
+ "'" + TEST_FUNCTION + "'";
+
+ String dropFunctionDDL = "drop function addOne";
+ testUserDefinedCatalogFunction(tableEnv, functionDDL);
+ // delete the function
+ tableEnv.sqlUpdate(dropFunctionDDL);
+ }
+
+ @Test
+ public void testUserDefinedTemporaryCatalogFunction() throws Exception {
+ TableEnvironment tableEnv = getTableEnvironment();
+ String functionDDL = "create temporary function addOne as " +
+ "'" + TEST_FUNCTION + "'";
+
+ String dropFunctionDDL = "drop temporary function addOne";
+ testUserDefinedCatalogFunction(tableEnv, functionDDL);
+ // delete the function
+ tableEnv.sqlUpdate(dropFunctionDDL);
+ }
+
+ @Test
+ public void testUserDefinedTemporarySystemFunction() throws Exception {
+ TableEnvironment tableEnv = getTableEnvironment();
+ String functionDDL = "create temporary system function addOne as " +
+ "'" + TEST_FUNCTION + "'";
+
+ String dropFunctionDDL = "drop temporary system function addOne";
+ testUserDefinedCatalogFunction(tableEnv, functionDDL);
+ // delete the function
+ tableEnv.sqlUpdate(dropFunctionDDL);
+ }
+
+ /**
+ * Test udf class.
+ */
+ public static class TestUDF extends ScalarFunction {
+
+ public Integer eval(Integer a, Integer b) {
+ return a + b;
+ }
+ }
+
+ private void testUserDefinedCatalogFunction(TableEnvironment tableEnv, String createFunctionDDL) throws Exception {
+ List<Row> sourceData = Arrays.asList(
+ Row.of(1, "1000", 2),
+ Row.of(2, "1", 3),
+ Row.of(3, "2000", 4),
+ Row.of(1, "2", 2),
+ Row.of(2, "3000", 3)
+ );
+
+ TestCollectionTableFactory.reset();
+ TestCollectionTableFactory.initData(sourceData, new ArrayList<>(), -1);
+
+ String sourceDDL = "create table t1(a int, b varchar, c int) with ('connector' = 'COLLECTION')";
+ String sinkDDL = "create table t2(a int, b varchar, c int) with ('connector' = 'COLLECTION')";
+ String query = " insert into t2 select t1.a, t1.b, addOne(t1.a, 1) as c from t1";
+
+ tableEnv.sqlUpdate(sourceDDL);
+ tableEnv.sqlUpdate(sinkDDL);
+ tableEnv.sqlUpdate(createFunctionDDL);
+ tableEnv.sqlUpdate(query);
+ tableEnv.execute("Test job");
+
+ Row[] result = TestCollectionTableFactory.RESULT().toArray(new Row[0]);
+ Row[] expected = sourceData.toArray(new Row[0]);
+ assertArrayEquals(expected, result);
- @BeforeClass
- public static void setup() {
- streamExecEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
- StreamTableEnvironment streamTableEnvironment = StreamTableEnvironment.create(streamExecEnvironment);
- setTableEnv(streamTableEnvironment);
+ tableEnv.sqlUpdate("drop table t1");
+ tableEnv.sqlUpdate("drop table t2");
}
- @Override
- public void execute() throws Exception {
- streamExecEnvironment.execute();
+ private TableEnvironment getTableEnvironment() {
+ StreamExecutionEnvironment streamExecEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
+ return StreamTableEnvironment.create(streamExecEnvironment);
}
}
[flink] 02/04: [FLINK-12283][table] Relax UDF constraints by using
the ClosureCleaner
Posted by tw...@apache.org.
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 91629d67a6c42852ab3d218238f3dce5f55ca752
Author: Timo Walther <tw...@apache.org>
AuthorDate: Mon Dec 9 17:34:12 2019 +0100
[FLINK-12283][table] Relax UDF constraints by using the ClosureCleaner
This relaxes the UDF constraints as much as possible by using the
ClosureCleaner. For Java users, this is very convenient. For Scala users, it
improves the current status but we still need to fix FLINK-15162.
It ensures that the validation and cleaning happens at exactly 2 locations.
Either during registration in the function catalog or during resolution of
inline, unregistered functions.
This closes #10519.
---
.../client/gateway/local/ExecutionContext.java | 27 ++++---
.../java/internal/StreamTableEnvironmentImpl.java | 2 +-
.../internal/StreamTableEnvironmentImplTest.java | 5 +-
.../table/api/internal/TableEnvironmentImpl.java | 6 +-
.../flink/table/catalog/FunctionCatalog.java | 36 ++++-----
.../expressions/resolver/ExpressionResolver.java | 18 ++++-
.../resolver/rules/ResolveCallByArgumentsRule.java | 50 +++++++++++-
.../expressions/resolver/rules/ResolverRule.java | 6 ++
.../table/functions/UserDefinedFunctionHelper.java | 40 ++++++++--
.../operations/utils/OperationTreeBuilder.java | 36 +++++++--
.../flink/table/catalog/FunctionCatalogTest.java | 2 +
.../flink/table/utils/TableEnvironmentMock.java | 12 ++-
.../internal/StreamTableEnvironmentImpl.scala | 2 +-
.../internal/StreamTableEnvironmentImplTest.scala | 5 +-
.../functions/AggregateFunctionDefinition.java | 4 +
.../table/functions/ScalarFunctionDefinition.java | 4 +
.../TableAggregateFunctionDefinition.java | 4 +
.../table/functions/TableFunctionDefinition.java | 4 +
.../flink/table/functions/UserDefinedFunction.java | 2 +-
.../expressions/CallExpressionResolver.java | 1 +
.../planner/codegen/CodeGeneratorContext.scala | 2 +-
.../flink/table/planner/expressions/call.scala | 4 -
.../operations/SqlToOperationConverterTest.java | 5 +-
.../codegen/WatermarkGeneratorCodeGenTest.scala | 5 +-
.../metadata/AggCallSelectivityEstimatorTest.scala | 3 +-
.../plan/metadata/FlinkRelMdHandlerTestBase.scala | 2 +-
.../plan/metadata/SelectivityEstimatorTest.scala | 2 +-
.../planner/plan/utils/RexNodeExtractorTest.scala | 7 +-
.../planner/runtime/stream/table/CalcITCase.scala | 77 ++++++++++++++++++-
.../flink/table/planner/utils/TableTestBase.scala | 2 +-
.../flink/table/api/internal/TableEnvImpl.scala | 3 +-
.../apache/flink/table/codegen/CodeGenerator.scala | 2 +-
.../org/apache/flink/table/expressions/call.scala | 4 -
.../PushFilterIntoTableSourceScanRule.scala | 4 +-
.../table/sqlexec/SqlToOperationConverterTest.java | 5 +-
.../api/stream/StreamTableEnvironmentTest.scala | 2 +-
.../flink/table/api/stream/sql/AggregateTest.scala | 5 +-
.../flink/table/plan/RexProgramExtractorTest.scala | 2 +
.../table/runtime/stream/table/CalcITCase.scala | 88 ++++++++++++++++++++++
.../apache/flink/table/utils/TableTestBase.scala | 2 +-
40 files changed, 405 insertions(+), 87 deletions(-)
diff --git a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/ExecutionContext.java b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/ExecutionContext.java
index 26f5322..8a1f082 100644
--- a/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/ExecutionContext.java
+++ b/flink-table/flink-sql-client/src/main/java/org/apache/flink/table/client/gateway/local/ExecutionContext.java
@@ -393,13 +393,12 @@ public class ExecutionContext<ClusterID> {
private static TableEnvironment createStreamTableEnvironment(
StreamExecutionEnvironment env,
EnvironmentSettings settings,
+ TableConfig config,
Executor executor,
CatalogManager catalogManager,
ModuleManager moduleManager,
FunctionCatalog functionCatalog) {
- final TableConfig config = TableConfig.getDefault();
-
final Map<String, String> plannerProperties = settings.toPlannerProperties();
final Planner planner = ComponentFactoryService.find(PlannerFactory.class, plannerProperties)
.create(plannerProperties, executor, config, functionCatalog, catalogManager);
@@ -441,6 +440,10 @@ public class ExecutionContext<ClusterID> {
//--------------------------------------------------------------------------------------------------------------
// Step.1 Create environments
//--------------------------------------------------------------------------------------------------------------
+ // Step 1.0 Initialize the table configuration.
+ final TableConfig config = new TableConfig();
+ environment.getConfiguration().asMap().forEach((k, v) ->
+ config.getConfiguration().setString(k, v));
// Step 1.1 Initialize the CatalogManager if required.
final CatalogManager catalogManager = new CatalogManager(
settings.getBuiltInCatalogName(),
@@ -450,12 +453,12 @@ public class ExecutionContext<ClusterID> {
// Step 1.2 Initialize the ModuleManager if required.
final ModuleManager moduleManager = new ModuleManager();
// Step 1.3 Initialize the FunctionCatalog if required.
- final FunctionCatalog functionCatalog = new FunctionCatalog(catalogManager, moduleManager);
+ final FunctionCatalog functionCatalog = new FunctionCatalog(config, catalogManager, moduleManager);
// Step 1.4 Set up session state.
- this.sessionState = SessionState.of(catalogManager, moduleManager, functionCatalog);
+ this.sessionState = SessionState.of(config, catalogManager, moduleManager, functionCatalog);
// Must initialize the table environment before actually the
- createTableEnvironment(settings, catalogManager, moduleManager, functionCatalog);
+ createTableEnvironment(settings, config, catalogManager, moduleManager, functionCatalog);
//--------------------------------------------------------------------------------------------------------------
// Step.2 Create modules and load them into the TableEnvironment.
@@ -487,6 +490,7 @@ public class ExecutionContext<ClusterID> {
this.sessionState = sessionState;
createTableEnvironment(
settings,
+ sessionState.config,
sessionState.catalogManager,
sessionState.moduleManager,
sessionState.functionCatalog);
@@ -495,6 +499,7 @@ public class ExecutionContext<ClusterID> {
private void createTableEnvironment(
EnvironmentSettings settings,
+ TableConfig config,
CatalogManager catalogManager,
ModuleManager moduleManager,
FunctionCatalog functionCatalog) {
@@ -507,6 +512,7 @@ public class ExecutionContext<ClusterID> {
tableEnv = createStreamTableEnvironment(
streamExecEnv,
settings,
+ config,
executor,
catalogManager,
moduleManager,
@@ -517,15 +523,12 @@ public class ExecutionContext<ClusterID> {
executor = null;
tableEnv = new BatchTableEnvironmentImpl(
execEnv,
- TableConfig.getDefault(),
+ config,
catalogManager,
moduleManager);
} else {
throw new SqlExecutionException("Unsupported execution type specified.");
}
- // set table configuration
- environment.getConfiguration().asMap().forEach((k, v) ->
- tableEnv.getConfig().getConfiguration().setString(k, v));
}
private void initializeCatalogs() {
@@ -760,24 +763,28 @@ public class ExecutionContext<ClusterID> {
/** Represents the state that should be reused in one session. **/
public static class SessionState {
+ public final TableConfig config;
public final CatalogManager catalogManager;
public final ModuleManager moduleManager;
public final FunctionCatalog functionCatalog;
private SessionState(
+ TableConfig config,
CatalogManager catalogManager,
ModuleManager moduleManager,
FunctionCatalog functionCatalog) {
+ this.config = config;
this.catalogManager = catalogManager;
this.moduleManager = moduleManager;
this.functionCatalog = functionCatalog;
}
public static SessionState of(
+ TableConfig config,
CatalogManager catalogManager,
ModuleManager moduleManager,
FunctionCatalog functionCatalog) {
- return new SessionState(catalogManager, moduleManager, functionCatalog);
+ return new SessionState(config, catalogManager, moduleManager, functionCatalog);
}
}
}
diff --git a/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/api/java/internal/StreamTableEnvironmentImpl.java b/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/api/java/internal/StreamTableEnvironmentImpl.java
index 760a934..d269fc4 100644
--- a/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/api/java/internal/StreamTableEnvironmentImpl.java
+++ b/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/api/java/internal/StreamTableEnvironmentImpl.java
@@ -110,7 +110,7 @@ public final class StreamTableEnvironmentImpl extends TableEnvironmentImpl imple
ModuleManager moduleManager = new ModuleManager();
- FunctionCatalog functionCatalog = new FunctionCatalog(catalogManager, moduleManager);
+ FunctionCatalog functionCatalog = new FunctionCatalog(tableConfig, catalogManager, moduleManager);
Map<String, String> executorProperties = settings.toExecutorProperties();
Executor executor = lookupExecutor(executorProperties, executionEnvironment);
diff --git a/flink-table/flink-table-api-java-bridge/src/test/java/org/apache/flink/table/api/java/internal/StreamTableEnvironmentImplTest.java b/flink-table/flink-table-api-java-bridge/src/test/java/org/apache/flink/table/api/java/internal/StreamTableEnvironmentImplTest.java
index 1a0cf5a..3ddf5c8 100644
--- a/flink-table/flink-table-api-java-bridge/src/test/java/org/apache/flink/table/api/java/internal/StreamTableEnvironmentImplTest.java
+++ b/flink-table/flink-table-api-java-bridge/src/test/java/org/apache/flink/table/api/java/internal/StreamTableEnvironmentImplTest.java
@@ -90,13 +90,14 @@ public class StreamTableEnvironmentImplTest {
private StreamTableEnvironmentImpl getStreamTableEnvironment(
StreamExecutionEnvironment env,
DataStreamSource<Integer> elements) {
+ TableConfig config = new TableConfig();
CatalogManager catalogManager = new CatalogManager("cat", new GenericInMemoryCatalog("cat", "db"));
ModuleManager moduleManager = new ModuleManager();
return new StreamTableEnvironmentImpl(
catalogManager,
moduleManager,
- new FunctionCatalog(catalogManager, moduleManager),
- new TableConfig(),
+ new FunctionCatalog(config, catalogManager, moduleManager),
+ config,
env,
new TestPlanner(elements.getTransformation()),
new ExecutorMock(),
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
index 920124e..adcd639 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
@@ -169,6 +169,7 @@ public class TableEnvironmentImpl implements TableEnvironment {
this.planner = planner;
this.parser = planner.getParser();
this.operationTreeBuilder = OperationTreeBuilder.create(
+ tableConfig,
functionCatalog,
path -> {
try {
@@ -189,18 +190,19 @@ public class TableEnvironmentImpl implements TableEnvironment {
public static TableEnvironmentImpl create(EnvironmentSettings settings) {
+ TableConfig tableConfig = new TableConfig();
+
CatalogManager catalogManager = new CatalogManager(
settings.getBuiltInCatalogName(),
new GenericInMemoryCatalog(settings.getBuiltInCatalogName(), settings.getBuiltInDatabaseName()));
ModuleManager moduleManager = new ModuleManager();
- FunctionCatalog functionCatalog = new FunctionCatalog(catalogManager, moduleManager);
+ FunctionCatalog functionCatalog = new FunctionCatalog(tableConfig, catalogManager, moduleManager);
Map<String, String> executorProperties = settings.toExecutorProperties();
Executor executor = ComponentFactoryService.find(ExecutorFactory.class, executorProperties)
.create(executorProperties);
- TableConfig tableConfig = new TableConfig();
Map<String, String> plannerProperties = settings.toPlannerProperties();
Planner planner = ComponentFactoryService.find(PlannerFactory.class, plannerProperties)
.create(plannerProperties, executor, tableConfig, functionCatalog, catalogManager);
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 be94b0d..1838c13 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
@@ -20,6 +20,7 @@ package org.apache.flink.table.catalog;
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.api.TableException;
import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
@@ -55,6 +56,7 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
*/
@Internal
public class FunctionCatalog implements FunctionLookup {
+ private final TableConfig config;
private final CatalogManager catalogManager;
private final ModuleManager moduleManager;
@@ -66,7 +68,11 @@ public class FunctionCatalog implements FunctionLookup {
*/
private PlannerTypeInferenceUtil plannerTypeInferenceUtil;
- public FunctionCatalog(CatalogManager catalogManager, ModuleManager moduleManager) {
+ public FunctionCatalog(
+ TableConfig config,
+ CatalogManager catalogManager,
+ ModuleManager moduleManager) {
+ this.config = checkNotNull(config);
this.catalogManager = checkNotNull(catalogManager);
this.moduleManager = checkNotNull(moduleManager);
}
@@ -76,7 +82,8 @@ public class FunctionCatalog implements FunctionLookup {
}
public void registerTempSystemScalarFunction(String name, ScalarFunction function) {
- UserDefinedFunctionHelper.validateInstantiation(function.getClass());
+ UserDefinedFunctionHelper.prepareFunction(config, function);
+
registerTempSystemFunction(
name,
new ScalarFunctionDefinition(name, function)
@@ -87,10 +94,7 @@ public class FunctionCatalog implements FunctionLookup {
String name,
TableFunction<T> function,
TypeInformation<T> resultType) {
- // check if class not Scala object
- UserDefinedFunctionHelper.validateNotSingleton(function.getClass());
- // check if class could be instantiated
- UserDefinedFunctionHelper.validateInstantiation(function.getClass());
+ UserDefinedFunctionHelper.prepareFunction(config, function);
registerTempSystemFunction(
name,
@@ -106,10 +110,7 @@ public class FunctionCatalog implements FunctionLookup {
UserDefinedAggregateFunction<T, ACC> function,
TypeInformation<T> resultType,
TypeInformation<ACC> accType) {
- // check if class not Scala object
- UserDefinedFunctionHelper.validateNotSingleton(function.getClass());
- // check if class could be instantiated
- UserDefinedFunctionHelper.validateInstantiation(function.getClass());
+ UserDefinedFunctionHelper.prepareFunction(config, function);
final FunctionDefinition definition;
if (function instanceof AggregateFunction) {
@@ -135,7 +136,8 @@ public class FunctionCatalog implements FunctionLookup {
}
public void registerTempCatalogScalarFunction(ObjectIdentifier oi, ScalarFunction function) {
- UserDefinedFunctionHelper.validateInstantiation(function.getClass());
+ UserDefinedFunctionHelper.prepareFunction(config, function);
+
registerTempCatalogFunction(
oi,
new ScalarFunctionDefinition(oi.getObjectName(), function)
@@ -146,10 +148,7 @@ public class FunctionCatalog implements FunctionLookup {
ObjectIdentifier oi,
TableFunction<T> function,
TypeInformation<T> resultType) {
- // check if class not Scala object
- UserDefinedFunctionHelper.validateNotSingleton(function.getClass());
- // check if class could be instantiated
- UserDefinedFunctionHelper.validateInstantiation(function.getClass());
+ UserDefinedFunctionHelper.prepareFunction(config, function);
registerTempCatalogFunction(
oi,
@@ -165,10 +164,7 @@ public class FunctionCatalog implements FunctionLookup {
UserDefinedAggregateFunction<T, ACC> function,
TypeInformation<T> resultType,
TypeInformation<ACC> accType) {
- // check if class not Scala object
- UserDefinedFunctionHelper.validateNotSingleton(function.getClass());
- // check if class could be instantiated
- UserDefinedFunctionHelper.validateInstantiation(function.getClass());
+ UserDefinedFunctionHelper.prepareFunction(config, function);
final FunctionDefinition definition;
if (function instanceof AggregateFunction) {
@@ -378,7 +374,7 @@ public class FunctionCatalog implements FunctionLookup {
)).orElseGet(() -> resolvePreciseFunctionReference(oi));
}
- @Override
+ @Override
public PlannerTypeInferenceUtil getPlannerTypeInferenceUtil() {
Preconditions.checkNotNull(
plannerTypeInferenceUtil,
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/resolver/ExpressionResolver.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/resolver/ExpressionResolver.java
index 1763404..78a18a7 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/resolver/ExpressionResolver.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/resolver/ExpressionResolver.java
@@ -21,6 +21,7 @@ package org.apache.flink.table.expressions.resolver;
import org.apache.flink.annotation.Internal;
import org.apache.flink.table.api.GroupWindow;
import org.apache.flink.table.api.OverWindow;
+import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.api.TableException;
import org.apache.flink.table.catalog.FunctionLookup;
import org.apache.flink.table.expressions.CallExpression;
@@ -95,6 +96,8 @@ public class ExpressionResolver {
private static final VerifyResolutionVisitor VERIFY_RESOLUTION_VISITOR = new VerifyResolutionVisitor();
+ private final TableConfig config;
+
private final FieldReferenceLookup fieldLookup;
private final TableReferenceLookup tableLookup;
@@ -108,11 +111,13 @@ public class ExpressionResolver {
private final Map<Expression, LocalOverWindow> localOverWindows;
private ExpressionResolver(
+ TableConfig config,
TableReferenceLookup tableLookup,
FunctionLookup functionLookup,
FieldReferenceLookup fieldLookup,
List<OverWindow> localOverWindows,
List<LocalReferenceExpression> localReferences) {
+ this.config = Preconditions.checkNotNull(config);
this.tableLookup = Preconditions.checkNotNull(tableLookup);
this.fieldLookup = Preconditions.checkNotNull(fieldLookup);
this.functionLookup = Preconditions.checkNotNull(functionLookup);
@@ -128,16 +133,18 @@ public class ExpressionResolver {
* Creates a builder for {@link ExpressionResolver}. One can add additional properties to the resolver
* like e.g. {@link GroupWindow} or {@link OverWindow}. You can also add additional {@link ResolverRule}.
*
+ * @param config general configuration
* @param tableCatalog a way to lookup a table reference by name
* @param functionLookup a way to lookup call by name
* @param inputs inputs to use for field resolution
* @return builder for resolver
*/
public static ExpressionResolverBuilder resolverFor(
+ TableConfig config,
TableReferenceLookup tableCatalog,
FunctionLookup functionLookup,
QueryOperation... inputs) {
- return new ExpressionResolverBuilder(inputs, tableCatalog, functionLookup);
+ return new ExpressionResolverBuilder(inputs, config, tableCatalog, functionLookup);
}
/**
@@ -242,6 +249,11 @@ public class ExpressionResolver {
private class ExpressionResolverContext implements ResolverRule.ResolutionContext {
@Override
+ public TableConfig configuration() {
+ return config;
+ }
+
+ @Override
public FieldReferenceLookup referenceLookup() {
return fieldLookup;
}
@@ -341,6 +353,7 @@ public class ExpressionResolver {
*/
public static class ExpressionResolverBuilder {
+ private final TableConfig config;
private final List<QueryOperation> queryOperations;
private final TableReferenceLookup tableCatalog;
private final FunctionLookup functionLookup;
@@ -349,8 +362,10 @@ public class ExpressionResolver {
private ExpressionResolverBuilder(
QueryOperation[] queryOperations,
+ TableConfig config,
TableReferenceLookup tableCatalog,
FunctionLookup functionLookup) {
+ this.config = config;
this.queryOperations = Arrays.asList(queryOperations);
this.tableCatalog = tableCatalog;
this.functionLookup = functionLookup;
@@ -368,6 +383,7 @@ public class ExpressionResolver {
public ExpressionResolver build() {
return new ExpressionResolver(
+ config,
tableCatalog,
functionLookup,
new FieldReferenceLookup(queryOperations),
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/resolver/rules/ResolveCallByArgumentsRule.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/resolver/rules/ResolveCallByArgumentsRule.java
index 122b207..a31efa4 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/resolver/rules/ResolveCallByArgumentsRule.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/resolver/rules/ResolveCallByArgumentsRule.java
@@ -29,10 +29,16 @@ import org.apache.flink.table.expressions.Expression;
import org.apache.flink.table.expressions.ResolvedExpression;
import org.apache.flink.table.expressions.UnresolvedCallExpression;
import org.apache.flink.table.expressions.ValueLiteralExpression;
+import org.apache.flink.table.functions.AggregateFunctionDefinition;
import org.apache.flink.table.functions.BuiltInFunctionDefinition;
import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
import org.apache.flink.table.functions.FunctionDefinition;
import org.apache.flink.table.functions.FunctionIdentifier;
+import org.apache.flink.table.functions.ScalarFunctionDefinition;
+import org.apache.flink.table.functions.TableAggregateFunctionDefinition;
+import org.apache.flink.table.functions.TableFunctionDefinition;
+import org.apache.flink.table.functions.UserDefinedFunction;
+import org.apache.flink.table.functions.UserDefinedFunctionHelper;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.inference.CallContext;
import org.apache.flink.table.types.inference.TypeInference;
@@ -58,13 +64,14 @@ import static org.apache.flink.table.types.utils.TypeConversions.fromLegacyInfoT
/**
* This rule checks if a {@link UnresolvedCallExpression} can work with the given arguments and infers
- * the output data type. All function calls are resolved {@link CallExpression} after applying this
- * rule.
+ * the output data type. All function calls are resolved {@link CallExpression} after applying this rule.
*
* <p>This rule also resolves {@code flatten()} calls on composite types.
*
* <p>If the call expects different types of arguments, but the given arguments have types that can
* be casted, a {@link BuiltInFunctionDefinitions#CAST} expression is inserted.
+ *
+ * <p>It validates and prepares inline, unregistered {@link UserDefinedFunction}s.
*/
@Internal
final class ResolveCallByArgumentsRule implements ResolverRule {
@@ -89,7 +96,7 @@ final class ResolveCallByArgumentsRule implements ResolverRule {
@Override
public List<ResolvedExpression> visit(UnresolvedCallExpression unresolvedCall) {
- final FunctionDefinition definition = unresolvedCall.getFunctionDefinition();
+ final FunctionDefinition definition = prepareUserDefinedFunction(unresolvedCall.getFunctionDefinition());
final String name = unresolvedCall.getFunctionIdentifier()
.map(FunctionIdentifier::toString)
@@ -220,6 +227,43 @@ final class ResolveCallByArgumentsRule implements ResolverRule {
})
.collect(Collectors.toList());
}
+
+ /**
+ * Validates and cleans an inline, unregistered {@link UserDefinedFunction}.
+ */
+ private FunctionDefinition prepareUserDefinedFunction(FunctionDefinition definition) {
+ if (definition instanceof ScalarFunctionDefinition) {
+ final ScalarFunctionDefinition sf = (ScalarFunctionDefinition) definition;
+ UserDefinedFunctionHelper.prepareFunction(resolutionContext.configuration(), sf.getScalarFunction());
+ return new ScalarFunctionDefinition(
+ sf.getName(),
+ sf.getScalarFunction());
+ } else if (definition instanceof TableFunctionDefinition) {
+ final TableFunctionDefinition tf = (TableFunctionDefinition) definition;
+ UserDefinedFunctionHelper.prepareFunction(resolutionContext.configuration(), tf.getTableFunction());
+ return new TableFunctionDefinition(
+ tf.getName(),
+ tf.getTableFunction(),
+ tf.getResultType());
+ } else if (definition instanceof AggregateFunctionDefinition) {
+ final AggregateFunctionDefinition af = (AggregateFunctionDefinition) definition;
+ UserDefinedFunctionHelper.prepareFunction(resolutionContext.configuration(), af.getAggregateFunction());
+ return new AggregateFunctionDefinition(
+ af.getName(),
+ af.getAggregateFunction(),
+ af.getResultTypeInfo(),
+ af.getAccumulatorTypeInfo());
+ } else if (definition instanceof TableAggregateFunctionDefinition) {
+ final TableAggregateFunctionDefinition taf = (TableAggregateFunctionDefinition) definition;
+ UserDefinedFunctionHelper.prepareFunction(resolutionContext.configuration(), taf.getTableAggregateFunction());
+ return new TableAggregateFunctionDefinition(
+ taf.getName(),
+ taf.getTableAggregateFunction(),
+ taf.getResultTypeInfo(),
+ taf.getAccumulatorTypeInfo());
+ }
+ return definition;
+ }
}
// --------------------------------------------------------------------------------------------
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/resolver/rules/ResolverRule.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/resolver/rules/ResolverRule.java
index 0a7fed0..d8525c8 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/resolver/rules/ResolverRule.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/resolver/rules/ResolverRule.java
@@ -19,6 +19,7 @@
package org.apache.flink.table.expressions.resolver.rules;
import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.catalog.FunctionLookup;
import org.apache.flink.table.expressions.Expression;
import org.apache.flink.table.expressions.LocalReferenceExpression;
@@ -48,6 +49,11 @@ public interface ResolverRule {
interface ResolutionContext {
/**
+ * Access to configuration.
+ */
+ TableConfig configuration();
+
+ /**
* Access to available {@link org.apache.flink.table.expressions.FieldReferenceExpression} in inputs.
*/
FieldReferenceLookup referenceLookup();
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/functions/UserDefinedFunctionHelper.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/functions/UserDefinedFunctionHelper.java
index 6069cb9..35398ec 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/functions/UserDefinedFunctionHelper.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/functions/UserDefinedFunctionHelper.java
@@ -21,6 +21,7 @@ package org.apache.flink.table.functions;
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.ExecutionConfig.ClosureCleanerLevel;
import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.ClosureCleaner;
import org.apache.flink.api.java.typeutils.TypeExtractor;
import org.apache.flink.configuration.PipelineOptions;
import org.apache.flink.table.api.TableConfig;
@@ -146,9 +147,20 @@ public class UserDefinedFunctionHelper {
}
/**
+ * Prepares a {@link UserDefinedFunction} for usage in the API.
+ */
+ public static void prepareFunction(TableConfig config, UserDefinedFunction function) {
+ if (function instanceof TableFunction) {
+ UserDefinedFunctionHelper.validateNotSingleton(function.getClass());
+ }
+ UserDefinedFunctionHelper.validateInstantiation(function.getClass());
+ UserDefinedFunctionHelper.cleanFunction(config, function);
+ }
+
+ /**
* Checks if a user-defined function can be easily instantiated.
*/
- public static void validateInstantiation(Class<?> clazz) {
+ private static void validateInstantiation(Class<?> clazz) {
if (!InstantiationUtil.isPublic(clazz)) {
throw new ValidationException(String.format("Function class %s is not public.", clazz.getCanonicalName()));
} else if (!InstantiationUtil.isProperClass(clazz)) {
@@ -159,18 +171,36 @@ public class UserDefinedFunctionHelper {
}
/**
- * Check whether this is a Scala object. It is forbidden to use {@link TableFunction} implemented
- * by a Scala object, since concurrent risks.
+ * Check whether this is a Scala object. Using Scala objects can lead to concurrency issues,
+ * e.g., due to a shared collector.
*/
- public static void validateNotSingleton(Class<?> clazz) {
+ private static void validateNotSingleton(Class<?> clazz) {
// TODO it is not a good way to check singleton. Maybe improve it further.
if (Arrays.stream(clazz.getFields()).anyMatch(f -> f.getName().equals("MODULE$"))) {
throw new ValidationException(String.format(
- "TableFunction implemented by class %s is a Scala object. This is forbidden because of concurrency" +
+ "Function implemented by class %s is a Scala object. This is forbidden because of concurrency" +
" problems when using them.", clazz.getCanonicalName()));
}
}
+ /**
+ * Modifies a function instance by removing any reference to outer classes. This enables
+ * non-static inner function classes.
+ */
+ private static void cleanFunction(TableConfig config, UserDefinedFunction function) {
+ final ClosureCleanerLevel level = config.getConfiguration().get(PipelineOptions.CLOSURE_CLEANER_LEVEL);
+ try {
+ ClosureCleaner.clean(function, level, true);
+ } catch (Throwable t) {
+ throw new ValidationException(
+ String.format(
+ "Function class '%s' is not serializable. Make sure that the class is self-contained " +
+ "(i.e. no references to outer classes) and all inner fields are serializable as well.",
+ function.getClass()),
+ t);
+ }
+ }
+
private UserDefinedFunctionHelper() {
}
}
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/utils/OperationTreeBuilder.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/utils/OperationTreeBuilder.java
index a27aeba..05fa8b3 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/utils/OperationTreeBuilder.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/operations/utils/OperationTreeBuilder.java
@@ -23,6 +23,7 @@ import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.table.api.GroupWindow;
import org.apache.flink.table.api.OverWindow;
+import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.api.TableException;
import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.catalog.FunctionLookup;
@@ -85,6 +86,7 @@ import static org.apache.flink.table.operations.SetQueryOperation.SetQueryOperat
@Internal
public final class OperationTreeBuilder {
+ private final TableConfig config;
private final FunctionLookup functionCatalog;
private final TableReferenceLookup tableReferenceLookup;
private final LookupCallResolver lookupResolver;
@@ -100,6 +102,7 @@ public final class OperationTreeBuilder {
private final JoinOperationFactory joinOperationFactory;
private OperationTreeBuilder(
+ TableConfig config,
FunctionLookup functionLookup,
TableReferenceLookup tableReferenceLookup,
ProjectionOperationFactory projectionOperationFactory,
@@ -108,6 +111,7 @@ public final class OperationTreeBuilder {
SetOperationFactory setOperationFactory,
AggregateOperationFactory aggregateOperationFactory,
JoinOperationFactory joinOperationFactory) {
+ this.config = config;
this.functionCatalog = functionLookup;
this.tableReferenceLookup = tableReferenceLookup;
this.projectionOperationFactory = projectionOperationFactory;
@@ -120,10 +124,12 @@ public final class OperationTreeBuilder {
}
public static OperationTreeBuilder create(
+ TableConfig config,
FunctionLookup functionCatalog,
TableReferenceLookup tableReferenceLookup,
boolean isStreamingMode) {
return new OperationTreeBuilder(
+ config,
functionCatalog,
tableReferenceLookup,
new ProjectionOperationFactory(),
@@ -168,7 +174,11 @@ public final class OperationTreeBuilder {
boolean explicitAlias,
List<OverWindow> overWindows) {
- ExpressionResolver resolver = ExpressionResolver.resolverFor(tableReferenceLookup, functionCatalog, child)
+ ExpressionResolver resolver = ExpressionResolver.resolverFor(
+ config,
+ tableReferenceLookup,
+ functionCatalog,
+ child)
.withOverWindows(overWindows)
.build();
List<ResolvedExpression> projections = resolver.resolve(projectList);
@@ -233,6 +243,7 @@ public final class OperationTreeBuilder {
ResolvedGroupWindow resolvedWindow = aggregateOperationFactory.createResolvedWindow(window, resolver);
ExpressionResolver resolverWithWindowReferences = ExpressionResolver.resolverFor(
+ config,
tableReferenceLookup,
functionCatalog,
child)
@@ -282,9 +293,10 @@ public final class OperationTreeBuilder {
// Step3: resolve expressions, including grouping, aggregates and window properties.
ResolvedGroupWindow resolvedWindow = aggregateOperationFactory.createResolvedWindow(window, resolver);
ExpressionResolver resolverWithWindowReferences = ExpressionResolver.resolverFor(
- tableReferenceLookup,
- functionCatalog,
- child)
+ config,
+ tableReferenceLookup,
+ functionCatalog,
+ child)
.withLocalReferences(
new LocalReferenceExpression(
resolvedWindow.getAlias(),
@@ -327,7 +339,12 @@ public final class OperationTreeBuilder {
JoinType joinType,
Optional<Expression> condition,
boolean correlated) {
- ExpressionResolver resolver = ExpressionResolver.resolverFor(tableReferenceLookup, functionCatalog, left, right)
+ ExpressionResolver resolver = ExpressionResolver.resolverFor(
+ config,
+ tableReferenceLookup,
+ functionCatalog,
+ left,
+ right)
.build();
Optional<ResolvedExpression> resolvedCondition = condition.map(expr -> resolveSingleExpression(expr, resolver));
@@ -356,6 +373,7 @@ public final class OperationTreeBuilder {
public Expression resolveExpression(Expression expression, QueryOperation... tableOperation) {
ExpressionResolver resolver = ExpressionResolver.resolverFor(
+ config,
tableReferenceLookup,
functionCatalog,
tableOperation).build();
@@ -660,6 +678,7 @@ public final class OperationTreeBuilder {
ResolvedGroupWindow resolvedWindow = aggregateOperationFactory.createResolvedWindow(window, resolver);
ExpressionResolver resolverWithWindowReferences = ExpressionResolver.resolverFor(
+ config,
tableReferenceLookup,
functionCatalog,
child)
@@ -758,7 +777,12 @@ public final class OperationTreeBuilder {
}
private ExpressionResolver getResolver(QueryOperation child) {
- return ExpressionResolver.resolverFor(tableReferenceLookup, functionCatalog, child).build();
+ return ExpressionResolver.resolverFor(
+ config,
+ tableReferenceLookup,
+ functionCatalog,
+ child)
+ .build();
}
private static class NoWindowPropertyChecker extends ApiExpressionDefaultVisitor<Void> {
diff --git a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/FunctionCatalogTest.java b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/FunctionCatalogTest.java
index 05be6d3..bdf64fb 100644
--- a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/FunctionCatalogTest.java
+++ b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/FunctionCatalogTest.java
@@ -18,6 +18,7 @@
package org.apache.flink.table.catalog;
+import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException;
import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
import org.apache.flink.table.catalog.exceptions.FunctionAlreadyExistException;
@@ -61,6 +62,7 @@ public class FunctionCatalogTest {
catalog = new GenericInMemoryCatalog(testCatalogName);
moduleManager = new ModuleManager();
functionCatalog = new FunctionCatalog(
+ TableConfig.getDefault(),
new CatalogManager(testCatalogName, catalog), moduleManager);
}
diff --git a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/utils/TableEnvironmentMock.java b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/utils/TableEnvironmentMock.java
index fa2c58c..57af810 100644
--- a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/utils/TableEnvironmentMock.java
+++ b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/utils/TableEnvironmentMock.java
@@ -65,14 +65,15 @@ public class TableEnvironmentMock extends TableEnvironmentImpl {
}
private static TableEnvironmentMock getInstance(boolean isStreamingMode) {
+ final TableConfig config = createTableConfig();
final CatalogManager catalogManager = createCatalogManager();
final ModuleManager moduleManager = new ModuleManager();
return new TableEnvironmentMock(
catalogManager,
moduleManager,
- createTableConfig(),
+ config,
createExecutor(),
- createFunctionCatalog(catalogManager, moduleManager),
+ createFunctionCatalog(config, catalogManager, moduleManager),
createPlanner(),
isStreamingMode);
}
@@ -93,8 +94,11 @@ public class TableEnvironmentMock extends TableEnvironmentImpl {
return new ExecutorMock();
}
- private static FunctionCatalog createFunctionCatalog(CatalogManager catalogManager, ModuleManager moduleManager) {
- return new FunctionCatalog(catalogManager, moduleManager);
+ private static FunctionCatalog createFunctionCatalog(
+ TableConfig config,
+ CatalogManager catalogManager,
+ ModuleManager moduleManager) {
+ return new FunctionCatalog(config, catalogManager, moduleManager);
}
private static PlannerMock createPlanner() {
diff --git a/flink-table/flink-table-api-scala-bridge/src/main/scala/org/apache/flink/table/api/scala/internal/StreamTableEnvironmentImpl.scala b/flink-table/flink-table-api-scala-bridge/src/main/scala/org/apache/flink/table/api/scala/internal/StreamTableEnvironmentImpl.scala
index fcddda9..5913d8a 100644
--- a/flink-table/flink-table-api-scala-bridge/src/main/scala/org/apache/flink/table/api/scala/internal/StreamTableEnvironmentImpl.scala
+++ b/flink-table/flink-table-api-scala-bridge/src/main/scala/org/apache/flink/table/api/scala/internal/StreamTableEnvironmentImpl.scala
@@ -315,7 +315,7 @@ object StreamTableEnvironmentImpl {
new GenericInMemoryCatalog(settings.getBuiltInCatalogName, settings.getBuiltInDatabaseName))
val moduleManager = new ModuleManager
- val functionCatalog = new FunctionCatalog(catalogManager, moduleManager)
+ val functionCatalog = new FunctionCatalog(tableConfig, catalogManager, moduleManager)
val executorProperties = settings.toExecutorProperties
val executor = lookupExecutor(executorProperties, executionEnvironment)
diff --git a/flink-table/flink-table-api-scala-bridge/src/test/scala/org/apache/flink/table/api/scala/internal/StreamTableEnvironmentImplTest.scala b/flink-table/flink-table-api-scala-bridge/src/test/scala/org/apache/flink/table/api/scala/internal/StreamTableEnvironmentImplTest.scala
index e238ddc..48bebee 100644
--- a/flink-table/flink-table-api-scala-bridge/src/test/scala/org/apache/flink/table/api/scala/internal/StreamTableEnvironmentImplTest.scala
+++ b/flink-table/flink-table-api-scala-bridge/src/test/scala/org/apache/flink/table/api/scala/internal/StreamTableEnvironmentImplTest.scala
@@ -81,6 +81,7 @@ class StreamTableEnvironmentImplTest {
private def getStreamTableEnvironment(
env: StreamExecutionEnvironment,
elements: DataStream[Int]) = {
+ val config = new TableConfig
val catalogManager = new CatalogManager(
"cat",
new GenericInMemoryCatalog("cat", "db"))
@@ -88,8 +89,8 @@ class StreamTableEnvironmentImplTest {
new StreamTableEnvironmentImpl(
catalogManager,
moduleManager,
- new FunctionCatalog(catalogManager, moduleManager),
- new TableConfig,
+ new FunctionCatalog(config, catalogManager, moduleManager),
+ config,
env,
new TestPlanner(elements.javaStream.getTransformation),
new ExecutorMock,
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/AggregateFunctionDefinition.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/AggregateFunctionDefinition.java
index b06c55a..df01db0 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/AggregateFunctionDefinition.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/AggregateFunctionDefinition.java
@@ -49,6 +49,10 @@ public final class AggregateFunctionDefinition implements FunctionDefinition {
this.accumulatorTypeInfo = Preconditions.checkNotNull(accTypeInfo);
}
+ public String getName() {
+ return name;
+ }
+
public AggregateFunction<?, ?> getAggregateFunction() {
return aggregateFunction;
}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/ScalarFunctionDefinition.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/ScalarFunctionDefinition.java
index fc365eb..279d436 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/ScalarFunctionDefinition.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/ScalarFunctionDefinition.java
@@ -40,6 +40,10 @@ public final class ScalarFunctionDefinition implements FunctionDefinition {
this.scalarFunction = Preconditions.checkNotNull(scalarFunction);
}
+ public String getName() {
+ return name;
+ }
+
public ScalarFunction getScalarFunction() {
return scalarFunction;
}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/TableAggregateFunctionDefinition.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/TableAggregateFunctionDefinition.java
index 03c4e33..d5c4f1c 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/TableAggregateFunctionDefinition.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/TableAggregateFunctionDefinition.java
@@ -49,6 +49,10 @@ public final class TableAggregateFunctionDefinition implements FunctionDefinitio
this.accumulatorTypeInfo = Preconditions.checkNotNull(accTypeInfo);
}
+ public String getName() {
+ return name;
+ }
+
public TableAggregateFunction<?, ?> getTableAggregateFunction() {
return aggregateFunction;
}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/TableFunctionDefinition.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/TableFunctionDefinition.java
index 0dcf569..8a6392f 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/TableFunctionDefinition.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/TableFunctionDefinition.java
@@ -46,6 +46,10 @@ public final class TableFunctionDefinition implements FunctionDefinition {
this.resultType = Preconditions.checkNotNull(resultType);
}
+ public String getName() {
+ return name;
+ }
+
public TableFunction<?> getTableFunction() {
return tableFunction;
}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/UserDefinedFunction.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/UserDefinedFunction.java
index 3565a77..a4c3261 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/UserDefinedFunction.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/UserDefinedFunction.java
@@ -43,7 +43,7 @@ public abstract class UserDefinedFunction implements FunctionDefinition, Seriali
*/
public final String functionIdentifier() {
final String md5 = EncodingUtils.hex(EncodingUtils.md5(EncodingUtils.encodeObjectToString(this)));
- return getClass().getCanonicalName().replace('.', '$').concat("$").concat(md5);
+ return getClass().getName().replace('.', '$').concat("$").concat(md5);
}
/**
diff --git a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/expressions/CallExpressionResolver.java b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/expressions/CallExpressionResolver.java
index c2ef623..61cc347 100644
--- a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/expressions/CallExpressionResolver.java
+++ b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/expressions/CallExpressionResolver.java
@@ -45,6 +45,7 @@ public class CallExpressionResolver {
.build()
.getCluster().getPlanner().getContext().unwrap(FlinkContext.class);
this.resolver = ExpressionResolver.resolverFor(
+ context.getTableConfig(),
name -> Optional.empty(),
context.getFunctionCatalog()).build();
}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/CodeGeneratorContext.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/CodeGeneratorContext.scala
index c63bef5..cc7d681 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/CodeGeneratorContext.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/CodeGeneratorContext.scala
@@ -614,7 +614,7 @@ class CodeGeneratorContext(val tableConfig: TableConfig) {
function: UserDefinedFunction,
functionContextClass: Class[_ <: FunctionContext] = classOf[FunctionContext],
contextTerm: String = null): String = {
- val classQualifier = function.getClass.getCanonicalName
+ val classQualifier = function.getClass.getName
val fieldTerm = CodeGenUtils.udfFieldName(function)
addReusableObjectInternal(function, fieldTerm, classQualifier)
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/expressions/call.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/expressions/call.scala
index b4aeb41..b1bf10a 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/expressions/call.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/expressions/call.scala
@@ -200,10 +200,6 @@ case class PlannerTableFunctionCall(
override private[flink] def children: Seq[PlannerExpression] = parameters
override def validateInput(): ValidationResult = {
- // check if not Scala object
- UserDefinedFunctionHelper.validateNotSingleton(tableFunction.getClass)
- // check if class could be instantiated
- UserDefinedFunctionHelper.validateInstantiation(tableFunction.getClass)
// look for a signature that matches the input types
val signature = parameters.map(_.resultType).map(fromLegacyInfoToDataType)
val foundMethod = getUserDefinedMethod(
diff --git a/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/operations/SqlToOperationConverterTest.java b/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/operations/SqlToOperationConverterTest.java
index c5bdef7..bf3567b 100644
--- a/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/operations/SqlToOperationConverterTest.java
+++ b/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/operations/SqlToOperationConverterTest.java
@@ -89,7 +89,10 @@ public class SqlToOperationConverterTest {
private final CatalogManager catalogManager =
new CatalogManager("builtin", catalog);
private final ModuleManager moduleManager = new ModuleManager();
- private final FunctionCatalog functionCatalog = new FunctionCatalog(catalogManager, moduleManager);
+ private final FunctionCatalog functionCatalog = new FunctionCatalog(
+ tableConfig,
+ catalogManager,
+ moduleManager);
private final PlannerContext plannerContext =
new PlannerContext(tableConfig,
functionCatalog,
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/codegen/WatermarkGeneratorCodeGenTest.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/codegen/WatermarkGeneratorCodeGenTest.scala
index d42db1b..5e101d5 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/codegen/WatermarkGeneratorCodeGenTest.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/codegen/WatermarkGeneratorCodeGenTest.scala
@@ -45,11 +45,12 @@ import java.util.Collections
class WatermarkGeneratorCodeGenTest {
// mock FlinkPlannerImpl to avoid discovering TableEnvironment and Executor.
+ val config = new TableConfig
val catalog = new GenericInMemoryCatalog("MockCatalog", "default")
val catalogManager = new CatalogManager("builtin", catalog)
- val functionCatalog = new FunctionCatalog(catalogManager, new ModuleManager)
+ val functionCatalog = new FunctionCatalog(config, catalogManager, new ModuleManager)
val plannerContext = new PlannerContext(
- new TableConfig,
+ config,
functionCatalog,
catalogManager,
asRootSchema(new CatalogManagerCalciteSchema(catalogManager, false)),
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/metadata/AggCallSelectivityEstimatorTest.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/metadata/AggCallSelectivityEstimatorTest.scala
index e248e49..ce25e6a 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/metadata/AggCallSelectivityEstimatorTest.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/metadata/AggCallSelectivityEstimatorTest.scala
@@ -84,7 +84,8 @@ class AggCallSelectivityEstimatorTest {
val planner = mock(classOf[AbstractRelOptPlanner])
val catalogManager = mock(classOf[CatalogManager])
val moduleManager = mock(classOf[ModuleManager])
- val functionCatalog = new FunctionCatalog(catalogManager, moduleManager)
+ val config = new TableConfig
+ val functionCatalog = new FunctionCatalog(config, catalogManager, moduleManager)
val context = new FlinkContextImpl(new TableConfig, functionCatalog, catalogManager)
when(tableScan, "getCluster").thenReturn(cluster)
when(cluster, "getRexBuilder").thenReturn(rexBuilder)
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdHandlerTestBase.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdHandlerTestBase.scala
index aa3d48f..18cd61a 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdHandlerTestBase.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdHandlerTestBase.scala
@@ -91,7 +91,7 @@ class FlinkRelMdHandlerTestBase {
val plannerContext: PlannerContext =
new PlannerContext(
tableConfig,
- new FunctionCatalog(catalogManager, moduleManager),
+ new FunctionCatalog(tableConfig, catalogManager, moduleManager),
catalogManager,
CalciteSchema.from(rootSchema),
util.Arrays.asList(
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/metadata/SelectivityEstimatorTest.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/metadata/SelectivityEstimatorTest.scala
index f89332b..6de6fa5 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/metadata/SelectivityEstimatorTest.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/metadata/SelectivityEstimatorTest.scala
@@ -88,7 +88,7 @@ class SelectivityEstimatorTest {
val planner = mock(classOf[AbstractRelOptPlanner])
val catalogManager = mock(classOf[CatalogManager])
val moduleManager = mock(classOf[ModuleManager])
- val functionCatalog = new FunctionCatalog(catalogManager, moduleManager)
+ val functionCatalog = new FunctionCatalog(tableConfig, catalogManager, moduleManager)
val context: FlinkContext = new FlinkContextImpl(tableConfig, functionCatalog, catalogManager)
when(tableScan, "getCluster").thenReturn(cluster)
when(cluster, "getRexBuilder").thenReturn(rexBuilder)
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/utils/RexNodeExtractorTest.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/utils/RexNodeExtractorTest.scala
index 35942d8..62c8e9f 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/utils/RexNodeExtractorTest.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/plan/utils/RexNodeExtractorTest.scala
@@ -19,7 +19,7 @@
package org.apache.flink.table.planner.plan.utils
import org.apache.flink.api.common.typeinfo.Types
-import org.apache.flink.table.api.DataTypes
+import org.apache.flink.table.api.{DataTypes, TableConfig}
import org.apache.flink.table.catalog.{CatalogManager, FunctionCatalog, GenericInMemoryCatalog}
import org.apache.flink.table.expressions.utils.ApiExpressionUtils.{unresolvedCall, unresolvedRef, valueLiteral}
import org.apache.flink.table.expressions.{Expression, ExpressionParser}
@@ -56,7 +56,10 @@ class RexNodeExtractorTest extends RexNodeTestBase {
val catalogManager = new CatalogManager(
defaultCatalog, new GenericInMemoryCatalog(defaultCatalog, "default_database"))
val moduleManager = new ModuleManager
- private val functionCatalog = new FunctionCatalog(catalogManager, moduleManager)
+ private val functionCatalog = new FunctionCatalog(
+ TableConfig.getDefault,
+ catalogManager,
+ moduleManager)
private val expressionBridge: ExpressionBridge[PlannerExpression] =
new ExpressionBridge[PlannerExpression](
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/runtime/stream/table/CalcITCase.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/runtime/stream/table/CalcITCase.scala
index 573274c..278a2b0 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/runtime/stream/table/CalcITCase.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/planner/runtime/stream/table/CalcITCase.scala
@@ -19,14 +19,14 @@
package org.apache.flink.table.planner.runtime.stream.table
import org.apache.flink.api.scala._
+import org.apache.flink.table.api.ValidationException
import org.apache.flink.table.api.scala._
import org.apache.flink.table.functions.ScalarFunction
-import org.apache.flink.table.planner.expressions.utils.{Func1, Func13, Func23, Func24, Func25, RichFunc1, RichFunc2}
+import org.apache.flink.table.planner.expressions.utils._
import org.apache.flink.table.planner.runtime.utils.StreamingWithStateTestBase.StateBackendMode
import org.apache.flink.table.planner.runtime.utils.TestData._
import org.apache.flink.table.planner.runtime.utils.{StreamingWithStateTestBase, TestingAppendSink, TestingRetractSink, UserDefinedFunctionTestUtils}
import org.apache.flink.types.Row
-
import org.junit.Assert._
import org.junit._
import org.junit.runner.RunWith
@@ -294,6 +294,79 @@ class CalcITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mode
}
@Test
+ def testInlineScalarFunction(): Unit = {
+ val t = env.fromElements(1, 2, 3, 4).toTable(tEnv).as('a)
+
+ val sink = new TestingAppendSink
+ val result = t.select(
+ (new ScalarFunction() {
+ def eval(i: Int, suffix: String): String = {
+ suffix + i
+ }
+ })('a, ">>"))
+ result.toAppendStream[Row].addSink(sink)
+ env.execute()
+
+ val expected = mutable.MutableList(
+ ">>1",
+ ">>2",
+ ">>3",
+ ">>4"
+ )
+ assertEquals(expected.sorted, sink.getAppendResults.sorted)
+ }
+
+ @Test
+ def testNonStaticObjectScalarFunction(): Unit = {
+ val t = env.fromElements(1, 2, 3, 4).toTable(tEnv).as('a)
+
+ val sink = new TestingAppendSink
+ val result = t.select(NonStaticObjectScalarFunction('a, ">>"))
+
+ result.toAppendStream[Row].addSink(sink)
+ env.execute()
+
+ val expected = mutable.MutableList(
+ ">>1",
+ ">>2",
+ ">>3",
+ ">>4"
+ )
+ assertEquals(expected.sorted, sink.getAppendResults.sorted)
+ }
+
+ object NonStaticObjectScalarFunction extends ScalarFunction {
+ def eval(i: Int, suffix: String): String = {
+ suffix + i
+ }
+ }
+
+ @Test(expected = classOf[ValidationException]) // see FLINK-15162
+ def testNonStaticClassScalarFunction(): Unit = {
+ val t = env.fromElements(1, 2, 3, 4).toTable(tEnv).as('a)
+
+ val sink = new TestingAppendSink
+ val result = t.select(new NonStaticClassScalarFunction()('a, ">>"))
+
+ result.toAppendStream[Row].addSink(sink)
+ env.execute()
+
+ val expected = mutable.MutableList(
+ ">>1",
+ ">>2",
+ ">>3",
+ ">>4"
+ )
+ assertEquals(expected.sorted, sink.getAppendResults.sorted)
+ }
+
+ class NonStaticClassScalarFunction extends ScalarFunction {
+ def eval(i: Int, suffix: String): String = {
+ suffix + i
+ }
+ }
+
+ @Test
def testMapType(): Unit = {
val ds = env.fromCollection(tupleData3).toTable(tEnv).select(map('_1, '_3))
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 c65f5c7..16edbbf 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
@@ -1087,7 +1087,7 @@ object TestingTableEnvironment {
settings.getBuiltInCatalogName, settings.getBuiltInDatabaseName))
}
val moduleManager = new ModuleManager
- val functionCatalog = new FunctionCatalog(catalogMgr, moduleManager)
+ val functionCatalog = new FunctionCatalog(tableConfig, catalogMgr, moduleManager)
val plannerProperties = settings.toPlannerProperties
val executorProperties = settings.toExecutorProperties
val executor = ComponentFactoryService.find(classOf[ExecutorFactory],
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 d6eba73..b726bc4 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
@@ -60,7 +60,7 @@ abstract class TableEnvImpl(
// Table API/SQL function catalog
private[flink] val functionCatalog: FunctionCatalog =
- new FunctionCatalog(catalogManager, moduleManager)
+ new FunctionCatalog(config, catalogManager, moduleManager)
// temporary utility until we don't use planner expressions anymore
functionCatalog.setPlannerTypeInferenceUtil(PlannerTypeInferenceUtilImpl.INSTANCE)
@@ -90,6 +90,7 @@ abstract class TableEnvImpl(
}
private[flink] val operationTreeBuilder = OperationTreeBuilder.create(
+ config,
functionCatalog,
tableLookup,
isStreamingMode)
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/codegen/CodeGenerator.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/codegen/CodeGenerator.scala
index 3bbbc65..9925f0d 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/codegen/CodeGenerator.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/codegen/CodeGenerator.scala
@@ -1649,7 +1649,7 @@ abstract class CodeGenerator(
function: UserDefinedFunction,
contextTerm: String = null,
functionContextClass: Class[_ <: FunctionContext] = classOf[FunctionContext]): String = {
- val classQualifier = function.getClass.getCanonicalName
+ val classQualifier = function.getClass.getName
val functionSerializedData = EncodingUtils.encodeObjectToString(function)
val fieldTerm = s"function_${function.functionIdentifier}"
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/call.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/call.scala
index b40053a..fc3d263 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/call.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/expressions/call.scala
@@ -298,10 +298,6 @@ case class PlannerTableFunctionCall(
override private[flink] def children: Seq[PlannerExpression] = parameters
override def validateInput(): ValidationResult = {
- // check if not Scala object
- UserDefinedFunctionHelper.validateNotSingleton(tableFunction.getClass)
- // check if class could be instantiated
- UserDefinedFunctionHelper.validateInstantiation(tableFunction.getClass)
// look for a signature that matches the input types
val signature = parameters.map(_.resultType)
val foundMethod = getUserDefinedMethod(tableFunction, "eval", typeInfoToClass(signature))
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/rules/logical/PushFilterIntoTableSourceScanRule.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/rules/logical/PushFilterIntoTableSourceScanRule.scala
index 334fcb3..44b100f 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/rules/logical/PushFilterIntoTableSourceScanRule.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/rules/logical/PushFilterIntoTableSourceScanRule.scala
@@ -23,7 +23,7 @@ import java.util
import org.apache.calcite.plan.RelOptRule.{none, operand}
import org.apache.calcite.plan.{RelOptRule, RelOptRuleCall}
import org.apache.calcite.rex.RexProgram
-import org.apache.flink.table.api.TableException
+import org.apache.flink.table.api.{TableConfig, TableException}
import org.apache.flink.table.catalog.{CatalogManager, FunctionCatalog, GenericInMemoryCatalog}
import org.apache.flink.table.expressions.{Expression, PlannerExpression}
import org.apache.flink.table.module.ModuleManager
@@ -74,7 +74,7 @@ class PushFilterIntoTableSourceScanRule extends RelOptRule(
RexProgramExtractor.extractConjunctiveConditions(
program,
call.builder().getRexBuilder,
- new FunctionCatalog(catalogManager, new ModuleManager))
+ new FunctionCatalog(TableConfig.getDefault, catalogManager, new ModuleManager))
if (predicates.isEmpty) {
// no condition can be translated to expression
return
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/sqlexec/SqlToOperationConverterTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/sqlexec/SqlToOperationConverterTest.java
index f5d9231..7609994 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/sqlexec/SqlToOperationConverterTest.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/sqlexec/SqlToOperationConverterTest.java
@@ -86,7 +86,10 @@ public class SqlToOperationConverterTest {
private final CatalogManager catalogManager =
new CatalogManager("builtin", catalog);
private final ModuleManager moduleManager = new ModuleManager();
- private final FunctionCatalog functionCatalog = new FunctionCatalog(catalogManager, moduleManager);
+ private final FunctionCatalog functionCatalog = new FunctionCatalog(
+ tableConfig,
+ catalogManager,
+ moduleManager);
private final PlanningConfigurationBuilder planningConfigurationBuilder =
new PlanningConfigurationBuilder(tableConfig,
functionCatalog,
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/StreamTableEnvironmentTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/StreamTableEnvironmentTest.scala
index a3450f4..b666039 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/StreamTableEnvironmentTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/StreamTableEnvironmentTest.scala
@@ -208,7 +208,7 @@ class StreamTableEnvironmentTest extends TableTestBase {
new GenericInMemoryCatalog("default_catalog", "default_database"))
val moduleManager: ModuleManager = new ModuleManager
val executor: StreamExecutor = new StreamExecutor(jStreamExecEnv)
- val functionCatalog = new FunctionCatalog(manager, moduleManager)
+ val functionCatalog = new FunctionCatalog(config, manager, moduleManager)
val streamPlanner = new StreamPlanner(executor, config, functionCatalog, manager)
val jTEnv = new JStreamTableEnvironmentImpl(
manager,
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/sql/AggregateTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/sql/AggregateTest.scala
index 25f875f..95b57f8 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/sql/AggregateTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/api/stream/sql/AggregateTest.scala
@@ -68,15 +68,16 @@ class AggregateTest extends TableTestBase {
@Test
def testUserDefinedAggregateFunctionWithScalaAccumulator(): Unit = {
val defaultCatalog = "default_catalog"
+ val config = new TableConfig
val catalogManager = new CatalogManager(
defaultCatalog, new GenericInMemoryCatalog(defaultCatalog, "default_database"))
val moduleManager = new ModuleManager
- val functionCatalog = new FunctionCatalog(catalogManager, moduleManager)
+ val functionCatalog = new FunctionCatalog(config, catalogManager, moduleManager)
val tablEnv = new StreamTableEnvironmentImpl(
catalogManager,
moduleManager,
functionCatalog,
- new TableConfig,
+ config,
Mockito.mock(classOf[StreamExecutionEnvironment]),
Mockito.mock(classOf[Planner]),
Mockito.mock(classOf[Executor]),
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/plan/RexProgramExtractorTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/plan/RexProgramExtractorTest.scala
index d109726..1f0da00 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/plan/RexProgramExtractorTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/plan/RexProgramExtractorTest.scala
@@ -28,6 +28,7 @@ import org.apache.calcite.sql.`type`.SqlTypeName
import org.apache.calcite.sql.`type`.SqlTypeName.{BIGINT, INTEGER, VARCHAR}
import org.apache.calcite.sql.fun.SqlStdOperatorTable
import org.apache.calcite.util.{DateString, TimeString, TimestampString}
+import org.apache.flink.table.api.TableConfig
import org.apache.flink.table.catalog.{CatalogManager, FunctionCatalog, GenericInMemoryCatalog}
import org.apache.flink.table.expressions._
import org.apache.flink.table.module.ModuleManager
@@ -43,6 +44,7 @@ import scala.collection.mutable
class RexProgramExtractorTest extends RexProgramTestBase {
private val functionCatalog: FunctionCatalog = new FunctionCatalog(
+ TableConfig.getDefault,
new CatalogManager("default_catalog", new GenericInMemoryCatalog("default_catalog")),
new ModuleManager
)
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/runtime/stream/table/CalcITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/runtime/stream/table/CalcITCase.scala
index 521b139..912f23b 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/runtime/stream/table/CalcITCase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/runtime/stream/table/CalcITCase.scala
@@ -20,8 +20,10 @@ package org.apache.flink.table.runtime.stream.table
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
+import org.apache.flink.table.api.ValidationException
import org.apache.flink.table.api.scala._
import org.apache.flink.table.expressions.utils._
+import org.apache.flink.table.functions.ScalarFunction
import org.apache.flink.table.runtime.utils.{StreamITCase, StreamTestData, UserDefinedFunctionTestUtils}
import org.apache.flink.test.util.AbstractTestBase
import org.apache.flink.types.Row
@@ -328,6 +330,92 @@ class CalcITCase extends AbstractTestBase {
}
@Test
+ def testInlineScalarFunction(): Unit = {
+ val env = StreamExecutionEnvironment.getExecutionEnvironment
+ val tEnv = StreamTableEnvironment.create(env)
+
+ StreamITCase.testResults = mutable.MutableList()
+
+ val t = env.fromElements(1, 2, 3, 4).toTable(tEnv).as('a)
+
+ val result = t.select(
+ (new ScalarFunction() {
+ def eval(i: Int, suffix: String): String = {
+ suffix + i
+ }
+ })('a, ">>"))
+
+ result.addSink(new StreamITCase.StringSink[Row])
+ env.execute()
+
+ val expected = mutable.MutableList(
+ ">>1",
+ ">>2",
+ ">>3",
+ ">>4"
+ )
+ assertEquals(expected.sorted, StreamITCase.testResults.sorted)
+ }
+
+ @Test
+ def testNonStaticObjectScalarFunction(): Unit = {
+ val env = StreamExecutionEnvironment.getExecutionEnvironment
+ val tEnv = StreamTableEnvironment.create(env)
+
+ StreamITCase.testResults = mutable.MutableList()
+
+ val t = env.fromElements(1, 2, 3, 4).toTable(tEnv).as('a)
+
+ val result = t.select(NonStaticObjectScalarFunction('a, ">>"))
+
+ result.addSink(new StreamITCase.StringSink[Row])
+ env.execute()
+
+ val expected = mutable.MutableList(
+ ">>1",
+ ">>2",
+ ">>3",
+ ">>4"
+ )
+ assertEquals(expected.sorted, StreamITCase.testResults.sorted)
+ }
+
+ object NonStaticObjectScalarFunction extends ScalarFunction {
+ def eval(i: Int, suffix: String): String = {
+ suffix + i
+ }
+ }
+
+ @Test(expected = classOf[ValidationException]) // see FLINK-15162
+ def testNonStaticClassScalarFunction(): Unit = {
+ val env = StreamExecutionEnvironment.getExecutionEnvironment
+ val tEnv = StreamTableEnvironment.create(env)
+
+ StreamITCase.testResults = mutable.MutableList()
+
+ val t = env.fromElements(1, 2, 3, 4).toTable(tEnv).as('a)
+
+ val result = t.select(new NonStaticClassScalarFunction()('a, ">>"))
+
+ result.addSink(new StreamITCase.StringSink[Row])
+ env.execute()
+
+ val expected = mutable.MutableList(
+ ">>1",
+ ">>2",
+ ">>3",
+ ">>4"
+ )
+ assertEquals(expected.sorted, StreamITCase.testResults.sorted)
+ }
+
+ class NonStaticClassScalarFunction extends ScalarFunction {
+ def eval(i: Int, suffix: String): String = {
+ suffix + i
+ }
+ }
+
+ @Test
def testMapType(): Unit = {
val env = StreamExecutionEnvironment.getExecutionEnvironment
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/utils/TableTestBase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/utils/TableTestBase.scala
index 5ebc79e..abe6f80 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/utils/TableTestBase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/utils/TableTestBase.scala
@@ -332,7 +332,7 @@ case class StreamTableTestUtil(
private val manager: CatalogManager = catalogManager.getOrElse(createCatalogManager())
private val moduleManager: ModuleManager = new ModuleManager
private val executor: StreamExecutor = new StreamExecutor(javaEnv)
- private val functionCatalog = new FunctionCatalog(manager, moduleManager)
+ private val functionCatalog = new FunctionCatalog(tableConfig, manager, moduleManager)
private val streamPlanner = new StreamPlanner(executor, tableConfig, functionCatalog, manager)
val javaTableEnv = new JavaStreamTableEnvironmentImpl(
[flink] 04/04: [hotfix][table] Use testing infrastacture in tests
Posted by tw...@apache.org.
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 524b443040f1d28115a587b1b261a25f29299081
Author: Timo Walther <tw...@apache.org>
AuthorDate: Tue Dec 17 11:47:57 2019 +0100
[hotfix][table] Use testing infrastacture in tests
---
.../org/apache/flink/table/catalog/CatalogTableITCase.scala | 4 +++-
.../table/runtime/batch/sql/PartitionableSinkITCase.scala | 11 ++++++-----
2 files changed, 9 insertions(+), 6 deletions(-)
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/catalog/CatalogTableITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/catalog/CatalogTableITCase.scala
index cece871..e406afd 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/catalog/CatalogTableITCase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/catalog/CatalogTableITCase.scala
@@ -30,11 +30,13 @@ import org.junit.runners.Parameterized
import org.junit.{Before, Ignore, Test}
import java.util
+import org.apache.flink.test.util.AbstractTestBase
+
import scala.collection.JavaConversions._
/** Test cases for catalog table. */
@RunWith(classOf[Parameterized])
-class CatalogTableITCase(isStreaming: Boolean) {
+class CatalogTableITCase(isStreaming: Boolean) extends AbstractTestBase {
private val batchExec: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
private var batchEnv: BatchTableEnvironment = _
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/runtime/batch/sql/PartitionableSinkITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/runtime/batch/sql/PartitionableSinkITCase.scala
index c7fe1be..219b2db 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/runtime/batch/sql/PartitionableSinkITCase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/runtime/batch/sql/PartitionableSinkITCase.scala
@@ -18,11 +18,14 @@
package org.apache.flink.table.runtime.batch.sql
+import java.util.{LinkedList => JLinkedList, Map => JMap}
+
import org.apache.flink.api.common.ExecutionConfig
import org.apache.flink.api.common.functions.MapFunction
import org.apache.flink.api.common.io.RichOutputFormat
import org.apache.flink.api.common.typeinfo.BasicTypeInfo.{INT_TYPE_INFO, LONG_TYPE_INFO, STRING_TYPE_INFO}
import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.api.java
import org.apache.flink.api.java.DataSet
import org.apache.flink.api.java.typeutils.RowTypeInfo
import org.apache.flink.api.scala.ExecutionEnvironment
@@ -34,19 +37,17 @@ import org.apache.flink.table.runtime.batch.sql.PartitionableSinkITCase._
import org.apache.flink.table.sinks.{BatchTableSink, PartitionableTableSink, TableSink}
import org.apache.flink.table.sources.BatchTableSource
import org.apache.flink.table.types.logical.{BigIntType, IntType, VarCharType}
+import org.apache.flink.test.util.AbstractTestBase
import org.apache.flink.types.Row
-
import org.junit.Assert.assertEquals
import org.junit.rules.ExpectedException
import org.junit.{Before, Rule, Test}
-import java.util.{LinkedList => JLinkedList, List => JList, Map => JMap}
-import org.apache.flink.api.java
-
import scala.collection.JavaConversions._
import scala.collection.Seq
-class PartitionableSinkITCase {
+class PartitionableSinkITCase extends AbstractTestBase {
+
private val batchExec: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
private var tEnv: BatchTableEnvironment = _
private val type3 = new RowTypeInfo(INT_TYPE_INFO, LONG_TYPE_INFO, STRING_TYPE_INFO)