Uh oh!
There was an error while loading. Please reload this page.
[SPARK-56870][SDP] Implement SCD1 Batch Processor; Extend Microbatch with CDC Metadata - #55970
Conversation
552e33c to
9a566ffCompare
szehon-ho
left a comment
There was a problem hiding this comment.
Review of the incremental diff on top of #55969 (extend microbatch with CDC metadata). Overall this looks good to merge with minor nits.
What looks good
- The delete/upsert encoding in
_cdc_metadatamatches the SPIP story: mutually exclusivedeleteSequence/upsertSequence, persisted beforecolumnSelectioncan dropdeleteConditioncolumns. resolvedSequencingTypeat processor construction is the right split (flow setup vs per-microbatch work); the Int→Long cast test and incompatible cast test are valuable.- Reserved-column conflict uses
conf.resolverandCaseSensitivityLabels— consistent with session case sensitivity. constructCdcMetadataColdriven offcdcMetadataColSchemawith ordered fields is clean; companion constants keep tests readable.AUTOCDC_RESERVED_COLUMN_NAME_CONFLICT/ SQLSTATE42710is appropriate.- Test coverage for classification, no delete condition, column ordering, cast success/failure, and reserved-name conflict is solid.
Incremental diff is focused and stacks cleanly on #55836 + #55969.
Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.
f9c2aed to
02473baCompare
szehon-ho
left a comment
There was a problem hiding this comment.
Re-reviewed the incremental diff on #55969. CDC metadata encoding and resolvedSequencingType casting look correct; reserved-column validation and tests LGTM.
Left three inline nits (comment wording, extend input contract, reserved-prefix scope) — all non-blocking. Approved.
Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.
Uh oh!
There was an error while loading. Please reload this page.
02473ba to
25387c3Compare…with CDC Metadata Approved AutoCDC SPIP: https://lists.apache.org/thread/j6sj9wo9odgdpgzlxtvhoy7szs0jplf7 -------- **Preamble:** The SCD type 1 flow is a foreachBatch streaming query on an input change-data-feed, and is responsible for reconciling the incoming change data onto some target table that follows SCD1 replication semantics. SCD1 flows also maintain an "auxiliary" table to keep track of early-arriving out-of-order received events state. Each microbatch will need to reconcile against this auxiliary table as well, and update the auxiliary table's state appropriately for future microbatches. **Extend Microbatch with CDC Metadata:** After deduplication, all of the incoming rows can be classified as either a delete event or an upsert event (mutually exclusive), and there's at most one per key. If we identify a row as a delete event, remember its sequencing as its `deleteSequence`. If we identify a row as an upsert event, remember its sequencing as its `upsertSequence`. That is, `deleteSequence`/`upsertSequence` encode both the sequencing for the row as well as the row classification (delete or upsert). We need to persist this encoded information now, because in future stages we may drop the columns that `deleteCondition` needed to do the classification in the first place, depending on which columns were selected by `ChangeArgs.columnSelection`. **Where is the CDC Metadata stored?** Within the microbatch, we append a `_cdc_metadata` struct column, that stores the `deleteSequence` and `upsertSequence`. This `_cdc_metadata` column will eventually also land in the persisted target and auxiliary tables, which are the artifacts of an AutoCDC flow. This column represents operational metadata that the AutoCDC flow has tagged a row with, and is necessary for out-of-order correctness of the SCD decomposition. Users will not be able to opt out of persisting this column in the target table using `ChangeArgs.columnSelection`, as it is necessary for correctness. The column will not have a stable public contract, and users should make no assumptions on its contents. Closes#55970 from AnishMahto/SPARK-56870-extend-microbatch-with-cdc-metadata. Authored-by: AnishMahto <anish.mahto99@gmail.com> Signed-off-by: DB Tsai <dbtsai@dbtsai.com> (cherry picked from commit 12807c5) Signed-off-by: DB Tsai <dbtsai@dbtsai.com>
…with CDC Metadata Approved AutoCDC SPIP: https://lists.apache.org/thread/j6sj9wo9odgdpgzlxtvhoy7szs0jplf7 -------- **Preamble:** The SCD type 1 flow is a foreachBatch streaming query on an input change-data-feed, and is responsible for reconciling the incoming change data onto some target table that follows SCD1 replication semantics. SCD1 flows also maintain an "auxiliary" table to keep track of early-arriving out-of-order received events state. Each microbatch will need to reconcile against this auxiliary table as well, and update the auxiliary table's state appropriately for future microbatches. **Extend Microbatch with CDC Metadata:** After deduplication, all of the incoming rows can be classified as either a delete event or an upsert event (mutually exclusive), and there's at most one per key. If we identify a row as a delete event, remember its sequencing as its `deleteSequence`. If we identify a row as an upsert event, remember its sequencing as its `upsertSequence`. That is, `deleteSequence`/`upsertSequence` encode both the sequencing for the row as well as the row classification (delete or upsert). We need to persist this encoded information now, because in future stages we may drop the columns that `deleteCondition` needed to do the classification in the first place, depending on which columns were selected by `ChangeArgs.columnSelection`. **Where is the CDC Metadata stored?** Within the microbatch, we append a `_cdc_metadata` struct column, that stores the `deleteSequence` and `upsertSequence`. This `_cdc_metadata` column will eventually also land in the persisted target and auxiliary tables, which are the artifacts of an AutoCDC flow. This column represents operational metadata that the AutoCDC flow has tagged a row with, and is necessary for out-of-order correctness of the SCD decomposition. Users will not be able to opt out of persisting this column in the target table using `ChangeArgs.columnSelection`, as it is necessary for correctness. The column will not have a stable public contract, and users should make no assumptions on its contents. Closes#55970 from AnishMahto/SPARK-56870-extend-microbatch-with-cdc-metadata. Authored-by: AnishMahto <anish.mahto99@gmail.com> Signed-off-by: DB Tsai <dbtsai@dbtsai.com> (cherry picked from commit 12807c5) Signed-off-by: DB Tsai <dbtsai@dbtsai.com>
dbtsai
commented
May 22, 2026
Merged into |
Approved AutoCDC SPIP: https://lists.apache.org/thread/j6sj9wo9odgdpgzlxtvhoy7szs0jplf7
Preamble:
The SCD type 1 flow is a foreachBatch streaming query on an input change-data-feed, and is responsible for reconciling the incoming change data onto some target table that follows SCD1 replication semantics.
SCD1 flows also maintain an "auxiliary" table to keep track of early-arriving out-of-order received events state. Each microbatch will need to reconcile against this auxiliary table as well, and update the auxiliary table's state appropriately for future microbatches.
Extend Microbatch with CDC Metadata:
After deduplication, all of the incoming rows can be classified as either a delete event or an upsert event (mutually exclusive), and there's at most one per key.
If we identify a row as a delete event, remember its sequencing as its
deleteSequence. If we identify a row as an upsert event, remember its sequencing as itsupsertSequence. That is,deleteSequence/upsertSequenceencode both the sequencing for the row as well as the row classification (delete or upsert).We need to persist this encoded information now, because in future stages we may drop the columns that
deleteConditionneeded to do the classification in the first place, depending on which columns were selected byChangeArgs.columnSelection.Where is the CDC Metadata stored?
Within the microbatch, we append a
_cdc_metadatastruct column, that stores thedeleteSequenceandupsertSequence.This
_cdc_metadatacolumn will eventually also land in the persisted target and auxiliary tables, which are the artifacts of an AutoCDC flow. This column represents operational metadata that the AutoCDC flow has tagged a row with, and is necessary for out-of-order correctness of the SCD decomposition.Users will not be able to opt out of persisting this column in the target table using
ChangeArgs.columnSelection, as it is necessary for correctness. The column will not have a stable public contract, and users should make no assumptions on its contents.