This is an automated email from the ASF dual-hosted git repository.
jose-torres pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 45a831a8cc07 [SPARK-57378][SDP] Implement SCD2 Batch Processor; Merge
Reconciled Rows into Aux and Target Tables
45a831a8cc07 is described below
commit 45a831a8cc079225461d67f540ccc5350b1f525c
Author: Andreas Neumann <[email protected]>
AuthorDate: Thu Jul 23 22:09:23 2026 -0700
[SPARK-57378][SDP] Implement SCD2 Batch Processor; Merge Reconciled Rows
into Aux and Target Tables
Note: This continues AnishMahto's #56457, opened from my fork while the
original author is out of office. It carries his implement merge into aux and
target + self-review commits and adds review fixes and additional test coverage
on top. #56457 should be closed in favor of this PR to keep review in one place.
### What changes were proposed in this pull request?
This is the next step in the SCD2 AutoCDC batch-processor series (following
SPARK-57356, "Cleanup Delete Encoding Rows Post-Reconciliation"). It adds the
final merge stage that writes a reconciled microbatch into the auxiliary and
target tables. Three methods are added to Scd2BatchProcessor:
- identifyAndTagAuxRows — tags each post-reconciliation row with
shouldRouteToAuxTableColName, marking which rows belong in the auxiliary table
(tombstones and other delete-boundary encodings) versus the visible target
table.
- mergeRowsIntoAuxiliaryTable — merges the reconciled rows tagged for the
auxiliary table into it, handling tombstone insertion/advancement and
deletedByBatchId-scoped garbage collection.
- mergeRowsIntoTargetTable — merges the reconciled visible rows into the
target table (the user-facing SCD2 history).
Together with the reconciliation and cleanup steps already merged, this
completes the per-microbatch transformation pipeline; the foreachBatch callback
that drives it end-to-end lands in the follow-up (SPARK-57395).
### Why are the changes needed?
The SCD2 batch processor is being built incrementally across the
SPARK-571xx series. After a microbatch is preprocessed, decomposed, reconciled,
and cleaned up, its rows must actually be written to the auxiliary and target
tables using the correct MERGE semantics. This PR supplies that final merge
step, which the engine's SCD2 execution path depends on.
### Does this PR introduce any user-facing change?
No. These are internal (private[autocdc]) batch-processor methods; SCD2
AutoCDC is not yet reachable end-to-end (the driving foreachBatch handler and
the API/planner wiring land in separate changes).
### How was this patch tested?
New unit tests:
- Scd2BatchProcessorMergeSuite — covers the aux/target merge behavior
(tombstone insertion and advancement, GC of superseded aux rows, target-table
history writes).
- Additional cases in Scd2BatchProcessorSuite — identifyAndTagAuxRows
routing, per-key window isolation, dotted / identifier-sensitive column names,
and combined-branch / multi-column merges.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8)
Closes #57444 from anew/SPARK-57378-merge-reconciled-rows.
Lead-authored-by: Andreas Neumann <[email protected]>
Co-authored-by: Anish Mahto <[email protected]>
Signed-off-by: Jose Torres <[email protected]>
(cherry picked from commit 0687ebf64c6370d55190f6b372f55119ed72831c)
Signed-off-by: Jose Torres <[email protected]>
---
.../sql/pipelines/autocdc/Scd2BatchProcessor.scala | 334 +++++++++++++++-
.../autocdc/Scd2BatchProcessorMergeSuite.scala | 440 +++++++++++++++++++++
.../autocdc/Scd2BatchProcessorSuite.scala | 310 +++++++++++++++
3 files changed, 1083 insertions(+), 1 deletion(-)
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 6e5af84216b5..34bd8a5d2929 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
@@ -20,11 +20,13 @@ package org.apache.spark.sql.pipelines.autocdc
import org.apache.spark.SparkException
import org.apache.spark.sql.{functions => F}
import org.apache.spark.sql.Column
+import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.expressions.{CreateMap, If, Literal,
RaiseError}
import org.apache.spark.sql.catalyst.util.QuotingUtils
import org.apache.spark.sql.classic.{DataFrame, ExpressionUtils}
import org.apache.spark.sql.expressions.{Window, WindowSpec}
-import org.apache.spark.sql.types.{BooleanType, DataType, StringType,
StructField, StructType}
+import org.apache.spark.sql.types.{BooleanType, DataType, LongType,
StringType, StructField,
+ StructType}
import org.apache.spark.util.ArrayImplicits._
/**
@@ -942,6 +944,315 @@ case class Scd2BatchProcessor(
.fieldNames
.toImmutableArraySeq
}
+
+ /**
+ * Tag each post-reconciliation row with
[[Scd2BatchProcessor.shouldRouteToAuxTableColName]]:
+ * `true` for rows that belong in the auxiliary table (tombstones and hidden
no-op upserts),
+ * `false` for rows that do not. The flag's contract is solely about
aux-table membership; it
+ * makes no claim about whether a `false` row belongs in the target table.
+ *
+ * @param reconciledDf the canonical post-reconciliation rows for the
affected keys.
+ * @return `reconciledDf` with an added boolean
+ * [[Scd2BatchProcessor.shouldRouteToAuxTableColName]] column; no
rows are added or
+ * removed.
+ */
+ private[autocdc] def identifyAndTagAuxRows(reconciledDf: DataFrame):
DataFrame = {
+ val recordStartAt =
+
Scd2BatchProcessor.recordStartAtOf(F.col(AutoCdcReservedNames.cdcMetadataColName))
+ val startAt = F.col(Scd2BatchProcessor.startAtColName)
+ val endAt = F.col(Scd2BatchProcessor.endAtColName)
+ val current = Scd2IntervalColumns(recordStartAt, startAt, endAt)
+ val next = current.leadBy(1, orderChronologicallyPerKeyWindow)
+
+ val trackedHistoryColumns = computeTrackedHistoryColumns(reconciledDf)
+ val areTrackedColumnsEqualInNextRow = trackedHistoryColumns
+ .map { c =>
+ val col = F.col(QuotingUtils.quoteIdentifier(c))
+ col <=> F.lead(col, 1).over(orderChronologicallyPerKeyWindow)
+ }
+ .reduceOption(_ && _)
+ .getOrElse(F.lit(true))
+
+ val isTombstone = RowClassifier.isTombstone(current)
+ val isHiddenNoOpUpsert = RowClassifier.isNoOpUpsertContinuation(
+ row = current,
+ next = next,
+ areTrackedColumnsEqual = areTrackedColumnsEqualInNextRow
+ )
+
+ reconciledDf.withColumn(
+ Scd2BatchProcessor.shouldRouteToAuxTableColName,
+ isTombstone || isHiddenNoOpUpsert
+ )
+ }
+
+ /**
+ * Merge the reconciled rows that belong in the auxiliary table (tombstones
and hidden no-op
+ * upserts, as tagged by [[identifyAndTagAuxRows]]) onto the auxiliary
table, and logically
+ * delete any previously-affected aux row that did not survive
reconciliation.
+ *
+ * Idempotency across `foreachBatch` retries: rows this batch logically
deletes are not
+ * physically removed but stamped with
[[Scd2BatchProcessor.deletedByBatchIdColName]] `=
+ * batchId`, so a retry of the same `batchId` still observes them via
+ * [[findAffectedRowsFromAuxiliaryTable]] and re-derives the same
reconciliation output. Aux
+ * rows logically deleted by an older, already-committed batch
+ * ([[Scd2BatchProcessor.deletedByBatchIdColName]] `!= batchId`) are
physically
+ * garbage-collected as part of this same merge.
+ *
+ * @param reconciledDfWithAuxRowsTagged reconciled rows tagged by
[[identifyAndTagAuxRows]].
+ * @param originalAffectedRowsFromAuxiliaryTable the affected aux rows
pulled in for this
+ * microbatch, in canonical SCD2 row schema (i.e. as returned by
+ * [[findAffectedRowsFromAuxiliaryTable]], with the aux-only
+ * [[Scd2BatchProcessor.deletedByBatchIdColName]] already dropped).
+ * @param auxiliaryTableIdentifier the identifier of the auxiliary table to
merge into.
+ * @param batchId the underlying Spark streaming query's batchId, used to
stamp logical deletes
+ * and scope garbage collection.
+ */
+ private[autocdc] def mergeRowsIntoAuxiliaryTable(
+ reconciledDfWithAuxRowsTagged: DataFrame,
+ originalAffectedRowsFromAuxiliaryTable: DataFrame,
+ auxiliaryTableIdentifier: TableIdentifier,
+ batchId: Long): Unit = {
+ val resolver =
reconciledDfWithAuxRowsTagged.sparkSession.sessionState.conf.resolver
+ val deletedByBatchIdCol = Scd2BatchProcessor.deletedByBatchIdColName
+
+ val reconciledAuxRows = reconciledDfWithAuxRowsTagged
+ .filter(F.col(Scd2BatchProcessor.shouldRouteToAuxTableColName))
+ .drop(Scd2BatchProcessor.shouldRouteToAuxTableColName)
+
+ // All of the reconciledAuxRows will be landing in the aux table as alive
(non-deleted) rows,
+ // so tag them with a null deleted-by batch id.
+ val auxRowsToUpsert = reconciledAuxRows
+ .withColumn(deletedByBatchIdCol, F.lit(null).cast(LongType))
+
+ // Any aux row pulled in for reconciliation but absent from the
post-reconciliation aux rows
+ // was either dropped as redundant or promoted to the (now-visible) target
table. Either way
+ // it must leave the aux table; stamp it with this batch's id for
deleted-by batch id. On this
+ // batch's MERGE, the existing aux row will be considered logically
deleted. In a future
+ // microbatch's merge, it will be physically deleted.
+ val auxRowsToDelete = antiJoinRowsByRecordStartAtPerKey(
+ leftRows = originalAffectedRowsFromAuxiliaryTable,
+ rightRows = reconciledAuxRows
+ )
+ .withColumn(deletedByBatchIdCol, F.lit(batchId))
+
+ val mergeSource = auxRowsToUpsert
+ .unionByName(auxRowsToDelete)
+ .as("source")
+
+ // At this point [[mergeSource]] must have the same columns/shape as the
persisted aux table.
+ val auxTableColumns = mergeSource.columns.toImmutableArraySeq
+
+ // Build predicate for whether rows represent the same event (match). In
SCD2 each key can have
+ // multiple records, so a row's identity is defined by (keys, sequencing).
+ val auxIdentQuoted = auxiliaryTableIdentifier.quotedString
+ val meta = AutoCdcReservedNames.cdcMetadataColName
+
+ val mergeSourceRecordStartAt =
+ Scd2BatchProcessor.recordStartAtOf(F.col(s"source.`$meta`"))
+ val auxTableRecordStartAt =
+ Scd2BatchProcessor.recordStartAtOf(F.col(s"$auxIdentQuoted.`$meta`"))
+
+ val doKeysMatch = changeArgs.keys
+ .map(k => F.col(s"source.${k.quoted}") ===
F.col(s"$auxIdentQuoted.${k.quoted}"))
+ .reduce(_ && _)
+ val doRowsMatch = doKeysMatch && (mergeSourceRecordStartAt <=>
auxTableRecordStartAt)
+
+ // On updates, MERGE requires only non-key columns are updated (remapped).
For inserts, all of
+ // the row's columns must explicitly be mapped.
+ val keyNames = changeArgs.keys.map(_.name)
+ def upsertAssignments(columnName: String): (String, Column) = {
+ val quoted = QuotingUtils.quoteIdentifier(columnName)
+ s"$auxIdentQuoted.$quoted" -> F.col(s"source.$quoted")
+ }
+ val nonKeyUpdateAssignments = auxTableColumns
+ .filterNot(c => keyNames.exists(resolver(_, c)))
+ .map(upsertAssignments)
+ .toMap
+ val insertAssignments = auxTableColumns.map(upsertAssignments).toMap
+
+ // Physically garbage-collect aux rows logically deleted by some other,
already-committed
+ // batch. Such rows are always excluded when pulling in the current
microbatch's affected set,
+ // so they can never match against a row being merged in.
+ val auxTableDeletedByBatchId =
F.col(s"$auxIdentQuoted.`$deletedByBatchIdCol`")
+ val isGarbageCollectableAuxRow =
+ auxTableDeletedByBatchId.isNotNull && auxTableDeletedByBatchId =!=
F.lit(batchId)
+
+ // Whether a row in the MERGE source represents a row being [logically]
deleted, as opposed to
+ // being upserted.
+ val shouldLogicallyDeleteAuxRow =
F.col(s"source.`$deletedByBatchIdCol`").isNotNull
+
+ mergeSource
+ .mergeInto(auxIdentQuoted, doRowsMatch)
+ // Keys in source row match against existing row in aux table, and
declare intent to delete
+ // the corresponding row in the aux; mark the row as logically deleted
in the aux table.
+ .whenMatched(shouldLogicallyDeleteAuxRow)
+ .update(
+ Map(
+ s"$auxIdentQuoted.`$deletedByBatchIdCol`" ->
+ F.col(s"source.`$deletedByBatchIdCol`")
+ )
+ )
+ // Keys in source row match against existing row in aux table and does
not represent a row
+ // being deleted; update the data/operational columns
+ .whenMatched(!shouldLogicallyDeleteAuxRow)
+ .update(nonKeyUpdateAssignments)
+ // Keys in source do not match against an existing row in aux table and
does not represent a
+ // row being deleted; insert the new key's row.
+ .whenNotMatched(!shouldLogicallyDeleteAuxRow)
+ .insert(insertAssignments)
+ // If this is a row not affected by the current microbatch but is
eligible for garbage
+ // collection now, proactively hard-delete it.
+ // TODO: This GC triggers a full scan of the aux table; revisit whether
to GC only
+ // periodically rather than on every microbatch.
+ .whenNotMatchedBySource(isGarbageCollectableAuxRow)
+ .delete()
+ .merge()
+ }
+
+ /**
+ * Merge the reconciled rows that are visible in the target table (run tails
that are
+ * upsert-representing and not routed to the aux table) onto the target
table, and delete any
+ * previously-affected target row that did not survive reconciliation.
+ *
+ * A previously-affected target row is deleted when it is absent from the
post-reconciliation
+ * visible rows. This covers both (a) a row reconciled away entirely (e.g.
closed by a
+ * coincident tombstone) and (b) a row demoted to a hidden aux row (e.g. a
previously-visible
+ * no-op tail superseded within its run by a later no-op upsert). In either
case the
+ * corresponding target row must go.
+ *
+ * @param reconciledDfWithAuxRowsTagged reconciled rows tagged by
[[identifyAndTagAuxRows]].
+ * @param affectedRowsFromTargetTable the affected target rows pulled in for
this microbatch.
+ * @param targetTableIdentifier the identifier of the target table to merge
into.
+ */
+ private[autocdc] def mergeRowsIntoTargetTable(
+ reconciledDfWithAuxRowsTagged: DataFrame,
+ affectedRowsFromTargetTable: DataFrame,
+ targetTableIdentifier: TableIdentifier): Unit = {
+ val targetIdentQuoted = targetTableIdentifier.quotedString
+ val meta = AutoCdcReservedNames.cdcMetadataColName
+
+ val resolver =
reconciledDfWithAuxRowsTagged.sparkSession.sessionState.conf.resolver
+ val keyNames = changeArgs.keys.map(_.name)
+
+ // Find the reconciled rows that should specifically land in the target
table, as per SCD2.
+ val recordStartAt =
+
Scd2BatchProcessor.recordStartAtOf(reconciledDfWithAuxRowsTagged.col(meta))
+ val startAt =
reconciledDfWithAuxRowsTagged.col(Scd2BatchProcessor.startAtColName)
+ val endAt =
reconciledDfWithAuxRowsTagged.col(Scd2BatchProcessor.endAtColName)
+ val isVisibleTargetRow =
+ // Aux table by definition only holds rows that should not be visible to
users, but are
+ // required for cross-microbatch stateful reconciliation (tombstones and
currently no-op
+ // upserts).
+ !F.col(Scd2BatchProcessor.shouldRouteToAuxTableColName) &&
+ // While rows headed to aux are mutually exclusive to rows that should
land in the target,
+ // it's possible there are rows that belong to neither aux nor target
(ex. decomposition
+ // tails). Only non-no-op upsert-representing rows should land in the
target, and no-op
+ // upsert-representing rows would have been marked as routed to aux
table; hence all
+ // remaining upserts should land in the target table.
+
RowClassifier.isUpsertRepresentingRow(Scd2IntervalColumns(recordStartAt,
startAt, endAt))
+
+ // Compute the set of affected rows that should be upserted back into the
target table, as well
+ // as rows that have been dropped or demoted out of the target table.
+ val reconciledTargetRows = reconciledDfWithAuxRowsTagged
+ .filter(isVisibleTargetRow)
+ .drop(Scd2BatchProcessor.shouldRouteToAuxTableColName)
+
+ val rowsToDeleteFromTarget = antiJoinRowsByRecordStartAtPerKey(
+ leftRows = affectedRowsFromTargetTable,
+ rightRows = reconciledTargetRows
+ )
+
+ // At this point, [[reconciledTargetRows]] shares the same columns/shape
as the target table.
+ val targetTableColumns = reconciledTargetRows.columns.toImmutableArraySeq
+
+ // Unlike the aux table the target table does not persist any delete
marker (logical or not),
+ // so we need to augment a column to all MERGE source rows to indicate
whether the row
+ // represents that a corresponding target table row should be deleted.
Once we explicitly mark
+ // each row as a deletion or not, we can union them without losing
information.
+ val shouldDeleteTargetRowCol =
Scd2BatchProcessor.shouldDeleteTargetRowColName
+ val mergeSource = reconciledTargetRows
+ .withColumn(shouldDeleteTargetRowCol, F.lit(false))
+ .unionByName(rowsToDeleteFromTarget.withColumn(shouldDeleteTargetRowCol,
F.lit(true)))
+ .select(
+ targetTableColumns.map(c => F.col(QuotingUtils.quoteIdentifier(c))) :+
+ F.col(shouldDeleteTargetRowCol): _*
+ )
+ .as("source")
+
+ val mergeSourceRecordStartAt =
Scd2BatchProcessor.recordStartAtOf(F.col(s"source.`$meta`"))
+ val targetTableRecordStartAt =
+ Scd2BatchProcessor.recordStartAtOf(F.col(s"$targetIdentQuoted.`$meta`"))
+ val doKeysMatch = changeArgs.keys
+ .map(k => F.col(s"source.${k.quoted}") ===
F.col(s"$targetIdentQuoted.${k.quoted}"))
+ .reduce(_ && _)
+ val doRowsMatch = doKeysMatch && (mergeSourceRecordStartAt <=>
targetTableRecordStartAt)
+
+ // On updates, MERGE requires only non-key columns are updated (remapped).
For inserts, all of
+ // the row's columns must explicitly be mapped.
+ def upsertAssignments(columnName: String): (String, Column) = {
+ val quoted = QuotingUtils.quoteIdentifier(columnName)
+ s"$targetIdentQuoted.$quoted" -> F.col(s"source.$quoted")
+ }
+ val nonKeyUpdateAssignments = targetTableColumns
+ .filterNot(c => keyNames.exists(resolver(_, c)))
+ .map(upsertAssignments)
+ .toMap
+ val insertAssignments = targetTableColumns.map(upsertAssignments).toMap
+
+ // Whether a row in the MERGE source represents a row being deleted, as
opposed to being
+ // upserted. Unlike the aux table, target deletes are physical (hard)
deletes.
+ val shouldDeleteTargetRow = F.col(s"source.`$shouldDeleteTargetRowCol`")
+
+ mergeSource
+ .mergeInto(targetIdentQuoted, doRowsMatch)
+ // Keys in source row match against an existing row in the target table,
and declare intent
+ // to delete the corresponding row; physically remove it from the target
table.
+ .whenMatched(shouldDeleteTargetRow)
+ .delete()
+ // Keys in source row match against an existing row in the target table
and does not
+ // represent a row being deleted; update the data/operational columns.
+ .whenMatched(!shouldDeleteTargetRow)
+ .update(nonKeyUpdateAssignments)
+ // Keys in source do not match against an existing row in the target
table and does not
+ // represent a row being deleted; insert the new key's row.
+ .whenNotMatched(!shouldDeleteTargetRow)
+ .insert(insertAssignments)
+ .merge()
+ }
+
+ /**
+ * Left-anti-join `leftRows` against `rightRows`, matching on the key
columns and
+ * [[Scd2BatchProcessor.recordStartAtFieldName]] (null-safe). Returns the
`leftRows` that have
+ * no counterpart in `rightRows` - i.e. the rows that were dropped between
the two sets -
+ * preserving `leftRows`' schema.
+ *
+ * Both inputs must carry the key columns and the
[[AutoCdcReservedNames.cdcMetadataColName]]
+ * column.
+ */
+ private[autocdc] def antiJoinRowsByRecordStartAtPerKey(
+ leftRows: DataFrame,
+ rightRows: DataFrame): DataFrame = {
+ val leftAlias = "left"
+ val rightAlias = "right"
+ val leftRecordStartAt = Scd2BatchProcessor.recordStartAtOf(
+ F.col(s"$leftAlias.`${AutoCdcReservedNames.cdcMetadataColName}`")
+ )
+ val rightRecordStartAt = Scd2BatchProcessor.recordStartAtOf(
+ F.col(s"$rightAlias.`${AutoCdcReservedNames.cdcMetadataColName}`")
+ )
+ val doKeysMatch = changeArgs.keys
+ .map(k => F.col(s"$leftAlias.${k.quoted}") ===
F.col(s"$rightAlias.${k.quoted}"))
+ .reduce(_ && _)
+
+ leftRows
+ .as(leftAlias)
+ .join(
+ rightRows.as(rightAlias),
+ doKeysMatch && (leftRecordStartAt <=> rightRecordStartAt),
+ joinType = "left_anti"
+ )
+ }
}
/**
@@ -1166,6 +1477,27 @@ object Scd2BatchProcessor {
*/
private val anchorSequenceColName: String =
s"${AutoCdcReservedNames.prefix}anchor_sequence"
+ /**
+ * Name of the temporary column projected by
[[Scd2BatchProcessor.identifyAndTagAuxRows]] to
+ * mark rows destined for the auxiliary table (tombstones and hidden no-op
upserts) rather than
+ * the target table.
+ *
+ * Temporary in that the column has no observable side effect or persistence
across microbatches.
+ */
+ private[autocdc] val shouldRouteToAuxTableColName: String =
+ s"${AutoCdcReservedNames.prefix}should_route_to_aux_table"
+
+ /**
+ * Name of the temporary column used by
[[Scd2BatchProcessor.mergeRowsIntoTargetTable]] to flag,
+ * within the unioned merge source, which rows should be deleted from
(rather than upserted
+ * into) the target table.
+ *
+ * Temporary in that the column has no observable side effect or persistence
across
+ * microbatches.
+ */
+ private val shouldDeleteTargetRowColName: String =
+ s"${AutoCdcReservedNames.prefix}should_delete_target_row"
+
/** Project the [[recordStartAtFieldName]] out of an SCD2 CDC metadata
column. */
private def recordStartAtOf(cdcMetadataCol: Column): Column =
cdcMetadataCol.getField(recordStartAtFieldName)
diff --git
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessorMergeSuite.scala
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessorMergeSuite.scala
new file mode 100644
index 000000000000..f77a35fecd15
--- /dev/null
+++
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessorMergeSuite.scala
@@ -0,0 +1,440 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.pipelines.autocdc
+
+import org.scalatest.BeforeAndAfter
+
+import org.apache.spark.sql.{functions => F, QueryTest, Row}
+import org.apache.spark.sql.classic.DataFrame
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types._
+
+/**
+ * Catalog-backed unit tests for the SCD2 `MERGE INTO` routines
+ * [[Scd2BatchProcessor.mergeRowsIntoAuxiliaryTable]] and
+ * [[Scd2BatchProcessor.mergeRowsIntoTargetTable]]. These exercise every merge
branch (insert,
+ * update, delete, and - for the aux table - logical delete and garbage
collection) against an
+ * in-memory v2 row-level catalog.
+ */
+class Scd2BatchProcessorMergeSuite
+ extends QueryTest
+ with SharedSparkSession
+ with BeforeAndAfter
+ with AutoCdcCatalogExecutionTestBase {
+
+ private val userSchema = new StructType()
+ .add("id", IntegerType)
+ .add("value", StringType)
+
+ /** Canonical SCD2 row schema: user columns + framework start/end + cdc
metadata struct. */
+ private val canonicalSchema = userSchema
+ .add(Scd2BatchProcessor.startAtColName, LongType, nullable = true)
+ .add(Scd2BatchProcessor.endAtColName, LongType, nullable = true)
+ .add(
+ AutoCdcReservedNames.cdcMetadataColName,
+ Scd2BatchProcessor.cdcMetadataColSchema(LongType),
+ nullable = false
+ )
+
+ /** Auxiliary table schema: canonical schema plus the aux-only
logical-delete marker column. */
+ private val auxSchema = canonicalSchema
+ .add(Scd2BatchProcessor.deletedByBatchIdColName, LongType, nullable = true)
+
+ /** Tagged schema consumed by the merges: canonical schema plus the routing
flag. */
+ private val taggedSchema = canonicalSchema
+ .add(Scd2BatchProcessor.shouldRouteToAuxTableColName, BooleanType,
nullable = false)
+
+ /** User schema with a dotted (identifier-sensitive) non-key column name. */
+ private val dottedUserSchema = new StructType()
+ .add("id", IntegerType)
+ .add("user.name", StringType)
+
+ /** Canonical schema variant whose user column name contains a dot. */
+ private val dottedCanonicalSchema = dottedUserSchema
+ .add(Scd2BatchProcessor.startAtColName, LongType, nullable = true)
+ .add(Scd2BatchProcessor.endAtColName, LongType, nullable = true)
+ .add(
+ AutoCdcReservedNames.cdcMetadataColName,
+ Scd2BatchProcessor.cdcMetadataColSchema(LongType),
+ nullable = false
+ )
+
+ /** Aux schema variant whose user column name contains a dot. */
+ private val dottedAuxSchema = dottedCanonicalSchema
+ .add(Scd2BatchProcessor.deletedByBatchIdColName, LongType, nullable = true)
+
+ /** Tagged schema variant whose user column name contains a dot. */
+ private val dottedTaggedSchema = dottedCanonicalSchema
+ .add(Scd2BatchProcessor.shouldRouteToAuxTableColName, BooleanType,
nullable = false)
+
+ /** User schema with several non-key columns, to exercise the non-key update
assignment map. */
+ private val wideUserSchema = new StructType()
+ .add("id", IntegerType)
+ .add("name", StringType)
+ .add("status", StringType)
+ .add("score", IntegerType)
+
+ /** Canonical schema variant with several non-key user columns. */
+ private val wideCanonicalSchema = wideUserSchema
+ .add(Scd2BatchProcessor.startAtColName, LongType, nullable = true)
+ .add(Scd2BatchProcessor.endAtColName, LongType, nullable = true)
+ .add(
+ AutoCdcReservedNames.cdcMetadataColName,
+ Scd2BatchProcessor.cdcMetadataColSchema(LongType),
+ nullable = false
+ )
+
+ /** Aux schema variant with several non-key user columns. */
+ private val wideAuxSchema = wideCanonicalSchema
+ .add(Scd2BatchProcessor.deletedByBatchIdColName, LongType, nullable = true)
+
+ /** Tagged schema variant with several non-key user columns. */
+ private val wideTaggedSchema = wideCanonicalSchema
+ .add(Scd2BatchProcessor.shouldRouteToAuxTableColName, BooleanType,
nullable = false)
+
+ private val processor = Scd2BatchProcessor(
+ changeArgs = ChangeArgs(
+ keys = Seq(UnqualifiedColumnName("id")),
+ sequencing = F.col("seq"),
+ storedAsScdType = ScdType.Type2
+ ),
+ resolvedSequencingType = LongType
+ )
+
+ /** Build a tagged-rows [[DataFrame]] (canonical columns + routing flag) for
the merges. */
+ private def taggedOf(rows: Row*): DataFrame =
microbatchOf(taggedSchema)(rows: _*)
+
+ /** Build a canonical SCD2 [[DataFrame]] (e.g. the `findAffected*` outputs)
for the merges. */
+ private def canonicalOf(rows: Row*): DataFrame =
microbatchOf(canonicalSchema)(rows: _*)
+
+ private def createAuxTable(seedRows: Row*): Unit =
+ createTable(defaultAuxIdent, defaultAuxTableIdentifier, auxSchema,
seedRows: _*)
+
+ private def createTargetTable(seedRows: Row*): Unit =
+ createTable(defaultTargetIdent, defaultTargetTableIdentifier,
canonicalSchema, seedRows: _*)
+
+ private def auxTable: DataFrame =
+ spark.read.table(defaultAuxTableIdentifier.quotedString)
+
+ private def targetTable: DataFrame =
+ spark.read.table(defaultTargetTableIdentifier.quotedString)
+
+ //
===========================================================================================
+ // mergeRowsIntoAuxiliaryTable tests
+ //
===========================================================================================
+
+ test("mergeRowsIntoAuxiliaryTable inserts a routed row absent from the aux
table") {
+ createAuxTable()
+
+ // A freshly-routed tombstone with no matching aux row is inserted live
(no logical-delete
+ // marker).
+ val tagged = taggedOf(Row(1, "x", 5L, 5L, Row(5L), true))
+ val affected = canonicalOf()
+
+ processor.mergeRowsIntoAuxiliaryTable(tagged, affected,
defaultAuxTableIdentifier, batchId = 1L)
+
+ checkAnswer(auxTable, Row(1, "x", 5L, 5L, Row(5L), null))
+ }
+
+ test("mergeRowsIntoAuxiliaryTable updates a surviving routed row in place") {
+ // An existing hidden no-op upsert at recordStartAt=5 that survives
reconciliation.
+ createAuxTable(Row(1, "old", 5L, null, Row(5L), null))
+
+ val tagged = taggedOf(Row(1, "new", 5L, null, Row(5L), true))
+ // The affected row survives (still present in the routed set), so it is
not logically
+ // deleted; only its non-key columns are refreshed.
+ val affected = canonicalOf(Row(1, "old", 5L, null, Row(5L)))
+
+ processor.mergeRowsIntoAuxiliaryTable(tagged, affected,
defaultAuxTableIdentifier, batchId = 2L)
+
+ checkAnswer(auxTable, Row(1, "new", 5L, null, Row(5L), null))
+ }
+
+ test("mergeRowsIntoAuxiliaryTable logically deletes an affected row that did
not survive") {
+ // A tombstone pulled in as affected but absent from this batch's routed
set.
+ createAuxTable(Row(1, "gone", 7L, 7L, Row(7L), null))
+
+ // The single tagged row is not routed to the aux table, so the affected
aux row has no
+ // surviving counterpart and must be logically deleted.
+ val tagged = taggedOf(Row(1, "vis", 10L, null, Row(10L), false))
+ val affected = canonicalOf(Row(1, "gone", 7L, 7L, Row(7L)))
+
+ processor.mergeRowsIntoAuxiliaryTable(tagged, affected,
defaultAuxTableIdentifier, batchId = 3L)
+
+ // Logical, not physical: the row stays but is stamped with this batch's
id.
+ checkAnswer(auxTable, Row(1, "gone", 7L, 7L, Row(7L), 3L))
+ }
+
+ test("mergeRowsIntoAuxiliaryTable garbage-collects rows logically deleted by
an older batch") {
+ createAuxTable(
+ // Logically deleted by an older batch (2 != 5) -> physically
garbage-collected.
+ Row(1, "gc-old", 7L, 7L, Row(7L), 2L),
+ // Still live (no marker) -> retained.
+ Row(2, "live", 3L, null, Row(3L), null),
+ // Logically deleted by the current batch (5 == 5) -> retained for
replay-stability.
+ Row(3, "this-batch", 4L, 4L, Row(4L), 5L)
+ )
+
+ // One brand-new routed insert keeps the merge source non-empty; none of
the seeded rows are
+ // in the affected set, so they are all evaluated by the
not-matched-by-source GC clause.
+ val tagged = taggedOf(Row(10, "newtomb", 8L, 8L, Row(8L), true))
+ val affected = canonicalOf()
+
+ processor.mergeRowsIntoAuxiliaryTable(tagged, affected,
defaultAuxTableIdentifier, batchId = 5L)
+
+ checkAnswer(
+ auxTable,
+ Seq(
+ Row(2, "live", 3L, null, Row(3L), null),
+ Row(3, "this-batch", 4L, 4L, Row(4L), 5L),
+ Row(10, "newtomb", 8L, 8L, Row(8L), null)
+ )
+ )
+ }
+
+ test("mergeRowsIntoAuxiliaryTable is replay-stable: re-running the same
batchId is a no-op") {
+ createAuxTable(
+ // Survives reconciliation -> re-upserted in place.
+ Row(1, "old", 5L, null, Row(5L), null),
+ // Affected but does not survive -> logically deleted by this batch.
+ Row(2, "gone", 7L, 7L, Row(7L), null)
+ )
+
+ val tagged = taggedOf(Row(1, "new", 5L, null, Row(5L), true))
+ val affected = canonicalOf(
+ Row(1, "old", 5L, null, Row(5L)),
+ Row(2, "gone", 7L, 7L, Row(7L))
+ )
+
+ // First application of the batch.
+ processor.mergeRowsIntoAuxiliaryTable(
+ tagged, affected, defaultAuxTableIdentifier, batchId = 7L)
+ val afterFirstRun = auxTable.collect().sortBy(_.getInt(0)).toSeq
+
+ // Replaying the exact same batch (same batchId, re-derived inputs) must
leave the aux table
+ // byte-identical: the surviving row stays updated in place, and the
non-surviving row stays
+ // logically deleted by this same batch rather than physically
garbage-collected (7 ==
+ // batchId, so the GC clause does not fire).
+ processor.mergeRowsIntoAuxiliaryTable(
+ tagged, affected, defaultAuxTableIdentifier, batchId = 7L)
+ val afterSecondRun = auxTable.collect().sortBy(_.getInt(0)).toSeq
+
+ assert(afterFirstRun == afterSecondRun)
+ checkAnswer(
+ auxTable,
+ Seq(
+ Row(1, "new", 5L, null, Row(5L), null),
+ Row(2, "gone", 7L, 7L, Row(7L), 7L)
+ )
+ )
+ }
+
+ test("mergeRowsIntoAuxiliaryTable applies insert, update, logical-delete,
and GC in one merge") {
+ createAuxTable(
+ // Affected and survives reconciliation -> non-key columns updated in
place.
+ Row(1, "old", 5L, null, Row(5L), null),
+ // Affected but does not survive -> logically deleted by this batch
(stamped with batchId).
+ Row(2, "gone", 7L, 7L, Row(7L), null),
+ // Not affected, logically deleted by an older batch (4 != 9) ->
physically GC'd.
+ Row(3, "gc", 3L, 3L, Row(3L), 4L)
+ )
+
+ val tagged = taggedOf(
+ // Matches key 1 -> update.
+ Row(1, "new", 5L, null, Row(5L), true),
+ // New key 4 routed to aux -> insert.
+ Row(4, "ins", 9L, 9L, Row(9L), true)
+ )
+ // Keys 1 and 2 were pulled in as affected; key 1 survives in the routed
set, key 2 does not.
+ val affected = canonicalOf(
+ Row(1, "old", 5L, null, Row(5L)),
+ Row(2, "gone", 7L, 7L, Row(7L))
+ )
+
+ processor.mergeRowsIntoAuxiliaryTable(tagged, affected,
defaultAuxTableIdentifier, batchId = 9L)
+
+ checkAnswer(
+ auxTable,
+ Seq(
+ Row(1, "new", 5L, null, Row(5L), null),
+ Row(2, "gone", 7L, 7L, Row(7L), 9L),
+ Row(4, "ins", 9L, 9L, Row(9L), null)
+ // key 3 physically garbage-collected.
+ )
+ )
+ }
+
+ test("mergeRowsIntoAuxiliaryTable updates every non-key column of a
surviving row") {
+ createTable(
+ defaultAuxIdent,
+ defaultAuxTableIdentifier,
+ wideAuxSchema,
+ Row(1, "old-name", "active", 10, 5L, null, Row(5L), null)
+ )
+
+ // Every non-key column (name, status, score) differs from the seeded row;
the in-place update
+ // must refresh all of them, not just the first.
+ val tagged = microbatchOf(wideTaggedSchema)(
+ Row(1, "new-name", "inactive", 42, 5L, null, Row(5L), true)
+ )
+ val affected = microbatchOf(wideCanonicalSchema)(
+ Row(1, "old-name", "active", 10, 5L, null, Row(5L))
+ )
+
+ processor.mergeRowsIntoAuxiliaryTable(tagged, affected,
defaultAuxTableIdentifier, batchId = 1L)
+
+ checkAnswer(auxTable, Row(1, "new-name", "inactive", 42, 5L, null,
Row(5L), null))
+ }
+
+ test("mergeRowsIntoAuxiliaryTable handles user columns whose names contain a
dot") {
+ // A user column named `user.name`: the merge must quote it for both the
insert column list
+ // and the non-key update assignments. Without quoting, `user.name` would
be parsed as a
+ // nested-field access and the MERGE would fail to resolve the column.
+ createTable(defaultAuxIdent, defaultAuxTableIdentifier, dottedAuxSchema)
+
+ val tagged = microbatchOf(dottedTaggedSchema)(Row(1, "alice", 5L, 5L,
Row(5L), true))
+ val affected = microbatchOf(dottedCanonicalSchema)()
+
+ processor.mergeRowsIntoAuxiliaryTable(tagged, affected,
defaultAuxTableIdentifier, batchId = 1L)
+
+ checkAnswer(auxTable, Row(1, "alice", 5L, 5L, Row(5L), null))
+ }
+
+ //
===========================================================================================
+ // mergeRowsIntoTargetTable tests
+ //
===========================================================================================
+
+ test("mergeRowsIntoTargetTable inserts a visible upsert absent from the
target table") {
+ createTargetTable()
+
+ val tagged = taggedOf(Row(1, "v", 5L, null, Row(5L), false))
+ val affected = canonicalOf()
+
+ processor.mergeRowsIntoTargetTable(tagged, affected,
defaultTargetTableIdentifier)
+
+ checkAnswer(targetTable, Row(1, "v", 5L, null, Row(5L)))
+ }
+
+ test("mergeRowsIntoTargetTable updates a visible row matched at the same
recordStartAt") {
+ createTargetTable(Row(1, "old", 5L, null, Row(5L)))
+
+ // The run head at recordStartAt=5 is now closed at 20 with refreshed
data; it matches and
+ // updates the existing target row in place.
+ val tagged = taggedOf(Row(1, "new", 5L, 20L, Row(5L), false))
+ val affected = canonicalOf(Row(1, "old", 5L, null, Row(5L)))
+
+ processor.mergeRowsIntoTargetTable(tagged, affected,
defaultTargetTableIdentifier)
+
+ checkAnswer(targetTable, Row(1, "new", 5L, 20L, Row(5L)))
+ }
+
+ test("mergeRowsIntoTargetTable deletes an affected row reconciled away") {
+ createTargetTable(Row(1, "old", 5L, null, Row(5L)))
+
+ // The only tagged row routes to the aux table (e.g. closed into a
tombstone), so no visible
+ // row survives for key 1: the previously-affected target row must be
deleted.
+ val tagged = taggedOf(Row(1, "x", 5L, 5L, Row(5L), true))
+ val affected = canonicalOf(Row(1, "old", 5L, null, Row(5L)))
+
+ processor.mergeRowsIntoTargetTable(tagged, affected,
defaultTargetTableIdentifier)
+
+ assert(targetTable.collect().isEmpty)
+ }
+
+ test("mergeRowsIntoTargetTable only writes visible upsert rows, filtering
aux/non-upsert rows") {
+ createTargetTable()
+
+ // Only the open upsert (key 1) is visible. The tombstone (routed) and
hidden no-op (routed)
+ // are filtered by the routing flag; the decomposition tail is filtered
because it is not an
+ // upsert-representing row.
+ val tagged = taggedOf(
+ Row(1, "vis", 5L, null, Row(5L), false),
+ Row(2, "tomb", 7L, 7L, Row(7L), true),
+ Row(3, "tail", null, 9L, Row(null), false),
+ Row(4, "hidden", 5L, null, Row(5L), true)
+ )
+ val affected = canonicalOf()
+
+ processor.mergeRowsIntoTargetTable(tagged, affected,
defaultTargetTableIdentifier)
+
+ checkAnswer(targetTable, Row(1, "vis", 5L, null, Row(5L)))
+ }
+
+ test("mergeRowsIntoTargetTable applies insert, update, and delete branches
in one merge") {
+ createTargetTable(
+ Row(1, "old1", 5L, null, Row(5L)),
+ Row(2, "old2", 8L, null, Row(8L))
+ )
+
+ val tagged = taggedOf(
+ // Matches key 1 at recordStartAt=5 -> update.
+ Row(1, "new1", 5L, 30L, Row(5L), false),
+ // New key 3 -> insert.
+ Row(3, "ins3", 12L, null, Row(12L), false)
+ // Key 2 has no surviving visible row -> delete.
+ )
+ val affected = canonicalOf(
+ Row(1, "old1", 5L, null, Row(5L)),
+ Row(2, "old2", 8L, null, Row(8L))
+ )
+
+ processor.mergeRowsIntoTargetTable(tagged, affected,
defaultTargetTableIdentifier)
+
+ checkAnswer(
+ targetTable,
+ Seq(
+ Row(1, "new1", 5L, 30L, Row(5L)),
+ Row(3, "ins3", 12L, null, Row(12L))
+ )
+ )
+ }
+
+ test("mergeRowsIntoTargetTable handles user columns whose names contain a
dot") {
+ // A user column named `user.name`: the merge quotes it for the source
projection, the insert
+ // column list, and the non-key update assignments. Without quoting,
`user.name` would be
+ // parsed as a nested-field access and the MERGE would fail to resolve the
column.
+ createTable(defaultTargetIdent, defaultTargetTableIdentifier,
dottedCanonicalSchema)
+
+ val tagged = microbatchOf(dottedTaggedSchema)(Row(1, "alice", 5L, null,
Row(5L), false))
+ val affected = microbatchOf(dottedCanonicalSchema)()
+
+ processor.mergeRowsIntoTargetTable(tagged, affected,
defaultTargetTableIdentifier)
+
+ checkAnswer(targetTable, Row(1, "alice", 5L, null, Row(5L)))
+ }
+
+ test("mergeRowsIntoTargetTable updates every non-key column of a matched
row") {
+ createTable(defaultTargetIdent, defaultTargetTableIdentifier,
wideCanonicalSchema)
+ microbatchOf(wideCanonicalSchema)(
+ Row(1, "old-name", "active", 10, 5L, null, Row(5L))
+ ).writeTo(defaultTargetTableIdentifier.quotedString).append()
+
+ // Every non-key column (name, status, score) differs from the existing
row; the in-place
+ // update must refresh all of them, exercising the full non-key update
assignment map.
+ val tagged = microbatchOf(wideTaggedSchema)(
+ Row(1, "new-name", "inactive", 42, 5L, 30L, Row(5L), false)
+ )
+ val affected = microbatchOf(wideCanonicalSchema)(
+ Row(1, "old-name", "active", 10, 5L, null, Row(5L))
+ )
+
+ processor.mergeRowsIntoTargetTable(tagged, affected,
defaultTargetTableIdentifier)
+
+ checkAnswer(targetTable, Row(1, "new-name", "inactive", 42, 5L, 30L,
Row(5L)))
+ }
+}
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 0e671721bacc..42b294248413 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
@@ -3040,4 +3040,314 @@ class Scd2BatchProcessorSuite extends QueryTest with
SharedSparkSession {
)
)
}
+
+ // =============== identifyAndTagAuxRows tests ===============
+
+ /** Select `id`, `value`, and the appended routing flag for compact
assertions. */
+ private def withRouteFlag(df: DataFrame): DataFrame =
+ df.select(
+ F.col("id"),
+ F.col("value"),
+ F.col(Scd2BatchProcessor.startAtColName),
+ F.col(Scd2BatchProcessor.endAtColName),
+ F.col(AutoCdcReservedNames.cdcMetadataColName),
+ F.col(Scd2BatchProcessor.shouldRouteToAuxTableColName)
+ )
+
+ test("identifyAndTagAuxRows routes a tombstone to the auxiliary table") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ // A tombstone (startAt == endAt == recordStartAt) is delete-encoded and
must live in the
+ // aux table, never the target table.
+ val df = targetTableOf(userSchema)(
+ Row(1, "t", 5L, 5L, Row(5L))
+ )
+
+ checkAnswer(
+ df = withRouteFlag(processor.identifyAndTagAuxRows(df)),
+ expectedAnswer = Seq(
+ Row(1, "t", 5L, 5L, Row(5L), true)
+ )
+ )
+ }
+
+ test("identifyAndTagAuxRows does not route a decomposition tail to the
auxiliary table") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ // A decomposition tail (recordStartAt == null) is neither a tombstone nor
an
+ // upsert-representing row, so it is tagged false. (Tails are
promoted/dropped upstream of
+ // the merges; this pins that the router itself never claims them for the
aux table.)
+ val df = targetTableOf(userSchema)(
+ Row(1, "tail", null, 10L, Row(null))
+ )
+
+ checkAnswer(
+ df = withRouteFlag(processor.identifyAndTagAuxRows(df)),
+ expectedAnswer = Seq(
+ Row(1, "tail", null, 10L, Row(null), false)
+ )
+ )
+ }
+
+ test("identifyAndTagAuxRows keeps a lone open upsert visible (no successor
to coalesce with)") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ // A single open upsert with no following row in its key window cannot be
a hidden no-op
+ // continuation - it is the visible tail of its (size-1) run.
+ val df = targetTableOf(userSchema)(
+ Row(1, "v", 5L, null, Row(5L))
+ )
+
+ checkAnswer(
+ df = withRouteFlag(processor.identifyAndTagAuxRows(df)),
+ expectedAnswer = Seq(
+ Row(1, "v", 5L, null, Row(5L), false)
+ )
+ )
+ }
+
+ test("identifyAndTagAuxRows routes every no-op run member except the visible
tail to the aux") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ // A reconciled no-op run: three upserts agreeing on the tracked `value`,
all sharing the
+ // run head's startAt=5 and open endAt. The first two coalesce into hidden
aux rows; only
+ // the last (the run tail) stays visible in the target table.
+ val df = targetTableOf(userSchema)(
+ Row(1, "a", 5L, null, Row(5L)),
+ Row(1, "a", 5L, null, Row(10L)),
+ Row(1, "a", 5L, null, Row(15L))
+ )
+
+ checkAnswer(
+ df = withRouteFlag(processor.identifyAndTagAuxRows(df)),
+ expectedAnswer = Seq(
+ Row(1, "a", 5L, null, Row(5L), true),
+ Row(1, "a", 5L, null, Row(10L), true),
+ Row(1, "a", 5L, null, Row(15L), false)
+ )
+ )
+ }
+
+ test("identifyAndTagAuxRows keeps a row visible when the next row changes a
tracked column") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ // A real state change between the two rows (value a -> b) breaks the run,
so the earlier
+ // closed upsert is a visible run tail rather than a hidden no-op
continuation.
+ val df = targetTableOf(userSchema)(
+ Row(1, "a", 5L, 10L, Row(5L)),
+ Row(1, "b", 10L, null, Row(10L))
+ )
+
+ checkAnswer(
+ df = withRouteFlag(processor.identifyAndTagAuxRows(df)),
+ expectedAnswer = Seq(
+ Row(1, "a", 5L, 10L, Row(5L), false),
+ Row(1, "b", 10L, null, Row(10L), false)
+ )
+ )
+ }
+
+ test("identifyAndTagAuxRows keeps a row visible when it closes strictly
before the next row") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ // The first row closes at 8 but the next row only begins at 10, leaving a
visible gap. Even
+ // though both rows agree on the tracked `value`, the earlier row cannot
be hidden - dropping
+ // it would erase the gap from the visible timeline.
+ val df = targetTableOf(userSchema)(
+ Row(1, "a", 5L, 8L, Row(5L)),
+ Row(1, "a", 10L, null, Row(10L))
+ )
+
+ checkAnswer(
+ df = withRouteFlag(processor.identifyAndTagAuxRows(df)),
+ expectedAnswer = Seq(
+ Row(1, "a", 5L, 8L, Row(5L), false),
+ Row(1, "a", 10L, null, Row(10L), false)
+ )
+ )
+ }
+
+ test("identifyAndTagAuxRows coalesces consecutive upserts when the tracked
set is empty") {
+ val processor = processorWithKeys(
+ keys = Seq("id"),
+ trackHistorySelection = Some(
+ ColumnSelection.ExcludeColumns(Seq(UnqualifiedColumnName("value")))
+ )
+ )
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ // With `value` excluded the effective tracked set is empty, so
consecutive gapless upserts
+ // are always tracked-equal: the earlier one is hidden even though the
user data differs.
+ val df = targetTableOf(userSchema)(
+ Row(1, "a", 5L, null, Row(5L)),
+ Row(1, "b", 5L, null, Row(10L))
+ )
+
+ checkAnswer(
+ df = withRouteFlag(processor.identifyAndTagAuxRows(df)),
+ expectedAnswer = Seq(
+ Row(1, "a", 5L, null, Row(5L), true),
+ Row(1, "b", 5L, null, Row(10L), false)
+ )
+ )
+ }
+
+ test("identifyAndTagAuxRows evaluates runs independently per key") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ // Two keys, each with its own no-op run. The per-key window must never
let one key's rows
+ // influence another's: the last row of key 1 must stay visible (it is its
run's tail, not a
+ // hidden continuation of key 2's run), and key 2's first row must be
evaluated against key
+ // 2's successor only. A window that leaked across keys would mistag one
of these boundary
+ // rows.
+ val df = targetTableOf(userSchema)(
+ Row(1, "a", 5L, null, Row(5L)),
+ Row(1, "a", 5L, null, Row(10L)),
+ Row(2, "b", 7L, null, Row(7L)),
+ Row(2, "b", 7L, null, Row(20L))
+ )
+
+ checkAnswer(
+ df = withRouteFlag(processor.identifyAndTagAuxRows(df)),
+ expectedAnswer = Seq(
+ Row(1, "a", 5L, null, Row(5L), true),
+ Row(1, "a", 5L, null, Row(10L), false),
+ Row(2, "b", 7L, null, Row(7L), true),
+ Row(2, "b", 7L, null, Row(20L), false)
+ )
+ )
+ }
+
+ test("identifyAndTagAuxRows quotes tracked columns whose names contain a
dot") {
+ // 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)
+
+ // The dotted column is the tracked column. Two gapless upserts agreeing
on it form a no-op
+ // run, so the earlier row is hidden. Without quoting, the
tracked-equality comparison would
+ // parse `user.name` as a nested-field access (struct `user`, field
`name`) and fail to
+ // resolve.
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", 5L, null, Row(5L)),
+ Row(1, "alice", 5L, null, Row(10L))
+ )
+
+ val result = processor.identifyAndTagAuxRows(df)
+
+ checkAnswer(
+ df = result.select(
+ F.col("id"),
+ F.col("`user.name`"),
+ F.col(Scd2BatchProcessor.startAtColName),
+ F.col(Scd2BatchProcessor.endAtColName),
+ F.col(AutoCdcReservedNames.cdcMetadataColName),
+ F.col(Scd2BatchProcessor.shouldRouteToAuxTableColName)
+ ),
+ expectedAnswer = Seq(
+ Row(1, "alice", 5L, null, Row(5L), true),
+ Row(1, "alice", 5L, null, Row(10L), false)
+ )
+ )
+ }
+
+ // =============== antiJoinRowsByRecordStartAtPerKey tests ===============
+
+ test("antiJoinRowsByRecordStartAtPerKey returns only left rows with no (key,
recordStartAt) " +
+ "match on the right") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ val left = targetTableOf(userSchema)(
+ Row(1, "L5", 5L, null, Row(5L)),
+ Row(1, "L9", 9L, null, Row(9L))
+ )
+ // Right matches the left row at recordStartAt=5 only; the left row at 9
has no counterpart.
+ val right = targetTableOf(userSchema)(
+ Row(1, "R5", 5L, null, Row(5L))
+ )
+
+ checkAnswer(
+ df = processor.antiJoinRowsByRecordStartAtPerKey(left, right),
+ expectedAnswer = Seq(
+ Row(1, "L9", 9L, null, Row(9L))
+ )
+ )
+ }
+
+ test("antiJoinRowsByRecordStartAtPerKey matches per key: same recordStartAt
under a different " +
+ "key is not a match") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ val left = targetTableOf(userSchema)(
+ Row(1, "a", 5L, null, Row(5L)),
+ Row(2, "b", 5L, null, Row(5L))
+ )
+ // Only key 1 at recordStartAt=5 matches; key 2 at the same recordStartAt
must survive.
+ val right = targetTableOf(userSchema)(
+ Row(1, "r", 5L, null, Row(5L))
+ )
+
+ checkAnswer(
+ df = processor.antiJoinRowsByRecordStartAtPerKey(left, right),
+ expectedAnswer = Seq(
+ Row(2, "b", 5L, null, Row(5L))
+ )
+ )
+ }
+
+ test("antiJoinRowsByRecordStartAtPerKey matches null recordStartAt
null-safely") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ // Both sides carry a null recordStartAt for the same key. A null-safe
equality treats the
+ // two as matching, so the left row is anti-joined away (an ordinary `=`
would keep it).
+ val left = targetTableOf(userSchema)(
+ Row(1, "tailL", null, 10L, Row(null))
+ )
+ val right = targetTableOf(userSchema)(
+ Row(1, "tailR", null, 20L, Row(null))
+ )
+
+ assert(processor.antiJoinRowsByRecordStartAtPerKey(left,
right).collect().isEmpty)
+ }
+
+ test("antiJoinRowsByRecordStartAtPerKey matches on the full composite key") {
+ val processor = processorWithKeys(Seq("id", "grp"))
+ val userSchema = new StructType()
+ .add("id", IntegerType)
+ .add("grp", StringType)
+ .add("value", StringType)
+
+ // Rows share id and recordStartAt but differ on the second key column
`grp`. Only the exact
+ // composite-key match (1, "g") is removed; (1, "h") survives.
+ val left = targetTableOf(userSchema)(
+ Row(1, "g", "a", 5L, null, Row(5L)),
+ Row(1, "h", "b", 5L, null, Row(5L))
+ )
+ val right = targetTableOf(userSchema)(
+ Row(1, "g", "r", 5L, null, Row(5L))
+ )
+
+ checkAnswer(
+ df = processor.antiJoinRowsByRecordStartAtPerKey(left, right),
+ expectedAnswer = Seq(
+ Row(1, "h", "b", 5L, null, Row(5L))
+ )
+ )
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
