Delta Lake Migration: Replace Delta Standalone with Delta Kernel

build.gradle

@@ -638,7 +638,8 @@ project(':iceberg-delta-lake') {
     annotationProcessor libs.immutables.value
     compileOnly libs.immutables.value
 
-    compileOnly "io.delta:delta-standalone_${scalaVersion}:${libs.versions.delta.standalone.get()}"
+    compileOnly "io.delta:delta-kernel-api:${libs.versions.delta.kernel.get()}"
+    compileOnly "io.delta:delta-kernel-defaults:${libs.versions.delta.kerneldefaults.get()}"
 
     compileOnly(libs.hadoop3.common) {
       exclude group: 'org.apache.avro', module: 'avro'

delta-lake/src/main/java/org/apache/iceberg/delta/DeltaActionUtils.java

@@ -0,0 +1,54 @@
+/*
+ * 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.delta;
+
+import io.delta.kernel.data.Row;
+import io.delta.kernel.internal.actions.AddFile;
+import io.delta.kernel.internal.actions.RemoveFile;
+
+public final class DeltaActionUtils {
+  private static final String ADD = "add";
+  private static final String REMOVE = "remove";
+
+  private DeltaActionUtils() {}
+
+  public static boolean isAdd(Row row) {
+    return !row.isNullAt(indexOf(row, ADD));
+  }
+
+  public static boolean isRemove(Row row) {
+    return !row.isNullAt(indexOf(row, REMOVE));
+  }
+
+  public static AddFile getAdd(Row row) {
+    return new AddFile(row.getStruct(indexOf(row, ADD)));
+  }
+
+  public static RemoveFile getRemove(Row row) {
+    return new RemoveFile(row.getStruct(indexOf(row, REMOVE)));
+  }
+
+  private static int indexOf(Row row, String name) {
+    int idx = row.getSchema().indexOf(name);
+    if (idx < 0) {
+      throw new IllegalArgumentException("Missing column in Delta action row: " + name);
+    }
+    return idx;
+  }
+}

delta-lake/src/main/java/org/apache/iceberg/delta/DeltaLakeDataTypeVisitor.java

@@ -18,19 +18,19 @@
  */
 package org.apache.iceberg.delta;
 
-import io.delta.standalone.types.ArrayType;
-import io.delta.standalone.types.DataType;
-import io.delta.standalone.types.MapType;
-import io.delta.standalone.types.StructField;
-import io.delta.standalone.types.StructType;
+import io.delta.kernel.types.ArrayType;
+import io.delta.kernel.types.DataType;
+import io.delta.kernel.types.MapType;
+import io.delta.kernel.types.StructField;
+import io.delta.kernel.types.StructType;
 import java.util.List;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 
 abstract class DeltaLakeDataTypeVisitor<T> {
   public static <T> T visit(DataType type, DeltaLakeDataTypeVisitor<T> visitor) {
     if (type instanceof StructType) {
-      StructField[] fields = ((StructType) type).getFields();
-      List<T> fieldResults = Lists.newArrayListWithExpectedSize(fields.length);
+      List<StructField> fields = ((StructType) type).fields();
+      List<T> fieldResults = Lists.newArrayListWithExpectedSize(fields.size());
 
       for (StructField field : fields) {
         fieldResults.add(visitor.field(field, visit(field.getDataType(), visitor)));

delta-lake/src/main/java/org/apache/iceberg/delta/DeltaLakeTypeToType.java

@@ -18,23 +18,24 @@
  */
 package org.apache.iceberg.delta;
 
-import io.delta.standalone.types.ArrayType;
-import io.delta.standalone.types.BinaryType;
-import io.delta.standalone.types.BooleanType;
-import io.delta.standalone.types.ByteType;
-import io.delta.standalone.types.DataType;
-import io.delta.standalone.types.DateType;
-import io.delta.standalone.types.DecimalType;
-import io.delta.standalone.types.DoubleType;
-import io.delta.standalone.types.FloatType;
-import io.delta.standalone.types.IntegerType;
-import io.delta.standalone.types.LongType;
-import io.delta.standalone.types.MapType;
-import io.delta.standalone.types.ShortType;
-import io.delta.standalone.types.StringType;
-import io.delta.standalone.types.StructField;
-import io.delta.standalone.types.StructType;
-import io.delta.standalone.types.TimestampType;
+import io.delta.kernel.types.ArrayType;
+import io.delta.kernel.types.BinaryType;
+import io.delta.kernel.types.BooleanType;
+import io.delta.kernel.types.ByteType;
+import io.delta.kernel.types.DataType;
+import io.delta.kernel.types.DateType;
+import io.delta.kernel.types.DecimalType;
+import io.delta.kernel.types.DoubleType;
+import io.delta.kernel.types.FloatType;
+import io.delta.kernel.types.IntegerType;
+import io.delta.kernel.types.LongType;
+import io.delta.kernel.types.MapType;
+import io.delta.kernel.types.ShortType;
+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.List;
 import org.apache.iceberg.exceptions.ValidationException;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
@@ -51,7 +52,7 @@ class DeltaLakeTypeToType extends DeltaLakeDataTypeVisitor<Type> {
 
   DeltaLakeTypeToType(StructType root) {
     this.root = root;
-    this.nextId = root.getFields().length;
+    this.nextId = root.fields().size();
   }
 
   private int getNextId() {
@@ -63,11 +64,11 @@ private int getNextId() {
   @Override
   @SuppressWarnings("ReferenceEquality")
   public Type struct(StructType struct, List<Type> types) {
-    StructField[] fields = struct.getFields();
-    List<Types.NestedField> newFields = Lists.newArrayListWithExpectedSize(fields.length);
+    List<StructField> fields = struct.fields();
+    List<Types.NestedField> newFields = Lists.newArrayListWithExpectedSize(fields.size());
     boolean isRoot = root == struct;
-    for (int i = 0; i < fields.length; i += 1) {
-      StructField field = fields[i];
+    for (int i = 0; i < fields.size(); i += 1) {
+      StructField field = fields.get(i);
       Type type = types.get(i);
 
       int id;
@@ -109,7 +110,7 @@ public Type array(ArrayType array, Type elementType) {
 
   @Override
   public Type map(MapType map, Type keyType, Type valueType) {
-    if (map.valueContainsNull()) {
+    if (map.isValueContainsNull()) {
       return Types.MapType.ofOptional(getNextId(), getNextId(), keyType, valueType);
     } else {
       return Types.MapType.ofRequired(getNextId(), getNextId(), keyType, valueType);
@@ -145,13 +146,16 @@ public Type atomic(DataType atomic) {
     } else if (atomic instanceof TimestampType) {
       return Types.TimestampType.withZone();
 
+    } else if (atomic instanceof TimestampNTZType) {
+      return Types.TimestampType.withoutZone();
+
     } else if (atomic instanceof DecimalType) {
       return Types.DecimalType.of(
           ((DecimalType) atomic).getPrecision(), ((DecimalType) atomic).getScale());
     } else if (atomic instanceof BinaryType) {
       return Types.BinaryType.get();
     }
 
-    throw new ValidationException("Not a supported type: %s", atomic.getCatalogString());
+    throw new ValidationException("Not a supported type: %s", atomic.toString());
   }
 }

delta-lake/src/test/java/org/apache/iceberg/delta/TestDeltaLakeTypeToType.java

@@ -19,19 +19,17 @@
 package org.apache.iceberg.delta;
 
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
-import io.delta.standalone.types.ArrayType;
-import io.delta.standalone.types.BinaryType;
-import io.delta.standalone.types.BooleanType;
-import io.delta.standalone.types.DoubleType;
-import io.delta.standalone.types.LongType;
-import io.delta.standalone.types.MapType;
-import io.delta.standalone.types.NullType;
-import io.delta.standalone.types.StringType;
-import io.delta.standalone.types.StructType;
+import io.delta.kernel.types.ArrayType;
+import io.delta.kernel.types.BinaryType;
+import io.delta.kernel.types.BooleanType;
+import io.delta.kernel.types.DoubleType;
+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 org.apache.iceberg.Schema;
-import org.apache.iceberg.exceptions.ValidationException;
 import org.apache.iceberg.types.Type;
 import org.apache.iceberg.types.Types;
 import org.junit.jupiter.api.BeforeEach;
@@ -44,30 +42,25 @@ public class TestDeltaLakeTypeToType {
   private static final String STRUCT_ARRAY_TYPE = "testStructArrayType";
   private static final String INNER_ATOMIC_SCHEMA = "testInnerAtomicSchema";
   private static final String STRING_LONG_MAP_TYPE = "testStringLongMap";
-  private static final String NULL_TYPE = "testNullType";
   private StructType deltaAtomicSchema;
   private StructType deltaNestedSchema;
-  private StructType deltaShallowNullTypeSchema;
-  private StructType deltaNullTypeSchema;
 
   @BeforeEach
   public void constructDeltaLakeSchema() {
     deltaAtomicSchema =
         new StructType()
-            .add(OPTIONAL_BOOLEAN_TYPE, new BooleanType())
-            .add(REQUIRED_BINARY_TYPE, new BinaryType(), false);
+            .add(new StructField(OPTIONAL_BOOLEAN_TYPE, BooleanType.BOOLEAN, true))
+            .add(new StructField(REQUIRED_BINARY_TYPE, BinaryType.BINARY, false));
     deltaNestedSchema =
         new StructType()
-            .add(INNER_ATOMIC_SCHEMA, deltaAtomicSchema)
-            .add(DOUBLE_ARRAY_TYPE, new ArrayType(new DoubleType(), true), false)
-            .add(STRUCT_ARRAY_TYPE, new ArrayType(deltaAtomicSchema, true), false)
-            .add(STRING_LONG_MAP_TYPE, new MapType(new StringType(), new LongType(), false), false);
-    deltaNullTypeSchema =
-        new StructType()
-            .add(INNER_ATOMIC_SCHEMA, deltaAtomicSchema)
-            .add(DOUBLE_ARRAY_TYPE, new ArrayType(new DoubleType(), true), false)
-            .add(STRING_LONG_MAP_TYPE, new MapType(new NullType(), new LongType(), false), false);
-    deltaShallowNullTypeSchema = new StructType().add(NULL_TYPE, new NullType(), false);
+            .add(new StructField(INNER_ATOMIC_SCHEMA, deltaAtomicSchema, true))
+            .add(new StructField(DOUBLE_ARRAY_TYPE, new ArrayType(DoubleType.DOUBLE, true), false))
+            .add(new StructField(STRUCT_ARRAY_TYPE, new ArrayType(deltaAtomicSchema, true), false))
+            .add(
+                new StructField(
+                    STRING_LONG_MAP_TYPE,
+                    new MapType(StringType.STRING, LongType.LONG, false),
+                    false));
   }
 
   @Test
@@ -162,21 +155,4 @@ public void testNestedTypeConversion() {
                 .isRequired())
         .isTrue();
   }
-
-  @Test
-  public void testNullTypeConversion() {
-    assertThatThrownBy(
-            () ->
-                DeltaLakeDataTypeVisitor.visit(
-                    deltaNullTypeSchema, new DeltaLakeTypeToType(deltaNullTypeSchema)))
-        .isInstanceOf(ValidationException.class)
-        .hasMessage("Not a supported type: %s", new NullType().getCatalogString());
-    assertThatThrownBy(
-            () ->
-                DeltaLakeDataTypeVisitor.visit(
-                    deltaShallowNullTypeSchema,
-                    new DeltaLakeTypeToType(deltaShallowNullTypeSchema)))
-        .isInstanceOf(ValidationException.class)
-        .hasMessage("Not a supported type: %s", new NullType().getCatalogString());
-  }
 }

gradle/libs.versions.toml

@@ -41,7 +41,8 @@ caffeine = "2.9.3"
 calcite = "1.41.0"
 comet = "0.12.0"
 datasketches = "6.2.0"
-delta-standalone = "3.3.2"
+delta-kernel = "4.0.0"
+delta-kerneldefaults = "4.0.0"
 delta-spark = "3.3.2"
 derby = "10.15.2.0"
 esotericsoftware-kryo = "4.0.3"
@@ -115,7 +116,8 @@ caffeine = { module = "com.github.ben-manes.caffeine:caffeine", version.ref = "c
 calcite-core = { module = "org.apache.calcite:calcite-core", version.ref = "calcite" }
 calcite-druid = { module = "org.apache.calcite:calcite-druid", version.ref = "calcite" }
 datasketches = { module = "org.apache.datasketches:datasketches-java", version.ref = "datasketches" }
-delta-standalone = { module = "io.delta:delta-standalone_2.12", version.ref = "delta-standalone" }
+delta-kernel-api = { module = "io.delta:delta-kernel-api", version.ref = "delta-kernel" }
+delta-kernel-defaults = { module = "io.delta:delta-kernel-defaults", version.ref = "delta-kernel" }
 errorprone-annotations = { module = "com.google.errorprone:error_prone_annotations", version.ref = "errorprone-annotations" }
 failsafe = { module = "dev.failsafe:failsafe", version.ref = "failsafe"}
 findbugs-jsr305 = { module = "com.google.code.findbugs:jsr305", version.ref = "findbugs-jsr305" }