[DO NOT MERGE] full iceberg cdc sink

model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/external_transforms.proto

@@ -109,6 +109,9 @@ message ManagedTransforms {
       "beam:schematransform:org.apache.beam:delta_lake_read:v1"];
     DELTA_LAKE_CDC_READ = 14 [(org.apache.beam.model.pipeline.v1.beam_urn) =
       "beam:schematransform:org.apache.beam:delta_lake_cdc_read:v1"];
+    // Applies a stream or batch of CDC row-level changes to Iceberg tables.
+    ICEBERG_CDC_WRITE = 15 [(org.apache.beam.model.pipeline.v1.beam_urn) =
+      "beam:schematransform:org.apache.beam:iceberg_cdc_write:v1"];
   }
 }
 

sdks/java/io/iceberg/build.gradle

@@ -46,6 +46,7 @@ dependencies {
     implementation library.java.vendored_guava_32_1_2_jre
     implementation project(path: ":sdks:java:core", configuration: "shadow")
     implementation project(path: ":model:pipeline", configuration: "shadow")
+    implementation project(path: ":sdks:java:extensions:sorter")
     implementation library.java.avro
     implementation library.java.slf4j_api
     implementation library.java.joda_time

sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergIO.java

@@ -26,6 +26,7 @@
 import org.apache.beam.sdk.annotations.Internal;
 import org.apache.beam.sdk.io.Read;
 import org.apache.beam.sdk.io.iceberg.cdc.IncrementalChangelogSource;
+import org.apache.beam.sdk.io.iceberg.cdc.sink.WriteCdcRows;
 import org.apache.beam.sdk.options.StreamingOptions;
 import org.apache.beam.sdk.schemas.Schema;
 import org.apache.beam.sdk.transforms.PTransform;
@@ -393,6 +394,23 @@ public static WriteRows writeRows(IcebergCatalogConfig catalog) {
         .build();
   }
 
+  /**
+   * Returns a {@link WriteCdcRows} transform: the CDC sink, applying inserts/updates/deletes from a
+   * {@code PCollection<Row>} of change records (each carrying a {@link
+   * org.apache.beam.sdk.values.ValueKind}) to one or more Iceberg V2+ tables via equality deletes;
+   * superseded rows are never written.
+   *
+   * <pre>{@code
+   * input.apply(IcebergIO.writeCdcRows(catalogConfig)
+   *     .to(tableId)
+   *     .withSequenceNumberColumn("seq")
+   *     .withTriggeringFrequency(Duration.standardMinutes(1)));
+   * }</pre>
+   */
+  public static WriteCdcRows writeCdcRows(IcebergCatalogConfig catalog) {
+    return WriteCdcRows.of(catalog);
+  }
+
   @AutoValue
   public abstract static class WriteRows extends PTransform<PCollection<Row>, IcebergWriteResult> {
 

sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergSchemaTransformTranslation.java

@@ -17,6 +17,7 @@
  */
 package org.apache.beam.sdk.io.iceberg;
 
+import static org.apache.beam.sdk.io.iceberg.IcebergCdcWriteSchemaTransformProvider.IcebergCdcWriteSchemaTransform;
 import static org.apache.beam.sdk.io.iceberg.IcebergReadSchemaTransformProvider.IcebergReadSchemaTransform;
 import static org.apache.beam.sdk.io.iceberg.IcebergWriteSchemaTransformProvider.IcebergWriteSchemaTransform;
 import static org.apache.beam.sdk.schemas.transforms.SchemaTransformTranslation.SchemaTransformPayloadTranslator;
@@ -87,6 +88,19 @@ public Row toConfigRow(IcebergWriteSchemaTransform transform) {
     }
   }
 
+  static class IcebergCdcWriteSchemaTransformTranslator
+      extends SchemaTransformPayloadTranslator<IcebergCdcWriteSchemaTransform> {
+    @Override
+    public SchemaTransformProvider provider() {
+      return new IcebergCdcWriteSchemaTransformProvider();
+    }
+
+    @Override
+    public Row toConfigRow(IcebergCdcWriteSchemaTransform transform) {
+      return transform.getConfigurationRow();
+    }
+  }
+
   @AutoService(TransformPayloadTranslatorRegistrar.class)
   public static class WriteRegistrar implements TransformPayloadTranslatorRegistrar {
     @Override
@@ -97,6 +111,7 @@ public static class WriteRegistrar implements TransformPayloadTranslatorRegistra
         getTransformPayloadTranslators() {
       return ImmutableMap.<Class<? extends PTransform>, TransformPayloadTranslator>builder()
           .put(IcebergWriteSchemaTransform.class, new IcebergWriteSchemaTransformTranslator())
+          .put(IcebergCdcWriteSchemaTransform.class, new IcebergCdcWriteSchemaTransformTranslator())
           .build();
     }
   }

sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergWriteResult.java

@@ -19,31 +19,110 @@
 
 import java.util.Map;
 import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.annotations.Internal;
 import org.apache.beam.sdk.transforms.PTransform;
 import org.apache.beam.sdk.values.KV;
 import org.apache.beam.sdk.values.PCollection;
 import org.apache.beam.sdk.values.PInput;
 import org.apache.beam.sdk.values.POutput;
 import org.apache.beam.sdk.values.PValue;
+import org.apache.beam.sdk.values.Row;
 import org.apache.beam.sdk.values.TupleTag;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
+import org.checkerframework.checker.nullness.qual.Nullable;
 
+/**
+ * The output of an {@code IcebergIO} write: the snapshots each destination table committed, plus
+ * the two diversion outputs the CDC sink can produce.
+ *
+ * <p>Only {@link #getSnapshots()} is always present. {@link #getDeadLetterRows()} (late-but-valid
+ * records) and {@link #getFailedRows()} (per-record poison rows) are non-null only for results
+ * built by {@link #cdc}, that is, by {@code IcebergIO.writeCdcRows}, and {@code getFailedRows()}
+ * additionally only when error handling was enabled. The append-only sink ({@code
+ * IcebergIO.writeRows}) leaves both null rather than exposing outputs that can never carry data.
+ */
 public final class IcebergWriteResult implements POutput {
 
   private static final TupleTag<KV<String, SnapshotInfo>> SNAPSHOTS_TAG =
       new TupleTag<KV<String, SnapshotInfo>>() {};
 
+  private static final TupleTag<Row> DEAD_LETTER_TAG = new TupleTag<Row>() {};
+
+  private static final TupleTag<Row> FAILED_ROWS_TAG = new TupleTag<Row>() {};
+
   private final Pipeline pipeline;
 
   private final PCollection<KV<String, SnapshotInfo>> snapshots;
 
+  private final @Nullable PCollection<Row> deadLetterRows;
+
+  private final @Nullable PCollection<Row> failedRows;
+
+  /**
+   * The committed snapshots, keyed by destination. A window committed in a commit fire that later
+   * fails may not re-emit its {@link SnapshotInfo} on the retry (table state is unaffected).
+   */
   public PCollection<KV<String, SnapshotInfo>> getSnapshots() {
     return snapshots;
   }
 
+  /**
+   * The replayable dead-letter {@link Row}s from the CDC sink: records whose grouped pane fired
+   * late, i.e. after the watermark had passed their commit window's end. Schema is {@code record
+   * ROW<dataSchema> + change_type STRING + sequence_number INT64 + destination STRING}. To replay,
+   * unnest {@code record}, map {@code change_type}/{@code sequence_number} as the sink's control
+   * columns, and route each row by {@code destination}. Replaying is only safe while no newer
+   * change for those keys has committed; a stale replay's equality delete removes the newer row.
+   *
+   * @return the dead-letter {@code PCollection<Row>} for results produced by {@code
+   *     IcebergIO.writeCdcRows}/{@link #cdc}, or {@code null} for results produced by the
+   *     append-only sink ({@code IcebergIO.writeRows}), which has no dead-letter output.
+   */
+  public @Nullable PCollection<Row> getDeadLetterRows() {
+    return deadLetterRows;
+  }
+
+  /**
+   * The per-record poison rows diverted by the CDC sink when error handling is enabled (see {@code
+   * WriteCdcRows.withErrorHandling}); schema is {@code failed_row ROW + error_message STRING} (see
+   * {@link org.apache.beam.sdk.schemas.transforms.providers.ErrorHandling}). Distinct from {@link
+   * #getDeadLetterRows()} (which carries late-but-valid records).
+   *
+   * @return the failed-rows {@code PCollection<Row>}, or {@code null} when error handling was not
+   *     enabled (or for the append-only sink).
+   */
+  public @Nullable PCollection<Row> getFailedRows() {
+    return failedRows;
+  }
+
   IcebergWriteResult(Pipeline pipeline, PCollection<KV<String, SnapshotInfo>> snapshots) {
+    this(pipeline, snapshots, null, null);
+  }
+
+  private IcebergWriteResult(
+      Pipeline pipeline,
+      PCollection<KV<String, SnapshotInfo>> snapshots,
+      @Nullable PCollection<Row> deadLetterRows,
+      @Nullable PCollection<Row> failedRows) {
     this.pipeline = pipeline;
     this.snapshots = snapshots;
+    this.deadLetterRows = deadLetterRows;
+    this.failedRows = failedRows;
+  }
+
+  /**
+   * Returns an {@link IcebergWriteResult} for the CDC sink, exposing the committed-snapshot {@code
+   * snapshots}, the replayable {@code deadLetterRows} (see {@link #getDeadLetterRows()}), and the
+   * optional per-record {@code failedRows} (see {@link #getFailedRows()}, {@code null} when error
+   * handling is off).
+   */
+  @Internal
+  public static IcebergWriteResult cdc(
+      Pipeline pipeline,
+      PCollection<KV<String, SnapshotInfo>> snapshots,
+      PCollection<Row> deadLetterRows,
+      @Nullable PCollection<Row> failedRows) {
+    return new IcebergWriteResult(pipeline, snapshots, deadLetterRows, failedRows);
   }
 
   @Override
@@ -55,6 +134,12 @@ public Pipeline getPipeline() {
   public Map<TupleTag<?>, PValue> expand() {
     ImmutableMap.Builder<TupleTag<?>, PValue> output = ImmutableMap.builder();
     output.put(SNAPSHOTS_TAG, snapshots);
+    if (deadLetterRows != null) {
+      output.put(DEAD_LETTER_TAG, deadLetterRows);
+    }
+    if (failedRows != null) {
+      output.put(FAILED_ROWS_TAG, failedRows);
+    }
     return output.build();
   }
 

sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergWriteSnapshotOutput.java

@@ -0,0 +1,79 @@
+/*
+ * 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;
+
+import org.apache.beam.sdk.schemas.NoSuchSchemaException;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.schemas.SchemaRegistry;
+import org.apache.beam.sdk.transforms.SimpleFunction;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.Row;
+
+/**
+ * Output-row helpers for the Iceberg CDC write SchemaTransform provider ({@link
+ * IcebergCdcWriteSchemaTransformProvider}): the {@code snapshots} output schema, the {@code
+ * SnapshotInfo}-to-Row mapping, and the configuration-row conversion.
+ *
+ * <p>These deliberately mirror {@link IcebergWriteSchemaTransformProvider}'s private equivalents —
+ * both providers expose an identical {@code snapshots} output shape — but the append provider is
+ * intentionally left untouched by the CDC change, so it keeps its own copies rather than sharing
+ * this helper (switching it over would modify a long-stable file for zero behavior change).
+ */
+final class IcebergWriteSnapshotOutput {
+
+  private IcebergWriteSnapshotOutput() {}
+
+  /**
+   * The {@code snapshots} output schema: a {@code table} string plus {@link SnapshotInfo}'s fields.
+   */
+  static final Schema OUTPUT_SCHEMA =
+      Schema.builder()
+          .addStringField("table")
+          .addFields(SnapshotInfo.getSchema().getFields())
+          .build();
+
+  /**
+   * Maps a committed {@code (table, SnapshotInfo)} to a {@link Row} matching {@link
+   * #OUTPUT_SCHEMA}.
+   */
+  static class SnapshotToRow extends SimpleFunction<KV<String, SnapshotInfo>, Row> {
+    @Override
+    public Row apply(KV<String, SnapshotInfo> input) {
+      return Row.withSchema(OUTPUT_SCHEMA)
+          .addValue(input.getKey())
+          .addValues(input.getValue().toRow().getValues())
+          .build();
+    }
+  }
+
+  /**
+   * Converts a SchemaTransform {@code Configuration} to its config {@link Row}, sorted
+   * lexicographically and snake_cased to match SchemaTransform config naming conventions.
+   */
+  static <T> Row configurationRow(T configuration, Class<T> configurationClass) {
+    try {
+      return SchemaRegistry.createDefault()
+          .getToRowFunction(configurationClass)
+          .apply(configuration)
+          .sorted()
+          .toSnakeCase();
+    } catch (NoSuchSchemaException e) {
+      throw new RuntimeException(e);
+    }
+  }
+}

sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/PortableIcebergDestinations.java

@@ -27,7 +27,7 @@
 import org.apache.iceberg.FileFormat;
 import org.checkerframework.checker.nullness.qual.Nullable;
 
-class PortableIcebergDestinations implements DynamicDestinations {
+public class PortableIcebergDestinations implements DynamicDestinations {
   private final RowFilter rowFilter;
   private final RowStringInterpolator interpolator;
   private final String fileFormat;