[FLINK-16414] create udaf/udtf function using sql casuing ValidationException: SQL validation failed. null

• Type: Bug
• Status: Resolved
• Priority: Critical
• Resolution: Fixed
• Affects Version/s: 1.10.0
• Fix Version/s: 1.10.1
• Component/s: Table SQL / API
• Labels: pull-request-available
• Assignee: Terry Wang
• Reporter: Terry Wang
• Created: 04/Mar/20 05:53
• Updated: 05/Mar/20 05:24
• Resolved: 05/Mar/20 05:24

Description

When using TableEnvironment#sqlupdate to create a udaf or udtf function, which doesn't override the getResultType() method, it's normal. But when using this function in later insert sql, some exception like following will be throwed:

Exception in thread "main" org.apache.flink.table.api.ValidationException: SQL validation failed. null
at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.org$apache$flink$table$planner$calcite$FlinkPlannerImpl$$validate(FlinkPlannerImpl.scala:130)
at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.validate(FlinkPlannerImpl.scala:105)
at org.apache.flink.table.planner.operations.SqlToOperationConverter.convert(SqlToOperationConverter.java:127)
at org.apache.flink.table.planner.operations.SqlToOperationConverter.convertSqlInsert(SqlToOperationConverter.java:342)
at org.apache.flink.table.planner.operations.SqlToOperationConverter.convert(SqlToOperationConverter.java:142)
at org.apache.flink.table.planner.delegation.ParserImpl.parse(ParserImpl.java:66)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.sqlUpdate(TableEnvironmentImpl.java:484)

The reason is in FunctionDefinitionUtil#createFunctionDefinition, we shouldn't direct call t.getResultType or a.getAccumulatorType() or a.getResultType() but using UserDefinedFunctionHelper#getReturnTypeOfTableFunction
UserDefinedFunctionHelper#getAccumulatorTypeOfAggregateFunction
UserDefinedFunctionHelper#getReturnTypeOfAggregateFunction instead.
```

if (udf instanceof ScalarFunction)

{ return new ScalarFunctionDefinition( name, (ScalarFunction) udf ); }

else if (udf instanceof TableFunction)

{ TableFunction t = (TableFunction) udf; return new TableFunctionDefinition( name, t, t.getResultType() ); }

else if (udf instanceof AggregateFunction)

{ AggregateFunction a = (AggregateFunction) udf; return new AggregateFunctionDefinition( name, a, a.getAccumulatorType(), a.getResultType() ); }

else if (udf instanceof TableAggregateFunction) {
TableAggregateFunction a = (TableAggregateFunction) udf;

return new TableAggregateFunctionDefinition(
name,
a,
a.getAccumulatorType(),
a.getResultType()
);
```

Issue Links

• links to: GitHub Pull Request #11302
• links to: GitHub Pull Request #11310