Copilot commented on code in PR #4360:
URL: https://github.com/apache/flink-cdc/pull/4360#discussion_r3031533042
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/main/java/org/apache/flink/cdc/connectors/iceberg/sink/v2/IcebergCommitter.java:
##########
@@ -91,74 +98,133 @@ private void commit(List<WriteResultWrapper>
writeResultWrappers) {
if (writeResultWrappers.isEmpty()) {
return;
}
- // all commits a same checkpoint-id
long checkpointId = writeResultWrappers.get(0).getCheckpointId();
String newFlinkJobId = writeResultWrappers.get(0).getJobId();
String operatorId = writeResultWrappers.get(0).getOperatorId();
- Map<TableId, List<WriteResult>> tableMap = new HashMap<>();
- for (WriteResultWrapper writeResultWrapper : writeResultWrappers) {
- List<WriteResult> writeResult =
- tableMap.getOrDefault(writeResultWrapper.getTableId(), new
ArrayList<>());
- writeResult.add(writeResultWrapper.getWriteResult());
- tableMap.put(writeResultWrapper.getTableId(), writeResult);
- LOGGER.info(writeResultWrapper.buildDescription());
+ Map<TableId, List<WriteResultWrapper>> tableMap = new HashMap<>();
+ for (WriteResultWrapper w : writeResultWrappers) {
+ tableMap.computeIfAbsent(w.getTableId(), k -> new
ArrayList<>()).add(w);
+ LOGGER.info(w.buildDescription());
}
- for (Map.Entry<TableId, List<WriteResult>> entry :
tableMap.entrySet()) {
+
+ for (Map.Entry<TableId, List<WriteResultWrapper>> entry :
tableMap.entrySet()) {
TableId tableId = entry.getKey();
+ // Sort ascending by batch index to guarantee correct Iceberg
sequence number ordering.
+ // Equality-delete files in batch N will have sequence number >
batch M (M < N), so
+ // they correctly supersede stale data written by earlier
same-checkpoint batches.
+ List<WriteResultWrapper> batches = entry.getValue();
+
batches.sort(Comparator.comparingInt(WriteResultWrapper::getBatchIndex));
+
Review Comment:
`commit()` currently sorts and iterates a `List<WriteResultWrapper>` per
table and commits each wrapper as its own Iceberg snapshot. In production, the
committer receives committables from *all* writer subtasks
(`committables.global()`), so there will be multiple wrappers with the same
`batchIndex` (one per subtask). Committing them separately can break
correctness: equality-deletes in one subtask’s wrapper may end up with a higher
sequence number than data files from another subtask in the same logical batch
and can incorrectly delete those rows. It also breaks idempotent retry because
the “already committed batchIndex” check can’t distinguish partially committed
subtasks within the same batchIndex.
Commit should be done once per (tableId, checkpointId, batchIndex) by
merging data/delete files from all wrappers with that batchIndex into a single
Append/RowDelta snapshot, then proceed to the next batchIndex.
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/main/java/org/apache/flink/cdc/connectors/iceberg/sink/v2/IcebergCommitter.java:
##########
@@ -91,74 +98,133 @@ private void commit(List<WriteResultWrapper>
writeResultWrappers) {
if (writeResultWrappers.isEmpty()) {
return;
}
- // all commits a same checkpoint-id
long checkpointId = writeResultWrappers.get(0).getCheckpointId();
String newFlinkJobId = writeResultWrappers.get(0).getJobId();
String operatorId = writeResultWrappers.get(0).getOperatorId();
- Map<TableId, List<WriteResult>> tableMap = new HashMap<>();
- for (WriteResultWrapper writeResultWrapper : writeResultWrappers) {
- List<WriteResult> writeResult =
- tableMap.getOrDefault(writeResultWrapper.getTableId(), new
ArrayList<>());
- writeResult.add(writeResultWrapper.getWriteResult());
- tableMap.put(writeResultWrapper.getTableId(), writeResult);
- LOGGER.info(writeResultWrapper.buildDescription());
+ Map<TableId, List<WriteResultWrapper>> tableMap = new HashMap<>();
+ for (WriteResultWrapper w : writeResultWrappers) {
+ tableMap.computeIfAbsent(w.getTableId(), k -> new
ArrayList<>()).add(w);
+ LOGGER.info(w.buildDescription());
}
- for (Map.Entry<TableId, List<WriteResult>> entry :
tableMap.entrySet()) {
+
+ for (Map.Entry<TableId, List<WriteResultWrapper>> entry :
tableMap.entrySet()) {
TableId tableId = entry.getKey();
+ // Sort ascending by batch index to guarantee correct Iceberg
sequence number ordering.
+ // Equality-delete files in batch N will have sequence number >
batch M (M < N), so
+ // they correctly supersede stale data written by earlier
same-checkpoint batches.
+ List<WriteResultWrapper> batches = entry.getValue();
+
batches.sort(Comparator.comparingInt(WriteResultWrapper::getBatchIndex));
+
Table table =
catalog.loadTable(
TableIdentifier.of(tableId.getSchemaName(),
tableId.getTableName()));
+ int startBatchIndex = 0;
Snapshot snapshot = table.currentSnapshot();
if (snapshot != null) {
Iterable<Snapshot> ancestors =
SnapshotUtil.ancestorsOf(snapshot.snapshotId(),
table::snapshot);
- long lastCheckpointId =
+ long lastCommittedCheckpointId =
getMaxCommittedCheckpointId(ancestors, newFlinkJobId,
operatorId);
- if (lastCheckpointId == checkpointId) {
+ if (lastCommittedCheckpointId >= checkpointId) {
LOGGER.warn(
"Checkpoint id {} has been committed to table {},
skipping",
checkpointId,
tableId.identifier());
continue;
}
+ ancestors = SnapshotUtil.ancestorsOf(snapshot.snapshotId(),
table::snapshot);
+ startBatchIndex =
+ getLastCommittedBatchIndex(
+ ancestors, newFlinkJobId, operatorId,
checkpointId)
+ + 1;
}
Review Comment:
`startBatchIndex` is computed from a *batch index* found in snapshot
summaries, but it is later used as a *list position* (`for (int i =
startBatchIndex; i < batches.size(); i++)`). This only works if `batches`
contains exactly one element per batch index and indices are contiguous
starting at 0. With multiple writer subtasks (multiple wrappers per
batchIndex), `startBatchIndex` will point into the middle of the list and can
either re-commit already committed wrappers (duplicates) or skip some wrappers
(data loss).
After grouping/merging by batchIndex, iterate over actual batchIndex values
(or find the first group with `batchIndex > lastCommittedBatchIndex`) rather
than using the batchIndex as a list offset.
##########
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:
##########
@@ -552,6 +556,485 @@ public void testWithRepeatCommit() throws Exception {
Assertions.assertThat(result).containsExactlyInAnyOrder("1, char1",
"2, char2");
}
+ @Test
+ public void testNoDuplicateWhenFlushSplitsSamePkUpdatesWithinCheckpoint()
throws Exception {
+ Map<String, String> catalogOptions = new HashMap<>();
+ String warehouse =
+ new File(temporaryFolder.toFile(),
UUID.randomUUID().toString()).toString();
+ catalogOptions.put("type", "hadoop");
+ catalogOptions.put("warehouse", warehouse);
+ catalogOptions.put("cache-enabled", "false");
+ Catalog catalog =
+ CatalogUtil.buildIcebergCatalog(
+ "cdc-iceberg-catalog", catalogOptions, new
Configuration());
+
+ String jobId = UUID.randomUUID().toString();
+ String operatorId = UUID.randomUUID().toString();
+ IcebergWriter icebergWriter =
+ new IcebergWriter(
+ catalogOptions, 1, 1, ZoneId.systemDefault(), 0,
jobId, operatorId);
+ IcebergMetadataApplier icebergMetadataApplier = new
IcebergMetadataApplier(catalogOptions);
+
+ TableId tableId = TableId.parse("test.iceberg_table");
+ CreateTableEvent createTableEvent =
+ new CreateTableEvent(
+ tableId,
+ Schema.newBuilder()
+ .physicalColumn("id",
DataTypes.BIGINT().notNull())
+ .physicalColumn("name", DataTypes.VARCHAR(100))
+ .primaryKey("id")
+ .build());
+ icebergMetadataApplier.applySchemaChange(createTableEvent);
+ icebergWriter.write(createTableEvent, null);
+
+ BinaryRecordDataGenerator generator =
+ new BinaryRecordDataGenerator(
+
createTableEvent.getSchema().getColumnDataTypes().toArray(new DataType[0]));
+
+ RecordData recordA =
+ generator.generate(new Object[] {1L,
BinaryStringData.fromString("a")});
+ RecordData recordB =
+ generator.generate(new Object[] {1L,
BinaryStringData.fromString("b")});
+
+ icebergWriter.write(DataChangeEvent.insertEvent(tableId, recordA),
null);
+ // flush(false) is a no-op; both events reach the same writer so the
position-delete
+ // mechanism within the writer handles dedup correctly.
+ icebergWriter.flush(false);
+ icebergWriter.write(DataChangeEvent.updateEvent(tableId, recordA,
recordB), null);
+
+ Collection<WriteResultWrapper> writeResults =
icebergWriter.prepareCommit();
+ IcebergCommitter icebergCommitter = new
IcebergCommitter(catalogOptions);
+ Collection<Committer.CommitRequest<WriteResultWrapper>> collection =
+
writeResults.stream().map(MockCommitRequestImpl::new).collect(Collectors.toList());
+ icebergCommitter.commit(collection);
+
+ List<String> result = fetchTableContent(catalog, tableId, null);
+ Assertions.assertThat(result).containsExactly("1, b");
+ }
+
+ /**
+ * Verifies that same-PK updates split across a schema-change-triggered
writer flush within one
+ * checkpoint do not produce duplicate records.
+ *
+ * <p>Root cause: when a schema-change event forces {@code
flushTableWriter}, the pre-change
+ * writes land in one {@code WriteResultWrapper} (batch 0) and the
post-change writes land in
+ * another (batch 1). Committing both in a single Iceberg {@code RowDelta}
would give all files
+ * the same sequence number N, so the equality-delete file from batch 1
(seq=N) cannot delete
+ * the data file from batch 0 (seq=N) — Iceberg equality-delete semantics
require strictly lower
+ * sequence numbers. The fix commits each batch as a separate sequential
snapshot, giving batch
+ * 1 a higher sequence number so its equality-delete correctly supersedes
batch 0's data.
+ */
+ @Test
+ public void testNoDuplicateWhenSchemaChangeFlushSplitsSamePkUpdates()
throws Exception {
+ Map<String, String> catalogOptions = new HashMap<>();
+ String warehouse =
+ new File(temporaryFolder.toFile(),
UUID.randomUUID().toString()).toString();
+ catalogOptions.put("type", "hadoop");
+ catalogOptions.put("warehouse", warehouse);
+ catalogOptions.put("cache-enabled", "false");
+ Catalog catalog =
+ CatalogUtil.buildIcebergCatalog(
+ "cdc-iceberg-catalog", catalogOptions, new
Configuration());
+
+ String jobId = UUID.randomUUID().toString();
+ String operatorId = UUID.randomUUID().toString();
+ IcebergWriter icebergWriter =
+ new IcebergWriter(
+ catalogOptions, 1, 1, ZoneId.systemDefault(), 0,
jobId, operatorId);
+ IcebergMetadataApplier icebergMetadataApplier = new
IcebergMetadataApplier(catalogOptions);
Review Comment:
The new tests exercise batch-splitting behavior using a single
`IcebergWriter` instance (one subtask). Since
`IcebergSink#addPreCommitTopology` uses `committables.global()`, the committer
will receive committables from multiple writer subtasks in production; it would
be valuable to add a test that creates 2 writers (different `taskId`s),
produces committables with the same `batchIndex` for a table, and verifies the
committer handles them correctly (no duplicates / no accidental deletes) when
committed together.
--
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]
