This is an automated email from the ASF dual-hosted git repository.
jose-torres pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 1baff34d3090 [SPARK-57322][SDP] Implement SCD2 Batch Processor;
Reconcile StartAt/EndAt
1baff34d3090 is described below
commit 1baff34d30904a0b7bce8c857216a47c3e9690d7
Author: Andreas Neumann <[email protected]>
AuthorDate: Thu Jul 16 19:56:16 2026 -0700
[SPARK-57322][SDP] Implement SCD2 Batch Processor; Reconcile StartAt/EndAt
Approved AutoCDC SPIP:
https://lists.apache.org/thread/j6sj9wo9odgdpgzlxtvhoy7szs0jplf7
------------
This is a stacked PR. Review incremental diff here:
https://github.com/AnishMahto/spark/compare/SPARK-57222-SCD2-decompose-affected-rows...SPARK-57322-reconcile-start-end-at
### What changes were proposed in this pull request?
**Preamble**:
The SCD type 2 flow is a foreachBatch streaming query on an input
change-data-feed, and is responsible for reconciling the incoming change data
onto some target table that follows SCD2 replication semantics.
SCD2 flows also maintain an "auxiliary" table to keep track of
early-arriving out-of-order received events state. Each microbatch will need to
reconcile against this auxiliary table as well, and update the auxiliary
table's state appropriately for future microbatches.
**Reconcile StartAt/EndAt**:
Once we have all of the relevant rows for microbatch SCD2 reconciliation
decomposed, we can begin setting the start/end-at for incoming microbatch rows,
and rewriting existing start/end-at of existing rows in the aux/target tables
to account for late arriving events.
When iterating on the affected microbatch rows in chronologically sorted
order _after decomposition_, we can reconcile a row's end-at by looking at its
next neighbor, which naturally succeeds the row.
Similarly we may need to rewrite both existing and new upsert-representing
rows' start-at depending on whether the upsert row is a now a run-head or not.
This can be done by looking backwards to identify the upsert run's head in
chronological sorted order.
### Why are the changes needed?
SCD2 core implementation.
### Does this PR introduce _any_ user-facing change?
No, this is core implementation of an unreleased feature (AutoCDC SCD2).
### How was this patch tested?
Unit tested in `Scd2BatchProcessorSuite`.
### Was this patch authored or co-authored using generative AI tooling?
Co-authored with Claude Opus 4.7/4.8
Closes #57192 from anew/SPARK-57322-reconcile-rebased.
Authored-by: Andreas Neumann <[email protected]>
Signed-off-by: Jose Torres <[email protected]>
---
.../spark/sql/pipelines/autocdc/ChangeArgs.scala | 48 +-
.../sql/pipelines/autocdc/Scd2BatchProcessor.scala | 283 +++++++-
.../sql/pipelines/autocdc/ChangeArgsSuite.scala | 42 +-
.../autocdc/Scd2BatchProcessorSuite.scala | 757 ++++++++++++++++++++-
4 files changed, 1104 insertions(+), 26 deletions(-)
diff --git
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/ChangeArgs.scala
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/ChangeArgs.scala
index 49636acc1f8f..15aa535994f9 100644
---
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/ChangeArgs.scala
+++
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/ChangeArgs.scala
@@ -17,6 +17,7 @@
package org.apache.spark.sql.pipelines.autocdc
+import org.apache.spark.SparkException
import org.apache.spark.sql.{AnalysisException, Column}
import org.apache.spark.sql.catalyst.parser.CatalystSqlParser
import org.apache.spark.sql.catalyst.util.QuotingUtils
@@ -153,23 +154,34 @@ object ScdType {
/**
* Configuration for an AutoCDC flow.
*
- * @param keys The column(s) that uniquely identify a row in the
source data.
- * @param sequencing Expression ordering CDC events to correctly resolve
out-of-order
- * arrivals. Must be a sortable type.
- * @param deleteCondition Expression that marks a source row as a DELETE. When
None, all
- * rows are treated as upserts.
- * @param storedAsScdType The SCD strategy these args should be applied to.
- * @param columnSelection Which source columns to select in the target table.
None means
- * all columns.
+ * @param keys The column(s) that uniquely identify a row in
the source data.
+ * @param sequencing Expression ordering CDC events to correctly
resolve out-of-order
+ * arrivals. Must be a sortable type.
+ * @param deleteCondition Expression that marks a source row as a
DELETE. When None, all
+ * rows are treated as upserts.
+ * @param storedAsScdType The SCD strategy these args should be applied
to.
+ * @param columnSelection Which source columns to select in the target
table. None means
+ * all columns.
+ * @param trackHistorySelection SCD2 only. Selects the selected user-data
columns whose values
+ * define a run: two consecutive upsert events
for the same key are
+ * coalesced into the same run iff they agree on
every selected
+ * tracking column. None means every eligible
selected user column
+ * (i.e. every selected source column that is
neither a key nor a
+ * framework column) is considered tracked. Must
be None under SCD1,
+ * which has no run concept and therefore no
history-tracking columns.
+ * See the "run of upsert events" concept in the
`Scd2BatchProcessor`
+ * scaladoc for the precise definition of a run.
*/
case class ChangeArgs(
keys: Seq[UnqualifiedColumnName],
sequencing: Column,
storedAsScdType: ScdType,
deleteCondition: Option[Column] = None,
- columnSelection: Option[ColumnSelection] = None
+ columnSelection: Option[ColumnSelection] = None,
+ trackHistorySelection: Option[ColumnSelection] = None
) {
ChangeArgs.validateNonEmptyKeys(keys)
+ ChangeArgs.validateTrackHistoryOnlyForScd2(storedAsScdType,
trackHistorySelection)
}
object ChangeArgs {
@@ -186,4 +198,22 @@ object ChangeArgs {
)
}
}
+
+ /**
+ * Validates that [[ChangeArgs.trackHistorySelection]] is only set under
SCD2. SCD1 has no run
+ * concept and therefore no history-tracking columns, so a non-None
selection is meaningless.
+ *
+ * User-facing validation is expected to reject this at the API
(GraphRegistration) layer; this
+ * is a defensive internal guard so downstream SCD1 code can assume the
selection is None. Hence
+ * it raises an internal error rather than a user-facing
[[AnalysisException]].
+ */
+ private def validateTrackHistoryOnlyForScd2(
+ storedAsScdType: ScdType,
+ trackHistorySelection: Option[ColumnSelection]): Unit = {
+ if (storedAsScdType == ScdType.Type1 && trackHistorySelection.isDefined) {
+ throw SparkException.internalError(
+ "trackHistorySelection must be None under SCD1; it has no
history-tracking columns."
+ )
+ }
+ }
}
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 b0c511958ab7..27b9036523f1 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
@@ -439,9 +439,12 @@ case class Scd2BatchProcessor(
* a dataframe with the same schema as the input. Every closed
non-tombstone row that
* was bisected has been replaced by its head + tail pair; every other row
is carried
* through as-is. Each output row can be classified as one of:
{decomposition head,
- * decomposition tail, instantaneous delete, open upsert,
closed-and-unbisected row}. It's
- * possible that some of the returned decomposition tails are logically
redundant, as
- * deletion markers that are immediately overtaken by a succeeding row.
+ * decomposition tail, tombstone, open upsert, closed-and-unbisected row}.
Tombstones here
+ * are aux-table-only rows representing early-arriving deletes (SCD2 has
no tombstone in the
+ * target table - a delete there is just a closed history record); see the
"tombstone" and
+ * "aux table" concepts in this class's scaladoc. It's possible that some
of the returned
+ * decomposition tails are logically redundant, as deletion markers that
are immediately
+ * overtaken by a succeeding row.
*/
private[autocdc] def decomposeOutOfOrderRows(rowsToDecomposePerKey:
DataFrame): DataFrame = {
val recordStartAtField =
@@ -532,8 +535,8 @@ case class Scd2BatchProcessor(
/**
* Asserts that every row in `decomposedRowsPerKey` conforms to one of the
four canonical
- * post-decomposition shapes - instantaneous delete, open upsert, closed
upsert, or
- * decomposition tail - and is otherwise a structural identity transform.
+ * post-decomposition shapes - tombstone, open upsert, closed upsert, or
decomposition
+ * tail - and is otherwise a structural identity transform.
*
* @param decomposedRowsPerKey
* the output of [[decomposeOutOfOrderRows]]: a dataframe conforming to
the canonical
@@ -556,7 +559,7 @@ case class Scd2BatchProcessor(
val row = Scd2IntervalColumns(recordStartAtField, startAtCol, endAtCol)
val isWellFormedRow =
RowClassifier.isDecompositionTail(row) ||
- RowClassifier.isDeleteRepresentingRow(row) ||
+ RowClassifier.isTombstone(row) ||
RowClassifier.isUpsertRepresentingRow(row)
def stringOrNullLit(c: Column): Column = F.coalesce(c.cast(StringType),
F.lit("null"))
@@ -639,6 +642,202 @@ case class Scd2BatchProcessor(
.filter(!isRedundantAtSameEffectiveSequence)
.drop(Scd2BatchProcessor.nextEffectiveRecordStartAtColName)
}
+
+ /**
+ * Recompute every row's [[startAtColName]] and [[endAtColName]] over the
per-key chronological
+ * window so the dataframe reflects the canonical SCD2 timeline that the
downstream aux- and
+ * target-table merges consume.
+ *
+ * Decomposition tails and tombstones round-trip unchanged. An open upsert
may close at its
+ * successor's effective sequence (becoming closed); the closing successor
may be either a
+ * later state-changing upsert or a delete, which is represented as an
instantaneous tombstone
+ * (see the "tombstone" concept in this class's scaladoc), so a delete
closes the preceding
+ * upsert at the tombstone's sequence like any other successor. A closed
upsert may conversely
+ * have its endAt cleared when absorbed into a run as a no-op continuation
(becoming open).
+ * [[recordStartAtFieldName]] is never modified.
+ *
+ * @param decomposedAndCleanedDf
+ * the output of [[dropRedundantRowsPostDecomposition]]: a dataframe
conforming to the
+ * canonical SCD2 row schema `[user_cols..., [[startAtColName]],
[[endAtColName]],
+ * [[cdcMetadataColName]]]` where every row is in one of the four
canonical post-
+ * decomposition shapes (decomposition tail, tombstone, open upsert,
closed upsert).
+ * If a row is a closed upsert in the input, it is assumed to not be
bisected by any other
+ * row in the input.
+ * @return
+ * a dataframe with the same schema and row count as the input, with each
row's
+ * [[startAtColName]] / [[endAtColName]] replaced by their reconciled
values.
+ */
+ private[autocdc] def reconcileStartAndEndAt(
+ decomposedAndCleanedDf: DataFrame): DataFrame = {
+ val trackedHistoryColumns =
computeTrackedHistoryColumns(decomposedAndCleanedDf)
+
+ val recordStartAtField =
+
Scd2BatchProcessor.recordStartAtOf(F.col(AutoCdcReservedNames.cdcMetadataColName))
+ val startAtCol = F.col(Scd2BatchProcessor.startAtColName)
+ val endAtCol = F.col(Scd2BatchProcessor.endAtColName)
+
+ // Decomposition tails carry no recordStartAt of their own, so they take
the closing
+ // sequence (`endAt`) as their effective ordering position - the same
convention used by
+ // [[orderChronologicallyPerKeyWindow]] and
[[dropRedundantRowsPostDecomposition]].
+ val current = Scd2IntervalColumns(recordStartAtField, startAtCol, endAtCol)
+ val previous = current.lagBy(1, orderChronologicallyPerKeyWindow)
+ val next = current.leadBy(1, orderChronologicallyPerKeyWindow)
+
+ // A row is the last in its per-key window when `LEAD(1)` has no
successor; a constant
+ // literal is sufficient since we only care whether one exists.
+ val isLastRowInKeyWindow =
+ F.lead(F.lit(true), 1).over(orderChronologicallyPerKeyWindow).isNull
+
+ // The current row's tracked-history equality is computed against both its
predecessor and
+ // its successor so the same window scan can decide both run-head start
(LAG-side) and no-op
+ // continuation closure (LEAD-side) without an extra pass. The comparison
is null-safe
+ // (`<=>`), so two rows with matching null values in the same tracked
column register as
+ // equal. An empty tracked-history column set collapses to a constant
`true`, which makes
+ // every consecutive upsert pair a no-op continuation - the correct
degenerate behavior when
+ // the user tracks nothing.
+ val areTrackedColumnsEqualInPreviousRow = trackedHistoryColumns
+ .map { c =>
+ val col = F.col(QuotingUtils.quoteIdentifier(c))
+ col <=> F.lag(col, 1).over(orderChronologicallyPerKeyWindow)
+ }
+ .reduceOption(_ && _)
+ .getOrElse(F.lit(true))
+
+ val areTrackedColumnsEqualInNextRow = trackedHistoryColumns
+ .map { c =>
+ val col = F.col(QuotingUtils.quoteIdentifier(c))
+ col <=> F.lead(col, 1).over(orderChronologicallyPerKeyWindow)
+ }
+ .reduceOption(_ && _)
+ .getOrElse(F.lit(true))
+
+ // Reconciliation of start/end at is dependent on the class of row being
reconciled. Build
+ // row classification predicates.
+ val isDecompositionTail = RowClassifier.isDecompositionTail(current)
+ val isUpsertRepresentingRow =
RowClassifier.isUpsertRepresentingRow(current)
+
+ // From the previous row's perspective, the current row is its successor.
+ val previousIsNoOpUpsertWithCurrent =
+ RowClassifier.isNoOpUpsertContinuation(
+ row = previous,
+ next = current,
+ areTrackedColumnsEqual = areTrackedColumnsEqualInPreviousRow
+ )
+
+ // "Window-local run head" means the current row begins a new run within
the affected
+ // window. The first row in the window is automatically considered
local-run-head since
+ // there's no predecessor to coalesce with. A non-first row is a local run
head iff its
+ // predecessor is not a no-op continuation that absorbs it.
+ val isWindowLocalUpsertRunHead =
+ isUpsertRepresentingRow && !previousIsNoOpUpsertWithCurrent
+ val isFirstRowInKeyWindow = previous.effectiveRecordStartAt.isNull
+ val runHeadStartAt =
+ 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`
+ // 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.
+ F.when(isFirstRowInKeyWindow, startAtCol).otherwise(recordStartAtField)
+ )
+
+ // Propagate the run head's `startAt` forward to every row in the run via
a running
+ // `last(...)` over `[unboundedPreceding, currentRow]`. `runHeadStartAt`
is non-null
+ // only on run heads, and `ignoreNulls = true` makes intermediate rows
inherit the most
+ // recent head's value.
+ val runStartAt =
+ F.last(runHeadStartAt, ignoreNulls = true).over(
+ orderChronologicallyPerKeyWindow.rowsBetween(
+ Window.unboundedPreceding,
+ Window.currentRow
+ )
+ )
+
+ val currentIsNoOpUpsertWithNext =
+ RowClassifier.isNoOpUpsertContinuation(
+ row = current,
+ next = next,
+ areTrackedColumnsEqual = areTrackedColumnsEqualInNextRow
+ )
+
+ val finalStartAt =
+ F.when(isDecompositionTail, F.lit(null).cast(resolvedSequencingType))
+ .when(isUpsertRepresentingRow, runStartAt)
+ .otherwise(startAtCol)
+
+ val finalEndAt =
+ F.when(isDecompositionTail, endAtCol)
+ .when(isLastRowInKeyWindow, endAtCol)
+ // A no-op continuation collapses into its run head, so its own
visible interval is
+ // erased by clearing `endAt` to null. Note this cleared `endAt` is
NOT by itself the
+ // aux-routing signal: the run tail is the row that represents the run
to the user, and
+ // a tail can also carry `endAt = null` (an open run whose latest
event is still active),
+ // so `endAt = null` alone does not mean "route to aux". Deciding
which reconciled rows
+ // land in the target table (the run tail) versus the aux table
(interior no-op rows)
+ // happens in a later transform (SPARK-57378), keyed off the
reconciled run shape rather
+ // than `endAt` alone. A run therefore never fully disappears from the
target.
+ .when(currentIsNoOpUpsertWithNext,
F.lit(null).cast(resolvedSequencingType))
+ // The row already closes strictly before the next event (e.g., a
tombstone or a
+ // closed upsert that ended before the next event arrived), so there
is nothing to
+ // re-close.
+ .when(
+ RowClassifier.rowClosesStrictlyBeforeNextRow(endAtCol,
next.effectiveRecordStartAt),
+ endAtCol)
+ .otherwise(next.effectiveRecordStartAt)
+
+ // Stage the recomputed start/end values into temporary columns first,
then project them
+ // back over the originals via `select`. Both staged values reference the
original
+ // `startAt`/`endAt`, so we cannot replace the originals in a single
`withColumn` step.
+ val staged = decomposedAndCleanedDf
+ .withColumn(Scd2BatchProcessor.finalStartAtColName, finalStartAt)
+ .withColumn(Scd2BatchProcessor.finalEndAtColName, finalEndAt)
+
+ // Determine columns for selection from staged dataframe using the
original input dataframe's
+ // schema. The final startAt/endAt should be remapped, all other columns
should pass through
+ // as-is.
+ val outputColumns = decomposedAndCleanedDf.columns.map {
+ case col if col == Scd2BatchProcessor.startAtColName =>
+ val startAtMetadata = decomposedAndCleanedDf.schema(col).metadata
+ F.col(Scd2BatchProcessor.finalStartAtColName)
+ .as(Scd2BatchProcessor.startAtColName, startAtMetadata)
+ case col if col == Scd2BatchProcessor.endAtColName =>
+ val endAtMetadata = decomposedAndCleanedDf.schema(col).metadata
+ F.col(Scd2BatchProcessor.finalEndAtColName)
+ .as(Scd2BatchProcessor.endAtColName, endAtMetadata)
+ case col =>
+ F.col(QuotingUtils.quoteIdentifier(col))
+ }
+ staged.select(outputColumns.toImmutableArraySeq: _*)
+ }
+
+ /**
+ * Return the schema field names of columns selected for history-tracking on
`df`:
+ * the eligible user-data columns (those not in [[ChangeArgs.keys]] or the
framework
+ * reserved set) filtered through [[ChangeArgs.trackHistorySelection]].
+ */
+ private def computeTrackedHistoryColumns(df: DataFrame): Seq[String] = {
+ val conf = df.sparkSession.sessionState.conf
+ val resolver = conf.resolver
+
+ val keyColNames = changeArgs.keys.map(_.name)
+ val reservedColNames = Scd2BatchProcessor.reservedFrameworkColNames
+
+ val eligibleSchema = StructType(df.schema.fields.filterNot { field =>
+ reservedColNames.exists(resolver(_, field.name)) ||
+ keyColNames.exists(resolver(_, field.name))
+ })
+
+ ColumnSelection
+ .applyToSchema(
+ schemaName = "trackHistorySelection",
+ schema = eligibleSchema,
+ columnSelection = changeArgs.trackHistorySelection,
+ caseSensitive = conf.caseSensitiveAnalysis
+ )
+ .fieldNames
+ .toImmutableArraySeq
+ }
}
/**
@@ -833,6 +1032,18 @@ object Scd2BatchProcessor {
private[autocdc] val nextEffectiveRecordStartAtColName: String =
s"${AutoCdcReservedNames.prefix}next_effective_record_start_at"
+ /**
+ * Names of temporary columns used by [[reconcileStartAndEndAt]] to stage
the recomputed
+ * [[startAtColName]] / [[endAtColName]] values before projecting them back
over the originals.
+ *
+ * Temporary in that the columns have no observable side effect or
persistence across
+ * microbatches.
+ */
+ private val finalStartAtColName: String =
+ s"${AutoCdcReservedNames.prefix}final_start_at"
+ private val finalEndAtColName: String =
+ s"${AutoCdcReservedNames.prefix}final_end_at"
+
/**
* 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
@@ -896,6 +1107,20 @@ private[autocdc] case class Scd2IntervalColumns(
* [[Scd2BatchProcessor.orderChronologicallyPerKeyWindow]].
*/
def effectiveRecordStartAt: Column = F.coalesce(recordStartAt, endAt)
+
+ /** The same interval columns read from the row `offset` positions earlier
in `window`. */
+ def lagBy(offset: Int, window: WindowSpec): Scd2IntervalColumns =
+ Scd2IntervalColumns(
+ F.lag(recordStartAt, offset).over(window),
+ F.lag(startAt, offset).over(window),
+ F.lag(endAt, offset).over(window))
+
+ /** The same interval columns read from the row `offset` positions later in
`window`. */
+ def leadBy(offset: Int, window: WindowSpec): Scd2IntervalColumns =
+ Scd2IntervalColumns(
+ F.lead(recordStartAt, offset).over(window),
+ F.lead(startAt, offset).over(window),
+ F.lead(endAt, offset).over(window))
}
object RowClassifier {
@@ -942,18 +1167,50 @@ object RowClassifier {
isOpenUpsert(row) || isClosedUpsert(row)
/**
- * Delete-representing row, encoded as an instantaneous (zero-width)
interval at `recordStartAt`
- * (`startAt == endAt == recordStartAt`). Never materializes in the target
table.
+ * Tombstone (delete-boundary) row, encoded as an instantaneous interval at
+ * `recordStartAt`. Never materializes in the target table, only in the aux
table.
*
- * User-data column values on these rows are not part of the SCD2 contract:
they may reflect
- * the originating delete event, the values of the upsert whose
closed-interval row was bisected
- * (when promoted from a decomposition tail), or be null altogether.
Reconciliation does not
- * consume these values for any semantic decision.
+ * User-data column values on tombstones are not part of the SCD2 contract:
they may
+ * reflect the originating delete event, the values of the upsert whose
closed-interval
+ * row was bisected (when the tombstone was promoted from a decomposition
tail), or be
+ * null altogether. Reconciliation does not consume these values for any
semantic
+ * decision.
*/
- private[autocdc] def isDeleteRepresentingRow(row: Scd2IntervalColumns):
Column =
+ private[autocdc] def isTombstone(row: Scd2IntervalColumns): Column =
row.recordStartAt.isNotNull &&
row.startAt.isNotNull &&
row.endAt.isNotNull &&
row.startAt === row.recordStartAt &&
row.endAt === row.recordStartAt
+
+ /**
+ * Whether a row closes (`endAt`) strictly before the next
chronologically-ordered row for the
+ * same key begins (`nextEffectiveRecordStartAt`), leaving a gap in the
visible timeline.
+ */
+ private[autocdc] def rowClosesStrictlyBeforeNextRow(
+ endAt: Column,
+ nextEffectiveRecordStartAt: Column
+ ): Column =
+ endAt.isNotNull && endAt < nextEffectiveRecordStartAt
+
+ /**
+ * Whether `row` carries no new information beyond its immediate successor
`next` and so
+ * collapses into that successor's run instead of standing as its own
visible interval. It is
+ * the caller's responsibility to pass `row` and `next` as successive rows
in chronological
+ * order, and `areTrackedColumnsEqual` as true iff the two rows hold equal
values for every
+ * tracked-history column.
+ *
+ * Returns true iff `row` and `next` are both upsert-representing, `row`'s
interval reaches
+ * `next` without leaving a gap, and the two are tracked-history equal. It
is false whenever
+ * there is no successor (the last row in a key window) or either row is not
upsert-representing.
+ */
+ private[autocdc] def isNoOpUpsertContinuation(
+ row: Scd2IntervalColumns,
+ next: Scd2IntervalColumns,
+ areTrackedColumnsEqual: Column
+ ): Column =
+ isUpsertRepresentingRow(row) &&
+ isUpsertRepresentingRow(next) &&
+ !rowClosesStrictlyBeforeNextRow(row.endAt, next.effectiveRecordStartAt)
&&
+ areTrackedColumnsEqual
}
diff --git
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/ChangeArgsSuite.scala
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/ChangeArgsSuite.scala
index 7be111003762..bf6a89652353 100644
---
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/ChangeArgsSuite.scala
+++
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/ChangeArgsSuite.scala
@@ -17,7 +17,7 @@
package org.apache.spark.sql.pipelines.autocdc
-import org.apache.spark.SparkFunSuite
+import org.apache.spark.{SparkException, SparkFunSuite}
import org.apache.spark.sql.{functions => F, AnalysisException, Row}
import org.apache.spark.sql.catalyst.parser.ParseException
import org.apache.spark.sql.test.SharedSparkSession
@@ -377,6 +377,46 @@ class ChangeArgsSuite extends SparkFunSuite with
SharedSparkSession {
)
}
+ test("ChangeArgs rejects trackHistorySelection under SCD1") {
+ // SCD1 has no run concept, so a non-None trackHistorySelection is
meaningless. User-facing
+ // validation is expected at the API layer; this defensive internal guard
raises an internal
+ // error so downstream SCD1 code can assume the selection is None.
+ val ex = intercept[SparkException] {
+ ChangeArgs(
+ keys = Seq(UnqualifiedColumnName("id")),
+ sequencing = F.col("seq"),
+ storedAsScdType = ScdType.Type1,
+ trackHistorySelection = Some(
+ ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("Name")))
+ )
+ )
+ }
+ assert(ex.getCondition == "INTERNAL_ERROR")
+ }
+
+ test("ChangeArgs allows trackHistorySelection under SCD2") {
+ // The same selection is valid under SCD2, where tracking columns define a
run.
+ val args = ChangeArgs(
+ keys = Seq(UnqualifiedColumnName("id")),
+ sequencing = F.col("seq"),
+ storedAsScdType = ScdType.Type2,
+ trackHistorySelection = Some(
+ ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("Name")))
+ )
+ )
+ assert(args.trackHistorySelection.isDefined)
+ }
+
+ test("ChangeArgs allows None trackHistorySelection under SCD1") {
+ // The default None selection is always valid under SCD1.
+ val args = ChangeArgs(
+ keys = Seq(UnqualifiedColumnName("id")),
+ sequencing = F.col("seq"),
+ storedAsScdType = ScdType.Type1
+ )
+ assert(args.trackHistorySelection.isEmpty)
+ }
+
test("UnqualifiedColumnName lets a ParseException from the SQL parser
propagate") {
checkError(
exception = intercept[ParseException] {
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 c29aa50e9e72..649126222a24 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
@@ -18,7 +18,7 @@
package org.apache.spark.sql.pipelines.autocdc
import org.apache.spark.{SparkException, SparkRuntimeException}
-import org.apache.spark.sql.{functions => F, Column, QueryTest, Row}
+import org.apache.spark.sql.{functions => F, AnalysisException, Column,
QueryTest, Row}
import org.apache.spark.sql.classic.DataFrame
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
@@ -97,14 +97,18 @@ class Scd2BatchProcessorSuite extends QueryTest with
SharedSparkSession {
*/
private def processorWithKeys(
keys: Seq[String],
- deleteCondition: Option[Column] = None
+ deleteCondition: Option[Column] = None,
+ columnSelection: Option[ColumnSelection] = None,
+ trackHistorySelection: Option[ColumnSelection] = None
): Scd2BatchProcessor =
Scd2BatchProcessor(
changeArgs = ChangeArgs(
keys = keys.map(UnqualifiedColumnName(_)),
sequencing = F.col("seq"),
storedAsScdType = ScdType.Type2,
- deleteCondition = deleteCondition
+ deleteCondition = deleteCondition,
+ columnSelection = columnSelection,
+ trackHistorySelection = trackHistorySelection
),
resolvedSequencingType = LongType
)
@@ -1910,4 +1914,751 @@ class Scd2BatchProcessorSuite extends QueryTest with
SharedSparkSession {
)
)
}
+
+ // =============== reconcileStartAndEndAt tests ===============
+
+ test("reconcileStartAndEndAt: a fresh-key run head propagates its startAt to
its " +
+ "no-op continuation") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ // Two open upserts at recordStartAt=5 and recordStartAt=10 with identical
tracked
+ // values. The first row begins a fresh run with startAt=5. The second
row, sharing the
+ // tracked value, is a continuation of that run and inherits the run
head's startAt.
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", 5L, null, Row(5L)),
+ Row(1, "alice", 10L, null, Row(10L))
+ )
+
+ val result = processor.reconcileStartAndEndAt(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", 5L, null, Row(5L)),
+ Row(1, "alice", 5L, null, Row(10L))
+ )
+ )
+ }
+
+ test("reconcileStartAndEndAt: a continuation that began before the affected
window " +
+ "keeps its earlier run start") {
+ 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
+ // 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.
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", 2L, null, Row(5L)),
+ Row(1, "alice", 10L, null, Row(10L))
+ )
+
+ val result = processor.reconcileStartAndEndAt(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", 2L, null, Row(5L)),
+ Row(1, "alice", 2L, null, Row(10L))
+ )
+ )
+ }
+
+ test("reconcileStartAndEndAt: a tracked-history transition opens a new run
anchored " +
+ "at the new value's recordStartAt") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ // Two open upserts that disagree on the tracked column. The transition
closes the
+ // first run at the second event's effective recordStartAt and starts a
new run whose
+ // startAt is the new event's recordStartAt.
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", 5L, null, Row(5L)),
+ Row(1, "bob", 10L, null, Row(10L))
+ )
+
+ val result = processor.reconcileStartAndEndAt(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", 5L, 10L, Row(5L)),
+ Row(1, "bob", 10L, null, Row(10L))
+ )
+ )
+ }
+
+ test("reconcileStartAndEndAt: a chain of no-op continuations all share a
single " +
+ "run start") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ // Three consecutive open upserts all agreeing on the tracked column form
one no-op
+ // run. Every row in the run must end up with the run head's startAt.
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", 5L, null, Row(5L)),
+ Row(1, "alice", 10L, null, Row(10L)),
+ Row(1, "alice", 15L, null, Row(15L))
+ )
+
+ val result = processor.reconcileStartAndEndAt(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", 5L, null, Row(5L)),
+ Row(1, "alice", 5L, null, Row(10L)),
+ Row(1, "alice", 5L, null, Row(15L))
+ )
+ )
+ }
+
+ test("reconcileStartAndEndAt: a chain of three no-op continuations collapses
then closes " +
+ "at a trailing transition") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ // Four open upserts: three tracked-equal rows (`alice`) that must all
collapse into one
+ // run, followed by a tracked-different row (`bob`) that opens a new run.
This exercises
+ // the interaction the classifier is easy to get wrong across a multi-row
collapse:
+ // - none of the two interior `alice` rows may be misclassified as a run
head (the
+ // LAG-side comparison must absorb each into its predecessor's run),
+ // - all three `alice` rows must inherit the head's startAt (5),
+ // - the last `alice` row (the run tail) must close at bob's
recordStartAt (20), not stay
+ // open and not close at any interior sequence.
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", 5L, null, Row(5L)),
+ Row(1, "alice", 10L, null, Row(10L)),
+ Row(1, "alice", 15L, null, Row(15L)),
+ Row(1, "bob", 20L, null, Row(20L))
+ )
+
+ val result = processor.reconcileStartAndEndAt(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", 5L, null, Row(5L)),
+ Row(1, "alice", 5L, null, Row(10L)),
+ Row(1, "alice", 5L, 20L, Row(15L)),
+ Row(1, "bob", 20L, null, Row(20L))
+ )
+ )
+ }
+
+ test("reconcileStartAndEndAt: default tracking treats every non-key selected
user column " +
+ "as tracked") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType()
+ .add("id", IntegerType)
+ .add("name", StringType)
+ .add("status", StringType)
+
+ // Default `trackHistorySelection` (None) treats every eligible selected
user column
+ // as tracked. With no columnSelection, both `name` and `status` are
selected. The
+ // two rows agree on `name` but disagree on `status`, so they start
distinct runs.
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", "active", 5L, null, Row(5L)),
+ Row(1, "alice", "inactive", 10L, null, Row(10L))
+ )
+
+ val result = processor.reconcileStartAndEndAt(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", "active", 5L, 10L, Row(5L)),
+ Row(1, "alice", "inactive", 10L, null, Row(10L))
+ )
+ )
+ }
+
+ test("reconcileStartAndEndAt: default tracking ignores columns excluded by
columnSelection") {
+ val processor = processorWithKeys(
+ keys = Seq("id"),
+ columnSelection = Some(
+ ColumnSelection.IncludeColumns(
+ Seq(UnqualifiedColumnName("id"), UnqualifiedColumnName("name"))
+ )
+ )
+ )
+ val sourceSchema = new StructType()
+ .add("id", IntegerType)
+ .add("seq", LongType)
+ .add("name", StringType)
+ .add("status", StringType)
+
+ // `status` differs but is excluded from the selected target schema.
Default tracking
+ // therefore considers only the selected non-key user column (`name`), and
the rows
+ // collapse into one no-op run.
+ val df = microbatchOf(sourceSchema)(
+ Row(1, 5L, "alice", "active"),
+ Row(1, 10L, "alice", "inactive")
+ )
+
+ val result =
processor.reconcileStartAndEndAt(processor.preprocessMicrobatch(df))
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", 5L, null, Row(5L)),
+ Row(1, "alice", 5L, null, Row(10L))
+ )
+ )
+ }
+
+ test("reconcileStartAndEndAt: trackHistorySelection cannot include a column
excluded by " +
+ "columnSelection") {
+ val processor = processorWithKeys(
+ keys = Seq("id"),
+ columnSelection = Some(
+ ColumnSelection.IncludeColumns(
+ Seq(UnqualifiedColumnName("id"), UnqualifiedColumnName("name"))
+ )
+ ),
+ trackHistorySelection = Some(
+ ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("status")))
+ )
+ )
+ val sourceSchema = new StructType()
+ .add("id", IntegerType)
+ .add("seq", LongType)
+ .add("name", StringType)
+ .add("status", StringType)
+ val df = microbatchOf(sourceSchema)(
+ Row(1, 5L, "alice", "active")
+ )
+
+ val ex = intercept[AnalysisException] {
+ processor.reconcileStartAndEndAt(processor.preprocessMicrobatch(df))
+ }
+ assert(ex.getCondition == "AUTOCDC_COLUMNS_NOT_FOUND_IN_SCHEMA")
+ }
+
+ test("reconcileStartAndEndAt: ExcludeColumns excludes only the listed
columns from " +
+ "the tracked set") {
+ val processor = processorWithKeys(
+ keys = Seq("id"),
+ trackHistorySelection = Some(
+ ColumnSelection.ExcludeColumns(Seq(UnqualifiedColumnName("status")))
+ )
+ )
+ val userSchema = new StructType()
+ .add("id", IntegerType)
+ .add("name", StringType)
+ .add("status", StringType)
+
+ // `status` is excluded from tracking, so the two rows are tracked-equal
on the
+ // remaining columns (`name`). They should collapse into a single no-op
run.
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", "active", 5L, null, Row(5L)),
+ Row(1, "alice", "inactive", 10L, null, Row(10L))
+ )
+
+ val result = processor.reconcileStartAndEndAt(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", "active", 5L, null, Row(5L)),
+ Row(1, "alice", "inactive", 5L, null, Row(10L))
+ )
+ )
+ }
+
+ test("reconcileStartAndEndAt: an empty effective tracked-history set
collapses every " +
+ "consecutive upsert pair into one run") {
+ val processor = processorWithKeys(
+ keys = Seq("id"),
+ trackHistorySelection = Some(
+ ColumnSelection.ExcludeColumns(
+ Seq(UnqualifiedColumnName("name"), UnqualifiedColumnName("status"))
+ )
+ )
+ )
+ val userSchema = new StructType()
+ .add("id", IntegerType)
+ .add("name", StringType)
+ .add("status", StringType)
+
+ // Both user data columns are excluded, so the effective tracked set is
empty. With
+ // nothing to compare, every consecutive upsert pair collapses into a
single run -
+ // even when the user-visible data differs on every column.
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", "active", 5L, null, Row(5L)),
+ Row(1, "bob", "inactive", 10L, null, Row(10L))
+ )
+
+ val result = processor.reconcileStartAndEndAt(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", "active", 5L, null, Row(5L)),
+ Row(1, "bob", "inactive", 5L, null, Row(10L))
+ )
+ )
+ }
+
+ test("reconcileStartAndEndAt: a tombstone's startAt and endAt pass through
unchanged") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ // An open upsert, then a tombstone at recordStartAt=10, then a new open
upsert. The
+ // tombstone is not an upsert and must not participate in any run; its
startAt/endAt
+ // pass through identically. The bracketing upserts close (10) and reopen
(15) around
+ // the tombstone.
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", 5L, null, Row(5L)),
+ Row(1, "alice", 10L, 10L, Row(10L)),
+ Row(1, "alice", 15L, null, Row(15L))
+ )
+
+ val result = processor.reconcileStartAndEndAt(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", 5L, 10L, Row(5L)),
+ Row(1, "alice", 10L, 10L, Row(10L)),
+ Row(1, "alice", 15L, null, Row(15L))
+ )
+ )
+ }
+
+ test("reconcileStartAndEndAt: a decomposition tail's startAt stays null and
its endAt " +
+ "passes through") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ // A head + tail pair: the head retains the run start while the tail
(recordStartAt
+ // null) is excluded from upsert reconciliation. The tail's startAt must
stay null
+ // and its endAt must pass through unchanged.
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", 5L, null, Row(5L)),
+ Row(1, "alice", null, 30L, Row(null))
+ )
+
+ val result = processor.reconcileStartAndEndAt(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", 5L, 30L, Row(5L)),
+ Row(1, "alice", null, 30L, Row(null))
+ )
+ )
+ }
+
+ test("reconcileStartAndEndAt: a bisecting event between a decomposition head
and tail closes " +
+ "the head at the event, not at the tail boundary") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ // The three-row shape produced when a previously-closed row [5, 30] is
bisected by a
+ // late-arriving, tracked-different event at recordStartAt=15: a
decomposition head
+ // (open, `alice`), the bisecting event (open, `bob`), and a decomposition
tail carrying
+ // the original right boundary 30 (recordStartAt null). The head must
close at the
+ // bisecting event's sequence (15), NOT at the tail's boundary (30) - the
bisecting event
+ // opens a new run in between. The bisecting event in turn closes at the
tail boundary
+ // (30), and the tail passes through unchanged.
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", 5L, null, Row(5L)),
+ Row(1, "bob", 15L, null, Row(15L)),
+ Row(1, "alice", null, 30L, Row(null))
+ )
+
+ val result = processor.reconcileStartAndEndAt(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", 5L, 15L, Row(5L)),
+ Row(1, "bob", 15L, 30L, Row(15L)),
+ Row(1, "alice", null, 30L, Row(null))
+ )
+ )
+ }
+
+ test("reconcileStartAndEndAt: a closed upsert that already closes before the
next event " +
+ "keeps its endAt") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ // A closed upsert [5, 15] is followed by a fresh-value run head at
recordStartAt=20.
+ // The closed upsert already ended at 15 - strictly before the next event
- so its
+ // endAt is left intact.
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", 5L, 15L, Row(5L)),
+ Row(1, "bob", 20L, null, Row(20L))
+ )
+
+ val result = processor.reconcileStartAndEndAt(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", 5L, 15L, Row(5L)),
+ Row(1, "bob", 20L, null, Row(20L))
+ )
+ )
+ }
+
+ test("reconcileStartAndEndAt: an open upsert closes at its successor's
effective sequence") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ // An open upsert at recordStartAt=5 is followed by a
tracked-history-different upsert
+ // at recordStartAt=20. The first run head must be closed at 20 because
the run ends
+ // there.
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", 5L, null, Row(5L)),
+ Row(1, "bob", 20L, null, Row(20L))
+ )
+
+ val result = processor.reconcileStartAndEndAt(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", 5L, 20L, Row(5L)),
+ Row(1, "bob", 20L, null, Row(20L))
+ )
+ )
+ }
+
+ test("reconcileStartAndEndAt: a tombstone coincident with a row that already
closes at the " +
+ "same sequence is preserved for the next transform to drop") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ // A closed upsert [5, 20] is followed by a tombstone at recordStartAt=20,
both
+ // ending at the same sequence. Reconciliation does not collapse them - it
leaves
+ // the now-redundant tombstone for the next transform to drop based on the
+ // shape locked in here.
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", 5L, 20L, Row(5L)),
+ Row(1, "alice", 20L, 20L, Row(20L))
+ )
+
+ val result = processor.reconcileStartAndEndAt(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", 5L, 20L, Row(5L)),
+ Row(1, "alice", 20L, 20L, Row(20L))
+ )
+ )
+ }
+
+ test("reconcileStartAndEndAt: a decomposition tail keeps its null
recordStartAt for " +
+ "downstream promotion") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ // Downstream transforms identify decomposition tails by recordStartAt =
null, so
+ // reconciliation must not synthesize a value into the tail's
_cdc_metadata.
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", 5L, null, Row(5L)),
+ Row(1, "alice", null, 30L, Row(null))
+ )
+
+ val result = processor.reconcileStartAndEndAt(df)
+
+ val tailRows = result.collect().filter(r => r.isNullAt(2))
+ assert(tailRows.length == 1)
+ val tailCdcMetadata = tailRows.head.getStruct(4)
+ assert(tailCdcMetadata.isNullAt(0))
+ }
+
+ test("reconcileStartAndEndAt preserves the input schema, column metadata,
and row count") {
+ val processor = processorWithKeys(Seq("id"))
+
+ def commentMetadata(comment: String): Metadata =
+ new MetadataBuilder().putString("comment", comment).build()
+
+ val cdcMetadataInnerSchema = new StructType().add(
+ Scd2BatchProcessor.recordStartAtFieldName,
+ LongType,
+ nullable = true,
+ metadata = commentMetadata("inner __RECORD_START_AT")
+ )
+
+ val schema = new StructType()
+ .add("id", IntegerType, nullable = false, metadata =
commentMetadata("user key"))
+ .add("value", StringType, nullable = true, metadata =
commentMetadata("user data"))
+ .add(
+ Scd2BatchProcessor.startAtColName, LongType, nullable = true,
+ metadata = commentMetadata("framework __START_AT"))
+ .add(
+ Scd2BatchProcessor.endAtColName, LongType, nullable = true,
+ metadata = commentMetadata("framework __END_AT"))
+ .add(
+ AutoCdcReservedNames.cdcMetadataColName, cdcMetadataInnerSchema,
nullable = false,
+ metadata = commentMetadata("framework _cdc_metadata"))
+
+ // Mix of canonical post-decomposition row shapes so we exercise multiple
reconciliation
+ // branches under the schema-preservation contract.
+ val df = microbatchOf(schema)(
+ Row(1, "alice", 5L, null, Row(5L)),
+ Row(1, "alice", null, 30L, Row(null)),
+ Row(1, "alice", 30L, 30L, Row(30L))
+ )
+
+ val result = processor.reconcileStartAndEndAt(df)
+
+ schema.fields.zip(result.schema.fields).foreach { case (in, out) =>
+ assert(in.name == out.name)
+ assert(in.dataType == out.dataType)
+ assert(in.nullable == out.nullable)
+ assert(in.metadata == out.metadata)
+ }
+ assert(result.count() == df.count())
+ }
+
+ test("reconcileStartAndEndAt: a single-event-per-key partition reconciles
without " +
+ "referring to neighbors") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ // Two keys, each with exactly one event - so neither partition has a
window predecessor
+ // or successor. Reconciliation must handle the missing neighbors cleanly
and pass the
+ // single rows through.
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", 5L, null, Row(5L)),
+ Row(2, "bob", 10L, 20L, Row(10L))
+ )
+
+ val result = processor.reconcileStartAndEndAt(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", 5L, null, Row(5L)),
+ Row(2, "bob", 10L, 20L, Row(10L))
+ )
+ )
+ }
+
+ test("reconcileStartAndEndAt: IncludeColumns selects only the listed columns
as tracked") {
+ val processor = processorWithKeys(
+ keys = Seq("id"),
+ trackHistorySelection = Some(
+ ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("name")))
+ )
+ )
+ val userSchema = new StructType()
+ .add("id", IntegerType)
+ .add("name", StringType)
+ .add("status", StringType)
+
+ // Only `name` is tracked. Two rows agreeing on name but differing on
status are
+ // tracked-equal and should collapse into one run.
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", "active", 5L, null, Row(5L)),
+ Row(1, "alice", "inactive", 10L, null, Row(10L))
+ )
+
+ val result = processor.reconcileStartAndEndAt(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", "active", 5L, null, Row(5L)),
+ Row(1, "alice", "inactive", 5L, null, Row(10L))
+ )
+ )
+ }
+
+ test("reconcileStartAndEndAt: a tracked-history column whose name contains a
dot is quoted " +
+ "and matched as a single column, not as a nested field path") {
+ // The backticks make `UnqualifiedColumnName` store the literal field name
"user.name".
+ val processor = processorWithKeys(
+ keys = Seq("id"),
+ trackHistorySelection = Some(
+
ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("`user.name`")))
+ )
+ )
+ val userSchema = new StructType()
+ .add("id", IntegerType)
+ .add("user.name", StringType)
+ .add("status", StringType)
+
+ // Only the dotted column is tracked. Two rows agreeing on "user.name" but
differing on
+ // status are tracked-equal and must collapse into a single run. Without
quoting,
+ // `F.col("user.name")` would be parsed as a nested-field access (struct
`user`, field
+ // `name`) and fail to resolve.
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", "active", 5L, null, Row(5L)),
+ Row(1, "alice", "inactive", 10L, null, Row(10L))
+ )
+
+ val result = processor.reconcileStartAndEndAt(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", "active", 5L, null, Row(5L)),
+ Row(1, "alice", "inactive", 5L, null, Row(10L))
+ )
+ )
+ }
+
+ test("reconcileStartAndEndAt: trackHistorySelection referring to an unknown
column raises " +
+ "AUTOCDC_COLUMNS_NOT_FOUND_IN_SCHEMA") {
+ val processor = processorWithKeys(
+ keys = Seq("id"),
+ trackHistorySelection = Some(
+
ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("does_not_exist")))
+ )
+ )
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", 5L, null, Row(5L))
+ )
+
+ val ex = intercept[AnalysisException] {
+ processor.reconcileStartAndEndAt(df)
+ }
+ assert(ex.getCondition == "AUTOCDC_COLUMNS_NOT_FOUND_IN_SCHEMA")
+ }
+
+ test("reconcileStartAndEndAt: null tracked-column values are treated as
tracked-equal") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("name",
StringType)
+
+ val df = targetTableOf(userSchema)(
+ Row(1, null, 5L, null, Row(5L)),
+ Row(1, null, 10L, null, Row(10L))
+ )
+
+ val result = processor.reconcileStartAndEndAt(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, null, 5L, null, Row(5L)),
+ Row(1, null, 5L, null, Row(10L))
+ )
+ )
+ }
+
+ test("reconcileStartAndEndAt: a no-op continuation followed by a tombstone
is absorbed " +
+ "into its run and closed at the tombstone") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", 5L, null, Row(5L)),
+ Row(1, "alice", 10L, null, Row(10L)),
+ Row(1, "alice", 15L, 15L, Row(15L))
+ )
+
+ val result = processor.reconcileStartAndEndAt(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", 5L, null, Row(5L)),
+ Row(1, "alice", 5L, 15L, Row(10L)),
+ Row(1, "alice", 15L, 15L, Row(15L))
+ )
+ )
+ }
+
+ test("reconcileStartAndEndAt isolates per-key reconciliation across multiple
key partitions") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ // Two keys, each with a fresh-key run head + tracked-equal continuation.
The two
+ // partitions must reconcile independently.
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", 5L, null, Row(5L)),
+ Row(1, "alice", 10L, null, Row(10L)),
+ Row(2, "bob", 20L, null, Row(20L)),
+ Row(2, "bob", 25L, null, Row(25L))
+ )
+
+ val result = processor.reconcileStartAndEndAt(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", 5L, null, Row(5L)),
+ Row(1, "alice", 5L, null, Row(10L)),
+ Row(2, "bob", 20L, null, Row(20L)),
+ Row(2, "bob", 20L, null, Row(25L))
+ )
+ )
+ }
+
+ test("reconcileStartAndEndAt: a closed upsert absorbed as a no-op
continuation has its " +
+ "endAt cleared and becomes open") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ // A closed upsert [5, 20] is immediately followed by a tracked-equal
upsert at
+ // recordStartAt=20. The closed upsert ends exactly where the next event
begins - not
+ // strictly before it - so it is absorbed into the run as a no-op
continuation: its
+ // endAt is reset to null (becoming open) and it keeps the run head's
startAt. The
+ // successor inherits the same run start. This is the documented "closed
upsert becomes
+ // open" transition, which every other no-op-continuation test leaves as a
no-op by
+ // starting from an already-open row.
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", 5L, 20L, Row(5L)),
+ Row(1, "alice", 20L, null, Row(20L))
+ )
+
+ val result = processor.reconcileStartAndEndAt(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", 5L, null, Row(5L)),
+ Row(1, "alice", 5L, null, Row(20L))
+ )
+ )
+ }
+
+ test("reconcileStartAndEndAt: tracked-equal upserts separated by a gap do
not coalesce; " +
+ "the closed run head keeps its endAt and the successor opens a new run") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ // A closed upsert [5, 10] is followed by an open upsert at
recordStartAt=15. Although
+ // the two rows are tracked-equal, the closed run head ends at 10 -
strictly before the
+ // successor begins at 15 - so the gap prevents coalescing. The closed row
is its own
+ // run head and keeps its endAt; the successor is a fresh window-local run
head anchored
+ // at its own recordStartAt. This exercises run-head startAt propagation
for a closed
+ // upsert, which the other run-head tests only cover with open rows.
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", 5L, 10L, Row(5L)),
+ Row(1, "alice", 15L, null, Row(15L))
+ )
+
+ val result = processor.reconcileStartAndEndAt(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", 5L, 10L, Row(5L)),
+ Row(1, "alice", 15L, null, Row(15L))
+ )
+ )
+ }
+
+ test("reconcileStartAndEndAt returns an empty DataFrame for empty input") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ val df = targetTableOf(userSchema)()
+
+ val result = processor.reconcileStartAndEndAt(df)
+
+ assert(result.collect().isEmpty)
+ assert(result.columns.toSeq == df.columns.toSeq)
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
