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 81b9188f409 Adds Delta Lake read support to Beam SQL (#39950)
81b9188f409 is described below
commit 81b9188f40919ae617a77db616c17543b2d9bb9e
Author: Chamikara Jayalath <[email protected]>
AuthorDate: Wed Sep 9 09:36:15 2026 -0700
Adds Delta Lake read support to Beam SQL (#39950)
---
.../beam_PostCommit_Java_Delta_IO_Dataflow.json | 2 +-
build.gradle.kts | 1 +
sdks/java/extensions/sql/delta/build.gradle | 50 +++
.../sql/meta/provider/delta/DeltaTable.java | 199 +++++++++
.../meta/provider/delta/DeltaTableProvider.java | 75 ++++
.../sql/meta/provider/delta/package-info.java | 20 +
.../sql/meta/provider/delta/BeamSqlDeltaTest.java | 475 +++++++++++++++++++++
.../provider/delta/DeltaTableProviderTest.java | 280 ++++++++++++
sdks/java/io/delta/build.gradle | 1 -
.../beam/sdk/io/delta/DeltaCDCSourceDoFn.java | 2 +-
.../java/org/apache/beam/sdk/io/delta/DeltaIO.java | 11 +-
.../apache/beam/sdk/io/delta/DeltaSourceDoFn.java | 30 +-
.../org/apache/beam/sdk/io/delta/DeltaIOIT.java | 17 +-
.../org/apache/beam/sdk/io/delta/DeltaIOTest.java | 145 +++++--
.../beam/sdk/io/delta/DeltaWriteTestUtils.java | 58 ++-
settings.gradle.kts | 3 +
16 files changed, 1304 insertions(+), 65 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 44932dfca50..0ca37f8c8e2 100644
--- a/.github/trigger_files/beam_PostCommit_Java_Delta_IO_Dataflow.json
+++ b/.github/trigger_files/beam_PostCommit_Java_Delta_IO_Dataflow.json
@@ -1,5 +1,5 @@
{
"comment": "Modify this file in a trivial way to cause this test suite to
run.",
- "modification": 3,
+ "modification": 1,
"https://github.com/apache/beam/pull/39990": "removing dead code from
FnApiDoFnRunner"
}
diff --git a/build.gradle.kts b/build.gradle.kts
index 8de3f762fb7..cbdc35f3369 100644
--- a/build.gradle.kts
+++ b/build.gradle.kts
@@ -396,6 +396,7 @@ tasks.register("sqlPreCommit") {
dependsOn(":sdks:java:extensions:sql:expansion-service:build")
dependsOn(":sdks:java:extensions:sql:hcatalog:build")
dependsOn(":sdks:java:extensions:sql:iceberg:build")
+ dependsOn(":sdks:java:extensions:sql:delta:build")
dependsOn(":sdks:java:extensions:sql:jdbc:build")
dependsOn(":sdks:java:extensions:sql:jdbc:preCommit")
dependsOn(":sdks:java:extensions:sql:perf-tests:build")
diff --git a/sdks/java/extensions/sql/delta/build.gradle
b/sdks/java/extensions/sql/delta/build.gradle
new file mode 100644
index 00000000000..07cce842854
--- /dev/null
+++ b/sdks/java/extensions/sql/delta/build.gradle
@@ -0,0 +1,50 @@
+/*
+ * 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.
+ */
+
+plugins { id 'org.apache.beam.module' }
+
+applyJavaNature(
+ automaticModuleName:
'org.apache.beam.sdk.extensions.sql.meta.provider.delta',
+ requireJavaVersion: JavaVersion.VERSION_17,
+)
+
+description = "Apache Beam :: SDKs :: Java :: Extensions :: SQL :: Delta Lake"
+ext.summary = "Delta Lake table provider for Beam SQL."
+
+dependencies {
+ implementation project(":sdks:java:extensions:sql")
+ implementation project(":sdks:java:core")
+ implementation project(":sdks:java:managed")
+ implementation project(":sdks:java:io:delta")
+ implementation library.java.jackson_databind
+ implementation library.java.jackson_core
+ implementation library.java.slf4j_api
+ implementation library.java.vendored_guava_32_1_2_jre
+ implementation library.java.vendored_calcite_1_40_0
+ // TODO(https://github.com/apache/beam/issues/21156): Determine how to build
without this dependency
+ provided "org.immutables:value:2.8.8"
+ permitUnusedDeclared "org.immutables:value:2.8.8"
+ permitUnusedDeclared project(":sdks:java:io:delta")
+ permitUnusedDeclared library.java.slf4j_api
+
+ testImplementation library.java.joda_time
+ testImplementation library.java.junit
+ testImplementation library.java.hadoop_common
+ testImplementation project(path: ":sdks:java:io:delta", configuration:
"testRuntimeMigration")
+ testImplementation project(path: ":runners:direct-java", configuration:
"shadow")
+}
diff --git
a/sdks/java/extensions/sql/delta/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/delta/DeltaTable.java
b/sdks/java/extensions/sql/delta/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/delta/DeltaTable.java
new file mode 100644
index 00000000000..290453a614c
--- /dev/null
+++
b/sdks/java/extensions/sql/delta/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/delta/DeltaTable.java
@@ -0,0 +1,199 @@
+/*
+ * 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.extensions.sql.meta.provider.delta;
+
+import static org.apache.beam.sdk.util.Preconditions.checkArgumentNotNull;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.beam.sdk.extensions.sql.TableUtils;
+import org.apache.beam.sdk.extensions.sql.meta.BeamSqlTableFilter;
+import org.apache.beam.sdk.extensions.sql.meta.DefaultTableFilter;
+import org.apache.beam.sdk.extensions.sql.meta.SchemaBaseBeamTable;
+import org.apache.beam.sdk.extensions.sql.meta.Table;
+import org.apache.beam.sdk.managed.Managed;
+import org.apache.beam.sdk.values.PBegin;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.POutput;
+import org.apache.beam.sdk.values.Row;
+import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.rex.RexNode;
+import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+class DeltaTable extends SchemaBaseBeamTable {
+ @VisibleForTesting static final String VERSION_FIELD = "version";
+ @VisibleForTesting static final String TIMESTAMP_FIELD = "timestamp";
+ @VisibleForTesting static final String HADOOP_CONFIG_FIELD = "hadoop_config";
+ @VisibleForTesting static final String HADOOP_CONFIG_CAMEL_FIELD =
"hadoopConfig";
+
+ static final String BEAM_WRITE_PROPERTY = "beam.write.";
+ static final String BEAM_READ_PROPERTY = "beam.read.";
+
+ @VisibleForTesting final String tableLocation;
+ @VisibleForTesting final @Nullable Long version;
+ @VisibleForTesting final @Nullable String timestamp;
+ @VisibleForTesting final @Nullable Map<String, String> hadoopConfig;
+
+ DeltaTable(Table table) {
+ this(
+ checkArgumentNotNull(
+ table.getLocation(),
+ "Delta Lake table location must be specified (catalog-based tables
are not supported)."),
+ table);
+ }
+
+ DeltaTable(String tableLocation, Table table) {
+ super(table.getSchema());
+ this.schema = table.getSchema();
+ this.tableLocation = tableLocation;
+
+ Long parsedVersion = null;
+ String parsedTimestamp = null;
+ Map<String, String> parsedHadoopConfig = new HashMap<>();
+
+ ObjectNode properties = table.getProperties();
+ for (Map.Entry<String, JsonNode> property : properties.properties()) {
+ String key = property.getKey();
+ String lowerKey = key.toLowerCase();
+ JsonNode val = property.getValue();
+
+ if (lowerKey.startsWith(BEAM_WRITE_PROPERTY)) {
+ // TODO: Support writing to Delta Lake tables once a Delta Lake sink is
+ // available.
+ throw new IllegalArgumentException(
+ String.format(
+ "Beam write property '%s' is not supported. Writing to Delta
Lake tables is currently not supported.",
+ key));
+ } else if (lowerKey.startsWith(BEAM_READ_PROPERTY)) {
+ // none supported yet
+ throw new IllegalArgumentException("Unknown Beam read property: " +
key);
+ } else if (lowerKey.equalsIgnoreCase(VERSION_FIELD)) {
+ parsedVersion = parseVersion(val);
+ } else if (lowerKey.equalsIgnoreCase(TIMESTAMP_FIELD)) {
+ parsedTimestamp = val.asText();
+ } else if (lowerKey.equalsIgnoreCase(HADOOP_CONFIG_FIELD)
+ || lowerKey.equalsIgnoreCase(HADOOP_CONFIG_CAMEL_FIELD)) {
+ parseHadoopConfig(val, parsedHadoopConfig);
+ } else {
+ throw new IllegalArgumentException(String.format("Unknown property
'%s'", key));
+ }
+ }
+
+ if (parsedVersion != null && parsedTimestamp != null) {
+ throw new IllegalArgumentException("Cannot set both version and
timestamp.");
+ }
+
+ this.version = parsedVersion;
+ this.timestamp = parsedTimestamp;
+ this.hadoopConfig = parsedHadoopConfig.isEmpty() ? null :
parsedHadoopConfig;
+ }
+
+ private static Long parseVersion(JsonNode val) {
+ if (val.isNumber()) {
+ return val.asLong();
+ }
+ return Long.parseLong(val.asText());
+ }
+
+ private static void parseHadoopConfig(JsonNode val, Map<String, String>
targetMap) {
+ if (val.isObject()) {
+ Map<String, String> map =
+ TableUtils.getObjectMapper()
+ .convertValue(val, new TypeReference<Map<String, String>>() {});
+ if (map != null) {
+ targetMap.putAll(map);
+ }
+ } else if (val.isTextual()) {
+ try {
+ Map<String, String> map =
+ TableUtils.getObjectMapper()
+ .readValue(val.asText(), new TypeReference<Map<String,
String>>() {});
+ if (map != null) {
+ targetMap.putAll(map);
+ }
+ } catch (Exception e) {
+ throw new IllegalArgumentException("Failed to parse hadoop_config
string as JSON", e);
+ }
+ }
+ }
+
+ @Override
+ public PCollection<Row> buildIOReader(PBegin begin) {
+ return begin
+ .apply(Managed.read(Managed.DELTA_LAKE).withConfig(getBaseConfig()))
+ .getSinglePCollection();
+ }
+
+ @Override
+ public PCollection<Row> buildIOReader(
+ PBegin begin, BeamSqlTableFilter filters, List<String> fieldNames) {
+ // TODO: Support predicate pushdown and column pruning when supported by
DeltaIO
+ // / Managed Delta
+ // Lake source.
+ String error = "%s does not support predicate/project push-down, yet
non-empty %s is passed.";
+ if (!(filters instanceof DefaultTableFilter)) {
+ throw new UnsupportedOperationException(
+ String.format(error, this.getClass().getName(), "'filters'"));
+ }
+ if (!fieldNames.isEmpty()) {
+ throw new UnsupportedOperationException(
+ String.format(error, this.getClass().getName(), "'fieldNames'"));
+ }
+ return buildIOReader(begin);
+ }
+
+ @Override
+ public POutput buildIOWriter(PCollection<Row> input) {
+ // TODO: Support writing to Delta Lake tables once a Delta Lake sink is
+ // available.
+ throw new UnsupportedOperationException(
+ "Writing to Delta Lake tables is currently not supported.");
+ }
+
+ @Override
+ public PCollection.IsBounded isBounded() {
+ return PCollection.IsBounded.BOUNDED;
+ }
+
+ @Override
+ public BeamSqlTableFilter constructFilter(List<RexNode> filter) {
+ // TODO: Support predicate pushdown when supported by DeltaIO / Managed
Delta
+ // Lake source.
+ return new DefaultTableFilter(filter);
+ }
+
+ private Map<String, Object> getBaseConfig() {
+ ImmutableMap.Builder<String, Object> managedConfigBuilder =
ImmutableMap.builder();
+ managedConfigBuilder.put("table", tableLocation);
+ if (version != null) {
+ managedConfigBuilder.put(VERSION_FIELD, version);
+ }
+ if (timestamp != null) {
+ managedConfigBuilder.put(TIMESTAMP_FIELD, timestamp);
+ }
+ if (hadoopConfig != null && !hadoopConfig.isEmpty()) {
+ managedConfigBuilder.put(HADOOP_CONFIG_FIELD, hadoopConfig);
+ }
+ return managedConfigBuilder.build();
+ }
+}
diff --git
a/sdks/java/extensions/sql/delta/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/delta/DeltaTableProvider.java
b/sdks/java/extensions/sql/delta/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/delta/DeltaTableProvider.java
new file mode 100644
index 00000000000..7e803825754
--- /dev/null
+++
b/sdks/java/extensions/sql/delta/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/delta/DeltaTableProvider.java
@@ -0,0 +1,75 @@
+/*
+ * 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.extensions.sql.meta.provider.delta;
+
+import static
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkArgument;
+
+import com.google.auto.service.AutoService;
+import org.apache.beam.sdk.extensions.sql.meta.BeamSqlTable;
+import org.apache.beam.sdk.extensions.sql.meta.Table;
+import
org.apache.beam.sdk.extensions.sql.meta.provider.InMemoryMetaTableProvider;
+import org.apache.beam.sdk.extensions.sql.meta.provider.TableProvider;
+
+/**
+ * {@link TableProvider} for Delta Lake tables.
+ *
+ * <p>A sample of Delta Lake table registration is:
+ *
+ * <pre>{@code
+ * CREATE EXTERNAL TABLE orders(
+ * id INTEGER,
+ * name VARCHAR,
+ * amount DOUBLE
+ * )
+ * TYPE 'delta'
+ * LOCATION '/path/to/delta/orders'
+ * TBLPROPERTIES '{"version": 1}'
+ * }</pre>
+ */
+@AutoService(TableProvider.class)
+public class DeltaTableProvider extends InMemoryMetaTableProvider {
+
+ @Override
+ public String getTableType() {
+ return "delta";
+ }
+
+ @Override
+ public void createTable(Table table) {
+ // TODO: Support catalog-based Delta Lake tables once Delta catalog
support is implemented.
+ checkArgument(
+ table.getLocation() != null,
+ "Delta Lake table location must be specified (catalog-based tables are
not supported).");
+ super.createTable(table);
+ }
+
+ @Override
+ public BeamSqlTable buildBeamSqlTable(Table table) {
+ // TODO: Support catalog-based Delta Lake tables once Delta catalog
support is implemented.
+ checkArgument(
+ table.getLocation() != null,
+ "Delta Lake table location must be specified (catalog-based tables are
not supported).");
+ return new DeltaTable(table);
+ }
+
+ @Override
+ public boolean supportsPartitioning(Table table) {
+ // TODO: Support partitioning when Delta Lake connector supports
partitioned reads.
+ return false;
+ }
+}
diff --git
a/sdks/java/extensions/sql/delta/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/delta/package-info.java
b/sdks/java/extensions/sql/delta/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/delta/package-info.java
new file mode 100644
index 00000000000..3f2eca1d856
--- /dev/null
+++
b/sdks/java/extensions/sql/delta/src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/delta/package-info.java
@@ -0,0 +1,20 @@
+/*
+ * 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.
+ */
+
+/** Table schema for Delta Lake. */
+package org.apache.beam.sdk.extensions.sql.meta.provider.delta;
diff --git
a/sdks/java/extensions/sql/delta/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/delta/BeamSqlDeltaTest.java
b/sdks/java/extensions/sql/delta/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/delta/BeamSqlDeltaTest.java
new file mode 100644
index 00000000000..3ac9444e2f5
--- /dev/null
+++
b/sdks/java/extensions/sql/delta/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/delta/BeamSqlDeltaTest.java
@@ -0,0 +1,475 @@
+/*
+ * 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.extensions.sql.meta.provider.delta;
+
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+
+import io.delta.kernel.defaults.engine.DefaultEngine;
+import io.delta.kernel.engine.Engine;
+import io.delta.kernel.types.DateType;
+import io.delta.kernel.types.DoubleType;
+import io.delta.kernel.types.IntegerType;
+import io.delta.kernel.types.StringType;
+import io.delta.kernel.types.StructType;
+import io.delta.kernel.types.TimestampNTZType;
+import io.delta.kernel.types.TimestampType;
+import java.io.File;
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.util.Arrays;
+import java.util.List;
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.extensions.sql.BeamSqlCli;
+import org.apache.beam.sdk.extensions.sql.impl.BeamSqlEnv;
+import org.apache.beam.sdk.extensions.sql.impl.rel.BeamRelNode;
+import org.apache.beam.sdk.extensions.sql.impl.rel.BeamSqlRelUtils;
+import org.apache.beam.sdk.extensions.sql.meta.catalog.InMemoryCatalogManager;
+import org.apache.beam.sdk.io.delta.DeltaWriteTestUtils;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.schemas.logicaltypes.SqlTypes;
+import org.apache.beam.sdk.schemas.logicaltypes.Timestamp;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.Row;
+import org.apache.hadoop.conf.Configuration;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/** Integration/functional tests for Delta Lake read support in Beam SQL. */
+@RunWith(JUnit4.class)
+public class BeamSqlDeltaTest {
+
+ @Rule public TestPipeline readPipeline = TestPipeline.create();
+ @Rule public TemporaryFolder tempFolder = new TemporaryFolder();
+
+ private InMemoryCatalogManager catalogManager;
+ private BeamSqlCli cli;
+ private BeamSqlEnv sqlEnv;
+
+ private static final Schema PERSON_SCHEMA =
+ Schema.builder()
+ .addStringField("name")
+ .addInt32Field("age")
+ .addDoubleField("score")
+ .addStringField("country")
+ .build();
+
+ private static final StructType PERSON_DELTA_SCHEMA =
+ new StructType()
+ .add("name", StringType.STRING)
+ .add("age", IntegerType.INTEGER)
+ .add("score", DoubleType.DOUBLE)
+ .add("country", StringType.STRING);
+
+ private static final Row PERSON_1 =
+ Row.withSchema(PERSON_SCHEMA).addValues("Alice", 30, 95.5,
"USA").build();
+ private static final Row PERSON_2 =
+ Row.withSchema(PERSON_SCHEMA).addValues("Bob", 20, 70.0, "USA").build();
+ private static final Row PERSON_3 =
+ Row.withSchema(PERSON_SCHEMA).addValues("Charlie", 40, 85.0,
"Canada").build();
+
+ @Before
+ public void setUp() {
+ catalogManager = new InMemoryCatalogManager();
+ catalogManager.registerTableProvider(new DeltaTableProvider());
+ cli = new BeamSqlCli().catalogManager(catalogManager);
+ sqlEnv =
+ BeamSqlEnv.builder(catalogManager)
+ .setPipelineOptions(PipelineOptionsFactory.create())
+ .build();
+ }
+
+ private void createDeltaTable(File tableDir, List<Row> rows) throws
Exception {
+ createDeltaTable(tableDir, rows, PERSON_DELTA_SCHEMA);
+ }
+
+ private void createDeltaTable(File tableDir, List<Row> rows, StructType
deltaSchema)
+ throws Exception {
+ Engine engine = DefaultEngine.create(new Configuration());
+ DeltaWriteTestUtils.writeAppendCommit(
+ engine, tableDir.getAbsolutePath(), 0L, 100000000000L, deltaSchema,
rows);
+ }
+
+ private void createTwoVersionDeltaTable(File tableDir) throws Exception {
+ Engine engine = DefaultEngine.create(new Configuration());
+ Schema schema = Schema.builder().addField("name",
Schema.FieldType.STRING).build();
+ StructType deltaSchema = new StructType().add("name", StringType.STRING);
+
+ Row row1 = Row.withSchema(schema).addValues("v0-row1").build();
+ Row row2 = Row.withSchema(schema).addValues("v0-row2").build();
+ Row row3 = Row.withSchema(schema).addValues("v1-row3").build();
+
+ // Commit 0 (version 0 at timestamp 100000000000L)
+ DeltaWriteTestUtils.writeAppendCommit(
+ engine,
+ tableDir.getAbsolutePath(),
+ 0L,
+ 100000000000L,
+ deltaSchema,
+ Arrays.asList(row1, row2));
+
+ // Commit 1 (version 1 at timestamp 200000000000L)
+ DeltaWriteTestUtils.writeAppendCommit(
+ engine, tableDir.getAbsolutePath(), 1L, 200000000000L, deltaSchema,
Arrays.asList(row3));
+ }
+
+ @Test
+ public void testSelectAll() throws Exception {
+ File tableDir = tempFolder.newFolder("delta-table-all");
+ createDeltaTable(tableDir, Arrays.asList(PERSON_1, PERSON_2, PERSON_3));
+
+ sqlEnv.executeDdl(
+ String.format(
+ "CREATE EXTERNAL TABLE persons (\n"
+ + " name VARCHAR,\n"
+ + " age INTEGER,\n"
+ + " score DOUBLE,\n"
+ + " country VARCHAR\n"
+ + ")\n"
+ + "TYPE 'delta'\n"
+ + "LOCATION '%s'",
+ tableDir.getAbsolutePath()));
+
+ BeamRelNode relNode = sqlEnv.parseQuery("SELECT * FROM persons");
+ PCollection<Row> output = BeamSqlRelUtils.toPCollection(readPipeline,
relNode);
+
+ PAssert.that(output).containsInAnyOrder(PERSON_1, PERSON_2, PERSON_3);
+ readPipeline.run().waitUntilFinish();
+ }
+
+ @Test
+ public void testSelectWithProjection() throws Exception {
+ File tableDir = tempFolder.newFolder("delta-table-project");
+ createDeltaTable(tableDir, Arrays.asList(PERSON_1, PERSON_2, PERSON_3));
+
+ sqlEnv.executeDdl(
+ String.format(
+ "CREATE EXTERNAL TABLE persons (\n"
+ + " name VARCHAR,\n"
+ + " age INTEGER,\n"
+ + " score DOUBLE,\n"
+ + " country VARCHAR\n"
+ + ")\n"
+ + "TYPE 'delta'\n"
+ + "LOCATION '%s'",
+ tableDir.getAbsolutePath()));
+
+ BeamRelNode relNode = sqlEnv.parseQuery("SELECT name, score FROM persons");
+ PCollection<Row> output = BeamSqlRelUtils.toPCollection(readPipeline,
relNode);
+
+ Schema projectedSchema =
+
Schema.builder().addStringField("name").addDoubleField("score").build();
+
+ PAssert.that(output)
+ .containsInAnyOrder(
+ Row.withSchema(projectedSchema).addValues("Alice", 95.5).build(),
+ Row.withSchema(projectedSchema).addValues("Bob", 70.0).build(),
+ Row.withSchema(projectedSchema).addValues("Charlie",
85.0).build());
+
+ readPipeline.run().waitUntilFinish();
+ }
+
+ @Test
+ public void testSelectWithFilter() throws Exception {
+ File tableDir = tempFolder.newFolder("delta-table-filter");
+ createDeltaTable(tableDir, Arrays.asList(PERSON_1, PERSON_2, PERSON_3));
+
+ sqlEnv.executeDdl(
+ String.format(
+ "CREATE EXTERNAL TABLE persons (\n"
+ + " name VARCHAR,\n"
+ + " age INTEGER,\n"
+ + " score DOUBLE,\n"
+ + " country VARCHAR\n"
+ + ")\n"
+ + "TYPE 'delta'\n"
+ + "LOCATION '%s'",
+ tableDir.getAbsolutePath()));
+
+ BeamRelNode relNode = sqlEnv.parseQuery("SELECT * FROM persons WHERE score
>= 85.0");
+ PCollection<Row> output = BeamSqlRelUtils.toPCollection(readPipeline,
relNode);
+
+ PAssert.that(output).containsInAnyOrder(PERSON_1, PERSON_3);
+ readPipeline.run().waitUntilFinish();
+ }
+
+ @Test
+ public void testSelectWithAggregation() throws Exception {
+ File tableDir = tempFolder.newFolder("delta-table-agg");
+ createDeltaTable(tableDir, Arrays.asList(PERSON_1, PERSON_2, PERSON_3));
+
+ sqlEnv.executeDdl(
+ String.format(
+ "CREATE EXTERNAL TABLE persons (\n"
+ + " name VARCHAR,\n"
+ + " age INTEGER,\n"
+ + " score DOUBLE,\n"
+ + " country VARCHAR\n"
+ + ")\n"
+ + "TYPE 'delta'\n"
+ + "LOCATION '%s'",
+ tableDir.getAbsolutePath()));
+
+ BeamRelNode relNode =
+ sqlEnv.parseQuery(
+ "SELECT country, COUNT(*) as cnt, AVG(score) as avg_score FROM
persons GROUP BY country");
+ PCollection<Row> output = BeamSqlRelUtils.toPCollection(readPipeline,
relNode);
+
+ Schema aggSchema =
+ Schema.builder()
+ .addStringField("country")
+ .addInt64Field("cnt")
+ .addDoubleField("avg_score")
+ .build();
+
+ PAssert.that(output)
+ .containsInAnyOrder(
+ Row.withSchema(aggSchema).addValues("USA", 2L, 82.75).build(),
+ Row.withSchema(aggSchema).addValues("Canada", 1L, 85.0).build());
+
+ readPipeline.run().waitUntilFinish();
+ }
+
+ @Test
+ public void testReadWithVersionTimeTravel() throws Exception {
+ File tableDir = tempFolder.newFolder("delta-table-ver");
+ createTwoVersionDeltaTable(tableDir);
+
+ Schema schema = Schema.builder().addField("name",
Schema.FieldType.STRING).build();
+ Row row1 = Row.withSchema(schema).addValues("v0-row1").build();
+ Row row2 = Row.withSchema(schema).addValues("v0-row2").build();
+
+ sqlEnv.executeDdl(
+ String.format(
+ "CREATE EXTERNAL TABLE table_v0 (\n"
+ + " name VARCHAR\n"
+ + ")\n"
+ + "TYPE 'delta'\n"
+ + "LOCATION '%s'\n"
+ + "TBLPROPERTIES '{\"version\": 0}'",
+ tableDir.getAbsolutePath()));
+
+ BeamRelNode relNode = sqlEnv.parseQuery("SELECT * FROM table_v0");
+ PCollection<Row> output = BeamSqlRelUtils.toPCollection(readPipeline,
relNode);
+
+ PAssert.that(output).containsInAnyOrder(row1, row2);
+ readPipeline.run().waitUntilFinish();
+ }
+
+ @Test
+ public void testReadWithTimestampTimeTravel() throws Exception {
+ File tableDir = tempFolder.newFolder("delta-table-ts");
+ createTwoVersionDeltaTable(tableDir);
+
+ Schema schema = Schema.builder().addField("name",
Schema.FieldType.STRING).build();
+ Row row1 = Row.withSchema(schema).addValues("v0-row1").build();
+ Row row2 = Row.withSchema(schema).addValues("v0-row2").build();
+
+ String timestampV0 =
java.time.Instant.ofEpochMilli(150000000000L).toString();
+
+ sqlEnv.executeDdl(
+ String.format(
+ "CREATE EXTERNAL TABLE table_ts (\n"
+ + " name VARCHAR\n"
+ + ")\n"
+ + "TYPE 'delta'\n"
+ + "LOCATION '%s'\n"
+ + "TBLPROPERTIES '{\"timestamp\": \"%s\"}'",
+ tableDir.getAbsolutePath(), timestampV0));
+
+ BeamRelNode relNode = sqlEnv.parseQuery("SELECT * FROM table_ts");
+ PCollection<Row> output = BeamSqlRelUtils.toPCollection(readPipeline,
relNode);
+
+ PAssert.that(output).containsInAnyOrder(row1, row2);
+ readPipeline.run().waitUntilFinish();
+ }
+
+ @Test
+ public void testInsertIntoTableFails() throws Exception {
+ File tableDir = tempFolder.newFolder("delta-table-insert");
+ createDeltaTable(tableDir, Arrays.asList(PERSON_1));
+
+ sqlEnv.executeDdl(
+ String.format(
+ "CREATE EXTERNAL TABLE persons (\n"
+ + " name VARCHAR,\n"
+ + " age INTEGER,\n"
+ + " score DOUBLE,\n"
+ + " country VARCHAR\n"
+ + ")\n"
+ + "TYPE 'delta'\n"
+ + "LOCATION '%s'",
+ tableDir.getAbsolutePath()));
+
+ BeamRelNode insertRel =
+ sqlEnv.parseQuery("INSERT INTO persons VALUES ('Dave', 28, 88.0,
'UK')");
+
+ Pipeline pipeline = Pipeline.create();
+ UnsupportedOperationException exception =
+ assertThrows(
+ UnsupportedOperationException.class,
+ () -> BeamSqlRelUtils.toPCollection(pipeline, insertRel));
+
+ assertTrue(
+ exception
+ .getMessage()
+ .contains("Writing to Delta Lake tables is currently not
supported."));
+ }
+
+ @Test
+ public void testCreateCatalogFails() {
+ UnsupportedOperationException exception =
+ assertThrows(
+ UnsupportedOperationException.class,
+ () ->
+ cli.execute(
+ "CREATE CATALOG delta_cat TYPE delta PROPERTIES
('warehouse' = '/path')"));
+
+ assertTrue(
+ exception.getMessage().contains("Could not find type 'delta' for
catalog 'delta_cat'."));
+ }
+
+ @Test
+ public void testCreateTableWithoutLocationFails() {
+ assertThrows(
+ Exception.class,
+ () -> sqlEnv.executeDdl("CREATE EXTERNAL TABLE no_location (name
VARCHAR) TYPE 'delta'"));
+ }
+
+ @Test
+ public void testSelectDateAndTimestamp() throws Exception {
+ File tableDir = tempFolder.newFolder("delta-table-date-ts");
+ Engine engine = DefaultEngine.create(new Configuration());
+
+ Schema beamSchema =
+ Schema.builder()
+ .addInt32Field("id")
+ .addField("ts_col", Schema.FieldType.logicalType(Timestamp.MICROS))
+ .addField("date_col", Schema.FieldType.logicalType(SqlTypes.DATE))
+ .build();
+
+ StructType deltaSchema =
+ new StructType()
+ .add("id", IntegerType.INTEGER)
+ .add("ts_col", TimestampType.TIMESTAMP)
+ .add("date_col", DateType.DATE);
+
+ Instant ts = Instant.parse("2026-08-31T12:00:00.000000Z");
+ LocalDate date = LocalDate.of(2026, 8, 31);
+ Row inputRow = Row.withSchema(beamSchema).addValues(1, ts, date).build();
+
+ DeltaWriteTestUtils.writeAppendCommit(
+ engine,
+ tableDir.getAbsolutePath(),
+ 0L,
+ 100000000000L,
+ deltaSchema,
+ Arrays.asList(inputRow));
+
+ sqlEnv.executeDdl(
+ String.format(
+ "CREATE EXTERNAL TABLE date_ts_table (\n"
+ + " id INTEGER,\n"
+ + " ts_col TIMESTAMP,\n"
+ + " date_col DATE\n"
+ + ")\n"
+ + "TYPE 'delta'\n"
+ + "LOCATION '%s'",
+ tableDir.getAbsolutePath()));
+
+ BeamRelNode relNode = sqlEnv.parseQuery("SELECT id, ts_col, date_col FROM
date_ts_table");
+ PCollection<Row> output = BeamSqlRelUtils.toPCollection(readPipeline,
relNode);
+
+ Schema sqlOutputSchema =
+ Schema.builder()
+ .addInt32Field("id")
+ .addField("ts_col", Schema.FieldType.DATETIME)
+ .addField("date_col", Schema.FieldType.logicalType(SqlTypes.DATE))
+ .build();
+ Row expectedRow =
+ Row.withSchema(sqlOutputSchema)
+ .addValues(1, new org.joda.time.Instant(ts.toEpochMilli()), date)
+ .build();
+
+ PAssert.that(output).containsInAnyOrder(expectedRow);
+ readPipeline.run().waitUntilFinish();
+ }
+
+ @Test
+ public void testSelectTimestampNtz() throws Exception {
+ File tableDir = tempFolder.newFolder("delta-table-ts-ntz");
+ Engine engine = DefaultEngine.create(new Configuration());
+
+ Schema beamSchema =
+ Schema.builder()
+ .addInt32Field("id")
+ .addField("ts_ntz_col",
Schema.FieldType.logicalType(SqlTypes.DATETIME))
+ .build();
+
+ StructType deltaSchema =
+ new StructType()
+ .add("id", IntegerType.INTEGER)
+ .add("ts_ntz_col", TimestampNTZType.TIMESTAMP_NTZ);
+
+ LocalDateTime dt = LocalDateTime.of(2026, 8, 31, 12, 0, 0);
+ Row inputRow = Row.withSchema(beamSchema).addValues(1, dt).build();
+
+ DeltaWriteTestUtils.writeAppendCommit(
+ engine,
+ tableDir.getAbsolutePath(),
+ 0L,
+ 100000000000L,
+ deltaSchema,
+ Arrays.asList(inputRow));
+
+ sqlEnv.executeDdl(
+ String.format(
+ "CREATE EXTERNAL TABLE ts_ntz_table (\n"
+ + " id INTEGER,\n"
+ + " ts_ntz_col TIMESTAMP\n"
+ + ")\n"
+ + "TYPE 'delta'\n"
+ + "LOCATION '%s'",
+ tableDir.getAbsolutePath()));
+
+ BeamRelNode relNode = sqlEnv.parseQuery("SELECT id, ts_ntz_col FROM
ts_ntz_table");
+ PCollection<Row> output = BeamSqlRelUtils.toPCollection(readPipeline,
relNode);
+
+ Schema sqlOutputSchema =
+ Schema.builder()
+ .addInt32Field("id")
+ .addField("ts_ntz_col", Schema.FieldType.DATETIME)
+ .build();
+ Instant dtInst = dt.toInstant(java.time.ZoneOffset.UTC);
+ Row expectedRow =
+ Row.withSchema(sqlOutputSchema)
+ .addValues(1, new org.joda.time.Instant(dtInst.toEpochMilli()))
+ .build();
+
+ PAssert.that(output).containsInAnyOrder(expectedRow);
+ readPipeline.run().waitUntilFinish();
+ }
+}
diff --git
a/sdks/java/extensions/sql/delta/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/delta/DeltaTableProviderTest.java
b/sdks/java/extensions/sql/delta/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/delta/DeltaTableProviderTest.java
new file mode 100644
index 00000000000..b48defa0b23
--- /dev/null
+++
b/sdks/java/extensions/sql/delta/src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/delta/DeltaTableProviderTest.java
@@ -0,0 +1,280 @@
+/*
+ * 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.extensions.sql.meta.provider.delta;
+
+import static org.apache.beam.sdk.schemas.Schema.toSchema;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+
+import java.util.Collections;
+import java.util.stream.Stream;
+import org.apache.beam.sdk.extensions.sql.TableUtils;
+import org.apache.beam.sdk.extensions.sql.meta.BeamSqlTable;
+import org.apache.beam.sdk.extensions.sql.meta.BeamSqlTableFilter;
+import org.apache.beam.sdk.extensions.sql.meta.DefaultTableFilter;
+import org.apache.beam.sdk.extensions.sql.meta.ProjectSupport;
+import org.apache.beam.sdk.extensions.sql.meta.Table;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.vendor.calcite.v1_40_0.org.apache.calcite.rex.RexNode;
+import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/** Unit tests for {@link DeltaTableProvider} and {@link DeltaTable}. */
+@RunWith(JUnit4.class)
+public class DeltaTableProviderTest {
+
+ private final DeltaTableProvider provider = new DeltaTableProvider();
+
+ private static Table fakeTable(String name, String location, String
properties) {
+ return Table.builder()
+ .name(name)
+ .comment(name + " table")
+ .location(location)
+ .schema(
+ Stream.of(
+ Schema.Field.nullable("id", Schema.FieldType.INT32),
+ Schema.Field.nullable("name", Schema.FieldType.STRING))
+ .collect(toSchema()))
+ .type("delta")
+ .properties(TableUtils.parseProperties(properties))
+ .build();
+ }
+
+ @Test
+ public void testGetTableType() {
+ assertEquals("delta", provider.getTableType());
+ }
+
+ @Test
+ public void testBuildBeamSqlTableBasic() {
+ Table table = fakeTable("my_table", "/path/to/delta/table", "{}");
+ BeamSqlTable sqlTable = provider.buildBeamSqlTable(table);
+
+ assertNotNull(sqlTable);
+ assertTrue(sqlTable instanceof DeltaTable);
+
+ DeltaTable deltaTable = (DeltaTable) sqlTable;
+ assertEquals("/path/to/delta/table", deltaTable.tableLocation);
+ assertNull(deltaTable.version);
+ assertNull(deltaTable.timestamp);
+ assertNull(deltaTable.hadoopConfig);
+ assertEquals(PCollection.IsBounded.BOUNDED, deltaTable.isBounded());
+ assertEquals(ProjectSupport.NONE, deltaTable.supportsProjects());
+ }
+
+ @Test
+ public void testBuildBeamSqlTableWithVersion() {
+ Table table = fakeTable("my_table", "/path/to/delta/table", "{\"version\":
5}");
+ BeamSqlTable sqlTable = provider.buildBeamSqlTable(table);
+
+ assertTrue(sqlTable instanceof DeltaTable);
+ DeltaTable deltaTable = (DeltaTable) sqlTable;
+ assertEquals(Long.valueOf(5L), deltaTable.version);
+ assertNull(deltaTable.timestamp);
+ }
+
+ @Test
+ public void testBuildBeamSqlTableWithTimestamp() {
+ Table table =
+ fakeTable("my_table", "/path/to/delta/table", "{\"timestamp\":
\"2026-05-20T15:43:26Z\"}");
+ BeamSqlTable sqlTable = provider.buildBeamSqlTable(table);
+
+ assertTrue(sqlTable instanceof DeltaTable);
+ DeltaTable deltaTable = (DeltaTable) sqlTable;
+ assertEquals("2026-05-20T15:43:26Z", deltaTable.timestamp);
+ assertNull(deltaTable.version);
+ }
+
+ @Test
+ public void testBuildBeamSqlTableWithHadoopConfig() {
+ Table table =
+ fakeTable(
+ "my_table",
+ "/path/to/delta/table",
+ "{\"hadoop_config\": {\"fs.gs.project.id\": \"my-project\",
\"foo\": \"bar\"}}");
+ BeamSqlTable sqlTable = provider.buildBeamSqlTable(table);
+
+ assertTrue(sqlTable instanceof DeltaTable);
+ DeltaTable deltaTable = (DeltaTable) sqlTable;
+ assertNotNull(deltaTable.hadoopConfig);
+ assertEquals("my-project",
deltaTable.hadoopConfig.get("fs.gs.project.id"));
+ assertEquals("bar", deltaTable.hadoopConfig.get("foo"));
+ }
+
+ @Test
+ public void testBuildBeamSqlTableWithoutLocationFails() {
+ Table table = fakeTable("my_table", null, "{}");
+ IllegalArgumentException exception =
+ assertThrows(IllegalArgumentException.class, () ->
provider.buildBeamSqlTable(table));
+ assertTrue(
+ exception
+ .getMessage()
+ .contains(
+ "Delta Lake table location must be specified (catalog-based
tables are not supported)."));
+ }
+
+ @Test
+ public void testCreateTableWithoutLocationFails() {
+ Table table = fakeTable("my_table", null, "{}");
+ IllegalArgumentException exception =
+ assertThrows(IllegalArgumentException.class, () ->
provider.createTable(table));
+ assertTrue(
+ exception
+ .getMessage()
+ .contains(
+ "Delta Lake table location must be specified (catalog-based
tables are not supported)."));
+ }
+
+ @Test
+ public void testBuildBeamSqlTableWithBothVersionAndTimestampFails() {
+ Table table =
+ fakeTable(
+ "my_table",
+ "/path/to/delta/table",
+ "{\"version\": 1, \"timestamp\": \"2026-05-20T15:43:26Z\"}");
+ IllegalArgumentException exception =
+ assertThrows(IllegalArgumentException.class, () ->
provider.buildBeamSqlTable(table));
+ assertTrue(exception.getMessage().contains("Cannot set both version and
timestamp."));
+ }
+
+ @Test
+ public void testBuildBeamSqlTableWithWritePropertyFails() {
+ Table table =
+ fakeTable(
+ "my_table",
+ "/path/to/delta/table",
+ "{\"beam.write.triggering_frequency_seconds\": 30}");
+ IllegalArgumentException exception =
+ assertThrows(IllegalArgumentException.class, () ->
provider.buildBeamSqlTable(table));
+ assertTrue(
+ exception
+ .getMessage()
+ .contains("Writing to Delta Lake tables is currently not
supported."));
+ }
+
+ @Test
+ public void testDirectDeltaTableConstructor() {
+ Table table = fakeTable("my_table", null, "{\"version\": 2}");
+ DeltaTable deltaTable = new DeltaTable("/direct/path", table);
+
+ assertEquals("/direct/path", deltaTable.tableLocation);
+ assertEquals(Long.valueOf(2L), deltaTable.version);
+ assertNull(deltaTable.timestamp);
+ assertNull(deltaTable.hadoopConfig);
+ }
+
+ @Test
+ public void testBuildBeamSqlTableWithCamelCaseHadoopConfig() {
+ Table table =
+ fakeTable(
+ "my_table",
+ "/path/to/delta/table",
+ "{\"hadoopConfig\": {\"fs.gs.project.id\": \"my-project\"}}");
+ BeamSqlTable sqlTable = provider.buildBeamSqlTable(table);
+
+ assertTrue(sqlTable instanceof DeltaTable);
+ DeltaTable deltaTable = (DeltaTable) sqlTable;
+ assertNotNull(deltaTable.hadoopConfig);
+ assertEquals("my-project",
deltaTable.hadoopConfig.get("fs.gs.project.id"));
+ }
+
+ @Test
+ public void testBuildBeamSqlTableWithUnknownBeamReadPropertyFails() {
+ Table table = fakeTable("my_table", "/path/to/delta/table",
"{\"beam.read.unsupported\": 123}");
+ IllegalArgumentException exception =
+ assertThrows(IllegalArgumentException.class, () ->
provider.buildBeamSqlTable(table));
+ assertTrue(
+ exception.getMessage().contains("Unknown Beam read property:
beam.read.unsupported"));
+ }
+
+ @Test
+ public void testBuildBeamSqlTableWithUnknownPropertyFails() {
+ Table table = fakeTable("my_table", "/path/to/delta/table",
"{\"unsupported_property\": 123}");
+ IllegalArgumentException exception =
+ assertThrows(IllegalArgumentException.class, () ->
provider.buildBeamSqlTable(table));
+ assertTrue(exception.getMessage().contains("Unknown property
'unsupported_property'"));
+ }
+
+ @Test
+ public void testSupportsPartitioning() {
+ Table table = fakeTable("my_table", "/path/to/delta/table", "{}");
+ assertFalse(provider.supportsPartitioning(table));
+ }
+
+ @Test
+ public void testAlterTableFails() {
+ assertThrows(UnsupportedOperationException.class, () ->
provider.alterTable("my_table"));
+ }
+
+ @Test
+ public void testBuildIOWriterFails() {
+ Table table = fakeTable("my_table", "/path/to/delta/table", "{}");
+ DeltaTable deltaTable = (DeltaTable) provider.buildBeamSqlTable(table);
+ UnsupportedOperationException exception =
+ assertThrows(UnsupportedOperationException.class, () ->
deltaTable.buildIOWriter(null));
+ assertTrue(
+ exception
+ .getMessage()
+ .contains("Writing to Delta Lake tables is currently not
supported."));
+ }
+
+ @Test
+ public void testConstructFilterReturnsDefaultTableFilter() {
+ Table table = fakeTable("my_table", "/path/to/delta/table", "{}");
+ DeltaTable deltaTable = (DeltaTable) provider.buildBeamSqlTable(table);
+ BeamSqlTableFilter filter =
deltaTable.constructFilter(Collections.emptyList());
+ assertTrue(filter instanceof DefaultTableFilter);
+ assertEquals(0, filter.numSupported());
+ }
+
+ @Test
+ public void testBuildIOReaderWithPushDownFails() {
+ Table table = fakeTable("my_table", "/path/to/delta/table", "{}");
+ DeltaTable deltaTable = (DeltaTable) provider.buildBeamSqlTable(table);
+
+ BeamSqlTableFilter nonDefaultFilter =
+ new BeamSqlTableFilter() {
+ @Override
+ public java.util.List<RexNode> getNotSupported() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public int numSupported() {
+ return 1;
+ }
+ };
+
+ assertThrows(
+ UnsupportedOperationException.class,
+ () -> deltaTable.buildIOReader(null, nonDefaultFilter,
Collections.emptyList()));
+
+ assertThrows(
+ UnsupportedOperationException.class,
+ () ->
+ deltaTable.buildIOReader(
+ null, new DefaultTableFilter(Collections.emptyList()),
ImmutableList.of("id")));
+ }
+}
diff --git a/sdks/java/io/delta/build.gradle b/sdks/java/io/delta/build.gradle
index a0de04ed854..2787c4bb606 100644
--- a/sdks/java/io/delta/build.gradle
+++ b/sdks/java/io/delta/build.gradle
@@ -43,7 +43,6 @@ dependencies {
permitUnusedDeclared library.java.delta_kernel_defaults
implementation library.java.hadoop_common
- implementation library.java.joda_time
// implementation library.java.slf4j_api
implementation "org.apache.parquet:parquet-column:$parquet_version"
implementation "org.apache.parquet:parquet-hadoop:$parquet_version"
diff --git
a/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/DeltaCDCSourceDoFn.java
b/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/DeltaCDCSourceDoFn.java
index 97d9c914a08..fe1f31e5be1 100644
---
a/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/DeltaCDCSourceDoFn.java
+++
b/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/DeltaCDCSourceDoFn.java
@@ -248,7 +248,7 @@ class DeltaCDCSourceDoFn extends DoFn<DeltaCDCReadTask,
Row> {
if (field.getName().equals(DeltaIO.COMMIT_VERSION_COLUMN)) {
value = task.getVersion();
} else if (field.getName().equals(DeltaIO.COMMIT_TIMESTAMP_COLUMN)) {
- value = new org.joda.time.Instant(task.getTimestamp());
+ value = java.time.Instant.ofEpochMilli(task.getTimestamp());
}
}
builder.addValue(value);
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 4701ff59c4a..c63fdcacc1c 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
@@ -36,12 +36,15 @@ import io.delta.kernel.types.MapType;
import io.delta.kernel.types.StringType;
import io.delta.kernel.types.StructField;
import io.delta.kernel.types.StructType;
+import io.delta.kernel.types.TimestampNTZType;
import io.delta.kernel.types.TimestampType;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import org.apache.beam.sdk.annotations.Internal;
import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.schemas.logicaltypes.SqlTypes;
+import org.apache.beam.sdk.schemas.logicaltypes.Timestamp;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.transforms.ParDo;
@@ -197,9 +200,11 @@ public class DeltaIO {
} else if (deltaType instanceof BinaryType) {
return Schema.FieldType.BYTES;
} else if (deltaType instanceof TimestampType) {
- return Schema.FieldType.DATETIME;
+ return Schema.FieldType.logicalType(Timestamp.MICROS);
+ } else if (deltaType instanceof TimestampNTZType) {
+ return Schema.FieldType.logicalType(SqlTypes.DATETIME);
} else if (deltaType instanceof DateType) {
- return Schema.FieldType.DATETIME;
+ return Schema.FieldType.logicalType(SqlTypes.DATE);
} else if (deltaType instanceof ArrayType) {
DataType elementType = ((ArrayType) deltaType).getElementType();
return Schema.FieldType.iterable(convertToBeamFieldType(elementType));
@@ -230,7 +235,7 @@ public class DeltaIO {
} else if (col.equals(COMMIT_VERSION_COLUMN)) {
builder.addField(COMMIT_VERSION_COLUMN, Schema.FieldType.INT64);
} else if (col.equals(COMMIT_TIMESTAMP_COLUMN)) {
- builder.addField(COMMIT_TIMESTAMP_COLUMN, Schema.FieldType.DATETIME);
+ builder.addField(COMMIT_TIMESTAMP_COLUMN,
Schema.FieldType.logicalType(Timestamp.MICROS));
}
}
return builder.build();
diff --git
a/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/DeltaSourceDoFn.java
b/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/DeltaSourceDoFn.java
index fca19e30cb8..0b368739312 100644
---
a/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/DeltaSourceDoFn.java
+++
b/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/DeltaSourceDoFn.java
@@ -44,10 +44,15 @@ import io.delta.kernel.types.ShortType;
import io.delta.kernel.types.StringType;
import io.delta.kernel.types.StructField;
import io.delta.kernel.types.StructType;
+import io.delta.kernel.types.TimestampNTZType;
import io.delta.kernel.types.TimestampType;
import io.delta.kernel.utils.CloseableIterator;
import io.delta.kernel.utils.FileStatus;
import java.math.BigDecimal;
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.ZoneOffset;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
@@ -250,10 +255,18 @@ class DeltaSourceDoFn extends DoFn<DeltaReadTask, Row> {
return row.getBinary(index);
} else if (type instanceof TimestampType) {
long microSeconds = row.getLong(index);
- return new org.joda.time.Instant(microSeconds / 1000L);
+ return Instant.ofEpochSecond(
+ Math.floorDiv(microSeconds, 1_000_000L),
+ Math.floorMod(microSeconds, 1_000_000L) * 1_000L);
+ } else if (type instanceof TimestampNTZType) {
+ long microSeconds = row.getLong(index);
+ return LocalDateTime.ofEpochSecond(
+ Math.floorDiv(microSeconds, 1_000_000L),
+ (int) (Math.floorMod(microSeconds, 1_000_000L) * 1_000L),
+ ZoneOffset.UTC);
} else if (type instanceof DateType) {
int daysSinceEpoch = row.getInt(index);
- return new org.joda.time.Instant(daysSinceEpoch * 86400000L);
+ return LocalDate.ofEpochDay(daysSinceEpoch);
} else if (type instanceof ArrayType) {
ArrayValue arrayValue = row.getArray(index);
int size = arrayValue.getSize();
@@ -312,11 +325,18 @@ class DeltaSourceDoFn extends DoFn<DeltaReadTask, Row> {
return vector.getBinary(index);
} else if (type instanceof TimestampType) {
long microSeconds = vector.getLong(index);
- return new org.joda.time.Instant(microSeconds / 1000L);
+ return Instant.ofEpochSecond(
+ Math.floorDiv(microSeconds, 1_000_000L),
+ Math.floorMod(microSeconds, 1_000_000L) * 1_000L);
+ } else if (type instanceof TimestampNTZType) {
+ long microSeconds = vector.getLong(index);
+ return LocalDateTime.ofEpochSecond(
+ Math.floorDiv(microSeconds, 1_000_000L),
+ (int) (Math.floorMod(microSeconds, 1_000_000L) * 1_000L),
+ ZoneOffset.UTC);
} else if (type instanceof DateType) {
- // Convert days since epoch to milliseconds since epoch.
int daysSinceEpoch = vector.getInt(index);
- return new org.joda.time.Instant(daysSinceEpoch * 24L * 60L * 60L *
1000L);
+ return LocalDate.ofEpochDay(daysSinceEpoch);
} else if (type instanceof ArrayType) {
ArrayValue arrayValue = vector.getArray(index);
int size = arrayValue.getSize();
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 ab7fd8e24f4..602c9b7f151 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
@@ -53,6 +53,7 @@ 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;
+import org.apache.beam.sdk.schemas.logicaltypes.Timestamp;
import org.apache.beam.sdk.testing.PAssert;
import org.apache.beam.sdk.testing.TestPipeline;
import org.apache.beam.sdk.transforms.DoFn;
@@ -61,7 +62,6 @@ import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.Row;
import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
import org.apache.hadoop.conf.Configuration;
-import org.joda.time.Instant;
import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
@@ -386,7 +386,8 @@ public class DeltaIOIT {
.addField("name", Schema.FieldType.STRING)
.addField(DeltaIO.CHANGE_TYPE_COLUMN, Schema.FieldType.STRING)
.addField(DeltaIO.COMMIT_VERSION_COLUMN, Schema.FieldType.INT64)
- .addField(DeltaIO.COMMIT_TIMESTAMP_COLUMN,
Schema.FieldType.DATETIME)
+ .addField(
+ DeltaIO.COMMIT_TIMESTAMP_COLUMN,
Schema.FieldType.logicalType(Timestamp.MICROS))
.build();
StructType cdcWriteDeltaSchema =
new StructType()
@@ -398,15 +399,21 @@ public class DeltaIOIT {
Row cdcRow1 =
Row.withSchema(cdcWriteSchema)
- .addValues(0, "name_0", "delete", 1L, new Instant(123456789000L))
+ .addValues(0, "name_0", "delete", 1L,
java.time.Instant.ofEpochMilli(123456789000L))
.build();
Row cdcRow2 =
Row.withSchema(cdcWriteSchema)
- .addValues(1, "name_1", "update_preimage", 1L, new
Instant(123456789000L))
+ .addValues(
+ 1, "name_1", "update_preimage", 1L,
java.time.Instant.ofEpochMilli(123456789000L))
.build();
Row cdcRow3 =
Row.withSchema(cdcWriteSchema)
- .addValues(1, "name_1_updated", "update_postimage", 1L, new
Instant(123456789000L))
+ .addValues(
+ 1,
+ "name_1_updated",
+ "update_postimage",
+ 1L,
+ java.time.Instant.ofEpochMilli(123456789000L))
.build();
DeltaWriteTestUtils.writeCdcCommit(
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 1b7f73566c1..342c4b94761 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
@@ -32,6 +32,7 @@ import io.delta.kernel.types.MapType;
import io.delta.kernel.types.StringType;
import io.delta.kernel.types.StructField;
import io.delta.kernel.types.StructType;
+import io.delta.kernel.types.TimestampNTZType;
import io.delta.kernel.types.TimestampType;
import java.io.File;
import java.nio.charset.StandardCharsets;
@@ -49,6 +50,8 @@ import org.apache.beam.sdk.io.delta.DeltaIO.ReadRows;
import org.apache.beam.sdk.io.parquet.ParquetIO;
import org.apache.beam.sdk.managed.Managed;
import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.schemas.logicaltypes.SqlTypes;
+import org.apache.beam.sdk.schemas.logicaltypes.Timestamp;
import org.apache.beam.sdk.testing.PAssert;
import org.apache.beam.sdk.testing.TestPipeline;
import org.apache.beam.sdk.transforms.Count;
@@ -62,7 +65,6 @@ import org.apache.beam.sdk.values.PCollectionRowTuple;
import org.apache.beam.sdk.values.Row;
import org.apache.beam.sdk.values.ValueKind;
import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
-import org.joda.time.Instant;
import org.junit.Assert;
import org.junit.Rule;
import org.junit.Test;
@@ -519,6 +521,7 @@ public class DeltaIOTest {
new StructField("boolean", BooleanType.BOOLEAN, false),
new StructField("binary", BinaryType.BINARY, false),
new StructField("timestamp", TimestampType.TIMESTAMP, false),
+ new StructField("timestamp_ntz",
TimestampNTZType.TIMESTAMP_NTZ, false),
new StructField("date", DateType.DATE, false),
new StructField("array", new ArrayType(StringType.STRING,
true), false),
new StructField(
@@ -542,8 +545,9 @@ public class DeltaIOTest {
.addField("double", Schema.FieldType.DOUBLE)
.addField("boolean", Schema.FieldType.BOOLEAN)
.addField("binary", Schema.FieldType.BYTES)
- .addField("timestamp", Schema.FieldType.DATETIME)
- .addField("date", Schema.FieldType.DATETIME)
+ .addField("timestamp",
Schema.FieldType.logicalType(Timestamp.MICROS))
+ .addField("timestamp_ntz",
Schema.FieldType.logicalType(SqlTypes.DATETIME))
+ .addField("date", Schema.FieldType.logicalType(SqlTypes.DATE))
.addField("array",
Schema.FieldType.iterable(Schema.FieldType.STRING))
.addField("map", Schema.FieldType.map(Schema.FieldType.STRING,
Schema.FieldType.INT32))
.addField("struct", Schema.FieldType.row(nestedSchema))
@@ -909,7 +913,8 @@ public class DeltaIOTest {
.addField("name", Schema.FieldType.STRING)
.addField(DeltaIO.CHANGE_TYPE_COLUMN, Schema.FieldType.STRING)
.addField(DeltaIO.COMMIT_VERSION_COLUMN, Schema.FieldType.INT64)
- .addField(DeltaIO.COMMIT_TIMESTAMP_COLUMN,
Schema.FieldType.DATETIME)
+ .addField(
+ DeltaIO.COMMIT_TIMESTAMP_COLUMN,
Schema.FieldType.logicalType(Timestamp.MICROS))
.build();
StructType cdcWriteDeltaSchema =
new StructType()
@@ -920,15 +925,20 @@ public class DeltaIOTest {
Row cdcRow1 =
Row.withSchema(cdcWriteSchema)
- .addValues("row-1", "update_preimage", 1L, new
Instant(123456789000L))
+ .addValues(
+ "row-1", "update_preimage", 1L,
java.time.Instant.ofEpochMilli(123456789000L))
.build();
Row cdcRow2 =
Row.withSchema(cdcWriteSchema)
- .addValues("row-1-updated", "update_postimage", 1L, new
Instant(123456789000L))
+ .addValues(
+ "row-1-updated",
+ "update_postimage",
+ 1L,
+ java.time.Instant.ofEpochMilli(123456789000L))
.build();
Row cdcRow3 =
Row.withSchema(cdcWriteSchema)
- .addValues("row-2", "delete", 1L, new Instant(123456789000L))
+ .addValues("row-2", "delete", 1L,
java.time.Instant.ofEpochMilli(123456789000L))
.build();
DeltaWriteTestUtils.writeCdcCommit(
@@ -986,7 +996,8 @@ public class DeltaIOTest {
.addField("name", Schema.FieldType.STRING)
.addField(DeltaIO.CHANGE_TYPE_COLUMN, Schema.FieldType.STRING)
.addField(DeltaIO.COMMIT_VERSION_COLUMN, Schema.FieldType.INT64)
- .addField(DeltaIO.COMMIT_TIMESTAMP_COLUMN,
Schema.FieldType.DATETIME)
+ .addField(
+ DeltaIO.COMMIT_TIMESTAMP_COLUMN,
Schema.FieldType.logicalType(Timestamp.MICROS))
.build();
StructType cdcWriteDeltaSchema =
new StructType()
@@ -997,7 +1008,7 @@ public class DeltaIOTest {
Row cdcRow =
Row.withSchema(cdcWriteSchema)
- .addValues("row-3", "insert", 1L, new Instant(123456789000L))
+ .addValues("row-3", "insert", 1L,
java.time.Instant.ofEpochMilli(123456789000L))
.build();
Row appendRow = Row.withSchema(tableSchema).addValues("row-3").build();
@@ -1063,7 +1074,8 @@ public class DeltaIOTest {
.addField("name", Schema.FieldType.STRING)
.addField(DeltaIO.CHANGE_TYPE_COLUMN, Schema.FieldType.STRING)
.addField(DeltaIO.COMMIT_VERSION_COLUMN, Schema.FieldType.INT64)
- .addField(DeltaIO.COMMIT_TIMESTAMP_COLUMN,
Schema.FieldType.DATETIME)
+ .addField(
+ DeltaIO.COMMIT_TIMESTAMP_COLUMN,
Schema.FieldType.logicalType(Timestamp.MICROS))
.build();
StructType cdcWriteDeltaSchema =
new StructType()
@@ -1074,15 +1086,20 @@ public class DeltaIOTest {
Row cdcRow1 =
Row.withSchema(cdcWriteSchema)
- .addValues("row-1", "update_preimage", 1L, new
Instant(123456789000L))
+ .addValues(
+ "row-1", "update_preimage", 1L,
java.time.Instant.ofEpochMilli(123456789000L))
.build();
Row cdcRow2 =
Row.withSchema(cdcWriteSchema)
- .addValues("row-1-updated", "update_postimage", 1L, new
Instant(123456789000L))
+ .addValues(
+ "row-1-updated",
+ "update_postimage",
+ 1L,
+ java.time.Instant.ofEpochMilli(123456789000L))
.build();
Row cdcRow3 =
Row.withSchema(cdcWriteSchema)
- .addValues("row-2", "delete", 1L, new Instant(123456789000L))
+ .addValues("row-2", "delete", 1L,
java.time.Instant.ofEpochMilli(123456789000L))
.build();
DeltaWriteTestUtils.writeCdcCommit(
@@ -1147,7 +1164,8 @@ public class DeltaIOTest {
.addField("name", Schema.FieldType.STRING)
.addField(DeltaIO.CHANGE_TYPE_COLUMN, Schema.FieldType.STRING)
.addField(DeltaIO.COMMIT_VERSION_COLUMN, Schema.FieldType.INT64)
- .addField(DeltaIO.COMMIT_TIMESTAMP_COLUMN,
Schema.FieldType.DATETIME)
+ .addField(
+ DeltaIO.COMMIT_TIMESTAMP_COLUMN,
Schema.FieldType.logicalType(Timestamp.MICROS))
.build();
StructType cdcWriteDeltaSchema =
new StructType()
@@ -1158,15 +1176,20 @@ public class DeltaIOTest {
Row cdcRow1 =
Row.withSchema(cdcWriteSchema)
- .addValues("row-1", "update_preimage", 1L, new
Instant(123456789000L))
+ .addValues(
+ "row-1", "update_preimage", 1L,
java.time.Instant.ofEpochMilli(123456789000L))
.build();
Row cdcRow2 =
Row.withSchema(cdcWriteSchema)
- .addValues("row-1-updated", "update_postimage", 1L, new
Instant(123456789000L))
+ .addValues(
+ "row-1-updated",
+ "update_postimage",
+ 1L,
+ java.time.Instant.ofEpochMilli(123456789000L))
.build();
Row cdcRow3 =
Row.withSchema(cdcWriteSchema)
- .addValues("row-2", "delete", 1L, new Instant(123456789000L))
+ .addValues("row-2", "delete", 1L,
java.time.Instant.ofEpochMilli(123456789000L))
.build();
DeltaWriteTestUtils.writeCdcCommit(
@@ -1450,7 +1473,8 @@ public class DeltaIOTest {
.addField("name", Schema.FieldType.STRING)
.addField(DeltaIO.CHANGE_TYPE_COLUMN, Schema.FieldType.STRING)
.addField(DeltaIO.COMMIT_VERSION_COLUMN, Schema.FieldType.INT64)
- .addField(DeltaIO.COMMIT_TIMESTAMP_COLUMN,
Schema.FieldType.DATETIME)
+ .addField(
+ DeltaIO.COMMIT_TIMESTAMP_COLUMN,
Schema.FieldType.logicalType(Timestamp.MICROS))
.build();
StructType cdcWriteDeltaSchema =
new StructType()
@@ -1461,15 +1485,20 @@ public class DeltaIOTest {
Row cdcRow1 =
Row.withSchema(cdcWriteSchema)
- .addValues("row-1", "update_preimage", 1L, new
Instant(123456789000L))
+ .addValues(
+ "row-1", "update_preimage", 1L,
java.time.Instant.ofEpochMilli(123456789000L))
.build();
Row cdcRow2 =
Row.withSchema(cdcWriteSchema)
- .addValues("row-1-updated", "update_postimage", 1L, new
Instant(123456789000L))
+ .addValues(
+ "row-1-updated",
+ "update_postimage",
+ 1L,
+ java.time.Instant.ofEpochMilli(123456789000L))
.build();
Row cdcRow3 =
Row.withSchema(cdcWriteSchema)
- .addValues("row-2", "delete", 1L, new Instant(123456789000L))
+ .addValues("row-2", "delete", 1L,
java.time.Instant.ofEpochMilli(123456789000L))
.build();
DeltaWriteTestUtils.writeCdcCommit(
@@ -1532,7 +1561,8 @@ public class DeltaIOTest {
.addField("name", Schema.FieldType.STRING)
.addField(DeltaIO.CHANGE_TYPE_COLUMN, Schema.FieldType.STRING)
.addField(DeltaIO.COMMIT_VERSION_COLUMN, Schema.FieldType.INT64)
- .addField(DeltaIO.COMMIT_TIMESTAMP_COLUMN,
Schema.FieldType.DATETIME)
+ .addField(
+ DeltaIO.COMMIT_TIMESTAMP_COLUMN,
Schema.FieldType.logicalType(Timestamp.MICROS))
.build();
StructType cdcWriteDeltaSchema =
new StructType()
@@ -1543,15 +1573,20 @@ public class DeltaIOTest {
Row cdcRow1 =
Row.withSchema(cdcWriteSchema)
- .addValues("row-1", "update_preimage", 1L, new
Instant(200000000000L))
+ .addValues(
+ "row-1", "update_preimage", 1L,
java.time.Instant.ofEpochMilli(200000000000L))
.build();
Row cdcRow2 =
Row.withSchema(cdcWriteSchema)
- .addValues("row-1-updated", "update_postimage", 1L, new
Instant(200000000000L))
+ .addValues(
+ "row-1-updated",
+ "update_postimage",
+ 1L,
+ java.time.Instant.ofEpochMilli(200000000000L))
.build();
Row cdcRow3 =
Row.withSchema(cdcWriteSchema)
- .addValues("row-2", "delete", 1L, new Instant(200000000000L))
+ .addValues("row-2", "delete", 1L,
java.time.Instant.ofEpochMilli(200000000000L))
.build();
DeltaWriteTestUtils.writeCdcCommit(
@@ -1645,7 +1680,8 @@ public class DeltaIOTest {
.addField("name", Schema.FieldType.STRING)
.addField(DeltaIO.CHANGE_TYPE_COLUMN, Schema.FieldType.STRING)
.addField(DeltaIO.COMMIT_VERSION_COLUMN, Schema.FieldType.INT64)
- .addField(DeltaIO.COMMIT_TIMESTAMP_COLUMN,
Schema.FieldType.DATETIME)
+ .addField(
+ DeltaIO.COMMIT_TIMESTAMP_COLUMN,
Schema.FieldType.logicalType(Timestamp.MICROS))
.build();
StructType cdcWriteDeltaSchema =
new StructType()
@@ -1656,15 +1692,20 @@ public class DeltaIOTest {
Row cdcRow1 =
Row.withSchema(cdcWriteSchema)
- .addValues("row-1", "update_preimage", 1L, new
Instant(200000000000L))
+ .addValues(
+ "row-1", "update_preimage", 1L,
java.time.Instant.ofEpochMilli(200000000000L))
.build();
Row cdcRow2 =
Row.withSchema(cdcWriteSchema)
- .addValues("row-1-updated", "update_postimage", 1L, new
Instant(200000000000L))
+ .addValues(
+ "row-1-updated",
+ "update_postimage",
+ 1L,
+ java.time.Instant.ofEpochMilli(200000000000L))
.build();
Row cdcRow3 =
Row.withSchema(cdcWriteSchema)
- .addValues("row-2", "delete", 1L, new Instant(200000000000L))
+ .addValues("row-2", "delete", 1L,
java.time.Instant.ofEpochMilli(200000000000L))
.build();
DeltaWriteTestUtils.writeCdcCommit(
@@ -1691,15 +1732,20 @@ public class DeltaIOTest {
// 4. Write parquet files for Version 3 (commit with updates and deletes)
Row cdcRow4 =
Row.withSchema(cdcWriteSchema)
- .addValues("row-3", "update_preimage", 3L, new
Instant(400000000000L))
+ .addValues(
+ "row-3", "update_preimage", 3L,
java.time.Instant.ofEpochMilli(400000000000L))
.build();
Row cdcRow5 =
Row.withSchema(cdcWriteSchema)
- .addValues("row-3-updated", "update_postimage", 3L, new
Instant(400000000000L))
+ .addValues(
+ "row-3-updated",
+ "update_postimage",
+ 3L,
+ java.time.Instant.ofEpochMilli(400000000000L))
.build();
Row cdcRow6 =
Row.withSchema(cdcWriteSchema)
- .addValues("row-1-updated", "delete", 3L, new
Instant(400000000000L))
+ .addValues("row-1-updated", "delete", 3L,
java.time.Instant.ofEpochMilli(400000000000L))
.build();
DeltaWriteTestUtils.writeCdcCommit(
@@ -1777,13 +1823,21 @@ public class DeltaIOTest {
private static final class FormatRowWithMetadata extends DoFn<Row, String> {
@ProcessElement
public void process(@Element Row row, OutputReceiver<String> out) {
- out.output(
- String.format(
- "%s:%s:v%d:t%d",
- row.getString("name"),
- row.getString(DeltaIO.CHANGE_TYPE_COLUMN),
- row.getInt64(DeltaIO.COMMIT_VERSION_COLUMN),
- row.getDateTime(DeltaIO.COMMIT_TIMESTAMP_COLUMN).getMillis()));
+ Object tsVal = row.getValue(DeltaIO.COMMIT_TIMESTAMP_COLUMN);
+ if (tsVal instanceof java.time.Instant) {
+ long millis = ((java.time.Instant) tsVal).toEpochMilli();
+ out.output(
+ String.format(
+ "%s:%s:v%d:t%d",
+ row.getString("name"),
+ row.getString(DeltaIO.CHANGE_TYPE_COLUMN),
+ row.getInt64(DeltaIO.COMMIT_VERSION_COLUMN),
+ millis));
+ } else {
+ throw new RuntimeException(
+ "Expected 'COMMIT_TIMESTAMP_COLUMN' to be of type
'java.time.Instant' but received: "
+ + tsVal);
+ }
}
}
@@ -1811,10 +1865,15 @@ public class DeltaIOTest {
private static final class FormatRowTimestampMetadata extends DoFn<Row,
String> {
@ProcessElement
public void process(@Element Row row, OutputReceiver<String> out) {
- out.output(
- row.getString("name")
- + ":"
- + row.getDateTime(DeltaIO.COMMIT_TIMESTAMP_COLUMN).getMillis());
+ Object tsVal = row.getValue(DeltaIO.COMMIT_TIMESTAMP_COLUMN);
+ if (tsVal instanceof java.time.Instant) {
+ long millis = ((java.time.Instant) tsVal).toEpochMilli();
+ out.output(row.getString("name") + ":" + millis);
+ } else {
+ throw new RuntimeException(
+ "Expected 'COMMIT_TIMESTAMP_COLUMN' to be of type
'java.time.Instant' but received: "
+ + tsVal);
+ }
}
}
}
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 55646de749f..90f987859d7 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
@@ -31,11 +31,13 @@ import io.delta.kernel.engine.Engine;
import io.delta.kernel.internal.data.GenericRow;
import io.delta.kernel.types.BooleanType;
import io.delta.kernel.types.DataType;
+import io.delta.kernel.types.DateType;
import io.delta.kernel.types.LongType;
import io.delta.kernel.types.MapType;
import io.delta.kernel.types.StringType;
import io.delta.kernel.types.StructField;
import io.delta.kernel.types.StructType;
+import io.delta.kernel.types.TimestampNTZType;
import io.delta.kernel.types.TimestampType;
import io.delta.kernel.utils.CloseableIterable;
import io.delta.kernel.utils.CloseableIterator;
@@ -50,10 +52,9 @@ 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;
/** Utility class for writing test commits (appends and CDC actions) to Delta
tables in tests. */
-final class DeltaWriteTestUtils {
+public final class DeltaWriteTestUtils {
private DeltaWriteTestUtils() {}
@@ -134,18 +135,63 @@ final class DeltaWriteTestUtils {
@Override
public int getInt(int rowId) {
+ if (dataType instanceof DateType) {
+ Object val = rows.get(rowId).getValue(fieldIndex);
+ if (val instanceof java.time.LocalDate) {
+ return (int) ((java.time.LocalDate) val).toEpochDay();
+ }
+ }
return rows.get(rowId).getInt32(fieldIndex);
}
@Override
public long getLong(int rowId) {
if (dataType instanceof TimestampType) {
- Instant instant =
rows.get(rowId).getDateTime(fieldIndex).toInstant();
- return instant.getMillis() * 1000L;
+ Object val = rows.get(rowId).getValue(fieldIndex);
+ if (val instanceof java.time.Instant) {
+ java.time.Instant inst = (java.time.Instant) val;
+ return inst.getEpochSecond() * 1_000_000L + inst.getNano() / 1000L;
+ } else {
+ throw new RuntimeException(
+ "Unexpected value for field " + rowId + " of type
'TimestampType': " + val);
+ }
+ }
+ if (dataType instanceof TimestampNTZType) {
+ Object val = rows.get(rowId).getValue(fieldIndex);
+ if (val instanceof java.time.LocalDateTime) {
+ java.time.Instant inst =
+ ((java.time.LocalDateTime)
val).toInstant(java.time.ZoneOffset.UTC);
+ return inst.getEpochSecond() * 1_000_000L + inst.getNano() / 1000L;
+ }
}
return rows.get(rowId).getInt64(fieldIndex);
}
+ @Override
+ public double getDouble(int rowId) {
+ return rows.get(rowId).getDouble(fieldIndex);
+ }
+
+ @Override
+ public float getFloat(int rowId) {
+ return rows.get(rowId).getFloat(fieldIndex);
+ }
+
+ @Override
+ public short getShort(int rowId) {
+ return rows.get(rowId).getInt16(fieldIndex);
+ }
+
+ @Override
+ public byte getByte(int rowId) {
+ return rows.get(rowId).getByte(fieldIndex);
+ }
+
+ @Override
+ public byte[] getBinary(int rowId) {
+ return rows.get(rowId).getBytes(fieldIndex);
+ }
+
@Override
public String getString(int rowId) {
return rows.get(rowId).getString(fieldIndex);
@@ -165,7 +211,7 @@ final class DeltaWriteTestUtils {
* @return the list of names of the written Parquet data files
* @throws Exception if any error occurs during write or commit
*/
- static List<String> writeAppendCommit(
+ public static List<String> writeAppendCommit(
Engine engine,
String tablePath,
long expectedVersion,
@@ -262,7 +308,7 @@ final class DeltaWriteTestUtils {
* @param cdcWriteSchema the schema used for writing the CDC files
* @throws Exception if any error occurs during write or commit
*/
- static void writeCdcCommit(
+ public static void writeCdcCommit(
Engine engine,
String tablePath,
long expectedVersion,
diff --git a/settings.gradle.kts b/settings.gradle.kts
index 60d9ed6060d..33f7447f320 100644
--- a/settings.gradle.kts
+++ b/settings.gradle.kts
@@ -397,5 +397,8 @@ include("it:clickhouse")
findProject(":it:clickhouse")?.name = "clickhouse"
include("sdks:java:extensions:sql:iceberg")
findProject(":sdks:java:extensions:sql:iceberg")?.name = "iceberg"
+include("sdks:java:extensions:sql:delta")
+findProject(":sdks:java:extensions:sql:delta")?.name = "delta"
include("examples:java:iceberg")
findProject(":examples:java:iceberg")?.name = "iceberg"
+
