[Date Prev][Date Next][Thread Prev][Thread Next][Date Index][Thread Index]

[jira] [Created] (FLINK-10019) Fix Composite getResultType of UDF cannot be chained with other operators

Rong Rong created FLINK-10019:

             Summary: Fix Composite getResultType of UDF cannot be chained with other operators
                 Key: FLINK-10019
             Project: Flink
          Issue Type: Sub-task
            Reporter: Rong Rong
            Assignee: Rong Rong

If explicitly return a CompositeType in {{udf.getResultType}}, will result in some failures in chained operators.
For example: consider a simple UDF,

object Func extends ScalarFunction {
  def eval(row: Row): Row = {

  override def getParameterTypes(signature: Array[Class[_]]): Array[TypeInformation[_]] =

  override def getResultType(signature: Array[Class[_]]): TypeInformation[_] =

This should work perfectly since it's just a simple pass through, however

  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", Func20)
    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])


Exception code:

java.lang.IndexOutOfBoundsException: index (1) must be less than size (1)

	at org.apache.calcite.sql.type.InferTypes$2.inferOperandTypes(
	at org.apache.calcite.sql.validate.SqlValidatorImpl.inferUnknownTypes(
	at org.apache.calcite.sql.validate.SqlValidatorImpl.expandSelectItem(
	at org.apache.calcite.sql.validate.SqlValidatorImpl.expandStar(

This is due to the fact that Calcite inferOperandTypes does not expect to infer a struct RelDataType.

This message was sent by Atlassian JIRA