[Kernel] Introduce IcebergCompatV3

kernel/kernel-api/src/main/java/io/delta/kernel/internal/TableConfig.java

@@ -242,6 +242,23 @@ public class TableConfig<T> {
           "needs to be a boolean.",
           true);
 
+  /**
+   * Table property that enables modifying the table in accordance with the Delta-Iceberg
+   * Compatibility V3 protocol.
+   *
+   * @see <a
+   *     href="https://github.com/delta-io/delta/blob/master/PROTOCOL.md#delta-iceberg-compatibility-v3">
+   *     Delta-Iceberg Compatibility V3 Protocol</a>
+   */
+  public static final TableConfig<Boolean> ICEBERG_COMPAT_V3_ENABLED =
+      new TableConfig<>(
+          "delta.enableIcebergCompatV3",
+          "false",
+          Boolean::valueOf,
+          value -> true,
+          "needs to be a boolean.",
+          true);
+
   /**
    * The number of columns to collect stats on for data skipping. A value of -1 means collecting
    * stats for all columns.
@@ -339,6 +356,7 @@ public static class UniversalFormats {
               addConfig(this, IN_COMMIT_TIMESTAMP_ENABLEMENT_TIMESTAMP);
               addConfig(this, COLUMN_MAPPING_MODE);
               addConfig(this, ICEBERG_COMPAT_V2_ENABLED);
+              addConfig(this, ICEBERG_COMPAT_V3_ENABLED);
               addConfig(this, ICEBERG_WRITER_COMPAT_V1_ENABLED);
               addConfig(this, COLUMN_MAPPING_MAX_COLUMN_ID);
               addConfig(this, DATA_SKIPPING_NUM_INDEXED_COLS);

kernel/kernel-api/src/main/java/io/delta/kernel/internal/actions/GenerateIcebergCompatActionUtils.java

@@ -75,6 +75,7 @@ public static Row generateIcebergCompatWriterV1AddAction(
     validateMaxRetriesSetToZero(transactionState);
 
     /* ----- Validate this is valid write given the table's protocol & configurations ----- */
+    // TODO: need to make it general
     checkState(
         TableConfig.ICEBERG_COMPAT_V2_ENABLED.fromMetadata(configuration),
         "icebergCompatV2 not enabled despite icebergWriterCompatV1 enabled");

kernel/kernel-api/src/main/java/io/delta/kernel/internal/data/TransactionStateRow.java

@@ -113,6 +113,19 @@ public static boolean isIcebergCompatV2Enabled(Row transactionState) {
             .getOrDefault(TableConfig.ICEBERG_COMPAT_V2_ENABLED.getKey(), "false"));
   }
 
