[Kernel] Update column mapping and schema evolution code to support usage with replace table (delta-io#4520)

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

Update SchemaUtils and ColumnMapping with unit tests in order to support
REPLACE TABLE with column mapping + fieldId re-use in PR #2.
Specifically this involves the following changes (not necessarily
related, but combined in this PR)

1) When a connector provides its own column mapping info in the schema
pre-populated we require that it's complete (i.e. fieldId AND
physicalName must be present)
2) We add an argument to our schema validation checks
`allowNewNonNullableFields`. This is useful in cases where we can be
sure the table state has been completely cleared, and thus new non-null
fields are valid (like REPLACE).
3) We don't allow adding a new column with a fieldId less than the
maxColId. For now, do this proactively for safety. In the future in the
case of something like RESTORE in the future we will likely need a
config to bypass this check.

## How was this patch tested?

Updates unit tests.

Also, all the changes in this PR are used by
delta-io#4520 which adds a lot more E2E
tests with multiple schema scenarios.

## Does this PR introduce _any_ user-facing changes?

No.

Author: allisonport-db

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

@@ -608,11 +608,10 @@ private void validateMetadataChange(
               .collect(toSet());
 
       SchemaUtils.validateUpdatedSchema(
-          oldMetadata.getSchema(),
-          newMetadata.getSchema(),
-          oldMetadata.getPartitionColNames(),
+          oldMetadata,
+          newMetadata,
           clusteringColumnPhysicalNames,
-          newMetadata);
+          false /* allowNewRequiredFields*/);
     }
   }
 

kernel/kernel-api/src/main/java/io/delta/kernel/internal/util/ColumnMapping.java

