szehon-ho commented on code in PR #58209:
URL: https://github.com/apache/spark/pull/58209#discussion_r3833439691
##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessorSuite.scala:
##########
@@ -1214,19 +1231,60 @@ 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)
+
+ val aux = auxTableOf(userSchema)(Row(1, "aux", 40L, null, 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 row 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)
+
+ val aux = auxTableOf(userSchema)(Row(1, "aux", 42L, null, Row(42L), null))
+ val target = targetTableOf(userSchema)(Row(1, "target", 40L, null,
Row(40L)))
+ val minSeq = minSeqOf(keySchema)(Row(1, 50L))
Review Comment:
Could this fixture be rebuilt from a state the pipeline can actually reach?
An aux row with `startAt == recordStartAt` and `endAt = null` is a hidden
run head, which by definition has its visible tail later - so the target row
for the same key would have to be at 42 or beyond, not at 40. As written the
state cannot occur in production.
What the test then records as correct is dropping an *active* target row
(`endAt = null`) from the affected set. That is harmless on an impossible
input, but in any reachable state nothing would close that row, and the batch
would leave two open intervals for the key. A reader who later hits a real case
of this has a green test telling them the behavior is intended.
An aux tombstone at 42 with the target row closed at 42 is reachable and
still exercises an aux-derived cutoff.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala:
##########
@@ -290,115 +277,121 @@ 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(
Review Comment:
Worth documenting the invariant that makes the target-side narrowing safe
here.
The old `findAffectedRowsFromTargetTable` included the active row
unconditionally via `endAt.isNull`, with a paragraph explaining why; selection
is now purely on `effectiveRecordStartAt >= cutoff` and that paragraph is gone.
The narrowing is fine, but only because no aux row ever sits inside a target
interval above that row's `recordStartAt` (a hidden no-op always precedes its
run's visible tail, and a delete inside a target interval bisects it rather
than leaving a tombstone there), so a target row covering `minSeq` always sets
the cutoff - the active row being the special case of that.
That chain is currently written nowhere. If a future change ever leaves an
aux row above the active target row, the active row is silently dropped from
the affected set and the batch opens a second interval for the key, with
nothing to catch it. Related to my note on `"affected sequence cutoff derives
from the auxiliary table"`.
##########
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,
Review Comment:
The "nulls sort first so a copy that recorded no boundary never displaces
one that did" rationale applies to `endAt` only. A null `startAt` occurs only
on decomposition tails, which `orderDecompositionTailsFirst` has already
separated by the time this key is consulted, so the null handling on this key
can never decide a tie. Might be worth splitting the comment so each key
carries its own reason.
##########
sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala:
##########
@@ -1492,12 +1492,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 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 anchorSequenceColName: String =
s"${AutoCdcReservedNames.prefix}anchor_sequence"
+ private[autocdc] val affectedSequenceCutoffColName: String =
Review Comment:
Can this be plain `private`? It has no use outside the object - the
`anchorSequenceColName` it replaces was `private`, as are
`effectiveRecordStartAtColName` and `latestSequenceBeforeMicrobatchColName`
added alongside it.
##########
sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandlerSuite.scala:
##########
@@ -715,6 +715,71 @@ 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") {
Review Comment:
Nice test. One gap: both copies here sit at `startAt = 5`, so only the
`endAt` key is exercised. The `startAt` key takes precedence and is
load-bearing in the case where the microbatch moves the run head, so the aux
copy carries a different run start than the target copy.
That case does resolve correctly as far as I can tell - a tie makes
`nextRecordStartAt == recordStartAt < endAt`, so a closed copy sorting first
always decomposes and its tail preserves the boundary, while a closed copy
sorting last survives intact - but that is a load-bearing argument with no test
behind it. Would be good to cover.
--
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]
