[FLINK-26039] Incorrect value getter in map unnest table function

• Type: Bug
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 1.14.3
• Fix Version/s: 1.14.4, 1.15.0, 1.13.7
• Component/s: Table SQL / Runtime
• Labels: pull-request-available
• Assignee: Han
• Reporter: Han
• Created: 09/Feb/22 07:50
• Updated: 16/Feb/22 18:23
• Resolved: 09/Feb/22 13:42

Description

Suppose we have a map field that needs to be expanded.

 

CREATE TABLE t (
    id INT,
    map_field MAP<STRING, INT>
) WITH (
    -- ...
);

SELECT id, k, v FROM t, unnest(map_field) as A(k, v);

 

 

We will get the following runtime exception:

Caused by: java.lang.ClassCastException: org.apache.flink.table.data.binary.BinaryStringData cannot be cast to java.lang.Integer
    at org.apache.flink.table.data.GenericRowData.getInt(GenericRowData.java:149)
    at org.apache.flink.table.data.utils.JoinedRowData.getInt(JoinedRowData.java:149)
    at org.apache.flink.table.data.RowData.lambda$createFieldGetter$245ca7d1$6(RowData.java:245)
    at org.apache.flink.table.data.RowData.lambda$createFieldGetter$25774257$1(RowData.java:296)
    at org.apache.flink.table.runtime.typeutils.RowDataSerializer.copyRowData(RowDataSerializer.java:170)
    at org.apache.flink.table.runtime.typeutils.RowDataSerializer.copy(RowDataSerializer.java:131)
    at org.apache.flink.table.runtime.typeutils.RowDataSerializer.copy(RowDataSerializer.java:48)
    at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:80)
    at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:57)
    at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:29)
    at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:56)
    at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:29)
    at org.apache.flink.table.runtime.util.StreamRecordCollector.collect(StreamRecordCollector.java:44)
    at org.apache.flink.table.runtime.collector.TableFunctionCollector.outputResult(TableFunctionCollector.java:68)
    at StreamExecCorrelate$10$TableFunctionCollector$4.collect(Unknown Source)
    at org.apache.flink.table.runtime.collector.WrappingCollector.outputResult(WrappingCollector.java:39)
    at StreamExecCorrelate$10$TableFunctionResultConverterCollector$8.collect(Unknown Source)
    at org.apache.flink.table.functions.TableFunction.collect(TableFunction.java:197)
    at org.apache.flink.table.runtime.functions.SqlUnnestUtils$MapUnnestTableFunction.eval(SqlUnnestUtils.java:169)
    at StreamExecCorrelate$10.processElement(Unknown Source)
    at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:82)
    at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:57)
    at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:29)
    at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:56)
    at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:29) 

Issue Links

• links to: GitHub Pull Request #18679