[SPARK-58118][SDP] Allow user-specified AUTO CDC schema to omit the reserved metadata column - #57644
[SPARK-58118][SDP] Allow user-specified AUTO CDC schema to omit the reserved metadata column#57644naveenp2708 wants to merge 2 commits into
Conversation
…eserved metadata column ### What changes were proposed in this pull request? An AUTO CDC (SCD1) flow appends a reserved metadata column (__spark_autocdc_metadata) to its output schema. Because validateUserSpecifiedSchemas required the declared schema to exactly equal the inferred schema, a user declaring a schema on an AUTO CDC streaming table was forced to also declare this engine-internal column. This makes two changes so the user schema can describe just the logical data columns: 1. GraphValidations.validateUserSpecifiedSchemas accepts a declared schema that differs from the inferred schema only by the reserved AUTO CDC metadata column(s). Any other mismatch still fails, and a schema that does declare the reserved column stays accepted. 2. DatasetManager.materializeTable appends the reserved metadata column(s) to the created table's schema when a user schema omits them, so the SCD1 MERGE resolves them at runtime. ### Why are the changes needed? The reserved column is an engine implementation detail; users should not need to know its name or shape. Relaxing validation alone is insufficient because materializeTable uses the user schema verbatim, so the target table would be created without the column and the MERGE would fail at runtime with UNRESOLVED_COLUMN. ### Does this PR introduce any user-facing change? Yes. A user-specified schema on an AUTO CDC streaming table may now omit the reserved __spark_autocdc_metadata column. ### How was this patch tested? UserSpecifiedSchemaValidationSuite: data-only schema accepted (implicit and named flows), wrong data columns still rejected, declaring the metadata column still accepted. ### Was this patch authored or co-authored using generative AI tooling? Yes.
HyukjinKwon
left a comment
HyukjinKwon
left a comment
There was a problem hiding this comment.
0 blocking, 0 non-blocking, 1 nit.
Clean UX fix that stops a leaky engine-internal column from being required; one optional micro-nit.
Nits: 1 minor item (see inline comments).
Verification
Validation and materialization use the same AutoCdcMergeFlow reserved-field helpers (reservedFields/stripReservedFields on AutoCdcReservedNames.prefix, exact complements), so the accepted schema and the materialized table agree; stripReservedFields only removes reserved fields, so a genuine logical-column mismatch still fails validation, and filterNot prevents duplication if the user did declare the reserved column.
| // so the created table matches what the AUTO CDC MERGE writes at runtime. | ||
| val omittedReservedFields = AutoCdcMergeFlow | ||
| .reservedFields(resolvedDataflowGraph.inferredSchema(table.identifier)) | ||
| .filterNot(f => ss.fieldNames.contains(f.name)) |
There was a problem hiding this comment.
Optional micro-nit: ss.fieldNames is recomputed (fresh Array) with a linear contains per reserved field. val specifiedNames = ss.fieldNames.toSet before the filterNot makes membership O(1). Materialization-time only, so take-it-or-leave-it.
There was a problem hiding this comment.
Good call @HyukjinKwon , done. Switched ss.fieldNames to a Set before the filterNot.
2 participants
What changes were proposed in this pull request?
An AUTO CDC (SCD1) flow appends a reserved metadata column (
__spark_autocdc_metadata) to its output schema. BecausevalidateUserSpecifiedSchemasrequired the declared schema to exactly equal the inferred schema, a user declaring a schema on an AUTO CDC streaming table was forced to also declare this engine-internal column. This makes two changes so the user schema can describe just the logical data columns:GraphValidations.validateUserSpecifiedSchemasaccepts a declared schema that differs from the inferred schema only by the reserved AUTO CDC metadata column(s). Any other mismatch still fails, and a schema that does declare the reserved column stays accepted.DatasetManager.materializeTableappends the reserved metadata column(s) to the created table's schema when a user schema omits them, so the SCD1 MERGE resolves them at runtime.Why are the changes needed?
The reserved column is an engine implementation detail; users should not need to know its name or shape. Relaxing validation alone is insufficient because
materializeTableuses the user schema verbatim, so the target table would be created without the column and the MERGE would fail at runtime withUNRESOLVED_COLUMN.Does this PR introduce any user-facing change?
Yes. A user-specified schema on an AUTO CDC streaming table may now omit the reserved
__spark_autocdc_metadatacolumn.How was this patch tested?
UserSpecifiedSchemaValidationSuite: data-only schema accepted (implicit and named flows), wrong data columns still rejected, declaring the metadata column still accepted.Was this patch authored or co-authored using generative AI tooling?
No