[Kernel] Fix schema validation for new nullable structs with non-nullable fields

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

@@ -424,7 +424,13 @@ static SchemaChanges computeSchemaChangesById(StructType currentSchema, StructTy
       }
 
       if (addedFieldIds.contains(id.getId())) {
-        // Skip early if this is a descendant of an added field.
+        // Skip early if this element's field ID was already recorded as added.
+        // This works for array/map synthetic children (key, value, element) because
+        // getSchemaElementId resolves their column ID to the nearest struct field
+        // ancestor's ID — so they share the parent's field ID in addedFieldIds.
+        // It does NOT work for struct children, which have their own unique column
+        // mapping IDs. Struct children of newly added fields are handled separately
+        // in validateUpdatedSchemaCompatibility via the descendantFieldIds set.
         continue;
       }
 
@@ -536,6 +542,34 @@ private static SchemaElementId getSchemaElementId(SchemaIterable.SchemaElement e
     return new SchemaElementId(element.getPathFromNearestStructFieldAncestor(""), columnId);
   }
 
+  /**
+   * Recursively collects the column mapping field IDs of all StructField descendants reachable
+   * through the given DataType. Traverses into StructType (collects each field's ID and recurses),
+   * ArrayType (recurses into element type), and MapType (recurses into key and value types).
+   *
+   * <p>This is used to build a set of field IDs that are descendants of newly added fields. During
+   * schema evolution validation, these descendants should not be independently checked for
+   * nullability because they are part of an entirely new container — no existing data could violate
+   * their constraints.
+   *
+   * @param type the DataType to walk
+   * @param fieldIds the set to collect descendant field IDs into
+   */
+  private static void collectDescendantFieldIds(DataType type, Set<Integer> fieldIds) {
+    if (type instanceof StructType) {
+      for (StructField field : ((StructType) type).fields()) {
+        fieldIds.add(getColumnId(field));
+        collectDescendantFieldIds(field.getDataType(), fieldIds);
+      }
+    } else if (type instanceof ArrayType) {
+      collectDescendantFieldIds(((ArrayType) type).getElementType(), fieldIds);
+    } else if (type instanceof MapType) {
+      collectDescendantFieldIds(((MapType) type).getKeyType(), fieldIds);
+      collectDescendantFieldIds(((MapType) type).getValueType(), fieldIds);
+    }
+    // Primitive types: no descendants, nothing to collect.
+  }
+
   private static SchemaElementId getSchemaElementId(
       SchemaIterable.ParentStructFieldInfo parentInfo) {
     int columnId = getColumnId(parentInfo.getParentField());
@@ -628,8 +662,15 @@ private static void validateClusteringColumnsNotDropped(
   }
 
   /**
-   * Verifies that no non-nullable fields are added, no existing field nullability is tightened and
-   * no invalid type changes are performed
+   * Verifies that no non-nullable fields are added (unless they are descendants of other newly
+   * added fields), no existing field nullability is tightened, and no invalid type changes are
+   * performed.
+   *
+   * <p>When a new nullable container (struct, array, or map) is added, its internal fields may be
+   * non-nullable. This is valid because no existing data contains the container, so the inner field
+   * nullability only constrains future writes. To handle this, we build a set of field IDs that are
+   * descendants of other added fields (the "descendantFieldIds" set) and skip the nullability check
+   * for those fields. The fieldId constraint is still enforced for all fields.
    *
    * <p>ToDo: Prevent moving fields outside of their containing struct
    */
@@ -639,12 +680,25 @@ private static void validateUpdatedSchemaCompatibility(
       boolean allowNewRequiredFields,
       boolean icebergWriterCompatV1Enabled,
       boolean typeWideningEnabled) {
+    // Build a set of field IDs whose ancestor is also a newly added field.
+    // These are descendants of entirely new containers — their nullability
+    // should not be independently validated because no existing data has them.
+    Set<Integer> descendantFieldIds = new HashSet<>();
     for (StructField addedField : schemaChanges.addedFields()) {
-      if (!allowNewRequiredFields && !addedField.isNullable()) {
+      collectDescendantFieldIds(addedField.getDataType(), descendantFieldIds);
+    }
+
+    for (StructField addedField : schemaChanges.addedFields()) {
+      int colId = getColumnId(addedField);
+      // Skip the nullability check for fields that are descendants of other added fields.
+      // Their nullability is a constraint on future data only (writer's responsibility),
+      // not a compatibility issue with existing data.
+      if (!allowNewRequiredFields
+          && !addedField.isNullable()
+          && !descendantFieldIds.contains(colId)) {
         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(

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

@@ -1582,6 +1582,299 @@ class SchemaUtilsSuite extends AnyFunSuite {
     }
   }
 
+  test("validateUpdatedSchema succeeds when adding nullable struct with " +
+    "non-nullable inner field") {
+    val schemaBefore = newSchema(
+      (1, new StructField("id", IntegerType.INTEGER, true)))
+
+    val schemaAfter = newSchema(
+      (1, new StructField("id", IntegerType.INTEGER, true)),
+      (
+        10,
+        new StructField(
+          "new_struct",
+          new StructType()
+            .add("inner", StringType.STRING, false, fieldMetadata(11, "inner")),
+          true)))
+
+    val schemaChanges = computeSchemaChangesById(schemaBefore, schemaAfter)
+    assert(
+      schemaChanges.addedFields().size() == 2,
+      s"Expected 2 added fields, got ${schemaChanges.addedFields().size()}")
+    assert(schemaChanges.removedFields().isEmpty)
+
+    validateUpdatedSchemaAndGetUpdatedSchema(
+      metadata(schemaBefore),
+      metadata(schemaAfter),
+      dummyProtocol,
+      emptySet(),
+      false /* allowNewRequiredFields */ )
+  }
+
+  test("validateUpdatedSchema succeeds when adding nullable struct with " +
+    "depth > 2 non-nullable chain") {
+    val schemaBefore = newSchema(
+      (1, new StructField("id", IntegerType.INTEGER, true)))
+
+    val schemaAfter = newSchema(
+      (1, new StructField("id", IntegerType.INTEGER, true)),
+      (
+        10,
+        new StructField(
+          "top",
+          new StructType()
+            .add(
+              "mid",
+              new StructType()
+                .add("leaf", IntegerType.INTEGER, false, fieldMetadata(12, "leaf")),
+              false,
+              fieldMetadata(11, "mid")),
+          true)))
+
+    val schemaChanges = computeSchemaChangesById(schemaBefore, schemaAfter)
+    assert(schemaChanges.addedFields().size() == 3, "Expected 3 added fields (leaf, mid, top)")
+
+    validateUpdatedSchemaAndGetUpdatedSchema(
+      metadata(schemaBefore),
+      metadata(schemaAfter),
+      dummyProtocol,
+      emptySet(),
+      false /* allowNewRequiredFields */ )
+  }
+
+  test("validateUpdatedSchema succeeds when adding nullable array with " +
+    "non-nullable struct element fields") {
+    val schemaBefore = newSchema(
+      (1, new StructField("id", IntegerType.INTEGER, true)))
+
+    val schemaAfter = newSchema(
+      (1, new StructField("id", IntegerType.INTEGER, true)),
+      (
+        10,
+        new StructField(
+          "arr",
+          new ArrayType(
+            new StructType()
+              .add("f", StringType.STRING, false, fieldMetadata(11, "f")),
+            false),
+          true)))
+
+    val schemaChanges = computeSchemaChangesById(schemaBefore, schemaAfter)
+    assert(schemaChanges.addedFields().size() == 2)
+
+    validateUpdatedSchemaAndGetUpdatedSchema(
+      metadata(schemaBefore),
+      metadata(schemaAfter),
+      dummyProtocol,
+      emptySet(),
+      false /* allowNewRequiredFields */ )
+  }
+
+  test("validateUpdatedSchema succeeds when adding nullable map with " +
+    "non-nullable struct value fields") {
+    val schemaBefore = newSchema(
+      (1, new StructField("id", IntegerType.INTEGER, true)))
+
+    val schemaAfter = newSchema(
+      (1, new StructField("id", IntegerType.INTEGER, true)),
+      (
+        10,
+        new StructField(
+          "m",
+          new MapType(
+            StringType.STRING,
+            new StructType()
+              .add("f", StringType.STRING, false, fieldMetadata(11, "f")),
+            false),
+          true)))
+
+    val schemaChanges = computeSchemaChangesById(schemaBefore, schemaAfter)
+    assert(
+      schemaChanges.addedFields().size() == 2,
+      "Expected 2 added fields (m via map key, f via self)")
+
+    validateUpdatedSchemaAndGetUpdatedSchema(
+      metadata(schemaBefore),
+      metadata(schemaAfter),
+      dummyProtocol,
+      emptySet(),
+      false /* allowNewRequiredFields */ )
+  }
+
+  test("validateUpdatedSchema succeeds when adding nullable struct with " +
+    "deeply nested non-nullable field through map and array") {
+    val schemaBefore = newSchema(
+      (1, new StructField("id", IntegerType.INTEGER, true)))
+
+    val schemaAfter = newSchema(
+      (1, new StructField("id", IntegerType.INTEGER, true)),
+      (
+        10,
+        new StructField(
+          "outer",
+          new StructType()
+            .add(
+              "m",
+              new MapType(
+                StringType.STRING,
+                new ArrayType(
+                  new StructType()
+                    .add("f", StringType.STRING, false, fieldMetadata(12, "f")),
+                  false),
+                false),
+              true,
+              fieldMetadata(11, "m")),
+          true)))
+
+    val schemaChanges = computeSchemaChangesById(schemaBefore, schemaAfter)
+    assert(
+      schemaChanges.addedFields().size() == 3,
+      s"Expected 3 added fields (m, f, outer), got ${schemaChanges.addedFields().size()}")
+    val addedFieldNames = (0 until schemaChanges.addedFields().size())
+      .map(i => schemaChanges.addedFields().get(i).getName).toSet
+    assert(
+      addedFieldNames == Set("m", "f", "outer"),
+      s"Expected {m, f, outer} in addedFields, got: $addedFieldNames")
+
+    validateUpdatedSchemaAndGetUpdatedSchema(
+      metadata(schemaBefore),
+      metadata(schemaAfter),
+      dummyProtocol,
+      emptySet(),
+      false /* allowNewRequiredFields */ )
+  }
+
+  test("validateUpdatedSchema succeeds when adding nullable struct with " +
+    "non-nullable inner field inside existing struct") {
+    val schemaBefore = newSchema(
+      (
+        1,
+        new StructField(
+          "parent",
+          new StructType()
+            .add("existing", IntegerType.INTEGER, true, fieldMetadata(2, "existing")),
+          true)))
+
+    val schemaAfter = newSchema(
+      (
+        1,
+        new StructField(
+          "parent",
+          new StructType()
+            .add("existing", IntegerType.INTEGER, true, fieldMetadata(2, "existing"))
+            .add(
+              "new_child",
+              new StructType()
+                .add("f", StringType.STRING, false, fieldMetadata(11, "f")),
+              true,
+              fieldMetadata(10, "new_child")),
+          true)))
+
+    val schemaChanges = computeSchemaChangesById(schemaBefore, schemaAfter)
+    assert(schemaChanges.updatedFields().size() >= 1, "Expected parent in updatedFields")
+    val addedFieldNames = (0 until schemaChanges.addedFields().size())
+      .map(i => schemaChanges.addedFields().get(i).getName).toSet
+    assert(
+      addedFieldNames.contains("f") && addedFieldNames.contains("new_child"),
+      s"Expected f and new_child in addedFields, got: $addedFieldNames")
+    assert(
+      !addedFieldNames.contains("parent"),
+      "parent should be in updatedFields, not addedFields")
+
+    validateUpdatedSchemaAndGetUpdatedSchema(
+      metadata(schemaBefore),
+      metadata(schemaAfter),
+      dummyProtocol,
+      emptySet(),
+      false /* allowNewRequiredFields */ )
+  }
+
+  test("validateUpdatedSchema fails when adding non-nullable struct " +
+    "with non-nullable inner field") {
+    val schemaBefore = newSchema(
+      (1, new StructField("id", IntegerType.INTEGER, true)))
+
+    val schemaAfter = newSchema(
+      (1, new StructField("id", IntegerType.INTEGER, true)),
+      (
+        10,
+        new StructField(
+          "new_struct",
+          new StructType()
+            .add("inner", StringType.STRING, false, fieldMetadata(11, "inner")),
+          false /* NOT NULL struct — this must fail */ )))
+
+    val e = intercept[KernelException] {
+      validateUpdatedSchemaAndGetUpdatedSchema(
+        metadata(schemaBefore),
+        metadata(schemaAfter),
+        dummyProtocol,
+        emptySet(),
+        false /* allowNewRequiredFields */ )
+    }
+    assert(
+      e.getMessage.contains("Cannot add non-nullable field"),
+      s"Expected non-nullable rejection, got: ${e.getMessage}")
+  }
+
+  test("validateUpdatedSchema fails on the non-nullable parent struct " +
+    "not its skipSet child") {
+    val schemaBefore = newSchema(
+      (1, new StructField("id", IntegerType.INTEGER, true)))
+
+    val schemaAfter = newSchema(
+      (1, new StructField("id", IntegerType.INTEGER, true)),
+      (
+        10,
+        new StructField(
+          "new_struct",
+          new StructType()
+            .add("inner", StringType.STRING, false, fieldMetadata(11, "inner")),
+          false /* NOT NULL struct */ )))
+
+    val e = intercept[KernelException] {
+      validateUpdatedSchemaAndGetUpdatedSchema(
+        metadata(schemaBefore),
+        metadata(schemaAfter),
+        dummyProtocol,
+        emptySet(),
+        false /* allowNewRequiredFields */ )
+    }
+    assert(
+      e.getMessage.contains("new_struct"),
+      s"Expected failure on 'new_struct', got: ${e.getMessage}")
+  }
+
+  test("validateUpdatedSchema fails when new nullable struct has inner field " +
+    "with fieldId <= maxFieldId") {
+    val schemaBefore = newSchema(
+      (1, new StructField("id", IntegerType.INTEGER, true)),
+      (5, new StructField("data", StringType.STRING, true)))
+
+    val schemaAfter = newSchema(
+      (1, new StructField("id", IntegerType.INTEGER, true)),
+      (5, new StructField("data", StringType.STRING, true)),
+      (
+        10,
+        new StructField(
+          "new_struct",
+          new StructType()
+            .add("inner", StringType.STRING, true, fieldMetadata(3, "inner")),
+          true)))
+
+    val e = intercept[IllegalArgumentException] {
+      validateUpdatedSchemaAndGetUpdatedSchema(
+        metadata(schemaBefore),
+        metadata(schemaAfter),
+        dummyProtocol,
+        emptySet(),
+        false /* allowNewRequiredFields */ )
+    }
+    assert(
+      e.getMessage.contains("Cannot add a new column with a fieldId <= maxFieldId"),
+      s"Expected fieldId rejection, got: ${e.getMessage}")
+  }
+
   private val validateMetadataChange = Table(
     ("schemaBefore", "schemaWithAddedField"),
     // Adding column comment to id column