[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

---

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.