AnishMahto commented on code in PR #58209:
URL: https://github.com/apache/spark/pull/58209#discussion_r3834053190
##########
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:
Added to `selectRowsAtOrAfterCutoff`'s scaladoc.
Honestly the difference is just due to implementation detail rather than a
changing guarantee or relying on some other invariant.
Before, it was reasonable to use different conditions to select the cutoff
between the aux and target tables. For the target table specifically, we don't
necessarily even care about the row that first precedes the microbatch, we only
care about the first row that is literally bisected by the microbatch. For the
aux table however, we had to look at rows by whether they precede the
microbatch, because hidden upserts have no `endAt`, and therefore no concept of
being bisected (but rather superseded).
Now that I'm union-ing the aux and target tables to do a global cutoff
calculation rather than per table, it's easier to just take the more
conservative filter and apply it to all rows equally, which is looking at rows
that precede the microbatch.
That means its now theoretically possible that we select a target row that
fully precedes the microbatch rather than actually being bisected by it, and
that _is_ technically a redundancy. But it makes the implementation much
cleaner, is very negligible computation wise, and doesn't affect correctness.
--
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]
