From bf0eb84c43ef0fac8df0f72a0d895a5a4972a7f5 Mon Sep 17 00:00:00 2001 From: Ahmed Abualsaud Date: Thu, 3 Sep 2026 11:50:48 -0700 Subject: [PATCH 1/6] config and writer --- .../io/iceberg/cdc/sink/CdcWriteConfig.java | 231 ++++++ .../cdc/sink/RecordDeltaTaskWriter.java | 462 +++++++++++ .../iceberg/cdc/sink/CdcWriteConfigTest.java | 226 ++++++ .../cdc/sink/RecordDeltaTaskWriterTest.java | 768 ++++++++++++++++++ 4 files changed, 1687 insertions(+) create mode 100644 sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcWriteConfig.java create mode 100644 sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/RecordDeltaTaskWriter.java create mode 100644 sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcWriteConfigTest.java create mode 100644 sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/RecordDeltaTaskWriterTest.java diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcWriteConfig.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcWriteConfig.java new file mode 100644 index 000000000000..ac91213710b7 --- /dev/null +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcWriteConfig.java @@ -0,0 +1,231 @@ +/* + * 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.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkArgument; + +import com.google.auto.value.AutoValue; +import java.io.Serializable; +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import org.apache.beam.sdk.io.iceberg.cdc.IcebergCdcMetadataColumns; +import org.apache.beam.sdk.values.ValueKind; +import org.checkerframework.checker.nullness.qual.Nullable; + +/** Configuration for the CDC sink. */ +@AutoValue +abstract class CdcWriteConfig implements Serializable { + static final String DEFAULT_SEQUENCE_NUMBER_COLUMN = + IcebergCdcMetadataColumns.COMMIT_SNAPSHOT_SEQUENCE_NUMBER; + + /** + * Every shard that touches a partition writes a file per commit window, so {@code num_shards x + * touched partitions x windows per day} files. On a partitioned table {@link + * #getShardsPerPartition()} caps the per-partition factor without lowering this number, trading + * per-partition write parallelism for proportionally fewer files. + */ + static final int DEFAULT_NUM_SHARDS = 16; + + static final int DEFAULT_SORTER_MEMORY_MB = 100; + + /** + * Columns that define a row's identity (the Iceberg equality-delete fields). If unspecified, will + * try to use the destination table's identifier fields. + */ + abstract @Nullable List getEqualityColumns(); + + /** + * The column holding the per-primary-key monotonic sequence number used to order a single key's + * changes. Defaults to {@value #DEFAULT_SEQUENCE_NUMBER_COLUMN}. + */ + abstract String getSequenceNumberColumn(); + + /** + * If set, the change kind is read from this string column instead of the element's native {@link + * ValueKind}. The column is stripped from the data row and never written to Iceberg. + */ + abstract @Nullable String getChangeTypeColumn(); + + /** + * Optional mapping from {@link #getChangeTypeColumn()} values to {@link ValueKind} names (e.g. + * {@code {"c": "INSERT", "u": "UPDATE_AFTER", "d": "DELETE"}}). If {@code null}, {@link + * #getChangeTypeColumn()} values must already be {@link ValueKind} names. + */ + abstract @Nullable Map getChangeTypeMap(); + + /** + * The number of deterministic primary-key-hash shards (logical write buckets) per destination. + * Defaults to {@value #DEFAULT_NUM_SHARDS}; set it to about your pipeline's write parallelism. + */ + abstract int getNumShards(); + + /** + * The maximum number of shards a single partition's rows may occupy on a partitioned + * destination. A {@code (destination, window)} writes about {@code min(shards_per_partition, + * distinct keys)} files per touched partition, and per-partition write parallelism is capped at + * this value. {@code 1} pins each partition to a single writer, {@code num_shards} is plain + * primary-key sharding. Ignored for an unpartitioned destination, which always shards by primary + * key. + */ + abstract int getShardsPerPartition(); + + /** + * The in-memory buffer size (MB) for the sorter that orders each shard's records by primary key, + * then sequence number, then change kind, before writing. Must be {@code >= 1}. Defaults to + * {@value #DEFAULT_SORTER_MEMORY_MB}. + */ + abstract int getSorterMemoryMB(); + + /** + * If {@code true}, {@code UPDATE_BEFORE} records are dropped and {@code INSERT}/{@code + * UPDATE_AFTER} are applied as upserts (equality-delete-then-insert on the primary key). Defaults + * to {@code false}. + */ + abstract boolean getUpsert(); + + /** + * If set, a destination that has committed at least once emits a periodic empty token-refresh + * commit while idle, keeping this sink's committed-through token snapshot recent. Disabled + * ({@code null}) by default. + */ + abstract @Nullable Long getTokenHeartbeatMillis(); + + /** + * A stable identifier for this sink, used to namespace the idempotency tokens written to each + * commit's Iceberg snapshot summary. + */ + abstract String getSinkId(); + + /** + * Extra user properties to add to every commit's Iceberg snapshot summary. Keys prefixed with + * {@code beam.cdc.} are reserved for the sink's own idempotency/diagnostic tokens. + */ + abstract @Nullable Map getSnapshotProperties(); + + /** + * If {@code true}, a poison record (unknown change type, missing/null sequence number, null + * equality value, an unresolvable destination) is diverted to the sink's failed-rows output + * instead of failing the pipeline. Defaults to {@code false} (fail-fast). + */ + abstract boolean getErrorHandling(); + + static Builder builder() { + return new AutoValue_CdcWriteConfig.Builder() + .setSequenceNumberColumn(DEFAULT_SEQUENCE_NUMBER_COLUMN) + .setNumShards(DEFAULT_NUM_SHARDS) + .setShardsPerPartition(DEFAULT_NUM_SHARDS) + .setSorterMemoryMB(DEFAULT_SORTER_MEMORY_MB) + .setUpsert(false) + .setErrorHandling(false); + } + + void validate() { + checkArgument(getNumShards() >= 1, "num_shards must be >= 1, got %s", getNumShards()); + checkArgument( + getShardsPerPartition() >= 1 && getShardsPerPartition() <= getNumShards(), + "shards_per_partition must be between 1 and num_shards (%s); got %s", + getNumShards(), + getShardsPerPartition()); + checkArgument( + getSorterMemoryMB() >= 1, "sorter_memory_mb must be >= 1, got %s", getSorterMemoryMB()); + + @Nullable List equalityColumns = getEqualityColumns(); + checkArgument( + equalityColumns == null || !equalityColumns.isEmpty(), + "equality_columns must be non-empty or unset (leave unset to use the table's identifier " + + "fields)."); + + checkArgument( + !getSequenceNumberColumn().equals(getChangeTypeColumn()), + "sequence_number_column and change_type_column must be distinct, both are '%s'.", + getSequenceNumberColumn()); + + @Nullable Map changeTypeMap = getChangeTypeMap(); + checkArgument( + changeTypeMap == null || getChangeTypeColumn() != null, + "change_type_map requires change_type_column to also be set (it defines the source " + + "values mapped for that column)."); + if (changeTypeMap != null) { + for (String value : changeTypeMap.values()) { + checkArgument( + isValueKindName(value), + "change_type_map value '%s' is not a valid ValueKind name; must be one of %s.", + value, + Arrays.toString(ValueKind.values())); + } + } + + @Nullable Long heartbeatMillis = getTokenHeartbeatMillis(); + checkArgument( + heartbeatMillis == null || heartbeatMillis > 0, + "token heartbeat (withTokenHeartbeat / token_heartbeat_seconds) must be > 0 when set, " + + "got %s ms", + heartbeatMillis); + + @Nullable Map snapshotProperties = getSnapshotProperties(); + if (snapshotProperties != null) { + for (String key : snapshotProperties.keySet()) { + checkArgument( + !key.startsWith("beam.cdc."), + "snapshot_properties key '%s' uses the reserved 'beam.cdc.' prefix; choose a " + + "different key.", + key); + } + } + } + + private static boolean isValueKindName(String value) { + for (ValueKind kind : ValueKind.values()) { + if (kind.name().equals(value)) { + return true; + } + } + return false; + } + + @AutoValue.Builder + abstract static class Builder { + + abstract Builder setEqualityColumns(@Nullable List equalityColumns); + + abstract Builder setSequenceNumberColumn(String sequenceNumberColumn); + + abstract Builder setChangeTypeColumn(@Nullable String changeTypeColumn); + + abstract Builder setChangeTypeMap(@Nullable Map changeTypeMap); + + abstract Builder setNumShards(int numShards); + + abstract Builder setShardsPerPartition(int shardsPerPartition); + + abstract Builder setSorterMemoryMB(int sorterMemoryMB); + + abstract Builder setUpsert(boolean upsert); + + abstract Builder setTokenHeartbeatMillis(@Nullable Long tokenHeartbeatMillis); + + abstract Builder setSinkId(String sinkId); + + abstract Builder setSnapshotProperties(@Nullable Map 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..0fa13df54c09 --- /dev/null +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/RecordDeltaTaskWriter.java @@ -0,0 +1,462 @@ +/* + * 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 none. + *
  • 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()]; + List allFields = schema.columns(); + for (int i = 0; i < pkFields.size(); i++) { + int fieldId = pkFields.get(i).fieldId(); + int pos = -1; + for (int j = 0; j < allFields.size(); j++) { + if (allFields.get(j).fieldId() == fieldId) { + pos = j; + break; + } + } + if (pos < 0) { + 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. */ + public void abort() throws IOException { + close(); + List locations = new ArrayList<>(); + for (PartitionDeltaWriter writer : partitionWriters) { + for (DataFile file : writer.dataFiles()) { + locations.add(file.location()); + } + for (DeleteFile file : writer.deleteFiles()) { + locations.add(file.location()); + } + } + Tasks.foreach(locations).throwFailureWhenFinished().noRetry().run(io::deleteFile); + } + + 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; + + 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(); + } + } finally { + if (deleteWriter != null) { + deleteWriter.close(); + } + } + } + + 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/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..ba9dae468245 --- /dev/null +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/RecordDeltaTaskWriterTest.java @@ -0,0 +1,768 @@ +/* + * 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.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +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.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.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.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()); + } + + // --------------------------------------------------------------------------------------------- + // 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()); + } + + // 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()); + } + } +} From f0a51d1c5fa3a4c3267f0c89cf2e2b7e8c3e4912 Mon Sep 17 00:00:00 2001 From: Ahmed Abualsaud Date: Thu, 3 Sep 2026 16:01:08 -0700 Subject: [PATCH 2/6] test utils --- .../io/iceberg/cdc/sink/CdcSinkTestUtils.java | 193 ++++++++++++++++++ 1 file changed, 193 insertions(+) create mode 100644 sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcSinkTestUtils.java 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..49776e5aeff7 --- /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(); + } + }); + } +} From 60e6ccfb1c874b3f53e2aa98e6b463c543a66667 Mon Sep 17 00:00:00 2001 From: Ahmed Abualsaud Date: Thu, 3 Sep 2026 16:17:25 -0700 Subject: [PATCH 3/6] spotless --- .../io/iceberg/cdc/sink/CdcSinkTestUtils.java | 98 +++++++++---------- 1 file changed, 49 insertions(+), 49 deletions(-) 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 index 49776e5aeff7..9f67e45f0801 100644 --- 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 @@ -65,23 +65,23 @@ static Catalog hadoopCatalog(File warehouseDir) { /** 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(); + .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) { + 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); @@ -94,17 +94,17 @@ static Table createTable( */ 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())); + 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()); + catalog, + TableIdentifier.of("db", name), + destTableSchema, + ImmutableSet.of(1), + 2, + PartitionSpec.unpartitioned()); } } @@ -113,20 +113,20 @@ static void createDestTables(Catalog catalog, String tableA, String tableB) { * 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) { + 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(); + .buildTable(id, schemaWithIds) + .withPartitionSpec(spec) + .withSortOrder(sortOrder) + .withProperties(ImmutableMap.of("format-version", String.valueOf(formatVersion))) + .create(); } /** @@ -134,18 +134,18 @@ static Table createSortedTable( * the current spec as the pinned spec. */ static RecordDeltaTaskWriter deltaWriter( - Table table, Set equalityFieldIds, boolean upsert, long targetFileSizeBytes) { + 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); + 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. */ @@ -183,11 +183,11 @@ static PCollection withKinds(String name, PCollection> t 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(); - } - }); + new DoFn, Row>() { + @ProcessElement + public void process(@Element KV e, OutputReceiver out) { + out.builder(e.getValue()).setValueKind(e.getKey()).output(); + } + }); } } From 93991469028f475d77d1bede37ce5ff9d7abbf8a Mon Sep 17 00:00:00 2001 From: Ahmed Abualsaud Date: Thu, 3 Sep 2026 17:31:26 -0700 Subject: [PATCH 4/6] spotless --- .../java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcRecord.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcRecord.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcRecord.java index 1aa82914fde9..3759443b735d 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcRecord.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcRecord.java @@ -17,10 +17,10 @@ */ package org.apache.beam.sdk.io.iceberg.cdc.sink; -import com.google.common.base.MoreObjects; import java.util.Objects; 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.base.MoreObjects; import org.checkerframework.checker.nullness.qual.Nullable; /** From b9fd7e8d13eb06f0fa5a3415ebfeafeeef07bf54 Mon Sep 17 00:00:00 2001 From: Ahmed Abualsaud Date: Fri, 4 Sep 2026 02:39:04 -0700 Subject: [PATCH 5/6] pull test fix --- .../sdk/io/iceberg/cdc/sink/CdcRecordCoderTest.java | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) 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); From d8e82762736ca6c5805454200c30425172989512 Mon Sep 17 00:00:00 2001 From: Ahmed Abualsaud Date: Fri, 4 Sep 2026 06:41:03 -0700 Subject: [PATCH 6/6] remove dep --- sdks/java/io/iceberg/build.gradle | 1 - 1 file changed, 1 deletion(-) diff --git a/sdks/java/io/iceberg/build.gradle b/sdks/java/io/iceberg/build.gradle index 7340d1eeb411..e2e8a12d01eb 100644 --- a/sdks/java/io/iceberg/build.gradle +++ b/sdks/java/io/iceberg/build.gradle @@ -49,7 +49,6 @@ dependencies { implementation library.java.avro implementation library.java.slf4j_api implementation library.java.joda_time - implementation library.java.guava implementation "org.apache.parquet:parquet-column:$parquet_version" implementation "org.apache.parquet:parquet-hadoop:$parquet_version" implementation "org.apache.parquet:parquet-common:$parquet_version"