WIP: Adding support for Delta to Iceberg migration

api/src/main/java/org/apache/iceberg/actions/ActionsProvider.java

@@ -35,6 +35,12 @@ default MigrateTable migrateTable(String tableIdent) {
         this.getClass().getName() + " does not implement migrateTable");
   }
 
+  /** Instantiates an action to delete orphan files. */
+  default MigrateDeltaLakeTable migrateDeltaLakeTable(String tableIdent, String deltaS3Location) {
+    throw new UnsupportedOperationException(
+        this.getClass().getName() + " does not implement migrateDeltaLakeTable");
+  }
+
   /** Instantiates an action to delete orphan files. */
   default DeleteOrphanFiles deleteOrphanFiles(Table table) {
     throw new UnsupportedOperationException(

api/src/main/java/org/apache/iceberg/actions/MigrateDeltaLakeTable.java

@@ -0,0 +1,34 @@
+/*
+ * 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.actions;
+
+import java.util.Map;
+
+/** Migrates a Delta Lake table to Iceberg in place. */
+public interface MigrateDeltaLakeTable
+    extends Action<MigrateDeltaLakeTable, MigrateDeltaLakeTable.Result> {
+
+  MigrateDeltaLakeTable tableProperties(Map<String, String> properties);
+
+  interface Result {
+
+    /** Returns the number of imported data files. */
+    long importedDataFilesCount();
+  }
+}

build.gradle

@@ -98,6 +98,11 @@ allprojects {
   repositories {
     mavenCentral()
     mavenLocal()
+    // TODO: remove once Delta Lake 2.1.0 is officially released
+    maven {
+      name = 'staging-repo'
+      url = 'https://oss.sonatype.org/content/repositories/iodelta-1087/'
+    }
   }
 }
 

core/src/main/java/org/apache/iceberg/actions/BaseMigrateDeltaLakeTableActionResult.java

@@ -0,0 +1,33 @@
+/*
+ * 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.actions;
+
+public class BaseMigrateDeltaLakeTableActionResult implements MigrateDeltaLakeTable.Result {
+
+  private final long importedDataFilesCount;
+
+  public BaseMigrateDeltaLakeTableActionResult(long importedDataFilesCount) {
+    this.importedDataFilesCount = importedDataFilesCount;
+  }
+
+  @Override
+  public long importedDataFilesCount() {
+    return importedDataFilesCount;
+  }
+}

data/src/main/java/org/apache/iceberg/data/TableMigrationUtil.java

@@ -161,7 +161,7 @@ private static Metrics getAvroMetrics(Path path, Configuration conf) {
     }
   }
 
-  private static Metrics getParquetMetrics(
+  public static Metrics getParquetMetrics(
       Path path, Configuration conf, MetricsConfig metricsSpec, NameMapping mapping) {
     try {
       InputFile file = HadoopInputFile.fromPath(path, conf);

spark/v3.3/build.gradle

@@ -63,6 +63,8 @@ project(":iceberg-spark:iceberg-spark-${sparkMajorVersion}_${scalaVersion}") {
     implementation("org.apache.parquet:parquet-column")
     implementation("org.apache.parquet:parquet-hadoop")
 
+    compileOnly ("io.delta:delta-standalone_${scalaVersion}")
+
     implementation("org.apache.orc:orc-core::nohive") {
       exclude group: 'org.apache.hadoop'
       exclude group: 'commons-lang'
@@ -77,6 +79,9 @@ project(":iceberg-spark:iceberg-spark-${sparkMajorVersion}_${scalaVersion}") {
       exclude group: 'com.google.code.findbugs', module: 'jsr305'
     }
 
+    // Needed to write Delta Lake tables for testing
+    testImplementation "io.delta:delta-core_${scalaVersion}"
+
     testImplementation("org.apache.hadoop:hadoop-minicluster") {
       exclude group: 'org.apache.avro', module: 'avro'
       // to make sure io.netty.buffer only comes from project(':iceberg-arrow')

spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/SparkSchemaUtil.java

@@ -338,7 +338,7 @@ private static PartitionSpec identitySpec(Schema schema, Collection<Column> colu
     return identitySpec(schema, names);
   }
 
-  private static PartitionSpec identitySpec(Schema schema, List<String> partitionNames) {
+  public static PartitionSpec identitySpec(Schema schema, List<String> partitionNames) {
     if (partitionNames == null || partitionNames.isEmpty()) {
       return null;
     }

spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/SparkTypeToType.java

@@ -42,15 +42,15 @@
 import org.apache.spark.sql.types.TimestampType;
 import org.apache.spark.sql.types.VarcharType;
 
-class SparkTypeToType extends SparkTypeVisitor<Type> {
+public class SparkTypeToType extends SparkTypeVisitor<Type> {
   private final StructType root;
   private int nextId = 0;
 
   SparkTypeToType() {
     this.root = null;
   }
 
-  SparkTypeToType(StructType root) {
+  public SparkTypeToType(StructType root) {
     this.root = root;
     // the root struct's fields use the first ids
     this.nextId = root.fields().length;

spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/SparkTypeVisitor.java

@@ -27,8 +27,8 @@
 import org.apache.spark.sql.types.StructType;
 import org.apache.spark.sql.types.UserDefinedType;
 
-class SparkTypeVisitor<T> {
-  static <T> T visit(DataType type, SparkTypeVisitor<T> visitor) {
+public class SparkTypeVisitor<T> {
+  public static <T> T visit(DataType type, SparkTypeVisitor<T> visitor) {
     if (type instanceof StructType) {
       StructField[] fields = ((StructType) type).fields();
       List<T> fieldResults = Lists.newArrayListWithExpectedSize(fields.length);