WIP: Support Snapshot Copy-On-Write Hudi Table to Iceberg Table

.github/workflows/hudi-conversion-ci.yaml

@@ -0,0 +1,86 @@
+#
+# 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.
+#
+
+name: "Hudi Conversion CI"
+on:
+  push:
+    branches:
+      - 'master'
+      - '0.**'
+    tags:
+      - 'apache-iceberg-**'
+  pull_request:
+    paths-ignore:
+      - '.github/ISSUE_TEMPLATE/iceberg_bug_report.yml'
+      - '.github/workflows/python-ci.yml'
+      - '.github/workflows/flink-ci.yml'
+      - '.github/workflows/hive-ci.yml'
+      - '.gitignore'
+      - '.asf.yml'
+      - 'dev/**'
+      - 'mr/**'
+      - 'hive3/**'
+      - 'hive3-orc-bundle/**'
+      - 'hive-runtime/**'
+      - 'flink/**'
+      - 'pig/**'
+      - 'python/**'
+      - 'python_legacy/**'
+      - 'docs/**'
+      - 'open-api/**'
+      - 'format/**'
+      - '.gitattributes'
+      - 'README.md'
+      - 'CONTRIBUTING.md'
+      - 'LICENSE'
+      - 'NOTICE'
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.ref }}
+  cancel-in-progress: ${{ github.event_name == 'pull_request' }}
+
+jobs:
+  hudi-conversion-scala-2-12-tests:
+    runs-on: ubuntu-20.04
+    strategy:
+      matrix:
+        jvm: [8, 11]
+    env:
+      SPARK_LOCAL_IP: localhost
+    steps:
+      - uses: actions/checkout@v3
+      - uses: actions/setup-java@v3
+        with:
+          distribution: zulu
+          java-version: ${{ matrix.jvm }}
+      - uses: actions/cache@v3
+        with:
+          path: |
+            ~/.gradle/caches
+            ~/.gradle/wrapper
+          key: ${{ runner.os }}-gradle-${{ hashFiles('**/*.gradle*', '**/gradle-wrapper.properties') }}
+          restore-keys: ${{ runner.os }}-gradle-
+      - run: echo -e "$(ip addr show eth0 | grep "inet\b" | awk '{print $2}' | cut -d/ -f1)\t$(hostname -f) $(hostname -s)" | sudo tee -a /etc/hosts
+      - run: ./gradlew -DsparkVersions=3.3 -DscalaVersion=2.12 -DhiveVersions= -DflinkVersions= :iceberg-hudi:check -Pquick=true -x javadoc
+      - uses: actions/upload-artifact@v3
+        if: failure()
+        with:
+          name: test logs
+          path: |
+            **/build/testlogs

build.gradle

@@ -48,6 +48,10 @@ plugins {
   id 'nebula.dependency-recommender' version '11.0.0'
 }
 
+String scalaVersion = System.getProperty("scalaVersion") != null ? System.getProperty("scalaVersion") : System.getProperty("defaultScalaVersion")
+String sparkVersionsString = System.getProperty("sparkVersions") != null ? System.getProperty("sparkVersions") : System.getProperty("defaultSparkVersions")
+List<String> sparkVersions = sparkVersionsString != null && !sparkVersionsString.isEmpty() ? sparkVersionsString.split(",") : []
+
 try {
   // apply these plugins in a try-catch block so that we can handle cases without .git directory
   apply plugin: 'com.palantir.git-version'
@@ -438,6 +442,81 @@ project(':iceberg-aws') {
   }
 }
 
