Hi , i really need the feature which could write CDC event to the iceberg table。 I see the feature has completed in the branch 0.11.0 ，and i try this today ， but this feature dos not works 。
First insert a record that Id = 1， and then use sql ”update tableA set xx = xxx where id = 1“ to update the record in mysql .
This CDC event is fetched by my flink stream job and transform it to a RowData record which rowkind is "UPDATE_AFTER" .
But the iceberg table can't distinguish the update event, instead two records that id = 1 appeared in the table ( I have correctly set the "equalityFieldColumns" fields ) 。
And then i delete this record in mysql , but this record still in the iceberg table
`
List pkNames = pkNames(database, tableName, parameterTool);
FlinkSink.forRowData(env.addSource(consumer).flatMap(new IcebergTypeFunction()))
.table(table)
.tableLoader(tableLoader)
.writeParallelism(1)
.equalityFieldColumns(pkNames)
.build();
`
Hi , i really need the feature which could write CDC event to the iceberg table。 I see the feature has completed in the branch 0.11.0 ，and i try this today ， but this feature dos not works 。
First insert a record that Id = 1， and then use sql ”update tableA set xx = xxx where id = 1“ to update the record in mysql .
This CDC event is fetched by my flink stream job and transform it to a RowData record which rowkind is "UPDATE_AFTER" .
But the iceberg table can't distinguish the update event, instead two records that id = 1 appeared in the table ( I have correctly set the "equalityFieldColumns" fields ) 。
And then i delete this record in mysql , but this record still in the iceberg table
`
List pkNames = pkNames(database, tableName, parameterTool);
`