@@ -468,6 +468,15 @@ private static StructField assignColumnIdAndPhysicalNameToField(
       AtomicInteger maxColumnId,
       boolean isNewTable,
       boolean useColumnIdForPhysicalName) {
+    if (hasColumnId(field) ^ hasPhysicalName(field)) {
+      // If a connector is providing column mapping metadata in the given schema we require it to be
+      // complete
+      throw new IllegalArgumentException(
+          String.format(
+              "Both columnId and physicalName must be present if one is present. "
+                  + "Found this field with incomplete column mapping metadata: %s",
+              field));
+    }
     if (!hasColumnId(field)) {
       field =
           field.withNewMetadata(

kernel/kernel-api/src/main/java/io/delta/kernel/internal/util/SchemaUtils.java

@@ -101,18 +101,27 @@ public static void validateSchema(StructType schema, boolean isColumnMappingEnab
    * </ul>
    */
   public static void validateUpdatedSchema(
-      StructType currentSchema,
-      StructType newSchema,
-      Set<String> currentPartitionColumns,
+      Metadata currentMetadata,
+      Metadata newMetadata,
       Set<String> clusteringColumnPhysicalNames,
-      Metadata newMetadata) {
+      boolean allowNewRequiredFields) {
     checkArgument(
         isColumnMappingModeEnabled(
             ColumnMapping.getColumnMappingMode(newMetadata.getConfiguration())),
         "Cannot validate updated schema when column mapping is disabled");
-    validateSchema(newSchema, true /*columnMappingEnabled*/);
-    validatePartitionColumns(newSchema, new ArrayList<>(currentPartitionColumns));
-    validateSchemaEvolution(currentSchema, newSchema, newMetadata, clusteringColumnPhysicalNames);
+    validateSchema(newMetadata.getSchema(), true /*columnMappingEnabled*/);
+    validatePartitionColumns(
+        newMetadata.getSchema(), new ArrayList<>(newMetadata.getPartitionColNames()));
+    int currentMaxFieldId =
+        Integer.parseInt(
+            currentMetadata.getConfiguration().getOrDefault(COLUMN_MAPPING_MAX_COLUMN_ID_KEY, "0"));
+    validateSchemaEvolution(
+        currentMetadata.getSchema(),
+        newMetadata.getSchema(),
+        ColumnMapping.getColumnMappingMode(newMetadata.getConfiguration()),
+        clusteringColumnPhysicalNames,
+        currentMaxFieldId,
+        allowNewRequiredFields);
   }
 
   /**
@@ -426,14 +435,19 @@ private static void validatePhysicalNameConsistency(
   private static void validateSchemaEvolution(
       StructType currentSchema,
       StructType newSchema,
-      Metadata metadata,
-      Set<String> clusteringColumnPhysicalNames) {
-    ColumnMappingMode columnMappingMode =
-        ColumnMapping.getColumnMappingMode(metadata.getConfiguration());
+      ColumnMappingMode columnMappingMode,
+      Set<String> clusteringColumnPhysicalNames,
+      int currentMaxFieldId,
+      boolean allowNewRequiredFields) {
     switch (columnMappingMode) {
       case ID:
       case NAME:
-        validateSchemaEvolutionById(currentSchema, newSchema, clusteringColumnPhysicalNames);
+        validateSchemaEvolutionById(
+            currentSchema,
+            newSchema,
+            clusteringColumnPhysicalNames,
+            currentMaxFieldId,
+            allowNewRequiredFields);
         return;
       case NONE:
         throw new UnsupportedOperationException(
@@ -449,14 +463,18 @@ private static void validateSchemaEvolution(
    * fields
    */
   private static void validateSchemaEvolutionById(
-      StructType currentSchema, StructType newSchema, Set<String> clusteringColumnPhysicalNames) {
+      StructType currentSchema,
+      StructType newSchema,
+      Set<String> clusteringColumnPhysicalNames,
+      int oldMaxFieldId,
+      boolean allowNewRequiredFields) {
     Map<Integer, StructField> currentFieldsById = fieldsById(currentSchema);
     Map<Integer, StructField> updatedFieldsById = fieldsById(newSchema);
     SchemaChanges schemaChanges = computeSchemaChangesById(currentFieldsById, updatedFieldsById);
     validatePhysicalNameConsistency(schemaChanges.updatedFields());
     // Validates that the updated schema does not contain breaking changes in terms of types and
     // nullability
-    validateUpdatedSchemaCompatibility(schemaChanges);
+    validateUpdatedSchemaCompatibility(schemaChanges, oldMaxFieldId, allowNewRequiredFields);
     validateClusteringColumnsNotDropped(
         schemaChanges.removedFields(), clusteringColumnPhysicalNames);
     // ToDo Potentially validate IcebergCompatV2 nested IDs
@@ -480,12 +498,21 @@ private static void validateClusteringColumnsNotDropped(
    *
    * <p>ToDo: Prevent moving fields outside of their containing struct
    */
-  private static void validateUpdatedSchemaCompatibility(SchemaChanges schemaChanges) {
+  private static void validateUpdatedSchemaCompatibility(
+      SchemaChanges schemaChanges, int oldMaxFieldId, boolean allowNewRequiredFields) {
     for (StructField addedField : schemaChanges.addedFields()) {
-      if (!addedField.isNullable()) {
+      if (!allowNewRequiredFields && !addedField.isNullable()) {
         throw new KernelException(
             String.format("Cannot add non-nullable field %s", addedField.getName()));
       }
+      int colId = getColumnId(addedField);
+      if (colId <= oldMaxFieldId) {
+        throw new IllegalArgumentException(
+            String.format(
+                "Cannot add a new column with a fieldId <= maxFieldId. Found field: %s with"
+                    + "fieldId=%s. Current maxFieldId in the table is: %s",
+                addedField, colId, oldMaxFieldId));
+      }
     }
 
     for (Tuple2<StructField, StructField> updatedFields : schemaChanges.updatedFields()) {

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

@@ -626,6 +626,35 @@ class ColumnMappingSuite extends AnyFunSuite with ColumnMappingSuiteBase {
     }
   }
 
+  test("both id and physical name must be provided if one is provided") {
+    val schemaWithoutPhysicalName = new StructType()
+      .add(
+        new StructField(
+          "col1",
+          StringType.STRING,
+          true,
+          FieldMetadata.builder()
+            .putLong(ColumnMapping.COLUMN_MAPPING_ID_KEY, 0)
+            .build()))
+    val schemaWithoutId = new StructType()
+      .add(
+        new StructField(
+          "col1",
+          StringType.STRING,
+          true,
+          FieldMetadata.builder()
+            .putString(ColumnMapping.COLUMN_MAPPING_PHYSICAL_NAME_KEY, "physical-name-col1")
+            .build()))
+
+    Seq(schemaWithoutId, schemaWithoutPhysicalName).foreach { schema =>
+      val e = intercept[IllegalArgumentException] {
+        updateColumnMappingMetadataIfNeeded(testMetadata(schema).withColumnMappingEnabled(), true)
+      }
+      assert(e.getMessage.contains(
+        "Both columnId and physicalName must be present if one is present"))
+    }
+  }
+
   /**
    * A struct type with all necessary CM info won't cause metadata change by
    * [[updateColumnMappingMetadataIfNeeded]]