+project(':iceberg-hudi') {
+
+  configurations {
+    integrationImplementation.extendsFrom testImplementation
+    integrationRuntime.extendsFrom testRuntimeOnly
+  }
+
+  dependencies {
+    implementation project(path: ':iceberg-bundled-guava', configuration: 'shadow')
+    api project(':iceberg-api')
+    implementation project(':iceberg-common')
+    implementation project(':iceberg-core')
+    implementation project(':iceberg-parquet')
+    implementation project(':iceberg-orc')
+    implementation "com.fasterxml.jackson.core:jackson-databind"
+
+    // Hudi uses java8, may need to assess if we can use hudi in java11.
+    compileOnly("org.apache.hudi:hudi-common")
+    // Added to resolve dependency conflicts with hudi-spark-bundle
+    compileOnly("org.apache.hudi:hudi-client-common")
+    implementation("org.apache.avro:avro") {
+      exclude group: 'org.tukaani' // xz compression is not supported
+    }
+
+    compileOnly("org.apache.hadoop:hadoop-common") {
+      exclude group: 'org.apache.avro', module: 'avro'
+      exclude group: 'org.slf4j', module: 'slf4j-log4j12'
+      exclude group: 'javax.servlet', module: 'servlet-api'
+      exclude group: 'com.google.code.gson', module: 'gson'
+    }
+    if (sparkVersions.contains("3.3") && scalaVersion == "2.12") {
+      integrationImplementation project(':iceberg-data')
+      integrationImplementation("org.apache.hudi:hudi-spark3.3-bundle_2.12") {
+        exclude group: 'org.apache.hudi', module: 'hudi-common'
+        exclude group: 'org.apache.hudi', module: 'hudi-client-common'
+      }
+      integrationImplementation project(path: ":iceberg-spark:iceberg-spark-3.3_${scalaVersion}")
+      integrationImplementation("org.apache.hadoop:hadoop-minicluster") {
+        exclude group: 'org.apache.avro', module: 'avro'
+        // to make sure netty libs only come from project(':iceberg-arrow')
+        exclude group: 'io.netty', module: 'netty-buffer'
+        exclude group: 'io.netty', module: 'netty-common'
+      }
+      integrationImplementation project(path: ':iceberg-hive-metastore')
+      integrationImplementation project(path: ':iceberg-hive-metastore', configuration: 'testArtifacts')
+      integrationImplementation("org.apache.spark:spark-hive_${scalaVersion}:3.3.1") {
+        exclude group: 'org.apache.avro', module: 'avro'
+        exclude group: 'org.apache.arrow'
+        exclude group: 'org.apache.parquet'
+        // to make sure netty libs only come from project(':iceberg-arrow')
+        exclude group: 'io.netty', module: 'netty-buffer'
+        exclude group: 'io.netty', module: 'netty-common'
+        exclude group: 'org.roaringbitmap'
+      }
+    }
+  }
+
+  if (sparkVersions.contains("3.3") && scalaVersion == "2.12") {
+    sourceSets {
+      integration {
+        java.srcDir "$projectDir/src/integration/java"
+        resources.srcDir "$projectDir/src/integration/resources"
+        compileClasspath += main.output + test.output
+        runtimeClasspath += main.output + test.output
+      }
+    }
+
+    task integrationTest(type: Test) {
+      testClassesDirs = sourceSets.integration.output.classesDirs
+      classpath = sourceSets.integration.runtimeClasspath
+    }
+    check.dependsOn integrationTest
+  }
+}
+
 project(':iceberg-gcp') {
   dependencies {
     implementation project(path: ':iceberg-bundled-guava', configuration: 'shadow')

hudi/src/integration/java/org/apache/iceberg/hudi/HudiToIcebergMigrationSparkIntegration.java

@@ -0,0 +1,51 @@
+/*
+ * 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.iceberg.hudi;
+
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.spark.Spark3Util;
+import org.apache.spark.sql.SparkSession;
+import org.apache.spark.sql.connector.catalog.CatalogPlugin;
+
+public class HudiToIcebergMigrationSparkIntegration {
+  private HudiToIcebergMigrationSparkIntegration() {}
+
+  static SnapshotHudiTable snapshotHudiTable(
+      SparkSession spark, String hudiTablePath, String newTableIdentifier) {
+    Preconditions.checkArgument(
+        spark != null, "The SparkSession cannot be null, please provide a valid SparkSession");
+    Preconditions.checkArgument(
+        newTableIdentifier != null,
+        "The table identifier cannot be null, please provide a valid table identifier for the new iceberg table");
+    Preconditions.checkArgument(
+        hudiTablePath != null,
+        "The hudi table location cannot be null, please provide a valid location of the delta lake table to be snapshot");
+    String ctx = "hudi snapshot target";
+    CatalogPlugin defaultCatalog = spark.sessionState().catalogManager().currentCatalog();
+    Spark3Util.CatalogAndIdentifier catalogAndIdentifier =
+        Spark3Util.catalogAndIdentifier(ctx, spark, newTableIdentifier, defaultCatalog);
+    return HudiToIcebergMigrationActionsProvider.defaultProvider()
+        .snapshotHudiTable(hudiTablePath)
+        .as(TableIdentifier.parse(catalogAndIdentifier.identifier().toString()))
+        .hoodieConfiguration(spark.sessionState().newHadoopConf())
+        .icebergCatalog(
+            Spark3Util.loadIcebergCatalog(spark, catalogAndIdentifier.catalog().name()));
+  }
+}

hudi/src/integration/java/org/apache/iceberg/hudi/SparkHudiMigrationTestBase.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.iceberg.hudi;
+
+import java.util.Map;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.iceberg.hive.TestHiveMetastore;
+import org.apache.spark.sql.SparkSession;
+import org.apache.spark.sql.internal.SQLConf;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+
+@SuppressWarnings("VisibilityModifier")
+public abstract class SparkHudiMigrationTestBase {
+  protected static TestHiveMetastore metastore = null;
+  protected static HiveConf hiveConf = null;
+  protected static SparkSession spark = null;
+
+  @BeforeClass
+  public static void startMetastoreAndSpark() {
+    SparkHudiMigrationTestBase.metastore = new TestHiveMetastore();
+    metastore.start();
+    SparkHudiMigrationTestBase.hiveConf = metastore.hiveConf();
+
+    SparkHudiMigrationTestBase.spark =
+        SparkSession.builder()
+            .master("local[2]")
+            .config(SQLConf.PARTITION_OVERWRITE_MODE().key(), "dynamic")
+            .config(
+                "spark.hadoop." + HiveConf.ConfVars.METASTOREURIS.varname,
+                hiveConf.get(HiveConf.ConfVars.METASTOREURIS.varname))
+            .config("spark.hadoop.parquet.avro.write-old-list-structure", "false")
+            .config("spark.sql.legacy.respectNullabilityInTextDatasetConversion", "true")
+            .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
+            .config("spark.sql.extensions", "org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
+            .enableHiveSupport()
+            .getOrCreate();
+  }
+
+  @AfterClass
+  public static void stopMetastoreAndSpark() throws Exception {
+    if (metastore != null) {
+      metastore.stop();
+      SparkHudiMigrationTestBase.metastore = null;
+    }
+    if (spark != null) {
+      spark.stop();
+      SparkHudiMigrationTestBase.spark = null;
+    }
+  }
+
+  public SparkHudiMigrationTestBase(
+      String catalogName, String implementation, Map<String, String> config) {
+
+    spark.conf().set("spark.sql.catalog." + catalogName, implementation);
+    config.forEach(
+        (key, value) -> spark.conf().set("spark.sql.catalog." + catalogName + "." + key, value));
+  }
+}