[Kernel][#1] IcebergCompatV3 - add icebergCompatV3 table feature (delta-io#4731)

<!--
Thanks for sending a pull request!  Here are some tips for you:
1. If this is your first time, please read our contributor guidelines:
https://github.com/delta-io/delta/blob/master/CONTRIBUTING.md
2. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP]
Your PR title ...'.
  3. Be sure to keep the PR description updated to reflect all changes.
  4. Please write your PR title to summarize what this PR proposes.
5. If possible, provide a concise example to reproduce the issue for a
faster review.
6. If applicable, include the corresponding issue number in the PR title
and link it in the body.
-->

#### Which Delta project/connector is this regarding?
<!--
Please add the component selected below to the beginning of the pull
request title
For example: [Spark] Title of my pull request
-->

- [ ] Spark
- [ ] Standalone
- [ ] Flink
- [x] Kernel
- [ ] Other (fill in here)

## Description

<!--
- Describe what this PR changes.
- Describe why we need the change.
 
If this PR resolves an issue be sure to include "Resolves #XXX" to
correctly link and close the issue upon merge.
-->
Add the IcebergCompatV3 table feature, which requires the columnMapping
and rowTracking to be enabled.
## How was this patch tested?

<!--
If tests were added, say they were added here. Please make sure to test
the changes thoroughly including negative and positive cases if
possible.
If the changes were tested in any way other than unit tests, please
clarify how you tested step by step (ideally copy and paste-able, so
that other reviewers can test and check, and descendants can verify in
the future).
If the changes were not tested, please explain why.
-->
Unit tests
## Does this PR introduce _any_ user-facing changes?

<!--
If yes, please clarify the previous behavior and the change this PR
proposes - provide the console output, description and/or an example to
show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change
compared to the released Delta Lake versions or within the unreleased
branches such as master.
If no, write 'No'.
-->

Author: KaiqiJinWow

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

@@ -242,6 +242,19 @@ 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. TODO: add the delta protocol link once updated
+   */
+  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.
@@ -357,6 +370,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/data/TransactionStateRow.java

