[Iceberg CDC] Finish wiring CDC source together and add external API

.github/trigger_files/IO_Iceberg_Integration_Tests.json

@@ -1,4 +1,4 @@
 {
     "comment": "Modify this file in a trivial way to cause this test suite to run.",
-    "modification": 3
+    "modification": 1
 }

.github/trigger_files/IO_Iceberg_Integration_Tests_Dataflow.json

@@ -1,4 +1,4 @@
 {
   "comment": "Modify this file in a trivial way to cause this test suite to run.",
-  "modification": 2
+  "modification": 1
 }

CHANGES.md

@@ -68,6 +68,7 @@
 * Support for X source added (Java/Python) ([#X](https://github.com/apache/beam/issues/X)).
 * Add ArrowFlight IO (Java) ([#20116](https://github.com/apache/beam/issues/20116)).
 * (Python) JmsIO (IBM MQ, ActiveMQ, and other providers) is now supported in Python via cross-language ([#30716](https://github.com/apache/beam/issues/30716)).
+* Added a full Iceberg batch and streaming changelog source (CDC) ([#38831](https://github.com/apache/beam/issues/38831))
 
 ## New Features / Improvements
 

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

@@ -117,7 +117,10 @@ public PCollectionRowTuple expand(PCollectionRowTuple input) {
               .streaming(configuration.getStreaming())
               .keeping(configuration.getKeep())
               .dropping(configuration.getDrop())
-              .withFilter(configuration.getFilter());
+              .withFilter(configuration.getFilter())
+              .withWatermarkColumn(configuration.getWatermarkColumn())
+              .withWatermarkColumnTimeUnit(configuration.getWatermarkColumnTimeUnit())
+              .withMetadataColumns(configuration.getIncludeMetadataColumns());
 
       @Nullable Integer pollIntervalSeconds = configuration.getPollIntervalSeconds();
       if (pollIntervalSeconds != null) {
@@ -193,6 +196,26 @@ static Builder builder() {
         "A subset of column names to exclude from reading. If null or empty, all columns will be read.")
     abstract @Nullable List<String> getDrop();
 
+    @SchemaFieldDescription(
+        "Column used to derive the source's output watermark. "
+            + "Must be an existing, required, top-level column of type 'long' or 'timestamp'. "
+            + "If not set, the watermark advances according to snapshot commit timestamp.")
+    abstract @Nullable String getWatermarkColumn();
+
+    @SchemaFieldDescription(
+        "Time unit used to interpret watermark column of type LONG. One of NANOSECONDS, MICROSECONDS, "
+            + "MILLISECONDS, SECONDS, MINUTES, HOURS, DAYS. Defaults to MICROSECONDS.")
+    abstract @Nullable String getWatermarkColumnTimeUnit();
+
+    @SchemaFieldDescription(
+        "List of top-level metadata columns to include with CDC output rows. Supported columns: \n"
+            + "- `_change_type`\n"
+            + "- `_row_id`\n"
+            + "- `_last_updated_sequence_number`\n"
+            + "- `_commit_snapshot_id`\n"
+            + "- `_commit_snapshot_sequence_number`\n")
+    abstract @Nullable List<String> getIncludeMetadataColumns();
+
     @AutoValue.Builder
     abstract static class Builder {
       abstract Builder setTable(String table);
@@ -223,6 +246,12 @@ abstract static class Builder {
 
       abstract Builder setFilter(String filter);
 
+      abstract Builder setWatermarkColumn(String watermarkColumn);
+
+      abstract Builder setWatermarkColumnTimeUnit(String timeUnit);
+
+      abstract Builder setIncludeMetadataColumns(List<String> metadataColumns);
+
       abstract Configuration build();
     }
 

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

@@ -25,6 +25,7 @@
 import java.util.Map;
 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.options.StreamingOptions;
 import org.apache.beam.sdk.schemas.Schema;
 import org.apache.beam.sdk.transforms.PTransform;
@@ -33,6 +34,7 @@
 import org.apache.beam.sdk.values.Row;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Predicates;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList;
 import org.apache.iceberg.DistributionMode;
 import org.apache.iceberg.Table;
 import org.apache.iceberg.catalog.Catalog;
@@ -576,6 +578,7 @@ public static ReadRows readRows(IcebergCatalogConfig catalogConfig) {
     return new AutoValue_IcebergIO_ReadRows.Builder()
         .setCatalogConfig(catalogConfig)
         .setUseCdc(false)
+        .setMetadataColumns(ImmutableList.of())
         .build();
   }
 
@@ -612,6 +615,12 @@ public enum StartingStrategy {
 
     abstract @Nullable String getFilter();
 
+    abstract @Nullable String getWatermarkColumn();
+
+    abstract @Nullable String getWatermarkColumnTimeUnit();
+
+    abstract List<String> getMetadataColumns();
+
     abstract Builder toBuilder();
 
     @AutoValue.Builder
@@ -642,6 +651,12 @@ abstract static class Builder {
 
       abstract Builder setFilter(@Nullable String filter);
 
+      abstract Builder setWatermarkColumn(@Nullable String watermarkColumn);
+
+      abstract Builder setWatermarkColumnTimeUnit(@Nullable String timeUnit);
+
+      abstract Builder setMetadataColumns(List<String> metadataColumns);
+
       abstract ReadRows build();
     }
 
@@ -693,6 +708,31 @@ public ReadRows withFilter(@Nullable String filter) {
       return toBuilder().setFilter(filter).build();
     }
 
+    public ReadRows withWatermarkColumn(@Nullable String watermarkColumn) {
+      return toBuilder().setWatermarkColumn(watermarkColumn).build();
+    }
+
+    public ReadRows withWatermarkColumnTimeUnit(@Nullable String timeUnit) {
+      return toBuilder().setWatermarkColumnTimeUnit(timeUnit).build();
+    }
+
+    /**
+     * Appends top-level metadata columns to CDC output rows.
+     *
+     * <p>Supported values are {@code _change_type}, {@code _commit_snapshot_id}, {@code
+     * _commit_snapshot_sequence_number}, {@code _row_id}, and {@code
+     * _last_updated_sequence_number}. The row metadata columns are read from Iceberg data files and
+     * require a row-lineage table. The changelog metadata columns come from the emitted change kind
+     * and snapshot context and are appended when final Beam rows are emitted.
+     *
+     * <p>This option is only valid {@link #withCdc()}.
+     */
+    public ReadRows withMetadataColumns(@Nullable List<String> metadataColumns) {
+      return toBuilder()
+          .setMetadataColumns(metadataColumns == null ? ImmutableList.of() : metadataColumns)
+          .build();
+    }
+
     @Override
     public PCollection<Row> expand(PBegin input) {
       TableIdentifier tableId =
@@ -728,12 +768,15 @@ public PCollection<Row> expand(PBegin input) {
               .setKeepFields(getKeep())
               .setDropFields(getDrop())
               .setFilterString(getFilter())
+              .setWatermarkColumn(getWatermarkColumn())
+              .setWatermarkColumnTimeUnit(getWatermarkColumnTimeUnit())
+              .setMetadataColumns(getMetadataColumns())
               .build();
       scanConfig.validate(table);
 
       PTransform<PBegin, PCollection<Row>> source =
           getUseCdc()
-              ? new IncrementalScanSource(scanConfig)
+              ? new IncrementalChangelogSource(scanConfig)
               : Read.from(new ScanSource(scanConfig));
 
       return input.apply(source);

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

@@ -271,9 +271,6 @@ public Expression getFilter() {
   @Pure
   public abstract @Nullable String getWatermarkColumnTimeUnit();
 
-  @Pure
-  public abstract @Nullable Duration getMaxSnapshotDiscoveryDelay();
-
   @Pure
   public abstract List<String> getMetadataColumns();
 
@@ -371,8 +368,6 @@ public abstract Builder setUpdateCompatibilityVersion(
 
     public abstract Builder setWatermarkColumnTimeUnit(@Nullable String timeUnit);
 
-    public abstract Builder setMaxSnapshotDiscoveryDelay(@Nullable Duration delay);
-
     public abstract Builder setMetadataColumns(List<String> metadataColumns);
 
     public abstract IcebergScanConfig build();