[
https://issues.apache.org/jira/browse/FLINK-10019?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Rong Rong updated FLINK-10019:
------------------------------
Summary: singleton row type objects returned from sub query cannot be
chained with other operators (was: Fix Composite getResultType of UDF cannot
be chained with other operators)
> singleton row type objects returned from sub query cannot be chained with
> other operators
> -----------------------------------------------------------------------------------------
>
> Key: FLINK-10019
> URL: https://issues.apache.org/jira/browse/FLINK-10019
> Project: Flink
> Issue Type: Bug
> Components: SQL / Planner
> Reporter: Rong Rong
> Assignee: Rong Rong
> Priority: Major
>
> If explicitly return a CompositeType in {{udf.getResultType}}, will result in
> some failures in chained operators.
> For example: consider a simple UDF,
> {code:scala}
> object Func extends ScalarFunction {
> def eval(row: Row): Row = {
> row
> }
> override def getParameterTypes(signature: Array[Class[_]]):
> Array[TypeInformation[_]] =
> Array(Types.ROW(Types.INT))
> override def getResultType(signature: Array[Class[_]]): TypeInformation[_] =
> Types.ROW(Types.INT)
> }
> {code}
> This should work perfectly since it's just a simple pass through, however
> {code:scala}
> @Test
> def testRowType(): Unit = {
> val data = List(
> Row.of(Row.of(12.asInstanceOf[Integer]), "1")
> )
> val env = StreamExecutionEnvironment.getExecutionEnvironment
> val stream = env.fromCollection(data)(Types.ROW(Types.ROW(Types.INT),
> Types.STRING))
> val tEnv = TableEnvironment.getTableEnvironment(env)
> val table = stream.toTable(tEnv, 'a, 'b)
> tEnv.registerFunction("func", Func)
> tEnv.registerTable("t", table)
> // This works perfectly
> val result1 = tEnv.sqlQuery("SELECT func(a) FROM t").toAppendStream[Row]
> result1.addSink(new StreamITCase.StringSink[Row])
> // This throws exception
> val result2 = tEnv.sqlQuery("SELECT func(a) as myRow FROM
> t").toAppendStream[Row]
> result2.addSink(new StreamITCase.StringSink[Row])
> env.execute()
> }
> {code}
> Exception code:
> {code:java}
> java.lang.IndexOutOfBoundsException: index (1) must be less than size (1)
> at
> com.google.common.base.Preconditions.checkElementIndex(Preconditions.java:310)
> at
> com.google.common.base.Preconditions.checkElementIndex(Preconditions.java:293)
> at
> com.google.common.collect.SingletonImmutableList.get(SingletonImmutableList.java:41)
> at
> org.apache.calcite.sql.type.InferTypes$2.inferOperandTypes(InferTypes.java:83)
> at
> org.apache.calcite.sql.validate.SqlValidatorImpl.inferUnknownTypes(SqlValidatorImpl.java:1777)
> at
> org.apache.calcite.sql.validate.SqlValidatorImpl.expandSelectItem(SqlValidatorImpl.java:459)
> at
> org.apache.calcite.sql.validate.SqlValidatorImpl.expandStar(SqlValidatorImpl.java:349)
> ...
> {code}
> This is due to the fact that Calcite inferOperandTypes does not expect to
> infer a struct RelDataType.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
