[SPARK-59164][SDP] Support nested column schema evolution #58465
+683
−87
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| 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 | ||
|
|
@@ -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] = | ||
| diffStructs( | ||
| currentStruct = currentSchema, | ||
| targetStruct = targetSchema, | ||
| // Root call: path is empty because current and target are the top-level schemas. | ||
| pathToStruct = Seq.empty | ||
| ) | ||
|
|
||
| /** | ||
| * 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 |
||
| 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] = { | ||
| 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) | ||
| ) | ||
| } | ||
|
|
||
| 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 _ => | ||
| } | ||
| } | ||
| } | ||
| } | ||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
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.