This is an automated email from the ASF dual-hosted git repository.
chamikaramj pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 2a4f4c7fc5e Adds Delta Lake bounded CDC read support to Beam Python
and YAML SDKs (#40085)
2a4f4c7fc5e is described below
commit 2a4f4c7fc5ef0c184fe1b5808724812322689113
Author: Chamikara Jayalath <[email protected]>
AuthorDate: Tue Sep 15 09:12:21 2026 -0700
Adds Delta Lake bounded CDC read support to Beam Python and YAML SDKs
(#40085)
---
.../beam_PostCommit_Java_Delta_IO_Dataflow.json | 2 +-
.../beam_PostCommit_Python_Xlang_IO_Dataflow.json | 2 +-
.../beam_PostCommit_Python_Xlang_IO_Direct.json | 2 +-
.../beam_PostCommit_Yaml_Xlang_Direct.json | 2 +-
.../DeltaCdcReadSchemaTransformProviderTest.java | 161 ++++++++++++++++++
sdks/python/apache_beam/transforms/external.py | 1 +
sdks/python/apache_beam/transforms/managed.py | 12 +-
.../transforms/managed_delta_it_test.py | 189 +++++++++++++++++++++
sdks/python/apache_beam/transforms/managed_test.py | 64 +++++++
sdks/python/apache_beam/yaml/integration_tests.py | 23 +++
sdks/python/apache_beam/yaml/standard_io.yaml | 1 +
sdks/python/apache_beam/yaml/tests/delta_cdc.yaml | 47 +++++
sdks/python/apache_beam/yaml/yaml_io.py | 37 ++++
sdks/python/apache_beam/yaml/yaml_io_test.py | 19 +++
sdks/python/setup.py | 3 +-
15 files changed, 559 insertions(+), 6 deletions(-)
diff --git a/.github/trigger_files/beam_PostCommit_Java_Delta_IO_Dataflow.json
b/.github/trigger_files/beam_PostCommit_Java_Delta_IO_Dataflow.json
index 0ca37f8c8e2..9a4fbff0c0b 100644
--- a/.github/trigger_files/beam_PostCommit_Java_Delta_IO_Dataflow.json
+++ b/.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"
}
diff --git
a/.github/trigger_files/beam_PostCommit_Python_Xlang_IO_Dataflow.json
b/.github/trigger_files/beam_PostCommit_Python_Xlang_IO_Dataflow.json
index 4f9719d7185..e3ce593c117 100644
--- a/.github/trigger_files/beam_PostCommit_Python_Xlang_IO_Dataflow.json
+++ b/.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": 5,
"https://github.com/apache/beam/pull/39990": "removing dead code from
FnApiDoFnRunner"
}
diff --git a/.github/trigger_files/beam_PostCommit_Python_Xlang_IO_Direct.json
b/.github/trigger_files/beam_PostCommit_Python_Xlang_IO_Direct.json
index 1b1ef86e917..8b8cd389b3d 100644
--- a/.github/trigger_files/beam_PostCommit_Python_Xlang_IO_Direct.json
+++ b/.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": 4,
"https://github.com/apache/beam/pull/39990": "removing dead code from
FnApiDoFnRunner"
}
diff --git a/.github/trigger_files/beam_PostCommit_Yaml_Xlang_Direct.json
b/.github/trigger_files/beam_PostCommit_Yaml_Xlang_Direct.json
index 86bf1193abd..8ed972c9f57 100644
--- a/.github/trigger_files/beam_PostCommit_Yaml_Xlang_Direct.json
+++ b/.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
}
diff --git
a/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaCdcReadSchemaTransformProviderTest.java
b/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaCdcReadSchemaTransformProviderTest.java
new file mode 100644
index 00000000000..b82bb97cf6f
--- /dev/null
+++
b/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();
+ }
+}
diff --git a/sdks/python/apache_beam/transforms/external.py
b/sdks/python/apache_beam/transforms/external.py
index 90de7aed24a..33926281074 100644
--- a/sdks/python/apache_beam/transforms/external.py
+++ b/sdks/python/apache_beam/transforms/external.py
@@ -87,6 +87,7 @@ MANAGED_TRANSFORM_URN_TO_JAR_TARGET_MAPPING = {
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,
}
diff --git a/sdks/python/apache_beam/transforms/managed.py
b/sdks/python/apache_beam/transforms/managed.py
index ba4cb38a011..c252b10de5d 100644
--- a/sdks/python/apache_beam/transforms/managed.py
+++ b/sdks/python/apache_beam/transforms/managed.py
@@ -89,8 +89,17 @@ POSTGRES = "postgres"
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__(
diff --git a/sdks/python/apache_beam/transforms/managed_delta_it_test.py
b/sdks/python/apache_beam/transforms/managed_delta_it_test.py
new file mode 100644
index 00000000000..59446a65ab9
--- /dev/null
+++ b/sdks/python/apache_beam/transforms/managed_delta_it_test.py
@@ -0,0 +1,189 @@
+#
+# 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 logging
+import os
+import shutil
+import tempfile
+import unittest
+import uuid
+
+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.io.filesystem import CompressionTypes
+from apache_beam.io.filesystems import FileSystems
+from apache_beam.options.pipeline_options import StandardOptions
+from apache_beam.testing.test_pipeline import TestPipeline
+from apache_beam.testing.util import assert_that
+from apache_beam.testing.util import equal_to
+
+_LOGGER = logging.getLogger(__name__)
+
+# GCS location used to stage the Delta table when the pipeline runs on a remote
+# runner. Beam's integration test suites have read/write access to this bucket.
+_REMOTE_TABLE_ROOT = 'gs://temp-storage-for-end-to-end-tests/managed_delta_it'
+
+
[email protected]_io_java_expansion_service
[email protected](
+ os.environ.get('EXPANSION_JARS'),
+ "EXPANSION_JARS environment var is not provided, "
+ "indicating that jars have not been built")
[email protected](write_deltalake is None, 'deltalake is not installed.')
+class ManagedDeltaIT(unittest.TestCase):
+ # The table is written once and shared by all test methods, since every test
+ # only reads from it.
+ test_pipeline = None
+ args = None
+ # Path handed to the Managed transform; either `local_dir` or `staged_table`.
+ table = None
+ local_dir = None
+ staged_table = None
+
+ @classmethod
+ def setUpClass(cls):
+ cls.test_pipeline = TestPipeline(is_integration_test=True)
+ cls.args = cls.test_pipeline.get_full_options_as_args()
+
+ cls.local_dir = tempfile.mkdtemp()
+ cls._write_table(cls.local_dir)
+
+ # The Delta table is read by the runner's workers, so a local path only
+ # works when those workers run on this machine (e.g. DirectRunner). For
+ # remote runners the table has to be staged somewhere they can reach.
+ if cls._is_remote_runner():
+ cls.staged_table = FileSystems.join(_REMOTE_TABLE_ROOT, uuid.uuid4().hex)
+ _LOGGER.info('Staging Delta table at %s', cls.staged_table)
+ cls._upload_table(cls.local_dir, cls.staged_table)
+ cls.table = cls.staged_table
+ else:
+ cls.table = cls.local_dir
+
+ @classmethod
+ def tearDownClass(cls):
+ if cls.local_dir:
+ shutil.rmtree(cls.local_dir, ignore_errors=True)
+ if cls.staged_table:
+ try:
+ paths = [
+ metadata.path
+ for match_result in FileSystems.match([cls.staged_table + '/**'])
+ for metadata in match_result.metadata_list
+ ]
+ if paths:
+ FileSystems.delete(paths)
+ except Exception: # pylint: disable=broad-except
+ # Not fatal: the staged table is a handful of small files under a
+ # unique prefix in a shared test bucket.
+ _LOGGER.warning(
+ 'Failed to clean up staged Delta table at %s',
+ cls.staged_table,
+ exc_info=True)
+
+ @classmethod
+ def _is_remote_runner(cls):
+ runner = cls.test_pipeline.get_pipeline_options().view_as(
+ StandardOptions).runner or ''
+ return 'dataflow' in runner.lower()
+
+ @staticmethod
+ def _write_table(path):
+ """Writes a two-version Delta table with change data feed enabled."""
+ # Version 0 commit
+ write_deltalake(
+ path,
+ pa.table({"name": ["a", "b"]}),
+ mode="overwrite",
+ configuration={"delta.enableChangeDataFeed": "true"})
+
+ # Version 1 commit
+ write_deltalake(path, pa.table({"name": ["c"]}), mode="append")
+
+ @staticmethod
+ def _upload_table(local_dir, dest_dir):
+ """Copies a local Delta table directory to ``dest_dir``.
+
+ Delta transaction logs reference data files using paths relative to the
+ table root, so a verbatim copy of the directory is itself a valid table.
+ """
+ for root, _, files in os.walk(local_dir):
+ for name in files:
+ local_path = os.path.join(root, name)
+ relative_path = os.path.relpath(local_path, local_dir)
+ dest_path = FileSystems.join(dest_dir, *relative_path.split(os.sep))
+ with open(local_path, 'rb') as src, FileSystems.create(
+ dest_path, compression_type=CompressionTypes.UNCOMPRESSED) as dest:
+ shutil.copyfileobj(src, dest)
+
+ def test_read_delta(self):
+ with beam.Pipeline(argv=self.args) as p:
+ output = (
+ p
+ | beam.managed.Read(beam.managed.DELTA, config={"table": self.table})
+ | beam.Map(lambda row: row.name))
+ assert_that(output, equal_to(["a", "b", "c"]))
+
+ def test_read_delta_cdc_all_versions(self):
+ with beam.Pipeline(argv=self.args) as p:
+ output = (
+ p
+ | beam.managed.Read(
+ beam.managed.DELTA_CDC,
+ config={
+ "table": self.table, "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 beam.Pipeline(argv=self.args) as p:
+ output = (
+ p
+ | beam.managed.Read(
+ beam.managed.DELTA_CDC,
+ config={
+ "table": self.table, "start_version": 1
+ })
+ | beam.Map(lambda row: row.name))
+ assert_that(output, equal_to(["c"]))
+
+ def test_read_delta_cdc_version_range(self):
+ with beam.Pipeline(argv=self.args) as p:
+ output = (
+ p
+ | beam.managed.Read(
+ beam.managed.DELTA_CDC,
+ config={
+ "table": self.table, "start_version": 0, "end_version": 0
+ })
+ | beam.Map(lambda row: row.name))
+ assert_that(output, equal_to(["a", "b"]))
+
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/sdks/python/apache_beam/transforms/managed_test.py
b/sdks/python/apache_beam/transforms/managed_test.py
new file mode 100644
index 00000000000..0eda998e744
--- /dev/null
+++ b/sdks/python/apache_beam/transforms/managed_test.py
@@ -0,0 +1,64 @@
+#
+# 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.
+#
+
+"""Unit tests for the managed transforms module."""
+
+import unittest
+
+import apache_beam as beam
+from apache_beam.portability.common_urns import ManagedTransforms
+from apache_beam.transforms.external import
MANAGED_TRANSFORM_URN_TO_JAR_TARGET_MAPPING
+from apache_beam.transforms.managed import DELTA
+from apache_beam.transforms.managed import DELTA_CDC
+from apache_beam.transforms.managed import Read
+
+
+class ManagedTest(unittest.TestCase):
+ def test_delta_constants(self):
+ self.assertEqual(DELTA, "delta")
+ self.assertEqual(DELTA_CDC, "delta_cdc")
+ self.assertIn("DELTA", beam.managed.__all__)
+ self.assertIn("DELTA_CDC", beam.managed.__all__)
+
+ def test_read_delta_transforms(self):
+ read_delta = Read(DELTA, config={"table": "test_table"})
+ self.assertEqual(read_delta._source, "delta")
+ self.assertEqual(
+ read_delta._underlying_identifier,
+ ManagedTransforms.Urns.DELTA_LAKE_READ.urn)
+
+ read_delta_cdc = Read(DELTA_CDC, config={"table": "test_table"})
+ self.assertEqual(read_delta_cdc._source, "delta_cdc")
+ self.assertEqual(
+ read_delta_cdc._underlying_identifier,
+ ManagedTransforms.Urns.DELTA_LAKE_CDC_READ.urn)
+
+ def test_invalid_source_raises(self):
+ with self.assertRaises(ValueError):
+ Read("unsupported_source", config={})
+
+ def test_expansion_service_resolution(self):
+ self.assertIn(
+ ManagedTransforms.Urns.DELTA_LAKE_READ.urn,
+ MANAGED_TRANSFORM_URN_TO_JAR_TARGET_MAPPING)
+ self.assertIn(
+ ManagedTransforms.Urns.DELTA_LAKE_CDC_READ.urn,
+ MANAGED_TRANSFORM_URN_TO_JAR_TARGET_MAPPING)
+
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/sdks/python/apache_beam/yaml/integration_tests.py
b/sdks/python/apache_beam/yaml/integration_tests.py
index 71a2c3770c9..2ad64c40435 100644
--- a/sdks/python/apache_beam/yaml/integration_tests.py
+++ b/sdks/python/apache_beam/yaml/integration_tests.py
@@ -1137,6 +1137,29 @@ def temp_delta_table():
yield temp_dir
[email protected]
+def temp_delta_cdc_table():
+ try:
+ from deltalake import write_deltalake
+ except ImportError as exn:
+ raise unittest.SkipTest('deltalake is not installed') from exn
+
+ with tempfile.TemporaryDirectory() as temp_dir:
+ # Version 0 commit
+ table_data = pa.table({"name": ["a", "b"]})
+ write_deltalake(
+ temp_dir,
+ table_data,
+ mode="overwrite",
+ configuration={"delta.enableChangeDataFeed": "true"})
+
+ # Version 1 commit
+ table_data_1 = pa.table({"name": ["c"]})
+ write_deltalake(temp_dir, table_data_1, mode="append")
+
+ yield temp_dir
+
+
def replace_recursive(spec, vars):
"""Recursively replaces string placeholders in a spec with values from vars.
diff --git a/sdks/python/apache_beam/yaml/standard_io.yaml
b/sdks/python/apache_beam/yaml/standard_io.yaml
index 173236dc5e9..0ba02d1fdda 100644
--- a/sdks/python/apache_beam/yaml/standard_io.yaml
+++ b/sdks/python/apache_beam/yaml/standard_io.yaml
@@ -118,6 +118,7 @@
'ReadFromMongoDB': 'apache_beam.yaml.yaml_io.read_from_mongodb'
'WriteToMongoDB': 'apache_beam.yaml.yaml_io.write_to_mongodb'
'ReadFromDelta': 'apache_beam.yaml.yaml_io.read_from_delta'
+ 'ReadFromDeltaCDC': 'apache_beam.yaml.yaml_io.read_from_delta_cdc'
'DicomSearch': 'apache_beam.yaml.yaml_io.dicom_search'
# General File Formats
diff --git a/sdks/python/apache_beam/yaml/tests/delta_cdc.yaml
b/sdks/python/apache_beam/yaml/tests/delta_cdc.yaml
new file mode 100644
index 00000000000..319403c6d10
--- /dev/null
+++ b/sdks/python/apache_beam/yaml/tests/delta_cdc.yaml
@@ -0,0 +1,47 @@
+#
+# 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.
+#
+
+fixtures:
+ - name: DELTA_CDC_TABLE
+ type: "apache_beam.yaml.integration_tests.temp_delta_cdc_table"
+
+pipelines:
+ - pipeline:
+ type: chain
+ transforms:
+ - type: ReadFromDeltaCDC
+ config:
+ table: "{DELTA_CDC_TABLE}"
+ start_version: 0
+ - type: AssertEqual
+ config:
+ elements:
+ - {name: "a"}
+ - {name: "b"}
+ - {name: "c"}
+
+ - pipeline:
+ type: chain
+ transforms:
+ - type: ReadFromDeltaCDC
+ config:
+ table: "{DELTA_CDC_TABLE}"
+ start_version: 1
+ - type: AssertEqual
+ config:
+ elements:
+ - {name: "c"}
diff --git a/sdks/python/apache_beam/yaml/yaml_io.py
b/sdks/python/apache_beam/yaml/yaml_io.py
index 47d2f4685e4..b7ce98b3599 100644
--- a/sdks/python/apache_beam/yaml/yaml_io.py
+++ b/sdks/python/apache_beam/yaml/yaml_io.py
@@ -601,6 +601,43 @@ def read_from_delta(
hadoop_config=hadoop_config))
+def read_from_delta_cdc(
+ table: str,
+ start_version: Optional[int] = None,
+ start_timestamp: Optional[str] = None,
+ end_version: Optional[int] = None,
+ end_timestamp: Optional[str] = None,
+ include_metadata_columns: Optional[Iterable[str]] = None,
+ hadoop_config: Optional[Mapping[str, str]] = None,
+):
+ """Reads change records from a Delta Lake table.
+
+ Args:
+ table: Identifier of the Delta Lake table.
+ start_version: Start version of the Delta Lake table to read changes from.
+ Either this or start_timestamp has to be provided.
+ start_timestamp: Start timestamp of the Delta Lake table to read changes
+ from. Should be specified in the ISO 8601 standard. Either this or
+ start_version has to be provided.
+ end_version: End version of the Delta Lake table to read changes up to.
+ end_timestamp: End timestamp of the Delta Lake table to read changes up to.
+ Should be specified in the ISO 8601 standard.
+ include_metadata_columns: Metadata columns to include in the output rows.
+ Supported columns are: _change_type, _commit_version, and
_commit_timestamp.
+ hadoop_config: Properties passed to the Hadoop Configuration.
+ """
+ return beam.managed.Read(
+ "delta_cdc",
+ config=dict(
+ table=table,
+ start_version=start_version,
+ start_timestamp=start_timestamp,
+ end_version=end_version,
+ end_timestamp=end_timestamp,
+ include_metadata_columns=include_metadata_columns,
+ hadoop_config=hadoop_config))
+
+
def write_to_iceberg(
table: str,
catalog_name: Optional[str] = None,
diff --git a/sdks/python/apache_beam/yaml/yaml_io_test.py
b/sdks/python/apache_beam/yaml/yaml_io_test.py
index 6f982be6f5b..7276aa5f072 100644
--- a/sdks/python/apache_beam/yaml/yaml_io_test.py
+++ b/sdks/python/apache_beam/yaml/yaml_io_test.py
@@ -983,6 +983,25 @@ class YamlDicomSearchTest(unittest.TestCase):
'''))
+class YamlDeltaTest(unittest.TestCase):
+ def test_read_from_delta(self):
+ from apache_beam.yaml.yaml_io import read_from_delta
+ transform = read_from_delta(
+ table="my_table", version=5, timestamp="2026-01-01T00:00:00Z")
+ self.assertIsInstance(transform, beam.managed.Read)
+ self.assertEqual(transform._source, "delta")
+
+ def test_read_from_delta_cdc(self):
+ from apache_beam.yaml.yaml_io import read_from_delta_cdc
+ transform = read_from_delta_cdc(
+ table="my_table",
+ start_version=0,
+ end_version=10,
+ include_metadata_columns=["_change_type"])
+ self.assertIsInstance(transform, beam.managed.Read)
+ self.assertEqual(transform._source, "delta_cdc")
+
+
if __name__ == '__main__':
logging.getLogger().setLevel(logging.INFO)
unittest.main()
diff --git a/sdks/python/setup.py b/sdks/python/setup.py
index 36a3c8a9e61..5ea8257a4b4 100644
--- a/sdks/python/setup.py
+++ b/sdks/python/setup.py
@@ -513,7 +513,8 @@ if __name__ == '__main__':
'sqlalchemy-pytds>=1.0.2',
'pg8000>=1.31.5',
"PyMySQL>=1.1.0",
- 'oracledb>=3.1.1'
+ 'oracledb>=3.1.1',
+ 'deltalake>=0.15.0',
],
'gcp': [
'cachetools>=3.1.0,<7',