+  /**
+   * Get the iceberg compatibility enabled or not from the transaction state {@link Row} returned by
+   * {@link Transaction#getTransactionState(Engine)}
+   *
+   * @param transactionState Transaction state state {@link Row}
+   * @return True if iceberg compatibility is enabled, false otherwise.
+   */
+  public static boolean isIcebergCompatV3Enabled(Row transactionState) {
+    return Boolean.parseBoolean(
+        getConfiguration(transactionState)
+            .getOrDefault(TableConfig.ICEBERG_COMPAT_V3_ENABLED.getKey(), "false"));
+  }
+
   /**
    * Get the list of partition column names from the transaction state {@link Row} returned by
    * {@link Transaction#getTransactionState(Engine)}

kernel/kernel-api/src/main/java/io/delta/kernel/internal/icebergcompat/IcebergCompatV3MetadataValidatorAndUpdater.java

@@ -0,0 +1,202 @@
+/*
+ * Copyright (2023) The Delta Lake Project Authors.
+ *
+ * Licensed 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 io.delta.kernel.internal.icebergcompat;
+
+import static io.delta.kernel.internal.tablefeatures.TableFeatures.*;
+import static io.delta.kernel.internal.util.Preconditions.checkArgument;
+import static java.util.Collections.singletonList;
+import static java.util.stream.Collectors.toList;
+
+import io.delta.kernel.internal.DeltaErrors;
+import io.delta.kernel.internal.TableConfig;
+import io.delta.kernel.internal.actions.Metadata;
+import io.delta.kernel.internal.actions.Protocol;
+import io.delta.kernel.internal.tablefeatures.TableFeature;
+import io.delta.kernel.internal.util.ColumnMapping.ColumnMappingMode;
+import io.delta.kernel.internal.util.SchemaUtils;
+import io.delta.kernel.internal.util.Tuple2;
+import io.delta.kernel.types.*;
+import io.delta.kernel.utils.DataFileStatus;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Stream;
+
+/** Utility methods for validation and compatibility checks for Iceberg V3. */
+public class IcebergCompatV3MetadataValidatorAndUpdater
+    extends IcebergCompatMetadataValidatorAndUpdater {
+  /**
+   * Validate and update the given Iceberg V3 metadata.
+   *
+   * @param newMetadata Metadata after the current updates
+   * @param newProtocol Protocol after the current updates
+   * @return The updated metadata if the metadata is valid and updated, otherwise empty.
+   * @throws UnsupportedOperationException if the metadata is not compatible with Iceberg V3
+   *     requirements
+   */
+  public static Optional<Metadata> validateAndUpdateIcebergCompatV3Metadata(
+      boolean isCreatingNewTable, Metadata newMetadata, Protocol newProtocol) {
+    return INSTANCE.validateAndUpdateMetadata(
+        new IcebergCompatInputContext(isCreatingNewTable, newMetadata, newProtocol));
+  }
+
+  /**
+   * Validate the given {@link DataFileStatus} that is being added as a {@code add} action to Delta
+   * Log. Currently, it checks that the statistics are not empty.
+   *
+   * @param dataFileStatus The {@link DataFileStatus} to validate.
+   */
+  public static void validateDataFileStatus(DataFileStatus dataFileStatus) {
+    if (!dataFileStatus.getStatistics().isPresent()) {
+      // presence of stats means always has a non-null `numRecords`
+      throw DeltaErrors.icebergCompatMissingNumRecordsStats(
+          INSTANCE.compatFeatureName(), dataFileStatus);
+    }
+  }
+
+  /// //////////////////////////////////////////////////////////////////////////////
+  /// Define the compatibility and update checks for icebergCompatV3             ///
+  /// //////////////////////////////////////////////////////////////////////////////
+
+  private static final IcebergCompatV3MetadataValidatorAndUpdater INSTANCE =
+      new IcebergCompatV3MetadataValidatorAndUpdater();
+
+  private static final IcebergCompatRequiredTablePropertyEnforcer ICEBERG_COMPAT_V3_CM_REQUIREMENT =
+      new IcebergCompatRequiredTablePropertyEnforcer<>(
+          TableConfig.COLUMN_MAPPING_MODE,
+          (value) -> ColumnMappingMode.NAME == value || ColumnMappingMode.ID == value,
+          ColumnMappingMode.NAME.value);
+
+  private static final IcebergCompatCheck ICEBERG_COMPAT_V3_CHECK_NO_LOWER_COMPAT_ENABLED =
+      (inputContext) -> {
+        List<String> incompatibleProperties =
+            Arrays.asList("delta.enableIcebergCompatV1", "delta.enableIcebergCompatV2");
+
+        incompatibleProperties.forEach(
+            prop -> {
+              if (Boolean.parseBoolean(
+                  inputContext.newMetadata.getConfiguration().getOrDefault(prop, "false"))) {
+                throw DeltaErrors.icebergCompatIncompatibleVersionEnabled(
+                    INSTANCE.compatFeatureName(), prop);
+              }
+            });
+      };
+
+  private static final IcebergCompatCheck ICEBERG_COMPAT_V3_CHECK_HAS_SUPPORTED_TYPES =
+      (inputContext) -> {
+        List<Tuple2<List<String>, StructField>> matches =
+            SchemaUtils.filterRecursively(
+                inputContext.newMetadata.getSchema(),
+                /* recurseIntoMapAndArrayTypes= */ true,
+                /* stopOnFirstMatch = */ false,
+                field -> {
+                  DataType dataType = field.getDataType();
+                  return !(dataType instanceof ByteType
+                      || dataType instanceof ShortType
+                      || dataType instanceof IntegerType
+                      || dataType instanceof LongType
+                      || dataType instanceof FloatType
+                      || dataType instanceof DoubleType
+                      || dataType instanceof DecimalType
+                      || dataType instanceof StringType
+                      || dataType instanceof BinaryType
+                      || dataType instanceof BooleanType
+                      || dataType instanceof DateType
+                      || dataType instanceof TimestampType
+                      || dataType instanceof TimestampNTZType
+                      || dataType instanceof ArrayType
+                      || dataType instanceof MapType
+                      || dataType instanceof StructType);
+                });
+
+        if (!matches.isEmpty()) {
+          throw DeltaErrors.icebergCompatUnsupportedTypeColumns(
+              INSTANCE.compatFeatureName(),
+              matches.stream().map(tuple -> tuple._2.getDataType()).collect(toList()));
+        }
+      };
+
+  private static final IcebergCompatCheck ICEBERG_COMPAT_V3_CHECK_HAS_ALLOWED_PARTITION_TYPES =
+      (inputContext) ->
+          inputContext
+              .newMetadata
+              .getPartitionColNames()
+              .forEach(
+                  partitionCol -> {
+                    int partitionFieldIndex =
+                        inputContext.newMetadata.getSchema().indexOf(partitionCol);
+                    checkArgument(
+                        partitionFieldIndex != -1,
+                        "Partition column %s not found in the schema",
+                        partitionCol);
+                    DataType dataType =
+                        inputContext.newMetadata.getSchema().at(partitionFieldIndex).getDataType();
+                    boolean validType =
+                        dataType instanceof ByteType
+                            || dataType instanceof ShortType
+                            || dataType instanceof IntegerType
+                            || dataType instanceof LongType
+                            || dataType instanceof FloatType
+                            || dataType instanceof DoubleType
+                            || dataType instanceof DecimalType
+                            || dataType instanceof StringType
+                            || dataType instanceof BinaryType
+                            || dataType instanceof BooleanType
+                            || dataType instanceof DateType
+                            || dataType instanceof TimestampType
+                            || dataType instanceof TimestampNTZType;
+                    if (!validType) {
+                      throw DeltaErrors.icebergCompatUnsupportedTypePartitionColumn(
+                          INSTANCE.compatFeatureName(), dataType);
+                    }
+                  });
+
+  private static final IcebergCompatCheck ICEBERG_COMPAT_V3_CHECK_HAS_NO_PARTITION_EVOLUTION =
+      (inputContext) -> {
+        // TODO: Kernel doesn't support replace table yet. When it is supported, extend
+        // this to allow checking the partition columns aren't changed
+      };
+
+  @Override
+  String compatFeatureName() {
+    return "icebergCompatV3";
+  }
+
+  @Override
+  TableConfig<Boolean> requiredDeltaTableProperty() {
+    return TableConfig.ICEBERG_COMPAT_V3_ENABLED;
+  }
+
+  @Override
+  List<IcebergCompatRequiredTablePropertyEnforcer> requiredDeltaTableProperties() {
+    return singletonList(ICEBERG_COMPAT_V3_CM_REQUIREMENT);
+  }
+
+  @Override
+  List<TableFeature> requiredDependencyTableFeatures() {
+    return Stream.of(ICEBERG_COMPAT_V3_W_FEATURE, COLUMN_MAPPING_RW_FEATURE).collect(toList());
+  }
+
+  @Override
+  List<IcebergCompatCheck> icebergCompatChecks() {
+    return Stream.of(
+            ICEBERG_COMPAT_V3_CHECK_NO_LOWER_COMPAT_ENABLED,
+            ICEBERG_COMPAT_V3_CHECK_HAS_SUPPORTED_TYPES,
+            ICEBERG_COMPAT_V3_CHECK_HAS_ALLOWED_PARTITION_TYPES,
+            ICEBERG_COMPAT_V3_CHECK_HAS_NO_PARTITION_EVOLUTION)
+        .collect(toList());
+  }
+}

