[Managed Iceberg] unbounded source

.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": 2
+    "modification": 3
 }

CHANGES.md

@@ -65,6 +65,7 @@
 
 * Support for X source added (Java/Python) ([#X](https://github.com/apache/beam/issues/X)).
 * [Java] Use API compatible with both com.google.cloud.bigdataoss:util 2.x and 3.x in BatchLoads ([#34105](https://github.com/apache/beam/pull/34105))
+* [Iceberg] Added new CDC source for batch and streaming, available as `Managed.ICEBERG_CDC` ([#33504](https://github.com/apache/beam/pull/33504))
 
 ## New Features / Improvements
 

contributor-docs/discussion-docs/2025.md

@@ -15,5 +15,9 @@ limitations under the License.
 # List Of Documents Submitted To [email protected] In 2025
 | No. | Author | Subject | Date (UTC) |
 |---|---|---|---|
-| 1 | Danny McCormick | [Beam Python Dependency Extras](https://docs.google.com/document/d/1c84Gc-cZRCfrU8f7kWGsNR2o8oSRjCM-dGHO9KvPWPw) | 2025-01-27 17:50:00 |
-| 2 | Danny McCormick | [How vLLM Model Handler Works (Plus a Summary of Model Memory Management in Beam ML)](https://docs.google.com/document/d/1UB4umrtnp1Eg45fiUB3iLS7kPK3BE6pcf0YRDkA289Q) | 2025-01-31 17:50:00 |
+| 1 | Kenneth Knowles | [Apache Beam Release Acceptance Criteria  - Google Sheets](https://docs.google.com/spreadsheets/d/1qk-N5vjXvbcEk68GjbkSZTR8AGqyNUM-oLFo_ZXBpJw) | 2025-01-13 10:54:22 |
+| 2 | Danny McCormick | [Apache Beam Vendored Dependencies Release Guide](https://s.apache.org/beam-release-vendored-artifacts) | 2025-01-13 15:00:51 |
+| 3 | Danny McCormick | [Beam Python & ML Dependency Extras](https://docs.google.com/document/d/1c84Gc-cZRCfrU8f7kWGsNR2o8oSRjCM-dGHO9KvPWPw) | 2025-01-27 15:33:36 |
+| 4 | Danny McCormick | [How vLLM Model Handler Works (Plus a Summary of Model Memory Management in Beam ML)](https://docs.google.com/document/d/1UB4umrtnp1Eg45fiUB3iLS7kPK3BE6pcf0YRDkA289Q) | 2025-01-31 11:56:59 |
+| 5 | Shunping Huang | [Improve Logging Dependencies in Beam Java SDK](https://docs.google.com/document/d/1IkbiM4m8D-aB3NYI1aErFZHt6M7BQ-8eCULh284Davs) | 2025-02-04 15:13:14 |
+| 6 | Ahmed Abualsaud | [Iceberg Incremental Source design](https://s.apache.org/beam-iceberg-incremental-source) | 2025-03-03 14:52:42 |

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

@@ -74,6 +74,8 @@ message ManagedTransforms {
       "beam:schematransform:org.apache.beam:bigquery_storage_read:v1"];
     BIGQUERY_WRITE = 5 [(org.apache.beam.model.pipeline.v1.beam_urn) =
       "beam:schematransform:org.apache.beam:bigquery_write:v1"];
+    ICEBERG_CDC_READ = 6 [(org.apache.beam.model.pipeline.v1.beam_urn) =
+      "beam:schematransform:org.apache.beam:iceberg_cdc_read:v1"];
   }
 }
 

sdks/java/io/iceberg/build.gradle

@@ -50,6 +50,7 @@ dependencies {
     implementation library.java.slf4j_api
     implementation library.java.joda_time
     implementation "org.apache.parquet:parquet-column:$parquet_version"
+    implementation "org.apache.parquet:parquet-hadoop:$parquet_version"
     implementation "org.apache.orc:orc-core:$orc_version"
     implementation "org.apache.iceberg:iceberg-core:$iceberg_version"
     implementation "org.apache.iceberg:iceberg-api:$iceberg_version"
@@ -88,7 +89,7 @@ dependencies {
     testImplementation library.java.google_api_services_bigquery
 
     testRuntimeOnly library.java.slf4j_jdk14
-    testRuntimeOnly project(path: ":runners:direct-java", configuration: "shadow")
+    testImplementation project(path: ":runners:direct-java", configuration: "shadow")
     testRuntimeOnly project(path: ":runners:google-cloud-dataflow-java")
     hadoopVersions.each {kv ->
         "hadoopVersion$kv.key" "org.apache.hadoop:hadoop-client:$kv.value"

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

@@ -82,7 +82,7 @@ public String apply(FileWriteResult input) {
         .apply(
             "Append metadata updates to tables",
             ParDo.of(new AppendFilesToTablesDoFn(catalogConfig, manifestFilePrefix)))
-        .setCoder(KvCoder.of(StringUtf8Coder.of(), SnapshotInfo.CODER));
+        .setCoder(KvCoder.of(StringUtf8Coder.of(), SnapshotInfo.getCoder()));
   }
 
   private static class AppendFilesToTablesDoFn

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

@@ -0,0 +1,122 @@
+/*
+ * 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 static org.apache.beam.sdk.util.Preconditions.checkStateNotNull;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.ExecutionException;
+import org.apache.beam.sdk.metrics.Counter;
+import org.apache.beam.sdk.metrics.Metrics;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.values.KV;
+import org.apache.iceberg.CombinedScanTask;
+import org.apache.iceberg.DataOperations;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.IncrementalAppendScan;
+import org.apache.iceberg.ScanTaskParser;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.io.CloseableIterable;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Scans the given snapshot and creates multiple {@link ReadTask}s. Each task represents a portion
+ * of a data file that was appended within the snapshot range.
+ */
+class CreateReadTasksDoFn
+    extends DoFn<KV<String, List<SnapshotInfo>>, KV<ReadTaskDescriptor, ReadTask>> {
+  private static final Logger LOG = LoggerFactory.getLogger(CreateReadTasksDoFn.class);
+  private static final Counter totalScanTasks =
+      Metrics.counter(CreateReadTasksDoFn.class, "totalScanTasks");
+  // TODO(ahmedabu98): should we expose a metric that tracks the latest observed snapshot sequence
+  // number?
+
+  private final IcebergScanConfig scanConfig;
+  private final @Nullable String watermarkColumnName;
+
+  CreateReadTasksDoFn(IcebergScanConfig scanConfig) {
+    this.scanConfig = scanConfig;
+    this.watermarkColumnName = scanConfig.getWatermarkColumn();
+  }
+
+  @ProcessElement
+  public void process(
+      @Element KV<String, List<SnapshotInfo>> element,
+      OutputReceiver<KV<ReadTaskDescriptor, ReadTask>> out)
+      throws IOException, ExecutionException {
+    Table table = TableCache.get(element.getKey(), scanConfig.getCatalogConfig().catalog());
+    List<SnapshotInfo> snapshots = element.getValue();
+
+    // scan snapshots individually and assign commit timestamp to files
+    for (SnapshotInfo snapshot : snapshots) {
+      @Nullable Long fromSnapshot = snapshot.getParentId();
+      long toSnapshot = snapshot.getSnapshotId();
+
+      if (!DataOperations.APPEND.equals(snapshot.getOperation())) {
+        LOG.info(
+            "Skipping non-append snapshot of operation '{}'. Sequence number: {}, id: {}",
+            snapshot.getOperation(),
+            snapshot.getSequenceNumber(),
+            snapshot.getSnapshotId());
+      }
+
+      LOG.info("Planning to scan snapshot id range ({}, {}]", fromSnapshot, toSnapshot);
+      IncrementalAppendScan scan = table.newIncrementalAppendScan().toSnapshot(toSnapshot);
+      if (fromSnapshot != null) {
+        scan = scan.fromSnapshotExclusive(fromSnapshot);
+      }
+      if (watermarkColumnName != null) {
+        scan = scan.includeColumnStats(Collections.singletonList(watermarkColumnName));
+      }
+
+      createAndOutputReadTasks(scan, snapshot, out);
+    }
+  }
+
+  private void createAndOutputReadTasks(
+      IncrementalAppendScan scan,
+      SnapshotInfo snapshot,
+      OutputReceiver<KV<ReadTaskDescriptor, ReadTask>> out)
+      throws IOException {
+    try (CloseableIterable<CombinedScanTask> combinedScanTasks = scan.planTasks()) {
+      for (CombinedScanTask combinedScanTask : combinedScanTasks) {
+        // A single DataFile can be broken up into multiple FileScanTasks
+        for (FileScanTask fileScanTask : combinedScanTask.tasks()) {
+          ReadTask task =
+              ReadTask.builder()
+                  .setFileScanTaskJson(ScanTaskParser.toJson(fileScanTask))
+                  .setByteSize(fileScanTask.file().fileSizeInBytes())
+                  .setOperation(snapshot.getOperation())
+                  .setSnapshotTimestampMillis(snapshot.getTimestampMillis())
+                  .build();
+          ReadTaskDescriptor descriptor =
+              ReadTaskDescriptor.builder()
+                  .setTableIdentifierString(checkStateNotNull(snapshot.getTableIdentifierString()))
+                  .build();
+
+          out.output(KV.of(descriptor, task));
+          totalScanTasks.inc();
+        }
+      }
+    }
+  }
+}

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

@@ -51,26 +51,25 @@ public static Builder builder() {
   }
 
   public org.apache.iceberg.catalog.Catalog catalog() {
-    if (cachedCatalog != null) {
-      return cachedCatalog;
+    if (cachedCatalog == null) {
+      String catalogName = getCatalogName();
+      if (catalogName == null) {
+        catalogName = "apache-beam-" + ReleaseInfo.getReleaseInfo().getVersion();
+      }
+      Map<String, String> catalogProps = getCatalogProperties();
+      if (catalogProps == null) {
+        catalogProps = Maps.newHashMap();
+      }
+      Map<String, String> confProps = getConfigProperties();
+      if (confProps == null) {
+        confProps = Maps.newHashMap();
+      }
+      Configuration config = new Configuration();
+      for (Map.Entry<String, String> prop : confProps.entrySet()) {
+        config.set(prop.getKey(), prop.getValue());
+      }
+      cachedCatalog = CatalogUtil.buildIcebergCatalog(catalogName, catalogProps, config);
     }
-    String catalogName = getCatalogName();
-    if (catalogName == null) {
-      catalogName = "apache-beam-" + ReleaseInfo.getReleaseInfo().getVersion();
-    }
-    Map<String, String> catalogProps = getCatalogProperties();
-    if (catalogProps == null) {
-      catalogProps = Maps.newHashMap();
-    }
-    Map<String, String> confProps = getConfigProperties();
-    if (confProps == null) {
-      confProps = Maps.newHashMap();
-    }
-    Configuration config = new Configuration();
-    for (Map.Entry<String, String> prop : confProps.entrySet()) {
-      config.set(prop.getKey(), prop.getValue());
-    }
-    cachedCatalog = CatalogUtil.buildIcebergCatalog(catalogName, catalogProps, config);
     return cachedCatalog;
   }