[Kernel] Support column mapping mode in REPLACE TABLE including with fieldId re-use (#4518)

<!--
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

PR#1 here #4520 separates some of
this out.

This PR re-enables support for column mapping tables with REPLACE, and
allows fieldId re-use when the type/nullability is compatible. To check
for this compatibility we essentially re-use our schema evolution code.
Valid schema transformations in the normal case should also be valid
changes when reusing fieldIds across REPLACE. Any fields in the
replaceSchema that are not reusing existing fieldIds will be treated as
new fields during the schema evolution validation.

This validation checks that nullability changes are valid (cannot change
from nullable to non-nullable) and that type changes are valid (for now,
cannot change types). Columns are able to be moved (or nested columns
dropped) if the other invariants hold.

Note, this currently restricts REPLACE to operations that maintain the
same column mapping mode. In the future, we will loosen this restriction
but for now this lessens the scope of this PR.

## How was this patch tested?

Adds unit tests.

Author: allisonport-db

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

@@ -19,13 +19,13 @@
 import static io.delta.kernel.internal.DeltaErrors.tableAlreadyExists;
 import static io.delta.kernel.internal.TransactionImpl.DEFAULT_READ_VERSION;
 import static io.delta.kernel.internal.TransactionImpl.DEFAULT_WRITE_VERSION;
-import static io.delta.kernel.internal.util.ColumnMapping.getColumnMappingMode;
 import static io.delta.kernel.internal.util.ColumnMapping.isColumnMappingModeEnabled;
 import static io.delta.kernel.internal.util.Preconditions.checkArgument;
 import static io.delta.kernel.internal.util.SchemaUtils.casePreservingPartitionColNames;
 import static io.delta.kernel.internal.util.VectorUtils.buildArrayValue;
 import static io.delta.kernel.internal.util.VectorUtils.stringStringMapValue;
 import static java.util.Collections.emptyList;
+import static java.util.Collections.emptySet;
 import static java.util.Objects.requireNonNull;
 import static java.util.stream.Collectors.toSet;
 
@@ -300,7 +300,12 @@ protected TransactionImpl buildTransactionInternal(
             : ClusteringUtils.getClusteringColumnsOptional(latestSnapshot.get());
     Tuple2<Optional<Protocol>, Optional<Metadata>> updatedProtocolAndMetadata =
         validateAndUpdateProtocolAndMetadata(
-            engine, baseMetadata, baseProtocol, isCreateOrReplace, existingClusteringCols);
+            engine,
+            baseMetadata,
+            baseProtocol,
+            isCreateOrReplace,
+            existingClusteringCols,
+            latestSnapshot);
     Optional<Protocol> newProtocol = updatedProtocolAndMetadata._1;
     Optional<Metadata> newMetadata = updatedProtocolAndMetadata._2;
 
@@ -321,11 +326,6 @@ protected TransactionImpl buildTransactionInternal(
 
     // Block this for now - in a future PR we will enable this
     if (operation == Operation.REPLACE_TABLE) {
-      if (getColumnMappingMode(newMetadata.orElse(baseMetadata).getConfiguration())
-          != ColumnMappingMode.NONE) {
-        throw new UnsupportedOperationException(
-            "REPLACE TABLE is not yet supported with column mapping");
-      }
       if (newProtocol.orElse(baseProtocol).supportsFeature(TableFeatures.ROW_TRACKING_W_FEATURE)) {
         // Block this for now to be safe, we will return to this in the future
         throw new UnsupportedOperationException(
@@ -365,7 +365,8 @@ protected Tuple2<Optional<Protocol>, Optional<Metadata>> validateAndUpdateProtoc
       Metadata baseMetadata,
       Protocol baseProtocol,
       boolean isCreateOrReplace,
-      Optional<List<Column>> existingClusteringCols) {
+      Optional<List<Column>> existingClusteringCols,
+      Optional<SnapshotImpl> latestSnapshot) {
     if (isCreateOrReplace) {
       checkArgument(!existingClusteringCols.isPresent());
     }
@@ -477,7 +478,11 @@ protected Tuple2<Optional<Protocol>, Optional<Metadata>> validateAndUpdateProtoc
     // Now that all the config and schema changes have been made validate the old vs new metadata
     if (newMetadata.isPresent()) {
       validateMetadataChange(
-          existingClusteringCols, baseMetadata, newMetadata.get(), isCreateOrReplace);
+          existingClusteringCols,
+          baseMetadata,
+          newMetadata.get(),
+          isCreateOrReplace,
+          latestSnapshot);
     }
 
     return new Tuple2(newProtocol, newMetadata);
@@ -572,7 +577,8 @@ private void validateMetadataChange(
       Optional<List<Column>> existingClusteringCols,
       Metadata oldMetadata,
       Metadata newMetadata,
-      boolean isCreateOrReplace) {
+      boolean isCreateOrReplace,
+      Optional<SnapshotImpl> latestSnapshot) {
     ColumnMapping.verifyColumnMappingChange(
         oldMetadata.getConfiguration(), newMetadata.getConfiguration(), isCreateOrReplace);
     IcebergWriterCompatV1MetadataValidatorAndUpdater.validateIcebergWriterCompatV1Change(
@@ -611,7 +617,36 @@ private void validateMetadataChange(
           oldMetadata,
           newMetadata,
           clusteringColumnPhysicalNames,
-          false /* allowNewRequiredFields*/);
+          false /* allowNewRequiredFields */);
+    }
+
+    // For replace table we need to do special validation in the case of fieldId re-use
+    if (isCreateOrReplace && latestSnapshot.isPresent()) {
+      // For now, we don't support changing column mapping mode during replace, in a future PR we
+      // will loosen this restriction
+      ColumnMappingMode oldMode =
+          ColumnMapping.getColumnMappingMode(latestSnapshot.get().getMetadata().getConfiguration());
+      ColumnMappingMode newMode =
+          ColumnMapping.getColumnMappingMode(newMetadata.getConfiguration());
+      if (oldMode != newMode) {
+        throw new UnsupportedOperationException(
+            String.format(
+                "Changing column mapping mode from %s to %s is not currently supported in Kernel "
+                    + "during REPLACE TABLE operations",
+                oldMode, newMode));
+      }
+
+      // We only need to check fieldId re-use when cmMode != none
+      if (newMode != ColumnMappingMode.NONE) {
+        SchemaUtils.validateUpdatedSchema(
+            latestSnapshot.get().getMetadata(),
+            newMetadata,
+            // We already validate clustering columns elsewhere for isCreateOrReplace no need to
+            // duplicate this check here
+            emptySet() /* clusteringCols */,
+            // We allow new non-null fields in REPLACE since we know all existing data is removed
+            true /* allowNewRequiredFields */);
+      }
     }
   }
 

kernel/kernel-api/src/test/scala/io/delta/kernel/internal/util/ColumnMappingSuiteBase.scala

@@ -129,8 +129,9 @@ trait ColumnMappingSuiteBase extends VectorTestUtils {
       metadata: Metadata,
       isNewTable: Boolean = true,
       enableIcebergCompatV2: Boolean = true,
-      enableIcebergWriterCompatV1: Boolean = false): Unit = {
-    var fieldId: Long = 0L
+      enableIcebergWriterCompatV1: Boolean = false,
+      initialFieldId: Long = 0L): Unit = {
+    var fieldId: Long = initialFieldId
 
     def nextFieldId: Long = {
       fieldId += 1