snapshotProperties);
+
+ abstract Builder setErrorHandling(boolean errorHandling);
+
+ abstract CdcWriteConfig build();
+ }
+}
diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/RecordDeltaTaskWriter.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/RecordDeltaTaskWriter.java
new file mode 100644
index 000000000000..63792e2f6fc6
--- /dev/null
+++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/RecordDeltaTaskWriter.java
@@ -0,0 +1,506 @@
+/*
+ * 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.beam.sdk.io.iceberg.cdc.sink;
+
+import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.apache.beam.sdk.values.ValueKind;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Maps;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Sets;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.primitives.Ints;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionKey;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.data.GenericFileWriterFactory;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.data.InternalRecordWrapper;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.FileWriterFactory;
+import org.apache.iceberg.io.OutputFileFactory;
+import org.apache.iceberg.io.RollingDataWriter;
+import org.apache.iceberg.io.RollingEqualityDeleteWriter;
+import org.apache.iceberg.io.WriteResult;
+import org.apache.iceberg.types.TypeUtil;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.PropertyUtil;
+import org.apache.iceberg.util.Tasks;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+/**
+ * Writes one sorted {@code (destination, shard, window)} group, collapsing each primary key's
+ * changes into at most one equality delete and one data row.
+ *
+ * The group arrives sorted by {@link CdcSortKey}, so one key's records are contiguous and in
+ * (sequence, kind) order. The writer holds a block (the sequence of records for the current key)
+ * and flushes it when the key changes.
+ *
+ *
The block's last record is the key's final state; its first record tells us whether
+ * anything preceded this window.
+ *
+ *
+ * - Opens with INSERT: the key was born this window, so no earlier commit holds it and
+ * no delete is written, even if the key dies again before the window ends.
+ *
- Opens with anything else: an earlier commit may hold the key, so a delete is written
+ * once an UPDATE_BEFORE or DELETE appears. A block of only UPDATE_AFTERs writes no deletes.
+ *
- Upsert mode: always creates a delete.
+ *
- Ends with INSERT or UPDATE_AFTER: the final row is written. Otherwise, the key is
+ * gone and nothing is written.
+ *
+ *
+ * Partition routing
+ *
+ * The data row routes by the block's last record (the partition the key now lives in).
+ * The equality delete routes by the block's first record (the partition the committed row
+ * still lives in). Those differ whenever an update moved the row. {@code kindRank} ranks
+ * UPDATE_BEFORE and DELETE ahead of the after-images at an equal sequence, so the block opens with
+ * a before-image whenever the window's first change carries one, a guarantee that holds within one
+ * commit window only. Upsert has no before-images, but it requires partition source columns to be
+ * equality columns, so there every record of a block routes alike.
+ *
+ *
Hence the input contract for a table partitioned on non-key columns: every update must carry
+ * its UPDATE_BEFORE. A block opening with an after-image can only route its delete to the partition
+ * the row moved to, leaving the committed row unreachable in the one it moved from.
+ *
+ *
This never writes position deletes, and no deletion vectors on V3. Those exist to retract a
+ * row that was already flushed when a later change superseded it. Collapsing means the superseded
+ * row is never written at all.
+ */
+abstract class RecordDeltaTaskWriter {
+
+ private final PartitionSpec spec;
+ private final FileWriterFactory writerFactory;
+ private final OutputFileFactory fileFactory;
+ private final FileIO io;
+ private final long targetFileSize;
+ private final Schema deleteSchema;
+
+ /** Column position in the table schema of each {@link #deleteSchema} field. */
+ private final int[] pkPos;
+
+ private final boolean upsert;
+
+ private final List partitionWriters = new ArrayList<>();
+
+ /** The previous record's sort key, for the unsorted-input tripwire in {@link #write}. */
+ private byte @Nullable [] lastSortKey;
+
+ /** The current block: sort key, opening and latest records/kinds, and delete-trigger flag. */
+ private byte @Nullable [] blockKey;
+
+ private @Nullable Record latestRecord;
+ private @Nullable ValueKind latestKind;
+ private @Nullable Record firstRecord;
+ private @Nullable ValueKind firstKind;
+ private boolean sawUbOrDelete;
+
+ RecordDeltaTaskWriter(
+ PartitionSpec spec,
+ FileWriterFactory writerFactory,
+ OutputFileFactory fileFactory,
+ FileIO io,
+ long targetFileSize,
+ Schema schema,
+ Schema deleteSchema,
+ boolean upsert) {
+ this.spec = spec;
+ this.writerFactory = writerFactory;
+ this.fileFactory = fileFactory;
+ this.io = io;
+ this.targetFileSize = targetFileSize;
+ this.deleteSchema = deleteSchema;
+ List pkFields = deleteSchema.columns();
+ this.pkPos = new int[pkFields.size()];
+ // pk should only be in top-level columns
+ List allFields = schema.columns();
+ Map positionById = Maps.newHashMapWithExpectedSize(allFields.size());
+ for (int j = 0; j < allFields.size(); j++) {
+ positionById.put(allFields.get(j).fieldId(), j);
+ }
+ for (int i = 0; i < pkFields.size(); i++) {
+ @Nullable Integer pos = positionById.get(pkFields.get(i).fieldId());
+ if (pos == null) {
+ throw new IllegalStateException(
+ "Equality field "
+ + pkFields.get(i).name()
+ + " is not a top-level column of schema: "
+ + schema);
+ }
+ this.pkPos[i] = pos;
+ }
+ this.upsert = upsert;
+ }
+
+ /** Routes a record to the {@link PartitionDeltaWriter} responsible for its partition. */
+ abstract PartitionDeltaWriter route(Record row);
+
+ /**
+ * Buffers {@code row} into the current block, flushing the previous block first when {@code
+ * sortKey} starts a new primary key.
+ */
+ public void write(byte[] sortKey, Record row, ValueKind kind) {
+ // The collapse is only correct over sorted input, so a regressing key must not be accepted.
+ if (lastSortKey != null && Arrays.compareUnsigned(sortKey, lastSortKey) < 0) {
+ throw new IllegalStateException(
+ "RecordDeltaTaskWriter received unsorted input: a record's sort key sorts below its "
+ + "predecessor's within the group.");
+ }
+ lastSortKey = sortKey.clone();
+ if (blockKey != null && !CdcSortKey.samePk(blockKey, sortKey)) {
+ // we're encountering a new PK. flush the current one
+ flushBlock();
+ }
+ if (blockKey == null) {
+ blockKey = sortKey.clone();
+ firstRecord = row;
+ firstKind = kind;
+ }
+ if (kind == ValueKind.UPDATE_BEFORE || kind == ValueKind.DELETE) {
+ sawUbOrDelete = true;
+ }
+ latestRecord = row;
+ latestKind = kind;
+ }
+
+ /** Flushes the current block per the class javadoc's rule and resets the block state. */
+ private void flushBlock() {
+ Record row = checkStateNotNull(latestRecord);
+ boolean deleteExistingRow;
+ if (upsert) {
+ deleteExistingRow = true; // any key may replace a row from an earlier commit
+ } else if (firstKind == ValueKind.INSERT) {
+ deleteExistingRow = false; // key born this window: no earlier commit holds it
+ } else {
+ // delete if we see a UPDATE_BEFORE/DELETE
+ deleteExistingRow = sawUbOrDelete;
+ }
+ boolean writeRow = latestKind == ValueKind.INSERT || latestKind == ValueKind.UPDATE_AFTER;
+
+ // The delete routes (and projects its key) by the block's first record: kindRank sorts
+ // UPDATE_BEFORE/DELETE ahead of after-images at an equal sequence, so the block opens with a
+ // before-image whenever the window's first change carries one.
+ // Upsert drops before-images, but it also requires partition sources to be equality columns,
+ // so there every record of the block routes alike.
+ // The write routes by the latest record, the key's final state: the block is sorted by
+ // sequence, with kindRank putting the after-image last at an equal sequence.
+ if (deleteExistingRow) {
+ Record first = checkStateNotNull(firstRecord);
+ route(first).delete(projectKey(first));
+ }
+ if (writeRow) {
+ route(row).write(row);
+ }
+ blockKey = null;
+ latestRecord = null;
+ latestKind = null;
+ firstRecord = null;
+ firstKind = null;
+ sawUbOrDelete = false;
+ }
+
+ /** Flushes the last block, closes every file, and returns the completed files. */
+ public WriteResult complete() throws IOException {
+ if (blockKey != null) {
+ flushBlock();
+ }
+ close();
+ WriteResult.Builder result = WriteResult.builder();
+ for (PartitionDeltaWriter writer : partitionWriters) {
+ result.addDataFiles(writer.dataFiles());
+ result.addDeleteFiles(writer.deleteFiles());
+ }
+ return result.build();
+ }
+
+ /**
+ * Closes every file and deletes it: a failed group must leave nothing behind. A close failure
+ * does not stop the deletes; it is rethrown once they are done.
+ */
+ public void abort() throws IOException {
+ @Nullable Exception closeFailure = null;
+ try {
+ close();
+ } catch (IOException | RuntimeException e) {
+ closeFailure = e;
+ }
+ List locations = new ArrayList<>();
+ for (PartitionDeltaWriter writer : partitionWriters) {
+ locations.addAll(writer.fileLocations());
+ }
+ try {
+ Tasks.foreach(locations).throwFailureWhenFinished().noRetry().run(io::deleteFile);
+ } catch (RuntimeException deleteFailure) {
+ if (closeFailure != null) {
+ deleteFailure.addSuppressed(closeFailure);
+ }
+ throw deleteFailure;
+ }
+ if (closeFailure instanceof IOException) {
+ throw (IOException) closeFailure;
+ }
+ if (closeFailure != null) {
+ throw (RuntimeException) closeFailure;
+ }
+ }
+
+ private void close() throws IOException {
+ Tasks.foreach(partitionWriters)
+ .throwFailureWhenFinished()
+ .noRetry()
+ .run(PartitionDeltaWriter::close, IOException.class);
+ }
+
+ /** Projects a full record onto a PK-only {@link Record} matching {@link #deleteSchema}. */
+ private Record projectKey(Record row) {
+ GenericRecord key = GenericRecord.create(deleteSchema);
+ for (int i = 0; i < pkPos.length; i++) {
+ key.set(i, row.get(pkPos[i], Object.class));
+ }
+ return key;
+ }
+
+ PartitionDeltaWriter newPartitionWriter(@Nullable PartitionKey partition) {
+ PartitionDeltaWriter writer = new PartitionDeltaWriter(partition);
+ partitionWriters.add(writer);
+ return writer;
+ }
+
+ @SuppressWarnings("argument")
+ private RollingDataWriter newDataWriter(@Nullable PartitionKey partition) {
+ return new RollingDataWriter<>(writerFactory, fileFactory, io, targetFileSize, spec, partition);
+ }
+
+ @SuppressWarnings("argument")
+ private RollingEqualityDeleteWriter newDeleteWriter(@Nullable PartitionKey partition) {
+ return new RollingEqualityDeleteWriter<>(
+ writerFactory, fileFactory, io, targetFileSize, spec, partition);
+ }
+
+ /** One partition's rolling data and equality-delete writers, each opened on first use. */
+ protected class PartitionDeltaWriter {
+ private final @Nullable PartitionKey partition;
+ private @Nullable RollingDataWriter dataWriter;
+ private @Nullable RollingEqualityDeleteWriter deleteWriter;
+ private boolean dataClosed;
+ private boolean deleteClosed;
+
+ PartitionDeltaWriter(@Nullable PartitionKey partition) {
+ this.partition = partition;
+ }
+
+ void write(Record row) {
+ @Nullable RollingDataWriter writer = dataWriter;
+ if (writer == null) {
+ writer = newDataWriter(partition);
+ dataWriter = writer;
+ }
+ writer.write(row);
+ }
+
+ void delete(Record key) {
+ @Nullable RollingEqualityDeleteWriter writer = deleteWriter;
+ if (writer == null) {
+ writer = newDeleteWriter(partition);
+ deleteWriter = writer;
+ }
+ writer.write(key);
+ }
+
+ void close() throws IOException {
+ try {
+ if (dataWriter != null) {
+ dataWriter.close();
+ dataClosed = true;
+ }
+ } finally {
+ if (deleteWriter != null) {
+ deleteWriter.close();
+ deleteClosed = true;
+ }
+ }
+ }
+
+ /** Completed files, plus the open file of a writer whose close failed. */
+ List fileLocations() {
+ List locations = new ArrayList<>();
+ @Nullable RollingDataWriter data = dataWriter;
+ if (data != null) {
+ if (dataClosed) {
+ for (DataFile file : data.result().dataFiles()) {
+ locations.add(file.location());
+ }
+ } else {
+ locations.add(data.currentFilePath().toString());
+ }
+ }
+ @Nullable RollingEqualityDeleteWriter deletes = deleteWriter;
+ if (deletes != null) {
+ if (deleteClosed) {
+ for (DeleteFile file : deletes.result().deleteFiles()) {
+ locations.add(file.location());
+ }
+ } else {
+ locations.add(deletes.currentFilePath().toString());
+ }
+ }
+ return locations;
+ }
+
+ List dataFiles() {
+ return dataWriter == null ? ImmutableList.of() : dataWriter.result().dataFiles();
+ }
+
+ List deleteFiles() {
+ return deleteWriter == null ? ImmutableList.of() : deleteWriter.result().deleteFiles();
+ }
+ }
+
+ /** Record writer for an unpartitioned table. */
+ static class UnpartitionedRecordDeltaWriter extends RecordDeltaTaskWriter {
+ private final PartitionDeltaWriter writer;
+
+ @SuppressWarnings("method.invocation")
+ UnpartitionedRecordDeltaWriter(
+ PartitionSpec spec,
+ FileWriterFactory writerFactory,
+ OutputFileFactory fileFactory,
+ FileIO io,
+ long targetFileSize,
+ Schema schema,
+ Schema deleteSchema,
+ boolean upsert) {
+ super(spec, writerFactory, fileFactory, io, targetFileSize, schema, deleteSchema, upsert);
+ this.writer = newPartitionWriter(null);
+ }
+
+ @Override
+ PartitionDeltaWriter route(Record row) {
+ return writer;
+ }
+ }
+
+ /**
+ * Partitioned table: a fanout delta writer per partition key, created lazily on first touch and
+ * held open, because the group is sorted by PK and partitions interleave.
+ */
+ static class PartitionedRecordDeltaWriter extends RecordDeltaTaskWriter {
+ private final PartitionKey partitionKey;
+ private final InternalRecordWrapper wrapper;
+ private final Map writers = Maps.newHashMap();
+
+ PartitionedRecordDeltaWriter(
+ PartitionSpec spec,
+ FileWriterFactory writerFactory,
+ OutputFileFactory fileFactory,
+ FileIO io,
+ long targetFileSize,
+ Schema schema,
+ Schema deleteSchema,
+ boolean upsert) {
+ super(spec, writerFactory, fileFactory, io, targetFileSize, schema, deleteSchema, upsert);
+ this.partitionKey = new PartitionKey(spec, schema);
+ this.wrapper = new InternalRecordWrapper(schema.asStruct());
+ }
+
+ @Override
+ PartitionDeltaWriter route(Record row) {
+ partitionKey.partition(wrapper.wrap(row));
+
+ @Nullable PartitionDeltaWriter writer = writers.get(partitionKey);
+ if (writer == null) {
+ // The shared partitionKey is mutated on every route() call; copy before keying the map.
+ PartitionKey copiedKey = partitionKey.copy();
+ writer = newPartitionWriter(copiedKey);
+ writers.put(copiedKey, writer);
+ }
+
+ return writer;
+ }
+ }
+
+ /** Builds a {@link RecordDeltaTaskWriter} writing under a specified {@code spec}. */
+ static RecordDeltaTaskWriter create(
+ Table table,
+ PartitionSpec spec,
+ Set equalityFieldIds,
+ boolean upsert,
+ long targetFileSizeBytes,
+ OutputFileFactory fileFactory,
+ FileFormat dataFormat,
+ FileFormat deleteFormat) {
+ Schema deleteSchema = TypeUtil.select(table.schema(), Sets.newHashSet(equalityFieldIds));
+ FileWriterFactory writerFactory =
+ new GenericFileWriterFactory.Builder(table)
+ .dataSchema(table.schema())
+ .dataFileFormat(dataFormat)
+ .deleteFileFormat(deleteFormat)
+ .equalityFieldIds(Ints.toArray(equalityFieldIds))
+ .equalityDeleteRowSchema(deleteSchema)
+ .build();
+
+ if (spec.isUnpartitioned()) {
+ return new UnpartitionedRecordDeltaWriter(
+ spec,
+ writerFactory,
+ fileFactory,
+ table.io(),
+ targetFileSizeBytes,
+ table.schema(),
+ deleteSchema,
+ upsert);
+ } else {
+ return new PartitionedRecordDeltaWriter(
+ spec,
+ writerFactory,
+ fileFactory,
+ table.io(),
+ targetFileSizeBytes,
+ table.schema(),
+ deleteSchema,
+ upsert);
+ }
+ }
+
+ /** The table's default data file format ({@code write.format.default}, Parquet fallback). */
+ static FileFormat dataFileFormat(Table table) {
+ return FileFormat.fromString(
+ PropertyUtil.propertyAsString(
+ table.properties(),
+ TableProperties.DEFAULT_FILE_FORMAT,
+ TableProperties.DEFAULT_FILE_FORMAT_DEFAULT));
+ }
+
+ /** The equality-delete file format: {@code write.delete.format.default}, else the data format. */
+ static FileFormat deleteFileFormat(Table table, FileFormat dataFormat) {
+ return FileFormat.fromString(
+ PropertyUtil.propertyAsString(
+ table.properties(), TableProperties.DELETE_DEFAULT_FILE_FORMAT, dataFormat.name()));
+ }
+}
diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcRecordCoderTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcRecordCoderTest.java
index a53b98a10aa9..b37c00e1d7a2 100644
--- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcRecordCoderTest.java
+++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcRecordCoderTest.java
@@ -82,10 +82,10 @@ public void encodePinnedWireMapping() throws Exception {
RowCoder.of(DATA_SCHEMA).encode(row, rowOnly);
int rowLen = rowOnly.toByteArray().length;
- assertPinnedKindCode(row, ValueKind.INSERT, rowLen, 0);
- assertPinnedKindCode(row, ValueKind.UPDATE_BEFORE, rowLen, 1);
- assertPinnedKindCode(row, ValueKind.UPDATE_AFTER, rowLen, 2);
- assertPinnedKindCode(row, ValueKind.DELETE, rowLen, 3);
+ assertPinnedKindCode(row, ValueKind.INSERT, rowLen, 1);
+ assertPinnedKindCode(row, ValueKind.UPDATE_BEFORE, rowLen, 2);
+ assertPinnedKindCode(row, ValueKind.UPDATE_AFTER, rowLen, 3);
+ assertPinnedKindCode(row, ValueKind.DELETE, rowLen, 4);
}
private static void assertPinnedKindCode(Row row, ValueKind kind, int rowLen, int expectedCode)
@@ -103,7 +103,7 @@ public void decodeRejectsUnknownKindCode() throws Exception {
ByteArrayOutputStream out = new ByteArrayOutputStream();
RowCoder.of(DATA_SCHEMA)
.encode(Row.withSchema(DATA_SCHEMA).addValues(1, "a", "x").build(), out);
- VarIntCoder.of().encode(4, out);
+ VarIntCoder.of().encode(5, out);
VarLongCoder.of().encode(1L, out);
CdcRecordCoder coder = CdcRecordCoder.of(DATA_SCHEMA);
diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcSinkTestUtils.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcSinkTestUtils.java
new file mode 100644
index 000000000000..9f67e45f0801
--- /dev/null
+++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcSinkTestUtils.java
@@ -0,0 +1,193 @@
+/*
+ * 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.beam.sdk.io.iceberg.cdc.sink;
+
+import java.io.File;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.apache.beam.sdk.io.iceberg.IcebergCatalogConfig;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.OutputBuilder;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.Row;
+import org.apache.beam.sdk.values.ValueKind;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableSet;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.CatalogUtil;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.RowDelta;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.SortOrder;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.hadoop.HadoopCatalog;
+import org.apache.iceberg.io.OutputFileFactory;
+import org.apache.iceberg.io.WriteResult;
+import org.apache.iceberg.types.Types;
+
+/**
+ * Shared test helpers for the {@code cdc/sink} suites. The TableCache and catalog caches are
+ * process-wide statics, so tests must use unique table names per test method.
+ */
+final class CdcSinkTestUtils {
+
+ private CdcSinkTestUtils() {}
+
+ /** An in-process {@link HadoopCatalog} rooted at {@code warehouseDir}. */
+ static Catalog hadoopCatalog(File warehouseDir) {
+ Configuration hadoopConf = new Configuration();
+ return new HadoopCatalog(hadoopConf, warehouseDir.getAbsolutePath());
+ }
+
+ /** An {@link IcebergCatalogConfig} resolving to the same warehouse as {@link #hadoopCatalog}. */
+ static IcebergCatalogConfig catalogConfig(File warehouseDir) {
+ return IcebergCatalogConfig.builder()
+ .setCatalogProperties(
+ ImmutableMap.of(
+ "type",
+ CatalogUtil.ICEBERG_CATALOG_TYPE_HADOOP,
+ "warehouse",
+ "file:" + warehouseDir.getAbsolutePath()))
+ .build();
+ }
+
+ /** Creates a table with the given identifier field ids format version. */
+ static Table createTable(
+ Catalog catalog,
+ TableIdentifier id,
+ Schema schema,
+ Set identifierFieldIds,
+ int formatVersion,
+ PartitionSpec spec) {
+ Schema schemaWithIds = new Schema(schema.columns(), identifierFieldIds);
+ Map props = ImmutableMap.of("format-version", String.valueOf(formatVersion));
+ return catalog.createTable(id, schemaWithIds, spec, props);
+ }
+
+ /**
+ * Creates the two fresh unpartitioned V2 routing targets {@code db.} and {@code
+ * db.}, both with columns {@code (id INT pk, dest STRING)}: the dynamic-destination
+ * fixture where the routing column is also a data column.
+ */
+ static void createDestTables(Catalog catalog, String tableA, String tableB) {
+ Schema destTableSchema =
+ new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "dest", Types.StringType.get()));
+ for (String name : ImmutableList.of(tableA, tableB)) {
+ createTable(
+ catalog,
+ TableIdentifier.of("db", name),
+ destTableSchema,
+ ImmutableSet.of(1),
+ 2,
+ PartitionSpec.unpartitioned());
+ }
+ }
+
+ /**
+ * Creates a table born WITH {@code sortOrder}, distinct from altering afterwards: such a table
+ * stores only sort order id 1, no id 0 (the id every sink equality delete carries).
+ */
+ static Table createSortedTable(
+ Catalog catalog,
+ TableIdentifier id,
+ Schema schema,
+ Set identifierFieldIds,
+ int formatVersion,
+ PartitionSpec spec,
+ SortOrder sortOrder) {
+ Schema schemaWithIds = new Schema(schema.columns(), identifierFieldIds);
+ return catalog
+ .buildTable(id, schemaWithIds)
+ .withPartitionSpec(spec)
+ .withSortOrder(sortOrder)
+ .withProperties(ImmutableMap.of("format-version", String.valueOf(formatVersion)))
+ .create();
+ }
+
+ /**
+ * A {@link RecordDeltaTaskWriter} through the production factory path: table-resolved formats,
+ * the current spec as the pinned spec.
+ */
+ static RecordDeltaTaskWriter deltaWriter(
+ Table table, Set equalityFieldIds, boolean upsert, long targetFileSizeBytes) {
+ FileFormat dataFormat = RecordDeltaTaskWriter.dataFileFormat(table);
+ FileFormat deleteFormat = RecordDeltaTaskWriter.deleteFileFormat(table, dataFormat);
+ return RecordDeltaTaskWriter.create(
+ table,
+ table.spec(),
+ equalityFieldIds,
+ upsert,
+ targetFileSizeBytes,
+ OutputFileFactory.builderFor(table, 1, 1).build(),
+ dataFormat,
+ deleteFormat);
+ }
+
+ /** An {@link DoFn.OutputReceiver} appending to {@code out}, for driving a DoFn directly. */
+ static DoFn.OutputReceiver collectInto(List out) {
+ return new DoFn.OutputReceiver() {
+ @Override
+ public OutputBuilder builder(T value) {
+ throw new UnsupportedOperationException("test receiver: use output(value)");
+ }
+
+ @Override
+ public void output(T value) {
+ out.add(value);
+ }
+ };
+ }
+
+ /** Commits a {@link WriteResult}'s data and delete files to the table as one row delta. */
+ static void commitRowDelta(Table table, WriteResult result) {
+ RowDelta rowDelta = table.newRowDelta();
+ Arrays.stream(result.dataFiles()).forEach(rowDelta::addRows);
+ Arrays.stream(result.deleteFiles()).forEach(rowDelta::addDeletes);
+ rowDelta.commit();
+ }
+
+ /** Attaches each element's {@link ValueKind} to its {@link Row}: the sink's input contract. */
+ static PCollection withKinds(PCollection> tagged) {
+ return tagged.apply(kindsFn());
+ }
+
+ /** {@link #withKinds(PCollection)} with an explicit step name, for multi-application tests. */
+ static PCollection withKinds(String name, PCollection> tagged) {
+ return tagged.apply(name, kindsFn());
+ }
+
+ private static ParDo.SingleOutput, Row> kindsFn() {
+ return ParDo.of(
+ new DoFn, Row>() {
+ @ProcessElement
+ public void process(@Element KV e, OutputReceiver out) {
+ out.builder(e.getValue()).setValueKind(e.getKey()).output();
+ }
+ });
+ }
+}
diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcWriteConfigTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcWriteConfigTest.java
new file mode 100644
index 000000000000..fb8ab133b49a
--- /dev/null
+++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcWriteConfigTest.java
@@ -0,0 +1,226 @@
+/*
+ * 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.beam.sdk.io.iceberg.cdc.sink;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.nullValue;
+import static org.junit.Assert.assertThrows;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.beam.sdk.util.SerializableUtils;
+import org.apache.beam.sdk.values.ValueKind;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/** Tests for {@link CdcWriteConfig}. */
+@RunWith(JUnit4.class)
+public class CdcWriteConfigTest {
+
+ private static final String SINK_ID = "test-sink-id";
+
+ @Test
+ public void builderAppliesDefaults() {
+ CdcWriteConfig config = CdcWriteConfig.builder().setSinkId(SINK_ID).build();
+
+ assertThat(
+ config.getSequenceNumberColumn(), equalTo(CdcWriteConfig.DEFAULT_SEQUENCE_NUMBER_COLUMN));
+ assertThat(config.getNumShards(), equalTo(CdcWriteConfig.DEFAULT_NUM_SHARDS));
+ // Unset shards_per_partition resolves to num_shards: the resolved int carries no cap.
+ assertThat(config.getShardsPerPartition(), equalTo(CdcWriteConfig.DEFAULT_NUM_SHARDS));
+ assertThat(config.getSorterMemoryMB(), equalTo(CdcWriteConfig.DEFAULT_SORTER_MEMORY_MB));
+ assertThat(config.getUpsert(), equalTo(false));
+ assertThat(config.getTokenHeartbeatMillis(), nullValue());
+ assertThat(config.getErrorHandling(), equalTo(false));
+ assertThat(config.getEqualityColumns(), nullValue());
+ assertThat(config.getChangeTypeColumn(), nullValue());
+ assertThat(config.getChangeTypeMap(), nullValue());
+ assertThat(config.getSnapshotProperties(), nullValue());
+ assertThat(config.getSinkId(), equalTo(SINK_ID));
+ }
+
+ /** Both extremes of a legal config pass: everything optional unset, and everything set. */
+ @Test
+ public void validatePassesForDefaultsOnlyAndFullyPopulatedConfigs() {
+ CdcWriteConfig.builder().setSinkId(SINK_ID).build().validate();
+ fullyPopulatedBuilder().build().validate();
+ }
+
+ /** The config carries the resolved int as set; validation accepts the whole legal range. */
+ @Test
+ public void builderCarriesExplicitShardsPerPartition() {
+ CdcWriteConfig config =
+ CdcWriteConfig.builder()
+ .setSinkId(SINK_ID)
+ .setNumShards(16)
+ .setShardsPerPartition(4)
+ .build();
+
+ assertThat(config.getShardsPerPartition(), equalTo(4));
+ config.validate();
+ }
+
+ /** The whole rejection matrix: every field {@code validate()} bounds, one facet each. */
+ @Test
+ public void validateRejectsEachInvalidConfigField() {
+ // facet: num_shards below one.
+ CdcWriteConfig zeroShards = CdcWriteConfig.builder().setSinkId(SINK_ID).setNumShards(0).build();
+ IllegalArgumentException numShards =
+ assertThrows(IllegalArgumentException.class, () -> zeroShards.validate());
+ assertThat(numShards.getMessage(), containsString("num_shards"));
+
+ // facet: shards_per_partition below one.
+ CdcWriteConfig sppZeroConfig =
+ CdcWriteConfig.builder().setSinkId(SINK_ID).setShardsPerPartition(0).build();
+ IllegalArgumentException sppZero =
+ assertThrows(IllegalArgumentException.class, () -> sppZeroConfig.validate());
+ assertThat(sppZero.getMessage(), containsString("shards_per_partition"));
+ assertThat(sppZero.getMessage(), containsString("between 1 and num_shards"));
+
+ // facet: shards_per_partition above num_shards.
+ CdcWriteConfig sppAboveConfig =
+ CdcWriteConfig.builder()
+ .setSinkId(SINK_ID)
+ .setNumShards(16)
+ .setShardsPerPartition(32)
+ .build();
+ IllegalArgumentException sppAbove =
+ assertThrows(IllegalArgumentException.class, () -> sppAboveConfig.validate());
+ assertThat(sppAbove.getMessage(), containsString("shards_per_partition"));
+ assertThat(sppAbove.getMessage(), containsString("32"));
+ assertThat(sppAbove.getMessage(), containsString("16"));
+
+ // facet: sorter_memory_mb below one.
+ CdcWriteConfig sorterConfig =
+ CdcWriteConfig.builder().setSinkId(SINK_ID).setSorterMemoryMB(0).build();
+ IllegalArgumentException sorter =
+ assertThrows(IllegalArgumentException.class, () -> sorterConfig.validate());
+ assertThat(sorter.getMessage(), containsString("sorter_memory_mb"));
+
+ // facet: explicitly empty equality_columns.
+ CdcWriteConfig emptyEqConfig =
+ CdcWriteConfig.builder()
+ .setSinkId(SINK_ID)
+ .setEqualityColumns(Collections.emptyList())
+ .build();
+ IllegalArgumentException emptyEq =
+ assertThrows(IllegalArgumentException.class, () -> emptyEqConfig.validate());
+ assertThat(emptyEq.getMessage(), containsString("equality_columns"));
+
+ // facet: change-type column colliding with the sequence-number column.
+ CdcWriteConfig collidingConfig =
+ CdcWriteConfig.builder()
+ .setSinkId(SINK_ID)
+ .setSequenceNumberColumn("seq")
+ .setChangeTypeColumn("seq")
+ .build();
+ IllegalArgumentException colliding =
+ assertThrows(IllegalArgumentException.class, () -> collidingConfig.validate());
+ assertThat(colliding.getMessage(), containsString("sequence_number_column"));
+ assertThat(colliding.getMessage(), containsString("change_type_column"));
+
+ // facet: change_type_map without a change_type_column.
+ Map orphanMap = new HashMap<>();
+ orphanMap.put("c", "INSERT");
+ CdcWriteConfig orphanMapConfig =
+ CdcWriteConfig.builder().setSinkId(SINK_ID).setChangeTypeMap(orphanMap).build();
+ IllegalArgumentException orphan =
+ assertThrows(IllegalArgumentException.class, () -> orphanMapConfig.validate());
+ assertThat(orphan.getMessage(), containsString("change_type_map"));
+ assertThat(orphan.getMessage(), containsString("change_type_column"));
+
+ // facet: a change_type_map value that is not a ValueKind name lists the legal names.
+ Map typoMap = new HashMap<>();
+ typoMap.put("c", "INSSERT"); // typo: not a ValueKind name
+ CdcWriteConfig typoMapConfig =
+ CdcWriteConfig.builder()
+ .setSinkId(SINK_ID)
+ .setChangeTypeColumn("op")
+ .setChangeTypeMap(typoMap)
+ .build();
+ IllegalArgumentException typo =
+ assertThrows(IllegalArgumentException.class, () -> typoMapConfig.validate());
+ assertThat(typo.getMessage(), containsString("change_type_map"));
+ assertThat(typo.getMessage(), containsString("INSSERT"));
+ assertThat(typo.getMessage(), containsString("INSERT"));
+ assertThat(typo.getMessage(), containsString("UPDATE_BEFORE"));
+ assertThat(typo.getMessage(), containsString("UPDATE_AFTER"));
+ assertThat(typo.getMessage(), containsString("DELETE"));
+
+ // facet: non-positive token heartbeat, named by the real option names.
+ CdcWriteConfig heartbeatConfig =
+ CdcWriteConfig.builder().setSinkId(SINK_ID).setTokenHeartbeatMillis(0L).build();
+ IllegalArgumentException heartbeat =
+ assertThrows(IllegalArgumentException.class, () -> heartbeatConfig.validate());
+ assertThat(heartbeat.getMessage(), containsString("withTokenHeartbeat"));
+ assertThat(heartbeat.getMessage(), containsString("token_heartbeat_seconds"));
+
+ // facet: reserved beam.cdc. snapshot-property prefix.
+ Map reserved = new HashMap<>();
+ reserved.put("beam.cdc.sink-id", "x");
+ CdcWriteConfig reservedConfig =
+ CdcWriteConfig.builder().setSinkId(SINK_ID).setSnapshotProperties(reserved).build();
+ IllegalArgumentException reservedThrown =
+ assertThrows(IllegalArgumentException.class, () -> reservedConfig.validate());
+ assertThat(reservedThrown.getMessage(), containsString("snapshot_properties"));
+ assertThat(reservedThrown.getMessage(), containsString("beam.cdc."));
+ }
+
+ @Test
+ public void configIsJavaSerializable() {
+ CdcWriteConfig config = fullyPopulatedBuilder().build();
+
+ CdcWriteConfig deserialized = SerializableUtils.ensureSerializable(config);
+
+ assertThat(deserialized, equalTo(config));
+ }
+
+ private static CdcWriteConfig.Builder fullyPopulatedBuilder() {
+ Map snapshotProperties = new HashMap<>();
+ snapshotProperties.put("k", "v");
+
+ return CdcWriteConfig.builder()
+ .setSinkId(SINK_ID)
+ .setEqualityColumns(Arrays.asList("id", "region"))
+ .setSequenceNumberColumn("my_seq")
+ .setChangeTypeColumn("op")
+ .setChangeTypeMap(legalChangeTypeMap())
+ .setNumShards(8)
+ .setShardsPerPartition(1)
+ .setSorterMemoryMB(200)
+ .setUpsert(true)
+ .setTokenHeartbeatMillis(60000L)
+ .setSnapshotProperties(snapshotProperties)
+ .setErrorHandling(true);
+ }
+
+ /** A {@link CdcWriteConfig#getChangeTypeMap()} value naming every {@link ValueKind} constant. */
+ private static Map legalChangeTypeMap() {
+ Map changeTypeMap = new HashMap<>();
+ changeTypeMap.put("c", "INSERT");
+ changeTypeMap.put("u", "UPDATE_AFTER");
+ changeTypeMap.put("b", "UPDATE_BEFORE");
+ changeTypeMap.put("d", "DELETE");
+ return changeTypeMap;
+ }
+}
diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/RecordDeltaTaskWriterTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/RecordDeltaTaskWriterTest.java
new file mode 100644
index 000000000000..dc5b8371e873
--- /dev/null
+++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/RecordDeltaTaskWriterTest.java
@@ -0,0 +1,1079 @@
+/*
+ * 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.beam.sdk.io.iceberg.cdc.sink;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.arrayWithSize;
+import static org.hamcrest.Matchers.contains;
+import static org.hamcrest.Matchers.containsInAnyOrder;
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.empty;
+import static org.hamcrest.Matchers.emptyArray;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.hasSize;
+import static org.hamcrest.Matchers.not;
+import static org.hamcrest.Matchers.nullValue;
+import static org.junit.Assert.assertThrows;
+
+import java.io.File;
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.time.LocalDate;
+import java.time.OffsetDateTime;
+import java.time.ZoneOffset;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import org.apache.beam.sdk.io.iceberg.SerializableDataFile;
+import org.apache.beam.sdk.values.ValueKind;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Throwables;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableSet;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Iterables;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.primitives.Ints;
+import org.apache.iceberg.ContentFile;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileContent;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.SortOrder;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.data.IcebergGenerics;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.data.parquet.GenericParquetReaders;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.io.OutputFileFactory;
+import org.apache.iceberg.io.PositionOutputStream;
+import org.apache.iceberg.io.WriteResult;
+import org.apache.iceberg.parquet.Parquet;
+import org.apache.iceberg.transforms.Transforms;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.SerializableFunction;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/**
+ * Tests for {@link RecordDeltaTaskWriter}'s streaming collapse. Each case feeds one sorted group
+ * (same-key records contiguous, in (seq, kind) order) and asserts the flush truth table at
+ * file-content level: the produced Parquet files are read back row by row, and where a table state
+ * matters the result is committed with {@link Table#newRowDelta()} and read via {@link
+ * IcebergGenerics}.
+ *
+ * The writer emits at most one equality delete and one data row per key per group, and never
+ * writes position deletes or deletion vectors; same-window churn that cancels out reaches no file
+ * at all.
+ */
+@RunWith(JUnit4.class)
+public class RecordDeltaTaskWriterTest {
+
+ @Rule public transient TemporaryFolder tmp = new TemporaryFolder();
+
+ private static final Schema SCHEMA =
+ new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "name", Types.StringType.get()),
+ Types.NestedField.optional(3, "data", Types.StringType.get()));
+
+ private static final long TARGET_FILE_SIZE = 512L * 1024 * 1024;
+
+ private File warehouseDir;
+ private Catalog catalog;
+
+ @Before
+ public void setUp() throws Exception {
+ warehouseDir = tmp.newFolder("warehouse");
+ catalog = CdcSinkTestUtils.hadoopCatalog(warehouseDir);
+ }
+
+ /** Creates the canonical unpartitioned V2 table with {@code id} as the identifier/PK. */
+ private Table v2Table() {
+ return CdcSinkTestUtils.createTable(
+ catalog,
+ TableIdentifier.of("db", "t" + System.nanoTime()),
+ SCHEMA,
+ ImmutableSet.of(1),
+ 2,
+ PartitionSpec.unpartitioned());
+ }
+
+ /** Creates the canonical unpartitioned V3 table with {@code id} as the identifier/PK. */
+ private Table v3Table() {
+ return CdcSinkTestUtils.createTable(
+ catalog,
+ TableIdentifier.of("db", "v3_" + System.nanoTime()),
+ SCHEMA,
+ ImmutableSet.of(1),
+ 3,
+ PartitionSpec.unpartitioned());
+ }
+
+ /** Creates a V2 table partitioned by {@code bucket(id, 2)} (the PK). */
+ private Table v2BucketPartitionedTable() {
+ PartitionSpec spec = PartitionSpec.builderFor(SCHEMA).bucket("id", 2).build();
+ return CdcSinkTestUtils.createTable(
+ catalog,
+ TableIdentifier.of("db", "p" + System.nanoTime()),
+ SCHEMA,
+ ImmutableSet.of(1),
+ 2,
+ spec);
+ }
+
+ /** Creates a V2 table partitioned by {@code identity(name)}, a NON-key column (PK stays id). */
+ private Table v2NonKeyPartitionedTable() {
+ PartitionSpec spec = PartitionSpec.builderFor(SCHEMA).identity("name").build();
+ return CdcSinkTestUtils.createTable(
+ catalog,
+ TableIdentifier.of("db", "nk" + System.nanoTime()),
+ SCHEMA,
+ ImmutableSet.of(1),
+ 2,
+ spec);
+ }
+
+ /** A V2 table BORN with a sort order on a non-key column: its only sort order id is 1, not 0. */
+ private Table v2SortedTable() {
+ return CdcSinkTestUtils.createSortedTable(
+ catalog,
+ TableIdentifier.of("db", "s" + System.nanoTime()),
+ SCHEMA,
+ ImmutableSet.of(1),
+ 2,
+ PartitionSpec.unpartitioned(),
+ SortOrder.builderFor(SCHEMA).asc("name").build());
+ }
+
+ /** A production-path writer with this suite's PK ({@code id}) and target file size. */
+ private static RecordDeltaTaskWriter writer(Table t, boolean upsert) {
+ return CdcSinkTestUtils.deltaWriter(t, ImmutableSet.of(1), upsert, TARGET_FILE_SIZE);
+ }
+
+ private static Record rec(Table t, int id, String name, String data) {
+ GenericRecord r = GenericRecord.create(t.schema());
+ r.setField("id", id);
+ r.setField("name", name);
+ r.setField("data", data);
+ return r;
+ }
+
+ /** Writes one change: the sort key's pk prefix carries the record's encoded {@code id}. */
+ private static void write(RecordDeltaTaskWriter w, Record rec, long seq, ValueKind kind) {
+ byte[] pk = Ints.toByteArray((Integer) rec.getField("id"));
+ w.write(CdcSortKey.encode(pk, seq, kind), rec, kind);
+ }
+
+ /** Reads the table's current rows as sorted {@code "id:name:data"} strings. */
+ private static List readRows(Table t) throws IOException {
+ List rows = new ArrayList<>();
+ try (CloseableIterable reader = IcebergGenerics.read(t).build()) {
+ for (Record r : reader) {
+ rows.add(r.getField("id") + ":" + r.getField("name") + ":" + r.getField("data"));
+ }
+ }
+ Collections.sort(rows);
+ return rows;
+ }
+
+ /** Reads a Parquet data/delete file's rows with the given projection (matched by field id). */
+ private static List readParquetRows(Table t, String location, Schema projection)
+ throws IOException {
+ try (CloseableIterable reader =
+ Parquet.read(t.io().newInputFile(location))
+ .project(projection)
+ .createReaderFunc(
+ fileSchema -> GenericParquetReaders.buildReader(projection, fileSchema))
+ .build()) {
+ return ImmutableList.copyOf(reader);
+ }
+ }
+
+ /** A file's rows as {@code "id:name:data"} strings against the full table schema. */
+ private static List readFileRows(Table t, String location) throws IOException {
+ return readParquetRows(t, location, t.schema()).stream()
+ .map(r -> r.getField("id") + ":" + r.getField("name") + ":" + r.getField("data"))
+ .collect(Collectors.toList());
+ }
+
+ /** The single delete file, asserted to be a PK-only equality delete over the given ids. */
+ private static void assertEqualityDeleteOfIds(Table t, WriteResult r, Integer... ids)
+ throws IOException {
+ assertThat(r.deleteFiles(), arrayWithSize(1));
+ DeleteFile del = r.deleteFiles()[0];
+ assertThat(del.content(), equalTo(FileContent.EQUALITY_DELETES));
+ assertThat(del.equalityFieldIds(), contains(1));
+
+ // PK-only rows: projecting the full schema over the delete file yields nulls for name/data;
+ // a full-row equality delete would read the data columns back.
+ List deleteRows = readParquetRows(t, del.location(), t.schema());
+ List deletedIds = new ArrayList<>();
+ for (Record row : deleteRows) {
+ deletedIds.add((Integer) row.getField("id"));
+ assertThat(row.getField("name"), nullValue());
+ assertThat(row.getField("data"), nullValue());
+ }
+ assertThat(deletedIds, containsInAnyOrder(ids));
+ }
+
+ // ---------------------------------------------------------------------------------------------
+ // Flush truth table, non-upsert
+ // ---------------------------------------------------------------------------------------------
+
+ // [I] -> row only.
+ @Test
+ public void insertWritesDataFileOnlyAndReadsBack() throws Exception {
+ Table t = v2Table();
+ RecordDeltaTaskWriter w = writer(t, false);
+ write(w, rec(t, 1, "a", "x"), 1L, ValueKind.INSERT);
+ write(w, rec(t, 2, "b", "y"), 1L, ValueKind.INSERT);
+ WriteResult r = w.complete();
+
+ assertThat(r.dataFiles(), arrayWithSize(1));
+ assertThat(r.deleteFiles(), emptyArray());
+ assertThat(readFileRows(t, r.dataFiles()[0].location()), contains("1:a:x", "2:b:y"));
+
+ CdcSinkTestUtils.commitRowDelta(t, r);
+ assertThat(readRows(t), contains("1:a:x", "2:b:y"));
+ }
+
+ // [I, D] -> nothing: the key was born and died this window, so no file is written at all.
+ @Test
+ public void insertThenDeleteEmitsNothing() throws Exception {
+ Table t = v2Table();
+ RecordDeltaTaskWriter w = writer(t, false);
+ write(w, rec(t, 1, "a", "x"), 1L, ValueKind.INSERT);
+ write(w, rec(t, 1, "a", "x"), 2L, ValueKind.DELETE);
+ WriteResult r = w.complete();
+
+ assertThat(r.dataFiles(), emptyArray());
+ assertThat(r.deleteFiles(), emptyArray());
+ assertThat(dataFilesUnder(warehouseDir), empty());
+ }
+
+ // [I, UB, UA] -> row only: born this window, so its churn needs no delete.
+ @Test
+ public void insertUpdatedInWindowWritesFinalRowOnly() throws Exception {
+ Table t = v2Table();
+ RecordDeltaTaskWriter w = writer(t, false);
+ write(w, rec(t, 1, "a", "x"), 1L, ValueKind.INSERT);
+ write(w, rec(t, 1, "a", "x"), 2L, ValueKind.UPDATE_BEFORE);
+ write(w, rec(t, 1, "a2", "x2"), 2L, ValueKind.UPDATE_AFTER);
+ WriteResult r = w.complete();
+
+ assertThat(r.dataFiles(), arrayWithSize(1));
+ assertThat(r.deleteFiles(), emptyArray());
+ assertThat(readFileRows(t, r.dataFiles()[0].location()), contains("1:a2:x2"));
+
+ CdcSinkTestUtils.commitRowDelta(t, r);
+ assertThat(readRows(t), contains("1:a2:x2"));
+ }
+
+ // [UB, UA] -> delete + row, reaching a row committed by an earlier writer.
+ @Test
+ public void updatePairWritesEqualityDeleteAndFinalRow() throws Exception {
+ Table t = v2Table();
+ RecordDeltaTaskWriter a = writer(t, false);
+ write(a, rec(t, 1, "old", "x"), 1L, ValueKind.INSERT);
+ CdcSinkTestUtils.commitRowDelta(t, a.complete());
+
+ RecordDeltaTaskWriter b = writer(t, false);
+ write(b, rec(t, 1, "old", "x"), 2L, ValueKind.UPDATE_BEFORE);
+ write(b, rec(t, 1, "new", "y"), 2L, ValueKind.UPDATE_AFTER);
+ WriteResult r = b.complete();
+
+ assertThat(r.dataFiles(), arrayWithSize(1));
+ assertThat(readFileRows(t, r.dataFiles()[0].location()), contains("1:new:y"));
+ assertEqualityDeleteOfIds(t, r, 1);
+
+ CdcSinkTestUtils.commitRowDelta(t, r);
+ assertThat(readRows(t), contains("1:new:y"));
+ }
+
+ // [D, I] with an earlier committed row -> delete + row: the delete survives the block ending in
+ // INSERT (a reinsert must still remove the committed image).
+ @Test
+ public void deleteThenReinsertReplacesTheCommittedRow() throws Exception {
+ Table t = v2Table();
+ RecordDeltaTaskWriter a = writer(t, false);
+ write(a, rec(t, 1, "old", "x"), 1L, ValueKind.INSERT);
+ CdcSinkTestUtils.commitRowDelta(t, a.complete());
+
+ RecordDeltaTaskWriter b = writer(t, false);
+ write(b, rec(t, 1, "old", "x"), 2L, ValueKind.DELETE);
+ write(b, rec(t, 1, "new", "y"), 3L, ValueKind.INSERT);
+ WriteResult r = b.complete();
+
+ assertThat(r.dataFiles(), arrayWithSize(1));
+ assertThat(readFileRows(t, r.dataFiles()[0].location()), contains("1:new:y"));
+ assertEqualityDeleteOfIds(t, r, 1);
+
+ CdcSinkTestUtils.commitRowDelta(t, r);
+ assertThat(readRows(t), contains("1:new:y"));
+ }
+
+ // [D] -> delete only, removing a row committed by an earlier writer.
+ @Test
+ public void deleteWritesPkOnlyEqualityDeleteOnly() throws Exception {
+ Table t = v2Table();
+ RecordDeltaTaskWriter a = writer(t, false);
+ write(a, rec(t, 1, "a", "x"), 1L, ValueKind.INSERT);
+ CdcSinkTestUtils.commitRowDelta(t, a.complete());
+
+ RecordDeltaTaskWriter b = writer(t, false);
+ write(b, rec(t, 1, "a", "x"), 2L, ValueKind.DELETE);
+ WriteResult r = b.complete();
+
+ assertThat(r.dataFiles(), emptyArray());
+ assertEqualityDeleteOfIds(t, r, 1);
+
+ CdcSinkTestUtils.commitRowDelta(t, r);
+ assertThat(readRows(t), empty());
+ }
+
+ // [UA] -> row only: bare-UA parity, a lone after-image writes without deleting.
+ @Test
+ public void bareUpdateAfterWritesRowWithoutDelete() throws Exception {
+ Table t = v2Table();
+ RecordDeltaTaskWriter w = writer(t, false);
+ write(w, rec(t, 1, "a2", "x2"), 2L, ValueKind.UPDATE_AFTER);
+ WriteResult r = w.complete();
+
+ assertThat(r.dataFiles(), arrayWithSize(1));
+ assertThat(r.deleteFiles(), emptyArray());
+ assertThat(readFileRows(t, r.dataFiles()[0].location()), contains("1:a2:x2"));
+ w.abort();
+ }
+
+ // [UA, UB, UA] -> delete + row: the opening UA fails the sawUbOrDelete arm only until the UB.
+ @Test
+ public void updateAfterChurnEndingInUpdateAfterWritesDeleteAndRow() throws Exception {
+ Table t = v2Table();
+ RecordDeltaTaskWriter w = writer(t, false);
+ write(w, rec(t, 1, "a2", "x2"), 2L, ValueKind.UPDATE_AFTER);
+ write(w, rec(t, 1, "a2", "x2"), 3L, ValueKind.UPDATE_BEFORE);
+ write(w, rec(t, 1, "a3", "x3"), 3L, ValueKind.UPDATE_AFTER);
+ WriteResult r = w.complete();
+
+ assertThat(r.dataFiles(), arrayWithSize(1));
+ assertThat(readFileRows(t, r.dataFiles()[0].location()), contains("1:a3:x3"));
+ assertEqualityDeleteOfIds(t, r, 1);
+ w.abort();
+ }
+
+ // [I, I] -> row only, the LAST image: a duplicate insert supersedes the first in the writer.
+ @Test
+ public void duplicateInsertKeepsLastImage() throws Exception {
+ Table t = v2Table();
+ RecordDeltaTaskWriter w = writer(t, false);
+ write(w, rec(t, 1, "a", "x"), 1L, ValueKind.INSERT);
+ write(w, rec(t, 1, "b", "y"), 2L, ValueKind.INSERT);
+ WriteResult r = w.complete();
+
+ assertThat(r.dataFiles(), arrayWithSize(1));
+ assertThat(r.deleteFiles(), emptyArray());
+ assertThat(readFileRows(t, r.dataFiles()[0].location()), contains("1:b:y"));
+
+ CdcSinkTestUtils.commitRowDelta(t, r);
+ assertThat(readRows(t), contains("1:b:y"));
+ }
+
+ // ---------------------------------------------------------------------------------------------
+ // Flush truth table, upsert (every block deletes first; UPDATE_BEFOREs are dropped upstream)
+ // ---------------------------------------------------------------------------------------------
+
+ // upsert [I] -> delete + row, replacing a previously committed image of the key.
+ @Test
+ public void upsertInsertWritesDeleteAndRow() throws Exception {
+ Table t = v2Table();
+ RecordDeltaTaskWriter a = writer(t, false);
+ write(a, rec(t, 1, "a", "x"), 1L, ValueKind.INSERT);
+ CdcSinkTestUtils.commitRowDelta(t, a.complete());
+
+ RecordDeltaTaskWriter b = writer(t, true /* upsert */);
+ write(b, rec(t, 1, "b", "y"), 2L, ValueKind.INSERT);
+ WriteResult r = b.complete();
+
+ assertThat(r.dataFiles(), arrayWithSize(1));
+ assertEqualityDeleteOfIds(t, r, 1);
+
+ CdcSinkTestUtils.commitRowDelta(t, r);
+ assertThat(readRows(t), contains("1:b:y"));
+ }
+
+ // upsert [D] -> delete only.
+ @Test
+ public void upsertDeleteWritesDeleteOnly() throws Exception {
+ Table t = v2Table();
+ RecordDeltaTaskWriter w = writer(t, true /* upsert */);
+ write(w, rec(t, 1, "a", "x"), 2L, ValueKind.DELETE);
+ WriteResult r = w.complete();
+
+ assertThat(r.dataFiles(), emptyArray());
+ assertEqualityDeleteOfIds(t, r, 1);
+ w.abort();
+ }
+
+ // upsert [UA] -> delete + row.
+ @Test
+ public void upsertUpdateAfterWritesDeleteAndRow() throws Exception {
+ Table t = v2Table();
+ RecordDeltaTaskWriter w = writer(t, true /* upsert */);
+ write(w, rec(t, 1, "a2", "x2"), 2L, ValueKind.UPDATE_AFTER);
+ WriteResult r = w.complete();
+
+ assertThat(r.dataFiles(), arrayWithSize(1));
+ assertThat(readFileRows(t, r.dataFiles()[0].location()), contains("1:a2:x2"));
+ assertEqualityDeleteOfIds(t, r, 1);
+ w.abort();
+ }
+
+ // ---------------------------------------------------------------------------------------------
+ // Multi-key groups and partition fanout
+ // ---------------------------------------------------------------------------------------------
+
+ // A multi-key group flushes per block: dead key omitted, update pair collapsed, insert written.
+ @Test
+ public void multiKeyGroupFlushesPerBlock() throws Exception {
+ Table t = v2Table();
+ RecordDeltaTaskWriter a = writer(t, false);
+ write(a, rec(t, 2, "old", "o"), 1L, ValueKind.INSERT);
+ CdcSinkTestUtils.commitRowDelta(t, a.complete());
+
+ RecordDeltaTaskWriter b = writer(t, false);
+ write(b, rec(t, 1, "a", "x"), 2L, ValueKind.INSERT);
+ write(b, rec(t, 1, "a", "x"), 3L, ValueKind.DELETE);
+ write(b, rec(t, 2, "old", "o"), 2L, ValueKind.UPDATE_BEFORE);
+ write(b, rec(t, 2, "new", "n"), 2L, ValueKind.UPDATE_AFTER);
+ write(b, rec(t, 3, "c", "z"), 2L, ValueKind.INSERT);
+ WriteResult r = b.complete();
+
+ assertThat(r.dataFiles(), arrayWithSize(1));
+ assertThat(readFileRows(t, r.dataFiles()[0].location()), contains("2:new:n", "3:c:z"));
+ assertEqualityDeleteOfIds(t, r, 2);
+
+ CdcSinkTestUtils.commitRowDelta(t, r);
+ assertThat(readRows(t), contains("2:new:n", "3:c:z"));
+ }
+
+ /**
+ * Partitioned fanout: the sort is by PK, so partitions interleave (bucket A, B, then A again),
+ * and each block's data row and equality delete must land in the block's own partition. The
+ * return to bucket A pins fanout: a clustered (one-open-partition) writer would refuse it.
+ */
+ @Test
+ public void partitionedFanoutRoutesRowAndDeleteToTheBlockPartition() throws Exception {
+ Table t = v2BucketPartitionedTable();
+ SerializableFunction bucketOf =
+ Transforms.bucket(2).bind(Types.IntegerType.get());
+ // Four ascending ids whose buckets go A, B, A, A.
+ int firstA = nextIdInBucket(bucketOf, 1, 0);
+ int midB = nextIdInBucket(bucketOf, firstA + 1, 1);
+ int deadA = nextIdInBucket(bucketOf, midB + 1, 0);
+ int lastA = nextIdInBucket(bucketOf, deadA + 1, 0);
+
+ RecordDeltaTaskWriter a = writer(t, false);
+ write(a, rec(t, midB, "old", "o"), 1L, ValueKind.INSERT);
+ write(a, rec(t, deadA, "gone", "g"), 1L, ValueKind.INSERT);
+ CdcSinkTestUtils.commitRowDelta(t, a.complete());
+
+ RecordDeltaTaskWriter b = writer(t, false);
+ write(b, rec(t, firstA, "a", "x"), 2L, ValueKind.INSERT);
+ write(b, rec(t, midB, "old", "o"), 2L, ValueKind.UPDATE_BEFORE);
+ write(b, rec(t, midB, "new", "n"), 2L, ValueKind.UPDATE_AFTER);
+ write(b, rec(t, deadA, "gone", "g"), 2L, ValueKind.DELETE);
+ write(b, rec(t, lastA, "d", "w"), 2L, ValueKind.INSERT);
+ WriteResult r = b.complete();
+
+ // One data file per touched bucket; bucket A's holds both of its blocks' rows.
+ assertThat(r.dataFiles(), arrayWithSize(2));
+ for (DataFile file : r.dataFiles()) {
+ Integer bucket = file.partition().get(0, Integer.class);
+ List rows = readFileRows(t, file.location());
+ if (bucket == 0) {
+ assertThat(rows, contains(firstA + ":a:x", lastA + ":d:w"));
+ } else {
+ assertThat(rows, contains(midB + ":new:n"));
+ }
+ }
+
+ // One equality delete per touched bucket, in the bucket of the key it removes.
+ assertThat(r.deleteFiles(), arrayWithSize(2));
+ for (DeleteFile file : r.deleteFiles()) {
+ assertThat(file.content(), equalTo(FileContent.EQUALITY_DELETES));
+ Integer bucket = file.partition().get(0, Integer.class);
+ List keys = readParquetRows(t, file.location(), t.schema());
+ assertThat(keys, hasSize(1));
+ int deletedId = (Integer) keys.get(0).getField("id");
+ assertThat(deletedId, equalTo(bucket == 1 ? midB : deadA));
+ }
+
+ CdcSinkTestUtils.commitRowDelta(t, r);
+ assertThat(readRows(t), containsInAnyOrder(firstA + ":a:x", midB + ":new:n", lastA + ":d:w"));
+ }
+
+ /** The smallest id at or above {@code from} whose {@code bucketOf} value is {@code bucket}. */
+ private static int nextIdInBucket(
+ SerializableFunction bucketOf, int from, int bucket) {
+ int id = from;
+ while (bucketOf.apply(id) != bucket) {
+ id++;
+ }
+ return id;
+ }
+
+ // ---------------------------------------------------------------------------------------------
+ // Non-key partitioning: the delete routes by the block's OPENING record
+ // ---------------------------------------------------------------------------------------------
+
+ /** The single {@code identity(name)} partition value of a data or delete file. */
+ private static String partitionOf(ContentFile file) {
+ return file.partition().get(0, String.class);
+ }
+
+ // [UB(p1), UA(p2)]: the delete lands in the OLD partition, the row in the new one. Routing the
+ // delete by the latest record would leave the p1 row alive forever.
+ @Test
+ public void movedRowDeletesFromOldPartitionAndWritesToNew() throws Exception {
+ Table t = v2NonKeyPartitionedTable();
+ RecordDeltaTaskWriter a = writer(t, false);
+ write(a, rec(t, 1, "p1", "x"), 1L, ValueKind.INSERT);
+ CdcSinkTestUtils.commitRowDelta(t, a.complete());
+
+ RecordDeltaTaskWriter b = writer(t, false);
+ write(b, rec(t, 1, "p1", "x"), 2L, ValueKind.UPDATE_BEFORE);
+ write(b, rec(t, 1, "p2", "y"), 2L, ValueKind.UPDATE_AFTER);
+ WriteResult r = b.complete();
+
+ assertEqualityDeleteOfIds(t, r, 1);
+ assertThat(partitionOf(r.deleteFiles()[0]), equalTo("p1"));
+ assertThat(r.dataFiles(), arrayWithSize(1));
+ assertThat(partitionOf(r.dataFiles()[0]), equalTo("p2"));
+ assertThat(readFileRows(t, r.dataFiles()[0].location()), contains("1:p2:y"));
+
+ CdcSinkTestUtils.commitRowDelta(t, r);
+ assertThat(readRows(t), contains("1:p2:y"));
+ }
+
+ // [UB(p1), UA(p2), UB(p2), UA(p3)]: still one delete, at the OPENING partition p1 (the only one
+ // holding a committed row), and one row at the final p3; the p2 stopover reaches no file.
+ @Test
+ public void multiMoveDeletesOnceAtTheOpeningPartition() throws Exception {
+ Table t = v2NonKeyPartitionedTable();
+ RecordDeltaTaskWriter a = writer(t, false);
+ write(a, rec(t, 1, "p1", "x"), 1L, ValueKind.INSERT);
+ CdcSinkTestUtils.commitRowDelta(t, a.complete());
+
+ RecordDeltaTaskWriter b = writer(t, false);
+ write(b, rec(t, 1, "p1", "x"), 2L, ValueKind.UPDATE_BEFORE);
+ write(b, rec(t, 1, "p2", "y"), 2L, ValueKind.UPDATE_AFTER);
+ write(b, rec(t, 1, "p2", "y"), 3L, ValueKind.UPDATE_BEFORE);
+ write(b, rec(t, 1, "p3", "z"), 3L, ValueKind.UPDATE_AFTER);
+ WriteResult r = b.complete();
+
+ assertEqualityDeleteOfIds(t, r, 1);
+ assertThat(partitionOf(r.deleteFiles()[0]), equalTo("p1"));
+ assertThat(r.dataFiles(), arrayWithSize(1));
+ assertThat(partitionOf(r.dataFiles()[0]), equalTo("p3"));
+
+ CdcSinkTestUtils.commitRowDelta(t, r);
+ assertThat(readRows(t), contains("1:p3:z"));
+ }
+
+ // [I(p1), UA(p2)]: born this window, so the move needs no delete; only the p2 row is written.
+ @Test
+ public void bornThisWindowMoveWritesFinalRowOnly() throws Exception {
+ Table t = v2NonKeyPartitionedTable();
+ RecordDeltaTaskWriter w = writer(t, false);
+ write(w, rec(t, 1, "p1", "x"), 1L, ValueKind.INSERT);
+ write(w, rec(t, 1, "p2", "y"), 2L, ValueKind.UPDATE_AFTER);
+ WriteResult r = w.complete();
+
+ assertThat(r.deleteFiles(), emptyArray());
+ assertThat(r.dataFiles(), arrayWithSize(1));
+ assertThat(partitionOf(r.dataFiles()[0]), equalTo("p2"));
+
+ CdcSinkTestUtils.commitRowDelta(t, r);
+ assertThat(readRows(t), contains("1:p2:y"));
+ }
+
+ // [D(p1)]: a DELETE carries the row's actual old values (the input contract), so its equality
+ // delete lands in the partition the row occupies.
+ @Test
+ public void deleteCarryingOldValuesLandsInTheRowsPartition() throws Exception {
+ Table t = v2NonKeyPartitionedTable();
+ RecordDeltaTaskWriter a = writer(t, false);
+ write(a, rec(t, 1, "p1", "x"), 1L, ValueKind.INSERT);
+ CdcSinkTestUtils.commitRowDelta(t, a.complete());
+
+ RecordDeltaTaskWriter b = writer(t, false);
+ write(b, rec(t, 1, "p1", "x"), 2L, ValueKind.DELETE);
+ WriteResult r = b.complete();
+
+ assertThat(r.dataFiles(), emptyArray());
+ assertEqualityDeleteOfIds(t, r, 1);
+ assertThat(partitionOf(r.deleteFiles()[0]), equalTo("p1"));
+
+ CdcSinkTestUtils.commitRowDelta(t, r);
+ assertThat(readRows(t), empty());
+ }
+
+ // A composite key: the equality delete carries both key columns, and only records agreeing on
+ // both collapse into one block.
+ @Test
+ public void compositeKeyProjectsEveryKeyColumnAndCollapsesPerKey() throws Exception {
+ Schema schema =
+ new Schema(
+ Types.NestedField.required(1, "tenant", Types.StringType.get()),
+ Types.NestedField.required(2, "id", Types.LongType.get()),
+ Types.NestedField.optional(3, "name", Types.StringType.get()),
+ Types.NestedField.optional(4, "data", Types.StringType.get()));
+ Table t = table(schema, ImmutableSet.of(1, 2));
+ RecordDeltaTaskWriter a =
+ CdcSinkTestUtils.deltaWriter(t, ImmutableSet.of(1, 2), false, TARGET_FILE_SIZE);
+ writeKeyed(a, record(t, "a", 1L, "old", "x"), 0L, ValueKind.INSERT, "a", 1L);
+ writeKeyed(a, record(t, "b", 1L, "c", "w"), 0L, ValueKind.INSERT, "b", 1L);
+ CdcSinkTestUtils.commitRowDelta(t, a.complete());
+
+ RecordDeltaTaskWriter w =
+ CdcSinkTestUtils.deltaWriter(t, ImmutableSet.of(1, 2), false, TARGET_FILE_SIZE);
+ writeKeyed(w, record(t, "a", 1L, "old", "x"), 1L, ValueKind.UPDATE_BEFORE, "a", 1L);
+ writeKeyed(w, record(t, "a", 1L, "new", "y"), 1L, ValueKind.UPDATE_AFTER, "a", 1L);
+ writeKeyed(w, record(t, "a", 2L, "b", "z"), 1L, ValueKind.INSERT, "a", 2L);
+ writeKeyed(w, record(t, "b", 1L, "c", "w"), 1L, ValueKind.DELETE, "b", 1L);
+ WriteResult r = w.complete();
+
+ assertThat(r.dataFiles(), arrayWithSize(1));
+ assertThat(rowStrings(t, r.dataFiles()[0].location()), contains("a:1:new:y", "a:2:b:z"));
+ DeleteFile del = singleEqualityDelete(r, 1, 2);
+ assertThat(rowStrings(t, del.location()), containsInAnyOrder("a:1:null:null", "b:1:null:null"));
+
+ CdcSinkTestUtils.commitRowDelta(t, r);
+ assertThat(readRows(t, schema), contains("a:1:new:y", "a:2:b:z"));
+ }
+
+ // Non-integer key types round-trip through the PK-only delete file and match on read.
+ @Test
+ public void keyTypesRoundTripThroughEqualityDeletes() throws Exception {
+ Schema schema =
+ new Schema(
+ Types.NestedField.required(1, "k_long", Types.LongType.get()),
+ Types.NestedField.required(2, "k_str", Types.StringType.get()),
+ Types.NestedField.required(3, "k_date", Types.DateType.get()),
+ Types.NestedField.required(4, "k_ts", Types.TimestampType.withZone()),
+ Types.NestedField.required(5, "k_dec", Types.DecimalType.of(10, 2)),
+ Types.NestedField.required(6, "k_uuid", Types.UUIDType.get()),
+ Types.NestedField.optional(7, "payload", Types.StringType.get()));
+ ImmutableSet keyIds = ImmutableSet.of(1, 2, 3, 4, 5, 6);
+ Object[] key = {
+ 7L,
+ "seven",
+ LocalDate.of(2026, 9, 3),
+ OffsetDateTime.of(2026, 9, 3, 12, 30, 0, 0, ZoneOffset.UTC),
+ new BigDecimal("12.34"),
+ UUID.randomUUID()
+ };
+ Table t = table(schema, keyIds);
+ RecordDeltaTaskWriter a = CdcSinkTestUtils.deltaWriter(t, keyIds, false, TARGET_FILE_SIZE);
+ writeKeyed(a, record(t, concat(key, "old")), 0L, ValueKind.INSERT, key);
+ CdcSinkTestUtils.commitRowDelta(t, a.complete());
+
+ RecordDeltaTaskWriter w = CdcSinkTestUtils.deltaWriter(t, keyIds, false, TARGET_FILE_SIZE);
+ writeKeyed(w, record(t, concat(key, "old")), 1L, ValueKind.UPDATE_BEFORE, key);
+ writeKeyed(w, record(t, concat(key, "new")), 1L, ValueKind.UPDATE_AFTER, key);
+ WriteResult r = w.complete();
+
+ DeleteFile del = singleEqualityDelete(r, 1, 2, 3, 4, 5, 6);
+ Record deleteRow = Iterables.getOnlyElement(readParquetRows(t, del.location(), schema));
+ for (int i = 0; i < key.length; i++) {
+ assertThat(schema.columns().get(i).name(), deleteRow.get(i), equalTo(key[i]));
+ }
+ assertThat(deleteRow.getField("payload"), nullValue());
+
+ CdcSinkTestUtils.commitRowDelta(t, r);
+ List rows = ImmutableList.copyOf(IcebergGenerics.read(t).build());
+ assertThat(Iterables.getOnlyElement(rows).getField("payload"), equalTo("new"));
+ }
+
+ // The key column is projected by field id, not by position: here it is the last column.
+ @Test
+ public void keyNotInFirstPositionIsProjectedByFieldId() throws Exception {
+ Schema schema =
+ new Schema(
+ Types.NestedField.optional(1, "name", Types.StringType.get()),
+ Types.NestedField.optional(2, "data", Types.StringType.get()),
+ Types.NestedField.required(3, "id", Types.IntegerType.get()));
+ Table t = table(schema, ImmutableSet.of(3));
+ RecordDeltaTaskWriter a =
+ CdcSinkTestUtils.deltaWriter(t, ImmutableSet.of(3), false, TARGET_FILE_SIZE);
+ writeKeyed(a, record(t, "n1", "d1", 7), 0L, ValueKind.INSERT, 7);
+ CdcSinkTestUtils.commitRowDelta(t, a.complete());
+
+ RecordDeltaTaskWriter w =
+ CdcSinkTestUtils.deltaWriter(t, ImmutableSet.of(3), false, TARGET_FILE_SIZE);
+ writeKeyed(w, record(t, "n1", "d1", 7), 1L, ValueKind.UPDATE_BEFORE, 7);
+ writeKeyed(w, record(t, "n2", "d2", 7), 1L, ValueKind.UPDATE_AFTER, 7);
+ WriteResult r = w.complete();
+
+ assertThat(rowStrings(t, r.dataFiles()[0].location()), contains("n2:d2:7"));
+ DeleteFile del = singleEqualityDelete(r, 3);
+ assertThat(rowStrings(t, del.location()), contains("null:null:7"));
+
+ CdcSinkTestUtils.commitRowDelta(t, r);
+ assertThat(readRows(t, schema), contains("n2:d2:7"));
+ }
+
+ private Table table(Schema schema, ImmutableSet identifierFieldIds) {
+ return CdcSinkTestUtils.createTable(
+ catalog,
+ TableIdentifier.of("db", "k" + System.nanoTime()),
+ schema,
+ identifierFieldIds,
+ 2,
+ PartitionSpec.unpartitioned());
+ }
+
+ private static Record record(Table t, Object... values) {
+ GenericRecord r = GenericRecord.create(t.schema());
+ for (int i = 0; i < values.length; i++) {
+ r.set(i, values[i]);
+ }
+ return r;
+ }
+
+ private static Object[] concat(Object[] head, Object tail) {
+ Object[] all = new Object[head.length + 1];
+ System.arraycopy(head, 0, all, 0, head.length);
+ all[head.length] = tail;
+ return all;
+ }
+
+ /** Writes one change whose sort-key pk prefix encodes the given key values. */
+ private static void writeKeyed(
+ RecordDeltaTaskWriter w, Record rec, long seq, ValueKind kind, Object... key) {
+ w.write(CdcSortKey.encode(pkBytes(key), seq, kind), rec, kind);
+ }
+
+ /** Length-prefixed string forms: distinct key tuples get distinct, sortable bytes. */
+ private static byte[] pkBytes(Object... key) {
+ List parts = new ArrayList<>();
+ int size = 0;
+ for (Object value : key) {
+ byte[] part = String.valueOf(value).getBytes(StandardCharsets.UTF_8);
+ parts.add(part);
+ size += 4 + part.length;
+ }
+ ByteBuffer buf = ByteBuffer.allocate(size);
+ for (byte[] part : parts) {
+ buf.putInt(part.length).put(part);
+ }
+ return buf.array();
+ }
+
+ private static DeleteFile singleEqualityDelete(WriteResult r, Integer... equalityFieldIds) {
+ assertThat(r.deleteFiles(), arrayWithSize(1));
+ DeleteFile del = r.deleteFiles()[0];
+ assertThat(del.content(), equalTo(FileContent.EQUALITY_DELETES));
+ assertThat(del.equalityFieldIds(), contains(equalityFieldIds));
+ return del;
+ }
+
+ /** A file's rows as colon-joined field values in table column order. */
+ private static List rowStrings(Table t, String location) throws IOException {
+ List rows = new ArrayList<>();
+ for (Record r : readParquetRows(t, location, t.schema())) {
+ rows.add(joined(r, t.schema()));
+ }
+ return rows;
+ }
+
+ /** The table's current rows as sorted colon-joined strings in column order. */
+ private static List readRows(Table t, Schema schema) throws IOException {
+ List rows = new ArrayList<>();
+ try (CloseableIterable reader = IcebergGenerics.read(t).build()) {
+ for (Record r : reader) {
+ rows.add(joined(r, schema));
+ }
+ }
+ Collections.sort(rows);
+ return rows;
+ }
+
+ private static String joined(Record r, Schema schema) {
+ List values = new ArrayList<>();
+ for (Types.NestedField column : schema.columns()) {
+ values.add(String.valueOf(r.getField(column.name())));
+ }
+ return String.join(":", values);
+ }
+
+ // ---------------------------------------------------------------------------------------------
+ // Format versions, formats, abort, sort order ids
+ // ---------------------------------------------------------------------------------------------
+
+ // V3 gets the identical treatment: cross-commit deletes are Parquet equality deletes.
+ @Test
+ public void v3UpdatePairWritesParquetEqualityDelete() throws Exception {
+ Table t = v3Table();
+ RecordDeltaTaskWriter a = writer(t, false);
+ write(a, rec(t, 1, "a", "x"), 1L, ValueKind.INSERT);
+ CdcSinkTestUtils.commitRowDelta(t, a.complete());
+
+ RecordDeltaTaskWriter b = writer(t, false);
+ write(b, rec(t, 1, "a", "x"), 2L, ValueKind.UPDATE_BEFORE);
+ write(b, rec(t, 1, "b", "z"), 2L, ValueKind.UPDATE_AFTER);
+ WriteResult r = b.complete();
+
+ assertThat(r.deleteFiles(), arrayWithSize(1));
+ assertThat(r.deleteFiles()[0].content(), equalTo(FileContent.EQUALITY_DELETES));
+ assertThat(r.deleteFiles()[0].format(), equalTo(FileFormat.PARQUET));
+
+ CdcSinkTestUtils.commitRowDelta(t, r);
+ assertThat(readRows(t), contains("1:b:z"));
+ }
+
+ // An out-of-order pair (seq 2 before seq 1) trips the sort tripwire instead of miscollapsing.
+ @Test
+ public void unsortedInputThrowsNamingTheProblem() throws Exception {
+ Table t = v2Table();
+ RecordDeltaTaskWriter w = writer(t, false);
+ write(w, rec(t, 1, "a", "x"), 2L, ValueKind.UPDATE_AFTER);
+
+ IllegalStateException error =
+ assertThrows(
+ IllegalStateException.class,
+ () -> write(w, rec(t, 1, "a", "x"), 1L, ValueKind.UPDATE_BEFORE));
+
+ assertThat(error.getMessage(), containsString("unsorted input"));
+ w.abort();
+ }
+
+ // abort() after a partial write removes everything it wrote from the filesystem.
+ @Test
+ public void abortDeletesWrittenFiles() throws Exception {
+ Table t = v2Table();
+ // Avro materializes files at writer-open (Parquet buffers in memory), so the pre-abort
+ // existence check is non-vacuous.
+ t.updateProperties().set(TableProperties.DEFAULT_FILE_FORMAT, "avro").commit();
+ RecordDeltaTaskWriter w = writer(t, false);
+ write(w, rec(t, 1, "a", "x"), 1L, ValueKind.INSERT);
+ write(w, rec(t, 2, "b", "y"), 1L, ValueKind.INSERT);
+
+ assertThat(dataFilesUnder(warehouseDir), not(empty()));
+ w.abort();
+ assertThat(dataFilesUnder(warehouseDir), empty());
+ }
+
+ // abort() still deletes every file when closing one of them fails, and surfaces that failure.
+ @Test
+ public void abortDeletesFilesEvenWhenCloseFails() throws Exception {
+ Table t = v2Table();
+ t.updateProperties().set(TableProperties.DEFAULT_FILE_FORMAT, "avro").commit();
+ CloseFailingFileIO io = new CloseFailingFileIO(t.io());
+ FileFormat format = RecordDeltaTaskWriter.dataFileFormat(t);
+ RecordDeltaTaskWriter w =
+ RecordDeltaTaskWriter.create(
+ t,
+ t.spec(),
+ ImmutableSet.of(1),
+ true,
+ TARGET_FILE_SIZE,
+ OutputFileFactory.builderFor(t, 1, 1).ioSupplier(() -> io).build(),
+ format,
+ RecordDeltaTaskWriter.deleteFileFormat(t, format));
+ // The second key flushes the first block, opening the delete file (fails on close) and the
+ // data file (closes fine).
+ write(w, rec(t, 1, "a", "x"), 1L, ValueKind.INSERT);
+ write(w, rec(t, 2, "b", "y"), 1L, ValueKind.INSERT);
+ assertThat(dataFilesUnder(warehouseDir), not(empty()));
+
+ Exception failure = assertThrows(Exception.class, w::abort);
+ assertThat(Throwables.getRootCause(failure).getMessage(), containsString("simulated"));
+ assertThat(dataFilesUnder(warehouseDir), empty());
+ }
+
+ // Factory format resolution: write.format.default and write.delete.format.default.
+ @Test
+ public void resolvesDataAndDeleteFileFormats() {
+ Table t = v2Table();
+ assertThat(RecordDeltaTaskWriter.dataFileFormat(t), equalTo(FileFormat.PARQUET));
+ assertThat(
+ RecordDeltaTaskWriter.deleteFileFormat(t, FileFormat.PARQUET), equalTo(FileFormat.PARQUET));
+
+ t.updateProperties().set(TableProperties.DELETE_DEFAULT_FILE_FORMAT, "avro").commit();
+ assertThat(
+ RecordDeltaTaskWriter.deleteFileFormat(t, FileFormat.PARQUET), equalTo(FileFormat.AVRO));
+ }
+
+ /**
+ * Every equality delete this writer produces carries sort order id 0 (unsorted), INCLUDING on a
+ * table that declares a sort order. Pins the premise {@code
+ * CommitDeltas.sortOrdersForReconstruction} rests on: if the writer ever stamped the table's real
+ * sort order, that special case would become both unnecessary and wrong, and this test says so.
+ */
+ @Test
+ public void sinkEqualityDeletesCarryUnsortedSortOrderId() throws Exception {
+ for (Table t : ImmutableList.of(v2Table(), v2SortedTable())) {
+ // Writer A commits an INSERT so writer B's DELETE is a cross-commit equality delete.
+ RecordDeltaTaskWriter a = writer(t, false);
+ write(a, rec(t, 1, "a", "x"), 1L, ValueKind.INSERT);
+ CdcSinkTestUtils.commitRowDelta(t, a.complete());
+
+ RecordDeltaTaskWriter b = writer(t, false);
+ write(b, rec(t, 1, "a", "x"), 2L, ValueKind.DELETE);
+ DeleteFile[] deletes = b.complete().deleteFiles();
+
+ assertThat(deletes, arrayWithSize(1));
+ assertThat(deletes[0].content(), equalTo(FileContent.EQUALITY_DELETES));
+ assertThat(deletes[0].sortOrderId(), equalTo(SortOrder.unsorted().orderId()));
+ b.abort();
+ }
+ }
+
+ /**
+ * The data-file half of the same premise: sink data files also carry sort order id 0. DO NOT add
+ * {@code .dataSortOrder(...)} to the factory: {@code SerializableDataFile} carries no
+ * sortOrderId, so a real sort order would be silently RESET at reconstruction, and nothing but
+ * this test would notice.
+ */
+ @Test
+ public void sinkDataFilesCarryUnsortedSortOrderId() throws Exception {
+ for (Table t : ImmutableList.of(v2Table(), v2SortedTable())) {
+ RecordDeltaTaskWriter w = writer(t, false);
+ write(w, rec(t, 1, "a", "x"), 1L, ValueKind.INSERT);
+ DataFile[] dataFiles = w.complete().dataFiles();
+
+ assertThat(dataFiles, arrayWithSize(1));
+ assertThat(dataFiles[0].sortOrderId(), equalTo(SortOrder.unsorted().orderId()));
+
+ // ...and the transport round trip preserves it, which is only true while it IS 0:
+ // SerializableDataFile has no sortOrderId field to carry anything else.
+ DataFile rebuilt =
+ SerializableDataFile.from(dataFiles[0], t.spec()).createDataFile(t.specs());
+ assertThat(rebuilt.sortOrderId(), equalTo(dataFiles[0].sortOrderId()));
+ }
+ }
+
+ /** Regular files under any table's {@code data/} directory (excludes {@code metadata/}). */
+ private static List dataFilesUnder(File dir) throws IOException {
+ String dataSegment = File.separator + "data" + File.separator;
+ try (Stream walk = Files.walk(dir.toPath())) {
+ return walk.filter(Files::isRegularFile)
+ .filter(p -> p.toString().contains(dataSegment))
+ .collect(Collectors.toList());
+ }
+ }
+
+ /** Delegates to a real {@link FileIO}; the first file it creates fails on close after writing. */
+ private static final class CloseFailingFileIO implements FileIO {
+ private final FileIO delegate;
+ private boolean armed = true;
+
+ CloseFailingFileIO(FileIO delegate) {
+ this.delegate = delegate;
+ }
+
+ @Override
+ public InputFile newInputFile(String path) {
+ return delegate.newInputFile(path);
+ }
+
+ @Override
+ public void deleteFile(String path) {
+ delegate.deleteFile(path);
+ }
+
+ @Override
+ public OutputFile newOutputFile(String path) {
+ OutputFile file = delegate.newOutputFile(path);
+ if (!armed) {
+ return file;
+ }
+ armed = false;
+ return new OutputFile() {
+ @Override
+ public PositionOutputStream create() {
+ return failingOnClose(file.create());
+ }
+
+ @Override
+ public PositionOutputStream createOrOverwrite() {
+ return failingOnClose(file.createOrOverwrite());
+ }
+
+ @Override
+ public String location() {
+ return file.location();
+ }
+
+ @Override
+ public InputFile toInputFile() {
+ return file.toInputFile();
+ }
+ };
+ }
+
+ private static PositionOutputStream failingOnClose(PositionOutputStream out) {
+ return new PositionOutputStream() {
+ @Override
+ public long getPos() throws IOException {
+ return out.getPos();
+ }
+
+ @Override
+ public void write(int b) throws IOException {
+ out.write(b);
+ }
+
+ @Override
+ public void write(byte[] b, int off, int len) throws IOException {
+ out.write(b, off, len);
+ }
+
+ @Override
+ public void flush() throws IOException {
+ out.flush();
+ }
+
+ @Override
+ public void close() throws IOException {
+ out.close();
+ throw new IOException("simulated close failure");
+ }
+ };
+ }
+ }
+}