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 fd60684150742..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,8 @@ 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.{ caseInsensitiveResolution, @@ -34,10 +36,13 @@ 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, Metadata, MetadataBuilder, + StructField, StructType +} -object SchemaInferenceUtils { +object SchemaInferenceUtils extends Logging { def resolverFor(caseSensitive: Boolean): Resolver = { if (caseSensitive) { @@ -207,12 +212,12 @@ 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 @@ -227,59 +232,234 @@ 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( + (pathToStruct :+ fieldInTarget.name).toArray, + fieldInTarget.dataType, + fieldInTarget.nullable, + 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( + (pathToStruct :+ fieldInCurrent.name).toArray, + 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 to delete (in current but not in target) - val columnsToDelete = currentFields.keySet.diff(targetFields.keySet) - columnsToDelete.foreach { columnName => - changes += TableChange.deleteColumn(Array(columnName), false) + columnsAdded ++ columnsDeleted ++ columnsUpdated + } + + /** + * 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}") } + } - // 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) + private def stripMetadataComment(m: Metadata): Metadata = + new MetadataBuilder().withMetadata(m).remove("comment").build() - // If data types are different, add a type update change - if (currentField.dataType != targetField.dataType) { - changes += TableChange.updateColumnType(Array(columnName), targetField.dataType) - } + private def diffNullability( + currentNullable: Boolean, + 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) + ) + } - // If nullability is different, add a nullability update change - if (currentField.nullable != targetField.nullable) { - changes += TableChange.updateColumnNullability(Array(columnName), targetField.nullable) - } + 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. + * + * 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 { + case (currentStruct: StructType, targetStruct: StructType) => + diffStructs(currentStruct, targetStruct, pathToField) + + case (currentArray: ArrayType, targetArray: ArrayType) => + val elementPath = pathToField :+ "element" + 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 - // 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) + case (currentMap: MapType, targetMap: MapType) => + val keyPath = pathToField :+ "key" + 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 + + 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 _ => } } - - changes.toSeq } } 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..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,10 +19,16 @@ 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.TableChange +import org.apache.spark.sql.connector.catalog.{ + CatalogV2Util, + Identifier, + InMemoryTableCatalog, + TableChange, + TableInfo +} import org.apache.spark.sql.pipelines.graph.{ FlowFunction, FlowFunctionResult, @@ -35,8 +41,10 @@ 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._ /** A [[FlowFunction]] that throws if invoked; the inferSchemaFromFlows test builds resolved * flows directly. */ @@ -164,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 @@ -207,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") @@ -222,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 @@ -247,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) } @@ -395,19 +401,19 @@ 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 " + - "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 @@ -438,22 +444,386 @@ 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])) } } + 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 = false, "old")) + val targetSchema = new StructType() + .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") -> true)) + 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("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("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 @@ -483,3 +853,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 +}