Skip to content

[SPARK-56249][SDP] Implement SCD1 Batch Processor; Merge Tombstones onto Microbatch#55993

Closed
AnishMahto wants to merge 3 commits into
apache:masterfrom
AnishMahto:SPARK-56249-merge-tombstones-onto-microbatch
Closed

[SPARK-56249][SDP] Implement SCD1 Batch Processor; Merge Tombstones onto Microbatch#55993
AnishMahto wants to merge 3 commits into
apache:masterfrom
AnishMahto:SPARK-56249-merge-tombstones-onto-microbatch

Conversation

@AnishMahto
Copy link
Copy Markdown
Contributor

@AnishMahto AnishMahto commented May 19, 2026

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.

Merge Tombstones onto Microbatch:

The auxiliary table produced by an SCD1 flow will [strictly] store tombstones accumulated from the flow's change data feed source thus far.

In SCD1, a tombstone is defined as a delete event that has not been overtaken by any upsert event so far (an upsert event whose sequence is geq to the delete event's sequence).

These events/rows are called tombstones because they represent delete events that could still be relevant in closing a late-arriving upsert received in future microbatches. But we cannot store this type of row in the target table, as it would break the contract of what rows an SCD1 compliant replica table contains - hence these tombstones are stored in the auxiliary table.

When a new microbatch is processed, its possible it contains said late-arriving upsert events that should be swallowed by some known tombstone(s). We need to left anti join the incoming microbatch with the auxiliary table on tombstones that do indeed match to the microbatch's late-arriving upserts.

Copy link
Copy Markdown
Member

@szehon-ho szehon-ho left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this looks good to me.

Questions, would we have validation to protect against corrupt cases like:

  • deleteSequence = null
  • multiple tombstones for the same key
  • updateSequence and deleteSequence set

or do we have fallback behavior? Or just its an assumption these dont happen?

Should we also add a test that verify that empty auxiliary has no-op?

@AnishMahto
Copy link
Copy Markdown
Contributor Author

The assumption (or more specifically, these are invariants given how auxiliary table and CDC metadata column projection are computed) is none of these will never happen.

If a user manually alters these values in the auxiliary/target table to break the invariants, then correct onus falls on them.

That being said the implementation is robust OOTB for all of these cases (ex. merging tombstones onto microbatch via left-anti join doesn't care if there are multiple tombstones per key, behavior is preserved, deleteSequence=null tombstones are ignored because of null safe equality checks, etc.).

I'm going to go ahead and lock in the behavior for these fringe cases in tests though, they'll be nice stop catches in case future refactorings change the assumptions.

@AnishMahto AnishMahto force-pushed the SPARK-56249-merge-tombstones-onto-microbatch branch from 2f70d6a to 49c6d4b Compare May 21, 2026 22:14
@szehon-ho
Copy link
Copy Markdown
Member

Great, yea i was also thinking it'd be nice to add tests and wanted to ask you to if its valid , thanks!

@AnishMahto AnishMahto force-pushed the SPARK-56249-merge-tombstones-onto-microbatch branch from 49c6d4b to 849d28b Compare May 22, 2026 18:32
@AnishMahto AnishMahto force-pushed the SPARK-56249-merge-tombstones-onto-microbatch branch from 849d28b to 1e1027b Compare May 22, 2026 20:46
@dbtsai dbtsai closed this in 2a78a09 May 22, 2026
dbtsai pushed a commit that referenced this pull request May 22, 2026
…nto Microbatch

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.

**Merge Tombstones onto Microbatch:**

The auxiliary table produced by an SCD1 flow will [strictly] store tombstones accumulated from the flow's change data feed source thus far.

In SCD1, a tombstone is defined as a delete event that has not been overtaken by any upsert event so far (an upsert event whose sequence is geq to the delete event's sequence).

These events/rows are called tombstones because they represent delete events that could still be relevant in closing a late-arriving upsert received in future microbatches. But we cannot store this type of row in the target table, as it would break the contract of what rows an SCD1 compliant replica table contains - hence these tombstones are stored in the auxiliary table.

When a new microbatch is processed, its possible it contains said late-arriving upsert events that should be swallowed by some known tombstone(s). We need to left anti join the incoming microbatch with the auxiliary table on tombstones that do indeed match to the microbatch's late-arriving upserts.

Closes #55993 from AnishMahto/SPARK-56249-merge-tombstones-onto-microbatch.

Authored-by: AnishMahto <anish.mahto99@gmail.com>
Signed-off-by: DB Tsai <dbtsai@dbtsai.com>
(cherry picked from commit 2a78a09)
Signed-off-by: DB Tsai <dbtsai@dbtsai.com>
dbtsai pushed a commit that referenced this pull request May 22, 2026
…nto Microbatch

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.

**Merge Tombstones onto Microbatch:**

The auxiliary table produced by an SCD1 flow will [strictly] store tombstones accumulated from the flow's change data feed source thus far.

In SCD1, a tombstone is defined as a delete event that has not been overtaken by any upsert event so far (an upsert event whose sequence is geq to the delete event's sequence).

These events/rows are called tombstones because they represent delete events that could still be relevant in closing a late-arriving upsert received in future microbatches. But we cannot store this type of row in the target table, as it would break the contract of what rows an SCD1 compliant replica table contains - hence these tombstones are stored in the auxiliary table.

When a new microbatch is processed, its possible it contains said late-arriving upsert events that should be swallowed by some known tombstone(s). We need to left anti join the incoming microbatch with the auxiliary table on tombstones that do indeed match to the microbatch's late-arriving upserts.

Closes #55993 from AnishMahto/SPARK-56249-merge-tombstones-onto-microbatch.

Authored-by: AnishMahto <anish.mahto99@gmail.com>
Signed-off-by: DB Tsai <dbtsai@dbtsai.com>
(cherry picked from commit 2a78a09)
Signed-off-by: DB Tsai <dbtsai@dbtsai.com>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants