spoorthibasu commented on code in PR #4588:
URL: https://github.com/apache/flink-cdc/pull/4588#discussion_r4338633076
##########
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:
`flush()` drops the cached factories but not the writers, so after a
FlushEvent for any table (a new table's CreateTableEvent is enough), this
capability can come from a newer table load than the writer it feeds. On an
append-only table I inserted `(1, north)`, called `flush(false)`, set `id` as
an identifier field through the catalog, then updated to `(1, east)`. The
commit leaves `(1, north), (1, north), (1, east)`, the extra before-image this
PR fixes, because the DELETE row went to the append writer. Keeping the factory
while its table has an open writer, for example
`writerFactoryMap.keySet().retainAll(writerMap.keySet())` in `flush()`, avoids
it, though `testCatalogLoadsOnlyOnFactoryCacheMiss` would need its expected
load counts updated. Happy to share the test.
--
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]
