Implements the Delta Lake source with support for splitting (#38706)

Author: chamikaramj

sdks/java/io/delta/build.gradle

@@ -26,6 +26,10 @@ applyJavaNature(
 description = "Apache Beam :: SDKs :: Java :: IO :: Delta Lake"
 ext.summary = "Integration with Delta Lake."
 
+// We need to override the GCS bigdataos connector version to prevent conflicts.
+def bigdataoss_gcs_connector_version = "4.0.4"
+
+def parquet_version = "1.16.0"
 
 dependencies {
     implementation project(path: ":sdks:java:core", configuration: "shadow")
@@ -35,5 +39,45 @@ dependencies {
     permitUnusedDeclared library.java.delta_kernel_api
     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"
+
+    // We need to override the GCS connector version to prevent conflicts with
+    // latest Hadoop.
+    implementation "com.google.cloud.bigdataoss:gcs-connector:$bigdataoss_gcs_connector_version"
+    implementation "com.google.cloud.bigdataoss:util-hadoop:$bigdataoss_gcs_connector_version"
+    implementation "com.google.cloud.bigdataoss:gcsio:$bigdataoss_gcs_connector_version"
+    implementation "com.google.cloud.bigdataoss:util:$bigdataoss_gcs_connector_version"
+    permitUnusedDeclared "com.google.cloud.bigdataoss:gcs-connector:$bigdataoss_gcs_connector_version"
+    permitUnusedDeclared "com.google.cloud.bigdataoss:util-hadoop:$bigdataoss_gcs_connector_version"
+    permitUnusedDeclared "com.google.cloud.bigdataoss:gcsio:$bigdataoss_gcs_connector_version"
+    permitUnusedDeclared "com.google.cloud.bigdataoss:util:$bigdataoss_gcs_connector_version"
+
+    // For Avro conversions
+    testImplementation project(":sdks:java:extensions:avro")
+
+    testImplementation library.java.avro
     testImplementation library.java.junit
+    testImplementation library.java.hamcrest
+    testImplementation "org.apache.parquet:parquet-avro:$parquet_version"
+    testImplementation project(":sdks:java:io:parquet")
+    testImplementation project(":sdks:java:managed")
+    testRuntimeOnly "org.yaml:snakeyaml:2.0"
+    testImplementation project(path: ":runners:direct-java", configuration: "shadow")
+}
+
+configurations.all {
+    // Exclude conflicting logging frameworks
+    exclude group: "org.apache.logging.log4j", module: "log4j-slf4j2-impl"
+    exclude group: "org.apache.logging.log4j", module: "log4j-slf4j-impl"
+    exclude group: "org.slf4j", module: "slf4j-reload4j"
+
+    // Force overriding for all configurations
+    resolutionStrategy.force "com.google.cloud.bigdataoss:gcs-connector:$bigdataoss_gcs_connector_version"
+    resolutionStrategy.force "com.google.cloud.bigdataoss:util-hadoop:$bigdataoss_gcs_connector_version"
+    resolutionStrategy.force "com.google.cloud.bigdataoss:gcsio:$bigdataoss_gcs_connector_version"
+    resolutionStrategy.force "com.google.cloud.bigdataoss:util:$bigdataoss_gcs_connector_version"
 }

sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/BeamEngine.java

@@ -0,0 +1,55 @@
+/*
+ * 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.io.delta;
+
+import io.delta.kernel.engine.Engine;
+import io.delta.kernel.engine.ExpressionHandler;
+import io.delta.kernel.engine.FileSystemClient;
+import io.delta.kernel.engine.JsonHandler;
+import io.delta.kernel.engine.ParquetHandler;
+
+/** A Beam specific {@link Engine} wrapper that provides a custom {@link ParquetHandler}. */
+public class BeamEngine implements Engine {
+  private final Engine delegate;
+  private final ParquetHandler parquetHandler;
+
+  public BeamEngine(Engine delegate, ParquetHandler parquetHandler) {
+    this.delegate = delegate;
+    this.parquetHandler = parquetHandler;
+  }
+
+  @Override
+  public ExpressionHandler getExpressionHandler() {
+    return delegate.getExpressionHandler();
+  }
+
+  @Override
+  public JsonHandler getJsonHandler() {
+    return delegate.getJsonHandler();
+  }
+
+  @Override
+  public FileSystemClient getFileSystemClient() {
+    return delegate.getFileSystemClient();
+  }
+
+  @Override
+  public ParquetHandler getParquetHandler() {
+    return parquetHandler;
+  }
+}