[SPARK-57152][SDP] Implement SCD2 Batch Processor; Find Affected Aux/Target Table Rows by AnishMahto · Pull Request #56283 · apache/spark

AnishMahto · 2026-06-02T17:52:36Z

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

What changes were proposed in this pull request?

Preamble:

The SCD type 2 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 SCD2 replication semantics.

SCD2 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.

Find Affected Aux/Target Table Rows

After preprocessing the microbatch such that we have each incoming row's startAt, endAt, and recordStartAt projected, the next step in reconciliation is determining which existing rows in the auxiliary and target tables either might be affected by the incoming rows or they might affect the incoming rows themselves.

A no-op upsert run row in the auxiliary table can be affected by the microbatch if an incoming row makes the row no longer a no-op (i.e microbatch delivers an interleaving row that does indeed change history tracked columns). A tombstone in the auxiliary table can affect an incoming row if it now matches against an upsert in the microbatch.

A row in the target table can be affected by the microbatch if an incoming upsert makes the target table's row a no-op upsert, or an incoming delete/upsert event terminates an existing row in the target table. An active row (endAt=null) in the target table could become terminated, or an existing closed row in the target table could become bisected. Conversely, existing rows in the target table can dictate when an incoming upsert row should be considered closed from.

We take a practical, conservative approach in selecting the set of rows that could possibly be affected or affect the microbatch. Per key we retrieve all existing rows whose startAt comes after the youngest sequence in the incoming microbatch, as well as the first existing row in both the target/aux that comes before the youngest sequence.

This is opposed to doing a very complex and expensive join to determine which rows are definitively affected by/affecting the microbatch. In practice its not common for events to actually receive very old events out of order, so pulling in all existing rows that come after the oldest row in the microbatch will generally be a very small result set.

Why are the changes needed?

AutoCDC SCD2 core algorithm.

Does this PR introduce any user-facing change?

No, new feature.

How was this patch tested?

Unit tested in Scd2BatchProcessorSuite.

Was this patch authored or co-authored using generative AI tooling?

Co-authored with Claude Opus 4.7.

…2-SCD2-find-affected-rows

AnishMahto · 2026-06-03T17:40:15Z

+   * @param microbatchDf
+   *   the incoming CDC microbatch.
+   * @return
+   *   a dataframe that retains every input row 1:1 - no rows added, dropped, reordered, or
+   *   merged - with the following schema, in column order:
+   *     1. The user columns of `microbatchDf` that survive [[ChangeArgs.columnSelection]], in
+   *        the order they appeared in the input.
+   *     2. [[startAtColName]], populated with the sequence value of the row.
+   *     3. [[endAtColName]], populated with the sequence value of the row IFF it's a delete
+   *        event, null otherwise.
+   *     4. [[cdcMetadataColName]], conforming to [[targetCdcMetadataColSchema]].
+   */


An aside on my intentions with these scaladocs here and going forward:

I want each individual, unit-testable, function in SCD2 microbatch processor to have a well defined contracts; it should be clear exactly what dataframe each function expects and what dataframe each function returns, including the schema and logical invariants of each.

I'm a big believer in correctness by construction, I would love for there to be a way to encode this information into scala types which are both self-documenting and compile-time verifiable.

But we only actually know the user's key/data columns at runtime, so we can't leverage the typed Dataset API in a meaningful way here.

At best we could introduce case-class wrappers around the DataFrame that communicate intent/invariants of the wrapped DataFrame via the name. But that can't enforce someone doesn't incorrectly construct a wrapped DataFrame (ex. one the violates the declared invariants), so it doesn't buy us much more than just documenting like this.

AnishMahto · 2026-06-03T17:41:45Z

@jose-torres Ready for review, use the stacked diff link!

Similar to the other PR, 80% of the diff here is just tests+documentation. Real logic added is less than 200 LOC.

jose-torres · 2026-06-04T22:05:15Z

+   *   particular, non-null key columns and a non-null [[recordStartAtFieldName]] on every row.
+   * @return
+   *   a dataframe containing one row per distinct key. Schema, in column order:
+   *     1. The key columns ([[ChangeArgs.keys]]), in their declared order.


