[jira] [Assigned] (FLINK-35726) Data Stream to Table API converts Map to RAW 'java.util.Map'

["Martijn Visser (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/FLINK-35726?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Martijn Visser reassigned FLINK-35726:
--------------------------------------

    Assignee: Martijn Visser

> Data Stream to Table API converts Map to RAW 'java.util.Map'
> ------------------------------------------------------------
>
>                 Key: FLINK-35726
>                 URL: https://issues.apache.org/jira/browse/FLINK-35726
>             Project: Flink
>          Issue Type: Bug
>          Components: Table SQL / Runtime
>    Affects Versions: 1.17.2
>            Reporter: David Perkins
>            Assignee: Martijn Visser
>            Priority: Major
>
> We have a use case where we convert from the Table API to a Data Stream with a class, perform some operations, and then convert back to the Table API. When the data contains a Map, the conversion back to the Table API converts the Map to {{{}RAW('java.util.Map', '...'){}}}. This causes an 'Incompatible types for sink column' exception.
> In this particular case, the Map contains the Kafka headers, which we need to preserve and write to the output topic. Both topics/table definitions use the same schema. We have set a {{DataTypeHint}} annotation on the Map field in the Java class. We are currently working around this issue by using a UDF to simply perform a type conversion from the RAW Java Map to the Table API Map.
> One note is that if no operations are performed on the stream, it work's correctly. But adding a simple identity map causes the exception.
> Here's a simple example to reproduce the problem.
> CREATE TABLE Source (
> id STRING,
> headers MAP<STRING, BYTES> METADATA
> ) WITH (
> 'connector' = 'kafka',
> 'topic' = 'source',
> 'properties.bootstrap.servers' = 'kafka-bootstrap-server',
> 'format' = 'json'
> );
> CREATE TABLE Target (
> id STRING,
> headers MAP<STRING, BYTES> METADATA
> ) WITH (
> 'connector' = 'kafka',
> 'topic' = 'target',
> 'properties.bootstrap.servers' = 'kafka-bootstrap-server',
> 'format' = 'json'
> );
> public class MyRecord {
> private String id;
> @DataTypeHint(value = "MAP<STRING, BYTES>")
> private Map<String,byte[]> headers;
> ...
> }
> public class MyJob {
> public static void main(String[] args) throws Exception{
> final StreamExecutionEnvironment streamEnv = StreamExecutionEnvironment.getExecutionEnvironment();
> final StreamTableEnvironment tableEnv = StreamTableEnvironment.create(streamEnv); Table sourceTable = tableEnv.from("Source"); var sourceStream = tableEnv.toDataStream(sourceTable, MyRecord.class); var mappedStream = sourceStream.map(row -> row); Table outputTable = tableEnv.fromDataStream(mappedStream); tableEnv.createStatementSet().add(outputTable.insertInto("Target")) .attachAsDataStream(); streamEnv.executeAsync("Table Datastream test"); 
> }
> }



--
This message was sent by Atlassian Jira
(v8.20.10#820010)