From 9dfdf71087c6d5f089dd368429a13c6d7c80fa0d Mon Sep 17 00:00:00 2001 From: goutamadwant Date: Sat, 15 Aug 2026 21:15:35 -0700 Subject: [PATCH 1/3] Add Prism integration tests for Delta IO --- .../beam_PreCommit_Java_Delta_IO_Direct.yml | 10 ++ sdks/java/io/delta/build.gradle | 26 ++++ .../beam/sdk/io/delta/DeltaIOPrismIT.java | 141 ++++++++++++++++++ 3 files changed, 177 insertions(+) create mode 100644 sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaIOPrismIT.java diff --git a/.github/workflows/beam_PreCommit_Java_Delta_IO_Direct.yml b/.github/workflows/beam_PreCommit_Java_Delta_IO_Direct.yml index d7624d5ccd36..55abdeea47de 100644 --- a/.github/workflows/beam_PreCommit_Java_Delta_IO_Direct.yml +++ b/.github/workflows/beam_PreCommit_Java_Delta_IO_Direct.yml @@ -105,6 +105,16 @@ jobs: -PtestJavaVersion=17 \ -Pjava17Home=$JAVA_HOME_17_X64 \ --info + - name: run Delta IO Prism integration tests + uses: ./.github/actions/gradle-command-self-hosted-action + with: + gradle-command: :sdks:java:io:delta:prismIntegrationTest + arguments: | + -PdisableSpotlessCheck=true \ + -PdisableCheckStyle=true \ + -PtestJavaVersion=17 \ + -Pjava17Home=$JAVA_HOME_17_X64 \ + --info - name: Archive JUnit Test Results uses: actions/upload-artifact@v7 if: ${{ !success() }} diff --git a/sdks/java/io/delta/build.gradle b/sdks/java/io/delta/build.gradle index a0de04ed8544..663c30487070 100644 --- a/sdks/java/io/delta/build.gradle +++ b/sdks/java/io/delta/build.gradle @@ -79,6 +79,7 @@ dependencies { testImplementation library.java.testcontainers_localstack testImplementation library.java.hadoop_aws testRuntimeOnly project(path: ":runners:google-cloud-dataflow-java") + testRuntimeOnly project(path: ":runners:prism:java") testRuntimeOnly library.java.slf4j_simple } @@ -122,6 +123,7 @@ task dataflowIntegrationTest(type: Test) { include '**/*IT.class' exclude '**/DeltaIOS3IT.class' + exclude '**/DeltaIOPrismIT.class' maxParallelForks 4 classpath = sourceSets.test.runtimeClasspath @@ -133,9 +135,33 @@ task integrationTest(type: Test) { description = "Runs integration tests locally." outputs.upToDateWhen { false } include '**/*IT.class' + exclude '**/DeltaIOPrismIT.class' classpath = sourceSets.test.runtimeClasspath testClassesDirs = sourceSets.test.output.classesDirs systemProperty "org.slf4j.simpleLogger.log.org.apache.hadoop.fs.s3a", "info" systemProperty "org.slf4j.simpleLogger.log.software.amazon.awssdk", "info" } + +project.evaluationDependsOn(":runners:prism") + +task prismIntegrationTest(type: Test) { + group = "Verification" + description = "Runs Delta Lake integration tests with the Prism runner." + + def prismProject = project(":runners:prism") + dependsOn prismProject.tasks.named("build") + def args = [ + "--runner=TestPrismRunner", + "--defaultEnvironmentType=LOOPBACK", + "--enableWebUI=false", + "--prismLocation=${prismProject.property('buildTarget').toString()}", + "--prismLogLevel=warn", + ] + systemProperty "beamTestPipelineOptions", JsonOutput.toJson(args) + + outputs.upToDateWhen { false } + include '**/DeltaIOPrismIT.class' + classpath = sourceSets.test.runtimeClasspath + testClassesDirs = sourceSets.test.output.classesDirs +} diff --git a/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaIOPrismIT.java b/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaIOPrismIT.java new file mode 100644 index 000000000000..29d43fe9b576 --- /dev/null +++ b/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaIOPrismIT.java @@ -0,0 +1,141 @@ +/* + * 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 io.delta.kernel.defaults.engine.DefaultEngine; +import io.delta.kernel.engine.Engine; +import io.delta.kernel.types.IntegerType; +import io.delta.kernel.types.StringType; +import io.delta.kernel.types.StructType; +import java.io.File; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.util.Arrays; +import java.util.List; +import org.apache.avro.generic.GenericRecord; +import org.apache.beam.sdk.extensions.avro.schemas.utils.AvroUtils; +import org.apache.beam.sdk.managed.Managed; +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.values.PCollection; +import org.apache.beam.sdk.values.Row; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap; +import org.apache.hadoop.fs.Path; +import org.apache.parquet.avro.AvroParquetWriter; +import org.apache.parquet.hadoop.ParquetWriter; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +/** Local integration tests for {@link DeltaIO} using the Prism runner. */ +@RunWith(JUnit4.class) +public class DeltaIOPrismIT { + + private static final Schema ROW_SCHEMA = + Schema.builder().addInt32Field("id").addStringField("name").build(); + + private static final List TEST_ROWS = + Arrays.asList( + Row.withSchema(ROW_SCHEMA).addValues(1, "one").build(), + Row.withSchema(ROW_SCHEMA).addValues(2, "two").build(), + Row.withSchema(ROW_SCHEMA).addValues(3, "three").build()); + + @Rule public final TestPipeline readPipeline = TestPipeline.create(); + @Rule public final TemporaryFolder tempFolder = new TemporaryFolder(); + + @Test + public void testReadDeltaFourTable() throws Exception { + File tableDir = tempFolder.newFolder("delta-4-table"); + Engine engine = DefaultEngine.create(new org.apache.hadoop.conf.Configuration()); + StructType deltaSchema = + new StructType().add("id", IntegerType.INTEGER).add("name", StringType.STRING); + + DeltaWriteTestUtils.writeAppendCommit( + engine, tableDir.getAbsolutePath(), 0L, 123456789L, deltaSchema, TEST_ROWS); + + assertManagedRead(tableDir); + } + + @Test + public void testReadDeltaThreeTable() throws Exception { + File tableDir = tempFolder.newFolder("delta-3-table"); + writeDeltaThreeTable(tableDir); + + assertManagedRead(tableDir); + } + + private void assertManagedRead(File tableDir) { + PCollection output = + readPipeline + .apply( + Managed.read(Managed.DELTA_LAKE) + .withConfig(ImmutableMap.of("table", tableDir.getAbsolutePath()))) + .getSinglePCollection(); + + PAssert.that(output).containsInAnyOrder(TEST_ROWS); + readPipeline.run().waitUntilFinish(); + } + + private void writeDeltaThreeTable(File tableDir) throws Exception { + // Keep this fixture self-contained instead of adding a conflicting Delta 3 runtime. The log + // uses the protocol and metadata emitted by Delta 3.0, while the Parquet data is + // version-neutral. + org.apache.avro.Schema avroSchema = AvroUtils.toAvroSchema(ROW_SCHEMA); + List records = + TEST_ROWS.stream() + .map(row -> AvroUtils.toGenericRecord(row, avroSchema)) + .collect(java.util.stream.Collectors.toList()); + + File parquetFile = new File(tableDir, "part-00000.parquet"); + try (ParquetWriter writer = + AvroParquetWriter.builder(new Path(parquetFile.getAbsolutePath())) + .withSchema(avroSchema) + .withConf(new org.apache.hadoop.conf.Configuration()) + .build()) { + for (GenericRecord record : records) { + writer.write(record); + } + } + + File logDir = new File(tableDir, "_delta_log"); + if (!logDir.mkdirs()) { + throw new IllegalStateException("Could not create Delta log directory " + logDir); + } + + String commit = + "{\"commitInfo\":{\"timestamp\":123456789,\"operation\":\"WRITE\"," + + "\"engineInfo\":\"Apache-Spark/3.5.0 Delta-Lake/3.0.0\"}}\n" + + "{\"protocol\":{\"minReaderVersion\":1,\"minWriterVersion\":2}}\n" + + "{\"metaData\":{\"id\":\"delta-3-test\",\"format\":{\"provider\":\"parquet\"," + + "\"options\":{}},\"schemaString\":\"{\\\"type\\\":\\\"struct\\\",\\\"fields\\\":[" + + "{\\\"name\\\":\\\"id\\\",\\\"type\\\":\\\"integer\\\",\\\"nullable\\\":false," + + "\\\"metadata\\\":{}},{\\\"name\\\":\\\"name\\\",\\\"type\\\":\\\"string\\\"," + + "\\\"nullable\\\":false,\\\"metadata\\\":{}}]}\",\"partitionColumns\":[]," + + "\"configuration\":{},\"createdTime\":123456789}}\n" + + "{\"add\":{\"path\":\"part-00000.parquet\",\"partitionValues\":{},\"size\":" + + parquetFile.length() + + ",\"modificationTime\":123456789,\"dataChange\":true}}\n"; + + Files.write( + new File(logDir, "00000000000000000000.json").toPath(), + commit.getBytes(StandardCharsets.UTF_8)); + } +} From 50ddf9b5c291f2cfebd0f43a2bc072be8b01c1d8 Mon Sep 17 00:00:00 2001 From: goutamadwant Date: Thu, 22 Oct 2026 22:43:55 -0700 Subject: [PATCH 2/3] Reuse existing Managed Delta IO tests on Prism --- .../beam_PreCommit_Java_Delta_IO_Direct.json | 2 +- sdks/java/io/delta/build.gradle | 12 +- .../beam/sdk/io/delta/DeltaIOPrismIT.java | 141 ------------------ 3 files changed, 9 insertions(+), 146 deletions(-) delete mode 100644 sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaIOPrismIT.java diff --git a/.github/trigger_files/beam_PreCommit_Java_Delta_IO_Direct.json b/.github/trigger_files/beam_PreCommit_Java_Delta_IO_Direct.json index e3d6056a5de9..b26833333238 100644 --- a/.github/trigger_files/beam_PreCommit_Java_Delta_IO_Direct.json +++ b/.github/trigger_files/beam_PreCommit_Java_Delta_IO_Direct.json @@ -1,4 +1,4 @@ { "comment": "Modify this file in a trivial way to cause this test suite to run", - "modification": 1 + "modification": 2 } diff --git a/sdks/java/io/delta/build.gradle b/sdks/java/io/delta/build.gradle index f4fbad02a7cb..dc2e964ea878 100644 --- a/sdks/java/io/delta/build.gradle +++ b/sdks/java/io/delta/build.gradle @@ -122,7 +122,6 @@ task dataflowIntegrationTest(type: Test) { include '**/*IT.class' exclude '**/DeltaIOS3IT.class' - exclude '**/DeltaIOPrismIT.class' maxParallelForks 4 classpath = sourceSets.test.runtimeClasspath @@ -134,7 +133,6 @@ task integrationTest(type: Test) { description = "Runs integration tests locally." outputs.upToDateWhen { false } include '**/*IT.class' - exclude '**/DeltaIOPrismIT.class' classpath = sourceSets.test.runtimeClasspath testClassesDirs = sourceSets.test.output.classesDirs @@ -146,7 +144,7 @@ project.evaluationDependsOn(":runners:prism") task prismIntegrationTest(type: Test) { group = "Verification" - description = "Runs Delta Lake integration tests with the Prism runner." + description = "Runs local Delta Lake Managed IO tests with the Prism runner." def prismProject = project(":runners:prism") dependsOn prismProject.tasks.named("build") @@ -160,7 +158,13 @@ task prismIntegrationTest(type: Test) { systemProperty "beamTestPipelineOptions", JsonOutput.toJson(args) outputs.upToDateWhen { false } - include '**/DeltaIOPrismIT.class' + include '**/DeltaIOTest.class' + filter { + includeTestsMatching 'org.apache.beam.sdk.io.delta.DeltaIOTest.testManagedDeltaRead' + includeTestsMatching 'org.apache.beam.sdk.io.delta.DeltaIOTest.testManagedDeltaReadWithVersion' + includeTestsMatching 'org.apache.beam.sdk.io.delta.DeltaIOTest.testManagedDeltaReadWithTimestamp' + includeTestsMatching 'org.apache.beam.sdk.io.delta.DeltaIOTest.testReadChangesWithManaged' + } classpath = sourceSets.test.runtimeClasspath testClassesDirs = sourceSets.test.output.classesDirs } diff --git a/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaIOPrismIT.java b/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaIOPrismIT.java deleted file mode 100644 index 29d43fe9b576..000000000000 --- a/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaIOPrismIT.java +++ /dev/null @@ -1,141 +0,0 @@ -/* - * 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 io.delta.kernel.defaults.engine.DefaultEngine; -import io.delta.kernel.engine.Engine; -import io.delta.kernel.types.IntegerType; -import io.delta.kernel.types.StringType; -import io.delta.kernel.types.StructType; -import java.io.File; -import java.nio.charset.StandardCharsets; -import java.nio.file.Files; -import java.util.Arrays; -import java.util.List; -import org.apache.avro.generic.GenericRecord; -import org.apache.beam.sdk.extensions.avro.schemas.utils.AvroUtils; -import org.apache.beam.sdk.managed.Managed; -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.values.PCollection; -import org.apache.beam.sdk.values.Row; -import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap; -import org.apache.hadoop.fs.Path; -import org.apache.parquet.avro.AvroParquetWriter; -import org.apache.parquet.hadoop.ParquetWriter; -import org.junit.Rule; -import org.junit.Test; -import org.junit.rules.TemporaryFolder; -import org.junit.runner.RunWith; -import org.junit.runners.JUnit4; - -/** Local integration tests for {@link DeltaIO} using the Prism runner. */ -@RunWith(JUnit4.class) -public class DeltaIOPrismIT { - - private static final Schema ROW_SCHEMA = - Schema.builder().addInt32Field("id").addStringField("name").build(); - - private static final List TEST_ROWS = - Arrays.asList( - Row.withSchema(ROW_SCHEMA).addValues(1, "one").build(), - Row.withSchema(ROW_SCHEMA).addValues(2, "two").build(), - Row.withSchema(ROW_SCHEMA).addValues(3, "three").build()); - - @Rule public final TestPipeline readPipeline = TestPipeline.create(); - @Rule public final TemporaryFolder tempFolder = new TemporaryFolder(); - - @Test - public void testReadDeltaFourTable() throws Exception { - File tableDir = tempFolder.newFolder("delta-4-table"); - Engine engine = DefaultEngine.create(new org.apache.hadoop.conf.Configuration()); - StructType deltaSchema = - new StructType().add("id", IntegerType.INTEGER).add("name", StringType.STRING); - - DeltaWriteTestUtils.writeAppendCommit( - engine, tableDir.getAbsolutePath(), 0L, 123456789L, deltaSchema, TEST_ROWS); - - assertManagedRead(tableDir); - } - - @Test - public void testReadDeltaThreeTable() throws Exception { - File tableDir = tempFolder.newFolder("delta-3-table"); - writeDeltaThreeTable(tableDir); - - assertManagedRead(tableDir); - } - - private void assertManagedRead(File tableDir) { - PCollection output = - readPipeline - .apply( - Managed.read(Managed.DELTA_LAKE) - .withConfig(ImmutableMap.of("table", tableDir.getAbsolutePath()))) - .getSinglePCollection(); - - PAssert.that(output).containsInAnyOrder(TEST_ROWS); - readPipeline.run().waitUntilFinish(); - } - - private void writeDeltaThreeTable(File tableDir) throws Exception { - // Keep this fixture self-contained instead of adding a conflicting Delta 3 runtime. The log - // uses the protocol and metadata emitted by Delta 3.0, while the Parquet data is - // version-neutral. - org.apache.avro.Schema avroSchema = AvroUtils.toAvroSchema(ROW_SCHEMA); - List records = - TEST_ROWS.stream() - .map(row -> AvroUtils.toGenericRecord(row, avroSchema)) - .collect(java.util.stream.Collectors.toList()); - - File parquetFile = new File(tableDir, "part-00000.parquet"); - try (ParquetWriter writer = - AvroParquetWriter.builder(new Path(parquetFile.getAbsolutePath())) - .withSchema(avroSchema) - .withConf(new org.apache.hadoop.conf.Configuration()) - .build()) { - for (GenericRecord record : records) { - writer.write(record); - } - } - - File logDir = new File(tableDir, "_delta_log"); - if (!logDir.mkdirs()) { - throw new IllegalStateException("Could not create Delta log directory " + logDir); - } - - String commit = - "{\"commitInfo\":{\"timestamp\":123456789,\"operation\":\"WRITE\"," - + "\"engineInfo\":\"Apache-Spark/3.5.0 Delta-Lake/3.0.0\"}}\n" - + "{\"protocol\":{\"minReaderVersion\":1,\"minWriterVersion\":2}}\n" - + "{\"metaData\":{\"id\":\"delta-3-test\",\"format\":{\"provider\":\"parquet\"," - + "\"options\":{}},\"schemaString\":\"{\\\"type\\\":\\\"struct\\\",\\\"fields\\\":[" - + "{\\\"name\\\":\\\"id\\\",\\\"type\\\":\\\"integer\\\",\\\"nullable\\\":false," - + "\\\"metadata\\\":{}},{\\\"name\\\":\\\"name\\\",\\\"type\\\":\\\"string\\\"," - + "\\\"nullable\\\":false,\\\"metadata\\\":{}}]}\",\"partitionColumns\":[]," - + "\"configuration\":{},\"createdTime\":123456789}}\n" - + "{\"add\":{\"path\":\"part-00000.parquet\",\"partitionValues\":{},\"size\":" - + parquetFile.length() - + ",\"modificationTime\":123456789,\"dataChange\":true}}\n"; - - Files.write( - new File(logDir, "00000000000000000000.json").toPath(), - commit.getBytes(StandardCharsets.UTF_8)); - } -} From 56a26a5edb7e232346a69d458028e1a9acf4d18a Mon Sep 17 00:00:00 2001 From: goutamadwant Date: Sun, 25 Oct 2026 14:00:42 -0700 Subject: [PATCH 3/3] Use self-repository action reference for Delta Prism CI --- .github/workflows/beam_PreCommit_Java_Delta_IO_Direct.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/beam_PreCommit_Java_Delta_IO_Direct.yml b/.github/workflows/beam_PreCommit_Java_Delta_IO_Direct.yml index 057dc9525e95..0d316a73c9cb 100644 --- a/.github/workflows/beam_PreCommit_Java_Delta_IO_Direct.yml +++ b/.github/workflows/beam_PreCommit_Java_Delta_IO_Direct.yml @@ -96,7 +96,7 @@ jobs: -Pjava17Home=$JAVA_HOME_17_X64 \ --info - name: run Delta IO Prism integration tests - uses: ./.github/actions/gradle-command-self-hosted-action + uses: $/.github/actions/gradle-command-self-hosted-action with: gradle-command: :sdks:java:io:delta:prismIntegrationTest arguments: |