kernel/kernel-api/src/main/java/io/delta/kernel/internal/icebergcompat/IcebergUniversalFormatMetadataValidatorAndUpdater.java

@@ -18,6 +18,8 @@
 import io.delta.kernel.exceptions.InvalidConfigurationValueException;
 import io.delta.kernel.internal.TableConfig;
 import io.delta.kernel.internal.actions.Metadata;
+import java.util.Arrays;
+import java.util.List;
 import java.util.Set;
 
 /**
@@ -51,14 +53,33 @@ public static void validate(Metadata metadata) {
 
     Set<String> targetFormats = TableConfig.UNIVERSAL_FORMAT_ENABLED_FORMATS.fromMetadata(metadata);
     boolean isIcebergEnabled = targetFormats.contains(TableConfig.UniversalFormats.FORMAT_ICEBERG);
-    boolean isIcebergCompatV2Enabled = TableConfig.ICEBERG_COMPAT_V2_ENABLED.fromMetadata(metadata);
-    if (isIcebergEnabled && !isIcebergCompatV2Enabled) {
+
+    List<TableConfig<Boolean>> icebergCompatOptions =
+        Arrays.asList(TableConfig.ICEBERG_COMPAT_V2_ENABLED, TableConfig.ICEBERG_COMPAT_V3_ENABLED);
+    long enabledCompatCount =
+        icebergCompatOptions.stream().filter(opt -> opt.fromMetadata(metadata)).count();
+
+    if (isIcebergEnabled && enabledCompatCount == 0) {
+      String optionKeys =
+          String.join(
+              " or ",
+              icebergCompatOptions.stream().map(TableConfig::getKey).toArray(String[]::new));
       throw new InvalidConfigurationValueException(
           TableConfig.UNIVERSAL_FORMAT_ENABLED_FORMATS.getKey(),
           TableConfig.UniversalFormats.FORMAT_ICEBERG,
           String.format(
-              "'%s' must be set to \"true\" to enable iceberg uniform format.",
-              TableConfig.ICEBERG_COMPAT_V2_ENABLED.getKey()));
+              "One of %s must be set to \"true\" to enable iceberg uniform format.", optionKeys));
+    }
+
+    if (enabledCompatCount > 1) {
+      String optionKeys =
+          String.join(
+              "' and '",
+              icebergCompatOptions.stream().map(TableConfig::getKey).toArray(String[]::new));
+      throw new InvalidConfigurationValueException(
+          TableConfig.UNIVERSAL_FORMAT_ENABLED_FORMATS.getKey(),
+          TableConfig.UniversalFormats.FORMAT_ICEBERG,
+          String.format("'%s' cannot be enabled at the same time.", optionKeys));
     }
   }
 }

kernel/kernel-api/src/main/java/io/delta/kernel/internal/tablefeatures/TableFeatures.java

@@ -326,6 +326,24 @@ public boolean metadataRequiresFeatureToBeEnabled(Protocol protocol, Metadata me
     }
   }
 
+  public static final TableFeature ICEBERG_COMPAT_V3_W_FEATURE = new IcebergCompatV3TableFeature();
+
+  private static class IcebergCompatV3TableFeature extends TableFeature.WriterFeature
+      implements FeatureAutoEnabledByMetadata {
+    IcebergCompatV3TableFeature() {
+      super("icebergCompatV3", /* minWriterVersion = */ 7);
+    }
+
+    @Override
+    public boolean metadataRequiresFeatureToBeEnabled(Protocol protocol, Metadata metadata) {
+      return TableConfig.ICEBERG_COMPAT_V3_ENABLED.fromMetadata(metadata);
+    }
+
+    public @Override Set<TableFeature> requiredFeatures() {
+      return Collections.singleton(COLUMN_MAPPING_RW_FEATURE);
+    }
+  }
+
   /* ---- Start: type widening ---- */
   // Base class for typeWidening and typeWidening-preview features. Both features are same in terms
   // of behavior and given the feature is graduated, we will enable the `typeWidening` by default
