szehon-ho commented on code in PR #58209:
URL: https://github.com/apache/spark/pull/58209#discussion_r3834237888
##########
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:
This comment isn't right, and the state it describes can't occur in
production.
A delete that closes a run leaves no tombstone behind -
`dropLeftoverDeletesPostReconciliation` drops it, because the closed upsert
already carries the boundary. The deletion regression test in
`Scd2ForeachBatchHandlerSuite` shows it: `del(1, 12)` closes the run at 12 and
the auxiliary table afterwards holds only the hidden head at 10. So a target
row closed exactly at 42 and a tombstone at 42 never coexist.
Suggest `Row(1, "target", 40L, 41L, Row(40L))`. A tombstone survives only in
a gap, where nothing open precedes it, so with the run closed at 41 the state
is reachable: `upsert@40`, then `del@41` closes the run and its tombstone is
dropped, then a standalone `del@42` has nothing to close and persists. The
assertions and the aux-derived cutoff are unchanged.
The assertions themselves are correct either way - this is about the fixture
and the comment, not the behavior.
##########
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:
Suggest one more sentence in (2): an open row can never have an intervening
row above it, since that row would have closed it.
"Separated from the microbatch by at least one intervening row, and cannot
be affected by it" is the one step that doesn't hold for an open row on its own
- an open row is affected by anything after it. That same fact is what keeps
the live row above the cutoff, so it also backs the last sentence of (1).
--
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]
