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 49636acc1f8f1..eb09ac59d1ca5 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 @@ -153,21 +153,28 @@ 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 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 + * column. None means every eligible user column (i.e. every + * source column that is neither a key nor a framework column) is + * considered tracked. Ignored under SCD1. */ 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) } 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 b0c511958ab76..2ee3dac35cf58 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,9 @@ 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}. 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 +532,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 +556,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 +639,193 @@ 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); a closed upsert may 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, + areTrackedColumnsEqualInNextRow = 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, + areTrackedColumnsEqualInNextRow = 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 the row's visible interval + // disappears and `endAt` is reset to null to route the row to the aux table. + .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 +1020,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 +1095,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 +1155,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 `areTrackedColumnsEqualInNextRow` 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, + areTrackedColumnsEqualInNextRow: Column + ): Column = + isUpsertRepresentingRow(row) && + isUpsertRepresentingRow(next) && + !rowClosesStrictlyBeforeNextRow(row.endAt, next.effectiveRecordStartAt) && + areTrackedColumnsEqualInNextRow } 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 c29aa50e9e721..29063986c89af 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,16 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { */ private def processorWithKeys( keys: Seq[String], - deleteCondition: Option[Column] = None + deleteCondition: Option[Column] = None, + trackHistorySelection: Option[ColumnSelection] = None ): Scd2BatchProcessor = Scd2BatchProcessor( changeArgs = ChangeArgs( keys = keys.map(UnqualifiedColumnName(_)), sequencing = F.col("seq"), storedAsScdType = ScdType.Type2, - deleteCondition = deleteCondition + deleteCondition = deleteCondition, + trackHistorySelection = trackHistorySelection ), resolvedSequencingType = LongType ) @@ -1910,4 +1912,558 @@ 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: default tracking treats every non-key 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) is documented to treat every eligible user + // column as tracked. The two rows agree on `name` but disagree on `status`, which + // should therefore start a new 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, 10L, Row(5L)), + Row(1, "alice", "inactive", 10L, null, Row(10L)) + ) + ) + } + + 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 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)) + ) + ) + } }