@@ -463,6 +481,7 @@ public boolean metadataRequiresFeatureToBeEnabled(Protocol protocol, Metadata me
               GENERATED_COLUMNS_W_FEATURE,
               DOMAIN_METADATA_W_FEATURE,
               ICEBERG_COMPAT_V2_W_FEATURE,
+              ICEBERG_COMPAT_V3_W_FEATURE,
               IDENTITY_COLUMNS_W_FEATURE,
               IN_COMMIT_TIMESTAMP_W_FEATURE,
               INVARIANTS_W_FEATURE,

kernel/kernel-api/src/test/scala/io/delta/kernel/internal/icebergcompat/IcebergUniversalFormatMetadataValidatorAndUpdaterSuite.scala

@@ -51,19 +51,21 @@ class IcebergUniversalFormatMetadataValidatorAndUpdaterSuite extends AnyFunSuite
 
   Seq(
     Map(TableConfig.ICEBERG_COMPAT_V2_ENABLED.getKey -> "false"),
-    Map[String, String]()).foreach { disableIcebergCompatV2 =>
+    Map(TableConfig.ICEBERG_COMPAT_V3_ENABLED.getKey -> "false"),
+    Map[String, String]()).foreach { disableIcebergCompat =>
     test(
       "validate should throw when iceberg universal format is enabled and "
-        + s"icebergCompatV2 isn't $disableIcebergCompatV2") {
+        + s"icebergCompatV2 isn't $disableIcebergCompat") {
       val metadata = createMetadata(Map(
         TableConfig.UNIVERSAL_FORMAT_ENABLED_FORMATS.getKey -> "iceberg",
-        "unrelated_key" -> "unrelated_value") ++ disableIcebergCompatV2)
+        "unrelated_key" -> "unrelated_value") ++ disableIcebergCompat)
       val exc = intercept[InvalidConfigurationValueException] {
         IcebergUniversalFormatMetadataValidatorAndUpdater.validate(metadata)
       }
       assert(exc.getMessage == "Invalid value for table property " +
         "'delta.universalFormat.enabledFormats': 'iceberg'. " +
-        "'delta.enableIcebergCompatV2' must be set to \"true\" to enable iceberg uniform format.")
+        "One of delta.enableIcebergCompatV2 or delta.enableIcebergCompatV3 " +
+        "must be set to \"true\" to enable iceberg uniform format.")
     }
   }