[SPARK-59165][SDP] SCD2 Ignore-null support; Introduce version map in _cdc_metadata - #58483
[SPARK-59165][SDP] SCD2 Ignore-null support; Introduce version map in _cdc_metadata#58483AnishMahto wants to merge 10 commits into
_cdc_metadata#58483Conversation
ab72b27 to
2cb3476
Compare
2cb3476 to
ac73d9a
Compare
| private def scd2Meta(recordStartAt: Long): Row = Row(recordStartAt) | ||
| private def scd2Meta(recordStartAt: Long): Row = Row(recordStartAt, null) | ||
|
|
||
| test("legacy SCD2 CDC metadata schema evolves to include the version map") { |
There was a problem hiding this comment.
The major focus for this PR should be - what should we do with existing AutoCDC tables created before version map is added to the CDC metadata column?
What I rely on in this PR, as is illustrated by this test, is SDP + DSv2's ability to handle schema evolution for nested columns. That means existing AutoCDC tables will gain a version map sub-field in their CDC metadata column, set to null, on their next run after these changes lands.
Now what is technically not guaranteed, is whether each popular connector actually supports and respects the table alters and TableChanges we emit, via DSv2. If they don't, then those tables will be not be able to be refreshed going forward, on the first Spark version that includes these changes. But from SDP's perspective, we are fully compliant with and using DSv2 API.
Of the connectors that do support the MERGE operation (which is a prerequisite for AutoCDC anyway), I'm not aware of any that don't support nested column evolution via DSv2 - ex. both Delta and Iceberg do support nested evolution. But technically Parquet tables in HMS do not support nested schema evolution AFAIK. Now Parquet does not support MERGE either, so an AutoCDC table could never have been materialized using it.
Alternative option
If we're not willing to rely on the assumption that all AutoCDC compatible table formats and catalogs support nested schema evolution via DSv2, then another option is too just support two variants of the CDC metadata column schema going forward. Existing tables will use the schema definition that does not include the version map, and new tables will be created with the schema definition that includes the version map.
This does however mean:
- We need to support both schema variants forever going forward
- Existing AutoCDC tables are not allowed to convert to ignore-null. That also means they will never be eligible for AutoCDC multiflow, which relies on ignore-null
- We'll need to add extra logic to check the existing table's schema, and determine which schema variant its using
- The
Scd2BatchProcessorcode will have to be written in a way that supports both types of schemas
Which I am generally not a fan of, and would prefer to avoid if justifiable.
| private[pipelines] val recordStartAtFieldName: String = "__RECORD_START_AT" | ||
|
|
||
| /** CDC metadata field for the ignore-null version map. */ | ||
| private[pipelines] val versionMapFieldName: String = "__VERSION_MAP" |
There was a problem hiding this comment.
Authorship map could also be a good name instead of version map.
|
@jose-torres for review. This is a really small logical change, a bunch of tests just needed to be updated to adjust for the new Speaking of which though, there is one important migration discussion I started a thread on here that could change the approach we take: #58483 (comment). |
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.
ac73d9a to
e25ff00
Compare
e25ff00 to
ab65e68
Compare
Add a nullable __VERSION_MAP field (Map<String, Boolean>) to the SCD2 _cdc_metadata struct. This field will track per-column authorship for ignore-null semantics in a future change; for now it is always null. Source changes: - Introduce Scd2VersionMap with mapType and authorship contract docs. - Add versionMapFieldName, versionMapOf to Scd2BatchProcessor companion. - Extend cdcMetadataColSchema and constructCdcMetadataCol with the new field. All four call sites pass versionMap explicitly (no default). - Decomposition tails and tombstones always receive a null version map (synthetic rows carry no authorship claim). Test changes: - Update all Row(...) constructions for CDC metadata structs across Scd2BatchProcessorSuite, Scd2BatchProcessorMergeSuite, Scd2ForeachBatchHandlerSuite, and 7 graph execution test suites. - Fix schema-preservation tests to use the production two-field cdcMetadataInnerSchema. - Update scd2MetadataDdl in AutoCdcGraphExecutionTestMixin.
ab65e68 to
bdbe370
Compare
jose-torres
left a comment
jose-torres
left a comment
There was a problem hiding this comment.
The contents of this incremental diff make sense to me, although I'm not quite clear on what it is that it's a stack on top of - is there another PR before it I should also be looking at?
|
Just added an explicit link to the previous PR in the stack! Feel free to take a look, but I tagged @szehon-ho on that one since it's a more general SDP bug fix. It affects this PR, but is not AutoCDC specific. |
2 participants
This is a stacked PR. See incremental diff here: AnishMahto/spark@SPARK-59164-nested-schema-diff...SPARK-59165-scd2-version-map-schema
Previous PR in stack: #58465
What changes were proposed in this pull request?
In order to support ignore-null in SCD2, we need to persist additional metadata per row that tells us which column values were actually authored by the upsert event that created the row, and which columns were unauthored by the same upsert event and instead need to inherit a value from a previous row.
This PR is concerned with simply defining the version map schema/concept, and adding it to the
_cdc_metadataper row. The version map will not be used yet, but will start materializing in both old and new AutoCDC tables.API for ignore-null is not exposed yet, so a non-null version map will never be created yet.
Why are the changes needed?
To support the ignore-null feature for AutoCDC SCD2 (as per the approved SPIP), we need to introduce a new map data structure to persist column authorship. This is a per-row data structure and holds operational metadata, so it naturally fits in
_cdc_metadata.Does this PR introduce any user-facing change?
Yes. The (nullable) version map field is added to the
_cdc_metadatacolumn. That means existing AutoCDC target tables that do not contain this subfield will undergo schema evolution on their next pipeline run after this change.Nested-field schema evolution is supported in SDP as of the previous PR in this stack, and the evolution case is captured via unit tests using the in-memory DSv2 catalog.
How was this patch tested?
Unit tests.
Many existing unit test assertions on final table contents were updated to account for the (null) version map.
Was this patch authored or co-authored using generative AI tooling?
Yes, using Claude Opus 4.6.