[wip] DeltaLake yaml normalization

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

@@ -0,0 +1,127 @@
+/*
+ * 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.DeltaReadSchemaTransformProvider.Configuration;
+import static org.apache.beam.sdk.io.delta.DeltaReadSchemaTransformProvider.OUTPUT_TAG;
+
+import java.io.File;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+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 DeltaReadSchemaTransformProvider}. */
+@RunWith(JUnit4.class)
+public class DeltaReadSchemaTransformProviderTest {
+
+  @Rule public TestPipeline writePipeline = TestPipeline.create();
+  @Rule public TestPipeline readPipeline = TestPipeline.create();
+  @Rule public TemporaryFolder tempFolder = new TemporaryFolder();
+
+  @Test
+  public void testBuildTransformWithRow() {
+    java.util.Map<String, String> hadoopConfig = new java.util.HashMap<>();
+    hadoopConfig.put("fs.gs.project.id", "test-project");
+
+    Row config =
+        Row.withSchema(new DeltaReadSchemaTransformProvider().configurationSchema())
+            .withFieldValue("table", "/path/to/table")
+            .withFieldValue("version", 5L)
+            .withFieldValue("timestamp", "2026-06-04T12:00:00Z")
+            .withFieldValue("hadoop_config", hadoopConfig)
+            .build();
+
+    new DeltaReadSchemaTransformProvider().from(config);
+  }
+
+  @Test
+  public void testSimpleScan() throws Exception {
+    File tableDir = tempFolder.newFolder("delta-table-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
+    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\":{},\"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 DeltaReadSchemaTransformProvider
+    Configuration readConfig = Configuration.builder().setTable(tableDir.getAbsolutePath()).build();
+
+    PCollection<Row> output =
+        PCollectionRowTuple.empty(readPipeline)
+            .apply(new DeltaReadSchemaTransformProvider().from(readConfig))
+            .get(OUTPUT_TAG);
+
+    PAssert.that(output).containsInAnyOrder(row);
+
+    readPipeline.run().waitUntilFinish();
+  }
+}

sdks/java/io/expansion-service/build.gradle

@@ -97,6 +97,7 @@ dependencies {
 
   runtimeOnly project(":sdks:java:io:datadog")
   runtimeOnly project(":sdks:java:io:mongodb")
+  runtimeOnly project(":sdks:java:io:delta")
   runtimeOnly library.java.kafka_clients
   runtimeOnly library.java.slf4j_jdk14
 

sdks/python/apache_beam/transforms/external.py

@@ -86,6 +86,7 @@
     ManagedTransforms.Urns.MYSQL_WRITE.urn: _GCP_EXPANSION_SERVICE_JAR_TARGET,
     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,
 }
 
 

sdks/python/apache_beam/transforms/managed.py

@@ -88,8 +88,9 @@
 POSTGRES = "postgres"
 MYSQL = "mysql"
 SQL_SERVER = "sqlserver"
+DELTA = "delta"
 
-__all__ = ["ICEBERG", "KAFKA", "BIGQUERY", "Read", "Write"]
+__all__ = ["ICEBERG", "KAFKA", "BIGQUERY", "DELTA", "Read", "Write"]
 
 
 class Read(PTransform):
@@ -102,6 +103,7 @@ class Read(PTransform):
       POSTGRES: ManagedTransforms.Urns.POSTGRES_READ.urn,
       MYSQL: ManagedTransforms.Urns.MYSQL_READ.urn,
       SQL_SERVER: ManagedTransforms.Urns.SQL_SERVER_READ.urn,
+      DELTA: ManagedTransforms.Urns.DELTA_LAKE_READ.urn,
   }
 
   def __init__(

sdks/python/apache_beam/yaml/integration_tests.py

@@ -68,6 +68,8 @@ def get_impl(self):
     None, lambda payload, components, context: BigEndianIntegerCoder())
 
 import psycopg2
+import pyarrow as pa
+import pyarrow.parquet as pq
 import pytds
 import sqlalchemy
 import yaml
