From 249dbbe3f4f6de8c6bc3dd945a6534fe5d228d0a Mon Sep 17 00:00:00 2001 From: Chamikara Jayalath Date: Fri, 12 Jun 2026 14:38:41 -0700 Subject: [PATCH 1/5] Adds a Dataflow integration test suite for Delta Lake --- ...eam_PostCommit_Java_Delta_IO_Dataflow.json | 4 + ...beam_PostCommit_Java_Delta_IO_Dataflow.yml | 78 ++++++++ sdks/java/io/delta/build.gradle | 38 ++++ .../apache/beam/sdk/io/delta/DeltaIOIT.java | 174 ++++++++++++++++++ 4 files changed, 294 insertions(+) create mode 100644 .github/trigger_files/beam_PostCommit_Java_Delta_IO_Dataflow.json create mode 100644 .github/workflows/beam_PostCommit_Java_Delta_IO_Dataflow.yml create mode 100644 sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaIOIT.java diff --git a/.github/trigger_files/beam_PostCommit_Java_Delta_IO_Dataflow.json b/.github/trigger_files/beam_PostCommit_Java_Delta_IO_Dataflow.json new file mode 100644 index 000000000000..5abe02fc09c7 --- /dev/null +++ b/.github/trigger_files/beam_PostCommit_Java_Delta_IO_Dataflow.json @@ -0,0 +1,4 @@ +{ + "comment": "Modify this file in a trivial way to cause this test suite to run.", + "modification": 1 +} diff --git a/.github/workflows/beam_PostCommit_Java_Delta_IO_Dataflow.yml b/.github/workflows/beam_PostCommit_Java_Delta_IO_Dataflow.yml new file mode 100644 index 000000000000..604be67d90fa --- /dev/null +++ b/.github/workflows/beam_PostCommit_Java_Delta_IO_Dataflow.yml @@ -0,0 +1,78 @@ +# 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. + +name: PostCommit Java Delta IO Dataflow + +on: + schedule: + - cron: '30 4/6 * * *' + pull_request_target: + paths: [ 'release/trigger_all_tests.json', '.github/trigger_files/beam_PostCommit_Java_Delta_IO_Dataflow.json' ] + workflow_dispatch: + +# Setting explicit permissions for the action to avoid the default permissions which are `write-all` in case of pull_request_target event +permissions: + actions: write + pull-requests: write + checks: write + contents: read + deployments: read + id-token: none + issues: write + discussions: read + packages: read + pages: read + repository-projects: read + security-events: read + statuses: read + +# This allows a subsequently queued workflow run to interrupt previous runs +concurrency: + group: '${{ github.workflow }} @ ${{ github.event.pull_request.number || github.sha || github.head_ref || github.ref }}-${{ github.event.schedule || github.event.comment.id || github.event.sender.login }}' + cancel-in-progress: true + +env: + DEVELOCITY_ACCESS_KEY: ${{ secrets.DEVELOCITY_ACCESS_KEY }} + GRADLE_ENTERPRISE_CACHE_USERNAME: ${{ secrets.GE_CACHE_USERNAME }} + GRADLE_ENTERPRISE_CACHE_PASSWORD: ${{ secrets.GE_CACHE_PASSWORD }} + +jobs: + beam_PostCommit_Java_Delta_IO_Dataflow: + if: | + github.event_name == 'pull_request_target' || + github.event_name == 'workflow_dispatch' || + (github.event_name == 'schedule' && github.repository == 'apache/beam') || + github.event.comment.body == 'Run PostCommit Java Delta IO Dataflow' + runs-on: [self-hosted, ubuntu-24.04, main] + timeout-minutes: 120 + name: ${{ matrix.job_name }} (${{ matrix.job_phrase }}) + strategy: + matrix: + job_name: ["beam_PostCommit_Java_Delta_IO_Dataflow"] + job_phrase: ["Run PostCommit Java Delta IO Dataflow"] + steps: + - uses: actions/checkout@v6 + - name: Setup repository + uses: ./.github/actions/setup-action + with: + comment_phrase: ${{ matrix.job_phrase }} + github_token: ${{ secrets.GITHUB_TOKEN }} + github_job: ${{ matrix.job_name }} (${{ matrix.job_phrase }}) + - name: Setup environment + uses: ./.github/actions/setup-environment-action + - name: Run PostCommit Java Delta IO Dataflow + uses: ./.github/actions/gradle-command-self-hosted-action + with: + gradle-command: :sdks:java:io:delta:dataflowIntegrationTest --info diff --git a/sdks/java/io/delta/build.gradle b/sdks/java/io/delta/build.gradle index 57b1cd8ad876..5ee5442ecd18 100644 --- a/sdks/java/io/delta/build.gradle +++ b/sdks/java/io/delta/build.gradle @@ -16,6 +16,8 @@ * limitations under the License. */ +import groovy.json.JsonOutput + plugins { id 'org.apache.beam.module' } applyJavaNature( automaticModuleName: 'org.apache.beam.sdk.io.delta', @@ -68,6 +70,10 @@ dependencies { testImplementation project(":sdks:java:managed") testRuntimeOnly "org.yaml:snakeyaml:2.0" testImplementation project(path: ":runners:direct-java", configuration: "shadow") + testImplementation project(":sdks:java:io:google-cloud-platform") + testImplementation project(":sdks:java:extensions:google-cloud-platform-core") + testImplementation library.java.google_cloud_storage + testRuntimeOnly project(path: ":runners:google-cloud-dataflow-java") } configurations.all { @@ -82,3 +88,35 @@ configurations.all { resolutionStrategy.force "com.google.cloud.bigdataoss:gcsio:$bigdataoss_gcs_connector_version" resolutionStrategy.force "com.google.cloud.bigdataoss:util:$bigdataoss_gcs_connector_version" } + +def gcpProject = project.findProperty('gcpProject') ?: 'apache-beam-testing' +def gcpTempLocation = project.findProperty('gcpTempLocation') ?: 'gs://apache-beam-testing-delta-lake/temp' + +// Used by Dataflow integration tests. +task dataflowIntegrationTest(type: Test) { + group = "Verification" + evaluationDependsOn(":runners:google-cloud-dataflow-java") + dependsOn ":runners:google-cloud-dataflow-java:buildAndPushDockerJavaContainer" + finalizedBy ":runners:google-cloud-dataflow-java:cleanUpDockerJavaImages" + def dockerJavaImageName = project.project(':runners:google-cloud-dataflow-java').ext.dockerJavaImageName + + def args = [ + "--runner=DataflowRunner", + "--region=us-central1", + "--project=${gcpProject}", + "--tempLocation=${gcpTempLocation}", + "--tempRoot=${gcpTempLocation}", + "--sdkContainerImage=${dockerJavaImageName}", + "--experiments=use_runner_v2,use_staged_dataflow_worker_jar" + ] + systemProperty "beamTestPipelineOptions", JsonOutput.toJson(args) + + // Disable Gradle cache: these ITs interact with live service that should always be considered "out of date" + outputs.upToDateWhen { false } + + include '**/*IT.class' + + maxParallelForks 4 + 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/DeltaIOIT.java b/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaIOIT.java new file mode 100644 index 000000000000..36df75a05c7d --- /dev/null +++ b/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaIOIT.java @@ -0,0 +1,174 @@ +/* + * 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.junit.Assert.assertNotNull; + +import com.google.cloud.storage.Blob; +import com.google.cloud.storage.BlobId; +import com.google.cloud.storage.BlobInfo; +import com.google.cloud.storage.Storage; +import com.google.cloud.storage.StorageOptions; +import java.nio.charset.StandardCharsets; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; +import java.util.stream.IntStream; +import org.apache.avro.generic.GenericRecord; +import org.apache.beam.runners.direct.DirectOptions; +import org.apache.beam.runners.direct.DirectRunner; +import org.apache.beam.sdk.Pipeline; +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.managed.Managed; +import org.apache.beam.sdk.options.PipelineOptionsFactory; +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.MapElements; +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.Row; +import org.apache.beam.sdk.values.TypeDescriptor; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap; +import org.junit.After; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TestName; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** Integration tests for {@link DeltaIO}. */ +@RunWith(JUnit4.class) +public class DeltaIOIT { + private static final Logger LOG = LoggerFactory.getLogger(DeltaIOIT.class); + + private static final String BUCKET = "apache-beam-testing-delta-lake"; + + @Rule public final TestPipeline readPipeline = TestPipeline.create(); + @Rule public final TestName testName = new TestName(); + + private String repoPath; + private String repoPrefix; + private Storage storage; + + private static final Schema ROW_SCHEMA = + Schema.builder().addInt32Field("id").addStringField("name").build(); + + private static final List TEST_ROWS = + IntStream.range(0, 100) + .mapToObj(i -> Row.withSchema(ROW_SCHEMA).addValues(i, "name_" + i).build()) + .collect(Collectors.toList()); + + @Before + public void setup() throws Exception { + storage = StorageOptions.newBuilder().build().getService(); + long salt = System.currentTimeMillis(); + repoPrefix = "delta_io_it/" + testName.getMethodName() + "-" + salt; + repoPath = "gs://" + BUCKET + "/" + repoPrefix; + + LOG.info("Generating Delta Lake repository at {}", repoPath); + + // 1. Write Parquet file using a direct local pipeline + DirectOptions setupOptions = PipelineOptionsFactory.as(DirectOptions.class); + setupOptions.setRunner(DirectRunner.class); + setupOptions.setBlockOnRun(true); + Pipeline setupPipeline = Pipeline.create(setupOptions); + + org.apache.avro.Schema avroSchema = AvroUtils.toAvroSchema(ROW_SCHEMA); + setupPipeline + .apply(Create.of(TEST_ROWS).withRowSchema(ROW_SCHEMA)) + .apply( + MapElements.into(TypeDescriptor.of(GenericRecord.class)) + .via(AvroUtils.getRowToGenericRecordFunction(avroSchema))) + .setCoder(AvroCoder.of(avroSchema)) + .apply( + FileIO.write() + .via(ParquetIO.sink(avroSchema)) + .to(repoPath + "/") + .withNaming( + (BoundedWindow window, + PaneInfo paneInfo, + int numShards, + int shardIndex, + Compression compression) -> "part-00000.parquet")); + setupPipeline.run().waitUntilFinish(); + + // 2. Find written Parquet file to inspect its size + BlobId parquetBlobId = BlobId.of(BUCKET, repoPrefix + "/part-00000.parquet"); + Blob parquetBlob = storage.get(parquetBlobId); + assertNotNull("Parquet file not found on GCS: " + parquetBlobId, parquetBlob); + long fileSize = parquetBlob.getSize(); + + // 3. Create the Delta log commit file + String commitContent = + "{\"protocol\":{\"minReaderVersion\":1,\"minWriterVersion\":2}}\n" + + "{\"metaData\":{\"id\":\"" + + salt + + "\",\"format\":{\"provider\":\"parquet\",\"options\":{}},\"schemaString\":\"{\\\"type\\\":\\\"struct\\\",\\\"fields\\\":[{\\\"name\\\":\\\"id\\\",\\\"type\\\":\\\"integer\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}},{\\\"name\\\":\\\"name\\\",\\\"type\\\":\\\"string\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}}]}\",\"partitionColumns\":[],\"configuration\":{},\"createdAt\":123456789}}\n" + + "{\"add\":{\"path\":\"part-00000.parquet\",\"partitionValues\":{},\"size\":" + + fileSize + + ",\"modificationTime\":123456789,\"dataChange\":true}}"; + + BlobId commitBlobId = BlobId.of(BUCKET, repoPrefix + "/_delta_log/00000000000000000000.json"); + BlobInfo commitBlobInfo = + BlobInfo.newBuilder(commitBlobId).setContentType("application/json").build(); + storage.create(commitBlobInfo, commitContent.getBytes(StandardCharsets.UTF_8)); + LOG.info("Successfully generated Delta Lake repository"); + } + + @After + public void teardown() { + LOG.info("Cleaning up Delta Lake repository at {}", repoPath); + try { + Iterable blobs = + storage.list(BUCKET, Storage.BlobListOption.prefix(repoPrefix)).getValues(); + blobs.forEach(b -> storage.delete(b.getBlobId())); + storage.close(); + } catch (Exception e) { + LOG.warn("Failed to clean up GCS repository at {}", repoPath, e); + } + } + + @Test + public void testReadDeltaLakeTable() { + Map hadoopConfig = new HashMap<>(); + hadoopConfig.put("fs.gs.impl", "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem"); + hadoopConfig.put( + "fs.AbstractFileSystem.gs.impl", "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFS"); + + PCollection output = + readPipeline + .apply( + Managed.read(Managed.DELTA_LAKE) + .withConfig(ImmutableMap.of("table", repoPath, "hadoop_config", hadoopConfig))) + .getSinglePCollection(); + + PAssert.that(output).containsInAnyOrder(TEST_ROWS); + readPipeline.run().waitUntilFinish(); + } +} From 558122e013fa2f131b8b8efc72e735e2200a3204 Mon Sep 17 00:00:00 2001 From: Chamikara Jayalath Date: Fri, 12 Jun 2026 15:15:49 -0700 Subject: [PATCH 2/5] Updates --- .../apache/beam/sdk/io/delta/DeltaIOIT.java | 35 ++++++++++++++----- 1 file changed, 26 insertions(+), 9 deletions(-) diff --git a/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaIOIT.java b/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaIOIT.java index 36df75a05c7d..68592aea2559 100644 --- a/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaIOIT.java +++ b/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaIOIT.java @@ -67,11 +67,11 @@ public class DeltaIOIT { private static final Logger LOG = LoggerFactory.getLogger(DeltaIOIT.class); - private static final String BUCKET = "apache-beam-testing-delta-lake"; - + private static final String DEFAULT_BUCKET = "apache-beam-testing-delta-lake"; @Rule public final TestPipeline readPipeline = TestPipeline.create(); @Rule public final TestName testName = new TestName(); + private String bucket; private String repoPath; private String repoPrefix; private Storage storage; @@ -88,8 +88,18 @@ public class DeltaIOIT { public void setup() throws Exception { storage = StorageOptions.newBuilder().build().getService(); long salt = System.currentTimeMillis(); - repoPrefix = "delta_io_it/" + testName.getMethodName() + "-" + salt; - repoPath = "gs://" + BUCKET + "/" + repoPrefix; + + String tempLocation = readPipeline.getOptions().getTempLocation(); + if (tempLocation != null && tempLocation.startsWith("gs://")) { + org.apache.beam.sdk.extensions.gcp.util.gcs.GcsPath gcsPath = org.apache.beam.sdk.extensions.gcp.util.gcs.GcsPath + .fromUri(tempLocation); + bucket = gcsPath.getBucket(); + repoPrefix = gcsPath.getObject() + "/delta_io_it/" + testName.getMethodName() + "-" + salt; + } else { + bucket = DEFAULT_BUCKET; + repoPrefix = "delta_io_it/" + testName.getMethodName() + "-" + salt; + } + repoPath = "gs://" + bucket + "/" + repoPrefix; LOG.info("Generating Delta Lake repository at {}", repoPath); @@ -119,7 +129,7 @@ public void setup() throws Exception { setupPipeline.run().waitUntilFinish(); // 2. Find written Parquet file to inspect its size - BlobId parquetBlobId = BlobId.of(BUCKET, repoPrefix + "/part-00000.parquet"); + BlobId parquetBlobId = BlobId.of(bucket, repoPrefix + "/part-00000.parquet"); Blob parquetBlob = storage.get(parquetBlobId); assertNotNull("Parquet file not found on GCS: " + parquetBlobId, parquetBlob); long fileSize = parquetBlob.getSize(); @@ -134,7 +144,7 @@ public void setup() throws Exception { + fileSize + ",\"modificationTime\":123456789,\"dataChange\":true}}"; - BlobId commitBlobId = BlobId.of(BUCKET, repoPrefix + "/_delta_log/00000000000000000000.json"); + BlobId commitBlobId = BlobId.of(bucket, repoPrefix + "/_delta_log/00000000000000000000.json"); BlobInfo commitBlobInfo = BlobInfo.newBuilder(commitBlobId).setContentType("application/json").build(); storage.create(commitBlobInfo, commitContent.getBytes(StandardCharsets.UTF_8)); @@ -143,12 +153,14 @@ public void setup() throws Exception { @After public void teardown() { + if (storage == null) { + return; + } LOG.info("Cleaning up Delta Lake repository at {}", repoPath); try { - Iterable blobs = - storage.list(BUCKET, Storage.BlobListOption.prefix(repoPrefix)).getValues(); + Iterable blobs = + storage.list(bucket, Storage.BlobListOption.prefix(repoPrefix)).getValues(); blobs.forEach(b -> storage.delete(b.getBlobId())); - storage.close(); } catch (Exception e) { LOG.warn("Failed to clean up GCS repository at {}", repoPath, e); } @@ -160,6 +172,11 @@ public void testReadDeltaLakeTable() { hadoopConfig.put("fs.gs.impl", "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem"); hadoopConfig.put( "fs.AbstractFileSystem.gs.impl", "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFS"); + String project = readPipeline.getOptions().as(org.apache.beam.sdk.extensions.gcp.options.GcpOptions.class) + .getProject(); + if (project != null) { + hadoopConfig.put("fs.gs.project.id", project); + } PCollection output = readPipeline From ac9581dda9a571e687b0947aae65a56d84547ea2 Mon Sep 17 00:00:00 2001 From: Chamikara Jayalath Date: Fri, 12 Jun 2026 15:20:08 -0700 Subject: [PATCH 3/5] Updates --- .../test/java/org/apache/beam/sdk/io/delta/DeltaIOIT.java | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaIOIT.java b/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaIOIT.java index 68592aea2559..e4506771ebb2 100644 --- a/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaIOIT.java +++ b/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaIOIT.java @@ -91,7 +91,7 @@ public void setup() throws Exception { String tempLocation = readPipeline.getOptions().getTempLocation(); if (tempLocation != null && tempLocation.startsWith("gs://")) { - org.apache.beam.sdk.extensions.gcp.util.gcs.GcsPath gcsPath = org.apache.beam.sdk.extensions.gcp.util.gcs.GcsPath + org.apache.beam.sdk.extensions.gcp.util.gcsfs.GcsPath gcsPath = org.apache.beam.sdk.extensions.gcp.util.gcsfs.GcsPath .fromUri(tempLocation); bucket = gcsPath.getBucket(); repoPrefix = gcsPath.getObject() + "/delta_io_it/" + testName.getMethodName() + "-" + salt; @@ -158,7 +158,7 @@ public void teardown() { } LOG.info("Cleaning up Delta Lake repository at {}", repoPath); try { - Iterable blobs = + Iterable blobs = storage.list(bucket, Storage.BlobListOption.prefix(repoPrefix)).getValues(); blobs.forEach(b -> storage.delete(b.getBlobId())); } catch (Exception e) { @@ -172,7 +172,9 @@ public void testReadDeltaLakeTable() { hadoopConfig.put("fs.gs.impl", "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem"); hadoopConfig.put( "fs.AbstractFileSystem.gs.impl", "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFS"); - String project = readPipeline.getOptions().as(org.apache.beam.sdk.extensions.gcp.options.GcpOptions.class) + String project = readPipeline + .getOptions() + .as(org.apache.beam.sdk.extensions.gcp.options.GcpOptions.class) .getProject(); if (project != null) { hadoopConfig.put("fs.gs.project.id", project); From 23ddd1ad6de49168b50c4679767f0ec4099ae290 Mon Sep 17 00:00:00 2001 From: Chamikara Jayalath Date: Fri, 12 Jun 2026 15:27:00 -0700 Subject: [PATCH 4/5] updates --- .github/workflows/README.md | 1 + 1 file changed, 1 insertion(+) diff --git a/.github/workflows/README.md b/.github/workflows/README.md index 93fda448521b..dcc885feec9d 100644 --- a/.github/workflows/README.md +++ b/.github/workflows/README.md @@ -330,6 +330,7 @@ PostCommit Jobs run in a schedule against master branch and generally do not get | [ PostCommit Java BigQueryEarlyRollout ](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_BigQueryEarlyRollout.yml) | N/A |`beam_PostCommit_Java_BigQueryEarlyRollout.json`| [![.github/workflows/beam_PostCommit_Java_BigQueryEarlyRollout.yml](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_BigQueryEarlyRollout.yml/badge.svg?event=schedule)](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_BigQueryEarlyRollout.yml?query=event%3Aschedule) | | [ PostCommit Java Dataflow V1 ](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_DataflowV1.yml) | N/A |`beam_PostCommit_Java_DataflowV1.json`| [![.github/workflows/beam_PostCommit_Java_DataflowV1.yml](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_DataflowV1.yml/badge.svg?event=schedule)](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_DataflowV1.yml?query=event%3Aschedule) | | [ PostCommit Java Dataflow V2 ](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_DataflowV2.yml) | N/A |`beam_PostCommit_Java_DataflowV2.json`| [![.github/workflows/beam_PostCommit_Java_DataflowV2.yml](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_DataflowV2.yml/badge.svg?event=schedule)](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_DataflowV2.yml?query=event%3Aschedule) | +| [ PostCommit Java Delta IO Dataflow ](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_Delta_IO_Dataflow.yml) | N/A |`beam_PostCommit_Java_Delta_IO_Dataflow.json`| [![.github/workflows/beam_PostCommit_Java_Delta_IO_Dataflow.yml](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_Delta_IO_Dataflow.yml/badge.svg?event=schedule)](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_Delta_IO_Dataflow.yml?query=event%3Aschedule) | | [ PostCommit Java Examples Dataflow ARM ](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_Examples_Dataflow_ARM.yml) | ['8','11','17','21','25'] |`beam_PostCommit_Java_Examples_Dataflow_ARM.json`| [![.github/workflows/beam_PostCommit_Java_Examples_Dataflow_ARM.yml](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_Examples_Dataflow_ARM.yml/badge.svg?event=schedule)](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_Examples_Dataflow_ARM.yml?query=event%3Aschedule) | | [ PostCommit Java Examples Dataflow](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_Examples_Dataflow.yml) | N/A |`beam_PostCommit_Java_Examples_Dataflow.json`| [![.github/workflows/beam_PostCommit_Java_Examples_Dataflow.yml](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_Examples_Dataflow.yml/badge.svg?event=schedule)](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_Examples_Dataflow.yml?query=event%3Aschedule) | | [ PostCommit Java Examples Dataflow Java ](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_Examples_Dataflow_Java.yml) | ['8','17','21','25'] |`beam_PostCommit_Java_Examples_Dataflow_Java.json`| [![.github/workflows/beam_PostCommit_Java_Examples_Dataflow_Java.yml](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_Examples_Dataflow_Java.yml/badge.svg?event=schedule)](https://github.com/apache/beam/actions/workflows/beam_PostCommit_Java_Examples_Dataflow_Java.yml?query=event%3Aschedule) | From 8d347ac388ce5ed28c60656a092a4667b5cfa71d Mon Sep 17 00:00:00 2001 From: Chamikara Jayalath Date: Thu, 9 Jul 2026 15:20:21 -0700 Subject: [PATCH 5/5] Write data using the Delta Kernel API. --- .../apache/beam/sdk/io/delta/DeltaIOIT.java | 213 ++++++++++++------ 1 file changed, 142 insertions(+), 71 deletions(-) diff --git a/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaIOIT.java b/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaIOIT.java index e4506771ebb2..e0d35f30faa6 100644 --- a/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaIOIT.java +++ b/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaIOIT.java @@ -17,41 +17,43 @@ */ package org.apache.beam.sdk.io.delta; -import static org.junit.Assert.assertNotNull; - import com.google.cloud.storage.Blob; -import com.google.cloud.storage.BlobId; -import com.google.cloud.storage.BlobInfo; import com.google.cloud.storage.Storage; import com.google.cloud.storage.StorageOptions; -import java.nio.charset.StandardCharsets; +import io.delta.kernel.DataWriteContext; +import io.delta.kernel.Operation; +import io.delta.kernel.Table; +import io.delta.kernel.Transaction; +import io.delta.kernel.TransactionBuilder; +import io.delta.kernel.TransactionCommitResult; +import io.delta.kernel.data.ColumnVector; +import io.delta.kernel.data.ColumnarBatch; +import io.delta.kernel.data.FilteredColumnarBatch; +import io.delta.kernel.defaults.engine.DefaultEngine; +import io.delta.kernel.defaults.internal.data.DefaultColumnarBatch; +import io.delta.kernel.engine.Engine; +import io.delta.kernel.types.DataType; +import io.delta.kernel.types.IntegerType; +import io.delta.kernel.types.StringType; +import io.delta.kernel.types.StructType; +import io.delta.kernel.utils.CloseableIterable; +import io.delta.kernel.utils.CloseableIterator; +import io.delta.kernel.utils.DataFileStatus; +import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.stream.Collectors; import java.util.stream.IntStream; -import org.apache.avro.generic.GenericRecord; -import org.apache.beam.runners.direct.DirectOptions; -import org.apache.beam.runners.direct.DirectRunner; -import org.apache.beam.sdk.Pipeline; -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.managed.Managed; -import org.apache.beam.sdk.options.PipelineOptionsFactory; 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.MapElements; -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.Row; -import org.apache.beam.sdk.values.TypeDescriptor; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap; +import org.apache.hadoop.conf.Configuration; import org.junit.After; import org.junit.Before; import org.junit.Rule; @@ -91,8 +93,8 @@ public void setup() throws Exception { String tempLocation = readPipeline.getOptions().getTempLocation(); if (tempLocation != null && tempLocation.startsWith("gs://")) { - org.apache.beam.sdk.extensions.gcp.util.gcsfs.GcsPath gcsPath = org.apache.beam.sdk.extensions.gcp.util.gcsfs.GcsPath - .fromUri(tempLocation); + org.apache.beam.sdk.extensions.gcp.util.gcsfs.GcsPath gcsPath = + org.apache.beam.sdk.extensions.gcp.util.gcsfs.GcsPath.fromUri(tempLocation); bucket = gcsPath.getBucket(); repoPrefix = gcsPath.getObject() + "/delta_io_it/" + testName.getMethodName() + "-" + salt; } else { @@ -103,51 +105,119 @@ public void setup() throws Exception { LOG.info("Generating Delta Lake repository at {}", repoPath); - // 1. Write Parquet file using a direct local pipeline - DirectOptions setupOptions = PipelineOptionsFactory.as(DirectOptions.class); - setupOptions.setRunner(DirectRunner.class); - setupOptions.setBlockOnRun(true); - Pipeline setupPipeline = Pipeline.create(setupOptions); - - org.apache.avro.Schema avroSchema = AvroUtils.toAvroSchema(ROW_SCHEMA); - setupPipeline - .apply(Create.of(TEST_ROWS).withRowSchema(ROW_SCHEMA)) - .apply( - MapElements.into(TypeDescriptor.of(GenericRecord.class)) - .via(AvroUtils.getRowToGenericRecordFunction(avroSchema))) - .setCoder(AvroCoder.of(avroSchema)) - .apply( - FileIO.write() - .via(ParquetIO.sink(avroSchema)) - .to(repoPath + "/") - .withNaming( - (BoundedWindow window, - PaneInfo paneInfo, - int numShards, - int shardIndex, - Compression compression) -> "part-00000.parquet")); - setupPipeline.run().waitUntilFinish(); - - // 2. Find written Parquet file to inspect its size - BlobId parquetBlobId = BlobId.of(bucket, repoPrefix + "/part-00000.parquet"); - Blob parquetBlob = storage.get(parquetBlobId); - assertNotNull("Parquet file not found on GCS: " + parquetBlobId, parquetBlob); - long fileSize = parquetBlob.getSize(); - - // 3. Create the Delta log commit file - String commitContent = - "{\"protocol\":{\"minReaderVersion\":1,\"minWriterVersion\":2}}\n" - + "{\"metaData\":{\"id\":\"" - + salt - + "\",\"format\":{\"provider\":\"parquet\",\"options\":{}},\"schemaString\":\"{\\\"type\\\":\\\"struct\\\",\\\"fields\\\":[{\\\"name\\\":\\\"id\\\",\\\"type\\\":\\\"integer\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}},{\\\"name\\\":\\\"name\\\",\\\"type\\\":\\\"string\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}}]}\",\"partitionColumns\":[],\"configuration\":{},\"createdAt\":123456789}}\n" - + "{\"add\":{\"path\":\"part-00000.parquet\",\"partitionValues\":{},\"size\":" - + fileSize - + ",\"modificationTime\":123456789,\"dataChange\":true}}"; - - BlobId commitBlobId = BlobId.of(bucket, repoPrefix + "/_delta_log/00000000000000000000.json"); - BlobInfo commitBlobInfo = - BlobInfo.newBuilder(commitBlobId).setContentType("application/json").build(); - storage.create(commitBlobInfo, commitContent.getBytes(StandardCharsets.UTF_8)); + Configuration configuration = new Configuration(); + configuration.set("fs.gs.impl", "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem"); + configuration.set( + "fs.AbstractFileSystem.gs.impl", "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFS"); + configuration.set("fs.gs.auth.type", "APPLICATION_DEFAULT"); + String project = + readPipeline + .getOptions() + .as(org.apache.beam.sdk.extensions.gcp.options.GcpOptions.class) + .getProject(); + if (project != null) { + configuration.set("fs.gs.project.id", project); + } + + Engine engine = DefaultEngine.create(configuration); + Table table = Table.forPath(engine, repoPath); + + StructType deltaSchema = + new StructType().add("id", IntegerType.INTEGER).add("name", StringType.STRING); + + TransactionBuilder txnBuilder = + table.createTransactionBuilder(engine, "DeltaIOIT", Operation.CREATE_TABLE); + txnBuilder = txnBuilder.withSchema(engine, deltaSchema); + Transaction txn = txnBuilder.build(engine); + io.delta.kernel.data.Row txnState = txn.getTransactionState(engine); + + ColumnVector idVector = + new ColumnVector() { + @Override + public DataType getDataType() { + return IntegerType.INTEGER; + } + + @Override + public int getSize() { + return TEST_ROWS.size(); + } + + @Override + public void close() {} + + @Override + public boolean isNullAt(int rowId) { + return TEST_ROWS.get(rowId).getValue("id") == null; + } + + @Override + public int getInt(int rowId) { + return TEST_ROWS.get(rowId).getInt32("id"); + } + }; + + ColumnVector nameVector = + new ColumnVector() { + @Override + public DataType getDataType() { + return StringType.STRING; + } + + @Override + public int getSize() { + return TEST_ROWS.size(); + } + + @Override + public void close() {} + + @Override + public boolean isNullAt(int rowId) { + return TEST_ROWS.get(rowId).getValue("name") == null; + } + + @Override + public String getString(int rowId) { + return TEST_ROWS.get(rowId).getString("name"); + } + }; + + ColumnVector[] vectors = new ColumnVector[] {idVector, nameVector}; + ColumnarBatch columnarBatch = new DefaultColumnarBatch(TEST_ROWS.size(), deltaSchema, vectors); + FilteredColumnarBatch filteredBatch = + new FilteredColumnarBatch(columnarBatch, Optional.empty()); + + CloseableIterator data = + io.delta.kernel.internal.util.Utils.toCloseableIterator( + Collections.singletonList(filteredBatch).iterator()); + + CloseableIterator physicalData = + Transaction.transformLogicalData(engine, txnState, data, Collections.emptyMap()); + + DataWriteContext writeContext = + Transaction.getWriteContext(engine, txnState, Collections.emptyMap()); + + CloseableIterator dataFiles = + engine + .getParquetHandler() + .writeParquetFiles( + writeContext.getTargetDirectory(), + physicalData, + writeContext.getStatisticsColumns()); + + CloseableIterator dataActions = + Transaction.generateAppendActions(engine, txnState, dataFiles, writeContext); + + CloseableIterable dataActionsIterable = + CloseableIterable.inMemoryIterable(dataActions); + + TransactionCommitResult commitResult = txn.commit(engine, dataActionsIterable); + + if (commitResult.getVersion() < 0) { + throw new RuntimeException("Table creation/write failed"); + } + LOG.info("Successfully generated Delta Lake repository"); } @@ -172,10 +242,11 @@ public void testReadDeltaLakeTable() { hadoopConfig.put("fs.gs.impl", "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem"); hadoopConfig.put( "fs.AbstractFileSystem.gs.impl", "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFS"); - String project = readPipeline - .getOptions() - .as(org.apache.beam.sdk.extensions.gcp.options.GcpOptions.class) - .getProject(); + String project = + readPipeline + .getOptions() + .as(org.apache.beam.sdk.extensions.gcp.options.GcpOptions.class) + .getProject(); if (project != null) { hadoopConfig.put("fs.gs.project.id", project); }