AnishMahto opened a new pull request, #58209:
URL: https://github.com/apache/spark/pull/58209
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines:
https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR:
https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g.,
'[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a
faster review.
7. If you want to add a new configuration, please read the guideline first
for naming configurations in
'common/utils/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the
guideline first in
'common/utils/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
In order for SCD2 to reconcile a microbatch, it finds the existing rows in
the auxiliary and target tables that could be affected by said microbatch, and
pulls them in for reconciliation.
The current implementation finds the affected rows in the auxiliary table
and target table independently, by finding the first rows in each table that
independently precede the minimum sequence (per key) in the microbatch.
This is incorrect behavior as it can lead to pulling in additional rows from
the auxiliary table that are not actually affected by the microbatch. Per
reconciliation there can be at most one actual row per key that immediately
precedes the microbatch's events and therefore should be considered the
starting point for all rows that need to be pulled in. By calculating an
affected row anchor for the aux and target tables independently, up to two
anchor rows could be pulled in instead.
For any given microbatch its possible the aux table's anchor row is the one
that is actually redundant, and will be pulled in unnecessarily. Pulling in
these additional rows can actually be consequential for correctness, because
the auxiliary table holds hidden no-op upsert run rows. In several scenarios,
such upsert run rows can be then further be incorrectly promoted to run-tails
and moved to the target table.
The fix is to consider a unified or global timeline of all existing rows
across the auxiliary and target tables, before finding the single and truthful
anchor row (if one exists). That is, union the two tables before finding the
first row that immediately precedes the microbatch, rather than doing so
independently on each table.
### Why are the changes needed?
Necessary changes for correctness. Without these changes, AutoCDC flows that
use track history columns and have no-op upsert runs may reconcile into the
target table incorrectly. See regression tests added to
`Scd2ForeachBatchHandlerSuite` for concrete failure scenarios.
### Does this PR introduce _any_ user-facing change?
No, since SCD2 is not yet released.
### How was this patch tested?
Unit tests in `Scd2ForeachBatchHandlerSuite` and `Scd2BatchProcessorSuite`.
Also fixes the failures in the convergence tests introduced by
https://github.com/apache/spark/pull/58055.
### Was this patch authored or co-authored using generative AI tooling?
Yes, co-authored with Claude Opus 5 and GPT-5.6 Sol
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
