[Iceberg CDC sink] SchemaTransform (ManagedIO) layer

CHANGES.md

@@ -102,6 +102,7 @@
 * ClickHouseIO: support writing `Decimal(P, S)` / `Decimal32/64/128/256` columns (Java) ([#39840](https://github.com/apache/beam/issues/39840)).
 * SolaceIO now supports reading and writing user properties (message metadata) (Java) ([#40099](https://github.com/apache/beam/issues/40099)).
 * [IcebergIO] AddFiles (`IcebergAddFiles` in YAML) can evolve the table schema before registering files, with `schema_evolution_options`, `required_columns`, `incompatible_schema_handling` and `unverifiable_file_handling` (Java/YAML, batch only) ([#40144](https://github.com/apache/beam/issues/40144)).
+* [IcebergIO] Added batch and streaming CDC writes that applies INSERT/UPDATE_BEFORE/UPDATE_AFTER/DELETE changes to Iceberg V2+ tables by primary key. Invoke with `IcebergIO.writeCdcRows` (Java) or by setting `mode: merge-on-read` on the Managed `ICEBERG` write (Java, Python, YAML) ([#39979](https://github.com/apache/beam/issues/39979)).
 
 ## New Features / Improvements
 

sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CdcWriteConfig.java

@@ -31,9 +31,6 @@
 /** 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 <b>partitioned</b> table {@link
@@ -52,7 +49,7 @@ abstract class CdcWriteConfig implements Serializable {
 
   /**
    * 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}.
+   * changes. Defaults to {@value IcebergCdcMetadataColumns#COMMIT_SNAPSHOT_SEQUENCE_NUMBER}.
    */
   abstract String getSequenceNumberColumn();
 
@@ -131,7 +128,7 @@ abstract class CdcWriteConfig implements Serializable {
 
   static Builder builder() {
     return new AutoValue_CdcWriteConfig.Builder()
-        .setSequenceNumberColumn(DEFAULT_SEQUENCE_NUMBER_COLUMN)
+        .setSequenceNumberColumn(IcebergCdcMetadataColumns.COMMIT_SNAPSHOT_SEQUENCE_NUMBER)
         .setNumShards(DEFAULT_NUM_SHARDS)
         .setShardsPerPartition(DEFAULT_NUM_SHARDS)
         .setSorterMemoryMB(DEFAULT_SORTER_MEMORY_MB)

sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/WriteCdcRows.java

@@ -29,6 +29,7 @@
 import org.apache.beam.sdk.io.iceberg.IcebergCatalogConfig;
 import org.apache.beam.sdk.io.iceberg.IcebergWriteResult;
 import org.apache.beam.sdk.io.iceberg.SnapshotInfo;
+import org.apache.beam.sdk.io.iceberg.cdc.IcebergCdcMetadataColumns;
 import org.apache.beam.sdk.schemas.Schema;
 import org.apache.beam.sdk.transforms.PTransform;
 import org.apache.beam.sdk.transforms.display.DisplayData;
@@ -184,7 +185,7 @@ abstract static class Builder {
   public static WriteCdcRows of(IcebergCatalogConfig catalogConfig) {
     return new AutoValue_WriteCdcRows.Builder()
         .setCatalogConfig(catalogConfig)
-        .setSequenceNumberColumn(CdcWriteConfig.DEFAULT_SEQUENCE_NUMBER_COLUMN)
+        .setSequenceNumberColumn(IcebergCdcMetadataColumns.COMMIT_SNAPSHOT_SEQUENCE_NUMBER)
         .setNumShards(CdcWriteConfig.DEFAULT_NUM_SHARDS)
         .setSorterMemoryMB(CdcWriteConfig.DEFAULT_SORTER_MEMORY_MB)
         .setUpsert(false)
@@ -223,7 +224,7 @@ public WriteCdcRows withEqualityColumns(List<String> columns) {
    * The column holding the per-primary-key monotonic sequence number used to order a single key's
    * changes. Must be declared as a non-nullable {@code INT64} in the input schema. The column is
    * stripped from the written rows. Defaults to {@value
-   * CdcWriteConfig#DEFAULT_SEQUENCE_NUMBER_COLUMN}.
+   * IcebergCdcMetadataColumns#COMMIT_SNAPSHOT_SEQUENCE_NUMBER}.
    */
   public WriteCdcRows withSequenceNumberColumn(String column) {
     return toBuilder().setSequenceNumberColumn(column).build();

sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/IcebergSchemaTransformTranslationTest.java

@@ -98,6 +98,18 @@ public class IcebergSchemaTransformTranslationTest {
           .withFieldValue("keep", Collections.singletonList("str"))
           .build();
 
+  /** A merge-on-read write config: the mode plus the options gated to it. */
+  private static final Row WRITE_CDC_CONFIG_ROW =
+      Row.withSchema(WRITE_PROVIDER.configurationSchema())
+          .withFieldValue("table", "test_table_identifier")
+          .withFieldValue("catalog_properties", CATALOG_PROPERTIES)
+          .withFieldValue("mode", "merge-on-read")
+          .withFieldValue("change_type_column", "op")
+          .withFieldValue("upsert", true)
+          .withFieldValue("equality_columns", Collections.singletonList("id"))
+          .withFieldValue("sink_id", "stable-sink")
+          .build();
+
   private static final Row READ_CONFIG_ROW =
       Row.withSchema(READ_PROVIDER.configurationSchema())
           .withFieldValue("table", "test_table_identifier")
@@ -136,6 +148,21 @@ public void testReCreateWriteTransformFromRow() {
     assertEquals(WRITE_CONFIG_ROW, writeTransformFromRow.getConfigurationRow());
   }
 
+  @Test
+  public void testReCreateCdcWriteTransformFromRow() {
+    IcebergWriteSchemaTransform writeTransform =
+        (IcebergWriteSchemaTransform) WRITE_PROVIDER.from(WRITE_CDC_CONFIG_ROW);
+
+    IcebergSchemaTransformTranslation.IcebergWriteSchemaTransformTranslator translator =
+        new IcebergSchemaTransformTranslation.IcebergWriteSchemaTransformTranslator();
+    Row row = translator.toConfigRow(writeTransform);
+
+    IcebergWriteSchemaTransform writeTransformFromRow =
+        translator.fromConfigRow(row, PipelineOptionsFactory.create());
+
+    assertEquals(WRITE_CDC_CONFIG_ROW, writeTransformFromRow.getConfigurationRow());
+  }
+
   @Test
   public void testWriteTransformProtoTranslation()
       throws InvalidProtocolBufferException, IOException {