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 ab4daeae2349..3a009261f4f9 100644 --- a/.github/trigger_files/beam_PostCommit_Java_Delta_IO_Dataflow.json +++ b/.github/trigger_files/beam_PostCommit_Java_Delta_IO_Dataflow.json @@ -1,4 +1,4 @@ { "comment": "Modify this file in a trivial way to cause this test suite to run.", - "modification": 3 + "modification": 2 } diff --git a/build.gradle.kts b/build.gradle.kts index 4ffa09a93719..8e9b81caf36a 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 000000000000..ea6daebbe9af --- /dev/null +++ b/sdks/java/extensions/sql/delta/build.gradle @@ -0,0 +1,56 @@ +/* + * 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." + +def parquet_version = "1.16.0" + +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 library.java.avro + testImplementation "org.apache.parquet:parquet-avro:$parquet_version" + testImplementation "org.apache.parquet:parquet-hadoop:$parquet_version" + testImplementation project(path: ":sdks:java:io:delta", configuration: "testRuntimeMigration") + testImplementation project(":sdks:java:extensions:avro") + 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 000000000000..6c087bc836b6 --- /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,208 @@ +/* + * 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.ProjectSupport; +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 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 parsedHadoopConfig = new HashMap<>(); + + ObjectNode properties = table.getProperties(); + for (Map.Entry 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 targetMap) { + if (val.isObject()) { + Map map = + TableUtils.getObjectMapper() + .convertValue(val, new TypeReference>() {}); + if (map != null) { + targetMap.putAll(map); + } + } else if (val.isTextual()) { + try { + Map map = + TableUtils.getObjectMapper() + .readValue(val.asText(), new TypeReference>() {}); + if (map != null) { + targetMap.putAll(map); + } + } catch (Exception e) { + throw new IllegalArgumentException("Failed to parse hadoop_config string as JSON", e); + } + } + } + + @Override + public PCollection buildIOReader(PBegin begin) { + return begin + .apply(Managed.read(Managed.DELTA_LAKE).withConfig(getBaseConfig())) + .getSinglePCollection(); + } + + @Override + public PCollection buildIOReader( + PBegin begin, BeamSqlTableFilter filters, List 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 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 ProjectSupport supportsProjects() { + // TODO: Support project pushdown / column pruning when supported by DeltaIO / + // Managed Delta + // Lake source. + return ProjectSupport.NONE; + } + + @Override + public BeamSqlTableFilter constructFilter(List filter) { + // TODO: Support predicate pushdown when supported by DeltaIO / Managed Delta + // Lake source. + return new DefaultTableFilter(filter); + } + + private Map getBaseConfig() { + ImmutableMap.Builder 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 000000000000..7e803825754d --- /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. + * + *

A sample of Delta Lake table registration is: + * + * 

{@code
+ * CREATE EXTERNAL TABLE orders(
+ *   id INTEGER,
+ *   name VARCHAR,
+ *   amount DOUBLE
+ * )
+ * TYPE 'delta'
+ * LOCATION '/path/to/delta/orders'
+ * TBLPROPERTIES '{"version": 1}'
+ * }
+ */ +@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 000000000000..3f2eca1d8568 --- /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 000000000000..c5df55c1784a --- /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,566 @@ +/* + * 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.IntegerType; +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; +import java.nio.file.Files; +import java.time.Instant; +import java.time.LocalDate; +import java.time.LocalDateTime; +import java.util.Arrays; +import java.util.List; +import org.apache.avro.generic.GenericRecord; +import org.apache.beam.sdk.Pipeline; +import org.apache.beam.sdk.extensions.avro.schemas.utils.AvroUtils; +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.apache.hadoop.fs.Path; +import org.apache.parquet.avro.AvroParquetWriter; +import org.apache.parquet.hadoop.ParquetWriter; +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 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 writeParquetData(File tableDir, String filename, Schema schema, List rows) + throws Exception { + org.apache.avro.Schema avroSchema = AvroUtils.toAvroSchema(schema); + List records = + rows.stream().map(r -> AvroUtils.toGenericRecord(r, avroSchema)).toList(); + + Path parquetPath = new Path(tableDir.getAbsolutePath(), filename); + try (ParquetWriter writer = + AvroParquetWriter.builder(parquetPath) + .withSchema(avroSchema) + .withConf(new Configuration()) + .build()) { + for (GenericRecord record : records) { + writer.write(record); + } + } + } + + private void createDeltaTable( + File tableDir, Schema schema, List rows, String deltaSchemaString) throws Exception { + writeParquetData(tableDir, "part-00000.parquet", schema, rows); + + File parquetFile = new File(tableDir, "part-00000.parquet"); + byte[] fileBytes = Files.readAllBytes(parquetFile.toPath()); + + File logDir = new File(tableDir, "_delta_log"); + logDir.mkdirs(); + File commitFile = new File(logDir, "00000000000000000000.json"); + + String commitContent = + "{\"protocol\":{\"minReaderVersion\":1,\"minWriterVersion\":2}}\n" + + "{\"metaData\":{\"id\":\"test-id\",\"format\":{\"provider\":\"parquet\",\"options\":{}},\"schemaString\":\"" + + deltaSchemaString + + "\",\"partitionColumns\":[],\"configuration\":{},\"createdAt\":100000000000}}\n" + + "{\"add\":{\"path\":\"part-00000.parquet\",\"partitionValues\":{},\"size\":" + + fileBytes.length + + ",\"modificationTime\":100000000000,\"dataChange\":true}}"; + + Files.write(commitFile.toPath(), commitContent.getBytes(StandardCharsets.UTF_8)); + commitFile.setLastModified(100000000000L); + } + + private void createTwoVersionDeltaTable(File tableDir) throws Exception { + 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(); + Row row3 = Row.withSchema(schema).addValues("v1-row3").build(); + + // Commit 0 data + writeParquetData(tableDir, "part-00000.parquet", schema, Arrays.asList(row1, row2)); + File parquet0 = new File(tableDir, "part-00000.parquet"); + byte[] bytes0 = Files.readAllBytes(parquet0.toPath()); + + File logDir = new File(tableDir, "_delta_log"); + logDir.mkdirs(); + File commit0 = new File(logDir, "00000000000000000000.json"); + + String commit0Content = + "{\"protocol\":{\"minReaderVersion\":1,\"minWriterVersion\":2}}\n" + + "{\"metaData\":{\"id\":\"two-ver-id\",\"format\":{\"provider\":\"parquet\",\"options\":{}},\"schemaString\":\"{\\\"type\\\":\\\"struct\\\",\\\"fields\\\":[{\\\"name\\\":\\\"name\\\",\\\"type\\\":\\\"string\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}}]}\",\"partitionColumns\":[],\"configuration\":{},\"createdAt\":100000000000}}\n" + + "{\"add\":{\"path\":\"part-00000.parquet\",\"partitionValues\":{},\"size\":" + + bytes0.length + + ",\"modificationTime\":100000000000,\"dataChange\":true}}"; + + Files.write(commit0.toPath(), commit0Content.getBytes(StandardCharsets.UTF_8)); + commit0.setLastModified(100000000000L); + + // Commit 1 data + writeParquetData(tableDir, "part-00001.parquet", schema, Arrays.asList(row3)); + File parquet1 = new File(tableDir, "part-00001.parquet"); + byte[] bytes1 = Files.readAllBytes(parquet1.toPath()); + + File commit1 = new File(logDir, "00000000000000000001.json"); + String commit1Content = + "{\"commitInfo\":{\"timestamp\":200000000000,\"operation\":\"WRITE\"}}\n" + + "{\"add\":{\"path\":\"part-00001.parquet\",\"partitionValues\":{},\"size\":" + + bytes1.length + + ",\"modificationTime\":200000000000,\"dataChange\":true}}"; + + Files.write(commit1.toPath(), commit1Content.getBytes(StandardCharsets.UTF_8)); + commit1.setLastModified(200000000000L); + } + + @Test + public void testSelectAll() throws Exception { + File tableDir = tempFolder.newFolder("delta-table-all"); + String deltaSchemaStr = + "{\\\"type\\\":\\\"struct\\\",\\\"fields\\\":[" + + "{\\\"name\\\":\\\"name\\\",\\\"type\\\":\\\"string\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}}," + + "{\\\"name\\\":\\\"age\\\",\\\"type\\\":\\\"integer\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}}," + + "{\\\"name\\\":\\\"score\\\",\\\"type\\\":\\\"double\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}}," + + "{\\\"name\\\":\\\"country\\\",\\\"type\\\":\\\"string\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}}" + + "]}"; + + createDeltaTable( + tableDir, PERSON_SCHEMA, Arrays.asList(PERSON_1, PERSON_2, PERSON_3), deltaSchemaStr); + + 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 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"); + String deltaSchemaStr = + "{\\\"type\\\":\\\"struct\\\",\\\"fields\\\":[" + + "{\\\"name\\\":\\\"name\\\",\\\"type\\\":\\\"string\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}}," + + "{\\\"name\\\":\\\"age\\\",\\\"type\\\":\\\"integer\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}}," + + "{\\\"name\\\":\\\"score\\\",\\\"type\\\":\\\"double\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}}," + + "{\\\"name\\\":\\\"country\\\",\\\"type\\\":\\\"string\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}}" + + "]}"; + + createDeltaTable( + tableDir, PERSON_SCHEMA, Arrays.asList(PERSON_1, PERSON_2, PERSON_3), deltaSchemaStr); + + 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 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"); + String deltaSchemaStr = + "{\\\"type\\\":\\\"struct\\\",\\\"fields\\\":[" + + "{\\\"name\\\":\\\"name\\\",\\\"type\\\":\\\"string\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}}," + + "{\\\"name\\\":\\\"age\\\",\\\"type\\\":\\\"integer\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}}," + + "{\\\"name\\\":\\\"score\\\",\\\"type\\\":\\\"double\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}}," + + "{\\\"name\\\":\\\"country\\\",\\\"type\\\":\\\"string\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}}" + + "]}"; + + createDeltaTable( + tableDir, PERSON_SCHEMA, Arrays.asList(PERSON_1, PERSON_2, PERSON_3), deltaSchemaStr); + + 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 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"); + String deltaSchemaStr = + "{\\\"type\\\":\\\"struct\\\",\\\"fields\\\":[" + + "{\\\"name\\\":\\\"name\\\",\\\"type\\\":\\\"string\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}}," + + "{\\\"name\\\":\\\"age\\\",\\\"type\\\":\\\"integer\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}}," + + "{\\\"name\\\":\\\"score\\\",\\\"type\\\":\\\"double\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}}," + + "{\\\"name\\\":\\\"country\\\",\\\"type\\\":\\\"string\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}}" + + "]}"; + + createDeltaTable( + tableDir, PERSON_SCHEMA, Arrays.asList(PERSON_1, PERSON_2, PERSON_3), deltaSchemaStr); + + 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 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 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 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"); + String deltaSchemaStr = + "{\\\"type\\\":\\\"struct\\\",\\\"fields\\\":[" + + "{\\\"name\\\":\\\"name\\\",\\\"type\\\":\\\"string\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}}," + + "{\\\"name\\\":\\\"age\\\",\\\"type\\\":\\\"integer\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}}," + + "{\\\"name\\\":\\\"score\\\",\\\"type\\\":\\\"double\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}}," + + "{\\\"name\\\":\\\"country\\\",\\\"type\\\":\\\"string\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}}" + + "]}"; + + createDeltaTable(tableDir, PERSON_SCHEMA, Arrays.asList(PERSON_1), deltaSchemaStr); + + 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 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 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 000000000000..b48defa0b23e --- /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 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 a0de04ed8544..2787c4bb606d 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 97d9c914a086..fe1f31e5be18 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 @@ private static Row projectRow(Row row, Schema targetSchema, DeltaCDCReadTask tas 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 4701ff59c4ad..c63fdcacc1c8 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.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 @@ static Schema.FieldType convertToBeamFieldType(DataType deltaType) { } 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 @@ static Schema buildPublicBeamSchema(Schema baseSchema, @Nullable List me } 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 fca19e30cb8f..0b3687393126 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.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 @@ static Row toBeamRow(io.delta.kernel.data.Row deltaRow, Schema beamSchema) { 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 @@ static Row toBeamRow(io.delta.kernel.data.Row deltaRow, Schema beamSchema) { 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 ab7fd8e24f4c..602c9b7f151c 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.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.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 void testReadChangesDeltaLake() throws Exception { .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 void testReadChangesDeltaLake() throws Exception { 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 1b7f73566c16..342c4b94761a 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.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.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.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 void testConvertToBeamSchema() { 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 void testConvertToBeamSchema() { .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 void testReadChanges() throws Exception { .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 void testReadChanges() throws Exception { 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 void testReadChangesAndNormalReadWithCDCAndAppend() throws Exception { .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 void testReadChangesAndNormalReadWithCDCAndAppend() throws Exception { 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 void testReadChangesWithSchemaTransformProvider() throws Exception { .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 void testReadChangesWithSchemaTransformProvider() throws Exception { 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 void testReadChangesWithMetadataColumns() throws Exception { .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 void testReadChangesWithMetadataColumns() throws Exception { 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 void testReadChangesWithManaged() throws Exception { .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 void testReadChangesWithManaged() throws Exception { 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 void testReadChangesRanges() throws Exception { .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 void testReadChangesRanges() throws Exception { 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 void testReadChangesPartialRange() throws Exception { .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 void testReadChangesPartialRange() throws Exception { 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 void testReadChangesPartialRange() throws Exception { // 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 void process( private static final class FormatRowWithMetadata extends DoFn { @ProcessElement public void process(@Element Row row, OutputReceiver 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 void process(@Element Row row, OutputReceiver out) { private static final class FormatRowTimestampMetadata extends DoFn { @ProcessElement public void process(@Element Row row, OutputReceiver 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 55646de749f1..36eb058eba1b 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.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 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,14 +135,34 @@ public boolean getBoolean(int rowId) { @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); } @@ -165,7 +186,7 @@ public String getString(int rowId) { * @return the list of names of the written Parquet data files * @throws Exception if any error occurs during write or commit */ - static List writeAppendCommit( + public static List writeAppendCommit( Engine engine, String tablePath, long expectedVersion, @@ -262,7 +283,7 @@ static List writeAppendCommit( * @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 60d9ed6060da..33f7447f3208 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" +