From f1cce12e9c92efa8a1bbe98f497add32ea472e25 Mon Sep 17 00:00:00 2001 From: Chamikara Jayalath Date: Fri, 14 Aug 2026 17:36:54 -0700 Subject: [PATCH] Adds support for deading at a given Delta Lake version or timestamp --- ...eam_PostCommit_Java_Delta_IO_Dataflow.json | 2 +- .../sdk/io/delta/CreateReadTasksDoFn.java | 23 ++- .../org/apache/beam/sdk/io/delta/DeltaIO.java | 39 +++- .../DeltaReadSchemaTransformProvider.java | 6 +- .../apache/beam/sdk/io/delta/DeltaIOIT.java | 110 +++++++++++ .../apache/beam/sdk/io/delta/DeltaIOTest.java | 179 ++++++++++++++++++ .../DeltaReadSchemaTransformProviderTest.java | 90 +++++++++ .../content/en/documentation/io/managed-io.md | 4 +- 8 files changed, 437 insertions(+), 16 deletions(-) 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 12481ae0dbc8..3a009261f4f9 100644 --- a/.github/trigger_files/beam_PostCommit_Java_Delta_IO_Dataflow.json +++ b/.github/trigger_files/beam_PostCommit_Java_Delta_IO_Dataflow.json @@ -1,4 +1,4 @@ { "comment": "Modify this file in a trivial way to cause this test suite to run.", - "modification": 4 + "modification": 2 } diff --git a/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/CreateReadTasksDoFn.java b/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/CreateReadTasksDoFn.java index 36c9a1a47f8c..9d4da4708e82 100644 --- a/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/CreateReadTasksDoFn.java +++ b/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/CreateReadTasksDoFn.java @@ -38,9 +38,20 @@ class CreateReadTasksDoFn extends DoFn { private static final long MAX_TASK_SIZE_BYTES = 1024L * 1024L * 1024L; // 1 GB private final @Nullable Map hadoopConfig; + private final @Nullable Long version; + private final @Nullable String timestamp; public CreateReadTasksDoFn(@Nullable Map hadoopConfig) { + this(hadoopConfig, null, null); + } + + public CreateReadTasksDoFn( + @Nullable Map hadoopConfig, + @Nullable Long version, + @Nullable String timestamp) { this.hadoopConfig = hadoopConfig; + this.version = version; + this.timestamp = timestamp; } @ProcessElement @@ -54,7 +65,17 @@ public void processElement(@Element String tablePath, OutputReceiverOnly one of version or timestamp should be provided. If neither is provided, the latest + * version (HEAD) is read. + */ public ReadRows withVersion(@Nullable Long version) { return toBuilder().setVersion(version).build(); } + /** + * Specifies the timestamp of the Delta Lake table to read as an ISO 8601 string (e.g. + * "2026-05-20T15:43:26Z"). + * + *

Only one of version or timestamp should be provided. If neither is provided, the latest + * version (HEAD) is read. + */ public ReadRows withTimestamp(@Nullable String timestamp) { return toBuilder().setTimestamp(timestamp).build(); } @@ -132,14 +145,8 @@ public PCollection expand(PBegin input) { if (path == null) { throw new IllegalArgumentException("Table path must be set."); } - if (getTimestamp() != null) { - throw new UnsupportedOperationException( - "Reading from a specific timestamp is not supported yet"); - } - - if (getVersion() != null) { - throw new UnsupportedOperationException( - "Reading from a specific version is not supported yet"); + if (getVersion() != null && getTimestamp() != null) { + throw new IllegalArgumentException("Cannot set both version and timestamp."); } Configuration conf = new Configuration(); @@ -151,7 +158,17 @@ public PCollection expand(PBegin input) { } Engine engine = DefaultEngine.create(conf); Table table = Table.forPath(engine, path); - io.delta.kernel.Snapshot snapshot = table.getLatestSnapshot(engine); + Snapshot snapshot; + Long versionVal = getVersion(); + String timestampVal = getTimestamp(); + if (versionVal != null) { + snapshot = table.getSnapshotAsOfVersion(engine, versionVal); + } else if (timestampVal != null) { + long timestampMillis = java.time.Instant.parse(timestampVal).toEpochMilli(); + snapshot = table.getSnapshotAsOfTimestamp(engine, timestampMillis); + } else { + snapshot = table.getLatestSnapshot(engine); + } StructType deltaSchema = snapshot.getSchema(); if (deltaSchema == null) { throw new IllegalStateException("Table schema is null."); @@ -160,7 +177,9 @@ public PCollection expand(PBegin input) { return input .apply("Create Path", Create.of(path)) - .apply("Plan Files", ParDo.of(new CreateReadTasksDoFn(hadoopConfig))) + .apply( + "Plan Files", + ParDo.of(new CreateReadTasksDoFn(hadoopConfig, getVersion(), getTimestamp()))) .apply("Read Logical Data", ParDo.of(new DeltaSourceDoFn(hadoopConfig))) .setRowSchema(beamSchema); } diff --git a/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/DeltaReadSchemaTransformProvider.java b/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/DeltaReadSchemaTransformProvider.java index 48dc3a2c7488..3121a36d4c3b 100644 --- a/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/DeltaReadSchemaTransformProvider.java +++ b/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/DeltaReadSchemaTransformProvider.java @@ -114,11 +114,13 @@ static Builder builder() { @SchemaFieldDescription("Identifier of the Delta Lake table.") abstract String getTable(); - @SchemaFieldDescription("Version of the Delta Lake table to read.") + @SchemaFieldDescription( + "Version of the Delta Lake table to read. Cannot be set if timestamp is set.") @Nullable abstract Long getVersion(); - @SchemaFieldDescription("Timestamp of the Delta Lake table to read.") + @SchemaFieldDescription( + "Timestamp of the Delta Lake table to read (in UTC ISO 8601 format, e.g. 2026-05-20T15:43:26Z). Cannot be set if version is set.") @Nullable abstract String getTimestamp(); 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 ad526008b20e..744252949e2d 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 @@ -302,6 +302,116 @@ public void testReadDeltaLakeTable() { readPipeline.run().waitUntilFinish(); } + @Test + public void testReadDeltaLakeTableAtTimestamp() throws Exception { + ExperimentalOptions options = readPipeline.getOptions().as(ExperimentalOptions.class); + ExperimentalOptions.addExperiment(options, "use_runner_v2"); + + 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"); + hadoopConfig.put("fs.gs.auth.type", "APPLICATION_DEFAULT"); + 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); + } + + org.apache.hadoop.conf.Configuration conf = new org.apache.hadoop.conf.Configuration(); + for (Map.Entry entry : hadoopConfig.entrySet()) { + conf.set(entry.getKey(), entry.getValue()); + } + Engine engine = DefaultEngine.create(conf); + + // Wait briefly to ensure timestamp is after version 0 commit + Thread.sleep(1000); + String timestampV0 = java.time.Instant.ofEpochMilli(System.currentTimeMillis()).toString(); + Thread.sleep(1000); + + // Write version 1 with additional rows + List additionalRows = + IntStream.range(100, 150) + .mapToObj(i -> Row.withSchema(ROW_SCHEMA).addValues(i, "name_" + i).build()) + .collect(Collectors.toList()); + + StructType deltaSchema = + new StructType().add("id", IntegerType.INTEGER).add("name", StringType.STRING); + + DeltaWriteTestUtils.writeAppendCommit( + engine, repoPath, 1L, System.currentTimeMillis(), deltaSchema, additionalRows); + + PCollection output = + readPipeline + .apply( + Managed.read(Managed.DELTA_LAKE) + .withConfig( + ImmutableMap.of( + "table", + repoPath, + "timestamp", + timestampV0, + "hadoop_config", + hadoopConfig))) + .getSinglePCollection(); + + PAssert.that(output).containsInAnyOrder(TEST_ROWS); + readPipeline.run().waitUntilFinish(); + } + + @Test + public void testReadDeltaLakeTableAtVersion() throws Exception { + ExperimentalOptions options = readPipeline.getOptions().as(ExperimentalOptions.class); + ExperimentalOptions.addExperiment(options, "use_runner_v2"); + + 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"); + hadoopConfig.put("fs.gs.auth.type", "APPLICATION_DEFAULT"); + 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); + } + + org.apache.hadoop.conf.Configuration conf = new org.apache.hadoop.conf.Configuration(); + for (Map.Entry entry : hadoopConfig.entrySet()) { + conf.set(entry.getKey(), entry.getValue()); + } + Engine engine = DefaultEngine.create(conf); + + // Write version 1 with additional rows + List additionalRows = + IntStream.range(100, 150) + .mapToObj(i -> Row.withSchema(ROW_SCHEMA).addValues(i, "name_" + i).build()) + .collect(Collectors.toList()); + + StructType deltaSchema = + new StructType().add("id", IntegerType.INTEGER).add("name", StringType.STRING); + + DeltaWriteTestUtils.writeAppendCommit( + engine, repoPath, 1L, System.currentTimeMillis(), deltaSchema, additionalRows); + + PCollection output = + readPipeline + .apply( + Managed.read(Managed.DELTA_LAKE) + .withConfig( + ImmutableMap.of( + "table", repoPath, "version", 0L, "hadoop_config", hadoopConfig))) + .getSinglePCollection(); + + PAssert.that(output).containsInAnyOrder(TEST_ROWS); + readPipeline.run().waitUntilFinish(); + } + @Test public void testReadChangesDeltaLake() throws Exception { ExperimentalOptions options = readPipeline.getOptions().as(ExperimentalOptions.class); diff --git a/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaIOTest.java b/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaIOTest.java index 0db0aef9e080..f26e79d798e3 100644 --- a/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaIOTest.java +++ b/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaIOTest.java @@ -109,6 +109,185 @@ public void testReadRowsNullDefaults() { Assert.assertNull(readRows.getHadoopConfig()); } + @Test + public void testReadRowsBothVersionAndTimestampThrows() { + org.apache.beam.sdk.Pipeline p = org.apache.beam.sdk.Pipeline.create(); + IllegalArgumentException exception = + Assert.assertThrows( + IllegalArgumentException.class, + () -> + p.apply( + DeltaIO.readRows() + .from("/path/to/table") + .withVersion(0L) + .withTimestamp("2026-05-20T15:43:26Z"))); + Assert.assertTrue(exception.getMessage().contains("Cannot set both version and timestamp.")); + } + + @Test + public void testReadRowsAtVersion() throws Exception { + File tableDir = tempFolder.newFolder("delta-table-read-version"); + Engine engine = DefaultEngine.create(new org.apache.hadoop.conf.Configuration()); + + Schema schema = Schema.builder().addField("name", Schema.FieldType.STRING).build(); + Row row1 = Row.withSchema(schema).addValues("row-1").build(); + Row row2 = Row.withSchema(schema).addValues("row-2").build(); + Row row3 = Row.withSchema(schema).addValues("row-3").build(); + StructType deltaSchema = new StructType().add("name", StringType.STRING); + + // Commit version 0 + DeltaWriteTestUtils.writeAppendCommit( + engine, + tableDir.getAbsolutePath(), + 0L, + 100000000000L, + deltaSchema, + java.util.Arrays.asList(row1, row2)); + + // Commit version 1 + DeltaWriteTestUtils.writeAppendCommit( + engine, + tableDir.getAbsolutePath(), + 1L, + 200000000000L, + deltaSchema, + java.util.Arrays.asList(row3)); + + // Read at version 0 + PCollection outputV0 = + readPipeline.apply(DeltaIO.readRows().from(tableDir.getAbsolutePath()).withVersion(0L)); + + PAssert.that(outputV0).containsInAnyOrder(row1, row2); + + readPipeline.run().waitUntilFinish(); + } + + @Test + public void testReadRowsAtTimestamp() throws Exception { + File tableDir = tempFolder.newFolder("delta-table-read-timestamp"); + Engine engine = DefaultEngine.create(new org.apache.hadoop.conf.Configuration()); + + Schema schema = Schema.builder().addField("name", Schema.FieldType.STRING).build(); + Row row1 = Row.withSchema(schema).addValues("row-1").build(); + Row row2 = Row.withSchema(schema).addValues("row-2").build(); + Row row3 = Row.withSchema(schema).addValues("row-3").build(); + StructType deltaSchema = new StructType().add("name", StringType.STRING); + + // Commit version 0 at timestamp 100000000000L + DeltaWriteTestUtils.writeAppendCommit( + engine, + tableDir.getAbsolutePath(), + 0L, + 100000000000L, + deltaSchema, + java.util.Arrays.asList(row1, row2)); + + // Commit version 1 at timestamp 200000000000L + DeltaWriteTestUtils.writeAppendCommit( + engine, + tableDir.getAbsolutePath(), + 1L, + 200000000000L, + deltaSchema, + java.util.Arrays.asList(row3)); + + // Read at timestamp between version 0 and version 1 + String timestampV0 = java.time.Instant.ofEpochMilli(150000000000L).toString(); + PCollection outputV0 = + readPipeline.apply( + DeltaIO.readRows().from(tableDir.getAbsolutePath()).withTimestamp(timestampV0)); + + PAssert.that(outputV0).containsInAnyOrder(row1, row2); + + readPipeline.run().waitUntilFinish(); + } + + @Test + public void testManagedDeltaReadWithVersion() throws Exception { + File tableDir = tempFolder.newFolder("managed-delta-table-version"); + Engine engine = DefaultEngine.create(new org.apache.hadoop.conf.Configuration()); + + Schema schema = Schema.builder().addField("name", Schema.FieldType.STRING).build(); + Row row1 = Row.withSchema(schema).addValues("row-1").build(); + Row row2 = Row.withSchema(schema).addValues("row-2").build(); + StructType deltaSchema = new StructType().add("name", StringType.STRING); + + // Commit version 0 + DeltaWriteTestUtils.writeAppendCommit( + engine, + tableDir.getAbsolutePath(), + 0L, + 100000000000L, + deltaSchema, + Collections.singletonList(row1)); + + // Commit version 1 + DeltaWriteTestUtils.writeAppendCommit( + engine, + tableDir.getAbsolutePath(), + 1L, + 200000000000L, + deltaSchema, + Collections.singletonList(row2)); + + // Read version 0 using Managed + PCollection output = + readPipeline + .apply( + Managed.read(Managed.DELTA_LAKE) + .withConfig( + ImmutableMap.of("table", tableDir.getAbsolutePath(), "version", 0L))) + .getSinglePCollection(); + + PAssert.that(output).containsInAnyOrder(row1); + + readPipeline.run().waitUntilFinish(); + } + + @Test + public void testManagedDeltaReadWithTimestamp() throws Exception { + File tableDir = tempFolder.newFolder("managed-delta-table-timestamp"); + Engine engine = DefaultEngine.create(new org.apache.hadoop.conf.Configuration()); + + Schema schema = Schema.builder().addField("name", Schema.FieldType.STRING).build(); + Row row1 = Row.withSchema(schema).addValues("row-1").build(); + Row row2 = Row.withSchema(schema).addValues("row-2").build(); + StructType deltaSchema = new StructType().add("name", StringType.STRING); + + // Commit version 0 at timestamp 100000000000L + DeltaWriteTestUtils.writeAppendCommit( + engine, + tableDir.getAbsolutePath(), + 0L, + 100000000000L, + deltaSchema, + Collections.singletonList(row1)); + + // Commit version 1 at timestamp 200000000000L + DeltaWriteTestUtils.writeAppendCommit( + engine, + tableDir.getAbsolutePath(), + 1L, + 200000000000L, + deltaSchema, + Collections.singletonList(row2)); + + // Read timestamp after version 0 using Managed + String timestampV0 = java.time.Instant.ofEpochMilli(150000000000L).toString(); + PCollection output = + readPipeline + .apply( + Managed.read(Managed.DELTA_LAKE) + .withConfig( + ImmutableMap.of( + "table", tableDir.getAbsolutePath(), "timestamp", timestampV0))) + .getSinglePCollection(); + + PAssert.that(output).containsInAnyOrder(row1); + + readPipeline.run().waitUntilFinish(); + } + @Test public void testPrintScanStateSchema() throws Exception { File tableDir = tempFolder.newFolder("delta-table-schema"); diff --git a/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaReadSchemaTransformProviderTest.java b/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaReadSchemaTransformProviderTest.java index 77aef7bce494..dfdea6534676 100644 --- a/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaReadSchemaTransformProviderTest.java +++ b/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaReadSchemaTransformProviderTest.java @@ -20,6 +20,10 @@ import static org.apache.beam.sdk.io.delta.DeltaReadSchemaTransformProvider.Configuration; import static org.apache.beam.sdk.io.delta.DeltaReadSchemaTransformProvider.OUTPUT_TAG; +import io.delta.kernel.defaults.engine.DefaultEngine; +import io.delta.kernel.engine.Engine; +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; @@ -124,4 +128,90 @@ public void testSimpleScan() throws Exception { readPipeline.run().waitUntilFinish(); } + + @Test + public void testReadWithVersion() throws Exception { + File tableDir = tempFolder.newFolder("delta-table-provider-version"); + Engine engine = DefaultEngine.create(new org.apache.hadoop.conf.Configuration()); + + Schema schema = Schema.builder().addField("name", Schema.FieldType.STRING).build(); + Row row1 = Row.withSchema(schema).addValues("row-1").build(); + Row row2 = Row.withSchema(schema).addValues("row-2").build(); + StructType deltaSchema = new StructType().add("name", StringType.STRING); + + // Commit version 0 + DeltaWriteTestUtils.writeAppendCommit( + engine, + tableDir.getAbsolutePath(), + 0L, + 100000000000L, + deltaSchema, + java.util.Collections.singletonList(row1)); + + // Commit version 1 + DeltaWriteTestUtils.writeAppendCommit( + engine, + tableDir.getAbsolutePath(), + 1L, + 200000000000L, + deltaSchema, + java.util.Collections.singletonList(row2)); + + Configuration readConfig = + Configuration.builder().setTable(tableDir.getAbsolutePath()).setVersion(0L).build(); + + PCollection output = + PCollectionRowTuple.empty(readPipeline) + .apply(new DeltaReadSchemaTransformProvider().from(readConfig)) + .get(OUTPUT_TAG); + + PAssert.that(output).containsInAnyOrder(row1); + + readPipeline.run().waitUntilFinish(); + } + + @Test + public void testReadWithTimestamp() throws Exception { + File tableDir = tempFolder.newFolder("delta-table-provider-timestamp"); + Engine engine = DefaultEngine.create(new org.apache.hadoop.conf.Configuration()); + + Schema schema = Schema.builder().addField("name", Schema.FieldType.STRING).build(); + Row row1 = Row.withSchema(schema).addValues("row-1").build(); + Row row2 = Row.withSchema(schema).addValues("row-2").build(); + StructType deltaSchema = new StructType().add("name", StringType.STRING); + + // Commit version 0 at timestamp 100000000000L + DeltaWriteTestUtils.writeAppendCommit( + engine, + tableDir.getAbsolutePath(), + 0L, + 100000000000L, + deltaSchema, + java.util.Collections.singletonList(row1)); + + // Commit version 1 at timestamp 200000000000L + DeltaWriteTestUtils.writeAppendCommit( + engine, + tableDir.getAbsolutePath(), + 1L, + 200000000000L, + deltaSchema, + java.util.Collections.singletonList(row2)); + + String timestampV0 = java.time.Instant.ofEpochMilli(150000000000L).toString(); + Configuration readConfig = + Configuration.builder() + .setTable(tableDir.getAbsolutePath()) + .setTimestamp(timestampV0) + .build(); + + PCollection output = + PCollectionRowTuple.empty(readPipeline) + .apply(new DeltaReadSchemaTransformProvider().from(readConfig)) + .get(OUTPUT_TAG); + + PAssert.that(output).containsInAnyOrder(row1); + + readPipeline.run().waitUntilFinish(); + } } diff --git a/website/www/site/content/en/documentation/io/managed-io.md b/website/www/site/content/en/documentation/io/managed-io.md index 70b5efe9ed17..5eb6f04ab80e 100644 --- a/website/www/site/content/en/documentation/io/managed-io.md +++ b/website/www/site/content/en/documentation/io/managed-io.md @@ -304,7 +304,7 @@ and Beam SQL is invoked via the Managed API under the hood. str - Timestamp of the Delta Lake table to read. + Timestamp of the Delta Lake table to read (in UTC ISO 8601 format, e.g. 2026-05-20T15:43:26Z). Cannot be set if version is set. @@ -315,7 +315,7 @@ and Beam SQL is invoked via the Managed API under the hood. int64 - Version of the Delta Lake table to read. + Version of the Delta Lake table to read. Cannot be set if timestamp is set.