This is an automated email from the ASF dual-hosted git repository.
szehon-ho pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new b8f7cdd0c33f [SPARK-58937][SDP] Consider unified timeline when
determining affected rows for SCD2
b8f7cdd0c33f is described below
commit b8f7cdd0c33f23b879d7de1370e8dcc32c14bba4
Author: AnishMahto <[email protected]>
AuthorDate: Sat Aug 22 08:44:40 2026 -0700
[SPARK-58937][SDP] Consider unified timeline when determining affected rows
for SCD2
### 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 will then 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
Closes #58209 from
AnishMahto/SPARK-58937-find-affected-rows-on-unified-timeline.
Authored-by: AnishMahto <[email protected]>
Signed-off-by: Szehon Ho <[email protected]>
(cherry picked from commit a6108a5db606b7d5f7539da5bce688cf0c2ca129)
Signed-off-by: Szehon Ho <[email protected]>
---
.../sql/pipelines/autocdc/Scd2BatchProcessor.scala | 312 ++++++++++++---------
.../autocdc/Scd2ForeachBatchHandler.scala | 14 +-
.../autocdc/Scd2BatchProcessorSuite.scala | 310 +++++++++++---------
.../autocdc/Scd2ForeachBatchHandlerSuite.scala | 197 ++++++++++++-
4 files changed, 574 insertions(+), 259 deletions(-)
diff --git
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala
index c93460c2ad81..3f2a2e741532 100644
---
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala
+++
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala
@@ -94,17 +94,27 @@ 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, and the
+ // two keys below break that tie deterministically -
dropRedundantRowsPostDecomposition
+ // drops a tie's leading row, so the copy sorting last is the one that
survives.
+ //
+ // A batch that moves a run's start leaves the copies disagreeing on
where their interval
+ // begins. Nulls are inert on this key: only decomposition tails carry
a null startAt, and
+ // orderDecompositionTailsFirst has already separated those by the
time it is consulted.
+ startAtCol.desc_nulls_first,
+ // Copies agreeing on their run start may still disagree on its
closure, so nulls sort
+ // first and are dropped: a copy that recorded no boundary never
displaces one that did.
+ endAtCol.desc_nulls_first
)
}
@@ -254,33 +264,15 @@ case class Scd2BatchProcessor(
}
/**
- * Find the auxiliary-table rows whose state matters for reconciling the
microbatch.
- *
- * @param rawAuxiliaryTableDf
- * the auxiliary table in its native schema, which is expected to contain
- * [[deletedByBatchIdColName]] in addition to all of the columns in the
target table.
- * @param perKeyMinimumSequenceInMicrobatchDf
- * one row per distinct key as produced by
[[computeMinimumSequencePerKey]], representing
- * the minimum sequence for that key in the microbatch.
- * @param batchId
- * the underlying Spark streaming query's batchId, used to scope aux-row
visibility for
- * replay-stability across retries of the same microbatch.
- * @return
- * a dataframe containing all the affected aux rows, with the aux-only
- * [[deletedByBatchIdColName]] column dropped so the result is
union-compatible with
- * preprocessed microbatch rows and target-table rows downstream.
+ * Restrict the auxiliary table to the rows that are live for this
microbatch, and drop the
+ * aux-only [[deletedByBatchIdColName]] column so the result shares the
canonical SCD2 row
+ * schema with target-table rows and preprocessed-microbatch rows.
*/
- private[autocdc] def findAffectedRowsFromAuxiliaryTable(
+ private def filterLiveAuxiliaryRows(
rawAuxiliaryTableDf: DataFrame,
- perKeyMinimumSequenceInMicrobatchDf: DataFrame,
- batchId: Long
- ): DataFrame = {
- val auxTableRecordStartAtField = Scd2BatchProcessor.recordStartAtOf(
- F.col(AutoCdcReservedNames.cdcMetadataColName)
- )
+ batchId: Long): DataFrame = {
val auxTableDeletedByBatchIdCol =
F.col(Scd2BatchProcessor.deletedByBatchIdColName)
-
- val reducedAuxiliaryTableDf = rawAuxiliaryTableDf
+ rawAuxiliaryTableDf
.filter(
// [[deletedByBatchIdColName]] carries the batchId whose MERGE
logically deleted the
// row, or null on live aux rows. Rows deleted by other batches are
excluded - those
@@ -290,115 +282,136 @@ case class Scd2BatchProcessor(
auxTableDeletedByBatchIdCol.isNull ||
auxTableDeletedByBatchIdCol === F.lit(batchId)
)
- // Drop the aux-only idempotency column so the output schema matches
target-table rows
- // and preprocessed-microbatch rows (which share the same canonical SCD2
row schema).
.drop(Scd2BatchProcessor.deletedByBatchIdColName)
+ }
- val perKeyMinimumSequenceInMicrobatchCol =
F.col(Scd2BatchProcessor.minSequenceColName)
+ /**
+ * Project a table of canonical SCD2 rows down to `[key1, ... keyN,
effectiveRecordStartAt]`.
+ */
+ private def projectEffectiveRecordStartAtPerRow(rowsDf: DataFrame):
DataFrame =
+ rowsDf.select(
+ keysQuoted.map(F.col) :+
+ Scd2BatchProcessor.canonicalRowIntervalColumns.effectiveRecordStartAt
+ .as(Scd2BatchProcessor.effectiveRecordStartAtColName): _*
+ )
- // Per key, identify the sequence value associated with the anchor row in
the aux table.
- //
- // The anchor row is the aux row with the largest
[[recordStartAtFieldName]] strictly less
- // than the min sequence in the incoming microbatch for that key. The
reconciler needs this
- // "left context" in two cases:
- // (1) Incoming no-op upsert: without the anchor, it would look like a
new run head, when in
- // reality it's a part of an existing no-op run/head.
- // (2) Incoming state-changing upsert that bisects two aux no-ops: the
anchor surfaces
- // the before-half so both halves can be promoted to target. (The
after-half is
- // picked up by the >= minSeq branch.)
- //
- // Because no-op upserts are stored only in the aux table, the anchor
concept only exists when
- // pulling in rows from the aux table, and is not relevant for the target
table.
- //
- // Keys with no aux row strictly before the min sequence have no anchor;
their affected set
- // reduces to "all aux rows at or after the min sequence."
- //
- // The shape of this DataFrame is: [key1, key2, ... keyN, anchorSequence]
- val perKeyAnchorSequenceDf = reducedAuxiliaryTableDf
+ /**
+ * Per key; calculate the earliest point in time (sequence) at or after
which all existing rows
+ * across the auxiliary and target tables may be affected by the microbatch,
and therefore should
+ * be pulled in for reconciliation. The row sitting exactly at the cutoff is
itself included.
+ *
+ * Returns a dataframe with one row per distinct key in
[[perKeyMinimumSequenceInMicrobatchDf]],
+ * with the key columns and the calculated [[affectedSequenceCutoffColName]]
column.
+ */
+ private[autocdc] def computePerKeyAffectedSequenceCutoff(
+ rawAuxiliaryTableDf: DataFrame,
+ targetTableDf: DataFrame,
+ perKeyMinimumSequenceInMicrobatchDf: DataFrame,
+ batchId: Long
+ ): DataFrame = {
+ val effectiveRecordStartAtCol =
F.col(Scd2BatchProcessor.effectiveRecordStartAtColName)
+ val perKeyMinimumSequenceInMicrobatchCol =
F.col(Scd2BatchProcessor.minSequenceColName)
+ val latestSequenceBeforeMicrobatchCol =
+ F.col(Scd2BatchProcessor.latestSequenceBeforeMicrobatchColName)
+
+ // In order to determine the affected sequence cutoff per key, we first
need a "global" (across
+ // both auxiliary and target tables, hence unioned) timeline of existing
effective sequences per
+ // key.
+ val allRowsByEffectiveRecordStartAt =
+
projectEffectiveRecordStartAtPerRow(filterLiveAuxiliaryRows(rawAuxiliaryTableDf,
batchId))
+ .unionByName(projectEffectiveRecordStartAtPerRow(targetTableDf))
+
+ // Across all existing rows per key, we need to find the latest one that
starts
+ // (i.e effectiveRecordStartAt) before the first event for that key in the
microbatch. This is
+ // an "anchor", where any existing row that predates this one on the
timeline will definitely
+ // not be affected by the microbatch.
+ val perKeyLatestSequenceBeforeMicrobatchDf =
allRowsByEffectiveRecordStartAt
// The number of rows in [[perKeyMinimumSequenceInMicrobatchDf]] is
bounded by the
- // number of unique keys in the microbatch, which should typically be
small. The
- // auxiliary table should generally also be small, containing only no-op
upsert runs
- // and tombstones per key. Therefore this join should be cheap, and
broadcast joinable.
+ // number of unique keys in the microbatch, which should typically be
small, so this
+ // join should be cheap and broadcast joinable.
.join(perKeyMinimumSequenceInMicrobatchDf, keysRaw)
- .filter(auxTableRecordStartAtField <
perKeyMinimumSequenceInMicrobatchCol)
+ // We only care about rows that definitely start before the first event
for the same key in
+ // the microbatch.
+ .filter(effectiveRecordStartAtCol < perKeyMinimumSequenceInMicrobatchCol)
.groupBy(keysQuoted.map(F.col): _*)
+ // Of all the existing rows that start before the earliest event in the
microbatch, we want
+ // the latest, hence max-by.
.agg(
-
F.max(auxTableRecordStartAtField).as(Scd2BatchProcessor.anchorSequenceColName)
+ F.max(effectiveRecordStartAtCol)
+ .as(Scd2BatchProcessor.latestSequenceBeforeMicrobatchColName)
)
- val anchorSequenceCol = F.col(Scd2BatchProcessor.anchorSequenceColName)
- val auxRowIsAnchorRow = auxTableRecordStartAtField === anchorSequenceCol
-
- // Now that we have the minimum sequence in the microbatch and the
sequence of the anchor row,
- // we have enough information to compute the full set of auxiliary rows
that may affect or
- // be affected by the microbatch. Membership here is a conservative
superset: every row that
- // could possibly participate in reconciliation is included, but
downstream reconciliation
- // determines the actual outcome per row.
- val auxRowIsAtOrAfterMinSequenceInMicrobatch =
- auxTableRecordStartAtField >= perKeyMinimumSequenceInMicrobatchCol
-
- val auxRowAffectsMicrobatch = auxRowIsAtOrAfterMinSequenceInMicrobatch ||
auxRowIsAnchorRow
-
- val affectedRowsFromAuxiliaryTable = reducedAuxiliaryTableDf
- // Per row, project the minimum microbatch sequence and anchor sequence
for that row's key
- // set onto the row, so the affected-row predicate can be evaluated in a
single filter.
- .join(perKeyMinimumSequenceInMicrobatchDf, keysRaw)
- .join(
- perKeyAnchorSequenceDf,
- keysRaw,
- joinType = "left"
- )
- .filter(auxRowAffectsMicrobatch)
- .drop(perKeyMinimumSequenceInMicrobatchCol, anchorSequenceCol)
- affectedRowsFromAuxiliaryTable
+ // Now for all unique keys in the microbatch, we need to calculate its
affected sequence
+ // cutoff. If a key has existing rows and at least one of them precedes
the earliest event
+ // for that same key in the microbatch, use its sequence as the cutoff. In
all other cases
+ // (key does not yet exist in aux/target, or microbatch contains event for
key that precedes
+ // all existing rows), the min sequence in the microbatch will be the
cutoff point for the key;
+ // all existing rows will necessarily be considered affected. This is a
very cheap join to
+ // make, where both dataframes cardinality is limited by the number of
unique keys in the
+ // microbatch.
+ perKeyMinimumSequenceInMicrobatchDf
+ .join(perKeyLatestSequenceBeforeMicrobatchDf, keysRaw, joinType = "left")
+ .select(
+ keysQuoted.map(F.col) :+
+ F.coalesce(latestSequenceBeforeMicrobatchCol,
perKeyMinimumSequenceInMicrobatchCol)
+ .as(Scd2BatchProcessor.affectedSequenceCutoffColName): _*
+ )
}
/**
- * Find the target-table rows whose state matters for reconciling the
microbatch.
+ * Per key, keep only rows in [[rowsDf]] that are included by the sequence
cutoff.
*
- * @param targetTableDf
- * the target table in its native schema.
- * @param perKeyMinimumSequenceInMicrobatchDf
- * one row per distinct key as produced by
[[computeMinimumSequencePerKey]], representing
- * the minimum sequence for that key in the microbatch.
- * @return
- * a dataframe containing the affected target rows, with all columns
passed-through.
+ * A plain `effectiveRecordStartAt >= cutoff` threshold suffices, in both
directions:
+ *
+ * 1. Nothing needed for reconciliation is missed. Once a row is pulled in,
every later row for
+ * that key must come with it, so that boundaries reconcile and rows are
promoted or demoted
+ * correctly - even when the two live in different tables, as when a
run's visible tail sits
+ * in the target table after the run's hidden head in the auxiliary
table. The cutoff is a
+ * single threshold per key over the unified ordering across both tables,
and so selects
+ * exactly a complete suffix of the global, unified view. This includes
the live rows (open
+ * upserts in the target) too.
+ * 2. Nothing dropped was needed for reconciliation. The cutoff is the
position of the last row
+ * preceding the microbatch per key, so every row below it is separated
from the microbatch
+ * by at least one intervening row, and cannot be affected by it. An open
row is the one case
+ * that does not follow from separation alone, since it is affected by
anything after it -
+ * but no row can intervene above an open row, as that row would have
closed it.
*/
- private[autocdc] def findAffectedRowsFromTargetTable(
- targetTableDf: DataFrame,
- perKeyMinimumSequenceInMicrobatchDf: DataFrame
- ): DataFrame = {
- val targetEndAtCol = F.col(Scd2BatchProcessor.endAtColName)
- val perKeyMinimumSequenceInMicrobatchCol =
F.col(Scd2BatchProcessor.minSequenceColName)
+ private def selectRowsAtOrAfterCutoff(
+ rowsDf: DataFrame,
+ perKeyAffectedSequenceCutoffDf: DataFrame): DataFrame = {
+ val affectedSequenceCutoffCol =
F.col(Scd2BatchProcessor.affectedSequenceCutoffColName)
+ rowsDf
+ .join(perKeyAffectedSequenceCutoffDf, keysRaw)
+ .filter(
+ Scd2BatchProcessor.canonicalRowIntervalColumns.effectiveRecordStartAt
>=
+ affectedSequenceCutoffCol)
+ .drop(affectedSequenceCutoffCol)
+ }
- // Per key, identify all the rows in the target table that may be affected
by the
- // incoming microbatch.
- //
- // Unlike the auxiliary table, the target table holds visible rows only:
no hidden open
- // no-op upsert rows, no tombstones. Visible rows for a given key form a
non-overlapping
- // interval partition over the sequencing axis, and at most one row has a
null [[endAtColName]]
- // (the currently active row per key).
- //
- // Hence we can simply grab all rows that were active at some point after
the min sequencing
- // 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
- // smallest incoming sequence: the consecutive left neighbor of that
incoming event. This
- // provides "left context" for the smallest event, analogous to the anchor
row in
- // [[findAffectedRowsFromAuxiliaryTable]]. It may need to be demoted from
a target run
- // boundary to an aux no-op continuation if the incoming event at minSeq
turns out to
- // extend an earlier run.
- val rowEndsAfterMinimumSequence = targetEndAtCol >=
perKeyMinimumSequenceInMicrobatchCol
- val rowMayBeAffected = isCurrentlyActiveRow || rowEndsAfterMinimumSequence
-
- val affectedRowsFromTargetTable = targetTableDf
- .join(perKeyMinimumSequenceInMicrobatchDf, keysRaw)
- .filter(rowMayBeAffected)
- .drop(perKeyMinimumSequenceInMicrobatchCol)
+ /**
+ * Retrieve all rows from the auxiliary table that are possibly affected by
this microbatch, and
+ * need to participate in reconciliation.
+ */
+ private[autocdc] def findAffectedRowsFromAuxiliaryTable(
+ rawAuxiliaryTableDf: DataFrame,
+ perKeyAffectedSequenceCutoffDf: DataFrame,
+ batchId: Long
+ ): DataFrame = selectRowsAtOrAfterCutoff(
+ rowsDf = filterLiveAuxiliaryRows(rawAuxiliaryTableDf, batchId),
+ perKeyAffectedSequenceCutoffDf = perKeyAffectedSequenceCutoffDf
+ )
- affectedRowsFromTargetTable
- }
+ /**
+ * Retrieve all rows from the target table that are possibly affected by
this microbatch, and
+ * need to participate in reconciliation.
+ */
+ private[autocdc] def findAffectedRowsFromTargetTable(
+ targetTableDf: DataFrame,
+ perKeyAffectedSequenceCutoffDf: DataFrame
+ ): DataFrame = selectRowsAtOrAfterCutoff(
+ rowsDf = targetTableDf,
+ perKeyAffectedSequenceCutoffDf = perKeyAffectedSequenceCutoffDf
+ )
/**
* For every closed non-tombstone row in the input dataframe whose immediate
window-order
@@ -611,6 +624,12 @@ case class Scd2BatchProcessor(
* upsert drops in favor of a same-sequence tombstone (delete wins over
upsert at the
* same instant).
*
+ * Two copies of one row that differ only in the boundaries each recorded -
which the
+ * auxiliary merge can produce for the target merge to read - collide the
same way, and
+ * are ordered by [[startAtColName]] first and [[endAtColName]] second, each
descending
+ * with nulls first. A copy that recorded no boundary therefore never
outlives one that
+ * did, so of an open and a closed copy at the same recordStartAt the closed
one survives.
+ *
* @param decomposedRowsPerKey
* the output of [[decomposeOutOfOrderRows]]: a dataframe conforming to
the canonical
* SCD2 row schema `[user_cols..., [[startAtColName]], [[endAtColName]],
@@ -738,7 +757,8 @@ case class Scd2BatchProcessor(
F.when(
isWindowLocalUpsertRunHead,
// The first row in the window may be a window-local run head but not
a global run
- // head (e.g., an aux anchor row pulled in for left context). In that
case, `startAt`
+ // head (e.g., the row at the affected sequence cutoff, pulled in from
either the
+ // auxiliary or the target table for left context). In that case,
`startAt`
// may be strictly less than `recordStartAt`, encoding the true global
run start, and
// we propagate it forward to later in-window continuations of the
same run.
// For every later window-local upsert run head, `recordStartAt` is
the run start.
@@ -1492,12 +1512,35 @@ object Scd2BatchProcessor {
s"${AutoCdcReservedNames.prefix}is_redundant_delete_encoding"
/**
- * Name of the temporary column used to identify the sequence associated
with the anchor
- * row found in the auxiliary table for the incoming microbatch. Since
sequences must be unique
- * amongst all rows for a key (or risk undefined behavior), this sequence
value uniquely
- * identifies an exact row in the aux.
+ * Name of the temporary column carrying a row's
[[Scd2IntervalColumns.effectiveRecordStartAt]]
+ * in the narrow, key-plus-ordering-metadata union of the auxiliary and
target tables that
+ * [[Scd2BatchProcessor.computePerKeyAffectedSequenceCutoff]] aggregates
over.
+ *
+ * Temporary in that the column has no observable side effect or persistence
across microbatches.
*/
- private val anchorSequenceColName: String =
s"${AutoCdcReservedNames.prefix}anchor_sequence"
+ private val effectiveRecordStartAtColName: String =
+ s"${AutoCdcReservedNames.prefix}effective_record_start_at"
+
+ /**
+ * Name of the temporary column holding, per key, the largest effective
ordering position
+ * strictly below the key's minimum microbatch sequence, across the
auxiliary AND target tables
+ * jointly. Null for a key with no such row, in which case the affected
sequence cutoff falls
+ * back to the microbatch minimum.
+ *
+ * Temporary in that the column has no observable side effect or persistence
across microbatches.
+ */
+ private val latestSequenceBeforeMicrobatchColName: String =
+ s"${AutoCdcReservedNames.prefix}latest_sequence_before_microbatch"
+
+ /**
+ * Name of the temporary column holding the single per-key cutoff on
effective ordering
+ * position that gates affected-row selection from both the auxiliary and
the target table, as
+ * computed by [[Scd2BatchProcessor.computePerKeyAffectedSequenceCutoff]].
+ *
+ * Temporary in that the column has no observable side effect or persistence
across microbatches.
+ */
+ private val affectedSequenceCutoffColName: String =
+ s"${AutoCdcReservedNames.prefix}affected_sequence_cutoff"
/**
* Name of the temporary column projected by
[[Scd2BatchProcessor.identifyAndTagAuxRows]] to
@@ -1524,6 +1567,17 @@ object Scd2BatchProcessor {
private def recordStartAtOf(cdcMetadataCol: Column): Column =
cdcMetadataCol.getField(recordStartAtFieldName)
+ /**
+ * The [[Scd2IntervalColumns]] of a row read from either the auxiliary or
the target table, in
+ * the canonical SCD2 row schema. The columns are unresolved name
references, so they read from
+ * whichever dataframe the expressions are applied to.
+ */
+ private def canonicalRowIntervalColumns: Scd2IntervalColumns =
Scd2IntervalColumns(
+ recordStartAt =
recordStartAtOf(F.col(AutoCdcReservedNames.cdcMetadataColName)),
+ startAt = F.col(startAtColName),
+ endAt = F.col(endAtColName)
+ )
+
/**
* Schema of the CDC metadata struct column for SCD2 rows.
*/
diff --git
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandler.scala
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandler.scala
index 42d9a55126d2..e4adfcb516be 100644
---
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandler.scala
+++
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandler.scala
@@ -75,16 +75,24 @@ case class Scd2ForeachBatchHandler(
)
val auxTableDf =
batchDf.sparkSession.read.table(auxiliaryTableIdentifier.quotedString)
- val affectedRowsFromAuxiliaryTable =
batchProcessor.findAffectedRowsFromAuxiliaryTable(
+ val targetTableDf =
batchDf.sparkSession.read.table(targetTableIdentifier.quotedString)
+
+ val perKeyAffectedSequenceCutoffDf =
batchProcessor.computePerKeyAffectedSequenceCutoff(
rawAuxiliaryTableDf = auxTableDf,
+ targetTableDf = targetTableDf,
perKeyMinimumSequenceInMicrobatchDf =
perKeyMinimumSequenceInMicrobatchDf,
batchId = batchId
)
- val targetTableDf =
batchDf.sparkSession.read.table(targetTableIdentifier.quotedString)
+ val affectedRowsFromAuxiliaryTable =
batchProcessor.findAffectedRowsFromAuxiliaryTable(
+ rawAuxiliaryTableDf = auxTableDf,
+ perKeyAffectedSequenceCutoffDf = perKeyAffectedSequenceCutoffDf,
+ batchId = batchId
+ )
+
val affectedRowsFromTargetTable =
batchProcessor.findAffectedRowsFromTargetTable(
targetTableDf = targetTableDf,
- perKeyMinimumSequenceInMicrobatchDf = perKeyMinimumSequenceInMicrobatchDf
+ perKeyAffectedSequenceCutoffDf = perKeyAffectedSequenceCutoffDf
)
// The three inputs share the canonical SCD2 row schema by name, but not
necessarily by column
diff --git
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessorSuite.scala
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessorSuite.scala
index 42b294248413..de82f05047cd 100644
---
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessorSuite.scala
+++
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessorSuite.scala
@@ -89,6 +89,44 @@ class Scd2BatchProcessorSuite extends QueryTest with
SharedSparkSession {
spark.createDataFrame(spark.sparkContext.parallelize(rows), schema)
}
+ /**
+ * Select the affected aux rows the way [[Scd2ForeachBatchHandler]] does:
compute the shared
+ * per-key affected-sequence cutoff across both tables first, then apply it
to the aux table.
+ */
+ private def findAffectedAuxRows(
+ processor: Scd2BatchProcessor,
+ aux: DataFrame,
+ target: DataFrame,
+ minSeq: DataFrame,
+ batchId: Long = 100L): DataFrame =
+ processor.findAffectedRowsFromAuxiliaryTable(
+ rawAuxiliaryTableDf = aux,
+ perKeyAffectedSequenceCutoffDf =
processor.computePerKeyAffectedSequenceCutoff(
+ rawAuxiliaryTableDf = aux,
+ targetTableDf = target,
+ perKeyMinimumSequenceInMicrobatchDf = minSeq,
+ batchId = batchId
+ ),
+ batchId = batchId
+ )
+
+ /** Target-side counterpart of [[findAffectedAuxRows]]. */
+ private def findAffectedTargetRows(
+ processor: Scd2BatchProcessor,
+ target: DataFrame,
+ aux: DataFrame,
+ minSeq: DataFrame,
+ batchId: Long = 100L): DataFrame =
+ processor.findAffectedRowsFromTargetTable(
+ targetTableDf = target,
+ perKeyAffectedSequenceCutoffDf =
processor.computePerKeyAffectedSequenceCutoff(
+ rawAuxiliaryTableDf = aux,
+ targetTableDf = target,
+ perKeyMinimumSequenceInMicrobatchDf = minSeq,
+ batchId = batchId
+ )
+ )
+
/**
* Build a [[Scd2BatchProcessor]] suitable for `findAffected*` and
* `computeMinimumSequencePerKey` tests. The `sequencing` is fixed to
`F.col("seq")`,
@@ -822,17 +860,21 @@ class Scd2BatchProcessorSuite extends QueryTest with
SharedSparkSession {
val keySchema = new StructType().add("id", IntegerType)
val userSchema = keySchema.add("value", StringType)
- // Two keys to demonstrate per-key anchor isolation.
+ // Two keys to demonstrate per-key cutoff isolation.
//
// Input row shape per `auxTableOf`:
// (id, value, __START_AT, __END_AT, Row(recordStartAt),
deletedByBatchId)
//
- // Key 1: aux rows at recordStartAt 3, 5, 10. minSeq = 10.
- // - 3 -> older than the anchor; dropped.
- // - 5 -> anchor (max < 10); included.
- // - 10 -> at minSeq; included via the >= branch (NOT as anchor;
selection is strict <).
- // Key 2: only one aux row at 7, minSeq = 7.
- // - 7 -> at minSeq; included via >= branch. No anchor (no rows < 7 for
this key).
+ // The target table is empty throughout, so every cutoff below comes from
the aux table.
+ //
+ // Key 1: aux rows at recordStartAt 3, 5, 10. minSeq = 10, so the cutoff
is 5 (the largest
+ // recordStartAt strictly below minSeq).
+ // - 3 -> below the cutoff; dropped.
+ // - 5 -> sits at the cutoff; included.
+ // - 10 -> after the cutoff; included.
+ // Key 2: only one aux row at 7, minSeq = 7. Nothing precedes minSeq, so
the cutoff falls
+ // back to minSeq itself.
+ // - 7 -> sits at the cutoff; included.
val aux = auxTableOf(userSchema)(
Row(1, "v1.3", 3L, null, Row(3L), null),
Row(1, "v1.5", 5L, null, Row(5L), null),
@@ -843,19 +885,16 @@ class Scd2BatchProcessorSuite extends QueryTest with
SharedSparkSession {
Row(1, 10L),
Row(2, 7L)
)
+ val target = targetTableOf(userSchema)()
- val result = processor.findAffectedRowsFromAuxiliaryTable(
- rawAuxiliaryTableDf = aux,
- perKeyMinimumSequenceInMicrobatchDf = minSeq,
- batchId = 100L
- )
+ val result = findAffectedAuxRows(processor, aux = aux, target = target,
minSeq = minSeq)
checkAnswer(
df = result,
expectedAnswer = Seq(
- Row(1, "v1.5", 5L, null, Row(5L)), // anchor for key=1
- Row(1, "v1.10", 10L, null, Row(10L)), // >= minSeq for key=1
- Row(2, "v2.7", 7L, null, Row(7L)) // >= minSeq for key=2 (no
anchor)
+ Row(1, "v1.5", 5L, null, Row(5L)), // sits at key=1's cutoff
+ Row(1, "v1.10", 10L, null, Row(10L)), // after key=1's cutoff
+ Row(2, "v2.7", 7L, null, Row(7L)) // sits at key=2's cutoff, which
fell back to minSeq
)
)
}
@@ -867,27 +906,23 @@ class Scd2BatchProcessorSuite extends QueryTest with
SharedSparkSession {
// Aux carries a mix of row kinds for one key. The find function does NOT
distinguish
// between them - it filters purely on `recordStartAt` - so a tombstone, a
no-op upsert
- // run head, and a continuation are all eligible anchor candidates and all
eligible for
- // the >= minSeq inclusion branch.
+ // run head, and a continuation can all set the cutoff, and can all be
selected by it.
val aux = auxTableOf(userSchema)(
// Tombstone at recordStartAt = 3 (deleted at sequence 3): startAt =
endAt = 3.
- // Older than the anchor; dropped.
+ // Below the cutoff; dropped.
Row(1, null, 3L, 3L, Row(3L), null),
// No-op upsert continuation at recordStartAt = 7: startAt inherits its
run head's
- // recordStartAt, endAt is null. Anchor for minSeq=10 (max < 10).
+ // recordStartAt, endAt is null. Sets the cutoff for minSeq=10 (nearest
below it).
Row(1, "alice", 5L, null, Row(7L), null),
- // Tombstone at recordStartAt = 12: at-or-after minSeq, included via >=
branch.
+ // Tombstone at recordStartAt = 12: after the cutoff; included.
Row(1, null, 12L, 12L, Row(12L), null),
- // No-op upsert continuation at recordStartAt = 15: included via >=
branch.
+ // No-op upsert continuation at recordStartAt = 15: after the cutoff;
included.
Row(1, "bob", 13L, null, Row(15L), null)
)
val minSeq = minSeqOf(keySchema)(Row(1, 10L))
+ val target = targetTableOf(userSchema)()
- val result = processor.findAffectedRowsFromAuxiliaryTable(
- rawAuxiliaryTableDf = aux,
- perKeyMinimumSequenceInMicrobatchDf = minSeq,
- batchId = 100L
- )
+ val result = findAffectedAuxRows(processor, aux = aux, target = target,
minSeq = minSeq)
checkAnswer(
df = result,
@@ -910,19 +945,16 @@ class Scd2BatchProcessorSuite extends QueryTest with
SharedSparkSession {
Row(1, "alice", 2L, null, Row(12L), null)
)
val minSeq = minSeqOf(keySchema)(Row(1, 10L))
+ val target = targetTableOf(userSchema)()
- val result = processor.findAffectedRowsFromAuxiliaryTable(
- rawAuxiliaryTableDf = aux,
- perKeyMinimumSequenceInMicrobatchDf = minSeq,
- batchId = 100L
- )
+ val result = findAffectedAuxRows(processor, aux = aux, target = target,
minSeq = minSeq)
checkAnswer(
df = result,
expectedAnswer = Seq(
- // Row with record start at of 8 gets pulled in as an anchor,
+ // Row with record start at of 8 sets the cutoff,
Row(1, "alice", 2L, null, Row(8L)),
- // Row with record start at of 12 gets pulled in as a regular affected
row.
+ // Row with record start at of 12 sits after the cutoff.
Row(1, "alice", 2L, null, Row(12L))
)
)
@@ -933,31 +965,28 @@ class Scd2BatchProcessorSuite extends QueryTest with
SharedSparkSession {
val keySchema = new StructType().add("id", IntegerType)
val userSchema = keySchema.add("value", StringType)
- // Tombstone-as-anchor is incidental: the find function selects the anchor
purely on
- // `max recordStartAt < minSeq`, so a tombstone qualifies just like any
other row kind.
- // Downstream reconciliation does not actually rely on the anchor when it
is a
- // tombstone (a delete already closed the prior run, so any subsequent
incoming event
- // is necessarily a fresh run head regardless of whether the anchor is
surfaced). We
- // still pull it in as a harmless side effect of the range filter, and
this behavior is
+ // A tombstone setting the cutoff is incidental: the cutoff is the largest
effective
+ // record start below minSeq across both tables, so a tombstone qualifies
just like any
+ // other row kind. Downstream reconciliation does not actually rely on
that row when it
+ // is a tombstone (a delete already closed the prior run, so any
subsequent incoming
+ // event is necessarily a fresh run head regardless of whether it is
surfaced). We still
+ // pull it in as a harmless side effect of the range filter, and this
behavior is
// documented via test.
val aux = auxTableOf(userSchema)(
Row(1, null, 7L, 7L, Row(7L), null),
Row(1, null, 12L, 12L, Row(12L), null)
)
val minSeq = minSeqOf(keySchema)(Row(1, 10L))
+ val target = targetTableOf(userSchema)()
- val result = processor.findAffectedRowsFromAuxiliaryTable(
- rawAuxiliaryTableDf = aux,
- perKeyMinimumSequenceInMicrobatchDf = minSeq,
- batchId = 100L
- )
+ val result = findAffectedAuxRows(processor, aux = aux, target = target,
minSeq = minSeq)
checkAnswer(
df = result,
expectedAnswer = Seq(
- // Pulled in as anchor.
+ // Sets the cutoff.
Row(1, null, 7L, 7L, Row(7L)),
- // Pulled in as regular affected row.
+ // Sits after the cutoff.
Row(1, null, 12L, 12L, Row(12L))
)
)
@@ -973,9 +1002,9 @@ class Scd2BatchProcessorSuite extends QueryTest with
SharedSparkSession {
// The idempotency filter retains rows deleted by `currentBatchId` (so a
mid-flight
// retry sees its own prior writes) and drops rows deleted by any other
batch. This
- // applies uniformly to both the anchor and non-anchor affected rows.
+ // applies uniformly to the row at the cutoff and to the rows after it.
val aux = auxTableOf(userSchema)(
- // Anchor candidate (recordStartAt < minSeq):
+ // Cutoff candidate (recordStartAt < minSeq):
Row(1, "anchor", 5L, null, Row(5L), currentBatchId), // deleted by
current -> kept
// At-or-after minSeq:
Row(1, "live", 10L, null, Row(10L), null), // not deleted -> kept
@@ -983,10 +1012,13 @@ class Scd2BatchProcessorSuite extends QueryTest with
SharedSparkSession {
Row(1, "ignored", 12L, null, Row(12L), differentBatchId) // deleted by
another -> dropped
)
val minSeq = minSeqOf(keySchema)(Row(1, 10L))
+ val target = targetTableOf(userSchema)()
- val result = processor.findAffectedRowsFromAuxiliaryTable(
- rawAuxiliaryTableDf = aux,
- perKeyMinimumSequenceInMicrobatchDf = minSeq,
+ val result = findAffectedAuxRows(
+ processor,
+ aux = aux,
+ target = target,
+ minSeq = minSeq,
batchId = currentBatchId
)
@@ -1009,22 +1041,25 @@ class Scd2BatchProcessorSuite extends QueryTest with
SharedSparkSession {
val currentBatchId = 100L
val differentBatchId = 99L
- // Codifies the step-ordering invariant inside
`findAffectedRowsFromAuxiliaryTable`: the
- // idempotency filter MUST run before the anchor `max(...)` aggregation.
Here the closest
+ // Codifies the step-ordering invariant inside
`computePerKeyAffectedSequenceCutoff`: the
+ // idempotency filter MUST run before the cutoff `max(...)` aggregation.
Here the closest
// pre-minSeq candidate (recordStartAt=7) was logically deleted by a
different batch, so
- // it is filtered out and the anchor falls back to recordStartAt=3. If a
future refactor
+ // it is filtered out and the cutoff falls back to recordStartAt=3. If a
future refactor
// were to flip these two steps (e.g. as a "perf optimization"), this test
would catch it
- // because the natural-anchor row (7) would otherwise be selected and then
dropped, leaving
- // no anchor at all.
+ // because row 7 would set the cutoff and then be dropped, leaving nothing
at the cutoff
+ // at all.
val aux = auxTableOf(userSchema)(
Row(1, "live3", 3L, null, Row(3L), null),
Row(1, "stale7", 7L, null, Row(7L), differentBatchId)
)
val minSeq = minSeqOf(keySchema)(Row(1, 10L))
+ val target = targetTableOf(userSchema)()
- val result = processor.findAffectedRowsFromAuxiliaryTable(
- rawAuxiliaryTableDf = aux,
- perKeyMinimumSequenceInMicrobatchDf = minSeq,
+ val result = findAffectedAuxRows(
+ processor,
+ aux = aux,
+ target = target,
+ minSeq = minSeq,
batchId = currentBatchId
)
@@ -1046,12 +1081,9 @@ class Scd2BatchProcessorSuite extends QueryTest with
SharedSparkSession {
// `_cdc_metadata` struct schema untouched.
val aux = auxTableOf(userSchema)(Row(1, "v", 5L, null, Row(5L), null))
val minSeq = minSeqOf(keySchema)(Row(1, 10L))
+ val target = targetTableOf(userSchema)()
- val result = processor.findAffectedRowsFromAuxiliaryTable(
- rawAuxiliaryTableDf = aux,
- perKeyMinimumSequenceInMicrobatchDf = minSeq,
- batchId = 100L
- )
+ val result = findAffectedAuxRows(processor, aux = aux, target = target,
minSeq = minSeq)
assert(!result.columns.contains(Scd2BatchProcessor.deletedByBatchIdColName))
val cdcMetadataField =
result.schema(AutoCdcReservedNames.cdcMetadataColName)
@@ -1065,14 +1097,11 @@ class Scd2BatchProcessorSuite extends QueryTest with
SharedSparkSession {
val aux = auxTableOf(userSchema)(Row(1, "v", 5L, null, Row(5L), null))
val minSeq = minSeqOf(keySchema)(Row(1, 10L))
+ val target = targetTableOf(userSchema)()
- val result = processor.findAffectedRowsFromAuxiliaryTable(
- rawAuxiliaryTableDf = aux,
- perKeyMinimumSequenceInMicrobatchDf = minSeq,
- batchId = 100L
- )
+ val result = findAffectedAuxRows(processor, aux = aux, target = target,
minSeq = minSeq)
- // The lone aux row is the anchor (recordStartAt=5 < minSeq=10, no other
candidates).
+ // The lone aux row sets the cutoff (recordStartAt=5 < minSeq=10, no other
candidates).
checkAnswer(
df = result,
expectedAnswer = Seq(Row(1, "v", 5L, null, Row(5L)))
@@ -1087,9 +1116,9 @@ class Scd2BatchProcessorSuite extends QueryTest with
SharedSparkSession {
val processor = processorWithKeys(Seq("region", "customer_id"))
- // Three composite keys: (US, 1), (EU, 1), (US, 2). Each is independent.
- // (US, 1): anchor at 3; row at 10 included via >=.
- // (EU, 1): anchor at 4; no rows at or after 12 -> only the anchor.
+ // Three composite keys: (US, 1), (EU, 1), (US, 2). Each gets its own
cutoff.
+ // (US, 1): cutoff at 3; the row at 10 follows it.
+ // (EU, 1): cutoff at 4; nothing follows it, so only the cutoff row is
selected.
// (US, 2): no aux rows -> contributes nothing.
val aux = auxTableOf(userSchema)(
Row("US", 1, "us1.3", 3L, null, Row(3L), null),
@@ -1101,12 +1130,9 @@ class Scd2BatchProcessorSuite extends QueryTest with
SharedSparkSession {
Row("EU", 1, 12L),
Row("US", 2, 100L)
)
+ val target = targetTableOf(userSchema)()
- val result = processor.findAffectedRowsFromAuxiliaryTable(
- rawAuxiliaryTableDf = aux,
- perKeyMinimumSequenceInMicrobatchDf = minSeq,
- batchId = 100L
- )
+ val result = findAffectedAuxRows(processor, aux = aux, target = target,
minSeq = minSeq)
checkAnswer(
df = result,
@@ -1125,12 +1151,9 @@ class Scd2BatchProcessorSuite extends QueryTest with
SharedSparkSession {
val aux = auxTableOf(userSchema)()
val minSeq = minSeqOf(keySchema)(Row(1, 10L))
+ val target = targetTableOf(userSchema)()
- val result = processor.findAffectedRowsFromAuxiliaryTable(
- rawAuxiliaryTableDf = aux,
- perKeyMinimumSequenceInMicrobatchDf = minSeq,
- batchId = 100L
- )
+ val result = findAffectedAuxRows(processor, aux = aux, target = target,
minSeq = minSeq)
assert(result.collect().isEmpty)
}
@@ -1144,12 +1167,9 @@ class Scd2BatchProcessorSuite extends QueryTest with
SharedSparkSession {
// Aux only has rows for key=1. Microbatch only sees key=2.
val aux = auxTableOf(userSchema)(Row(1, "v", 5L, null, Row(5L), null))
val minSeq = minSeqOf(keySchema)(Row(2, 10L))
+ val target = targetTableOf(userSchema)()
- val result = processor.findAffectedRowsFromAuxiliaryTable(
- rawAuxiliaryTableDf = aux,
- perKeyMinimumSequenceInMicrobatchDf = minSeq,
- batchId = 100L
- )
+ val result = findAffectedAuxRows(processor, aux = aux, target = target,
minSeq = minSeq)
assert(result.collect().isEmpty)
}
@@ -1166,12 +1186,9 @@ class Scd2BatchProcessorSuite extends QueryTest with
SharedSparkSession {
Row(2, "v2", 7L, null, Row(7L), null)
)
val minSeq = minSeqOf(keySchema)(Row(1, 10L))
+ val target = targetTableOf(userSchema)()
- val result = processor.findAffectedRowsFromAuxiliaryTable(
- rawAuxiliaryTableDf = aux,
- perKeyMinimumSequenceInMicrobatchDf = minSeq,
- batchId = 100L
- )
+ val result = findAffectedAuxRows(processor, aux = aux, target = target,
minSeq = minSeq)
checkAnswer(
df = result,
@@ -1186,11 +1203,13 @@ class Scd2BatchProcessorSuite extends QueryTest with
SharedSparkSession {
val keySchema = new StructType().add("id", IntegerType)
val userSchema = keySchema.add("value", StringType)
- // Single key with four target rows:
- // - row closed at endAt=5 -> < minSeq=10 -> excluded
- // - row closed at endAt=10 -> = minSeq=10 -> included (>=)
- // - row closed at endAt=15 -> > minSeq=10 -> included
- // - row active (endAt=null) -> always included
+ // Single key with four target rows and an empty aux table, so the cutoff
is the target
+ // row nearest below minSeq=10: recordStartAt=5. Selection is on
recordStartAt, not on
+ // endAt, so a row's interval width is irrelevant - only where it starts
matters.
+ // - recordStartAt=1 -> below the cutoff -> excluded
+ // - recordStartAt=5 -> sits at the cutoff -> included
+ // - recordStartAt=10 -> after the cutoff -> included
+ // - recordStartAt=15 -> after the cutoff -> included (and is the
active row)
val target = targetTableOf(userSchema)(
Row(1, "old", 1L, 5L, Row(1L)),
Row(1, "edge", 5L, 10L, Row(5L)),
@@ -1198,11 +1217,9 @@ class Scd2BatchProcessorSuite extends QueryTest with
SharedSparkSession {
Row(1, "active", 15L, null, Row(15L))
)
val minSeq = minSeqOf(keySchema)(Row(1, 10L))
+ val aux = auxTableOf(userSchema)()
- val result = processor.findAffectedRowsFromTargetTable(
- targetTableDf = target,
- perKeyMinimumSequenceInMicrobatchDf = minSeq
- )
+ val result = findAffectedTargetRows(processor, target = target, aux = aux,
minSeq = minSeq)
checkAnswer(
df = result,
@@ -1214,19 +1231,66 @@ class Scd2BatchProcessorSuite extends QueryTest with
SharedSparkSession {
)
}
+ test("affected sequence cutoff derives from the target table") {
+ val processor = processorWithKeys(Seq("id"))
+ val keySchema = new StructType().add("id", IntegerType)
+ val userSchema = keySchema.add("value", StringType)
+
+ // A standalone delete at 40 found nothing live to close, so it survives
in the auxiliary
+ // table as a tombstone. The upsert at 42 then opened a run in the gap
after it.
+ val aux = auxTableOf(userSchema)(Row(1, null, 40L, 40L, Row(40L), null))
+ val target = targetTableOf(userSchema)(Row(1, "target", 42L, null,
Row(42L)))
+ val minSeq = minSeqOf(keySchema)(Row(1, 50L))
+
+ // The target's row at 42 is the cutoff, so the auxiliary tombstone at 40
falls below it.
+ checkAnswer(
+ df = findAffectedTargetRows(processor, target = target, aux = aux,
minSeq = minSeq),
+ expectedAnswer = Seq(Row(1, "target", 42L, null, Row(42L)))
+ )
+ checkAnswer(
+ df = findAffectedAuxRows(processor, aux = aux, target = target, minSeq =
minSeq),
+ expectedAnswer = Seq.empty[Row]
+ )
+ }
+
+ test("affected sequence cutoff derives from the auxiliary table") {
+ val processor = processorWithKeys(Seq("id"))
+ val keySchema = new StructType().add("id", IntegerType)
+ val userSchema = keySchema.add("value", StringType)
+
+ // A delete at 41 closed the target's run, leaving no tombstone of its own
since the closed
+ // row already carries that boundary. A later standalone delete at 42
landed in the gap after
+ // it with nothing to close, and so survives in the auxiliary table.
+ val aux = auxTableOf(userSchema)(Row(1, null, 42L, 42L, Row(42L), null))
+ val target = targetTableOf(userSchema)(Row(1, "target", 40L, 41L,
Row(40L)))
+ val minSeq = minSeqOf(keySchema)(Row(1, 50L))
+
+ // The tombstone at 42 is the cutoff, so the target's row at 40 falls
below it - correctly,
+ // since that interval already closed at 41, before anything in the
microbatch.
+ checkAnswer(
+ df = findAffectedAuxRows(processor, aux = aux, target = target, minSeq =
minSeq),
+ expectedAnswer = Seq(Row(1, null, 42L, 42L, Row(42L)))
+ )
+ checkAnswer(
+ df = findAffectedTargetRows(processor, target = target, aux = aux,
minSeq = minSeq),
+ expectedAnswer = Seq.empty[Row]
+ )
+ }
+
test("findAffectedRowsFromTargetTable computes inclusion independently per
key") {
val processor = processorWithKeys(Seq("id"))
val keySchema = new StructType().add("id", IntegerType)
val userSchema = keySchema.add("value", StringType)
- // Two keys with overlapping endAt ranges but different per-key minSeqs.
Each key is
- // reconciled independently against its own minSeq.
+ // Two keys with overlapping intervals but different per-key minSeqs. Each
key gets its
+ // own cutoff, computed independently.
val target = targetTableOf(userSchema)(
- // Key 1: minSeq=10. "active" (null) and "recent" (15) are at/after 10.
+ // Key 1: minSeq=10, so the cutoff is recordStartAt=5. "k1.recent" sits
at it and
+ // "k1.active" follows it; "k1.old" at recordStartAt=1 falls below it.
Row(1, "k1.old", 1L, 5L, Row(1L)),
Row(1, "k1.recent", 5L, 15L, Row(5L)),
Row(1, "k1.active", 15L, null, Row(15L)),
- // Key 2: minSeq=20. Only "active" (null) is at/after 20.
+ // Key 2: minSeq=20, so the cutoff is recordStartAt=18. Only "k2.active"
survives.
Row(2, "k2.old", 1L, 10L, Row(1L)),
Row(2, "k2.recent", 10L, 18L, Row(10L)),
Row(2, "k2.active", 18L, null, Row(18L))
@@ -1235,11 +1299,9 @@ class Scd2BatchProcessorSuite extends QueryTest with
SharedSparkSession {
Row(1, 10L),
Row(2, 20L)
)
+ val aux = auxTableOf(userSchema)()
- val result = processor.findAffectedRowsFromTargetTable(
- targetTableDf = target,
- perKeyMinimumSequenceInMicrobatchDf = minSeq
- )
+ val result = findAffectedTargetRows(processor, target = target, aux = aux,
minSeq = minSeq)
checkAnswer(
df = result,
@@ -1259,9 +1321,10 @@ class Scd2BatchProcessorSuite extends QueryTest with
SharedSparkSession {
val processor = processorWithKeys(Seq("region", "customer_id"))
- // (US, 1) and (EU, 1) are distinct composite keys. (US, 1)'s active row
is included
- // for minSeq=10; (EU, 1)'s active row is included for minSeq=12; (EU,
1)'s old closed
- // row at endAt=5 is excluded (5 < 12). (US, 2) has no target rows.
+ // (US, 1) and (EU, 1) are distinct composite keys, each getting its own
cutoff.
+ // (US, 1)'s only row is at recordStartAt=1, which becomes its cutoff for
minSeq=10.
+ // (EU, 1)'s cutoff for minSeq=12 is recordStartAt=5, so its older row at
+ // recordStartAt=1 falls below it. (US, 2) has no target rows.
val target = targetTableOf(userSchema)(
Row("US", 1, "us1", 1L, null, Row(1L)),
Row("EU", 1, "eu1.old", 1L, 5L, Row(1L)),
@@ -1272,11 +1335,9 @@ class Scd2BatchProcessorSuite extends QueryTest with
SharedSparkSession {
Row("EU", 1, 12L),
Row("US", 2, 100L)
)
+ val aux = auxTableOf(userSchema)()
- val result = processor.findAffectedRowsFromTargetTable(
- targetTableDf = target,
- perKeyMinimumSequenceInMicrobatchDf = minSeq
- )
+ val result = findAffectedTargetRows(processor, target = target, aux = aux,
minSeq = minSeq)
checkAnswer(
df = result,
@@ -1294,11 +1355,9 @@ class Scd2BatchProcessorSuite extends QueryTest with
SharedSparkSession {
val target = targetTableOf(userSchema)()
val minSeq = minSeqOf(keySchema)(Row(1, 10L))
+ val aux = auxTableOf(userSchema)()
- val result = processor.findAffectedRowsFromTargetTable(
- targetTableDf = target,
- perKeyMinimumSequenceInMicrobatchDf = minSeq
- )
+ val result = findAffectedTargetRows(processor, target = target, aux = aux,
minSeq = minSeq)
assert(result.collect().isEmpty)
}
@@ -1312,11 +1371,9 @@ class Scd2BatchProcessorSuite extends QueryTest with
SharedSparkSession {
// Target only has rows for key=1. Microbatch only sees key=2.
val target = targetTableOf(userSchema)(Row(1, "v", 1L, null, Row(1L)))
val minSeq = minSeqOf(keySchema)(Row(2, 10L))
+ val aux = auxTableOf(userSchema)()
- val result = processor.findAffectedRowsFromTargetTable(
- targetTableDf = target,
- perKeyMinimumSequenceInMicrobatchDf = minSeq
- )
+ val result = findAffectedTargetRows(processor, target = target, aux = aux,
minSeq = minSeq)
assert(result.collect().isEmpty)
}
@@ -1946,7 +2003,8 @@ class Scd2BatchProcessorSuite extends QueryTest with
SharedSparkSession {
val processor = processorWithKeys(Seq("id"))
val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
- // The first row is an aux anchor (startAt < recordStartAt), pulled in as
left context
+ // The first row sits at the affected sequence cutoff (startAt <
recordStartAt), pulled in
+ // as left context
// for a run that began at startAt=2. Because the row sits at the front of
the window,
// its existing startAt encodes the true global run start and must be
preserved -
// and propagated to the in-window continuation.
diff --git
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandlerSuite.scala
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandlerSuite.scala
index 287447cc0664..1384bcd078aa 100644
---
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandlerSuite.scala
+++
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandlerSuite.scala
@@ -20,7 +20,7 @@ package org.apache.spark.sql.pipelines.autocdc
import org.scalatest.BeforeAndAfter
import org.apache.spark.sql.{functions => F, AnalysisException, QueryTest, Row}
-import org.apache.spark.sql.classic.DataFrame
+import org.apache.spark.sql.classic.{DataFrame, Dataset}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.types._
@@ -715,6 +715,132 @@ class Scd2ForeachBatchHandlerSuite
checkAnswer(auxTable, auxRow(1, null, 20L, 20L, 20L, 2L))
}
+ test("the auxiliary merge's own writes do not duplicate the rows the target
merge reads") {
+ // [[Scd2ForeachBatchHandler.execute]] builds ONE reconciliation source
and hands it to two
+ // MERGEs in turn. The auxiliary merge commits first, and the source reads
the auxiliary table,
+ // so the target merge re-evaluates that source over a table this very
batch just rewrote. A
+ // batch that DEMOTES a visible target row into a hidden auxiliary no-op
stresses that: the
+ // same (id, recordStartAt) - the pair both MERGEs match on - ends up in
both tables at once.
+ // Reconciliation has to collapse the two copies, keeping the one that
knows the run's closure.
+ //
+ // Seed: the "a" run covers [5, 20), its head hidden at 5 and its visible
tail at 10, closed by
+ // "b" at 20.
+ createAuxTable(auxRow(1, "a", 5L, null, 5L, null))
+ createTargetTable(
+ targetRow(1, "a", 5L, 20L, 10L),
+ targetRow(1, "b", 20L, null, 20L)
+ )
+
+ // A no-op "a" at 15 lands inside the run and becomes its visible tail,
demoting the tail at 10
+ // into the auxiliary table - where the target table still holds its own
copy of it.
+ val batchId = 11L
+ val batchDf = microbatchOf(sourceSchema)(upsert(1, "a", 15L))
+ val reconciled = exec.reconcileMicrobatch(batchDf, batchId)
+ val sourceTheAuxMergeReads =
reconciled.reconciledAndRoutedDf.collect().toSeq
+ processor.mergeRowsIntoAuxiliaryTable(
+ reconciledDfWithAuxRowsTagged = reconciled.reconciledAndRoutedDf,
+ originalAffectedRowsFromAuxiliaryTable =
reconciled.affectedRowsFromAuxiliaryTable,
+ auxiliaryTableIdentifier = defaultAuxTableIdentifier,
+ batchId = batchId
+ )
+
+ // The demotion landed: the auxiliary table now holds a copy of the event
at 10 alongside the
+ // run head at 5, and the target table has not been touched yet, so it
holds one too.
+ checkAnswer(
+ auxTable,
+ Seq(
+ auxRow(1, "a", 5L, null, 5L, null),
+ auxRow(1, "a", 5L, null, 10L, null)
+ )
+ )
+
+ // Evaluate the source the way the target merge does: a fresh execution
over the same plan, so
+ // its scans are rebuilt against the auxiliary table as the merge above
left it. It must yield
+ // exactly what the auxiliary merge consumed - the duplicated event at 10
collapsing back to
+ // the one copy that was there before, the copy that knows the run's
closure at 20.
+ checkAnswer(
+ Dataset.ofRows(spark, reconciled.reconciledAndRoutedDf.logicalPlan),
+ sourceTheAuxMergeReads
+ )
+
+ processor.mergeRowsIntoTargetTable(
+ reconciledDfWithAuxRowsTagged = reconciled.reconciledAndRoutedDf,
+ affectedRowsFromTargetTable = reconciled.affectedRowsFromTargetTable,
+ targetTableIdentifier = defaultTargetTableIdentifier
+ )
+
+ // The merge lands exactly those rows: the run still covers [5, 20) with
the event at 15 as its
+ // visible tail, and "b" is untouched.
+ checkAnswer(
+ targetTable,
+ Seq(
+ targetRow(1, "a", 5L, 20L, 15L),
+ targetRow(1, "b", 20L, null, 20L)
+ )
+ )
+ }
+
+ test("a demoted row's two copies reconcile even when they disagree on where
the run starts") {
+ // Same cross-merge duplication as the test above, but here the batch also
moves the run's
+ // start, so the two copies of the demoted event disagree on startAt
rather than only on
+ // endAt - the case the startAt sort key exists to decide.
+ //
+ // Seed: the "a" run covers [10, 20) as a single visible event, closed by
"b" at 20. Nothing
+ // is hidden yet.
+ createAuxTable()
+ createTargetTable(
+ targetRow(1, "a", 10L, 20L, 10L),
+ targetRow(1, "b", 20L, null, 20L)
+ )
+
+ // Two no-op "a" events. The one at 5 predates the run and pulls its start
back to 5; the one
+ // at 15 becomes the run's visible tail, demoting the event at 10 into the
auxiliary table.
+ val batchId = 11L
+ val batchDf = microbatchOf(sourceSchema)(upsert(1, "a", 5L), upsert(1,
"a", 15L))
+ val reconciled = exec.reconcileMicrobatch(batchDf, batchId)
+ val sourceTheAuxMergeReads =
reconciled.reconciledAndRoutedDf.collect().toSeq
+ processor.mergeRowsIntoAuxiliaryTable(
+ reconciledDfWithAuxRowsTagged = reconciled.reconciledAndRoutedDf,
+ originalAffectedRowsFromAuxiliaryTable =
reconciled.affectedRowsFromAuxiliaryTable,
+ auxiliaryTableIdentifier = defaultAuxTableIdentifier,
+ batchId = batchId
+ )
+
+ // Both hidden members of the run now record the run's new start at 5,
while the target table
+ // still holds its own copy of the event at 10 recording the old start at
10.
+ checkAnswer(
+ auxTable,
+ Seq(
+ auxRow(1, "a", 5L, null, 5L, null),
+ auxRow(1, "a", 5L, null, 10L, null)
+ )
+ )
+ checkAnswer(targetTable.filter("value = 'a'"), targetRow(1, "a", 10L, 20L,
10L))
+
+ // The target merge re-evaluates the same plan against the auxiliary table
as the merge above
+ // left it, so the two disagreeing copies of the event at 10 meet in the
window. They must
+ // collapse back to what the auxiliary merge already consumed.
+ checkAnswer(
+ Dataset.ofRows(spark, reconciled.reconciledAndRoutedDf.logicalPlan),
+ sourceTheAuxMergeReads
+ )
+
+ processor.mergeRowsIntoTargetTable(
+ reconciledDfWithAuxRowsTagged = reconciled.reconciledAndRoutedDf,
+ affectedRowsFromTargetTable = reconciled.affectedRowsFromTargetTable,
+ targetTableIdentifier = defaultTargetTableIdentifier
+ )
+
+ // The run now covers [5, 20) with the event at 15 as its visible tail,
and "b" is untouched.
+ checkAnswer(
+ targetTable,
+ Seq(
+ targetRow(1, "a", 5L, 20L, 15L),
+ targetRow(1, "b", 20L, null, 20L)
+ )
+ )
+ }
+
test("duplicate events at the same key and sequence in one microbatch
collapse to one record") {
createAuxTable()
createTargetTable()
@@ -908,6 +1034,75 @@ class Scd2ForeachBatchHandlerSuite
resolvedSequencingType = LongType
)
+ test("a closed run of untracked-only changes is one record holding the last
event's values") {
+ // SPARK-58937 Regression test; A row hidden inside a no-op run should not
be prematurely
+ // considered affected by the microbatch and promoted into the target
table.
+ createTable(defaultAuxIdent, defaultAuxTableIdentifier, trackedAuxSchema)
+ createTable(defaultTargetIdent, defaultTargetTableIdentifier,
trackedCanonicalSchema)
+ val handler = execWith(trackedProcessor)
+
+ Seq(
+ // Start a run whose tracked `name` is "alice".
+ Seq(Row(1, "alice", 39, 39L, false)),
+ // Continue that run with an untracked-only score change.
+ Seq(Row(1, "alice", 41, 41L, false)),
+ // Close the run, then process a later event that needs left context
from the same key.
+ Seq(Row(1, "bob", 42, 42L, false)),
+ Seq(Row(1, "carol", 43, 43L, false))
+ ).zipWithIndex.foreach { case (batch, batchId) =>
+ handler.execute(microbatchOf(trackedSourceSchema)(batch: _*), batchId)
+ }
+
+ checkAnswer(
+ targetTable,
+ Seq(
+ Row(1, "alice", 41, 39L, 42L, meta(41L)),
+ Row(1, "bob", 42, 42L, 43L, meta(42L)),
+ Row(1, "carol", 43, 43L, null, meta(43L))
+ )
+ )
+
+ // The run head stays hidden in the auxiliary table, holding the values it
had before the
+ // untracked-only change continued the run.
+ checkAnswer(auxTable, Row(1, "alice", 39, 39L, null, meta(39L), null))
+ }
+
+ test("an event after a deletion leaves the run the deletion closed exactly
as it was") {
+ // SPARK-58937 Regression test; A row hidden behind a run that a deletion
closed should not be
+ // considered affected by a microbatch that only touches instants after
that deletion.
+ createAuxTable()
+ createTargetTable()
+
+ // Batch 1: a two-event no-op run for "a". The head at 10 is hidden in the
aux table and the
+ // tail at 11 is the single visible row covering the run.
+ runBatch(1L)(upsert(1, "a", 10L), upsert(1, "a", 11L))
+ checkAnswer(targetTable, targetRow(1, "a", 10L, null, 11L))
+ checkAnswer(auxTable, auxRow(1, "a", 10L, null, 10L, null))
+
+ // Batch 2: a delete closes that run at 12. The key is now absent from 12
onwards, so no
+ // target row covers any instant at or after 12 - the history has a gap
there.
+ runBatch(2L)(del(1, 12L))
+ checkAnswer(targetTable, targetRow(1, "a", 10L, 12L, 11L))
+ checkAnswer(auxTable, auxRow(1, "a", 10L, null, 10L, null))
+
+ // Batch 3: a new value at 20, strictly after the gap. It cannot interact
with the run that
+ // already closed at 12, so that run's single visible row must be left
exactly as it is - in
+ // particular the head hidden at 10 must not be pulled back into the
target table beside it.
+ runBatch(3L)(upsert(1, "b", 20L))
+
+ checkAnswer(
+ targetTable,
+ Seq(
+ targetRow(1, "a", 10L, 12L, 11L),
+ targetRow(1, "b", 20L, null, 20L)
+ )
+ )
+
+ // The head hidden at 10 is also left untouched - the new event neither
closed it nor
+ // duplicated it.
+ checkAnswer(auxTable, auxRow(1, "a", 10L, null, 10L, null))
+ }
+
test("changing only an untracked column updates the current record without
adding history") {
createTable(defaultAuxIdent, defaultAuxTableIdentifier, trackedAuxSchema)
createTable(defaultTargetIdent, defaultTargetTableIdentifier,
trackedCanonicalSchema)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