@@ -114,6 +114,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 column mapping mode from the transaction state {@link Row} returned by {@link
    * Transaction#getTransactionState(Engine)}

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

@@ -322,6 +322,25 @@ 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.unmodifiableSet(
+          new HashSet<>(Arrays.asList(COLUMN_MAPPING_RW_FEATURE, ROW_TRACKING_W_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
@@ -459,6 +478,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,22 +51,37 @@ 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") {
+      "validate should throw when iceberg universal format is enabled but  "
+        + s"no IcebergCompat version is enabled: $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.")
     }
   }
 
+  test("validate should throw when both IcebergCompatV2 and V3 are enabled") {
+    val metadata = createMetadata(Map(
+      TableConfig.UNIVERSAL_FORMAT_ENABLED_FORMATS.getKey -> "iceberg",
+      TableConfig.ICEBERG_COMPAT_V2_ENABLED.getKey -> "true",
+      TableConfig.ICEBERG_COMPAT_V3_ENABLED.getKey -> "true"))
+    val exc = intercept[InvalidConfigurationValueException] {
+      IcebergUniversalFormatMetadataValidatorAndUpdater.validate(metadata)
+    }
+    assert(exc.getMessage.contains(
+      "'delta.enableIcebergCompatV2' and 'delta.enableIcebergCompatV3' " +
+        "cannot be enabled at the same time."))
+  }
+
   def createMetadata(tblProps: Map[String, String] = Map.empty): Metadata = {
     val schema = new StructType()
       .add("c1", IntegerType.INTEGER)

kernel/kernel-api/src/test/scala/io/delta/kernel/internal/tablefeatures/TableFeaturesSuite.scala

@@ -61,6 +61,7 @@ class TableFeaturesSuite extends AnyFunSuite {
     "rowTracking",
     "domainMetadata",
     "icebergCompatV2",
+    "icebergCompatV3",
     "inCommitTimestamp",
     "icebergWriterCompatV1",
     "clustering")
@@ -153,6 +154,14 @@ class TableFeaturesSuite extends AnyFunSuite {
       "icebergCompatV2",
       testMetadata(tblProps = Map("delta.enableIcebergCompatV2" -> "false")),
       false),
+    (
+      "icebergCompatV3",
+      testMetadata(tblProps = Map("delta.enableIcebergCompatV3" -> "true")),
+      true),
+    (
+      "icebergCompatV3",
+      testMetadata(tblProps = Map("delta.enableIcebergCompatV3" -> "false")),
+      false),
     (
       "inCommitTimestamp",
       testMetadata(tblProps = Map("delta.enableInCommitTimestamps" -> "true")),
@@ -238,6 +247,7 @@ class TableFeaturesSuite extends AnyFunSuite {
       "rowTracking",
       "domainMetadata",
       "icebergCompatV2",
+      "icebergCompatV3",
       "inCommitTimestamp",
       "appendOnly",
       "invariants",
@@ -568,6 +578,11 @@ class TableFeaturesSuite extends AnyFunSuite {
     new Protocol(3, 7, emptySet(), singleton("icebergCompatV2")),
     testMetadata(tblProps = Map("delta.enableIcebergCompatV2" -> "true")))
 
+  checkWriteSupported(
+    "validateKernelCanWriteToTable: protocol 7 with icebergCompatV3",
+    new Protocol(3, 7, emptySet(), singleton("icebergCompatV3")),
+    testMetadata(tblProps = Map("delta.enableIcebergCompatV3" -> "true")))
+
   checkWriteSupported(
     "validateKernelCanWriteToTable: protocol 7 with v2Checkpoint, " +
       "metadata enables v2Checkpoint",
@@ -791,6 +806,21 @@ class TableFeaturesSuite extends AnyFunSuite {
         set(),
         set("columnMapping", "appendOnly", "invariants", "icebergCompatV2")),
       set("icebergCompatV2", "columnMapping")),
+    (
+      testMetadata(tblProps = Map("delta.enableIcebergCompatV3" -> "true")),
+      new Protocol(1, 2),
+      new Protocol(
+        2,
+        7,
+        set(),
+        set(
+          "columnMapping",
+          "appendOnly",
+          "invariants",
+          "icebergCompatV3",
+          "domainMetadata",
+          "rowTracking")),
+      set("icebergCompatV3", "domainMetadata", "columnMapping", "rowTracking")),
     (
       testMetadata(tblProps =
         Map("delta.enableIcebergCompatV2" -> "true", "delta.enableDeletionVectors" -> "true")),
@@ -819,6 +849,21 @@ class TableFeaturesSuite extends AnyFunSuite {
         set("columnMapping", "deletionVectors"),
         set("columnMapping", "icebergCompatV2", "deletionVectors")),
       set("icebergCompatV2")),
+    (
+      testMetadata(tblProps =
+        Map("delta.enableIcebergCompatV3" -> "true")),
+      new Protocol(3, 7, set("columnMapping", "deletionVectors"), set("columnMapping")),
+      new Protocol(
+        3,
+        7,
+        set("columnMapping", "deletionVectors"),
+        set(
+          "columnMapping",
+          "icebergCompatV3",
+          "deletionVectors",
+          "domainMetadata",
+          "rowTracking")),
+      set("icebergCompatV3", "domainMetadata", "rowTracking")),
     (
       testMetadata(tblProps = Map("delta.enableIcebergWriterCompatV1" -> "true")),
       new Protocol(1, 2),
@@ -963,6 +1008,15 @@ class TableFeaturesSuite extends AnyFunSuite {
       // that is of partial (writer only) table feature support
       testMetadata(tblProps = Map("delta.enableIcebergCompatV2" -> "true")),
       new Protocol(2, 7, set(), set("columnMapping", "icebergCompatV2"))),
+    (
+      // try to enable the feature that is already supported on a protocol
+      // that is of partial (writer only) table feature support
+      testMetadata(tblProps = Map("delta.enableIcebergCompatV3" -> "true")),
+      new Protocol(
+        2,
+        7,
+        set(),
+        set("columnMapping", "icebergCompatV3", "domainMetadata", "rowTracking"))),
     (
       // try to enable the feature that is already supported on a protocol
       // that is of table feature support