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)); + } +}