This is an automated email from the ASF dual-hosted git repository.
chamikaramj pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new c1017953ac3 Adds support for reading at a given Delta Lake version or
timestamp (#39758)
c1017953ac3 is described below
commit c1017953ac3b213a951fe9534a33afcc1c1190ee
Author: Chamikara Jayalath <[email protected]>
AuthorDate: Fri Aug 21 16:54:16 2026 -0700
Adds support for reading at a given Delta Lake version or timestamp (#39758)
---
.../beam_PostCommit_Java_Delta_IO_Dataflow.json | 2 +-
.../beam/sdk/io/delta/CreateReadTasksDoFn.java | 23 +++-
.../java/org/apache/beam/sdk/io/delta/DeltaIO.java | 26 ++--
.../io/delta/DeltaReadSchemaTransformProvider.java | 6 +-
.../org/apache/beam/sdk/io/delta/DeltaIOIT.java | 141 ++++++++++++++-------
.../org/apache/beam/sdk/io/delta/DeltaIOTest.java | 102 +++++++++++++++
.../DeltaReadSchemaTransformProviderTest.java | 51 ++++++++
.../beam/sdk/io/delta/DeltaWriteTestUtils.java | 30 +++++
.../site/content/en/documentation/io/managed-io.md | 4 +-
9 files changed, 326 insertions(+), 59 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 12481ae0dbc..ab4daeae234 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": 3
}
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 36c9a1a47f8..9d4da4708e8 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 @@ import org.checkerframework.checker.nullness.qual.Nullable;
class CreateReadTasksDoFn extends DoFn<String, DeltaReadTask> {
private static final long MAX_TASK_SIZE_BYTES = 1024L * 1024L * 1024L; // 1
GB
private final @Nullable Map<String, String> hadoopConfig;
+ private final @Nullable Long version;
+ private final @Nullable String timestamp;
public CreateReadTasksDoFn(@Nullable Map<String, String> hadoopConfig) {
+ this(hadoopConfig, null, null);
+ }
+
+ public CreateReadTasksDoFn(
+ @Nullable Map<String, String> hadoopConfig,
+ @Nullable Long version,
+ @Nullable String timestamp) {
this.hadoopConfig = hadoopConfig;
+ this.version = version;
+ this.timestamp = timestamp;
}
@ProcessElement
@@ -54,7 +65,17 @@ class CreateReadTasksDoFn extends DoFn<String,
DeltaReadTask> {
}
Engine engine = DefaultEngine.create(conf);
Table table = Table.forPath(engine, tablePath);
- Snapshot snapshot = table.getLatestSnapshot(engine);
+ Snapshot snapshot;
+ Long versionVal = version;
+ String timestampVal = timestamp;
+ 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);
+ }
Scan scan = snapshot.getScanBuilder().build();
Row scanState = scan.getScanState(engine);
SerializableRow serializableScanState = new SerializableRow(scanState);
diff --git
a/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/DeltaIO.java
b/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/DeltaIO.java
index 3a53b5c7620..4701ff59c4a 100644
--- a/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/DeltaIO.java
+++ b/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/DeltaIO.java
@@ -132,14 +132,8 @@ public class DeltaIO {
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 +145,17 @@ public class DeltaIO {
}
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 +164,9 @@ public class DeltaIO {
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 48dc3a2c748..3121a36d4c3 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 @@ public class DeltaReadSchemaTransformProvider extends
TypedSchemaTransformProvid
@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 ad526008b20..ab7fd8e24f4 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
@@ -49,6 +49,7 @@ import java.util.Map;
import java.util.Optional;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
+import org.apache.beam.sdk.extensions.gcp.options.GcpOptions;
import org.apache.beam.sdk.managed.Managed;
import org.apache.beam.sdk.options.ExperimentalOptions;
import org.apache.beam.sdk.schemas.Schema;
@@ -114,19 +115,7 @@ public class DeltaIOIT {
LOG.info("Generating Delta Lake repository at {}", repoPath);
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);
- }
-
+ getHadoopConfig().forEach(configuration::set);
Engine engine = DefaultEngine.create(configuration);
Table table = Table.forPath(engine, repoPath);
@@ -278,18 +267,7 @@ public class DeltaIOIT {
ExperimentalOptions options =
readPipeline.getOptions().as(ExperimentalOptions.class);
ExperimentalOptions.addExperiment(options, "use_runner_v2");
- Map<String, String> 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");
- 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);
- }
+ Map<String, String> hadoopConfig = getHadoopConfig();
PCollection<Row> output =
readPipeline
@@ -302,6 +280,85 @@ public class DeltaIOIT {
readPipeline.run().waitUntilFinish();
}
+ @Test
+ public void testReadDeltaLakeTableAtTimestamp() throws Exception {
+ ExperimentalOptions options =
readPipeline.getOptions().as(ExperimentalOptions.class);
+ ExperimentalOptions.addExperiment(options, "use_runner_v2");
+
+ Map<String, String> hadoopConfig = getHadoopConfig();
+ Configuration conf = new Configuration();
+ hadoopConfig.forEach(conf::set);
+ Engine engine = DefaultEngine.create(conf);
+
+ Table table = Table.forPath(engine, repoPath);
+ long commitTimestampV0 = table.getSnapshotAsOfVersion(engine,
0L).getTimestamp(engine);
+ String timestampV0 =
java.time.Instant.ofEpochMilli(commitTimestampV0).toString();
+
+ // Write version 1 with additional rows
+ List<Row> 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<Row> 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<String, String> hadoopConfig = getHadoopConfig();
+ Configuration conf = new Configuration();
+ hadoopConfig.forEach(conf::set);
+ Engine engine = DefaultEngine.create(conf);
+
+ // Write version 1 with additional rows
+ List<Row> 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<Row> 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);
@@ -314,24 +371,9 @@ public class DeltaIOIT {
options.setExperiments(modifiableExperiments);
}
- Map<String, String> 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<String, String> entry : hadoopConfig.entrySet()) {
- conf.set(entry.getKey(), entry.getValue());
- }
+ Map<String, String> hadoopConfig = getHadoopConfig();
+ Configuration conf = new Configuration();
+ hadoopConfig.forEach(conf::set);
Engine engine = DefaultEngine.create(conf);
StructType deltaSchema =
@@ -413,6 +455,19 @@ public class DeltaIOIT {
readPipeline.run().waitUntilFinish();
}
+ private Map<String, String> getHadoopConfig() {
+ Map<String, String> 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(GcpOptions.class).getProject();
+ if (project != null) {
+ hadoopConfig.put("fs.gs.project.id", project);
+ }
+ return hadoopConfig;
+ }
+
private static final class FormatITRowWithMetadata extends DoFn<Row, String>
{
@ProcessElement
public void process(@Element Row row, OutputReceiver<String> out) {
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 0db0aef9e08..1b7f73566c1 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
@@ -38,6 +38,7 @@ import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.util.Collections;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import org.apache.avro.generic.GenericRecord;
import org.apache.beam.sdk.extensions.avro.coders.AvroCoder;
@@ -109,6 +110,107 @@ public class DeltaIOTest {
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());
+
+ List<Row> rows = DeltaWriteTestUtils.setupTwoVersionTable(engine,
tableDir.getAbsolutePath());
+ Row row1 = rows.get(0);
+ Row row2 = rows.get(1);
+
+ // Read at version 0
+ PCollection<Row> 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());
+
+ List<Row> rows = DeltaWriteTestUtils.setupTwoVersionTable(engine,
tableDir.getAbsolutePath());
+ Row row1 = rows.get(0);
+ Row row2 = rows.get(1);
+
+ // Read at timestamp between version 0 and version 1
+ String timestampV0 =
java.time.Instant.ofEpochMilli(150000000000L).toString();
+ PCollection<Row> 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());
+
+ List<Row> rows = DeltaWriteTestUtils.setupTwoVersionTable(engine,
tableDir.getAbsolutePath());
+ Row row1 = rows.get(0);
+ Row row2 = rows.get(1);
+
+ // Read version 0 using Managed
+ PCollection<Row> output =
+ readPipeline
+ .apply(
+ Managed.read(Managed.DELTA_LAKE)
+ .withConfig(
+ ImmutableMap.of("table", tableDir.getAbsolutePath(),
"version", 0L)))
+ .getSinglePCollection();
+
+ PAssert.that(output).containsInAnyOrder(row1, row2);
+
+ 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());
+
+ List<Row> rows = DeltaWriteTestUtils.setupTwoVersionTable(engine,
tableDir.getAbsolutePath());
+ Row row1 = rows.get(0);
+ Row row2 = rows.get(1);
+
+ // Read timestamp after version 0 using Managed
+ String timestampV0 =
java.time.Instant.ofEpochMilli(150000000000L).toString();
+ PCollection<Row> output =
+ readPipeline
+ .apply(
+ Managed.read(Managed.DELTA_LAKE)
+ .withConfig(
+ ImmutableMap.of(
+ "table", tableDir.getAbsolutePath(), "timestamp",
timestampV0)))
+ .getSinglePCollection();
+
+ PAssert.that(output).containsInAnyOrder(row1, row2);
+
+ 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 77aef7bce49..2e2060b81ea 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,9 +20,12 @@ 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 io.delta.kernel.defaults.engine.DefaultEngine;
+import io.delta.kernel.engine.Engine;
import java.io.File;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
+import java.util.List;
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;
@@ -124,4 +127,52 @@ public class DeltaReadSchemaTransformProviderTest {
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());
+
+ List<Row> rows = DeltaWriteTestUtils.setupTwoVersionTable(engine,
tableDir.getAbsolutePath());
+ Row row1 = rows.get(0);
+ Row row2 = rows.get(1);
+
+ Configuration readConfig =
+
Configuration.builder().setTable(tableDir.getAbsolutePath()).setVersion(0L).build();
+
+ PCollection<Row> output =
+ PCollectionRowTuple.empty(readPipeline)
+ .apply(new DeltaReadSchemaTransformProvider().from(readConfig))
+ .get(OUTPUT_TAG);
+
+ PAssert.that(output).containsInAnyOrder(row1, row2);
+
+ 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());
+
+ List<Row> rows = DeltaWriteTestUtils.setupTwoVersionTable(engine,
tableDir.getAbsolutePath());
+ Row row1 = rows.get(0);
+ Row row2 = rows.get(1);
+
+ String timestampV0 =
java.time.Instant.ofEpochMilli(150000000000L).toString();
+ Configuration readConfig =
+ Configuration.builder()
+ .setTable(tableDir.getAbsolutePath())
+ .setTimestamp(timestampV0)
+ .build();
+
+ PCollection<Row> output =
+ PCollectionRowTuple.empty(readPipeline)
+ .apply(new DeltaReadSchemaTransformProvider().from(readConfig))
+ .get(OUTPUT_TAG);
+
+ PAssert.that(output).containsInAnyOrder(row1, row2);
+
+ readPipeline.run().waitUntilFinish();
+ }
}
diff --git
a/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaWriteTestUtils.java
b/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaWriteTestUtils.java
index 4ae75bcd47c..55646de749f 100644
---
a/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaWriteTestUtils.java
+++
b/sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaWriteTestUtils.java
@@ -48,6 +48,7 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import javax.annotation.Nullable;
+import org.apache.beam.sdk.schemas.Schema;
import org.apache.beam.sdk.values.Row;
import org.joda.time.Instant;
@@ -368,4 +369,33 @@ final class DeltaWriteTestUtils {
commitFile.setLastModified(timestamp);
}
}
+
+ /**
+ * Sets up a Delta table with two commit versions containing test rows.
+ *
+ * <p>Version 0 is committed at timestamp 100000000000L with rows ["row-1",
"row-2"]. Version 1 is
+ * committed at timestamp 200000000000L with row ["row-3"].
+ *
+ * @param engine the Delta Lake {@link Engine} instance to use
+ * @param tablePath the path of the Delta table to create
+ * @return the list of {@link Row} objects written [row1, row2, row3]
+ * @throws Exception if any error occurs during write or commit
+ */
+ static List<Row> setupTwoVersionTable(Engine engine, String tablePath)
throws Exception {
+ 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
+ writeAppendCommit(
+ engine, tablePath, 0L, 100000000000L, deltaSchema,
java.util.Arrays.asList(row1, row2));
+
+ // Commit version 1
+ writeAppendCommit(
+ engine, tablePath, 1L, 200000000000L, deltaSchema,
java.util.Arrays.asList(row3));
+
+ return java.util.Arrays.asList(row1, row2, row3);
+ }
}
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 70b5efe9ed1..5eb6f04ab80 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.
<code style="color: green">str</code>
</td>
<td>
- Timestamp of the Delta Lake table to read.
+ Timestamp of the Delta Lake table to read (in UTC ISO 8601 format,
e.g. <code>2026-05-20T15:43:26Z</code>). Cannot be set if version is set.
</td>
</tr>
<tr>
@@ -315,7 +315,7 @@ and Beam SQL is invoked via the Managed API under the hood.
<code style="color: #f54251">int64</code>
</td>
<td>
- Version of the Delta Lake table to read.
+ Version of the Delta Lake table to read. Cannot be set if timestamp is
set.
</td>
</tr>
</table>
