[FLINK-13869] Hive functions can not work in blink planner stream mode

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Fix Version/s: 1.10.0
• Component/s: Connectors / Hive, Table SQL / Planner
• Labels: pull-request-available
• Assignee: Jingsong Lee
• Reporter: Terry Wang
• Created: 27/Aug/19 07:48
• Updated: 30/Oct/19 12:19
• Resolved: 30/Oct/19 12:12

Description

In flink, specifying the StreamTableEnvironment through the EnvironmentSetting using the blink planner, when using the UDAF in hive in the table API, the error is reported.

The hive function should been make by correct constants and argTypes. Otherwise it will throw an exception. (See HiveAggSqlFunction)
In this isTableAggregate, it just want to check the aggregate function class type, so the better way is get the function instead of make a function.

Caused by: java.lang.NullPointerException
	at java.util.Arrays.stream(Arrays.java:5004)
	at java.util.stream.Stream.of(Stream.java:1000)
	at org.apache.flink.table.types.utils.TypeConversions.fromLogicalToDataType(TypeConversions.java:67)
	at org.apache.flink.table.planner.functions.utils.HiveFunctionUtils.invokeSetArgs(HiveFunctionUtils.java:59)
	at org.apache.flink.table.planner.functions.utils.HiveAggSqlFunction.makeFunction(HiveAggSqlFunction.java:68)
	at org.apache.flink.table.planner.functions.utils.HiveAggSqlFunction.makeFunction(HiveAggSqlFunction.java:47)
	at org.apache.flink.table.planner.plan.utils.AggregateUtil$$anonfun$isTableAggregate$2.apply(AggregateUtil.scala:750)
	at org.apache.flink.table.planner.plan.utils.AggregateUtil$$anonfun$isTableAggregate$2.apply(AggregateUtil.scala:750)
	at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
	at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
	at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
	at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
	at scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
	at scala.collection.AbstractTraversable.map(Traversable.scala:104)
	at org.apache.flink.table.planner.plan.utils.AggregateUtil$.isTableAggregate(AggregateUtil.scala:750)
	at org.apache.flink.table.planner.plan.utils.RelExplainUtil$.streamGroupAggregationToString(RelExplainUtil.scala:346)
	at org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecGroupAggregate.explainTerms(StreamExecGroupAggregate.scala:109)
	at org.apache.calcite.rel.AbstractRelNode.explain(AbstractRelNode.java:307)
	at org.apache.calcite.rel.AbstractRelNode.computeDigest(AbstractRelNode.java:388)
	at org.apache.calcite.rel.AbstractRelNode.recomputeDigest(AbstractRelNode.java:351)
	at org.apache.calcite.rel.AbstractRelNode.onRegister(AbstractRelNode.java:345)
	at org.apache.calcite.plan.volcano.VolcanoPlanner.registerImpl(VolcanoPlanner.java:1668)
	at org.apache.calcite.plan.volcano.VolcanoPlanner.register(VolcanoPlanner.java:846)
	at org.apache.calcite.plan.volcano.VolcanoPlanner.ensureRegistered(VolcanoPlanner.java:868)
	at org.apache.calcite.plan.volcano.VolcanoPlanner.ensureRegistered(VolcanoPlanner.java:1939)
	at org.apache.calcite.plan.volcano.VolcanoRuleCall.transformTo(VolcanoRuleCall.java:129)
	... 60 more

Issue Links

• links to: GitHub Pull Request #10013