Adds Delta Lake bounded CDC read support to Beam Python and YAML SDKs

.github/trigger_files/beam_PostCommit_Java_Delta_IO_Dataflow.json

@@ -1,5 +1,5 @@
 {
   "comment": "Modify this file in a trivial way to cause this test suite to run.",
-  "modification": 1,
+  "modification": 2,
   "https://github.com/apache/beam/pull/39990": "removing dead code from FnApiDoFnRunner"
 }

.github/trigger_files/beam_PostCommit_Python_Xlang_IO_Dataflow.json

@@ -1,5 +1,5 @@
 {
   "comment": "Modify this file in a trivial way to cause this test suite to run",
-  "modification": 3,
+  "modification": 4,
   "https://github.com/apache/beam/pull/39990": "removing dead code from FnApiDoFnRunner"
 }

.github/trigger_files/beam_PostCommit_Python_Xlang_IO_Direct.json

@@ -1,5 +1,5 @@
 {
   "comment": "Modify this file in a trivial way to cause this test suite to run",
-  "modification": 2,
+  "modification": 3,
   "https://github.com/apache/beam/pull/39990": "removing dead code from FnApiDoFnRunner"
 }

.github/trigger_files/beam_PostCommit_Yaml_Xlang_Direct.json

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

sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaCdcReadSchemaTransformProviderTest.java

