[Kernel][#4] Introduce IcebergWriterCompatV3 (delta-io#4774)

<!--
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.
-->
- Implement IcebergWriterCompatV3. We skipped v2 to align with
IcebergCompatV3 which maps to iceberg format v3.


## 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'.
-->
Yes. Introduces IcebergWriterCompatV3.

Author: murali-db

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

@@ -41,6 +41,7 @@
 import io.delta.kernel.internal.icebergcompat.IcebergCompatV3MetadataValidatorAndUpdater;
 import io.delta.kernel.internal.icebergcompat.IcebergUniversalFormatMetadataValidatorAndUpdater;
 import io.delta.kernel.internal.icebergcompat.IcebergWriterCompatV1MetadataValidatorAndUpdater;
+import io.delta.kernel.internal.icebergcompat.IcebergWriterCompatV3MetadataValidatorAndUpdater;
 import io.delta.kernel.internal.lang.Lazy;
 import io.delta.kernel.internal.metrics.SnapshotMetrics;
 import io.delta.kernel.internal.metrics.SnapshotQueryContext;
@@ -481,6 +482,16 @@ protected Tuple2<Optional<Protocol>, Optional<Metadata>> validateAndUpdateProtoc
       newMetadata = icebergWriterCompatV1;
     }
 
+    Optional<Metadata> icebergWriterCompatV3 =
+        IcebergWriterCompatV3MetadataValidatorAndUpdater
+            .validateAndUpdateIcebergWriterCompatV3Metadata(
+                isCreateOrReplace,
+                newMetadata.orElse(baseMetadata),
+                newProtocol.orElse(baseProtocol));
+    if (icebergWriterCompatV3.isPresent()) {
+      newMetadata = icebergWriterCompatV3;
+    }
+
     // TODO: refactor this method to use a single validator and updater.
     Optional<Metadata> icebergCompatV2Metadata =
         IcebergCompatV2MetadataValidatorAndUpdater.validateAndUpdateIcebergCompatV2Metadata(

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

@@ -34,6 +34,7 @@
 import io.delta.kernel.internal.data.TransactionStateRow;
 import io.delta.kernel.internal.fs.Path;
 import io.delta.kernel.internal.icebergcompat.IcebergCompatV2MetadataValidatorAndUpdater;
+import io.delta.kernel.internal.icebergcompat.IcebergCompatV3MetadataValidatorAndUpdater;
 import io.delta.kernel.statistics.DataFileStatistics;
 import io.delta.kernel.types.StructType;
 import io.delta.kernel.utils.CloseableIterable;
@@ -107,6 +108,57 @@ public static Row generateIcebergCompatWriterV1AddAction(
         transactionState, fileStatus, partitionValues, dataChange, Collections.emptyMap());
   }
 
