[jira] [Created] (FLINK-20181) RowData cannot cast to Tuple2

[Shang Yuanchun (Jira)]

Xianxun Ye created FLINK-20181:
----------------------------------

             Summary: RowData cannot cast to Tuple2
                 Key: FLINK-20181
                 URL: https://issues.apache.org/jira/browse/FLINK-20181
             Project: Flink
          Issue Type: Bug
            Reporter: Xianxun Ye


I want to emit CDC data by my own StreamOperator.

flink version :1.11.2, blink planner.
{code:java}

//代码占位符
  getTableEnv().registerTableSource(
        "source",
        new StreamTableSource<RowData>() {
          TableSchema tableSchema = TableSchema.builder()
              .field("id", new AtomicDataType(new IntType(false)))
              .field("name", DataTypes.STRING())
              .field("type", DataTypes.STRING())
              .primaryKey("id")
              .build();          @Override
          public DataStream<RowData> getDataStream(StreamExecutionEnvironment execEnv) {
            return execEnv.addSource(new DebugSourceFunction(tableSchema.toRowDataType()));
          }          @Override
          public TableSchema getTableSchema() {
            return tableSchema;
          }          @Override
          public DataType getProducedDataType() {
            return getTableSchema().toRowDataType().bridgedTo(RowData.class);
          }
        }
    );
    sql("insert into Test.testdb.animal "
            + " SELECT id, name, type, '2020' as da, '11' as hr"
            + " from source"
    );  

class DebugSourceFunction extends RichParallelSourceFunction<RowData> implements ResultTypeQueryable<RowData> {    DataType dataType;    public DebugSourceFunction(DataType dataType) {
      this.dataType = dataType;
    }    @Override
    public TypeInformation<RowData> getProducedType() {
      return (TypeInformation<RowData>) createTypeInformation(dataType);
    }    @Override
    public void run(SourceContext<RowData> ctx) throws Exception {
      ctx.collect(GenericRowData.ofKind(RowKind.INSERT, 1, StringData.fromString("monkey"), StringData.fromString("small")));
    }    @Override
    public void cancel() {    }    public TypeInformation<?> createTypeInformation(DataType producedDataType) {
      final DataType internalDataType = DataTypeUtils.transform(
          producedDataType,
          TypeTransformations.TO_INTERNAL_CLASS);
      return fromDataTypeToTypeInfo(internalDataType);
    }
  }  

public class TestUpsertTableSink implements UpsertStreamTableSink<RowData>, OverwritableTableSink, PartitionableTableSink {
     @Override
    public DataStreamSink<?> consumeDataStream(DataStream<Tuple2<Boolean, RowData>> dataStream) {
     
      DataStream<Void> returnStream = dataStream
          .map(
              (MapFunction<Tuple2<Boolean, RowData>, RowData>)
                  value -> value.f1
          )
          ......      return returnStream
          .addSink(new DiscardingSink<>())
          .setParallelism(1);
    }
  }
{code}
when I execute sql with `insert into ...`, occurs class cast fail exception:
{code:java}
//代码占位符
Caused by: java.lang.ClassCastException: org.apache.flink.table.data.GenericRowData cannot be cast to org.apache.flink.api.java.tuple.Tuple2Caused by: java.lang.ClassCastException: org.apache.flink.table.data.GenericRowData cannot be cast to org.apache.flink.api.java.tuple.Tuple2 at org.apache.flink.streaming.api.operators.StreamMap.processElement(StreamMap.java:41) at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator(OperatorChain.java:717) at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:692) at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:672) at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:52) at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:30) at StreamExecCalc$8.processElement(Unknown Source) at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator(OperatorChain.java:717) at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:692) at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:672) at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:52) at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:30) at org.apache.flink.streaming.api.operators.StreamSourceContexts$NonTimestampContext.collect(StreamSourceContexts.java:104)
{code}



--
This message was sent by Atlassian Jira
(v8.3.4#803005)