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..ab4daeae2349 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 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, OutputReceiver 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 +145,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 +164,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..ab7fd8e24f4c 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.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 void setup() throws Exception { 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 void testReadDeltaLakeTable() { 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"); - 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 hadoopConfig = getHadoopConfig(); PCollection output = readPipeline @@ -302,6 +280,85 @@ 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 = 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 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 = getHadoopConfig(); + Configuration conf = new Configuration(); + hadoopConfig.forEach(conf::set); + 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); @@ -314,24 +371,9 @@ public void testReadChangesDeltaLake() throws Exception { options.setExperiments(modifiableExperiments); } - 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()); - } + Map hadoopConfig = getHadoopConfig(); + Configuration conf = new Configuration(); + hadoopConfig.forEach(conf::set); Engine engine = DefaultEngine.create(conf); StructType deltaSchema = @@ -413,6 +455,19 @@ public void testReadChangesDeltaLake() throws Exception { readPipeline.run().waitUntilFinish(); } + private Map getHadoopConfig() { + 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(GcpOptions.class).getProject(); + if (project != null) { + hadoopConfig.put("fs.gs.project.id", project); + } + return hadoopConfig; + } + private static final class FormatITRowWithMetadata extends DoFn { @ProcessElement public void process(@Element Row row, OutputReceiver 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 0db0aef9e080..1b7f73566c16 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.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 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()); + + List rows = DeltaWriteTestUtils.setupTwoVersionTable(engine, tableDir.getAbsolutePath()); + Row row1 = rows.get(0); + Row row2 = rows.get(1); + + // 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()); + + List 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 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 rows = DeltaWriteTestUtils.setupTwoVersionTable(engine, tableDir.getAbsolutePath()); + Row row1 = rows.get(0); + Row row2 = rows.get(1); + + // 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, 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 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 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 77aef7bce494..2e2060b81ead 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 @@ 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 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()); + + List 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 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 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 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 4ae75bcd47cd..55646de749f1 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.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 @@ static void writeCdcCommit( commitFile.setLastModified(timestamp); } } + + /** + * Sets up a Delta table with two commit versions containing test rows. + * + *

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 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 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.