@@ -0,0 +1,161 @@
+/*
+ * 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.delta;
+
+import static org.apache.beam.sdk.io.delta.DeltaCdcReadSchemaTransformProvider.Configuration;
+import static org.apache.beam.sdk.io.delta.DeltaCdcReadSchemaTransformProvider.OUTPUT_TAG;
+
+import io.delta.kernel.defaults.engine.DefaultEngine;
+import io.delta.kernel.engine.Engine;
+import java.io.File;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.beam.sdk.extensions.avro.coders.AvroCoder;
+import org.apache.beam.sdk.extensions.avro.schemas.utils.AvroUtils;
+import org.apache.beam.sdk.io.Compression;
+import org.apache.beam.sdk.io.FileIO;
+import org.apache.beam.sdk.io.parquet.ParquetIO;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.transforms.windowing.PaneInfo;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PCollectionRowTuple;
+import org.apache.beam.sdk.values.Row;
+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 DeltaCdcReadSchemaTransformProvider}. */
+@RunWith(JUnit4.class)
+public class DeltaCdcReadSchemaTransformProviderTest {
+
+  @Rule public TestPipeline writePipeline = TestPipeline.create();
+  @Rule public TestPipeline readPipeline = TestPipeline.create();
+  @Rule public TemporaryFolder tempFolder = new TemporaryFolder();
+
+  @Test
+  public void testBuildTransformWithRow() {
+    Map<String, String> hadoopConfig = new HashMap<>();
+    hadoopConfig.put("fs.gs.project.id", "test-project");
+
+    Row config =
+        Row.withSchema(new DeltaCdcReadSchemaTransformProvider().configurationSchema())
+            .withFieldValue("table", "/path/to/table")
+            .withFieldValue("start_version", 0L)
+            .withFieldValue("end_version", 5L)
+            .withFieldValue("hadoop_config", hadoopConfig)
+            .withFieldValue("include_metadata_columns", Arrays.asList(DeltaIO.CHANGE_TYPE_COLUMN))
+            .build();
+
+    new DeltaCdcReadSchemaTransformProvider().from(config);
+  }
+
+  @Test
+  public void testSimpleScan() throws Exception {
+    File tableDir = tempFolder.newFolder("delta-table-cdc-simple");
+
+    // 1. Write a Parquet file using Beam
+    Schema schema = Schema.builder().addField("name", Schema.FieldType.STRING).build();
+    Row row = Row.withSchema(schema).addValues("test-name").build();
+
+    org.apache.avro.Schema avroSchema = AvroUtils.toAvroSchema(schema);
+    GenericRecord record = AvroUtils.toGenericRecord(row, avroSchema);
+
+    writePipeline
+        .apply("Create Input", Create.of(record).withCoder(AvroCoder.of(avroSchema)))
+        .apply(
+            "Write Parquet",
+            FileIO.<GenericRecord>write()
+                .via(ParquetIO.sink(avroSchema))
+                .to(tableDir.getAbsolutePath() + "/")
+                .withNaming(
+                    (BoundedWindow window,
+                        PaneInfo paneInfo,
+                        int numShards,
+                        int shardIndex,
+                        Compression compression) -> "part-00000.parquet"));
+
+    writePipeline.run().waitUntilFinish();
+
+    File parquetFile = new File(tableDir, "part-00000.parquet");
+    byte[] fileBytes = Files.readAllBytes(parquetFile.toPath());
+
+    // 2. Create the Delta log with CDF enabled
+    File logDir = new File(tableDir, "_delta_log");
+    logDir.mkdirs();
+    File commitFile = new File(logDir, "00000000000000000000.json");
+
+    String commitContent =
+        "{\"protocol\":{\"minReaderVersion\":1,\"minWriterVersion\":2}}\n"
+            + "{\"metaData\":{\"id\":\"test-id\",\"format\":{\"provider\":\"parquet\",\"options\":{}},\"schemaString\":\"{\\\"type\\\":\\\"struct\\\",\\\"fields\\\":[{\\\"name\\\":\\\"name\\\",\\\"type\\\":\\\"string\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}}]}\",\"partitionColumns\":[],\"configuration\":{\"delta.enableChangeDataFeed\":\"true\"},\"createdAt\":123456789}}\n"
+            + "{\"add\":{\"path\":\"part-00000.parquet\",\"partitionValues\":{},\"size\":"
+            + fileBytes.length
+            + ",\"modificationTime\":123456789,\"dataChange\":true}}";
+
+    Files.write(commitFile.toPath(), commitContent.getBytes(StandardCharsets.UTF_8));
+
+    // 3. Read it using DeltaCdcReadSchemaTransformProvider
+    Configuration readConfig =
+        Configuration.builder().setTable(tableDir.getAbsolutePath()).setStartVersion(0L).build();
+
+    PCollection<Row> output =
+        PCollectionRowTuple.empty(readPipeline)
+            .apply(new DeltaCdcReadSchemaTransformProvider().from(readConfig))
+            .get(OUTPUT_TAG);
+
+    PAssert.that(output).containsInAnyOrder(row);
+
+    readPipeline.run().waitUntilFinish();
+  }
+
+  @Test
+  public void testReadWithStartVersion() throws Exception {
+    File tableDir = tempFolder.newFolder("delta-table-cdc-version");
+    Engine engine = DefaultEngine.create(new org.apache.hadoop.conf.Configuration());
+
+    List<Row> rows = DeltaWriteTestUtils.setupTwoVersionTable(engine, tableDir.getAbsolutePath());
+    Row row1 = rows.get(0);
+    Row row2 = rows.get(1);
+
+    Configuration readConfig =
+        Configuration.builder()
+            .setTable(tableDir.getAbsolutePath())
+            .setStartVersion(0L)
+            .setEndVersion(0L)
+            .build();
+
+    PCollection<Row> output =
+        PCollectionRowTuple.empty(readPipeline)
+            .apply(new DeltaCdcReadSchemaTransformProvider().from(readConfig))
+            .get(OUTPUT_TAG);
+
+    PAssert.that(output).containsInAnyOrder(row1, row2);
+
+    readPipeline.run().waitUntilFinish();
+  }
+}

sdks/python/apache_beam/transforms/external.py

@@ -87,6 +87,7 @@
     ManagedTransforms.Urns.SQL_SERVER_READ.urn: _GCP_EXPANSION_SERVICE_JAR_TARGET,  # pylint: disable=line-too-long
     ManagedTransforms.Urns.SQL_SERVER_WRITE.urn: _GCP_EXPANSION_SERVICE_JAR_TARGET,  # pylint: disable=line-too-long
     ManagedTransforms.Urns.DELTA_LAKE_READ.urn: _IO_EXPANSION_SERVICE_JAR_TARGET,
+    ManagedTransforms.Urns.DELTA_LAKE_CDC_READ.urn: _IO_EXPANSION_SERVICE_JAR_TARGET,
 }
 
 

sdks/python/apache_beam/transforms/managed.py

@@ -89,8 +89,17 @@
 MYSQL = "mysql"
 SQL_SERVER = "sqlserver"
 DELTA = "delta"
+DELTA_CDC = "delta_cdc"
 
