AnishMahto commented on code in PR #58209:
URL: https://github.com/apache/spark/pull/58209#discussion_r3832053217
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala:
##########
@@ -94,17 +94,22 @@ case class Scd2BatchProcessor(
// tail can detect its own redundancy via LEAD(1): if the next row is
a non-tail at
// the same instant, the synthetic close the tail encodes is already
represented by
// that event and the tail is dropped downstream.
- //
- // Any tiebreaking beyond this rule only meaningfully fires when the
user's source
- // has emitted two or more events at the same sequence, violating the
uniqueness
- // contract above. Behavior in that case is publicly undefined and the
remaining
- // tiebreaker clauses exist as a best-effort to keep retries and
replays deterministic.
orderDecompositionTailsFirst,
// Upsert-representing rows sort before tombstones because rows detect
if they are being
// bisected by LEAD(1). This allows upserts to match against
same-sequence deletes, an
// arbitrary but deterministic convention. When this happens, the
delete event will survive
// and persist as a tombstone in the auxiliary table.
- orderUpsertRepresentingRowsFirst
+ orderUpsertRepresentingRowsFirst,
+ // Amongst upsert-representing rows, there's one valid case where rows
are still tied, even
+ // if the user's change feed source did not emit duplicate sequences:
the auxiliary merge
+ // commits before the target merge, which re-reads that table, so a
row this batch wrote to
+ // the auxiliary table re-enters the window beside the copy the
microbatch or the target
+ // table still holds. The copies differ only in the boundaries each
one recorded, so
+ // descending order keeps the earliest of them -
dropRedundantRowsPostDecomposition drops a
+ // tie's leading row - and nulls sort first so a copy that recorded no
boundary never
+ // displaces one that did.
+ startAtCol.desc_nulls_first,
+ endAtCol.desc_nulls_first
Review Comment:
In a nutshell; in order to construct a unioned timeline for determining
affected rows, both aux and target tables need to read each other.
But merging into each table is sequential, not atomic, and the aux table is
merged into first. That means its new row-set will reflect when the target
table determines its affected rows, which could (correctly) already include
rows from the incoming microbatch, that were merged into the aux.
This is a scenario where rows with duplicate keys, sequence, and data
columns can arise, which is valid. In such a scenario the calculated intervals
for the duplicate rows will be different, and by ordering deterministically we
can consistently drop one the correct redundant row on the second merge.
--
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]
