lvyanquan commented on code in PR #4607:
URL: https://github.com/apache/flink-cdc/pull/4607#discussion_r4451912941
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/test/java/org/apache/flink/cdc/connectors/iceberg/sink/v2/IcebergWriterTest.java:
##########
@@ -1767,6 +1767,172 @@ void testAppendOnlyUpdatesAndDeletes(boolean
partitioned, boolean cdcHasPrimaryK
}
}
+ @Test
+ public void testSamePartitionUpdatesWithinAndAcrossCheckpoints() throws
Exception {
+ assertSamePartitionUpdatesLeaveOnlyLatestRow(true);
+ }
+
+ @Test
+ public void testUnpartitionedUpdatesWithinAndAcrossCheckpoints() throws
Exception {
+ assertSamePartitionUpdatesLeaveOnlyLatestRow(false);
+ }
+
+ @Test
+ public void testPartitionedUpdateChangingKeyDeletesOldKey() throws
Exception {
+ assertUpdateChangingKeyDeletesOldKey(true);
+ }
+
+ @Test
+ public void testUnpartitionedUpdateChangingKeyDeletesOldKey() throws
Exception {
+ assertUpdateChangingKeyDeletesOldKey(false);
+ }
+
+ private void assertUpdateChangingKeyDeletesOldKey(boolean partitioned)
throws Exception {
+ Map<String, String> catalogOptions = new HashMap<>();
+ catalogOptions.put("type", "hadoop");
+ catalogOptions.put(
+ "warehouse",
+ new File(temporaryFolder.toFile(),
UUID.randomUUID().toString()).toString());
+ catalogOptions.put("cache-enabled", "false");
+ Catalog catalog =
+ CatalogUtil.buildIcebergCatalog(
+ "cdc-iceberg-catalog", catalogOptions, new
Configuration());
+ IcebergMetadataApplier icebergMetadataApplier = new
IcebergMetadataApplier(catalogOptions);
+ IcebergWriter icebergWriter =
+ new IcebergWriter(
+ catalogOptions,
+ 1,
+ 1,
+ ZoneId.systemDefault(),
+ 0,
+ UUID.randomUUID().toString(),
+ UUID.randomUUID().toString(),
+ new HashMap<>());
+
+ TableId tableId = TableId.parse("test.iceberg_table");
+ Schema.Builder schemaBuilder =
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.BIGINT().notNull())
+ .physicalColumn("region",
DataTypes.VARCHAR(100).notNull())
+ .primaryKey("id");
+ if (partitioned) {
+ schemaBuilder.partitionKey("region");
+ }
+ Schema schema = schemaBuilder.build();
+ CreateTableEvent createTableEvent = new CreateTableEvent(tableId,
schema);
+ icebergMetadataApplier.applySchemaChange(createTableEvent);
+ icebergWriter.write(createTableEvent, null);
+
+ BinaryRecordDataGenerator generator =
+ new
BinaryRecordDataGenerator(schema.getColumnDataTypes().toArray(new DataType[0]));
+ RecordData oldKey =
+ generator.generate(new Object[] {1L,
BinaryStringData.fromString("north")});
+ RecordData newKey =
+ generator.generate(new Object[] {2L,
BinaryStringData.fromString("north")});
+ IcebergCommitter committer = new IcebergCommitter(catalogOptions, new
HashMap<>());
+
+ icebergWriter.write(DataChangeEvent.insertEvent(tableId, oldKey),
null);
+ commit(committer, icebergWriter.prepareCommit());
+
+ // The upsert delete only covers the new key, so the old key needs an
explicit delete.
+ icebergWriter.write(DataChangeEvent.updateEvent(tableId, oldKey,
newKey), null);
+ commit(committer, icebergWriter.prepareCommit());
+ committer.close();
+
+ Assertions.assertThat(fetchTableContent(catalog, tableId, null))
+ .containsExactly("2, north");
+ }
+
+ private void assertSamePartitionUpdatesLeaveOnlyLatestRow(boolean
partitioned)
+ throws Exception {
+ Map<String, String> catalogOptions = new HashMap<>();
+ catalogOptions.put("type", "hadoop");
+ catalogOptions.put(
+ "warehouse",
+ new File(temporaryFolder.toFile(),
UUID.randomUUID().toString()).toString());
+ catalogOptions.put("cache-enabled", "false");
+ Catalog catalog =
+ CatalogUtil.buildIcebergCatalog(
+ "cdc-iceberg-catalog", catalogOptions, new
Configuration());
+ IcebergMetadataApplier icebergMetadataApplier = new
IcebergMetadataApplier(catalogOptions);
+ IcebergWriter icebergWriter =
+ new IcebergWriter(
+ catalogOptions,
+ 1,
+ 1,
+ ZoneId.systemDefault(),
+ 0,
+ UUID.randomUUID().toString(),
+ UUID.randomUUID().toString(),
+ new HashMap<>());
+
+ TableId tableId = TableId.parse("test.iceberg_table");
+ Schema.Builder schemaBuilder =
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.BIGINT().notNull())
+ .physicalColumn("region",
DataTypes.VARCHAR(100).notNull())
+ .physicalColumn("name", DataTypes.VARCHAR(100))
+ .primaryKey("id");
+ if (partitioned) {
+ schemaBuilder.partitionKey("region");
+ }
+ Schema schema = schemaBuilder.build();
+ CreateTableEvent createTableEvent = new CreateTableEvent(tableId,
schema);
+ icebergMetadataApplier.applySchemaChange(createTableEvent);
+ icebergWriter.write(createTableEvent, null);
+
+ BinaryRecordDataGenerator generator =
+ new
BinaryRecordDataGenerator(schema.getColumnDataTypes().toArray(new DataType[0]));
+ RecordData[] versions = new RecordData[4];
+ String[] names = {"a", "b", "c", "d"};
+ for (int i = 0; i < names.length; i++) {
+ versions[i] =
+ generator.generate(
+ new Object[] {
+ 1L,
+ BinaryStringData.fromString("north"),
+ BinaryStringData.fromString(names[i])
+ });
+ }
+ RecordData other =
+ generator.generate(
+ new Object[] {
+ 2L,
+ BinaryStringData.fromString("north"),
+ BinaryStringData.fromString("x")
+ });
+ IcebergCommitter committer = new IcebergCommitter(catalogOptions, new
HashMap<>());
+
+ // Checkpoint 1: insert, then update the same key twice before
committing.
+ icebergWriter.write(DataChangeEvent.insertEvent(tableId, versions[0]),
null);
+ icebergWriter.write(DataChangeEvent.insertEvent(tableId, other), null);
+ icebergWriter.write(DataChangeEvent.updateEvent(tableId, versions[0],
versions[1]), null);
+ icebergWriter.write(DataChangeEvent.updateEvent(tableId, versions[1],
versions[2]), null);
+ commit(committer, icebergWriter.prepareCommit());
+ Assertions.assertThat(fetchTableContent(catalog, tableId, null))
+ .containsExactlyInAnyOrder("1, north, c", "2, north, x");
+
+ // Checkpoint 2: update a row that only exists in an already committed
data file.
+ icebergWriter.write(DataChangeEvent.updateEvent(tableId, versions[2],
versions[3]), null);
+ commit(committer, icebergWriter.prepareCommit());
Review Comment:
The new writer tests only assert the final table contents, which also pass
on master, so nothing checks that the skip actually runs through
`IcebergWriter`. If the delete-key getters were always `null`, every writer
test would still pass. Could we also assert the equality-delete record count
from `prepareCommit()`? For example, checkpoint 2 (updating a row that only
exists in a committed file) should produce 1 equality-delete record instead of
2.
##########
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:
The delete-key getters look up columns by name in the CDC schema, but the
writer maps columns by position. For pre-existing tables nothing checks that
the two column orders match. If they differ and the types happen to be
compatible, we could compare the wrong column and skip a `DELETE` that is still
needed, which leaves a duplicate row. The misalignment itself is an existing
issue, but it might be worth checking that each column sits at the same
position in both schemas, and falling back to always emitting the `DELETE` if
not. The check would run only once, when the factory is built.
--
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]
