From d7138043a67a9236297db3564cfd3d9a849ee62b Mon Sep 17 00:00:00 2001 From: Anish Mahto Date: Wed, 2 Sep 2026 03:09:02 +0000 Subject: [PATCH 1/6] [SQL] Make diffSchemas recurse into nested structs diffSchemas previously compared top-level fields only. When a struct column gained or lost nested fields, the entire column type was emitted as an UpdateColumnType -- a change many DSv2 connectors reject because the semantics of replacing a whole struct type are ambiguous. This patch makes diffSchemas recurse into StructType columns: nested field additions become addColumn with multi-part field paths, deletions become deleteColumn, and type/nullability changes become the corresponding nested updateColumnType/updateColumnNullability. This is the DSv2-idiomatic way to evolve struct columns and is supported by Delta, Iceberg, and other connectors that handle nested schemas. --- .../pipelines/util/SchemaInferenceUtils.scala | 178 +++++--- .../util/SchemaInferenceUtilsSuite.scala | 420 +++++++++++++++++- 2 files changed, 527 insertions(+), 71 deletions(-) diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtils.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtils.scala index fd60684150742..4b1f788e66121 100644 --- a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtils.scala +++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtils.scala @@ -34,7 +34,7 @@ import org.apache.spark.sql.pipelines.graph.{ GraphErrors, ResolvedFlow } -import org.apache.spark.sql.types.{StructField, StructType} +import org.apache.spark.sql.types.{ArrayType, DataType, MapType, StructField, StructType} object SchemaInferenceUtils { @@ -207,19 +207,19 @@ object SchemaInferenceUtils { } /** - * Determines the column changes needed to transform the current schema into the target schema. - * - * This function compares the current schema with the target schema and produces a sequence of - * TableChange objects representing: - * 1. New columns that need to be added - * 2. Existing columns that need type updates + * Produces the [[TableChange]] sequence needed to transform `currentSchema` into + * `targetSchema`: additions, type updates, deletions, nullability and comment changes. + * Recurses into structs, arrays, and maps so changes are emitted at the leaf level. + * Similar to [[org.apache.spark.sql.catalyst.analysis.ResolveSchemaEvolution]], but + * produces a full bidirectional sync (deletes, nullability, and comment changes) rather + * than additive-only evolution. * * Column identity is keyed on the exact field name, not on a case-normalized one. On the * incremental streaming-table path, `targetSchema` is the merge of the current and desired * schemas, and [[SchemaMergingUtils.mergeSchemas]] has already folded an incoming - * case-only-differing field onto the persisted one. On the non-merging paths (materialized views - * and any full refresh), `targetSchema` is the run's declared schema as-is, so exact-name keying - * keeps a case-only rename visible as an explicit drop-then-add. + * case-only-differing field onto the persisted one. On the non-merging paths (materialized + * views and any full refresh), `targetSchema` is the run's declared schema as-is, so + * exact-name keying keeps a case-only rename visible as an explicit drop-then-add. * Exact keying also avoids silently collapsing two genuinely distinct declared columns that * differ only in case (`value` and `Value`) into an arbitrary one of the two. * @@ -227,59 +227,131 @@ object SchemaInferenceUtils { * @param targetSchema The target schema that we want the table to have * @return A sequence of TableChange objects representing the necessary changes */ - def diffSchemas(currentSchema: StructType, targetSchema: StructType): Seq[TableChange] = { - val changes = scala.collection.mutable.ArrayBuffer.empty[TableChange] + def diffSchemas(currentSchema: StructType, targetSchema: StructType): Seq[TableChange] = + diffStructs( + currentStruct = currentSchema, + targetStruct = targetSchema, + // Root call: path is empty because current and target are the top-level schemas. + pathToStruct = Seq.empty + ) - // Helper function to get a map of field name to field - def getFieldMap(schema: StructType): Map[String, StructField] = { - schema.fields.map(field => field.name -> field).toMap - } + /** + * Diffs two structs field-by-field, matching fields by exact name. + * + * @param currentStruct The struct as it exists in the current schema. + * @param targetStruct The struct as it should look in the target schema. + * @param pathToStruct Path segments from the top-level schema to this + * struct, if this is a nested struct. Empty for the + * root call. + */ + private def diffStructs( + currentStruct: StructType, + targetStruct: StructType, + pathToStruct: Seq[String]): Seq[TableChange] = { + val topLevelFieldsInCurrent = currentStruct.fields.map(field => field.name -> field).toMap + val topLevelFieldsInTarget = targetStruct.fields.map(field => field.name -> field).toMap - val currentFields = getFieldMap(currentSchema) - val targetFields = getFieldMap(targetSchema) + // Fields present in target but not in current are columns that need to be added. + val columnsAdded = topLevelFieldsInTarget.values.toSeq + .filterNot(fieldInTarget => + topLevelFieldsInCurrent.contains(fieldInTarget.name) + ) + .map { fieldInTarget => + TableChange.addColumn( + fieldNames = (pathToStruct :+ fieldInTarget.name).toArray, + dataType = fieldInTarget.dataType, + isNullable = fieldInTarget.nullable, + comment = fieldInTarget.getComment().orNull + ) + } - // Find columns to add (in target but not in current) - val columnsToAdd = targetFields.keySet.diff(currentFields.keySet) - columnsToAdd.foreach { columnName => - val field = targetFields(columnName) - changes += TableChange.addColumn( - Array(columnName), - field.dataType, - field.nullable, - field.getComment().orNull + // Fields present in current but not in target are columns that need to be removed. + val columnsDeleted = topLevelFieldsInCurrent.values.toSeq + .filterNot(fieldInCurrent => + topLevelFieldsInTarget.contains(fieldInCurrent.name) + ) + .map(fieldInCurrent => + TableChange + .deleteColumn( + fieldNames = (pathToStruct :+ fieldInCurrent.name).toArray, + ifExists = false + ) ) - } - // Find columns to delete (in current but not in target) - val columnsToDelete = currentFields.keySet.diff(targetFields.keySet) - columnsToDelete.foreach { columnName => - changes += TableChange.deleteColumn(Array(columnName), false) + // Fields in both current and target but vary in metadata or nested sub-fields represent + // columns that need to be updated. + val columnsUpdated = topLevelFieldsInCurrent.values.toSeq.flatMap { + fieldInCurrent => + topLevelFieldsInTarget.get(fieldInCurrent.name).toSeq.flatMap { + fieldInTarget => + diffField( + currentField = fieldInCurrent, + targetField = fieldInTarget, + pathToField = pathToStruct :+ fieldInCurrent.name + ) + } } - // Find columns with type changes (in both but with different types) - val commonColumns = currentFields.keySet.intersect(targetFields.keySet) - commonColumns.foreach { columnName => - val currentField = currentFields(columnName) - val targetField = targetFields(columnName) + columnsAdded ++ columnsDeleted ++ columnsUpdated + } - // If data types are different, add a type update change - if (currentField.dataType != targetField.dataType) { - changes += TableChange.updateColumnType(Array(columnName), targetField.dataType) - } + /** Diffs the type, nullability, and comment of one field present in both schemas. */ + private def diffField( + currentField: StructField, + targetField: StructField, + pathToField: Seq[String]): Seq[TableChange] = { + diffDataTypes(currentField.dataType, targetField.dataType, pathToField) ++ + diffNullability(currentField.nullable, targetField.nullable, pathToField) ++ + diffComment(currentField.getComment(), targetField.getComment(), pathToField) + } - // If nullability is different, add a nullability update change - if (currentField.nullable != targetField.nullable) { - changes += TableChange.updateColumnNullability(Array(columnName), targetField.nullable) - } + private def diffNullability( + currentNullable: Boolean, + targetNullable: Boolean, + pathToField: Seq[String] + ): Option[TableChange] = { + Option.when(currentNullable != targetNullable)( + TableChange.updateColumnNullability(pathToField.toArray, targetNullable) + ) + } - // If comments are different, add a comment update change - val currentComment = currentField.getComment().orNull - val targetComment = targetField.getComment().orNull - if (currentComment != targetComment) { - changes += TableChange.updateColumnComment(Array(columnName), targetComment) - } - } + private def diffComment( + currentComment: Option[String], + targetComment: Option[String], + pathToField: Seq[String] + ): Option[TableChange] = { + Option.when(currentComment != targetComment)( + TableChange.updateColumnComment(pathToField.toArray, targetComment.orNull) + ) + } + + /** Diffs two data types at `path`, descending through matching complex types. */ + private def diffDataTypes( + currentType: DataType, + targetType: DataType, + pathToField: Seq[String] + ): Seq[TableChange] = (currentType, targetType) match { + case (currentStruct: StructType, targetStruct: StructType) => + diffStructs(currentStruct, targetStruct, pathToField) + + case (currentArray: ArrayType, targetArray: ArrayType) => + val elementPath = pathToField :+ "element" + + diffDataTypes(currentArray.elementType, targetArray.elementType, elementPath) ++ + diffNullability(currentArray.containsNull, targetArray.containsNull, elementPath) + + case (currentMap: MapType, targetMap: MapType) => + val valuePath = pathToField :+ "value" + val keyPath = pathToField :+ "key" + + diffDataTypes(currentMap.keyType, targetMap.keyType, keyPath) ++ + diffDataTypes(currentMap.valueType, targetMap.valueType, valuePath) ++ + diffNullability(currentMap.valueContainsNull, targetMap.valueContainsNull, valuePath) + + case _ if currentType == targetType => + Seq.empty - changes.toSeq + case _ => + Seq(TableChange.updateColumnType(pathToField.toArray, targetType)) } } diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtilsSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtilsSuite.scala index 92e1c700b99ba..c696058f74ce8 100644 --- a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtilsSuite.scala +++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtilsSuite.scala @@ -215,6 +215,27 @@ class SchemaInferenceUtilsSuite extends QueryTest with SharedSparkSession { assert(descriptionCommentChange.newComment() === "Product description") } + test("determineColumnChanges - tightening nullability emits the change") { + // diffSchemas faithfully emits an UpdateColumnNullability when a column goes from + // nullable to non-nullable. Whether the catalog *accepts* this is a separate concern: + // Delta rejects tightening nullability on existing columns because existing data may + // contain nulls. The diff layer's job is to report the change; enforcement is downstream. + val currentSchema = new StructType() + .add("id", IntegerType) + .add("value", StringType, nullable = true) + val targetSchema = new StructType() + .add("id", IntegerType) + .add("value", StringType, nullable = false) + + val changes = + SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) + assert(changes.length === 1) + val nc = changes.head + .asInstanceOf[TableChange.UpdateColumnNullability] + assert(nc.fieldNames() === Array("value")) + assert(nc.nullable() === false) + } + test("determineColumnChanges - complex changes") { val currentSchema = new StructType() .add("id", IntegerType, nullable = false) @@ -395,15 +416,15 @@ class SchemaInferenceUtilsSuite extends QueryTest with SharedSparkSession { assert(merged === new StructType().add("s", expectedStruct)) // Unlike the case-insensitive test above (where the merge is a no-op and no changes are - // derived), evolution here must rewrite the top-level `s` column. `diffSchemas` compares nested - // types wholesale, so the growth of a nested field surfaces as a single UpdateColumnType on `s` - // carrying the full new struct -- not as an add of `s.Value`. + // derived), evolution here must grow the `s` struct. The growth surfaces as an add of the + // nested leaf `s.Value`, which is the portable shape; retyping `s` with the whole new struct + // would be rejected by `CheckAnalysis`, which fails ALTER COLUMN ... TYPE on a struct. val changes = SchemaInferenceUtils.diffSchemas(currentSchema, merged) assert(changes.length === 1) - val typeChange = changes.collect { case tc: TableChange.UpdateColumnType => tc } - assert(typeChange.length === 1) - assert(typeChange.head.fieldNames() === Array("s")) - assert(typeChange.head.newDataType() === expectedStruct) + val addChange = changes.collect { case ac: TableChange.AddColumn => ac } + assert(addChange.length === 1) + assert(addChange.head.fieldNames() === Array("s", "Value")) + assert(addChange.head.dataType() === StringType) } test("mergeSchemas - a nested case-only field whose type also changes fails to merge, and " + @@ -438,22 +459,385 @@ class SchemaInferenceUtilsSuite extends QueryTest with SharedSparkSession { } // Diffing the two schemas directly (rather than diffing against their merge, which fails - // above) reports a TYPE change on the enclosing `s` column -- not a field-name mismatch, i.e. - // not an add of `s.Value` plus a delete of `s.value`. `diffSchemas` keys column identity only - // at the top level and compares nested types wholesale, so the case difference inside the - // struct never surfaces as an add/delete pair. + // above) reports a drop-then-add of the nested leaf, since `diffSchemas` keys column identity + // on the exact name at every level. This is the nested analog of the top-level rule pinned by + // "a case-only difference is a drop-then-add, not a match", and it is only reachable on the + // non-merging paths, where the declared schema is used as-is. { val changes = SchemaInferenceUtils.diffSchemas(currentSchema, dataSchema) - assert(changes.length === 1, s"changes=$changes") - val typeChange = changes.collect { case tc: TableChange.UpdateColumnType => tc } - assert(typeChange.length === 1, s"changes=$changes") - assert(typeChange.head.fieldNames() === Array("s")) - assert(typeChange.head.newDataType() === new StructType().add("Value", LongType)) - assert(!changes.exists(_.isInstanceOf[TableChange.AddColumn])) - assert(!changes.exists(_.isInstanceOf[TableChange.DeleteColumn])) + assert(changes.length === 2, s"changes=$changes") + val added = changes.collect { case ac: TableChange.AddColumn => ac } + assert(added.length === 1, s"changes=$changes") + assert(added.head.fieldNames() === Array("s", "Value")) + assert(added.head.dataType() === LongType) + val deleted = changes.collect { case dc: TableChange.DeleteColumn => dc } + assert(deleted.length === 1, s"changes=$changes") + assert(deleted.head.fieldNames() === Array("s", "value")) + assert(!changes.exists(_.isInstanceOf[TableChange.UpdateColumnType])) } } + // ============================================================================================= + // Nested diffing + // + // `diffSchemas` recurses into structs, array elements, and map keys/values so that a change + // inside a complex type is emitted at the leaf that actually changed. The alternative -- an + // UpdateColumnType on the enclosing column carrying the whole new complex type -- is rejected by + // `CheckAnalysis`, which fails ALTER COLUMN ... TYPE when the new type is a struct, map, or + // array. Pipelines only ever got away with it by calling `catalog.alterTable` directly and + // testing against an in-memory catalog that applies it regardless. + // ============================================================================================= + + /** Added columns as path -> (type, nullable, comment). */ + private def addsOf( + changes: Seq[TableChange]): Map[Seq[String], (DataType, Boolean, String)] = + changes.collect { + case ac: TableChange.AddColumn => + ac.fieldNames().toSeq -> ((ac.dataType(), ac.isNullable(), ac.comment())) + }.toMap + + private def deletesOf(changes: Seq[TableChange]): Set[Seq[String]] = + changes.collect { + case dc: TableChange.DeleteColumn => dc.fieldNames().toSeq + }.toSet + + private def typeUpdatesOf( + changes: Seq[TableChange]): Map[Seq[String], DataType] = + changes.collect { + case tc: TableChange.UpdateColumnType => + tc.fieldNames().toSeq -> tc.newDataType() + }.toMap + + private def nullabilityUpdatesOf( + changes: Seq[TableChange]): Map[Seq[String], Boolean] = + changes.collect { + case nc: TableChange.UpdateColumnNullability => + nc.fieldNames().toSeq -> nc.nullable() + }.toMap + + private def commentUpdatesOf( + changes: Seq[TableChange]): Map[Seq[String], String] = + changes.collect { + case cc: TableChange.UpdateColumnComment => + cc.fieldNames().toSeq -> cc.newComment() + }.toMap + + test("diffSchemas - a leaf added to a struct is a nested add, " + + "not a retype of the parent") { + val currentSchema = new StructType() + .add("id", IntegerType) + .add( + "point", + new StructType().add("x", DoubleType).add("y", DoubleType)) + val targetSchema = new StructType() + .add("id", IntegerType) + .add( + "point", + new StructType() + .add("x", DoubleType) + .add("y", DoubleType) + .add("z", DoubleType)) + + val changes = + SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) + assert(changes.length === 1, s"changes=$changes") + assert( + addsOf(changes) === + Map(Seq("point", "z") -> ((DoubleType, true, null)))) + } + + test("diffSchemas - a leaf added to a deeply nested struct " + + "carries the full path") { + val inner = new StructType().add("c", IntegerType) + val currentSchema = + new StructType().add("a", new StructType().add("b", inner)) + val targetSchema = new StructType() + .add("a", new StructType().add("b", inner.add("d", StringType))) + + val changes = + SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) + assert(addsOf(changes).keySet === Set(Seq("a", "b", "d"))) + } + + test("diffSchemas - an added nested leaf keeps its declared " + + "nullability and comment") { + val currentSchema = new StructType() + .add("s", new StructType().add("a", IntegerType)) + val targetSchema = new StructType().add( + "s", + new StructType() + .add("a", IntegerType) + .add("b", StringType, nullable = false, "a comment")) + + val changes = + SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) + assert( + addsOf(changes) === + Map(Seq("s", "b") -> ((StringType, false, "a comment")))) + } + + test("diffSchemas - a leaf removed from a struct is a nested delete") { + val currentSchema = new StructType() + .add( + "s", + new StructType().add("a", IntegerType).add("b", StringType)) + val targetSchema = new StructType() + .add("s", new StructType().add("a", IntegerType)) + + val changes = + SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) + assert(changes.length === 1, s"changes=$changes") + assert(deletesOf(changes) === Set(Seq("s", "b"))) + } + + test("diffSchemas - a leaf type change inside a struct " + + "is a nested type update") { + val currentSchema = new StructType() + .add("s", new StructType().add("a", IntegerType)) + val targetSchema = new StructType() + .add("s", new StructType().add("a", LongType)) + + val changes = + SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) + assert(changes.length === 1, s"changes=$changes") + assert(typeUpdatesOf(changes) === Map(Seq("s", "a") -> LongType)) + } + + test("diffSchemas - nested leaf nullability and comment changes " + + "are emitted at the leaf") { + val currentSchema = new StructType() + .add( + "s", + new StructType().add("a", IntegerType, nullable = true, "old")) + val targetSchema = new StructType() + .add( + "s", + new StructType().add("a", IntegerType, nullable = false, "new")) + + val changes = + SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) + assert(changes.length === 2, s"changes=$changes") + assert( + nullabilityUpdatesOf(changes) === Map(Seq("s", "a") -> false)) + assert(commentUpdatesOf(changes) === Map(Seq("s", "a") -> "new")) + } + + test("diffSchemas - a field added to a struct inside an array " + + "uses the element path") { + val currentSchema = new StructType() + .add( + "points", + ArrayType(new StructType().add("x", DoubleType))) + val targetSchema = new StructType().add( + "points", + ArrayType( + new StructType() + .add("x", DoubleType) + .add("y", DoubleType))) + + val changes = + SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) + assert( + addsOf(changes).keySet === + Set(Seq("points", "element", "y"))) + } + + test("diffSchemas - a field added to a struct inside a map value " + + "uses the value path") { + val currentSchema = new StructType().add( + "points", + MapType(StringType, new StructType().add("x", DoubleType))) + val targetSchema = new StructType().add( + "points", + MapType( + StringType, + new StructType() + .add("x", DoubleType) + .add("y", DoubleType))) + + val changes = + SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) + assert( + addsOf(changes).keySet === + Set(Seq("points", "value", "y"))) + } + + test("diffSchemas - a field added to a struct inside a map key " + + "uses the key path") { + val currentSchema = new StructType().add( + "points", + MapType(new StructType().add("x", DoubleType), LongType)) + val targetSchema = new StructType().add( + "points", + MapType( + new StructType() + .add("x", DoubleType) + .add("y", DoubleType), + LongType)) + + val changes = + SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) + assert( + addsOf(changes).keySet === + Set(Seq("points", "key", "y"))) + } + + test("diffSchemas - an element type change inside an array " + + "uses the element path") { + val currentSchema = + new StructType().add("vals", ArrayType(IntegerType)) + val targetSchema = + new StructType().add("vals", ArrayType(LongType)) + + val changes = + SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) + assert( + typeUpdatesOf(changes) === + Map(Seq("vals", "element") -> LongType)) + } + + test("diffSchemas - a map value type change uses the value path") { + val currentSchema = new StructType() + .add("m", MapType(StringType, IntegerType)) + val targetSchema = new StructType() + .add("m", MapType(StringType, LongType)) + + val changes = + SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) + assert( + typeUpdatesOf(changes) === + Map(Seq("m", "value") -> LongType)) + } + + test("diffSchemas - an array containsNull change is a " + + "nullability update on the element") { + val currentSchema = new StructType() + .add("vals", ArrayType(IntegerType, containsNull = false)) + val targetSchema = new StructType() + .add("vals", ArrayType(IntegerType, containsNull = true)) + + val changes = + SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) + assert(changes.length === 1, s"changes=$changes") + assert( + nullabilityUpdatesOf(changes) === + Map(Seq("vals", "element") -> true)) + } + + test("diffSchemas - a map valueContainsNull change is a " + + "nullability update on the value") { + val currentSchema = new StructType().add( + "m", MapType(StringType, IntegerType, valueContainsNull = false)) + val targetSchema = new StructType().add( + "m", MapType(StringType, IntegerType, valueContainsNull = true)) + + val changes = + SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) + assert(changes.length === 1, s"changes=$changes") + assert( + nullabilityUpdatesOf(changes) === + Map(Seq("m", "value") -> true)) + } + + test("diffSchemas - a wholly new struct column stays " + + "a single top-level add") { + val newStruct = + new StructType().add("x", DoubleType).add("y", DoubleType) + val currentSchema = new StructType().add("id", IntegerType) + val targetSchema = new StructType() + .add("id", IntegerType) + .add("point", newStruct) + + val changes = + SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) + assert(changes.length === 1, s"changes=$changes") + assert( + addsOf(changes) === + Map(Seq("point") -> ((newStruct, true, null)))) + } + + test("diffSchemas - a whole struct column removed stays " + + "a single top-level delete") { + val currentSchema = new StructType() + .add("id", IntegerType) + .add("point", new StructType().add("x", DoubleType)) + val targetSchema = new StructType().add("id", IntegerType) + + val changes = + SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) + assert(changes.length === 1, s"changes=$changes") + assert(deletesOf(changes) === Set(Seq("point"))) + } + + test("diffSchemas - a struct replaced by an atomic type is a " + + "type update on the column") { + val currentSchema = new StructType() + .add("s", new StructType().add("a", IntegerType)) + val targetSchema = new StructType().add("s", StringType) + + val changes = + SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) + assert(changes.length === 1, s"changes=$changes") + assert(typeUpdatesOf(changes) === Map(Seq("s") -> StringType)) + } + + test("diffSchemas - identical nested schemas produce no changes") { + val schema = new StructType() + .add( + "s", + new StructType().add("a", IntegerType).add("b", StringType)) + .add( + "arr", + ArrayType(new StructType().add("x", DoubleType))) + .add( + "m", + MapType(StringType, new StructType().add("y", DoubleType))) + + assert( + SchemaInferenceUtils.diffSchemas(schema, schema).isEmpty) + } + + test("diffSchemas - independent nested changes are all " + + "emitted at their own leaves") { + val currentSchema = new StructType() + .add( + "s", + new StructType() + .add("a", IntegerType) + .add("gone", StringType)) + .add( + "arr", + ArrayType(new StructType().add("x", DoubleType))) + .add("top", IntegerType) + val targetSchema = new StructType() + .add( + "s", + new StructType() + .add("a", IntegerType) + .add("added", StringType)) + .add( + "arr", + ArrayType( + new StructType() + .add("x", DoubleType) + .add("z", DoubleType))) + .add("top", IntegerType) + .add("brand_new", BooleanType) + + val changes = + SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) + assert( + addsOf(changes).keySet === Set( + Seq("s", "added"), + Seq("arr", "element", "z"), + Seq("brand_new")), + s"changes=$changes") + assert( + deletesOf(changes) === Set(Seq("s", "gone")), + s"changes=$changes") + assert(typeUpdatesOf(changes).isEmpty, s"changes=$changes") + } + test("inferSchemaFromFlows folds a case-only column to the same spelling regardless of flow " + "order, even when identifier names contain dots") { // The merge order decides which spelling of a case-only-differing column survives, so it must From 19dad1b2062e060df0729c447f2fca8624826c7b Mon Sep 17 00:00:00 2001 From: Anish Mahto Date: Wed, 2 Sep 2026 07:34:41 +0000 Subject: [PATCH 2/6] cleanup --- .../pipelines/util/SchemaInferenceUtils.scala | 6 +- .../util/SchemaInferenceUtilsSuite.scala | 326 ++++++------------ 2 files changed, 104 insertions(+), 228 deletions(-) diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtils.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtils.scala index 4b1f788e66121..3829ce3d0c88d 100644 --- a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtils.scala +++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtils.scala @@ -217,9 +217,9 @@ object SchemaInferenceUtils { * Column identity is keyed on the exact field name, not on a case-normalized one. On the * incremental streaming-table path, `targetSchema` is the merge of the current and desired * schemas, and [[SchemaMergingUtils.mergeSchemas]] has already folded an incoming - * case-only-differing field onto the persisted one. On the non-merging paths (materialized - * views and any full refresh), `targetSchema` is the run's declared schema as-is, so - * exact-name keying keeps a case-only rename visible as an explicit drop-then-add. + * case-only-differing field onto the persisted one. On the non-merging paths (materialized views + * and any full refresh), `targetSchema` is the run's declared schema as-is, so exact-name keying + * keeps a case-only rename visible as an explicit drop-then-add. * Exact keying also avoids silently collapsing two genuinely distinct declared columns that * differ only in case (`value` and `Value`) into an arbitrary one of the two. * diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtilsSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtilsSuite.scala index c696058f74ce8..b279e4ed23f1e 100644 --- a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtilsSuite.scala +++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtilsSuite.scala @@ -37,6 +37,7 @@ import org.apache.spark.sql.test.SharedSparkSession import org.apache.spark.sql.types._ class SchemaInferenceUtilsSuite extends QueryTest with SharedSparkSession { + import TableChangeExtractors._ /** A [[FlowFunction]] that throws if invoked; the inferSchemaFromFlows test builds resolved * flows directly. */ @@ -215,27 +216,6 @@ class SchemaInferenceUtilsSuite extends QueryTest with SharedSparkSession { assert(descriptionCommentChange.newComment() === "Product description") } - test("determineColumnChanges - tightening nullability emits the change") { - // diffSchemas faithfully emits an UpdateColumnNullability when a column goes from - // nullable to non-nullable. Whether the catalog *accepts* this is a separate concern: - // Delta rejects tightening nullability on existing columns because existing data may - // contain nulls. The diff layer's job is to report the change; enforcement is downstream. - val currentSchema = new StructType() - .add("id", IntegerType) - .add("value", StringType, nullable = true) - val targetSchema = new StructType() - .add("id", IntegerType) - .add("value", StringType, nullable = false) - - val changes = - SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) - assert(changes.length === 1) - val nc = changes.head - .asInstanceOf[TableChange.UpdateColumnNullability] - assert(nc.fieldNames() === Array("value")) - assert(nc.nullable() === false) - } - test("determineColumnChanges - complex changes") { val currentSchema = new StructType() .add("id", IntegerType, nullable = false) @@ -477,58 +457,10 @@ class SchemaInferenceUtilsSuite extends QueryTest with SharedSparkSession { } } - // ============================================================================================= - // Nested diffing - // - // `diffSchemas` recurses into structs, array elements, and map keys/values so that a change - // inside a complex type is emitted at the leaf that actually changed. The alternative -- an - // UpdateColumnType on the enclosing column carrying the whole new complex type -- is rejected by - // `CheckAnalysis`, which fails ALTER COLUMN ... TYPE when the new type is a struct, map, or - // array. Pipelines only ever got away with it by calling `catalog.alterTable` directly and - // testing against an in-memory catalog that applies it regardless. - // ============================================================================================= - - /** Added columns as path -> (type, nullable, comment). */ - private def addsOf( - changes: Seq[TableChange]): Map[Seq[String], (DataType, Boolean, String)] = - changes.collect { - case ac: TableChange.AddColumn => - ac.fieldNames().toSeq -> ((ac.dataType(), ac.isNullable(), ac.comment())) - }.toMap - - private def deletesOf(changes: Seq[TableChange]): Set[Seq[String]] = - changes.collect { - case dc: TableChange.DeleteColumn => dc.fieldNames().toSeq - }.toSet - - private def typeUpdatesOf( - changes: Seq[TableChange]): Map[Seq[String], DataType] = - changes.collect { - case tc: TableChange.UpdateColumnType => - tc.fieldNames().toSeq -> tc.newDataType() - }.toMap - - private def nullabilityUpdatesOf( - changes: Seq[TableChange]): Map[Seq[String], Boolean] = - changes.collect { - case nc: TableChange.UpdateColumnNullability => - nc.fieldNames().toSeq -> nc.nullable() - }.toMap - - private def commentUpdatesOf( - changes: Seq[TableChange]): Map[Seq[String], String] = - changes.collect { - case cc: TableChange.UpdateColumnComment => - cc.fieldNames().toSeq -> cc.newComment() - }.toMap - - test("diffSchemas - a leaf added to a struct is a nested add, " + - "not a retype of the parent") { + test("diffSchemas - a leaf added to a struct is a nested add, not a retype of the parent") { val currentSchema = new StructType() .add("id", IntegerType) - .add( - "point", - new StructType().add("x", DoubleType).add("y", DoubleType)) + .add("point", new StructType().add("x", DoubleType).add("y", DoubleType)) val targetSchema = new StructType() .add("id", IntegerType) .add( @@ -538,29 +470,22 @@ class SchemaInferenceUtilsSuite extends QueryTest with SharedSparkSession { .add("y", DoubleType) .add("z", DoubleType)) - val changes = - SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) + val changes = SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) assert(changes.length === 1, s"changes=$changes") - assert( - addsOf(changes) === - Map(Seq("point", "z") -> ((DoubleType, true, null)))) + assert(addsOf(changes) === Map(Seq("point", "z") -> ((DoubleType, true, null)))) } - test("diffSchemas - a leaf added to a deeply nested struct " + - "carries the full path") { + test("diffSchemas - a leaf added to a deeply nested struct carries the full path") { val inner = new StructType().add("c", IntegerType) - val currentSchema = - new StructType().add("a", new StructType().add("b", inner)) + val currentSchema = new StructType().add("a", new StructType().add("b", inner)) val targetSchema = new StructType() .add("a", new StructType().add("b", inner.add("d", StringType))) - val changes = - SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) + val changes = SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) assert(addsOf(changes).keySet === Set(Seq("a", "b", "d"))) } - test("diffSchemas - an added nested leaf keeps its declared " + - "nullability and comment") { + test("diffSchemas - an added nested leaf keeps its declared nullability and comment") { val currentSchema = new StructType() .add("s", new StructType().add("a", IntegerType)) val targetSchema = new StructType().add( @@ -569,131 +494,86 @@ class SchemaInferenceUtilsSuite extends QueryTest with SharedSparkSession { .add("a", IntegerType) .add("b", StringType, nullable = false, "a comment")) - val changes = - SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) - assert( - addsOf(changes) === - Map(Seq("s", "b") -> ((StringType, false, "a comment")))) + val changes = SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) + assert(addsOf(changes) === Map(Seq("s", "b") -> ((StringType, false, "a comment")))) } test("diffSchemas - a leaf removed from a struct is a nested delete") { val currentSchema = new StructType() - .add( - "s", - new StructType().add("a", IntegerType).add("b", StringType)) + .add("s", new StructType().add("a", IntegerType).add("b", StringType)) val targetSchema = new StructType() .add("s", new StructType().add("a", IntegerType)) - val changes = - SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) + val changes = SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) assert(changes.length === 1, s"changes=$changes") assert(deletesOf(changes) === Set(Seq("s", "b"))) } - test("diffSchemas - a leaf type change inside a struct " + - "is a nested type update") { + test("diffSchemas - a leaf type change inside a struct is a nested type update") { val currentSchema = new StructType() .add("s", new StructType().add("a", IntegerType)) val targetSchema = new StructType() .add("s", new StructType().add("a", LongType)) - val changes = - SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) + val changes = SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) assert(changes.length === 1, s"changes=$changes") assert(typeUpdatesOf(changes) === Map(Seq("s", "a") -> LongType)) } - test("diffSchemas - nested leaf nullability and comment changes " + - "are emitted at the leaf") { + test("diffSchemas - nested leaf nullability and comment changes are emitted at the leaf") { val currentSchema = new StructType() - .add( - "s", - new StructType().add("a", IntegerType, nullable = true, "old")) + .add("s", new StructType().add("a", IntegerType, nullable = true, "old")) val targetSchema = new StructType() - .add( - "s", - new StructType().add("a", IntegerType, nullable = false, "new")) + .add("s", new StructType().add("a", IntegerType, nullable = false, "new")) - val changes = - SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) + val changes = SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) assert(changes.length === 2, s"changes=$changes") - assert( - nullabilityUpdatesOf(changes) === Map(Seq("s", "a") -> false)) + assert(nullabilityUpdatesOf(changes) === Map(Seq("s", "a") -> false)) assert(commentUpdatesOf(changes) === Map(Seq("s", "a") -> "new")) } - test("diffSchemas - a field added to a struct inside an array " + - "uses the element path") { + test("diffSchemas - a field added to a struct inside an array uses the element path") { val currentSchema = new StructType() - .add( - "points", - ArrayType(new StructType().add("x", DoubleType))) - val targetSchema = new StructType().add( - "points", - ArrayType( - new StructType() - .add("x", DoubleType) - .add("y", DoubleType))) + .add("points", ArrayType(new StructType().add("x", DoubleType))) + val targetSchema = new StructType() + .add("points", ArrayType(new StructType().add("x", DoubleType).add("y", DoubleType))) - val changes = - SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) - assert( - addsOf(changes).keySet === - Set(Seq("points", "element", "y"))) + val changes = SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) + assert(addsOf(changes).keySet === Set(Seq("points", "element", "y"))) } - test("diffSchemas - a field added to a struct inside a map value " + - "uses the value path") { + test("diffSchemas - a field added to a struct inside a map value uses the value path") { val currentSchema = new StructType().add( "points", MapType(StringType, new StructType().add("x", DoubleType))) - val targetSchema = new StructType().add( - "points", - MapType( - StringType, - new StructType() - .add("x", DoubleType) - .add("y", DoubleType))) + val targetSchema = new StructType() + .add( + "points", + MapType(StringType, new StructType().add("x", DoubleType).add("y", DoubleType))) - val changes = - SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) - assert( - addsOf(changes).keySet === - Set(Seq("points", "value", "y"))) + val changes = SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) + assert(addsOf(changes).keySet === Set(Seq("points", "value", "y"))) } - test("diffSchemas - a field added to a struct inside a map key " + - "uses the key path") { + test("diffSchemas - a field added to a struct inside a map key uses the key path") { val currentSchema = new StructType().add( "points", MapType(new StructType().add("x", DoubleType), LongType)) - val targetSchema = new StructType().add( - "points", - MapType( - new StructType() - .add("x", DoubleType) - .add("y", DoubleType), - LongType)) + val targetSchema = new StructType() + .add( + "points", + MapType(new StructType().add("x", DoubleType).add("y", DoubleType), LongType)) - val changes = - SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) - assert( - addsOf(changes).keySet === - Set(Seq("points", "key", "y"))) + val changes = SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) + assert(addsOf(changes).keySet === Set(Seq("points", "key", "y"))) } - test("diffSchemas - an element type change inside an array " + - "uses the element path") { - val currentSchema = - new StructType().add("vals", ArrayType(IntegerType)) - val targetSchema = - new StructType().add("vals", ArrayType(LongType)) + test("diffSchemas - an element type change inside an array uses the element path") { + val currentSchema = new StructType().add("vals", ArrayType(IntegerType)) + val targetSchema = new StructType().add("vals", ArrayType(LongType)) - val changes = - SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) - assert( - typeUpdatesOf(changes) === - Map(Seq("vals", "element") -> LongType)) + val changes = SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) + assert(typeUpdatesOf(changes) === Map(Seq("vals", "element") -> LongType)) } test("diffSchemas - a map value type change uses the value path") { @@ -702,112 +582,82 @@ class SchemaInferenceUtilsSuite extends QueryTest with SharedSparkSession { val targetSchema = new StructType() .add("m", MapType(StringType, LongType)) - val changes = - SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) - assert( - typeUpdatesOf(changes) === - Map(Seq("m", "value") -> LongType)) + val changes = SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) + assert(typeUpdatesOf(changes) === Map(Seq("m", "value") -> LongType)) } - test("diffSchemas - an array containsNull change is a " + - "nullability update on the element") { + test("diffSchemas - an array containsNull change is a nullability update on the element") { val currentSchema = new StructType() .add("vals", ArrayType(IntegerType, containsNull = false)) val targetSchema = new StructType() .add("vals", ArrayType(IntegerType, containsNull = true)) - val changes = - SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) + val changes = SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) assert(changes.length === 1, s"changes=$changes") - assert( - nullabilityUpdatesOf(changes) === - Map(Seq("vals", "element") -> true)) + assert(nullabilityUpdatesOf(changes) === Map(Seq("vals", "element") -> true)) } - test("diffSchemas - a map valueContainsNull change is a " + - "nullability update on the value") { + test("diffSchemas - a map valueContainsNull change is a nullability update on the value") { val currentSchema = new StructType().add( "m", MapType(StringType, IntegerType, valueContainsNull = false)) val targetSchema = new StructType().add( "m", MapType(StringType, IntegerType, valueContainsNull = true)) - val changes = - SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) + val changes = SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) assert(changes.length === 1, s"changes=$changes") - assert( - nullabilityUpdatesOf(changes) === - Map(Seq("m", "value") -> true)) + assert(nullabilityUpdatesOf(changes) === Map(Seq("m", "value") -> true)) } - test("diffSchemas - a wholly new struct column stays " + - "a single top-level add") { - val newStruct = - new StructType().add("x", DoubleType).add("y", DoubleType) + test("diffSchemas - a wholly new struct column stays a single top-level add") { + val newStruct = new StructType().add("x", DoubleType).add("y", DoubleType) val currentSchema = new StructType().add("id", IntegerType) val targetSchema = new StructType() .add("id", IntegerType) .add("point", newStruct) - val changes = - SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) + val changes = SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) assert(changes.length === 1, s"changes=$changes") - assert( - addsOf(changes) === - Map(Seq("point") -> ((newStruct, true, null)))) + assert(addsOf(changes) === Map(Seq("point") -> ((newStruct, true, null)))) } - test("diffSchemas - a whole struct column removed stays " + - "a single top-level delete") { + test("diffSchemas - a whole struct column removed stays a single top-level delete") { val currentSchema = new StructType() .add("id", IntegerType) .add("point", new StructType().add("x", DoubleType)) val targetSchema = new StructType().add("id", IntegerType) - val changes = - SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) + val changes = SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) assert(changes.length === 1, s"changes=$changes") assert(deletesOf(changes) === Set(Seq("point"))) } - test("diffSchemas - a struct replaced by an atomic type is a " + - "type update on the column") { + test("diffSchemas - a struct replaced by an atomic type is a type update on the column") { val currentSchema = new StructType() .add("s", new StructType().add("a", IntegerType)) val targetSchema = new StructType().add("s", StringType) - val changes = - SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) + val changes = SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) assert(changes.length === 1, s"changes=$changes") assert(typeUpdatesOf(changes) === Map(Seq("s") -> StringType)) } test("diffSchemas - identical nested schemas produce no changes") { val schema = new StructType() - .add( - "s", - new StructType().add("a", IntegerType).add("b", StringType)) - .add( - "arr", - ArrayType(new StructType().add("x", DoubleType))) - .add( - "m", - MapType(StringType, new StructType().add("y", DoubleType))) + .add("s", new StructType().add("a", IntegerType).add("b", StringType)) + .add("arr", ArrayType(new StructType().add("x", DoubleType))) + .add("m", MapType(StringType, new StructType().add("y", DoubleType))) - assert( - SchemaInferenceUtils.diffSchemas(schema, schema).isEmpty) + assert(SchemaInferenceUtils.diffSchemas(schema, schema).isEmpty) } - test("diffSchemas - independent nested changes are all " + - "emitted at their own leaves") { + test("diffSchemas - independent nested changes are all emitted at their own leaves") { val currentSchema = new StructType() .add( "s", new StructType() .add("a", IntegerType) .add("gone", StringType)) - .add( - "arr", - ArrayType(new StructType().add("x", DoubleType))) + .add("arr", ArrayType(new StructType().add("x", DoubleType))) .add("top", IntegerType) val targetSchema = new StructType() .add( @@ -815,17 +665,11 @@ class SchemaInferenceUtilsSuite extends QueryTest with SharedSparkSession { new StructType() .add("a", IntegerType) .add("added", StringType)) - .add( - "arr", - ArrayType( - new StructType() - .add("x", DoubleType) - .add("z", DoubleType))) + .add("arr", ArrayType(new StructType().add("x", DoubleType).add("z", DoubleType))) .add("top", IntegerType) .add("brand_new", BooleanType) - val changes = - SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) + val changes = SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) assert( addsOf(changes).keySet === Set( Seq("s", "added"), @@ -867,3 +711,35 @@ class SchemaInferenceUtilsSuite extends QueryTest with SharedSparkSession { } } } + +private[util] object TableChangeExtractors { + /** Added columns as path -> (type, nullable, comment). */ + def addsOf(changes: Seq[TableChange]): Map[Seq[String], (DataType, Boolean, String)] = + changes.collect { + case ac: TableChange.AddColumn => + ac.fieldNames().toSeq -> ((ac.dataType(), ac.isNullable(), ac.comment())) + }.toMap + + def deletesOf(changes: Seq[TableChange]): Set[Seq[String]] = + changes.collect { + case dc: TableChange.DeleteColumn => dc.fieldNames().toSeq + }.toSet + + def typeUpdatesOf(changes: Seq[TableChange]): Map[Seq[String], DataType] = + changes.collect { + case tc: TableChange.UpdateColumnType => + tc.fieldNames().toSeq -> tc.newDataType() + }.toMap + + def nullabilityUpdatesOf(changes: Seq[TableChange]): Map[Seq[String], Boolean] = + changes.collect { + case nc: TableChange.UpdateColumnNullability => + nc.fieldNames().toSeq -> nc.nullable() + }.toMap + + def commentUpdatesOf(changes: Seq[TableChange]): Map[Seq[String], String] = + changes.collect { + case cc: TableChange.UpdateColumnComment => + cc.fieldNames().toSeq -> cc.newComment() + }.toMap +} From 06b78b35c01abd7789bccefc1b1332a70443dc76 Mon Sep 17 00:00:00 2001 From: Anish Mahto Date: Wed, 2 Sep 2026 17:38:08 +0000 Subject: [PATCH 3/6] compile error + add catalog test --- .../pipelines/util/SchemaInferenceUtils.scala | 12 +++--- .../util/SchemaInferenceUtilsSuite.scala | 40 ++++++++++++++++++- 2 files changed, 45 insertions(+), 7 deletions(-) diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtils.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtils.scala index 3829ce3d0c88d..cf39a0b8d81e5 100644 --- a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtils.scala +++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtils.scala @@ -258,10 +258,10 @@ object SchemaInferenceUtils { ) .map { fieldInTarget => TableChange.addColumn( - fieldNames = (pathToStruct :+ fieldInTarget.name).toArray, - dataType = fieldInTarget.dataType, - isNullable = fieldInTarget.nullable, - comment = fieldInTarget.getComment().orNull + (pathToStruct :+ fieldInTarget.name).toArray, + fieldInTarget.dataType, + fieldInTarget.nullable, + fieldInTarget.getComment().orNull ) } @@ -273,8 +273,8 @@ object SchemaInferenceUtils { .map(fieldInCurrent => TableChange .deleteColumn( - fieldNames = (pathToStruct :+ fieldInCurrent.name).toArray, - ifExists = false + (pathToStruct :+ fieldInCurrent.name).toArray, + false ) ) diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtilsSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtilsSuite.scala index b279e4ed23f1e..e79d9b0ede309 100644 --- a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtilsSuite.scala +++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtilsSuite.scala @@ -22,7 +22,13 @@ import scala.util.Success import org.apache.spark.SparkException import org.apache.spark.sql.{QueryTest, Row} import org.apache.spark.sql.catalyst.TableIdentifier -import org.apache.spark.sql.connector.catalog.TableChange +import org.apache.spark.sql.connector.catalog.{ + CatalogV2Util, + Identifier, + InMemoryTableCatalog, + TableChange, + TableInfo +} import org.apache.spark.sql.pipelines.graph.{ FlowFunction, FlowFunctionResult, @@ -35,6 +41,7 @@ import org.apache.spark.sql.pipelines.graph.{ } import org.apache.spark.sql.test.SharedSparkSession import org.apache.spark.sql.types._ +import org.apache.spark.sql.util.CaseInsensitiveStringMap class SchemaInferenceUtilsSuite extends QueryTest with SharedSparkSession { import TableChangeExtractors._ @@ -682,6 +689,37 @@ class SchemaInferenceUtilsSuite extends QueryTest with SharedSparkSession { assert(typeUpdatesOf(changes).isEmpty, s"changes=$changes") } + test("diffSchemas - DSv2 catalog respects emitted nested changes") { + val currentNestedStruct = + new StructType().add("x", DoubleType, nullable = false, "old comment") + val currentSchema = new StructType() + .add("struct", currentNestedStruct) + .add("array", ArrayType(currentNestedStruct)) + .add("map", MapType(currentNestedStruct, currentNestedStruct)) + + val targetNestedStruct = + new StructType() + .add("x", DoubleType, nullable = true, "new comment") + .add("y", DoubleType) + // Add `y`, make `x` nullable, and update its comment in every nested struct. + val targetSchema = new StructType() + .add("struct", targetNestedStruct) + .add("array", ArrayType(targetNestedStruct)) + .add("map", MapType(targetNestedStruct, targetNestedStruct)) + + val catalog = new InMemoryTableCatalog + catalog.initialize("test", CaseInsensitiveStringMap.empty()) + val ident = Identifier.of(Array.empty, "t") + catalog.createTable(ident, new TableInfo.Builder().withSchema(currentSchema).build()) + + val changes = SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) + val updated = catalog.alterTable(ident, changes: _*) + + assert( + CatalogV2Util.clearIds(updated.columns()) === + CatalogV2Util.structTypeToV2Columns(targetSchema, keepIds = false)) + } + test("inferSchemaFromFlows folds a case-only column to the same spelling regardless of flow " + "order, even when identifier names contain dots") { // The merge order decides which spelling of a case-only-differing column survives, so it must From 9d7a431b4c9a1cc045dcbc18d8d6ac4a6a1f35e8 Mon Sep 17 00:00:00 2001 From: Anish Mahto Date: Wed, 2 Sep 2026 21:55:58 +0000 Subject: [PATCH 4/6] add logging for metadata drift --- .../pipelines/util/SchemaInferenceUtils.scala | 39 +++++++++++++++++-- 1 file changed, 36 insertions(+), 3 deletions(-) diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtils.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtils.scala index cf39a0b8d81e5..8040811adf239 100644 --- a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtils.scala +++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtils.scala @@ -25,6 +25,7 @@ import org.apache.spark.sql.catalyst.analysis.{ caseSensitiveResolution, Resolver } +import org.apache.spark.internal.Logging import org.apache.spark.sql.connector.catalog.TableChange import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.pipelines.common.DatasetType @@ -34,10 +35,13 @@ import org.apache.spark.sql.pipelines.graph.{ GraphErrors, ResolvedFlow } -import org.apache.spark.sql.types.{ArrayType, DataType, MapType, StructField, StructType} +import org.apache.spark.sql.types.{ + ArrayType, DataType, MapType, Metadata, MetadataBuilder, + StructField, StructType +} -object SchemaInferenceUtils { +object SchemaInferenceUtils extends Logging { def resolverFor(caseSensitive: Boolean): Resolver = { if (caseSensitive) { @@ -295,16 +299,45 @@ object SchemaInferenceUtils { columnsAdded ++ columnsDeleted ++ columnsUpdated } - /** Diffs the type, nullability, and comment of one field present in both schemas. */ + /** + * Diffs the type, nullability, and comment of one field present in both schemas. Other + * StructField.metadata entries (defaults, generated-column expressions, connector-specific + * metadata) are not diffed: pipeline schema synchronization does not support propagating + * them, and Spark's own ResolveSchemaEvolution likewise ignores them. + */ private def diffField( currentField: StructField, targetField: StructField, pathToField: Seq[String]): Seq[TableChange] = { + warnOnFieldMetadataDrift(currentField, targetField, pathToField) diffDataTypes(currentField.dataType, targetField.dataType, pathToField) ++ diffNullability(currentField.nullable, targetField.nullable, pathToField) ++ diffComment(currentField.getComment(), targetField.getComment(), pathToField) } + /** + * Logs a warning when two fields' metadata bags differ beyond the "comment" key + * (which is already handled by [[diffComment]]). Pipeline schema synchronization does not + * support propagating other metadata entries (defaults, generated-column expressions, + * connector-specific metadata), so these differences are left for the user to reconcile. + */ + private def warnOnFieldMetadataDrift( + currentField: StructField, + targetField: StructField, + pathToField: Seq[String]): Unit = { + val current = stripMetadataComment(currentField.metadata) + val target = stripMetadataComment(targetField.metadata) + if (current != target) { + logWarning( + s"Field ${pathToField.mkString(".")} has metadata changes that pipeline schema " + + s"synchronization does not propagate and will be ignored. " + + s"Current: ${current.json}, Target: ${target.json}") + } + } + + private def stripMetadataComment(m: Metadata): Metadata = + new MetadataBuilder().withMetadata(m).remove("comment").build() + private def diffNullability( currentNullable: Boolean, targetNullable: Boolean, From 832410da3296dfa74a9b57a21398760fee3c4c6f Mon Sep 17 00:00:00 2001 From: Anish Mahto Date: Thu, 3 Sep 2026 02:26:04 +0000 Subject: [PATCH 5/6] clean up --- .../apache/spark/sql/pipelines/util/SchemaInferenceUtils.scala | 2 +- .../spark/sql/pipelines/util/SchemaInferenceUtilsSuite.scala | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtils.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtils.scala index 8040811adf239..d5232c3350a9a 100644 --- a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtils.scala +++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtils.scala @@ -19,13 +19,13 @@ package org.apache.spark.sql.pipelines.util import scala.util.control.NonFatal +import org.apache.spark.internal.Logging import org.apache.spark.sql.catalyst.TableIdentifier import org.apache.spark.sql.catalyst.analysis.{ caseInsensitiveResolution, caseSensitiveResolution, Resolver } -import org.apache.spark.internal.Logging import org.apache.spark.sql.connector.catalog.TableChange import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.pipelines.common.DatasetType diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtilsSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtilsSuite.scala index e79d9b0ede309..11c533c52f3b5 100644 --- a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtilsSuite.scala +++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtilsSuite.scala @@ -415,7 +415,7 @@ class SchemaInferenceUtilsSuite extends QueryTest with SharedSparkSession { } test("mergeSchemas - a nested case-only field whose type also changes fails to merge, and " + - "diffSchemas reports it as a type change") { + "diffSchemas reports it as a drop-then-add") { // A nested field that differs only in case AND changes type is rejected rather than silently // resolved. Note this is a *type* incompatibility, not a case one: `StructType.merge` never // widens numeric types, so `int` -> `long` fails identically for a same-cased field and at the From de2a5dfd8b2bb88144f0b566accf316ead38f2aa Mon Sep 17 00:00:00 2001 From: Anish Mahto Date: Fri, 4 Sep 2026 06:15:09 +0000 Subject: [PATCH 6/6] loudly throw for nullability tightening and unsupported nested element evolution --- .../resources/error/error-conditions.json | 14 ++ .../pipelines/util/SchemaInferenceUtils.scala | 97 ++++++++++-- .../util/SchemaInferenceUtilsSuite.scala | 146 +++++++++++++++--- 3 files changed, 225 insertions(+), 32 deletions(-) diff --git a/common/utils/src/main/resources/error/error-conditions.json b/common/utils/src/main/resources/error/error-conditions.json index 3714c27d2a98b..4b4b7c22b661d 100644 --- a/common/utils/src/main/resources/error/error-conditions.json +++ b/common/utils/src/main/resources/error/error-conditions.json @@ -6742,6 +6742,14 @@ ], "sqlState" : "0A000" }, + "PIPELINE_NESTED_COMPLEX_TYPE_SCHEMA_EVOLUTION_UNSUPPORTED" : { + "message" : [ + "Schema evolution within a complex type (array or map) whose element is itself a complex type is not supported.", + "Column path has element type but the target element type is .", + "Flatten the nesting or apply the change manually." + ], + "sqlState" : "0A000" + }, "PIPELINE_RUN_FAILED" : { "message" : [ "" @@ -6762,6 +6770,12 @@ ], "sqlState" : "42K03" }, + "PIPELINE_TIGHTEN_NULLABILITY_UNSUPPORTED" : { + "message" : [ + "Cannot tighten nullability of from nullable to non-nullable. Existing data may already contain nulls." + ], + "sqlState" : "0A000" + }, "PIPE_OPERATOR_AGGREGATE_EXPRESSION_CONTAINS_NO_AGGREGATE_FUNCTION" : { "message" : [ "Non-grouping expression is provided as an argument to the |> AGGREGATE pipe operator but does not contain any aggregate function; please update it to include an aggregate function and then retry the query again." diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtils.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtils.scala index d5232c3350a9a..9255b950e68a1 100644 --- a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtils.scala +++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtils.scala @@ -19,6 +19,7 @@ package org.apache.spark.sql.pipelines.util import scala.util.control.NonFatal +import org.apache.spark.SparkUnsupportedOperationException import org.apache.spark.internal.Logging import org.apache.spark.sql.catalyst.TableIdentifier import org.apache.spark.sql.catalyst.analysis.{ @@ -343,6 +344,11 @@ object SchemaInferenceUtils extends Logging { targetNullable: Boolean, pathToField: Seq[String] ): Option[TableChange] = { + if (currentNullable && !targetNullable) { + throw new SparkUnsupportedOperationException( + errorClass = "PIPELINE_TIGHTEN_NULLABILITY_UNSUPPORTED", + messageParameters = Map("columnPath" -> pathToField.mkString("."))) + } Option.when(currentNullable != targetNullable)( TableChange.updateColumnNullability(pathToField.toArray, targetNullable) ) @@ -358,28 +364,72 @@ object SchemaInferenceUtils extends Logging { ) } - /** Diffs two data types at `path`, descending through matching complex types. */ + /** + * Diffs two data types at `path`, descending through matching complex types. + * + * Recurses freely through structs, arrays, and maps. After recursing into an array element + * or map key/value, any actual changes are rejected when the child type is itself an array + * or map, because [[org.apache.spark.sql.connector.catalog.CatalogV2Util]] cannot resolve + * paths with consecutive `element`/`key`/`value` segments (see SPARK-59188). + */ private def diffDataTypes( currentType: DataType, targetType: DataType, - pathToField: Seq[String] - ): Seq[TableChange] = (currentType, targetType) match { + pathToField: Seq[String]): Seq[TableChange] = (currentType, targetType) match { case (currentStruct: StructType, targetStruct: StructType) => diffStructs(currentStruct, targetStruct, pathToField) case (currentArray: ArrayType, targetArray: ArrayType) => val elementPath = pathToField :+ "element" - - diffDataTypes(currentArray.elementType, targetArray.elementType, elementPath) ++ - diffNullability(currentArray.containsNull, targetArray.containsNull, elementPath) + val dataTypeChanges = diffDataTypes( + currentType = currentArray.elementType, + targetType = targetArray.elementType, + pathToField = elementPath + ) + val nullabilityChanges = diffNullability( + currentNullable = currentArray.containsNull, + targetNullable = targetArray.containsNull, + pathToField = elementPath + ) + rejectUnsupportedNestedTypeChanges( + currentType = currentArray.elementType, + targetType = targetArray.elementType, + typeChanges = dataTypeChanges, + pathToElement = elementPath + ) + dataTypeChanges ++ nullabilityChanges case (currentMap: MapType, targetMap: MapType) => - val valuePath = pathToField :+ "value" val keyPath = pathToField :+ "key" - - diffDataTypes(currentMap.keyType, targetMap.keyType, keyPath) ++ - diffDataTypes(currentMap.valueType, targetMap.valueType, valuePath) ++ - diffNullability(currentMap.valueContainsNull, targetMap.valueContainsNull, valuePath) + val valuePath = pathToField :+ "value" + val keyTypeChanges = diffDataTypes( + currentType = currentMap.keyType, + targetType = targetMap.keyType, + pathToField = keyPath + ) + val valueTypeChanges = diffDataTypes( + currentType = currentMap.valueType, + targetType = targetMap.valueType, + pathToField = valuePath + ) + val valueNullabilityChanges = diffNullability( + currentNullable = currentMap.valueContainsNull, + targetNullable = targetMap.valueContainsNull, + pathToField = valuePath + ) + rejectUnsupportedNestedTypeChanges( + currentType = currentMap.keyType, + targetType = targetMap.keyType, + typeChanges = keyTypeChanges, + pathToElement = keyPath + ) + rejectUnsupportedNestedTypeChanges( + currentType = currentMap.valueType, + targetType = targetMap.valueType, + typeChanges = valueTypeChanges, + pathToElement = valuePath + ) + keyTypeChanges ++ valueTypeChanges ++ valueNullabilityChanges case _ if currentType == targetType => Seq.empty @@ -387,4 +437,29 @@ object SchemaInferenceUtils extends Logging { case _ => Seq(TableChange.updateColumnType(pathToField.toArray, targetType)) } + + /** + * Throws when an array element or map key/value is itself an array or map and the + * recursive diff found changes. The resulting paths would contain consecutive + * `element`/`key`/`value` segments that + * [[org.apache.spark.sql.connector.catalog.CatalogV2Util]] cannot resolve (SPARK-59188). + */ + private def rejectUnsupportedNestedTypeChanges( + currentType: DataType, + targetType: DataType, + typeChanges: Seq[TableChange], + pathToElement: Seq[String]): Unit = { + if (typeChanges.nonEmpty) { + (currentType, targetType) match { + case (_: ArrayType | _: MapType, _: ArrayType | _: MapType) => + throw new SparkUnsupportedOperationException( + errorClass = "PIPELINE_NESTED_COMPLEX_TYPE_SCHEMA_EVOLUTION_UNSUPPORTED", + messageParameters = Map( + "columnPath" -> pathToElement.mkString("."), + "currentType" -> currentType.simpleString, + "targetType" -> targetType.simpleString)) + case _ => + } + } + } } diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtilsSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtilsSuite.scala index 11c533c52f3b5..c02b1f9119339 100644 --- a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtilsSuite.scala +++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/util/SchemaInferenceUtilsSuite.scala @@ -19,7 +19,7 @@ package org.apache.spark.sql.pipelines.util import scala.util.Success -import org.apache.spark.SparkException +import org.apache.spark.{SparkException, SparkUnsupportedOperationException} import org.apache.spark.sql.{QueryTest, Row} import org.apache.spark.sql.catalyst.TableIdentifier import org.apache.spark.sql.connector.catalog.{ @@ -172,18 +172,18 @@ class SchemaInferenceUtilsSuite extends QueryTest with SharedSparkSession { test("determineColumnChanges - updating nullability and comments") { val currentSchema = new StructType() .add("id", IntegerType, nullable = false) - .add("name", StringType, nullable = true) + .add("name", StringType, nullable = false) .add("description", StringType, nullable = true, "Item description") val targetSchema = new StructType() - .add("id", IntegerType, nullable = true) // Changed nullability - .add("name", StringType, nullable = false) // Changed nullability + .add("id", IntegerType, nullable = true) // Widened nullability + .add("name", StringType, nullable = true) // Widened nullability .add("description", StringType, nullable = true, "Product description") // Changed comment val changes = SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) - // Should have 3 changes - updating nullability for 'id' and 'name', and comment for - // 'description' + // Should have 3 changes - widening nullability for 'id' and 'name', and comment + // for 'description' assert(changes.length === 3) // Verify the nullability changes @@ -215,9 +215,9 @@ class SchemaInferenceUtilsSuite extends QueryTest with SharedSparkSession { .get .asInstanceOf[TableChange.UpdateColumnComment] - // Verify the new nullability values + // Verify the new nullability values (both widened to nullable) assert(idNullabilityChange.nullable() === true) - assert(nameNullabilityChange.nullable() === false) + assert(nameNullabilityChange.nullable() === true) // Verify the new comment assert(descriptionCommentChange.newComment() === "Product description") @@ -230,21 +230,19 @@ class SchemaInferenceUtilsSuite extends QueryTest with SharedSparkSession { .add("old_field", BooleanType) val targetSchema = new StructType() - .add("id", LongType, nullable = true) // Changed type and nullability - // Added comment and changed nullability - .add("name", StringType, nullable = false, "Full name") + .add("id", LongType, nullable = true) // Changed type and widened nullability + .add("name", StringType, nullable = true, "Full name") // Added comment .add("new_field", StringType) // New field val changes = SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) // Should have these changes: // 1. Update id type - // 2. Update id nullability - // 3. Update name nullability - // 4. Update name comment - // 5. Add new_field - // 6. Remove old_field - assert(changes.length === 6) + // 2. Update id nullability (widened) + // 3. Update name comment + // 4. Add new_field + // 5. Remove old_field + assert(changes.length === 5) // Count the types of changes val typeChanges = changes.collect { case _: TableChange.UpdateColumnType => 1 }.size @@ -255,7 +253,7 @@ class SchemaInferenceUtilsSuite extends QueryTest with SharedSparkSession { val addColumnChanges = changes.collect { case _: TableChange.AddColumn => 1 }.size assert(typeChanges === 1) - assert(nullabilityChanges === 2) + assert(nullabilityChanges === 1) assert(commentChanges === 1) assert(addColumnChanges === 1) } @@ -529,13 +527,13 @@ class SchemaInferenceUtilsSuite extends QueryTest with SharedSparkSession { test("diffSchemas - nested leaf nullability and comment changes are emitted at the leaf") { val currentSchema = new StructType() - .add("s", new StructType().add("a", IntegerType, nullable = true, "old")) + .add("s", new StructType().add("a", IntegerType, nullable = false, "old")) val targetSchema = new StructType() - .add("s", new StructType().add("a", IntegerType, nullable = false, "new")) + .add("s", new StructType().add("a", IntegerType, nullable = true, "new")) val changes = SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) assert(changes.length === 2, s"changes=$changes") - assert(nullabilityUpdatesOf(changes) === Map(Seq("s", "a") -> false)) + assert(nullabilityUpdatesOf(changes) === Map(Seq("s", "a") -> true)) assert(commentUpdatesOf(changes) === Map(Seq("s", "a") -> "new")) } @@ -720,6 +718,112 @@ class SchemaInferenceUtilsSuite extends QueryTest with SharedSparkSession { CatalogV2Util.structTypeToV2Columns(targetSchema, keepIds = false)) } + test("diffSchemas - array> throws unsupported error") { + val innerCurrent = + ArrayType(new StructType().add("x", IntegerType).add("y", StringType)) + val innerTarget = ArrayType(new StructType().add("x", IntegerType)) + + val currentSchema = new StructType().add("a", ArrayType(innerCurrent)) + val targetSchema = new StructType().add("a", ArrayType(innerTarget)) + + // CatalogV2Util cannot resolve paths with consecutive element/key/value + // segments (SPARK-59188), so diffSchemas rejects nested complex type + // evolution rather than emitting changes the catalog cannot apply. + checkError( + intercept[SparkUnsupportedOperationException]( + SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema)), + condition = + "PIPELINE_NESTED_COMPLEX_TYPE_SCHEMA_EVOLUTION_UNSUPPORTED", + parameters = Map( + "columnPath" -> "a.element", + "currentType" -> innerCurrent.simpleString, + "targetType" -> innerTarget.simpleString)) + } + + test("diffSchemas - map> throws unsupported error") { + val valCurrent = ArrayType(new StructType().add("x", IntegerType)) + val valTarget = ArrayType( + new StructType().add("x", IntegerType).add("y", StringType)) + + val currentSchema = new StructType() + .add("m", MapType(StringType, valCurrent)) + val targetSchema = new StructType() + .add("m", MapType(StringType, valTarget)) + + checkError( + intercept[SparkUnsupportedOperationException]( + SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema)), + condition = + "PIPELINE_NESTED_COMPLEX_TYPE_SCHEMA_EVOLUTION_UNSUPPORTED", + parameters = Map( + "columnPath" -> "m.value", + "currentType" -> valCurrent.simpleString, + "targetType" -> valTarget.simpleString)) + } + + test("diffSchemas - map> recurses into struct key") { + val keyCurrent = new StructType().add("a", IntegerType) + val keyTarget = + new StructType().add("a", IntegerType).add("b", StringType) + + val currentSchema = new StructType() + .add("m", MapType(keyCurrent, MapType(StringType, IntegerType))) + val targetSchema = new StructType() + .add("m", MapType(keyTarget, MapType(StringType, IntegerType))) + + val changes = SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) + // The key is a struct, so leaf-level diff is emitted. + assert(changes.length === 1, s"changes=$changes") + assert( + addsOf(changes) === + Map(Seq("m", "key", "b") -> ((StringType, true, null)))) + } + + test("diffSchemas - identical nested array> produces no changes") { + val inner = ArrayType(new StructType().add("x", IntegerType)) + val schema = new StructType().add("a", ArrayType(inner)) + assert(SchemaInferenceUtils.diffSchemas(schema, schema).isEmpty) + } + + test("diffSchemas - non-type changes on nested elements are supported") { + // Due to SPARK-59188, type changes on nested array/map elements are unsupported. But other + // schema changes like nullability are still supported. + val currentSchema = new StructType() + .add("a", ArrayType(ArrayType(IntegerType), containsNull = false)) + val targetSchema = new StructType() + .add("a", ArrayType(ArrayType(IntegerType), containsNull = true)) + + val changes = SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema) + assert(changes.length === 1, s"changes=$changes") + assert(nullabilityUpdatesOf(changes) === Map(Seq("a", "element") -> true)) + } + + test("diffSchemas - tightening nullability throws") { + val currentSchema = new StructType() + .add("a", IntegerType, nullable = true) + val targetSchema = new StructType() + .add("a", IntegerType, nullable = false) + + checkError( + intercept[SparkUnsupportedOperationException]( + SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema)), + condition = "PIPELINE_TIGHTEN_NULLABILITY_UNSUPPORTED", + parameters = Map("columnPath" -> "a")) + } + + test("diffSchemas - tightening nullability inside a nested struct throws") { + val currentSchema = new StructType() + .add("s", new StructType().add("x", IntegerType, nullable = true)) + val targetSchema = new StructType() + .add("s", new StructType().add("x", IntegerType, nullable = false)) + + checkError( + intercept[SparkUnsupportedOperationException]( + SchemaInferenceUtils.diffSchemas(currentSchema, targetSchema)), + condition = "PIPELINE_TIGHTEN_NULLABILITY_UNSUPPORTED", + parameters = Map("columnPath" -> "s.x")) + } + test("inferSchemaFromFlows folds a case-only column to the same spelling regardless of flow " + "order, even when identifier names contain dots") { // The merge order decides which spelling of a case-only-differing column survives, so it must