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

[AnishMahto]

Approved AutoCDC SPIP: https://lists.apache.org/thread/j6sj9wo9odgdpgzlxtvhoy7szs0jplf7

---

This is a stacked PR. Review incremental diff here: AnishMahto/spark@SPARK-56882-SCD1-project-target-columns-onto-microbatch...SPARK-56249-merge-tombstones-onto-microbatch

Link to previous PR: #55991

---

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.

[szehon-ho]

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]

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.

[szehon-ho]

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