[Kernel] Expose withSchema API to enable schema evolution (delta-io#4361)

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

This PR exposes withSchema API which will enable users to do non
intent-based schema evolution with the documented validation .

## How was this patch tested?

Added the same E2E tests as in
delta-io#4196 except for nested ID
consistency, since that is being reworked as part of
delta-io#4360 and that validation is not
there.

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

Yeah the transaction API exposes a withSchema API which allows users to
pass in their own schema for evolution.

Author: amogh-jahagirdar

kernel/kernel-api/src/main/java/io/delta/kernel/TransactionBuilder.java

@@ -34,11 +34,29 @@
 @Evolving
 public interface TransactionBuilder {
   /**
-   * Set the schema of the table when creating a new table.
+   * Set the schema of the table. If setting the schema on an existing table for a schema evolution,
+   * then column mapping must be enabled. This API will preserve field metadata for fields such as
+   * field IDs and physical names. If field metadata is not specified for a field, it is considered
+   * as a new column and new IDs/physical names will be specified. The possible schema evolutions
+   * supported include column additions, removals, renames, and moves. If a schema evolution is
+   * performed, implementations must perform the following validations:
+   *
+   * <ul>
+   *   <li>No duplicate columns are allowed
+   *   <li>Column names contain only valid characters
+   *   <li>Data types are supported
+   *   <li>No new non-nullable fields are added
+   *   <li>Physical column name consistency is preserved in the new schema
+   *   <li>No type changes
+   *   <li>ToDo: Nested IDs for array/map types are preserved in the new schema
+   *   <li>ToDo: Validate invalid field reorderings
+   * </ul>
    *
    * @param engine {@link Engine} instance to use.
    * @param schema The new schema of the table.
    * @return updated {@link TransactionBuilder} instance.
+   * @throws io.delta.kernel.exceptions.KernelException in case column mapping is not enabled
+   * @throws IllegalArgumentException in case of any validation failure
    */
   TransactionBuilder withSchema(Engine engine, StructType schema);
 

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

@@ -29,6 +29,7 @@
 
 import io.delta.kernel.*;
 import io.delta.kernel.engine.Engine;
+import io.delta.kernel.exceptions.KernelException;
 import io.delta.kernel.exceptions.TableNotFoundException;
 import io.delta.kernel.internal.actions.*;
 import io.delta.kernel.internal.fs.Path;
@@ -162,7 +163,9 @@ public Transaction build(Engine engine) {
       newMetadata = Optional.of(snapshotMetadata.withMergedConfiguration(newProperties));
     }
 
-    // TODO In the future update metadata with new schema if provided
+    if (schema.isPresent() && !isNewTable) {
+      newMetadata = Optional.of(newMetadata.orElse(snapshotMetadata).withNewSchema(schema.get()));
+    }
 
     /* ----- 2: Update the PROTOCOL based on the table properties or schema ----- */
     // This is the only place we update the protocol action; takes care of any dependent features
@@ -269,12 +272,6 @@ private void validateTransactionInputs(Engine engine, SnapshotImpl snapshot, boo
         snapshot.getProtocol(), snapshot.getMetadata(), tablePath);
 
     if (!isNewTable) {
-      if (schema.isPresent()) {
-        throw tableAlreadyExists(
-            tablePath,
-            "Table already exists, but provided a new schema. "
-                + "Schema can only be set on a new table.");
-      }
       if (partitionColumns.isPresent()) {
         throw tableAlreadyExists(
             tablePath,
@@ -319,7 +316,26 @@ private void validateMetadataChange(
     IcebergWriterCompatV1MetadataValidatorAndUpdater.validateIcebergWriterCompatV1Change(
         oldMetadata.getConfiguration(), newMetadata.getConfiguration(), isNewTable);
 
-    // TODO In the future validate any schema change
+    // Validate the conditions for schema evolution and the updated schema if applicable
+    if (schema.isPresent() && !isNewTable) {
+      ColumnMappingMode updatedMappingMode =
+          ColumnMapping.getColumnMappingMode(newMetadata.getConfiguration());
+      ColumnMappingMode currentMappingMode =
+          ColumnMapping.getColumnMappingMode(oldMetadata.getConfiguration());
+      if (currentMappingMode != updatedMappingMode) {
+        throw new KernelException("Cannot update mapping mode and perform schema evolution");
+      }
+
+      if (!isColumnMappingModeEnabled(updatedMappingMode)) {
+        throw new KernelException("Cannot update schema for table when column mapping is disabled");
+      }
+
+      SchemaUtils.validateUpdatedSchema(
+          oldMetadata.getSchema(),
+          newMetadata.getSchema(),
+          oldMetadata.getPartitionColNames(),
+          newMetadata);
+    }
   }
 
   private class InitialSnapshot extends SnapshotImpl {

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

@@ -16,6 +16,7 @@
 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;
 
@@ -131,9 +132,15 @@ public static void validateDataFileStatus(DataFileStatus dataFileStatus) {
               .newMetadata
               .getPartitionColNames()
               .forEach(
-                  partitonCol -> {
+                  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().get(partitonCol).getDataType();
+                        inputContext.newMetadata.getSchema().at(partitionFieldIndex).getDataType();
                     boolean validType =
                         dataType instanceof ByteType
                             || dataType instanceof ShortType

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

@@ -99,12 +99,17 @@ public static void validateSchema(StructType schema, boolean isColumnMappingEnab
    * </ul>
    */
   public static void validateUpdatedSchema(
-      StructType currentSchema, StructType newSchema, Metadata metadata) {
+      StructType currentSchema,
+      StructType newSchema,
+      Set<String> currentPartitionColumns,
+      Metadata newMetadata) {
     checkArgument(
-        isColumnMappingModeEnabled(ColumnMapping.getColumnMappingMode(metadata.getConfiguration())),
+        isColumnMappingModeEnabled(
+            ColumnMapping.getColumnMappingMode(newMetadata.getConfiguration())),
         "Cannot validate updated schema when column mapping is disabled");
     validateSchema(newSchema, true /*columnMappingEnabled*/);
-    validateSchemaEvolution(currentSchema, newSchema, metadata);
+    validatePartitionColumns(newSchema, new ArrayList<>(currentPartitionColumns));
+    validateSchemaEvolution(currentSchema, newSchema, newMetadata);
   }
 
   /**

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

@@ -490,6 +490,7 @@ class SchemaUtilsSuite extends AnyFunSuite {
       validateUpdatedSchema(
         current,
         updated,
+        Set.empty.asJava,
         metadata(current, properties = tblProperties))
     }
 
@@ -682,7 +683,7 @@ class SchemaUtilsSuite extends AnyFunSuite {
   test("validateUpdatedSchema fails with schema with duplicate column ID") {
     forAll(updatedSchemaHasDuplicateColumnId) { (schemaBefore, schemaAfter) =>
       val e = intercept[IllegalArgumentException] {
-        validateUpdatedSchema(schemaBefore, schemaAfter, metadata(schemaBefore))
+        validateUpdatedSchema(schemaBefore, schemaAfter, Set.empty.asJava, metadata(schemaBefore))
       }
 
       assert(e.getMessage.matches("Field duplicate_id with id .* already exists"))
@@ -731,7 +732,7 @@ class SchemaUtilsSuite extends AnyFunSuite {
 
   test("validateUpdatedSchema succeeds with valid ID and physical name") {
     forAll(validUpdatedSchemas) { (schemaBefore, schemaAfter) =>
-      validateUpdatedSchema(schemaBefore, schemaAfter, metadata(schemaBefore))
+      validateUpdatedSchema(schemaBefore, schemaAfter, Set.empty.asJava, metadata(schemaBefore))
     }
   }
 
@@ -917,7 +918,7 @@ class SchemaUtilsSuite extends AnyFunSuite {
 
   test("validateUpdatedSchema succeeds when adding field") {
     forAll(validateAddedFields) { (schemaBefore, schemaAfter) =>
-      validateUpdatedSchema(schemaBefore, schemaAfter, metadata(schemaBefore))
+      validateUpdatedSchema(schemaBefore, schemaAfter, Set.empty.asJava, metadata(schemaBefore))
     }
   }
 
@@ -951,7 +952,7 @@ class SchemaUtilsSuite extends AnyFunSuite {
 
   test("validateUpdatedSchema succeeds when updating field metadata") {
     forAll(validateMetadataChange) { (schemaBefore, schemaAfter) =>
-      validateUpdatedSchema(schemaBefore, schemaAfter, metadata(schemaBefore))
+      validateUpdatedSchema(schemaBefore, schemaAfter, Set.empty.asJava, metadata(schemaBefore))
     }
   }
 
@@ -1001,6 +1002,7 @@ class SchemaUtilsSuite extends AnyFunSuite {
         validateUpdatedSchema(
           schemaBefore,
           schemaAfter,
+          Set.empty.asJava,
           metadata(schemaBefore, tableProperties))
       }