[SPARK-59164][SDP] Support nested column schema evolution - #58465
[SPARK-59164][SDP] Support nested column schema evolution#58465AnishMahto wants to merge 6 commits into
Conversation
|
@szehon-ho Mind taking a look for review? |
| case (currentStruct: StructType, targetStruct: StructType) => | ||
| diffStructs(currentStruct, targetStruct, pathToField) | ||
|
|
||
| case (currentArray: ArrayType, targetArray: ArrayType) => |
There was a problem hiding this comment.
Btw turns out DSv2 doesn't support TableChanges to an array or map nested into another array/map. But I believe this is a bug with DSv2 and not how we construct the nested path here.
Filed a spark issue at https://issues.apache.org/jira/browse/SPARK-59188
| * struct, if this is a nested struct. Empty for the | ||
| * root call. | ||
| */ | ||
| private def diffStructs( |
There was a problem hiding this comment.
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.
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.
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.
22a29af to
832410d
Compare
| case (currentStruct: StructType, targetStruct: StructType) => | ||
| diffStructs(currentStruct, targetStruct, pathToField) | ||
|
|
||
| case (currentArray: ArrayType, targetArray: ArrayType) => |
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.
There was a problem hiding this comment.
Chose to be explicit and reject via the PIPELINE_NESTED_COMPLEX_TYPE_SCHEMA_EVOLUTION_UNSUPPORTED exception.
Btw I just tested on Spark 4.1 and the Iceberg catalog + connector locally to confirm, all nested schema changes threw an IllegalArgumentException: Cannot update '<col>', not a primitive type: <type>, not just an array/map nested in another array/map.
So although SDP wasn't throwing in SchemaInferenceUtils specifically before (and emitting an UpdateColumnType instead), a production catalog would throw - and throw for a much wider range of schemas.
| TableChange.addColumn( | ||
| (pathToStruct :+ fieldInTarget.name).toArray, | ||
| fieldInTarget.dataType, | ||
| fieldInTarget.nullable, |
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?
There was a problem hiding this comment.
Added explicit validation that nullability may only be widening. Throws PIPELINE_TIGHTEN_NULLABILITY_UNSUPPORTED otherwise.
2 participants
What changes were proposed in this pull request?
SchemaInferenceUtils.diffSchemasis used to deduce theTableChangesthat need to be sent to the catalog in order to reflect the latest schema evolved state of datasets declared by a pipeline, during dataset materialization.Today however, the diff does not support nested column level evolution. If some nested column (ex. struct type) has one of its sub-fields changed between runs, but the column's top level name stays the same, it does not emit the correct (or any) column add or delete changes to the catalog.
The fix is to recursively diff the schema, traversing every nested column as necessary. This is similar to the existing
ResolveSchemaEvolution.computeSchemaChanges, except it is not just additive schema evolution - it supports dropping columns in the latest dataset's schema declaration.Two things intentionally not handled by this change:
metadata. That was not respected before, and continues to not be respected after these changes, because DSv2 does not yet supportTableChangestypes for changes to column metadataWhy are the changes needed?
Support nested schema evolution in SDP.
Does this PR introduce any user-facing change?
Yes. When a user changes the declared schema for a nested column in their pipeline datasets, the schema changes will actually be propagated to the catalog as a set of
TableChanges.How was this patch tested?
Unit tests.
Was this patch authored or co-authored using generative AI tooling?
Yes, using Claude Opus 4.6.