diff --git a/kernel/kernel-api/src/main/java/io/delta/kernel/internal/util/SchemaUtils.java b/kernel/kernel-api/src/main/java/io/delta/kernel/internal/util/SchemaUtils.java index 6d916775d53..cb51a4c38ef 100644 --- a/kernel/kernel-api/src/main/java/io/delta/kernel/internal/util/SchemaUtils.java +++ b/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). + * + *

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 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. + * + *

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. * *

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 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( diff --git a/kernel/kernel-api/src/test/scala/io/delta/kernel/internal/util/SchemaUtilsSuite.scala b/kernel/kernel-api/src/test/scala/io/delta/kernel/internal/util/SchemaUtilsSuite.scala index b78a5e97b64..f241ac906de 100644 --- a/kernel/kernel-api/src/test/scala/io/delta/kernel/internal/util/SchemaUtilsSuite.scala +++ b/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 diff --git a/kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/DeltaTableSchemaEvolutionSuite.scala b/kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/DeltaTableSchemaEvolutionSuite.scala index e78a0bba81b..3d8482964ef 100644 --- a/kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/DeltaTableSchemaEvolutionSuite.scala +++ b/kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/DeltaTableSchemaEvolutionSuite.scala @@ -1131,7 +1131,7 @@ trait DeltaTableSchemaEvolutionSuiteBase extends AnyFunSuite with AbstractWriteU } } - test("Adding non-nullable field fails") { + test("Adding non-nullable field inside new nullable struct succeeds") { withTempDirAndEngine { (tablePath, engine) => val table = Table.forPath(engine, tablePath) val initialSchema = new StructType() @@ -1159,11 +1159,13 @@ trait DeltaTableSchemaEvolutionSuiteBase extends AnyFunSuite with AbstractWriteU .add("s", new StringType(utf8Lcase), true, fieldMetadataForColumn(6, "s")) .add("c", IntegerType.INTEGER, true, currentSchema.get("c").getMetadata) - assertSchemaEvolutionFails[KernelException]( - table, - engine, - newSchema, - "Cannot add non-nullable field non_nullable_field") + updateTableMetadata(engine, tablePath, newSchema) + + val resultSchema = table.getLatestSnapshot(engine).getSchema + val bStruct = resultSchema.get("b").getDataType.asInstanceOf[StructType] + assert( + !bStruct.get("non_nullable_field").isNullable, + "non_nullable_field should remain non-nullable") } } @@ -1227,6 +1229,115 @@ trait DeltaTableSchemaEvolutionSuiteBase extends AnyFunSuite with AbstractWriteU } } + test("Adding nullable struct with deeply nested non-nullable field " + + "through map and array succeeds") { + withTempDirAndEngine { (tablePath, engine) => + val table = Table.forPath(engine, tablePath) + val initialSchema = new StructType() + .add("a", StringType.STRING, true) + + createEmptyTable( + engine, + tablePath, + initialSchema, + tableProperties = Map( + TableConfig.COLUMN_MAPPING_MODE.getKey -> "id", + TableConfig.ICEBERG_COMPAT_V2_ENABLED.getKey -> "true")) + + val currentSchema = table.getLatestSnapshot(engine).getSchema + val nestedIdsForM = FieldMetadata.builder() + .putLong("m.key", 6) + .putLong("m.value", 7) + .putLong("m.value.element", 8) + .build() + val mMetadata = FieldMetadata.builder() + .putLong(ColumnMapping.COLUMN_MAPPING_ID_KEY, 3) + .putString(ColumnMapping.COLUMN_MAPPING_PHYSICAL_NAME_KEY, "m") + .putFieldMetadata(ColumnMapping.COLUMN_MAPPING_NESTED_IDS_KEY, nestedIdsForM) + .build() + + val newSchema = new StructType() + .add("a", StringType.STRING, true, currentSchema.get("a").getMetadata) + .add( + "outer", + new StructType() + .add( + "m", + new MapType( + StringType.STRING, + new ArrayType( + new StructType() + .add("f", StringType.STRING, false, fieldMetadataForColumn(5, "f")), + false), + false), + true, + mMetadata), + true, + fieldMetadataForColumn(2, "outer")) + + updateTableMetadata(engine, tablePath, newSchema) + + val resultSchema = table.getLatestSnapshot(engine).getSchema + assertColumnMapping(resultSchema.get("outer"), 2, "outer") + val outerStruct = resultSchema.get("outer").getDataType.asInstanceOf[StructType] + assert(outerStruct.indexOf("m") >= 0, "m should exist inside outer") + val mapType = outerStruct.get("m").getDataType.asInstanceOf[MapType] + val arrayType = mapType.getValueType.asInstanceOf[ArrayType] + val innerStruct = arrayType.getElementType.asInstanceOf[StructType] + assert( + !innerStruct.get("f").isNullable, + "f should remain non-nullable through map -> array -> struct nesting") + } + } + + test("Adding nullable struct with non-nullable inner field " + + "alongside unchanged sibling succeeds") { + withTempDirAndEngine { (tablePath, engine) => + val table = Table.forPath(engine, tablePath) + val initialSchema = new StructType() + .add( + "parent", + new StructType() + .add("sibling", IntegerType.INTEGER, true), + true) + + createEmptyTable( + engine, + tablePath, + initialSchema, + tableProperties = Map( + TableConfig.COLUMN_MAPPING_MODE.getKey -> "id", + TableConfig.ICEBERG_COMPAT_V2_ENABLED.getKey -> "true")) + + val currentSchema = table.getLatestSnapshot(engine).getSchema + val parentMeta = currentSchema.get("parent").getMetadata + val parentStruct = currentSchema.get("parent").getDataType.asInstanceOf[StructType] + val siblingMeta = parentStruct.get("sibling").getMetadata + + val newSchema = new StructType() + .add( + "parent", + new StructType() + .add("sibling", IntegerType.INTEGER, true, siblingMeta) + .add( + "new_struct", + new StructType() + .add("f", StringType.STRING, false, fieldMetadataForColumn(4, "f")), + true, + fieldMetadataForColumn(3, "new_struct")), + true, + parentMeta) + + updateTableMetadata(engine, tablePath, newSchema) + + val resultSchema = table.getLatestSnapshot(engine).getSchema + val resultParent = resultSchema.get("parent").getDataType.asInstanceOf[StructType] + assert(resultParent.indexOf("sibling") >= 0, "sibling should still exist") + val newStruct = resultParent.get("new_struct").getDataType.asInstanceOf[StructType] + assert(!newStruct.get("f").isNullable, "f should remain non-nullable after schema evolution") + } + } + test("Cannot drop a partition column") { withTempDirAndEngine { (tablePath, engine) => val table = Table.forPath(engine, tablePath)