Adds Delta Lake read support to Beam SQL

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

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")

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")
+}

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<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 ProjectSupport supportsProjects() {
+    // TODO: Support project pushdown / column pruning when supported by DeltaIO /
+    // Managed Delta
+    // Lake source.
+    return ProjectSupport.NONE;
+  }
+
+  @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();
+  }
+}

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;
+  }
+}

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;