The order they're declared in the change args or the order they're declared during DF construction? (The former seems to be what's implemented which is OK by me)

Honestly now that I think about it, I don't want this function to provide any contract on column ordering. As far as microbatch reconciliation is concerned, this function needs to return a dataframe with some expected set of columns, but downstream callers shouldn't need to make any assumptions about the order that the columns are returned in. Dropped the column order comment, so contractually the returned column order is undefined.

It is deterministic which allows us to materialize its dataframe asSeq[Row] and do equality comparisons during tests, but production callers shouldn't even have to assume that (and instead do comparisons by column name, never by column index).

We can make an argument that tests should also do column based equality comparisons for full safety, but I think its cheap enough to just update the expected Rows in tests if the auxiliary schema changes in the future. In exchange we get simple readability.

jose-torres · 2026-06-04T22:09:26Z

+
+    val reducedAuxiliaryTableDf = rawAuxiliaryTableDf
+      .filter(
+        // Ignore any auxiliary table rows logically deleted by any microbatch other than this one


There's a code smell here, although maybe a benign one. If it's an idempotency key why don't we need to check it for all auxiliary table rows rather than just deleted ones?

So I think idempotency key might be the wrong terminology, I reworded the comment - let me know if it makes more sense now.

Hopefully its more clear now, but we are checking the deletedByBatchId for all auxiliary rows. Btw deletedByBatchId here doesn't have anything to do with deletion CDC events specifically, delete in this context means a row has been logically deleted from the table via MERGE.

In the aux table, there are both tombstones (deletion events) and no-op upserts; both can be deleted via MERGE in later reconciliation, at which point they will first be soft/logically deleted (non-null deletedByBatchId) before they can eventually [and safely] be hard deleted in a future microbatch.

jose-torres · 2026-06-04T22:13:24Z

+    // per key, which can be determined entirely by the row's [[endAtColName]].
+    val isCurrentlyActiveRow = targetEndAtCol.isNull
+
+    // `>=` (rather than strict `>`) additionally pulls in the row that closes exactly at the


I think it's possible to have multiple rows here given the duplicate discussion from the prior PR if multiple updates at exactly targetEndAtCol have been ingested. Maybe that's wrong and the dedup is implicit elsewhere here?

So this should actually never actually be possible by construction if we implement SCD2 correctly E2E.

In SCD2 the target table should never have records that overlap by [startAt, endAt) intervals for any given key. The input here should be the raw target table exactly as-is, which means there can never be rows with duplicate targetEndAtCol.

A large part of how we make this target-table invariant true by construction will come in following PRs, ex. we do deduplication post-decomposition here.

Ok. I'm a bit torn on this, because ideally it'd be best to avoid these action at a distance assumptions, but I don't see a way to achieve that without adding extra pointless (and potentially costly) structure to the query just to check for what's ultimately an edge case of an edge case. I suppose ultimately the fuzz testing would catch any violation of this; when we have more of the implementation, I would also consider adding a test for ingestion against a target table that does have overlapping records just to capture the behavior in case something crazy happens.

I actually agree, I think we should explicitly validate any required invariants at runtime, if we cannot make them true by construction at compile time.

In fact in the next PR, I add validation post-decomposition to assert all the rows are well-formed as per recordStartAt/startAt/endAt expected invariants. I suspect we'll have this discussion there too on whether the validation <=> complexity tradeoff is worth it, personally I think it is.

I would be happy to add similar validation when we pull in aux/target table rows to validate each row conforms to the expected shapes.

There could be some middle ground too where validations are gated by some test/experimental build only flag.

jose-torres · 2026-06-08T05:52:59Z

+    // per key, which can be determined entirely by the row's [[endAtColName]].
+    val isCurrentlyActiveRow = targetEndAtCol.isNull
+
+    // `>=` (rather than strict `>`) additionally pulls in the row that closes exactly at the


Ok. I'm a bit torn on this, because ideally it'd be best to avoid these action at a distance assumptions, but I don't see a way to achieve that without adding extra pointless (and potentially costly) structure to the query just to check for what's ultimately an edge case of an edge case. I suppose ultimately the fuzz testing would catch any violation of this; when we have more of the implementation, I would also consider adding a test for ingestion against a target table that does have overlapping records just to capture the behavior in case something crazy happens.

AnishMahto · 2026-06-08T17:56:57Z

cc @szehon-ho for optional pass on reviewing. I'll fix the linting errors soon.

AnishMahto · 2026-06-08T18:31:10Z

cc @hhhhhazelnut for optional review too

szehon-ho

LGTM. The findAffected* decomposition is clean and the test coverage (anchor selection, idempotency-filter-before-aggregation ordering, composite/dotted keys, empty/missing-key cases) is excellent. One tiny scaladoc grammar nit inline, non-blocking.

AnishMahto · 2026-06-09T18:34:03Z

Failures to org.apache.spark.sql.pipelines.graph.AutoCdcScd1FullRefreshSuite are not actually related to these changes.

There's a couple parallel PRs out to address that breakage, which is first caused by #56121.

All other jobs are passing.

jose-torres · 2026-06-10T06:40:53Z

Reran the tests but I guess the underlying issue isn't fixed yet. I'm going to merge, since this change is compact enough to be confident it's not related, but since it's in the same component please do make sure we get it resolved before launching too much additional work.

…Target Table Rows Approved AutoCDC SPIP: https://lists.apache.org/thread/j6sj9wo9odgdpgzlxtvhoy7szs0jplf7 -------- ### What changes were proposed in this pull request? **Preamble**: The SCD type 2 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 SCD2 replication semantics. SCD2 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. **Find Affected Aux/Target Table Rows** After preprocessing the microbatch such that we have each incoming row's startAt, endAt, and recordStartAt projected, the next step in reconciliation is determining which existing rows in the auxiliary and target tables either might be affected by the incoming rows or they might affect the incoming rows themselves. A no-op upsert run row in the auxiliary table can be affected by the microbatch if an incoming row makes the row no longer a no-op (i.e microbatch delivers an interleaving row that does indeed change history tracked columns). A tombstone in the auxiliary table can affect an incoming row if it now matches against an upsert in the microbatch. A row in the target table can be affected by the microbatch if an incoming upsert makes the target table's row a no-op upsert, or an incoming delete/upsert event terminates an existing row in the target table. An active row (endAt=null) in the target table could become terminated, or an existing closed row in the target table could become bisected. Conversely, existing rows in the target table can dictate when an incoming upsert row should be considered closed from. We take a practical, conservative approach in selecting the set of rows that could possibly be affected or affect the microbatch. Per key we retrieve all existing rows whose startAt comes after the youngest sequence in the incoming microbatch, as well as the first existing row in both the target/aux that comes before the youngest sequence. This is opposed to doing a very complex and expensive join to determine which rows are definitively affected by/affecting the microbatch. In practice its not common for events to actually receive very old events out of order, so pulling in all existing rows that come after the oldest row in the microbatch will generally be a very small result set. ### Why are the changes needed? AutoCDC SCD2 core algorithm. ### Does this PR introduce _any_ user-facing change? No, new feature. ### How was this patch tested? Unit tested in `Scd2BatchProcessorSuite`. ### Was this patch authored or co-authored using generative AI tooling? Co-authored with Claude Opus 4.7. Closes #56283 from AnishMahto/SPARK-57152-SCD2-find-affected-rows. Authored-by: AnishMahto <anish.mahto99@gmail.com> Signed-off-by: Jose Torres <jtorres@apache.org> (cherry picked from commit 62ae4db) Signed-off-by: Jose Torres <jtorres@apache.org>

jose-torres · 2026-06-10T06:46:03Z

merged to master and 4.x

AnishMahto added 14 commits June 2, 2026 16:17

SCD2 preprocess microbatch

f579359

add scaladocs

3a34e55

fixing indenting

b77e557

document decomposition tail concept

437bfa4

document event deduplication responisbilities and add test

75adaa5

core implementation

35b9d51

tests

95d7280

some test cleanup

841bcd9

cleanup tests

53a3c5d

cleanup

7e28a20

PR feedbavk

8fbca9f

link TODO

f2f6282

Merge branch 'SPARK-57134-SCD2-preprocess-microbatch' into SPARK-5715…

d7cc05d

…2-SCD2-find-affected-rows

test cleanup

8893202

AnishMahto commented Jun 3, 2026

View reviewed changes

AnishMahto added 2 commits June 4, 2026 16:37

lift deletedByBatchId into its own column

33892d6

self-review and cleanup

4f8e5e1

jose-torres reviewed Jun 4, 2026

View reviewed changes

AnishMahto added 2 commits June 4, 2026 22:20

Merge branch 'master' into SPARK-57152-SCD2-find-affected-rows

e441d38

PR feedback

b5abfad

AnishMahto requested a review from jose-torres June 5, 2026 18:01

jose-torres approved these changes Jun 8, 2026

View reviewed changes

szehon-ho approved these changes Jun 8, 2026

View reviewed changes

Comment thread sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala Outdated

AnishMahto added 2 commits June 8, 2026 22:33

whitespace linting

d12eefb

scaladoc phrasing

84011d2

jose-torres closed this in 62ae4db Jun 10, 2026

Conversation

AnishMahto commented Jun 2, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Was this patch authored or co-authored using generative AI tooling?

Uh oh!

Choose a reason for hiding this comment

Uh oh!

AnishMahto commented Jun 3, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

AnishMahto Jun 5, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

AnishMahto Jun 8, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

AnishMahto commented Jun 8, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

AnishMahto commented Jun 8, 2026

Uh oh!

szehon-ho left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

AnishMahto commented Jun 9, 2026

Uh oh!

jose-torres commented Jun 10, 2026

Uh oh!

jose-torres commented Jun 10, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

AnishMahto commented Jun 2, 2026 •

edited

Loading

AnishMahto Jun 5, 2026 •

edited

Loading

AnishMahto Jun 8, 2026 •

edited

Loading

AnishMahto commented Jun 8, 2026 •

edited

Loading