[FLINK-18520][table] Fix unresolvable catalog table functions

This closes apache#12857.
deepfield · Jul 9, 2020 · df4f9bc · df4f9bc
1 parent 7d52ddc
commit df4f9bc
Show file tree

Hide file tree

Showing 2 changed files with 32 additions and 29 deletions.
diff --git a/...nk/src/main/java/org/apache/flink/table/planner/catalog/FunctionCatalogOperatorTable.java b/...nk/src/main/java/org/apache/flink/table/planner/catalog/FunctionCatalogOperatorTable.java
@@ -154,7 +154,7 @@ private Optional<SqlFunction> convertToBridgingSqlFunction(
  FunctionIdentifier identifier,
  FunctionDefinition definition) {
 
- if (!verifyFunctionKind(category, definition)) {
+ if (!verifyFunctionKind(category, identifier, definition)) {
  return Optional.empty();
  }
 
@@ -197,6 +197,7 @@ private Optional<SqlFunction> convertToBridgingSqlFunction(
  @SuppressWarnings("RedundantIfStatement")
  private boolean verifyFunctionKind(
  @Nullable SqlFunctionCategory category,
+ FunctionIdentifier identifier,
  FunctionDefinition definition) {
 
  // for now, we don't allow other functions than user-defined ones
@@ -208,11 +209,17 @@ private boolean verifyFunctionKind(
  // it would be nice to give a more meaningful exception when a scalar function is used instead
  // of a table function and vice versa, but we can do that only once FLIP-51 is implemented
 
- if (definition.getKind() == FunctionKind.SCALAR &&
- (category == SqlFunctionCategory.USER_DEFINED_FUNCTION || category == SqlFunctionCategory.SYSTEM)) {
+ if (definition.getKind() == FunctionKind.SCALAR) {
+ if (category != null && category.isTableFunction()) {
+ throw new ValidationException(
+ String.format(
+ "Function '%s' cannot be used as a table function.",
+ identifier.asSummaryString()
+ )
+ );
+ }
  return true;
- } else if (definition.getKind() == FunctionKind.TABLE &&
- (category == SqlFunctionCategory.USER_DEFINED_TABLE_FUNCTION || category == SqlFunctionCategory.SYSTEM)) {
+ } else if (definition.getKind() == FunctionKind.TABLE) {
  return true;
  }
 

diff --git a/...blink/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/FunctionITCase.java b/...blink/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/FunctionITCase.java
@@ -446,7 +446,7 @@ private void testUserDefinedCatalogFunction(String createFunctionDDL) throws Exc
  }
 
  @Test
- public void testPrimitiveScalarFunction() throws Exception {
+ public void testPrimitiveScalarFunction() {
  final List<Row> sourceData = Arrays.asList(
  Row.of(1, 1L, "-"),
  Row.of(2, 2L, "--"),
@@ -471,7 +471,7 @@ public void testPrimitiveScalarFunction() throws Exception {
  }
 
  @Test
- public void testNullScalarFunction() throws Exception {
+ public void testNullScalarFunction() {
  final List<Row> sinkData = Collections.singletonList(
  Row.of("Boolean", "String", "<<unknown>>", "String", "Object", "Boolean"));
 
@@ -496,7 +496,7 @@ public void testNullScalarFunction() throws Exception {
  }
 
  @Test
- public void testRowScalarFunction() throws Exception {
+ public void testRowScalarFunction() {
  final List<Row> sourceData = Arrays.asList(
  Row.of(1, Row.of(1, "1")),
  Row.of(2, Row.of(2, "2")),
@@ -518,7 +518,7 @@ public void testRowScalarFunction() throws Exception {
  }
 
  @Test
- public void testComplexScalarFunction() throws Exception {
+ public void testComplexScalarFunction() {
  final List<Row> sourceData = Arrays.asList(
  Row.of(1, new byte[]{1, 2, 3}),
  Row.of(2, new byte[]{2, 3, 4}),
@@ -588,7 +588,7 @@ public void testComplexScalarFunction() throws Exception {
  }
 
  @Test
- public void testCustomScalarFunction() throws Exception {
+ public void testCustomScalarFunction() {
  final List<Row> sourceData = Arrays.asList(
  Row.of(1),
  Row.of(2),
@@ -622,7 +622,7 @@ public void testCustomScalarFunction() throws Exception {
  }
 
  @Test
- public void testRawLiteralScalarFunction() throws Exception {
+ public void testRawLiteralScalarFunction() {
  final List<Row> sourceData = Arrays.asList(
  Row.of(1, DayOfWeek.MONDAY),
  Row.of(2, DayOfWeek.FRIDAY),
@@ -748,7 +748,7 @@ public void testInvalidCustomScalarFunction() {
  }
 
  @Test
- public void testRowTableFunction() throws Exception {
+ public void testRowTableFunction() {
  final List<Row> sourceData = Arrays.asList(
  Row.of("1,2,3"),
  Row.of("2,3,4"),
@@ -801,7 +801,7 @@ public void testStructuredTableFunction() {
  }
 
  @Test
- public void testDynamicTableFunction() throws Exception {
+ public void testDynamicCatalogTableFunction() {
  final Row[] sinkData = new Row[]{
  Row.of("Test is a string"),
  Row.of("42"),
@@ -812,7 +812,7 @@ public void testDynamicTableFunction() throws Exception {
 
  tEnv().executeSql("CREATE TABLE SinkTable(s STRING) WITH ('connector' = 'COLLECTION')");
 
- tEnv().createTemporarySystemFunction("DynamicTableFunction", DynamicTableFunction.class);
+ tEnv().createFunction("DynamicTableFunction", DynamicTableFunction.class);
  execInsertSqlAndWaitResult(
  "INSERT INTO SinkTable " +
  "SELECT T1.s FROM TABLE(DynamicTableFunction('Test')) AS T1(s) " +
@@ -826,7 +826,7 @@ public void testDynamicTableFunction() throws Exception {
 
  @Test
  public void testInvalidUseOfScalarFunction() {
- tEnv().executeSql("CREATE TABLE SinkTable(s STRING) WITH ('connector' = 'COLLECTION')");
+ tEnv().executeSql("CREATE TABLE SinkTable(s BIGINT NOT NULL) WITH ('connector' = 'COLLECTION')");
 
  tEnv().createTemporarySystemFunction("PrimitiveScalarFunction", PrimitiveScalarFunction.class);
  try {
@@ -839,7 +839,7 @@ public void testInvalidUseOfScalarFunction() {
  e,
  hasMessage(
  containsString(
- "No match found for function signature PrimitiveScalarFunction(<NUMERIC>, <NUMERIC>, <CHARACTER>)")));
+ "SQL validation failed. Function 'PrimitiveScalarFunction' cannot be used as a table function.")));
  }
  }
 
@@ -863,21 +863,17 @@ public void testInvalidUseOfSystemScalarFunction() {
 
  @Test
  public void testInvalidUseOfTableFunction() {
- tEnv().executeSql("CREATE TABLE SinkTable(s STRING) WITH ('connector' = 'COLLECTION')");
+ TestCollectionTableFactory.reset();
+
+ tEnv().executeSql("CREATE TABLE SinkTable(s ROW<s STRING, sa ARRAY<STRING> NOT NULL>) WITH ('connector' = 'COLLECTION')");
 
  tEnv().createTemporarySystemFunction("RowTableFunction", RowTableFunction.class);
- try {
- tEnv().executeSql(
- "INSERT INTO SinkTable " +
- "SELECT RowTableFunction('test')");
- fail();
- } catch (ValidationException e) {
- assertThat(
- e,
- hasMessage(
- containsString(
- "No match found for function signature RowTableFunction(<CHARACTER>)")));
- }
+ tEnv().executeSql(
+ "INSERT INTO SinkTable " +
+ "SELECT RowTableFunction('test')");
+
+ // currently, calling a table function like a scalar function produces no result
+ assertThat(TestCollectionTableFactory.getResult(), equalTo(Collections.emptyList()));
  }
 
  // --------------------------------------------------------------------------------------------