lvyanquan commented on code in PR #4588:
URL: https://github.com/apache/flink-cdc/pull/4588#discussion_r4344982488
##########
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:
Thanks for pointing this out! I've updated the fix and added regression
tests.
--
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]
