[SPARK-59164][SDP] Support nested column schema evolution #58465
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -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.{ | ||
| ArrayType, DataType, MapType, Metadata, MetadataBuilder, | ||
| StructField, StructType | ||
| } | ||
|
|
||
|
|
||
| object SchemaInferenceUtils extends Logging { | ||
|
|
||
| def resolverFor(caseSensitive: Boolean): Resolver = { | ||
| if (caseSensitive) { | ||
|
|
@@ -207,12 +212,12 @@ object SchemaInferenceUtils { | |
| } | ||
|
|
||
| /** | ||
| * 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 | ||
|
|
@@ -225,61 +230,252 @@ object SchemaInferenceUtils { | |
| * | ||
| * @param currentSchema The current schema of the table | ||
| * @param targetSchema The target schema that we want the table to have | ||
| * @param rejectNullabilityTightening When true, throws if any field changes | ||
| * from nullable to non-nullable. Callers should set this when existing | ||
| * rows are retained (incremental streaming tables) because those rows | ||
| * may already contain nulls. Full-refresh and materialized-view paths | ||
| * truncate the table first, so tightening is safe. | ||
| * @return A sequence of TableChange objects representing the necessary changes | ||
| */ | ||
| def diffSchemas( | ||
| currentSchema: StructType, | ||
| targetSchema: StructType, | ||
| rejectNullabilityTightening: Boolean = false | ||
| ): Seq[TableChange] = { | ||
| val changes = diffStructs( | ||
| currentStruct = currentSchema, | ||
| targetStruct = targetSchema, | ||
| // Root call: path is empty because current and target are the top-level schemas. | ||
| pathToStruct = Seq.empty | ||
| ) | ||
| if (rejectNullabilityTightening) { | ||
| changes.foreach { | ||
| case nc: TableChange.UpdateColumnNullability if !nc.nullable() => | ||
| throw new SparkUnsupportedOperationException( | ||
| errorClass = "PIPELINE_TIGHTEN_NULLABILITY_UNSUPPORTED", | ||
| messageParameters = | ||
| Map("columnPath" -> nc.fieldNames().mkString("."))) | ||
| case _ => | ||
| } | ||
| } | ||
| changes | ||
| } | ||
|
|
||
| /** | ||
| * 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 | ||
|
|
||
| // 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, | ||
|
Member
There was a problem hiding this comment. Should newly added fields be forced nullable when existing rows are retained? On the incremental streaming-table path, a new nested field has no value in old rows. Preserving nullable = false here can either make the catalog reject evolution or expose null values under a non-null schema. ResolveSchemaEvolution handles this by adding missing fields as nullable and making their nested data types nullable as well. The new test that expects false appears to enshrine the unsafe contract; could the additive evolution path emit a nullable field instead?
Contributor
Author
There was a problem hiding this comment. Added explicit validation that nullability may only be widening. Throws
Member
There was a problem hiding this comment. Thanks, but I do not think this validation covers the scenario in the original comment. diffNullability is only called for fields present in both schemas. A newly added field takes the columnsAdded path above and still passes fieldInTarget.nullable directly to TableChange.addColumn; the test for an added nested leaf still expects false. On an incremental table, old rows have no value for that field. Could we make added fields nullable on the merge/incremental path, matching ResolveSchemaEvolution?
Contributor
Author
There was a problem hiding this comment. Gotcha, misunderstood previously. This got me thinking, both for the tightening nullability validation and for the required nullability on column add restriction - should SDP really even govern these things? I'm actually thinking we should pass through what the user declares as-is, and let the catalog/connector throw if the operation is actually unsupported. One reason why is because it's difficult to map out all of the restrictions correctly. Ex. you caught that tightening nullability is allowed on non-incremental updates, and adding a non-nullable column might actually be supported in some catalogs as long as the default value is set too? WDYT about:
Member
There was a problem hiding this comment. Thats a good idea, it actually matches what we do in DML statement (ie MERGE INTO, INSERT INTO) schema evolution, pass to Catalog and let it decide. The only thing I validate explicitly for is some basic things like no complex <=> primitive type. We should keep nullability and all that. |
||
| fieldInTarget.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 | ||
| ) | ||
| } | ||
| } | ||
|
|
||
| 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}") | ||
| } | ||
| } | ||
|
|
||
| private def stripMetadataComment(m: Metadata): Metadata = | ||
| new MetadataBuilder().withMetadata(m).remove("comment").build() | ||
|
|
||
| private def diffNullability( | ||
| currentNullable: Boolean, | ||
| targetNullable: Boolean, | ||
| pathToField: Seq[String] | ||
| ): Option[TableChange] = { | ||
| Option.when(currentNullable != targetNullable)( | ||
| TableChange.updateColumnNullability(pathToField.toArray, targetNullable) | ||
| ) | ||
| } | ||
|
|
||
| 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) => | ||
|
Contributor
Author
There was a problem hiding this comment. Btw turns out DSv2 doesn't support Filed a spark issue at https://issues.apache.org/jira/browse/SPARK-59188
Member
There was a problem hiding this comment. Following up on SPARK-59188: can we avoid recursing through arrays/maps until that issue is fixed, or reject unsupported shapes here? For example, dropping y from array<array<struct<x: int, y: int>>> emits the path [a, element, element, y], but CatalogV2Util.replace only descends through an array when its immediate element is a StructType. InMemoryTableCatalog therefore rejects this path, while the previous implementation emitted an UpdateColumnType for a and could apply it. The catalog test in this PR only covers one array/map directly wrapping a struct, so it misses this failure.
Contributor
Author
There was a problem hiding this comment. Chose to be explicit and reject via the Btw I just tested on Spark 4.1 and the Iceberg catalog + connector locally to confirm, all nested schema changes threw an So although SDP wasn't throwing in |
||
| 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 | ||
|
|
||
| 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 _ => | ||
| } | ||
| } | ||
| } | ||
| } | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
As mentioned in the PR description, field order is intentionally not considered when diffing two structs.
It's fair to argue that it should, but that would be out of scope for this PR; I'm choosing to keep previous behavior, albeit now it applies recursively to nested struct comparisons too.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Exact key matches (case-sensitive) is maintained behavior from the previous implementation too, and the scaladoc section that discusses it for incremental/full refresh cases is carried as-is.