+  /**
+   * Create an add action {@link Row} that can be passed to {@link Transaction#commit(Engine,
+   * CloseableIterable)} from an Iceberg add.
+   *
+   * @param transactionState the transaction state from the built transaction
+   * @param fileStatus the file status to create the add with (contains path, time, size, and stats)
+   * @param partitionValues the partition values for the add
+   * @param dataChange whether or not the add constitutes a dataChange (i.e. append vs. compaction)
+   * @param tags key-value metadata to be attached to the add action
+   * @return add action row that can be included in the transaction
+   * @throws UnsupportedOperationException if icebergWriterCompatV3 is not enabled
+   * @throws UnsupportedOperationException if maxRetries != 0 in the transaction
+   * @throws KernelException if stats are not present (required for icebergCompatV3)
+   * @throws UnsupportedOperationException if the table is partitioned (currently unsupported)
+   */
+  public static Row generateIcebergCompatWriterV3AddAction(
+      Row transactionState,
+      DataFileStatus fileStatus,
+      Map<String, Literal> partitionValues,
+      boolean dataChange,
+      Map<String, String> tags) {
+    Map<String, String> configuration = TransactionStateRow.getConfiguration(transactionState);
+
+    /* ----- Validate that this is a valid usage of this API ----- */
+    validateIcebergWriterCompatV3Enabled(configuration);
+    validateMaxRetriesSetToZero(transactionState);
+
+    /* ----- Validate this is valid write given the table's protocol & configurations ----- */
+    checkState(
+        TableConfig.ICEBERG_COMPAT_V3_ENABLED.fromMetadata(configuration),
+        "icebergCompatV3 not enabled despite icebergWriterCompatV3 enabled");
+    // We require field `numRecords` when icebergCompatV3 is enabled
+    IcebergCompatV3MetadataValidatorAndUpdater.validateDataFileStatus(fileStatus);
+
+    /* --- Validate and update partitionValues ---- */
+    // Currently we don't support partitioned tables; fail here
+    blockPartitionedTables(transactionState, partitionValues);
+
+    URI tableRoot = new Path(TransactionStateRow.getTablePath(transactionState)).toUri();
+    // This takes care of relativizing the file path and serializing the file statistics
+    AddFile addFile =
+        AddFile.convertDataFileStatus(
+            TransactionStateRow.getPhysicalSchema(transactionState),
+            tableRoot,
+            fileStatus,
+            partitionValues,
+            dataChange,
+            tags);
+    return SingleAction.createAddFileSingleAction(addFile.toRow());
+  }
+
   /**
    * Create a remove action {@link Row} that can be passed to {@link Transaction#commit(Engine,
    * CloseableIterable)} from an Iceberg remove.
@@ -134,13 +186,64 @@ public static Row generateIcebergCompatWriterV1RemoveAction(
     validateIcebergWriterCompatV1Enabled(config);
     validateMaxRetriesSetToZero(transactionState);
 
+    /* ----- Validate this is valid write given the table's protocol & configurations ----- */
+    // We only allow removes with dataChange=false when appendOnly=true
+    blockUpdatingAppendOnlyTables(dataChange, transactionState, config);
+
+    /* --- Validate and update partitionValues ---- */
+    // Currently we don't support partitioned tables; fail here
+    blockPartitionedTables(transactionState, partitionValues);
+
+    URI tableRoot = new Path(TransactionStateRow.getTablePath(transactionState)).toUri();
+    // This takes care of relativizing the file path and serializing the file statistics
+    Row removeFileRow =
+        convertRemoveDataFileStatus(
+            TransactionStateRow.getPhysicalSchema(transactionState),
+            tableRoot,
+            fileStatus,
+            partitionValues,
+            dataChange);
+    return SingleAction.createRemoveFileSingleAction(removeFileRow);
+  }
+
+  /**
+   * Create a remove action {@link Row} that can be passed to {@link Transaction#commit(Engine,
+   * CloseableIterable)} from an Iceberg remove.
+   *
+   * @param transactionState the transaction state from the built transaction
+   * @param fileStatus the file status to create the remove with (contains path, time, size, and
+   *     stats)
+   * @param partitionValues the partition values for the remove
+   * @param dataChange whether or not the remove constitutes a dataChange (i.e. delete vs.
+   *     compaction)
+   * @return remove action row that can be committed to the transaction
+   * @throws UnsupportedOperationException if icebergWriterCompatV3 is not enabled
+   * @throws UnsupportedOperationException if maxRetries != 0 in the transaction
+   * @throws KernelException if the table is an append-only table and dataChange=true
+   * @throws UnsupportedOperationException if the table is partitioned (currently unsupported)
+   */
+  public static Row generateIcebergCompatWriterV3RemoveAction(
+      Row transactionState,
+      DataFileStatus fileStatus,
+      Map<String, Literal> partitionValues,
+      boolean dataChange) {
+    Map<String, String> config = TransactionStateRow.getConfiguration(transactionState);
+
+    /* ----- Validate that this is a valid usage of this API ----- */
+    validateIcebergWriterCompatV3Enabled(config);
+    validateMaxRetriesSetToZero(transactionState);
+
     /* ----- Validate this is valid write given the table's protocol & configurations ----- */
     // We only allow removes with dataChange=false when appendOnly=true
     if (dataChange && TableConfig.APPEND_ONLY_ENABLED.fromMetadata(config)) {
       throw DeltaErrors.cannotModifyAppendOnlyTable(
           TransactionStateRow.getTablePath(transactionState));
     }
 
+    /* ----- Validate this is valid write given the table's protocol & configurations ----- */
+    // We only allow removes with dataChange=false when appendOnly=true
+    blockUpdatingAppendOnlyTables(dataChange, transactionState, config);
+
     /* --- Validate and update partitionValues ---- */
     // Currently we don't support partitioned tables; fail here
     blockPartitionedTables(transactionState, partitionValues);
@@ -177,6 +280,21 @@ private static void validateIcebergWriterCompatV1Enabled(Map<String, String> con
     }
   }
 