-__all__ = ["ICEBERG", "KAFKA", "BIGQUERY", "DELTA", "Read", "Write"]
+__all__ = [
+    "ICEBERG",
+    "KAFKA",
+    "BIGQUERY",
+    "DELTA",
+    "DELTA_CDC",
+    "Read",
+    "Write",
+]
 
 
 class Read(PTransform):
@@ -104,6 +113,7 @@ class Read(PTransform):
       MYSQL: ManagedTransforms.Urns.MYSQL_READ.urn,
       SQL_SERVER: ManagedTransforms.Urns.SQL_SERVER_READ.urn,
       DELTA: ManagedTransforms.Urns.DELTA_LAKE_READ.urn,
+      DELTA_CDC: ManagedTransforms.Urns.DELTA_LAKE_CDC_READ.urn,
   }
 
   def __init__(

sdks/python/apache_beam/transforms/managed_delta_it_test.py

@@ -0,0 +1,119 @@
+#
+# 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.
+#
+
+"""Integration tests for DeltaIO and Delta CDC using Managed Transforms."""
+
+import os
+import shutil
+import sys
+import tempfile
+import unittest
+
+import pyarrow as pa
+import pytest
+
+# pylint: disable=wrong-import-order, wrong-import-position, ungrouped-imports
+try:
+  from deltalake import write_deltalake
+except ImportError:
+  write_deltalake = None
+# pylint: enable=wrong-import-order, wrong-import-position, ungrouped-imports
+
+import apache_beam as beam
+from apache_beam.testing.test_pipeline import TestPipeline
+from apache_beam.testing.util import assert_that
+from apache_beam.testing.util import equal_to
+
+
+@pytest.mark.uses_io_java_expansion_service
+@unittest.skipUnless(
+    os.environ.get('EXPANSION_JARS'),
+    "EXPANSION_JARS environment var is not provided, "
+    "indicating that jars have not been built")
+@unittest.skipIf(write_deltalake is None, 'deltalake is not installed.')
+class ManagedDeltaIT(unittest.TestCase):
+  def setUp(self):
+    if any('DataflowRunner' in arg for arg in sys.argv):
+      self.skipTest(
+          'ManagedDeltaIT only supports direct runner execution with '
+          'local file paths.')
+
+    self.temp_dir = tempfile.mkdtemp()
+
+    # Version 0 commit
+    table_data_0 = pa.table({"name": ["a", "b"]})
+    write_deltalake(
+        self.temp_dir,
+        table_data_0,
+        mode="overwrite",
+        configuration={"delta.enableChangeDataFeed": "true"})
+
+    # Version 1 commit
+    table_data_1 = pa.table({"name": ["c"]})
+    write_deltalake(self.temp_dir, table_data_1, mode="append")
+
+  def tearDown(self):
+    shutil.rmtree(self.temp_dir, ignore_errors=True)
+
+  def test_read_delta(self):
+    with TestPipeline() as p:
+      output = (
+          p
+          | beam.managed.Read(
+              beam.managed.DELTA, config={"table": self.temp_dir})
+          | beam.Map(lambda row: row.name))
+      assert_that(output, equal_to(["a", "b", "c"]))
+
+  def test_read_delta_cdc_all_versions(self):
+    with TestPipeline() as p:
+      output = (
+          p
+          | beam.managed.Read(
+              beam.managed.DELTA_CDC,
+              config={
+                  "table": self.temp_dir, "start_version": 0
+              })
+          | beam.Map(lambda row: row.name))
+      assert_that(output, equal_to(["a", "b", "c"]))
+
+  def test_read_delta_cdc_from_version_1(self):
+    with TestPipeline() as p:
+      output = (
+          p
+          | beam.managed.Read(
+              beam.managed.DELTA_CDC,
+              config={
+                  "table": self.temp_dir, "start_version": 1
+              })
+          | beam.Map(lambda row: row.name))
+      assert_that(output, equal_to(["c"]))
+
+  def test_read_delta_cdc_version_range(self):
+    with TestPipeline() as p:
+      output = (
+          p
+          | beam.managed.Read(
+              beam.managed.DELTA_CDC,
+              config={
+                  "table": self.temp_dir, "start_version": 0, "end_version": 0
+              })
+          | beam.Map(lambda row: row.name))
+      assert_that(output, equal_to(["a", "b"]))
+
+
+if __name__ == '__main__':
+  unittest.main()