spoorthibasu commented on code in PR #4607:
URL: https://github.com/apache/flink-cdc/pull/4607#discussion_r4464367245
##########
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:
##########
@@ -156,7 +162,23 @@ private TableWriterFactory getTableWriterFactory(TableId
tableId) {
rowDataTaskWriterFactory.initialize(taskId, attemptId);
// 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());
+ boolean supportsEqualityDeletes = !equalityFieldIds.isEmpty();
+ return new TableWriterFactory(
+ rowDataTaskWriterFactory,
+ supportsEqualityDeletes,
+ supportsEqualityDeletes ? createDeleteKeyGetters(tableId,
table) : null);
+ }
+
+ /**
+ * Getters for the target table's identifier fields and partition source
columns. An UPDATE that
+ * changes none of them can skip its before-image DELETE.
+ */
+ private RecordData.FieldGetter[] createDeleteKeyGetters(TableId tableId,
Table table) {
+ Set<String> columnNames = new
LinkedHashSet<>(table.schema().identifierFieldNames());
+ for (PartitionField field : table.spec().fields()) {
+ columnNames.add(table.schema().findColumnName(field.sourceId()));
+ }
+ return
RowDataUtils.createDeleteKeyGetters(schemaMap.get(tableId).getSchema(),
columnNames);
}
Review Comment:
On this branch I can reproduce the mismatch @lvyanquan describes. With CDC
columns `(id BIGINT, other BIGINT, name)` and primary key `id`:
- On a pre-existing table with columns `(other, id, name)` and identifier
`id`, I inserted and committed `(1, 100, a)`, then updated it to `(1, 200, a)`,
so only the CDC column `other` changes. This branch leaves two rows, while
master leaves one.
- On a table created by the pipeline, I inserted and committed `(1, 100,
a)`, swapped the names of `id` and `other` in the catalog, then updated it to
`(2, 100, a)`. This branch leaves `(1, 100, a), (2, 100, a)`, while master
leaves only `(2, 100, a)`.
With the position check Yanquan suggests, both cases leave the same rows as
master. Happy to share the 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]