+  /**
+   * Validates that table feature `icebergWriterCompatV3` is enabled. We restrict usage of these
+   * APIs to require that this table feature is enabled to prevent any unsafe usage due to the table
+   * features that are blocked via `icebergWriterCompatV3`.
+   */
+  private static void validateIcebergWriterCompatV3Enabled(Map<String, String> config) {
+    if (!TableConfig.ICEBERG_WRITER_COMPAT_V3_ENABLED.fromMetadata(config)) {
+      throw new UnsupportedOperationException(
+          String.format(
+              "APIs within GenerateIcebergCompatActionUtils are only supported on tables with"
+                  + " '%s' set to true",
+              TableConfig.ICEBERG_WRITER_COMPAT_V3_ENABLED.getKey()));
+    }
+  }
+
   /**
    * Throws an exception if `maxRetries` was not set to 0 in the transaction. We restrict these APIs
    * to require `maxRetries = 0` since conflict resolution is not supported for operations other
@@ -192,6 +310,15 @@ private static void validateMaxRetriesSetToZero(Row transactionState) {
     }
   }
 
+  private static void blockUpdatingAppendOnlyTables(
+      boolean dataChange, Row transactionState, Map<String, String> config) {
+    // We only allow removes with dataChange=false when appendOnly=true
+    if (dataChange && TableConfig.APPEND_ONLY_ENABLED.fromMetadata(config)) {
+      throw DeltaErrors.cannotModifyAppendOnlyTable(
+          TransactionStateRow.getTablePath(transactionState));
+    }
+  }
+
   private static void blockPartitionedTables(
       Row transactionState, Map<String, Literal> partitionValues) {
     if (!TransactionStateRow.getPartitionColumnsList(transactionState).isEmpty()) {

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

@@ -27,6 +27,7 @@
 import io.delta.kernel.internal.util.SchemaIterable;
 import io.delta.kernel.types.*;
 import java.util.*;
+import java.util.stream.Stream;
 
 /**
  * Contains interfaces and common utility classes performing the validations and updates necessary
@@ -55,6 +56,49 @@ abstract class IcebergWriterCompatMetadataValidatorAndUpdater
               ColumnMapping.updateColumnMappingMetadataIfNeeded(
                   inputContext.newMetadata, inputContext.isCreatingNewTable));
 
+  /**
+   * Creates an IcebergCompatRequiredTablePropertyEnforcer for enabling a specific Iceberg
+   * compatibility version. The enforcer ensures the property is set to "true" and delegates
+   * validation to the appropriate metadata validator.
+   *
+   * @param tableConfigProperty the table configuration property to enforce
+   * @param postProcessor the version-specific validation and metadata update processor
+   * @return configured enforcer for the specified Iceberg compatibility version
+   */
+  protected static IcebergCompatRequiredTablePropertyEnforcer<Boolean> createIcebergCompatEnforcer(
+      TableConfig<Boolean> tableConfigProperty, PostMetadataProcessor postProcessor) {
+    return new IcebergCompatRequiredTablePropertyEnforcer<>(
+        tableConfigProperty, (value) -> value, "true", postProcessor);
+  }
+
+  /**
+   * Common set of allowed table features shared across all Iceberg writer compatibility versions.
+   * This includes the incompatible legacy features (invariants, changeDataFeed, checkConstraints,
+   * identityColumns, generatedColumns) because they may be present in the table protocol even when
+   * they are not in use. In later checks we validate that these incompatible features are inactive
+   * in the table. See the protocol spec for more details.
+   */
+  protected static final Set<TableFeature> COMMON_ALLOWED_FEATURES =
+      Stream.of(
+              // Incompatible, but not active, legacy table features
+              INVARIANTS_W_FEATURE,
+              CHANGE_DATA_FEED_W_FEATURE,
+              CONSTRAINTS_W_FEATURE,
+              IDENTITY_COLUMNS_W_FEATURE,
+              GENERATED_COLUMNS_W_FEATURE,
+              // Compatible table features
+              APPEND_ONLY_W_FEATURE,
+              COLUMN_MAPPING_RW_FEATURE,
+              DOMAIN_METADATA_W_FEATURE,
+              VACUUM_PROTOCOL_CHECK_RW_FEATURE,
+              CHECKPOINT_V2_RW_FEATURE,
+              IN_COMMIT_TIMESTAMP_W_FEATURE,
+              CLUSTERING_W_FEATURE,
+              TIMESTAMP_NTZ_RW_FEATURE,
+              TYPE_WIDENING_RW_FEATURE,
+              TYPE_WIDENING_RW_PREVIEW_FEATURE)
+          .collect(toSet());
+
   protected static IcebergCompatCheck createUnsupportedFeaturesCheck(
       IcebergWriterCompatMetadataValidatorAndUpdater instance) {
     return (inputContext) -> {
@@ -209,6 +253,16 @@ protected static IcebergCompatCheck createUnsupportedFeaturesCheck(
         }
       };
 
+  protected static final List<IcebergCompatCheck> COMMON_CHECKS =
+      Arrays.asList(
+          UNSUPPORTED_TYPES_CHECK,
+          PHYSICAL_NAMES_MATCH_FIELD_IDS_CHECK,
+          INVARIANTS_INACTIVE_CHECK,
+          CHANGE_DATA_FEED_INACTIVE_CHECK,
+          CHECK_CONSTRAINTS_INACTIVE_CHECK,
+          IDENTITY_COLUMNS_INACTIVE_CHECK,
+          GENERATED_COLUMNS_INACTIVE_CHECK);
+
   @Override
   abstract String compatFeatureName();
 

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

@@ -107,45 +107,29 @@ public static Optional<Metadata> validateAndUpdateIcebergWriterCompatV1Metadata(
   private static final IcebergWriterCompatV1MetadataValidatorAndUpdater INSTANCE =
       new IcebergWriterCompatV1MetadataValidatorAndUpdater();
 
-  private static final IcebergCompatRequiredTablePropertyEnforcer ICEBERG_COMPAT_V2_ENABLED =
-      new IcebergCompatRequiredTablePropertyEnforcer<>(
-          TableConfig.ICEBERG_COMPAT_V2_ENABLED,
-          (value) -> value,
-          "true",
-          (inputContext) ->
-              IcebergCompatV2MetadataValidatorAndUpdater.validateAndUpdateIcebergCompatV2Metadata(
-                  inputContext.isCreatingNewTable,
-                  inputContext.newMetadata,
-                  inputContext.newProtocol));
+  /**
+   * Enforcer for Iceberg compatibility V2 (required by V1). Ensures the ICEBERG_COMPAT_V2_ENABLED
+   * property is set to "true" and delegates validation to the V2 metadata validator.
+   */
+  private static final IcebergCompatRequiredTablePropertyEnforcer<Boolean>
+      ICEBERG_COMPAT_V2_ENABLED =
+          createIcebergCompatEnforcer(
+              TableConfig.ICEBERG_COMPAT_V2_ENABLED,
+              (inputContext) ->
+                  IcebergCompatV2MetadataValidatorAndUpdater
+                      .validateAndUpdateIcebergCompatV2Metadata(
+                          inputContext.isCreatingNewTable,
+                          inputContext.newMetadata,
+                          inputContext.newProtocol));
 
   /**
-   * Current set of allowed table features. This may evolve as the protocol evolves. This includes
-   * the incompatible legacy features (invariants, changeDataFeed, checkConstraints,
-   * identityColumns, generatedColumns) because they may be present in the table protocol even when
-   * they are not in use. In later checks we validate that these incompatible features are inactive
-   * in the table. See the protocol spec for more details.
+   * Current set of allowed table features for Iceberg writer compat V1. This combines the common
+   * features with V1-specific features (ICEBERG_COMPAT_V2_W_FEATURE, ICEBERG_WRITER_COMPAT_V1).
    */
   private static Set<TableFeature> ALLOWED_TABLE_FEATURES =
-      Stream.of(
-              // Incompatible legacy table features
-              INVARIANTS_W_FEATURE,
-              CHANGE_DATA_FEED_W_FEATURE,
-              CONSTRAINTS_W_FEATURE,
-              IDENTITY_COLUMNS_W_FEATURE,
-              GENERATED_COLUMNS_W_FEATURE,
-              // Compatible table features
-              APPEND_ONLY_W_FEATURE,
-              COLUMN_MAPPING_RW_FEATURE,
-              ICEBERG_COMPAT_V2_W_FEATURE,
-              ICEBERG_WRITER_COMPAT_V1,
-              DOMAIN_METADATA_W_FEATURE,
-              VACUUM_PROTOCOL_CHECK_RW_FEATURE,
-              CHECKPOINT_V2_RW_FEATURE,
-              IN_COMMIT_TIMESTAMP_W_FEATURE,
-              CLUSTERING_W_FEATURE,
-              TIMESTAMP_NTZ_RW_FEATURE,
-              TYPE_WIDENING_RW_FEATURE,
-              TYPE_WIDENING_RW_PREVIEW_FEATURE)
+      Stream.concat(
+              COMMON_ALLOWED_FEATURES.stream(),
+              Stream.of(ICEBERG_COMPAT_V2_W_FEATURE, ICEBERG_WRITER_COMPAT_V1))
           .collect(toSet());
 
   @Override
@@ -177,15 +161,7 @@ protected Set<TableFeature> getAllowedTableFeatures() {
 
   @Override
   List<IcebergCompatCheck> icebergCompatChecks() {
-    return Stream.of(
-            createUnsupportedFeaturesCheck(this), // Pass 'this' instance
-            UNSUPPORTED_TYPES_CHECK,
-            PHYSICAL_NAMES_MATCH_FIELD_IDS_CHECK,
-            INVARIANTS_INACTIVE_CHECK,
-            CHANGE_DATA_FEED_INACTIVE_CHECK,
-            CHECK_CONSTRAINTS_INACTIVE_CHECK,
-            IDENTITY_COLUMNS_INACTIVE_CHECK,
-            GENERATED_COLUMNS_INACTIVE_CHECK)
+    return Stream.concat(Stream.of(createUnsupportedFeaturesCheck(this)), COMMON_CHECKS.stream())
         .collect(toList());
   }
 }