dataengineervishal commented on code in PR #4588:
URL: https://github.com/apache/flink-cdc/pull/4588#discussion_r4344754968
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/main/java/org/apache/flink/cdc/connectors/iceberg/sink/v2/IcebergWriter.java:
##########
@@ -140,35 +140,44 @@ public Collection<WriteResultWrapper> prepareCommit()
throws IOException {
return list;
}
- private RowDataTaskWriterFactory getRowDataTaskWriterFactory(TableId
tableId) {
+ private TableWriterFactory getTableWriterFactory(TableId tableId) {
Table table =
catalog.loadTable(TableIdentifier.parse(tableId.identifier()));
RowType rowType = FlinkSchemaUtil.convert(table.schema());
+ List<Integer> equalityFieldIds = new
ArrayList<>(table.schema().identifierFieldIds());
RowDataTaskWriterFactory rowDataTaskWriterFactory =
new RowDataTaskWriterFactory(
table,
rowType,
DEFAULT_MAX_FILE_SIZE,
FileFormat.fromString(DEFAULT_FILE_FORMAT),
new HashMap<>(),
- new ArrayList<>(table.schema().identifierFieldIds()),
+ equalityFieldIds,
true);
rowDataTaskWriterFactory.initialize(taskId, attemptId);
- return rowDataTaskWriterFactory;
+ // Match the factory's actual append/delta writer selection, even for
existing tables
+ // whose identifier fields differ from the incoming CDC schema's
primary key.
+ return new TableWriterFactory(rowDataTaskWriterFactory,
!equalityFieldIds.isEmpty());
}
@Override
public void write(Event event, Context context) throws IOException {
if (event instanceof DataChangeEvent) {
DataChangeEvent dataChangeEvent = (DataChangeEvent) event;
TableId tableId = dataChangeEvent.tableId();
- writerFactoryMap.computeIfAbsent(tableId,
this::getRowDataTaskWriterFactory);
- TaskWriter<RowData> writer =
- writerMap.computeIfAbsent(
- tableId, tableId1 ->
writerFactoryMap.get(tableId1).create());
+ TableWriterFactory writerFactory =
+ writerFactoryMap.computeIfAbsent(tableId,
this::getTableWriterFactory);
TableSchemaWrapper tableSchemaWrapper = schemaMap.get(tableId);
List<RowData> rowDataList =
RowDataUtils.convertDataChangeEventToRowData(
- dataChangeEvent,
tableSchemaWrapper.getFieldGetters());
+ dataChangeEvent,
+ tableSchemaWrapper.getFieldGetters(),
+ writerFactory.supportsEqualityDeletes);
Review Comment:
I can also reproduce @spoorthibasu's flush() finding on this branch (JDK 17,
Flink 1.20.3). Append-only table, insert (1,north), flush(false), set id as an
identifier field via the catalog, update to (1,east): the result is [1,north],
[1,north], [1,east]. It would be good to add this as a regression test along
with the fix, whether that is retainAll or tying the capability to the open
writer. Happy to help with the fix or the test if useful.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
