This is an automated email from the ASF dual-hosted git repository.
yuxiqian pushed a commit to branch release-3.7
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/release-3.7 by this push:
new ea6595f39 [FLINK-41057][iceberg] Fix append-only update and delete
handling (#4588) (#4606)
ea6595f39 is described below
commit ea6595f39ff6aaa8ca211e35a46ceea12394833d
Author: Kunni <[email protected]>
AuthorDate: Mon Nov 2 15:53:15 2026 +0800
[FLINK-41057][iceberg] Fix append-only update and delete handling (#4588)
(#4606)
---
.../iceberg/sink/utils/RowDataUtils.java | 20 +-
.../connectors/iceberg/sink/v2/IcebergWriter.java | 50 ++-
.../iceberg/sink/utils/RowDataUtilsTest.java | 118 +++++++
.../sink/v2/IcebergWriterDeleteCapabilityTest.java | 375 +++++++++++++++++++++
.../iceberg/sink/v2/IcebergWriterTest.java | 128 ++++++-
5 files changed, 670 insertions(+), 21 deletions(-)
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/main/java/org/apache/flink/cdc/connectors/iceberg/sink/utils/RowDataUtils.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/main/java/org/apache/flink/cdc/connectors/iceberg/sink/utils/RowDataUtils.java
index 51b6a8595..cf72a8c42 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/main/java/org/apache/flink/cdc/connectors/iceberg/sink/utils/RowDataUtils.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/main/java/org/apache/flink/cdc/connectors/iceberg/sink/utils/RowDataUtils.java
@@ -31,25 +31,37 @@ import java.util.List;
public class RowDataUtils {
/**
- * Convert a {@link DataChangeEvent} to one or two {@link RowData}.
+ * Convert a {@link DataChangeEvent} to zero, one or two {@link RowData}.
*
- * <p>An {@code UPDATE} is split into a {@code DELETE} built from the
before-image followed by
- * an {@code INSERT} built from the after-image, rather than a single
after-image row. Iceberg
+ * <p>For tables with equality fields, an {@code UPDATE} is split into a
{@code DELETE} built
+ * from the before-image followed by an {@code INSERT} built from the
after-image. Iceberg
* derives both the equality-delete key and the delete's target partition
from the row passed to
* the writer; collapsing an UPDATE to just the after-image (as before)
means a partition key
* change is never deleted from its old partition, leaving a stale
duplicate behind.
+ *
+ * <p>Append-only writers do not interpret {@link RowKind#DELETE}. For
tables without equality
+ * fields, only the after-image of an update is appended, and deletes are
ignored.
*/
public static List<RowData> convertDataChangeEventToRowData(
- DataChangeEvent dataChangeEvent, List<RecordData.FieldGetter>
fieldGetters) {
+ DataChangeEvent dataChangeEvent,
+ List<RecordData.FieldGetter> fieldGetters,
+ boolean supportsEqualityDeletes) {
switch (dataChangeEvent.op()) {
case INSERT:
case REPLACE:
return Collections.singletonList(
toRowData(dataChangeEvent.after(), fieldGetters,
RowKind.INSERT));
case DELETE:
+ if (!supportsEqualityDeletes) {
+ return Collections.emptyList();
+ }
return Collections.singletonList(
toRowData(dataChangeEvent.before(), fieldGetters,
RowKind.DELETE));
case UPDATE:
+ if (!supportsEqualityDeletes) {
+ return Collections.singletonList(
+ toRowData(dataChangeEvent.after(), fieldGetters,
RowKind.INSERT));
+ }
return Arrays.asList(
toRowData(dataChangeEvent.before(), fieldGetters,
RowKind.DELETE),
toRowData(dataChangeEvent.after(), fieldGetters,
RowKind.INSERT));
diff --git
a/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
b/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
index 6945ed57f..f0afbc37d 100644
---
a/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
+++
b/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
@@ -64,7 +64,7 @@ public class IcebergWriter
public static final long DEFAULT_MAX_FILE_SIZE = 256 * 1024 * 1024;
- private Map<TableId, RowDataTaskWriterFactory> writerFactoryMap;
+ private Map<TableId, TableWriterFactory> writerFactoryMap;
private Map<TableId, TaskWriter<RowData>> writerMap;
@@ -140,9 +140,10 @@ public class IcebergWriter
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,
@@ -150,10 +151,12 @@ public class IcebergWriter
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
@@ -161,14 +164,20 @@ public class IcebergWriter
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);
+ if (rowDataList.isEmpty()) {
+ return;
+ }
+ TaskWriter<RowData> writer =
+ writerMap.computeIfAbsent(
+ tableId, tableId1 ->
writerFactory.taskWriterFactory.create());
for (RowData rowData : rowDataList) {
writer.write(rowData);
}
@@ -193,13 +202,15 @@ public class IcebergWriter
@Override
public void flush(boolean flush) throws IOException {
- // Clear the factory cache so the next write picks up the latest
catalog schema.
- // Writers keep running; schema-change splits are handled in
flushTableWriter.
- writerFactoryMap.clear();
+ // Keep each active writer paired with the factory that defines its
delete capability.
+ // Factories without writers can reload the latest catalog schema on
the next write.
+ writerFactoryMap.keySet().retainAll(writerMap.keySet());
}
private void flushTableWriter(TableId tableId) throws IOException {
TaskWriter<RowData> writer = writerMap.remove(tableId);
+ // An ignored append-only DELETE can populate the factory without
creating a writer.
+ writerFactoryMap.remove(tableId);
// Advance even when no writer exists, to keep batchIndex in sync
across subtasks.
int batchIndex = tableBatchIndexMap.getOrDefault(tableId, 0);
tableBatchIndexMap.put(tableId, batchIndex + 1);
@@ -216,7 +227,6 @@ public class IcebergWriter
batchIndex);
temporaryWriteResult.add(writeResultWrapper);
LOGGER.info(writeResultWrapper.buildDescription());
- writerFactoryMap.remove(tableId);
}
private List<WriteResultWrapper> getWriteResult() throws IOException {
@@ -271,4 +281,18 @@ public class IcebergWriter
catalog = null;
}
+
+ /** Cache the writer factory and its delete capability from the same
target-table schema. */
+ private static class TableWriterFactory {
+
+ private final RowDataTaskWriterFactory taskWriterFactory;
+
+ private final boolean supportsEqualityDeletes;
+
+ private TableWriterFactory(
+ RowDataTaskWriterFactory taskWriterFactory, boolean
supportsEqualityDeletes) {
+ this.taskWriterFactory = taskWriterFactory;
+ this.supportsEqualityDeletes = supportsEqualityDeletes;
+ }
+ }
}
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/test/java/org/apache/flink/cdc/connectors/iceberg/sink/utils/RowDataUtilsTest.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/test/java/org/apache/flink/cdc/connectors/iceberg/sink/utils/RowDataUtilsTest.java
new file mode 100644
index 000000000..d251042da
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/test/java/org/apache/flink/cdc/connectors/iceberg/sink/utils/RowDataUtilsTest.java
@@ -0,0 +1,118 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.iceberg.sink.utils;
+
+import org.apache.flink.cdc.common.data.RecordData;
+import org.apache.flink.cdc.common.data.binary.BinaryStringData;
+import org.apache.flink.cdc.common.event.DataChangeEvent;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.common.types.DataType;
+import org.apache.flink.cdc.common.types.DataTypes;
+import org.apache.flink.cdc.runtime.typeutils.BinaryRecordDataGenerator;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.types.RowKind;
+
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.time.ZoneId;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for converting CDC events for append-only and equality-delete
writers. */
+class RowDataUtilsTest {
+
+ private static final TableId TABLE_ID =
TableId.parse("test.iceberg_table");
+
+ private static final Schema SCHEMA =
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.BIGINT().notNull())
+ .physicalColumn("value", DataTypes.STRING())
+ .build();
+
+ private final List<RecordData.FieldGetter> fieldGetters =
+ IcebergTypeUtils.createFieldGetters(SCHEMA, ZoneId.of("UTC"));
+
+ private final BinaryRecordDataGenerator generator =
+ new
BinaryRecordDataGenerator(SCHEMA.getColumnDataTypes().toArray(new DataType[0]));
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testInsertAndReplacePreserveAfterImage(boolean
supportsEqualityDeletes) {
+ RecordData after = record("new");
+ for (DataChangeEvent event :
+ Arrays.asList(
+ DataChangeEvent.insertEvent(TABLE_ID, after),
+ DataChangeEvent.replaceEvent(TABLE_ID, after, null))) {
+ List<RowData> rows =
+ RowDataUtils.convertDataChangeEventToRowData(
+ event, fieldGetters, supportsEqualityDeletes);
+ assertThat(rows).hasSize(1);
+ assertRow(rows.get(0), RowKind.INSERT, "new");
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testUpdateUsesBeforeImageOnlyWithEqualityDeletes(boolean
supportsEqualityDeletes) {
+ List<RowData> rows =
+ RowDataUtils.convertDataChangeEventToRowData(
+ DataChangeEvent.updateEvent(TABLE_ID, record("old"),
record("new")),
+ fieldGetters,
+ supportsEqualityDeletes);
+
+ if (supportsEqualityDeletes) {
+ assertThat(rows).hasSize(2);
+ assertRow(rows.get(0), RowKind.DELETE, "old");
+ assertRow(rows.get(1), RowKind.INSERT, "new");
+ } else {
+ assertThat(rows).hasSize(1);
+ assertRow(rows.get(0), RowKind.INSERT, "new");
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testDeleteIsIgnoredWithoutEqualityDeletes(boolean
supportsEqualityDeletes) {
+ List<RowData> rows =
+ RowDataUtils.convertDataChangeEventToRowData(
+ DataChangeEvent.deleteEvent(TABLE_ID, record("old")),
+ fieldGetters,
+ supportsEqualityDeletes);
+
+ if (supportsEqualityDeletes) {
+ assertThat(rows).hasSize(1);
+ assertRow(rows.get(0), RowKind.DELETE, "old");
+ } else {
+ assertThat(rows).isEmpty();
+ }
+ }
+
+ private RecordData record(String value) {
+ return generator.generate(new Object[] {1L,
BinaryStringData.fromString(value)});
+ }
+
+ private static void assertRow(RowData row, RowKind rowKind, String value) {
+ assertThat(row.getRowKind()).isEqualTo(rowKind);
+ assertThat(row.getLong(0)).isEqualTo(1L);
+ assertThat(row.getString(1).toString()).isEqualTo(value);
+ }
+}
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/test/java/org/apache/flink/cdc/connectors/iceberg/sink/v2/IcebergWriterDeleteCapabilityTest.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/test/java/org/apache/flink/cdc/connectors/iceberg/sink/v2/IcebergWriterDeleteCapabilityTest.java
new file mode 100644
index 000000000..cd2da2250
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/test/java/org/apache/flink/cdc/connectors/iceberg/sink/v2/IcebergWriterDeleteCapabilityTest.java
@@ -0,0 +1,375 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.iceberg.sink.v2;
+
+import org.apache.flink.cdc.common.data.RecordData;
+import org.apache.flink.cdc.common.data.binary.BinaryStringData;
+import org.apache.flink.cdc.common.event.CreateTableEvent;
+import org.apache.flink.cdc.common.event.DataChangeEvent;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.common.types.DataType;
+import org.apache.flink.cdc.common.types.DataTypes;
+import org.apache.flink.cdc.connectors.iceberg.sink.IcebergMetadataApplier;
+import org.apache.flink.cdc.runtime.typeutils.BinaryRecordDataGenerator;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.CatalogUtil;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.data.IcebergGenerics;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.io.CloseableIterable;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.lang.reflect.Field;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Proxy;
+import java.nio.file.Path;
+import java.time.ZoneId;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests target-table delete capabilities and their per-table factory cache.
*/
+class IcebergWriterDeleteCapabilityTest {
+
+ @TempDir private Path temporaryFolder;
+
+ private final BinaryRecordDataGenerator generator =
+ new BinaryRecordDataGenerator(
+ schema(false, false).getColumnDataTypes().toArray(new
DataType[0]));
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testMixedTableDeleteCapabilities(boolean keyedTableFirst) throws
Exception {
+ TableId keyedTable = TableId.parse("test.keyed_table");
+ TableId appendTable = TableId.parse("test.append_table");
+ List<TableId> writeOrder =
+ keyedTableFirst
+ ? Arrays.asList(keyedTable, appendTable)
+ : Arrays.asList(appendTable, keyedTable);
+ RecordData north = record(1L, "north");
+ RecordData east = record(1L, "east");
+
+ try (TestContext context = new TestContext()) {
+ Schema keyedSchema = schema(true, true);
+ Schema appendSchema = schema(false, true);
+ context.createTable(keyedTable, keyedSchema, keyedSchema);
+ context.createTable(appendTable, appendSchema, appendSchema);
+ for (TableId tableId : writeOrder) {
+ context.writer.write(DataChangeEvent.insertEvent(tableId,
north), null);
+ }
+ context.commit();
+
+ // Rebuild both factories after the checkpoint, interleaving
tables with opposite
+ // capabilities. The keyed row must move partitions; the append
table keeps both.
+ for (TableId tableId : writeOrder) {
+ context.writer.write(DataChangeEvent.updateEvent(tableId,
north, east), null);
+ }
+ context.commit();
+ assertThat(context.read(keyedTable)).containsExactly("1, east");
+
assertThat(context.read(appendTable)).containsExactlyInAnyOrder("1, north", "1,
east");
+
+ for (TableId tableId : writeOrder) {
+ context.writer.write(DataChangeEvent.deleteEvent(tableId,
east), null);
+ }
+ context.commit();
+ assertThat(context.read(keyedTable)).isEmpty();
+
assertThat(context.read(appendTable)).containsExactlyInAnyOrder("1, north", "1,
east");
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testExistingKeyedTargetWithoutCdcPrimaryKey(boolean partitioned)
throws Exception {
+ TableId tableId = TableId.parse("test.existing_keyed_table");
+ RecordData north = record(1L, "north");
+ RecordData east = record(1L, "east");
+
+ try (TestContext context = new TestContext()) {
+ Schema targetSchema = schema(true, partitioned);
+ Schema cdcSchema = schema(false, partitioned);
+ assertThat(cdcSchema.primaryKeys()).isEmpty();
+ context.createTable(tableId, targetSchema, cdcSchema);
+ assertThat(
+ context.catalog
+
.loadTable(TableIdentifier.parse(tableId.identifier()))
+ .schema()
+ .identifierFieldIds())
+ .isNotEmpty();
+
+ context.writer.write(DataChangeEvent.insertEvent(tableId, north),
null);
+ context.commit();
+ context.writer.write(DataChangeEvent.updateEvent(tableId, north,
east), null);
+ context.commit();
+ assertThat(context.read(tableId)).containsExactly("1, east");
+
+ // A source-schema PK check would incorrectly ignore this delete.
+ context.writer.write(DataChangeEvent.deleteEvent(tableId, east),
null);
+ context.commit();
+ assertThat(context.read(tableId)).isEmpty();
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testCatalogLoadsOnlyOnFactoryCacheMiss(boolean hasPrimaryKey) throws
Exception {
+ TableId tableId = TableId.parse("test.cached_table");
+ int rowCount = 20;
+
+ try (TestContext context = new TestContext()) {
+ Schema schema = schema(hasPrimaryKey, false);
+ context.createTable(tableId, schema, schema);
+ AtomicInteger loadTableCalls =
countWriterCatalogLoads(context.writer);
+
+ for (long id = 1; id <= rowCount; id++) {
+ context.writer.write(
+ DataChangeEvent.insertEvent(tableId, record(id,
"north")), null);
+ context.writer.write(
+ DataChangeEvent.updateEvent(
+ tableId, record(id, "north"), record(id,
"east")),
+ null);
+ }
+ assertThat(loadTableCalls.get()).isEqualTo(1);
+
+ // Flush keeps each active writer paired with its factory.
+ context.writer.flush(false);
+ assertThat(loadTableCalls.get()).isEqualTo(1);
+ for (long id = 1; id <= rowCount; id++) {
+ context.writer.write(
+ DataChangeEvent.updateEvent(
+ tableId, record(id, "east"), record(id,
"south")),
+ null);
+ }
+ assertThat(loadTableCalls.get()).isEqualTo(1);
+ context.commit();
+ assertThat(loadTableCalls.get()).isEqualTo(1);
+
+ // Checkpoint invalidation causes exactly one new load, even for a
delete-only batch.
+ for (long id = 1; id <= rowCount; id++) {
+ context.writer.write(
+ DataChangeEvent.deleteEvent(tableId, record(id,
"south")), null);
+ }
+ assertThat(loadTableCalls.get()).isEqualTo(2);
+ context.commit();
+ assertThat(loadTableCalls.get()).isEqualTo(2);
+ assertThat(context.read(tableId)).hasSize(hasPrimaryKey ? 0 :
rowCount * 3);
+ }
+ }
+
+ @ParameterizedTest
+ @CsvSource({"false,false", "false,true", "true,false", "true,true"})
+ void testFlushPreservesActiveWriterCapability(boolean hasPrimaryKey,
boolean partitioned)
+ throws Exception {
+ TableId tableId = TableId.parse("test.active_table");
+ TableId otherTableId = TableId.parse("test.other_table");
+ RecordData north = record(1L, "north");
+ RecordData east = record(1L, "east");
+
+ try (TestContext context = new TestContext()) {
+ Schema schema = schema(hasPrimaryKey, partitioned);
+ context.createTable(tableId, schema, schema);
+ AtomicInteger loadTableCalls =
countWriterCatalogLoads(context.writer);
+ context.writer.write(DataChangeEvent.insertEvent(tableId, north),
null);
+
+ // A new table's FlushEvent must not detach another table's active
writer.
+ context.writer.flush(false);
+ context.createTable(otherTableId, schema, schema);
+ context.catalog
+ .loadTable(TableIdentifier.parse(tableId.identifier()))
+ .updateSchema()
+ .setIdentifierFields(hasPrimaryKey ? new String[0] : new
String[] {"id"})
+ .commit();
+ context.writer.write(DataChangeEvent.updateEvent(tableId, north,
east), null);
+ context.commit();
+
+ if (hasPrimaryKey) {
+ assertThat(context.read(tableId)).containsExactly("1, east");
+ } else {
+
assertThat(context.read(tableId)).containsExactlyInAnyOrder("1, north", "1,
east");
+ }
+ assertThat(loadTableCalls.get()).isEqualTo(1);
+
+ // After the checkpoint, both conversion and the new writer use
the new capability.
+ context.writer.write(DataChangeEvent.deleteEvent(tableId, north),
null);
+ context.writer.write(DataChangeEvent.deleteEvent(tableId, east),
null);
+ context.commit();
+ assertThat(loadTableCalls.get()).isEqualTo(2);
+ if (hasPrimaryKey) {
+ assertThat(context.read(tableId)).containsExactly("1, east");
+ } else {
+ assertThat(context.read(tableId)).isEmpty();
+ }
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testFlushInvalidatesFactoryWithoutWriter(boolean partitioned) throws
Exception {
+ TableId tableId = TableId.parse("test.idle_table");
+ RecordData north = record(1L, "north");
+ RecordData east = record(1L, "east");
+
+ try (TestContext context = new TestContext()) {
+ Schema schema = schema(false, partitioned);
+ context.createTable(tableId, schema, schema);
+ context.writer.write(DataChangeEvent.insertEvent(tableId, north),
null);
+ context.commit();
+ AtomicInteger loadTableCalls =
countWriterCatalogLoads(context.writer);
+
+ context.writer.write(DataChangeEvent.deleteEvent(tableId, north),
null);
+ assertThat(loadTableCalls.get()).isEqualTo(1);
+ context.writer.flush(false);
+ context.catalog
+ .loadTable(TableIdentifier.parse(tableId.identifier()))
+ .updateSchema()
+ .setIdentifierFields("id")
+ .commit();
+ context.writer.write(DataChangeEvent.updateEvent(tableId, north,
east), null);
+ context.commit();
+
+ assertThat(loadTableCalls.get()).isEqualTo(2);
+ assertThat(context.read(tableId)).containsExactly("1, east");
+ }
+ }
+
+ private static AtomicInteger countWriterCatalogLoads(IcebergWriter writer)
throws Exception {
+ Field catalogField = IcebergWriter.class.getDeclaredField("catalog");
+ catalogField.setAccessible(true);
+ Catalog delegate = (Catalog) catalogField.get(writer);
+ AtomicInteger loadTableCalls = new AtomicInteger();
+ Catalog countingCatalog =
+ (Catalog)
+ Proxy.newProxyInstance(
+ Catalog.class.getClassLoader(),
+ new Class<?>[] {Catalog.class},
+ (proxy, method, args) -> {
+ if ("loadTable".equals(method.getName())) {
+ loadTableCalls.incrementAndGet();
+ }
+ try {
+ return method.invoke(delegate, args);
+ } catch (InvocationTargetException e) {
+ throw e.getCause();
+ }
+ });
+ catalogField.set(writer, countingCatalog);
+ return loadTableCalls;
+ }
+
+ private static Schema schema(boolean hasPrimaryKey, boolean partitioned) {
+ Schema.Builder builder =
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.BIGINT().notNull())
+ .physicalColumn("region",
DataTypes.STRING().notNull());
+ if (hasPrimaryKey) {
+ builder.primaryKey("id");
+ }
+ if (partitioned) {
+ builder.partitionKey("region");
+ }
+ return builder.build();
+ }
+
+ private RecordData record(long id, String region) {
+ return generator.generate(new Object[] {id,
BinaryStringData.fromString(region)});
+ }
+
+ private class TestContext implements AutoCloseable {
+
+ private final Catalog catalog;
+
+ private final IcebergMetadataApplier metadataApplier;
+
+ private final IcebergWriter writer;
+
+ private final IcebergCommitter committer;
+
+ private TestContext() {
+ Map<String, String> catalogOptions = new HashMap<>();
+ catalogOptions.put("type", "hadoop");
+ catalogOptions.put(
+ "warehouse",
temporaryFolder.resolve(UUID.randomUUID().toString()).toString());
+ catalogOptions.put("cache-enabled", "false");
+ catalog =
+ CatalogUtil.buildIcebergCatalog(
+ "cdc-iceberg-catalog", catalogOptions, new
Configuration());
+ metadataApplier = new IcebergMetadataApplier(catalogOptions);
+ writer =
+ new IcebergWriter(
+ catalogOptions,
+ 1,
+ 1,
+ ZoneId.of("UTC"),
+ 0,
+ UUID.randomUUID().toString(),
+ UUID.randomUUID().toString(),
+ new HashMap<>());
+ committer = new IcebergCommitter(catalogOptions, new HashMap<>());
+ }
+
+ private void createTable(TableId tableId, Schema targetSchema, Schema
cdcSchema) {
+ metadataApplier.applySchemaChange(new CreateTableEvent(tableId,
targetSchema));
+ try {
+ writer.write(new CreateTableEvent(tableId, cdcSchema), null);
+ } catch (java.io.IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private void commit() throws Exception {
+ committer.commit(
+ writer.prepareCommit().stream()
+ .map(IcebergWriterTest.MockCommitRequestImpl::new)
+ .collect(Collectors.toList()));
+ }
+
+ private List<String> read(TableId tableId) throws Exception {
+ List<String> result = new ArrayList<>();
+ try (CloseableIterable<Record> records =
+ IcebergGenerics.read(
+
catalog.loadTable(TableIdentifier.parse(tableId.identifier())))
+ .build()) {
+ for (Record record : records) {
+ result.add(record.getField("id") + ", " +
record.getField("region"));
+ }
+ }
+ return result;
+ }
+
+ @Override
+ public void close() throws Exception {
+ writer.close();
+ committer.close();
+ if (catalog instanceof AutoCloseable) {
+ ((AutoCloseable) catalog).close();
+ }
+ }
+ }
+}
diff --git
a/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
b/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
index 4eda834ed..b9cb71b5d 100644
---
a/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
+++
b/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
@@ -60,6 +60,8 @@ import org.apache.iceberg.types.Types;
import org.assertj.core.api.Assertions;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
import java.io.File;
import java.math.BigDecimal;
@@ -649,15 +651,13 @@ public class IcebergWriterTest {
generator.generate(new Object[] {1L,
BinaryStringData.fromString("b")});
icebergWriter.write(DataChangeEvent.insertEvent(tableId, recordA),
null);
- // flush resets the factory cache but keeps the writer running, so the
update
- // lands in the same writer as the insert and uses a position delete.
+ // Flush keeps the active writer and its factory, so the update uses a
position delete.
icebergWriter.flush(false);
icebergWriter.write(DataChangeEvent.updateEvent(tableId, recordA,
recordB), null);
Collection<WriteResultWrapper> writeResults =
icebergWriter.prepareCommit();
- // Both writes went through the same writer (batchIndex 0) since flush
only
- // cleared the factory. Position delete handles dedup within the
snapshot.
+ // Both writes use the same writer (batchIndex 0) and deduplicate
within the snapshot.
List<Integer> batchIndexes =
writeResults.stream()
.map(WriteResultWrapper::getBatchIndex)
@@ -1647,6 +1647,126 @@ public class IcebergWriterTest {
Assertions.assertThat(result).containsExactly("1, east");
}
+ @ParameterizedTest
+ @CsvSource({"false,false", "false,true", "true,false", "true,true"})
+ void testAppendOnlyUpdatesAndDeletes(boolean partitioned, boolean
cdcHasPrimaryKey)
+ 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());
+ TableId tableId = TableId.parse("test.iceberg_table");
+ Schema.Builder schemaBuilder =
+ Schema.newBuilder()
+ .physicalColumn("id", DataTypes.BIGINT().notNull())
+ .physicalColumn("region",
DataTypes.STRING().notNull());
+ if (partitioned) {
+ schemaBuilder.partitionKey("region");
+ }
+ Schema targetSchema = schemaBuilder.build();
+ IcebergMetadataApplier metadataApplier = new
IcebergMetadataApplier(catalogOptions);
+ metadataApplier.applySchemaChange(new CreateTableEvent(tableId,
targetSchema));
+ Assertions.assertThat(
+
catalog.loadTable(TableIdentifier.parse(tableId.identifier()))
+ .schema()
+ .identifierFieldIds())
+ .isEmpty();
+
+ // An existing append-only target must stay append-only even if the
CDC schema has a PK.
+ Schema cdcSchema =
+ cdcHasPrimaryKey
+ ? Schema.newBuilder()
+ .setColumns(targetSchema.getColumns())
+ .partitionKey(targetSchema.partitionKeys())
+ .primaryKey("id")
+ .build()
+ : targetSchema;
+ BinaryRecordDataGenerator generator =
+ new BinaryRecordDataGenerator(
+ targetSchema.getColumnDataTypes().toArray(new
DataType[0]));
+ RecordData north =
+ generator.generate(new Object[] {1L,
BinaryStringData.fromString("north")});
+ RecordData east =
+ generator.generate(new Object[] {1L,
BinaryStringData.fromString("east")});
+ RecordData south =
+ generator.generate(new Object[] {1L,
BinaryStringData.fromString("south")});
+
+ try (IcebergWriter writer =
+ new IcebergWriter(
+ catalogOptions,
+ 1,
+ 1,
+ ZoneId.of("UTC"),
+ 0,
+ UUID.randomUUID().toString(),
+ UUID.randomUUID().toString(),
+ new HashMap<>());
+ IcebergCommitter committer =
+ new IcebergCommitter(catalogOptions, new HashMap<>()))
{
+ writer.write(new CreateTableEvent(tableId, cdcSchema), null);
+ writer.write(DataChangeEvent.insertEvent(tableId, north), null);
+ writer.write(DataChangeEvent.updateEvent(tableId, north, east),
null);
+ committer.commit(
+ writer.prepareCommit().stream()
+ .map(MockCommitRequestImpl::new)
+ .collect(Collectors.toList()));
+ Assertions.assertThat(fetchTableContent(catalog, tableId, null))
+ .containsExactlyInAnyOrder("1, north", "1, east");
+
+ // A delete-only checkpoint must neither append the old row nor
create a writer.
+ writer.write(DataChangeEvent.deleteEvent(tableId, north), null);
+ Assertions.assertThat(writer.prepareCommit()).isEmpty();
+ Assertions.assertThat(fetchTableContent(catalog, tableId, null))
+ .containsExactlyInAnyOrder("1, north", "1, east");
+
+ // The capability must also be correct after checkpoint cache
invalidation.
+ writer.write(DataChangeEvent.updateEvent(tableId, east, south),
null);
+ committer.commit(
+ writer.prepareCommit().stream()
+ .map(MockCommitRequestImpl::new)
+ .collect(Collectors.toList()));
+ Assertions.assertThat(fetchTableContent(catalog, tableId, null))
+ .containsExactlyInAnyOrder("1, north", "1, east", "1,
south");
+
+ // An ignored delete caches a factory without a writer. Schema
changes must still
+ // invalidate that factory so subsequent inserts use the new
physical schema.
+ writer.write(DataChangeEvent.deleteEvent(tableId, south), null);
+ AddColumnEvent addColumn =
+ new AddColumnEvent(
+ tableId,
+ Arrays.asList(
+ new AddColumnEvent.ColumnWithPosition(
+ new PhysicalColumn(
+ "extra",
DataTypes.STRING(), null))));
+ metadataApplier.applySchemaChange(addColumn);
+ writer.write(addColumn, null);
+ Schema evolvedSchema =
SchemaUtils.applySchemaChangeEvent(cdcSchema, addColumn);
+ BinaryRecordDataGenerator evolvedGenerator =
+ new BinaryRecordDataGenerator(
+ evolvedSchema.getColumnDataTypes().toArray(new
DataType[0]));
+ RecordData west =
+ evolvedGenerator.generate(
+ new Object[] {
+ 1L,
+ BinaryStringData.fromString("west"),
+ BinaryStringData.fromString("added")
+ });
+ writer.write(DataChangeEvent.insertEvent(tableId, west), null);
+ committer.commit(
+ writer.prepareCommit().stream()
+ .map(MockCommitRequestImpl::new)
+ .collect(Collectors.toList()));
+ Assertions.assertThat(fetchTableContent(catalog, tableId, null))
+ .containsExactlyInAnyOrder(
+ "1, north, null", "1, east, null", "1, south,
null", "1, west, added");
+ }
+ }
+
private static long countSnapshots(Table table) {
long count = 0;
for (Snapshot ignored : table.snapshots()) {