@@ -618,6 +620,26 @@ def temp_pubsub_emulator(project_id="apache-beam-testing"):
     yield created_topic_object.name
 
 
+@contextlib.contextmanager
+def temp_delta_table():
+  with tempfile.TemporaryDirectory() as temp_dir:
+    log_dir = os.path.join(temp_dir, "_delta_log")
+    os.makedirs(log_dir, exist_ok=True)
+    table_data = pa.table({"name": ["a", "b", "c"]})
+    parquet_path = os.path.join(temp_dir, "part-00000.parquet")
+    pq.write_table(table_data, parquet_path)
+    file_size = os.path.getsize(parquet_path)
+    commit_content = (
+        '{"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\":{},\"createdAt\":123456789}}\n'
+        f'{{"add":{{"path":"part-00000.parquet","partitionValues":{{}},"size":{file_size},"modificationTime":123456789,"dataChange":true}}}}\n'
+    )
+    commit_file = os.path.join(log_dir, "00000000000000000000.json")
+    with open(commit_file, "w") as f:
+      f.write(commit_content)
+    yield temp_dir
+
+
 def replace_recursive(spec, vars):
   """Recursively replaces string placeholders in a spec with values from vars.
 

sdks/python/apache_beam/yaml/standard_io.yaml

@@ -116,6 +116,7 @@
     'ReadFromTFRecord': 'apache_beam.yaml.yaml_io.read_from_tfrecord'
     'WriteToTFRecord': 'apache_beam.yaml.yaml_io.write_to_tfrecord'
     'WriteToMongoDB': 'apache_beam.yaml.yaml_io.write_to_mongodb'
+    'ReadFromDelta': 'apache_beam.yaml.yaml_io.read_from_delta'
 
 
 # General File Formats
@@ -470,3 +471,4 @@
         'WriteToMongoDB': 'beam:schematransform:org.apache.beam:mongodb_write:v1'
       config:
         gradle_target: 'sdks:java:io:expansion-service:shadowJar'
+

sdks/python/apache_beam/yaml/tests/delta.yaml

@@ -0,0 +1,34 @@
+#
+# 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_TABLE
+    type: "apache_beam.yaml.integration_tests.temp_delta_table"
+
+pipelines:
+  - pipeline:
+      type: chain
+      transforms:
+        - type: ReadFromDelta
+          config:
+            table: "{DELTA_TABLE}"
+        - type: AssertEqual
+          config:
+            elements:
+              - {name: "a"}
+              - {name: "b"}
+              - {name: "c"}

sdks/python/apache_beam/yaml/yaml_io.py

@@ -565,6 +565,29 @@ def read_from_iceberg(
           config_properties=config_properties))
 
 
+def read_from_delta(
+    table: str,
+    version: Optional[int] = None,
+    timestamp: Optional[str] = None,
+    hadoop_config: Optional[Mapping[str, str]] = None,
+):
+  """Reads a Delta Lake table.
+
+  Args:
+    table: Identifier of the Delta Lake table.
+    version: Version of the Delta Lake table to read.
+    timestamp: Timestamp of the Delta Lake table to read.
+    hadoop_config: Hadoop configuration properties.
+  """
+  return beam.managed.Read(
+      "delta",
+      config=dict(
+          table=table,
+          version=version,
+          timestamp=timestamp,
+          hadoop_config=hadoop_config))
+
+
 def write_to_iceberg(
     table: str,
     catalog_name: Optional[str] = None,

sdks/standard_expansion_services.yaml

@@ -52,6 +52,7 @@
     - 'beam:schematransform:org.apache.beam:iceberg_write:v1'
     - 'beam:schematransform:org.apache.beam:iceberg_read:v1'
     - 'beam:schematransform:org.apache.beam:iceberg_cdc_read:v1'
+    - 'beam:schematransform:org.apache.beam:delta_lake_read:v1'
 
 - gradle_target: 'sdks:java:io:messaging-expansion-service:shadowJar'
   destinations: