From 8c8d31578c86d60ae5259f169386c4425c782e96 Mon Sep 17 00:00:00 2001 From: Chamikara Jayalath Date: Wed, 9 Sep 2026 16:40:48 -0700 Subject: [PATCH 1/3] Adds Delta Lake bounded CDC read support to Beam Python and YAML SDKs --- ...eam_PostCommit_Java_Delta_IO_Dataflow.json | 2 +- ...ltaCdcReadSchemaTransformProviderTest.java | 161 ++++++++++++++++++ .../python/apache_beam/transforms/external.py | 1 + sdks/python/apache_beam/transforms/managed.py | 12 +- .../transforms/managed_delta_it_test.py | 107 ++++++++++++ .../apache_beam/transforms/managed_test.py | 64 +++++++ .../apache_beam/yaml/integration_tests.py | 23 +++ sdks/python/apache_beam/yaml/standard_io.yaml | 1 + .../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 +- 12 files changed, 474 insertions(+), 3 deletions(-) create mode 100644 sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaCdcReadSchemaTransformProviderTest.java create mode 100644 sdks/python/apache_beam/transforms/managed_delta_it_test.py create mode 100644 sdks/python/apache_beam/transforms/managed_test.py create mode 100644 sdks/python/apache_beam/yaml/tests/delta_cdc.yaml 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 0ca37f8c8e27..9a4fbff0c0b7 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/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 000000000000..b82bb97cf6fa --- /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 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.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 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 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 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 90de7aed24a6..33926281074d 100644 --- a/sdks/python/apache_beam/transforms/external.py +++ b/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, } diff --git a/sdks/python/apache_beam/transforms/managed.py b/sdks/python/apache_beam/transforms/managed.py index ba4cb38a011e..c252b10de5db 100644 --- a/sdks/python/apache_beam/transforms/managed.py +++ b/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__( 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 000000000000..0bcbb4eb78f4 --- /dev/null +++ b/sdks/python/apache_beam/transforms/managed_delta_it_test.py @@ -0,0 +1,107 @@ +# +# 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 shutil +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.skipIf(write_deltalake is None, 'deltalake is not installed.') +class ManagedDeltaIT(unittest.TestCase): + def setUp(self): + 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() 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 000000000000..0eda998e7447 --- /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 71a2c3770c9a..2ad64c404356 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 +@contextlib.contextmanager +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 173236dc5e94..0ba02d1fddae 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 000000000000..319403c6d108 --- /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 47d2f4685e43..b7ce98b35997 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 6f982be6f5ba..7276aa5f0728 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 @@ def test_dicom_search_without_error_handling_raises(self): ''')) +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 36a3c8a9e617..5ea8257a4b4b 100644 --- a/sdks/python/setup.py +++ b/sdks/python/setup.py @@ -513,7 +513,8 @@ def get_portability_package_data(): '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', From 58378cc2c20fe93bcb77e37bcc5d3c50cc346d8f Mon Sep 17 00:00:00 2001 From: Chamikara Jayalath Date: Thu, 10 Sep 2026 13:10:46 -0700 Subject: [PATCH 2/3] Fix yapf --- .../transforms/managed_delta_it_test.py | 15 ++++++++------- 1 file changed, 8 insertions(+), 7 deletions(-) diff --git a/sdks/python/apache_beam/transforms/managed_delta_it_test.py b/sdks/python/apache_beam/transforms/managed_delta_it_test.py index 0bcbb4eb78f4..c42a7176e613 100644 --- a/sdks/python/apache_beam/transforms/managed_delta_it_test.py +++ b/sdks/python/apache_beam/transforms/managed_delta_it_test.py @@ -63,8 +63,7 @@ def test_read_delta(self): output = ( p | beam.managed.Read( - beam.managed.DELTA, - config={"table": self.temp_dir}) + beam.managed.DELTA, config={"table": self.temp_dir}) | beam.Map(lambda row: row.name)) assert_that(output, equal_to(["a", "b", "c"])) @@ -74,7 +73,9 @@ def test_read_delta_cdc_all_versions(self): p | beam.managed.Read( beam.managed.DELTA_CDC, - config={"table": self.temp_dir, "start_version": 0}) + config={ + "table": self.temp_dir, "start_version": 0 + }) | beam.Map(lambda row: row.name)) assert_that(output, equal_to(["a", "b", "c"])) @@ -84,7 +85,9 @@ def test_read_delta_cdc_from_version_1(self): p | beam.managed.Read( beam.managed.DELTA_CDC, - config={"table": self.temp_dir, "start_version": 1}) + config={ + "table": self.temp_dir, "start_version": 1 + }) | beam.Map(lambda row: row.name)) assert_that(output, equal_to(["c"])) @@ -95,9 +98,7 @@ def test_read_delta_cdc_version_range(self): | beam.managed.Read( beam.managed.DELTA_CDC, config={ - "table": self.temp_dir, - "start_version": 0, - "end_version": 0 + "table": self.temp_dir, "start_version": 0, "end_version": 0 }) | beam.Map(lambda row: row.name)) assert_that(output, equal_to(["a", "b"])) From 2bd41d7a28aa4ab99f8ebe349f0e122982cacf80 Mon Sep 17 00:00:00 2001 From: Chamikara Jayalath Date: Thu, 10 Sep 2026 15:10:55 -0700 Subject: [PATCH 3/3] Updates and triggers tests --- .../beam_PostCommit_Python_Xlang_IO_Dataflow.json | 2 +- .../beam_PostCommit_Python_Xlang_IO_Direct.json | 2 +- .../beam_PostCommit_Yaml_Xlang_Direct.json | 2 +- .../apache_beam/transforms/managed_delta_it_test.py | 11 +++++++++++ 4 files changed, 14 insertions(+), 3 deletions(-) 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 4f9719d7185b..8b8cd389b3d8 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": 4, "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 1b1ef86e9175..4f9719d7185b 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": 3, "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 86bf1193abd9..8ed972c9f579 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/python/apache_beam/transforms/managed_delta_it_test.py b/sdks/python/apache_beam/transforms/managed_delta_it_test.py index c42a7176e613..fb1412b8897d 100644 --- a/sdks/python/apache_beam/transforms/managed_delta_it_test.py +++ b/sdks/python/apache_beam/transforms/managed_delta_it_test.py @@ -17,7 +17,9 @@ """Integration tests for DeltaIO and Delta CDC using Managed Transforms.""" +import os import shutil +import sys import tempfile import unittest @@ -38,9 +40,18 @@ @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