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 62ae4db28f3b [SPARK-57152][SDP] Implement SCD2 Batch Processor; Find
Affected Aux/Target Table Rows
62ae4db28f3b is described below
commit 62ae4db28f3be8a0ca2c3016d27ca5a62f02915d
Author: AnishMahto <[email protected]>
AuthorDate: Tue Jun 9 23:45:07 2026 -0700
[SPARK-57152][SDP] Implement SCD2 Batch Processor; Find Affected Aux/Target
Table Rows
Approved AutoCDC SPIP:
https://lists.apache.org/thread/j6sj9wo9odgdpgzlxtvhoy7szs0jplf7
--------
### 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.
**Find Affected Aux/Target Table Rows**
After preprocessing the microbatch such that we have each incoming row's
startAt, endAt, and recordStartAt projected, the next step in reconciliation is
determining which existing rows in the auxiliary and target tables either might
be affected by the incoming rows or they might affect the incoming rows
themselves.
A no-op upsert run row in the auxiliary table can be affected by the
microbatch if an incoming row makes the row no longer a no-op (i.e microbatch
delivers an interleaving row that does indeed change history tracked columns).
A tombstone in the auxiliary table can affect an incoming row if it now matches
against an upsert in the microbatch.
A row in the target table can be affected by the microbatch if an incoming
upsert makes the target table's row a no-op upsert, or an incoming
delete/upsert event terminates an existing row in the target table. An active
row (endAt=null) in the target table could become terminated, or an existing
closed row in the target table could become bisected. Conversely, existing rows
in the target table can dictate when an incoming upsert row should be
considered closed from.
We take a practical, conservative approach in selecting the set of rows
that could possibly be affected or affect the microbatch. Per key we retrieve
all existing rows whose startAt comes after the youngest sequence in the
incoming microbatch, as well as the first existing row in both the target/aux
that comes before the youngest sequence.
This is opposed to doing a very complex and expensive join to determine
which rows are definitively affected by/affecting the microbatch. In practice
its not common for events to actually receive very old events out of order, so
pulling in all existing rows that come after the oldest row in the microbatch
will generally be a very small result set.
### Why are the changes needed?
AutoCDC SCD2 core algorithm.
### Does this PR introduce _any_ user-facing change?
No, new feature.
### 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.
Closes #56283 from AnishMahto/SPARK-57152-SCD2-find-affected-rows.
Authored-by: AnishMahto <[email protected]>
Signed-off-by: Jose Torres <[email protected]>
---
.../sql/pipelines/autocdc/Scd2BatchProcessor.scala | 260 +++++++-
.../autocdc/Scd2BatchProcessorSuite.scala | 707 ++++++++++++++++++++-
2 files changed, 943 insertions(+), 24 deletions(-)
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 60fda36fea07..2815420836e0 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
@@ -26,7 +26,7 @@ import org.apache.spark.sql.types.{DataType, StructField,
StructType}
import org.apache.spark.util.ArrayImplicits._
/**
- * Per-microbatch processor for SCD Type 2 AutoCDC flows, complying to the
specified
+ * Per-microbatch processor for SCD Type 2 AutoCDC flows, complying with the
specified
* [[changeArgs]] configuration.
*
* @param changeArgs The CDC flow configuration.
@@ -37,6 +37,18 @@ case class Scd2BatchProcessor(
changeArgs: ChangeArgs,
resolvedSequencingType: DataType) {
+ /**
+ * Backtick-quoted key column names. Use when the name flows through an
expression parser
+ * (e.g., [[F.col]]), which interprets dotted names as struct-field accesses.
+ */
+ private lazy val keysQuoted: Seq[String] = changeArgs.keys.map(_.quoted)
+
+ /**
+ * Raw key column names. Use when the name is matched literally against a
schema field
+ * (e.g., DataFrame `.join(other, usingColumns)`), where backticks are NOT
stripped.
+ */
+ private lazy val keysRaw: Seq[String] = changeArgs.keys.map(_.name)
+
/**
* Reconcile a CDC microbatch into the canonical form the auxiliary- and
target-table merges
* consume.
@@ -50,14 +62,17 @@ case class Scd2BatchProcessor(
* whether across microbatches or within the same microbatch, is the
responsibility of
* downstream reconciliation - not preprocessing.
*
- * Produces a dataframe that retains every input row 1:1 - no rows added,
dropped, reordered,
- * or merged - with the following schema, in column order:
- * 1. The user columns of `microbatchDf` that survive
[[ChangeArgs.columnSelection]], in the
- * order they appeared in the input.
- * 2. `__START_AT` column, populated with the sequence value of the row.
- * 3. `__END_AT` column, populated with the sequence value of the row IFF
it's a delete event,
- * null otherwise.
- * 4. `__spark_autocdc_metadata` column.
+ * @param microbatchDf
+ * the incoming CDC microbatch.
+ * @return
+ * a dataframe that retains every input row 1:1 - no rows added, dropped,
reordered, or
+ * merged - with the following schema, in column order:
+ * 1. The user columns of `microbatchDf` that survive
[[ChangeArgs.columnSelection]], in
+ * the order they appeared in the input.
+ * 2. [[startAtColName]], populated with the sequence value of the row.
+ * 3. [[endAtColName]], populated with the sequence value of the row IFF
it's a delete
+ * event, null otherwise.
+ * 4. [[cdcMetadataColName]], conforming to [[cdcMetadataColSchema]].
*/
private[autocdc] def preprocessMicrobatch(microbatchDf: DataFrame):
DataFrame = {
microbatchDf
@@ -108,7 +123,7 @@ case class Scd2BatchProcessor(
private def extendMicrobatchRowsWithCdcMetadata(microbatchDf: DataFrame):
DataFrame = {
microbatchDf.withColumn(
colName = AutoCdcReservedNames.cdcMetadataColName,
- col = Scd2BatchProcessor.constructTargetCdcMetadataCol(
+ col = Scd2BatchProcessor.constructCdcMetadataCol(
recordStartAt = changeArgs.sequencing,
sequencingType = resolvedSequencingType
)
@@ -159,6 +174,172 @@ case class Scd2BatchProcessor(
microbatch.select(finalColumnsToSelect: _*)
}
+ /**
+ * For each key in the preprocessed microbatch, compute the earliest
[[recordStartAtFieldName]]
+ * across the key's events.
+ *
+ * @param preprocessedBatchDf
+ * a validated and preprocessed microbatch as produced by
[[preprocessMicrobatch]] - in
+ * particular, non-null key columns and a non-null
[[recordStartAtFieldName]] on every row.
+ * @return
+ * a dataframe containing one row per distinct key, with the key columns
+ * ([[ChangeArgs.keys]]) and a [[minSequenceColName]] column carrying the
min
+ * [[recordStartAtFieldName]] across that key's records in the microbatch.
+ */
+ private[autocdc] def computeMinimumSequencePerKey(preprocessedBatchDf:
DataFrame): DataFrame = {
+ val recordStartAt =
+
Scd2BatchProcessor.recordStartAtOf(F.col(AutoCdcReservedNames.cdcMetadataColName))
+ preprocessedBatchDf
+ .groupBy(keysQuoted.map(F.col): _*)
+ .agg(F.min(recordStartAt).alias(Scd2BatchProcessor.minSequenceColName))
+ }
+
+ /**
+ * Find the auxiliary-table rows whose state matters for reconciling the
microbatch.
+ *
+ * @param rawAuxiliaryTableDf
+ * the auxiliary table in its native schema, which is expected to contain
+ * [[deletedByBatchIdColName]] in addition to all of the columns in the
target table.
+ * @param perKeyMinimumSequenceInMicrobatchDf
+ * one row per distinct key as produced by
[[computeMinimumSequencePerKey]], representing
+ * the minimum sequence for that key in the microbatch.
+ * @param batchId
+ * the underlying Spark streaming query's batchId, used to scope aux-row
visibility for
+ * replay-stability across retries of the same microbatch.
+ * @return
+ * a dataframe containing all the affected aux rows, with the aux-only
+ * [[deletedByBatchIdColName]] column dropped so the result is
union-compatible with
+ * preprocessed microbatch rows and target-table rows downstream.
+ */
+ private[autocdc] def findAffectedRowsFromAuxiliaryTable(
+ rawAuxiliaryTableDf: DataFrame,
+ perKeyMinimumSequenceInMicrobatchDf: DataFrame,
+ batchId: Long
+ ): DataFrame = {
+ val auxTableRecordStartAtField = Scd2BatchProcessor.recordStartAtOf(
+ F.col(AutoCdcReservedNames.cdcMetadataColName)
+ )
+ val auxTableDeletedByBatchIdCol =
F.col(Scd2BatchProcessor.deletedByBatchIdColName)
+
+ val reducedAuxiliaryTableDf = rawAuxiliaryTableDf
+ .filter(
+ // [[deletedByBatchIdColName]] carries the batchId whose MERGE
logically deleted the
+ // row, or null on live aux rows. Rows deleted by other batches are
excluded - those
+ // decisions are final. Rows deleted by THIS batch are re-included -
this can only
+ // happen when the current execution is a retry of a partially-failed
prior attempt of
+ // the same microbatch.
+ auxTableDeletedByBatchIdCol.isNull ||
+ auxTableDeletedByBatchIdCol === F.lit(batchId)
+ )
+ // Drop the aux-only idempotency column so the output schema matches
target-table rows
+ // and preprocessed-microbatch rows (which share the same canonical SCD2
row schema).
+ .drop(Scd2BatchProcessor.deletedByBatchIdColName)
+
+ val perKeyMinimumSequenceInMicrobatchCol =
F.col(Scd2BatchProcessor.minSequenceColName)
+
+ // Per key, identify the sequence value associated with the anchor row in
the aux table.
+ //
+ // The anchor row is the aux row with the largest
[[recordStartAtFieldName]] strictly less
+ // than the min sequence in the incoming microbatch for that key. The
reconciler needs this
+ // "left context" in two cases:
+ // (1) Incoming no-op upsert: without the anchor, it would look like a
new run head, when in
+ // reality it's a part of an existing no-op run/head.
+ // (2) Incoming state-changing upsert that bisects two aux no-ops: the
anchor surfaces
+ // the before-half so both halves can be promoted to target. (The
after-half is
+ // picked up by the >= minSeq branch.)
+ //
+ // Because no-op upserts are stored only in the aux table, the anchor
concept only exists when
+ // pulling in rows from the aux table, and is not relevant for the target
table.
+ //
+ // Keys with no aux row strictly before the min sequence have no anchor;
their affected set
+ // reduces to "all aux rows at or after the min sequence."
+ //
+ // The shape of this DataFrame is: [key1, key2, ... keyN, anchorSequence]
+ val perKeyAnchorSequenceDf = reducedAuxiliaryTableDf
+ // The number of rows in [[perKeyMinimumSequenceInMicrobatchDf]] is
bounded by the
+ // number of unique keys in the microbatch, which should typically be
small. The
+ // auxiliary table should generally also be small, containing only no-op
upsert runs
+ // and tombstones per key. Therefore this join should be cheap, and
broadcast joinable.
+ .join(perKeyMinimumSequenceInMicrobatchDf, keysRaw)
+ .filter(auxTableRecordStartAtField <
perKeyMinimumSequenceInMicrobatchCol)
+ .groupBy(keysQuoted.map(F.col): _*)
+ .agg(
+
F.max(auxTableRecordStartAtField).as(Scd2BatchProcessor.anchorSequenceColName)
+ )
+ val anchorSequenceCol = F.col(Scd2BatchProcessor.anchorSequenceColName)
+ val auxRowIsAnchorRow = auxTableRecordStartAtField === anchorSequenceCol
+
+ // Now that we have the minimum sequence in the microbatch and the
sequence of the anchor row,
+ // we have enough information to compute the full set of auxiliary rows
that may affect or
+ // be affected by the microbatch. Membership here is a conservative
superset: every row that
+ // could possibly participate in reconciliation is included, but
downstream reconciliation
+ // determines the actual outcome per row.
+ val auxRowIsAtOrAfterMinSequenceInMicrobatch =
+ auxTableRecordStartAtField >= perKeyMinimumSequenceInMicrobatchCol
+
+ val auxRowAffectsMicrobatch = auxRowIsAtOrAfterMinSequenceInMicrobatch ||
auxRowIsAnchorRow
+
+ val affectedRowsFromAuxiliaryTable = reducedAuxiliaryTableDf
+ // Per row, project the minimum microbatch sequence and anchor sequence
for that row's key
+ // set onto the row, so the affected-row predicate can be evaluated in a
single filter.
+ .join(perKeyMinimumSequenceInMicrobatchDf, keysRaw)
+ .join(
+ perKeyAnchorSequenceDf,
+ keysRaw,
+ joinType = "left"
+ )
+ .filter(auxRowAffectsMicrobatch)
+ .drop(perKeyMinimumSequenceInMicrobatchCol, anchorSequenceCol)
+
+ affectedRowsFromAuxiliaryTable
+ }
+
+ /**
+ * Find the target-table rows whose state matters for reconciling the
microbatch.
+ *
+ * @param targetTableDf
+ * the target table in its native schema.
+ * @param perKeyMinimumSequenceInMicrobatchDf
+ * one row per distinct key as produced by
[[computeMinimumSequencePerKey]], representing
+ * the minimum sequence for that key in the microbatch.
+ * @return
+ * a dataframe containing the affected target rows, with all columns
passed-through.
+ */
+ private[autocdc] def findAffectedRowsFromTargetTable(
+ targetTableDf: DataFrame,
+ perKeyMinimumSequenceInMicrobatchDf: DataFrame
+ ): DataFrame = {
+ val targetEndAtCol = F.col(Scd2BatchProcessor.endAtColName)
+ val perKeyMinimumSequenceInMicrobatchCol =
F.col(Scd2BatchProcessor.minSequenceColName)
+
+ // Per key, identify all the rows in the target table that may be affected
by the
+ // incoming microbatch.
+ //
+ // Unlike the auxiliary table, the target table holds visible rows only:
no hidden open
+ // no-op upsert rows, no tombstones. Visible rows for a given key form a
non-overlapping
+ // interval partition over the sequencing axis, and at most one row has a
null [[endAtColName]]
+ // (the currently active row per key).
+ //
+ // Hence we can simply grab all rows that were active at some point after
the min sequencing
+ // per key, which can be determined entirely by the row's [[endAtColName]].
+ val isCurrentlyActiveRow = targetEndAtCol.isNull
+
+ // `>=` (rather than strict `>`) additionally pulls in the row that closes
exactly at the
+ // smallest incoming sequence: the consecutive left neighbor of that
incoming event. This
+ // provides "left context" for the smallest event, analogous to the anchor
row in
+ // [[findAffectedRowsFromAuxiliaryTable]]. It may need to be demoted from
a target run
+ // boundary to an aux no-op continuation if the incoming event at minSeq
turns out to
+ // extend an earlier run.
+ val rowEndsAfterMinimumSequence = targetEndAtCol >=
perKeyMinimumSequenceInMicrobatchCol
+ val rowMayBeAffected = isCurrentlyActiveRow || rowEndsAfterMinimumSequence
+
+ val affectedRowsFromTargetTable = targetTableDf
+ .join(perKeyMinimumSequenceInMicrobatchDf, keysRaw)
+ .filter(rowMayBeAffected)
+ .drop(perKeyMinimumSequenceInMicrobatchCol)
+
+ affectedRowsFromTargetTable
+ }
}
/**
@@ -204,6 +385,9 @@ case class Scd2BatchProcessor(
* latest row values for that run. The target table in its entirety represents
the SCD2
* representation of the CDC flow's source table.
*
+ * Lifetime invariant: As per the SCD2 contract, the target table should never
have overlapping
+ * rows by [startAt, endAt) intervals.
+ *
* -------------
* Concept: aux table.
*
@@ -248,11 +432,24 @@ case class Scd2BatchProcessor(
*/
object Scd2BatchProcessor {
/**
- * Metadata field that represents the exact time (sequence) of the CDC event
that produced
- * this row. Null only for synthetic decomposition tails.
+ * CDC metadata column field that represents the exact time (sequence) of
the CDC event that
+ * produced this row. Null only for synthetic decomposition tails.
*/
private[autocdc] val recordStartAtFieldName: String = "__RECORD_START_AT"
+ /**
+ * Aux-table only column that holds the microbatch id by which a row was
logically
+ * deleted (null if the row is still live). Future microbatches must treat
any row with a
+ * non-null value here, other than the current batch's id, as deleted, and
may safely
+ * physically reap them since prior microbatches commit before the next one
starts.
+ *
+ * Logically-deleted rows exist as a concept on the auxiliary table to
provide
+ * idempotency, should a microbatch fail between a MERGE executed against
the auxiliary
+ * table and the MERGE executed against the target table.
+ */
+ private[autocdc] val deletedByBatchIdColName: String =
+ s"${AutoCdcReservedNames.prefix}deleted_by_batch_id"
+
/**
* What this column represents depends on which AutoCDC artifact table it is
read from.
*
@@ -284,9 +481,9 @@ object Scd2BatchProcessor {
private[autocdc] val endAtColName: String = "__END_AT"
/**
- * Column names reserved by AutoCDC, that will be projected onto the
microbatch and target
- * tables. If the user's source dataframe contains any of these columns,
SCD2 reconciliation
- * will fail.
+ * Column names reserved by AutoCDC that will be projected onto the
microbatch and
+ * eventually persisted in the target table. If the user's source dataframe
contains any of
+ * these columns, SCD2 reconciliation will fail.
*
* TODO(SPARK-57251): validate at [[AutoCdcMergeFlow]] construction time
that the source
* schema and column selection do not collide with these reserved names,
so we fail fast
@@ -299,9 +496,27 @@ object Scd2BatchProcessor {
)
/**
- * Schema of the CDC metadata struct column for SCD2 target table rows.
+ * Name of temporary column projected onto microbatch to compute the min
sequencing value per
+ * key within the microbatch.
+ */
+ private[autocdc] val minSequenceColName: String =
s"${AutoCdcReservedNames.prefix}min_sequence"
+
+ /**
+ * 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
+ * amongst all rows for a key (or risk undefined behavior), this sequence
value uniquely
+ * identifies an exact row in the aux.
+ */
+ private val anchorSequenceColName: String =
s"${AutoCdcReservedNames.prefix}anchor_sequence"
+
+ /** Project the [[recordStartAtFieldName]] out of an SCD2 CDC metadata
column. */
+ private def recordStartAtOf(cdcMetadataCol: Column): Column =
+ cdcMetadataCol.getField(recordStartAtFieldName)
+
+ /**
+ * Schema of the CDC metadata struct column for SCD2 rows.
*/
- private[pipelines] def targetCdcMetadataColSchema(sequencingType: DataType):
StructType =
+ private[pipelines] def cdcMetadataColSchema(sequencingType: DataType):
StructType =
StructType(
Seq(
// The sequence value of the originating CDC event for this row.
Nullable because
@@ -312,20 +527,19 @@ object Scd2BatchProcessor {
)
/**
- * Construct the CDC metadata struct column for SCD2 target/microbatch rows,
following the
- * exact schema and field ordering defined by [[targetCdcMetadataColSchema]].
+ * Construct the CDC metadata struct column for SCD2 rows, following the
exact schema and
+ * field ordering defined by [[cdcMetadataColSchema]].
*/
- private def constructTargetCdcMetadataCol(
+ private def constructCdcMetadataCol(
recordStartAt: Column,
sequencingType: DataType
): Column = {
- val cdcMetadataFieldsInOrder =
targetCdcMetadataColSchema(sequencingType).fields.map { field =>
+ val cdcMetadataFieldsInOrder =
cdcMetadataColSchema(sequencingType).fields.map { field =>
val value = field.name match {
case `recordStartAtFieldName` => recordStartAt
case other =>
throw SparkException.internalError(
- s"Unable to construct SCD2 target CDC metadata column due to
unknown " +
- s"`${other}` field."
+ s"Unable to construct SCD2 CDC metadata column due to unknown
`${other}` field."
)
}
value.cast(field.dataType).as(field.name)
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 f02986320ab6..c2dee5f9f33a 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
@@ -17,7 +17,7 @@
package org.apache.spark.sql.pipelines.autocdc
-import org.apache.spark.sql.{functions => F, QueryTest, Row}
+import org.apache.spark.sql.{functions => F, Column, QueryTest, Row}
import org.apache.spark.sql.classic.DataFrame
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
@@ -29,6 +29,85 @@ class Scd2BatchProcessorSuite extends QueryTest with
SharedSparkSession {
private def microbatchOf(schema: StructType)(rows: Row*): DataFrame =
spark.createDataFrame(spark.sparkContext.parallelize(rows), schema)
+ /**
+ * Build an mock aux-table [[DataFrame]] from explicit user rows + framework
column values.
+ */
+ private def auxTableOf(
+ userSchema: StructType,
+ sequencingType: DataType = LongType
+ )(rows: Row*): DataFrame = {
+ val schema = userSchema
+ .add(Scd2BatchProcessor.startAtColName, sequencingType, nullable = true)
+ .add(Scd2BatchProcessor.endAtColName, sequencingType, nullable = true)
+ .add(
+ AutoCdcReservedNames.cdcMetadataColName,
+ Scd2BatchProcessor.cdcMetadataColSchema(sequencingType),
+ nullable = false
+ )
+ .add(Scd2BatchProcessor.deletedByBatchIdColName, LongType, nullable =
true)
+ spark.createDataFrame(spark.sparkContext.parallelize(rows), schema)
+ }
+
+ /**
+ * Build a target-table [[DataFrame]] from explicit user rows + framework
column values.
+ *
+ * Each input [[Row]] carries the user columns followed by:
+ * - the row's `__START_AT` value
+ * - the row's `__END_AT` value (null IFF the row is currently active)
+ * - the row's `_cdc_metadata` struct as a [[Row]] (e.g.,
`Row(recordStartAt)`)
+ */
+ private def targetTableOf(
+ userSchema: StructType,
+ sequencingType: DataType = LongType
+ )(rows: Row*): DataFrame = {
+ val schema = userSchema
+ .add(Scd2BatchProcessor.startAtColName, sequencingType, nullable = true)
+ .add(Scd2BatchProcessor.endAtColName, sequencingType, nullable = true)
+ .add(
+ AutoCdcReservedNames.cdcMetadataColName,
+ Scd2BatchProcessor.cdcMetadataColSchema(sequencingType),
+ nullable = false
+ )
+ spark.createDataFrame(spark.sparkContext.parallelize(rows), schema)
+ }
+
+ /**
+ * Build a minimum-sequence-per-key [[DataFrame]] used by the
`findAffected*` functions.
+ *
+ * Each input [[Row]] carries the key columns followed by the per-key
minimum sequence.
+ */
+ private def minSeqOf(
+ keySchema: StructType,
+ sequencingType: DataType = LongType
+ )(rows: Row*): DataFrame = {
+ val schema = keySchema.add(
+ Scd2BatchProcessor.minSequenceColName,
+ sequencingType,
+ nullable = false
+ )
+ spark.createDataFrame(spark.sparkContext.parallelize(rows), schema)
+ }
+
+ /**
+ * Build a [[Scd2BatchProcessor]] suitable for `findAffected*` and
+ * `computeMinimumSequencePerKey` tests. The `sequencing` is fixed to
`F.col("seq")`,
+ * so the input microbatch must include a `seq` column. `deleteCondition` is
optional
+ * and only needed by tests that exercise both event kinds.
+ */
+ private def processorWithKeys(
+ keys: Seq[String],
+ deleteCondition: Option[Column] = None
+ ): Scd2BatchProcessor =
+ Scd2BatchProcessor(
+ changeArgs = ChangeArgs(
+ keys = keys.map(UnqualifiedColumnName(_)),
+ sequencing = F.col("seq"),
+ storedAsScdType = ScdType.Type2,
+ deleteCondition = deleteCondition
+ ),
+ resolvedSequencingType = LongType
+ )
+
// =============== preprocessMicrobatch tests ===============
test("preprocessMicrobatch appends framework columns __START_AT, __END_AT, "
+
@@ -512,4 +591,630 @@ class Scd2BatchProcessorSuite extends QueryTest with
SharedSparkSession {
)
)
}
+
+ // =============== computeMinimumSequencePerKey tests ===============
+
+ test("computeMinimumSequencePerKey returns one row per distinct key and
aggregates across " +
+ "both upsert and delete events") {
+ val schema = new StructType()
+ .add("id", IntegerType)
+ .add("seq", LongType)
+ .add("is_delete", BooleanType)
+
+ val processor = processorWithKeys(
+ keys = Seq("id"),
+ deleteCondition = Some(F.col("is_delete"))
+ )
+
+ // Two keys, each with multiple events including at least one delete and
at least one
+ // out-of-order sequence. Delete events must feed into the per-key min
exactly like
+ // upserts: `preprocessMicrobatch` stamps `__RECORD_START_AT = sequencing`
on every
+ // row regardless of kind, so the min computation cannot legitimately
ignore deletes.
+ // (If it did, the early-delete-bisects-late-upsert reconciliation case
would silently
+ // lose its anchor pull-in via the find* paths.)
+ val raw = microbatchOf(schema)(
+ Row(1, 30L, false), // out-of-order: appears before lower sequences in
the input
+ Row(1, 10L, true), // delete - smallest sequence for key=1
+ Row(1, 20L, false),
+ Row(2, 50L, false),
+ Row(2, 40L, true) // delete - smallest sequence for key=2
+ )
+
+ val preprocessed = processor.preprocessMicrobatch(raw)
+ val result = processor.computeMinimumSequencePerKey(preprocessed)
+
+ assert(result.schema.fieldNames.toSeq == Seq(
+ "id", Scd2BatchProcessor.minSequenceColName
+ ))
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, 10L),
+ Row(2, 40L)
+ )
+ )
+ }
+
+ test("computeMinimumSequencePerKey is compatible with composite keys") {
+ val schema = new StructType()
+ .add("region", StringType)
+ .add("customer_id", IntegerType)
+ .add("seq", LongType)
+
+ val processor = processorWithKeys(keys = Seq("region", "customer_id"))
+
+ // Three composite-key tuples that share their first or second key column.
If the
+ // function mistakenly grouped by `region` alone, (US, 1) and (US, 2)
would collapse
+ // and we'd see only two output rows; if it grouped by `customer_id` alone,
+ // (US, 1) and (EU, 1) would collapse.
+ val raw = microbatchOf(schema)(
+ Row("US", 1, 100L),
+ Row("US", 1, 50L), // smaller sequence for (US, 1)
+ Row("US", 2, 200L),
+ Row("EU", 1, 30L)
+ )
+
+ val preprocessed = processor.preprocessMicrobatch(raw)
+ val result = processor.computeMinimumSequencePerKey(preprocessed)
+
+ assert(result.schema.fieldNames.toSeq == Seq(
+ "region", "customer_id", Scd2BatchProcessor.minSequenceColName
+ ))
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row("US", 1, 50L),
+ Row("US", 2, 200L),
+ Row("EU", 1, 30L)
+ )
+ )
+ }
+
+ test("computeMinimumSequencePerKey returns an empty result for an empty
microbatch") {
+ val schema = new StructType()
+ .add("id", IntegerType)
+ .add("seq", LongType)
+
+ val processor = processorWithKeys(keys = Seq("id"))
+
+ val raw = microbatchOf(schema)()
+ val preprocessed = processor.preprocessMicrobatch(raw)
+ val result = processor.computeMinimumSequencePerKey(preprocessed)
+
+ assert(result.collect().isEmpty)
+ }
+
+ test("computeMinimumSequencePerKey resolves key columns containing a literal
dot") {
+ // Symmetric to the dotted-name test for
findAffectedRowsFromAuxiliaryTable: the
+ // `groupBy(keysQuoted.map(F.col): _*)` site relies on `keysQuoted`
correctly
+ // backtick-quoting "a.b" so that F.col parses it as a literal column name
(rather
+ // than struct-field access). Pins the F.col axis of the keysQuoted vs
keysRaw split.
+ val schema = new StructType()
+ .add("a.b", IntegerType)
+ .add("seq", LongType)
+
+ val processor = processorWithKeys(keys = Seq("`a.b`"))
+
+ val raw = microbatchOf(schema)(
+ Row(1, 30L),
+ Row(1, 10L)
+ )
+ val preprocessed = processor.preprocessMicrobatch(raw)
+ val result = processor.computeMinimumSequencePerKey(preprocessed)
+
+ assert(result.schema.fieldNames.toSeq == Seq(
+ "a.b", Scd2BatchProcessor.minSequenceColName
+ ))
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(Row(1, 10L))
+ )
+ }
+
+ // =============== findAffectedRowsFromAuxiliaryTable tests ===============
+
+ test("findAffectedRowsFromAuxiliaryTable returns the anchor row per key") {
+ val processor = processorWithKeys(Seq("id"))
+ val keySchema = new StructType().add("id", IntegerType)
+ val userSchema = keySchema.add("value", StringType)
+
+ // Two keys to demonstrate per-key anchor isolation.
+ //
+ // Input row shape per `auxTableOf`:
+ // (id, value, __START_AT, __END_AT, Row(recordStartAt),
deletedByBatchId)
+ //
+ // Key 1: aux rows at recordStartAt 3, 5, 10. minSeq = 10.
+ // - 3 -> older than the anchor; dropped.
+ // - 5 -> anchor (max < 10); included.
+ // - 10 -> at minSeq; included via the >= branch (NOT as anchor;
selection is strict <).
+ // Key 2: only one aux row at 7, minSeq = 7.
+ // - 7 -> at minSeq; included via >= branch. No anchor (no rows < 7 for
this key).
+ val aux = auxTableOf(userSchema)(
+ Row(1, "v1.3", 3L, null, Row(3L), null),
+ Row(1, "v1.5", 5L, null, Row(5L), null),
+ Row(1, "v1.10", 10L, null, Row(10L), null),
+ Row(2, "v2.7", 7L, null, Row(7L), null)
+ )
+ val minSeq = minSeqOf(keySchema)(
+ Row(1, 10L),
+ Row(2, 7L)
+ )
+
+ val result = processor.findAffectedRowsFromAuxiliaryTable(
+ rawAuxiliaryTableDf = aux,
+ perKeyMinimumSequenceInMicrobatchDf = minSeq,
+ batchId = 100L
+ )
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "v1.5", 5L, null, Row(5L)), // anchor for key=1
+ Row(1, "v1.10", 10L, null, Row(10L)), // >= minSeq for key=1
+ Row(2, "v2.7", 7L, null, Row(7L)) // >= minSeq for key=2 (no
anchor)
+ )
+ )
+ }
+
+ test("findAffectedRowsFromAuxiliaryTable pulls in both tombstone and no-op
upsert rows") {
+ val processor = processorWithKeys(Seq("id"))
+ val keySchema = new StructType().add("id", IntegerType)
+ val userSchema = keySchema.add("value", StringType)
+
+ // Aux carries a mix of row kinds for one key. The find function does NOT
distinguish
+ // between them - it filters purely on `recordStartAt` - so a tombstone, a
no-op upsert
+ // run head, and a continuation are all eligible anchor candidates and all
eligible for
+ // the >= minSeq inclusion branch.
+ val aux = auxTableOf(userSchema)(
+ // Tombstone at recordStartAt = 3 (deleted at sequence 3): startAt =
endAt = 3.
+ // Older than the anchor; dropped.
+ Row(1, null, 3L, 3L, Row(3L), null),
+ // No-op upsert continuation at recordStartAt = 7: startAt inherits its
run head's
+ // recordStartAt, endAt is null. Anchor for minSeq=10 (max < 10).
+ Row(1, "alice", 5L, null, Row(7L), null),
+ // Tombstone at recordStartAt = 12: at-or-after minSeq, included via >=
branch.
+ Row(1, null, 12L, 12L, Row(12L), null),
+ // No-op upsert continuation at recordStartAt = 15: included via >=
branch.
+ Row(1, "bob", 13L, null, Row(15L), null)
+ )
+ val minSeq = minSeqOf(keySchema)(Row(1, 10L))
+
+ val result = processor.findAffectedRowsFromAuxiliaryTable(
+ rawAuxiliaryTableDf = aux,
+ perKeyMinimumSequenceInMicrobatchDf = minSeq,
+ batchId = 100L
+ )
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", 5L, null, Row(7L)),
+ Row(1, null, 12L, 12L, Row(12L)),
+ Row(1, "bob", 13L, null, Row(15L))
+ )
+ )
+ }
+
+ test("findAffectedRowsFromAuxiliaryTable pulls in both consecutive no-op
upsert events " +
+ "being interleaved by incoming microbatch row") {
+ val processor = processorWithKeys(Seq("id"))
+ val keySchema = new StructType().add("id", IntegerType)
+ val userSchema = keySchema.add("value", StringType)
+
+ val aux = auxTableOf(userSchema)(
+ Row(1, "alice", 2L, null, Row(8L), null),
+ Row(1, "alice", 2L, null, Row(12L), null)
+ )
+ val minSeq = minSeqOf(keySchema)(Row(1, 10L))
+
+ val result = processor.findAffectedRowsFromAuxiliaryTable(
+ rawAuxiliaryTableDf = aux,
+ perKeyMinimumSequenceInMicrobatchDf = minSeq,
+ batchId = 100L
+ )
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ // Row with record start at of 8 gets pulled in as an anchor,
+ Row(1, "alice", 2L, null, Row(8L)),
+ // Row with record start at of 12 gets pulled in as a regular affected
row.
+ Row(1, "alice", 2L, null, Row(12L))
+ )
+ )
+ }
+
+ test("findAffectedRowsFromAuxiliaryTable selects tombstones as anchor if
applicable") {
+ val processor = processorWithKeys(Seq("id"))
+ val keySchema = new StructType().add("id", IntegerType)
+ val userSchema = keySchema.add("value", StringType)
+
+ // Tombstone-as-anchor is incidental: the find function selects the anchor
purely on
+ // `max recordStartAt < minSeq`, so a tombstone qualifies just like any
other row kind.
+ // Downstream reconciliation does not actually rely on the anchor when it
is a
+ // tombstone (a delete already closed the prior run, so any subsequent
incoming event
+ // is necessarily a fresh run head regardless of whether the anchor is
surfaced). We
+ // still pull it in as a harmless side effect of the range filter, and
this behavior is
+ // documented via test.
+ val aux = auxTableOf(userSchema)(
+ Row(1, null, 7L, 7L, Row(7L), null),
+ Row(1, null, 12L, 12L, Row(12L), null)
+ )
+ val minSeq = minSeqOf(keySchema)(Row(1, 10L))
+
+ val result = processor.findAffectedRowsFromAuxiliaryTable(
+ rawAuxiliaryTableDf = aux,
+ perKeyMinimumSequenceInMicrobatchDf = minSeq,
+ batchId = 100L
+ )
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ // Pulled in as anchor.
+ Row(1, null, 7L, 7L, Row(7L)),
+ // Pulled in as regular affected row.
+ Row(1, null, 12L, 12L, Row(12L))
+ )
+ )
+ }
+
+ test("findAffectedRowsFromAuxiliaryTable filters logically-deleted aux
rows") {
+ val processor = processorWithKeys(Seq("id"))
+ val keySchema = new StructType().add("id", IntegerType)
+ val userSchema = keySchema.add("value", StringType)
+
+ val currentBatchId = 100L
+ val differentBatchId = 99L
+
+ // The idempotency filter retains rows deleted by `currentBatchId` (so a
mid-flight
+ // retry sees its own prior writes) and drops rows deleted by any other
batch. This
+ // applies uniformly to both the anchor and non-anchor affected rows.
+ val aux = auxTableOf(userSchema)(
+ // Anchor candidate (recordStartAt < minSeq):
+ Row(1, "anchor", 5L, null, Row(5L), currentBatchId), // deleted by
current -> kept
+ // At-or-after minSeq:
+ Row(1, "live", 10L, null, Row(10L), null), // not deleted -> kept
+ Row(1, "retried", 11L, null, Row(11L), currentBatchId), // deleted by
current -> kept
+ Row(1, "ignored", 12L, null, Row(12L), differentBatchId) // deleted by
another -> dropped
+ )
+ val minSeq = minSeqOf(keySchema)(Row(1, 10L))
+
+ val result = processor.findAffectedRowsFromAuxiliaryTable(
+ rawAuxiliaryTableDf = aux,
+ perKeyMinimumSequenceInMicrobatchDf = minSeq,
+ batchId = currentBatchId
+ )
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "anchor", 5L, null, Row(5L)),
+ Row(1, "live", 10L, null, Row(10L)),
+ Row(1, "retried", 11L, null, Row(11L))
+ )
+ )
+ }
+
+ test("findAffectedRowsFromAuxiliaryTable falls back to the next anchor when
the closest " +
+ "candidate was logically deleted by a different batch") {
+ val processor = processorWithKeys(Seq("id"))
+ val keySchema = new StructType().add("id", IntegerType)
+ val userSchema = keySchema.add("value", StringType)
+
+ val currentBatchId = 100L
+ val differentBatchId = 99L
+
+ // Codifies the step-ordering invariant inside
`findAffectedRowsFromAuxiliaryTable`: the
+ // idempotency filter MUST run before the anchor `max(...)` aggregation.
Here the closest
+ // pre-minSeq candidate (recordStartAt=7) was logically deleted by a
different batch, so
+ // it is filtered out and the anchor falls back to recordStartAt=3. If a
future refactor
+ // were to flip these two steps (e.g. as a "perf optimization"), this test
would catch it
+ // because the natural-anchor row (7) would otherwise be selected and then
dropped, leaving
+ // no anchor at all.
+ val aux = auxTableOf(userSchema)(
+ Row(1, "live3", 3L, null, Row(3L), null),
+ Row(1, "stale7", 7L, null, Row(7L), differentBatchId)
+ )
+ val minSeq = minSeqOf(keySchema)(Row(1, 10L))
+
+ val result = processor.findAffectedRowsFromAuxiliaryTable(
+ rawAuxiliaryTableDf = aux,
+ perKeyMinimumSequenceInMicrobatchDf = minSeq,
+ batchId = currentBatchId
+ )
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(Row(1, "live3", 3L, null, Row(3L)))
+ )
+ }
+
+ test("findAffectedRowsFromAuxiliaryTable drops the aux-only deletedByBatchId
column") {
+ val processor = processorWithKeys(Seq("id"))
+ val keySchema = new StructType().add("id", IntegerType)
+ val userSchema = keySchema.add("value", StringType)
+
+ // Pre-condition: aux carries the canonical SCD2 row schema plus the
top-level
+ // `__spark_autocdc_deleted_by_batch_id` idempotency column. The find
function must
+ // drop that aux-only column so the result is union-compatible with
target-table rows
+ // and preprocessed-microbatch rows downstream, while leaving the
(now-shared)
+ // `_cdc_metadata` struct schema untouched.
+ val aux = auxTableOf(userSchema)(Row(1, "v", 5L, null, Row(5L), null))
+ val minSeq = minSeqOf(keySchema)(Row(1, 10L))
+
+ val result = processor.findAffectedRowsFromAuxiliaryTable(
+ rawAuxiliaryTableDf = aux,
+ perKeyMinimumSequenceInMicrobatchDf = minSeq,
+ batchId = 100L
+ )
+
+
assert(!result.columns.contains(Scd2BatchProcessor.deletedByBatchIdColName))
+ val cdcMetadataField =
result.schema(AutoCdcReservedNames.cdcMetadataColName)
+ assert(cdcMetadataField.dataType ==
Scd2BatchProcessor.cdcMetadataColSchema(LongType))
+ }
+
+ test("findAffectedRowsFromAuxiliaryTable resolves key columns containing a
literal dot") {
+ val processor = processorWithKeys(Seq("`a.b`"))
+ val keySchema = new StructType().add("a.b", IntegerType)
+ val userSchema = keySchema.add("value", StringType)
+
+ val aux = auxTableOf(userSchema)(Row(1, "v", 5L, null, Row(5L), null))
+ val minSeq = minSeqOf(keySchema)(Row(1, 10L))
+
+ val result = processor.findAffectedRowsFromAuxiliaryTable(
+ rawAuxiliaryTableDf = aux,
+ perKeyMinimumSequenceInMicrobatchDf = minSeq,
+ batchId = 100L
+ )
+
+ // The lone aux row is the anchor (recordStartAt=5 < minSeq=10, no other
candidates).
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(Row(1, "v", 5L, null, Row(5L)))
+ )
+ }
+
+ test("findAffectedRowsFromAuxiliaryTable respects composite keys") {
+ val keySchema = new StructType()
+ .add("region", StringType)
+ .add("customer_id", IntegerType)
+ val userSchema = keySchema.add("name", StringType)
+
+ val processor = processorWithKeys(Seq("region", "customer_id"))
+
+ // Three composite keys: (US, 1), (EU, 1), (US, 2). Each is independent.
+ // (US, 1): anchor at 3; row at 10 included via >=.
+ // (EU, 1): anchor at 4; no rows at or after 12 -> only the anchor.
+ // (US, 2): no aux rows -> contributes nothing.
+ val aux = auxTableOf(userSchema)(
+ Row("US", 1, "us1.3", 3L, null, Row(3L), null),
+ Row("US", 1, "us1.10", 10L, null, Row(10L), null),
+ Row("EU", 1, "eu1.4", 4L, null, Row(4L), null)
+ )
+ val minSeq = minSeqOf(keySchema)(
+ Row("US", 1, 10L),
+ Row("EU", 1, 12L),
+ Row("US", 2, 100L)
+ )
+
+ val result = processor.findAffectedRowsFromAuxiliaryTable(
+ rawAuxiliaryTableDf = aux,
+ perKeyMinimumSequenceInMicrobatchDf = minSeq,
+ batchId = 100L
+ )
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row("US", 1, "us1.3", 3L, null, Row(3L)),
+ Row("US", 1, "us1.10", 10L, null, Row(10L)),
+ Row("EU", 1, "eu1.4", 4L, null, Row(4L))
+ )
+ )
+ }
+
+ test("findAffectedRowsFromAuxiliaryTable returns an empty result when the
aux table is empty") {
+ val processor = processorWithKeys(Seq("id"))
+ val keySchema = new StructType().add("id", IntegerType)
+ val userSchema = keySchema.add("value", StringType)
+
+ val aux = auxTableOf(userSchema)()
+ val minSeq = minSeqOf(keySchema)(Row(1, 10L))
+
+ val result = processor.findAffectedRowsFromAuxiliaryTable(
+ rawAuxiliaryTableDf = aux,
+ perKeyMinimumSequenceInMicrobatchDf = minSeq,
+ batchId = 100L
+ )
+
+ assert(result.collect().isEmpty)
+ }
+
+ test("findAffectedRowsFromAuxiliaryTable returns no rows for a microbatch
key that has " +
+ "no rows in the aux table") {
+ val processor = processorWithKeys(Seq("id"))
+ val keySchema = new StructType().add("id", IntegerType)
+ val userSchema = keySchema.add("value", StringType)
+
+ // Aux only has rows for key=1. Microbatch only sees key=2.
+ val aux = auxTableOf(userSchema)(Row(1, "v", 5L, null, Row(5L), null))
+ val minSeq = minSeqOf(keySchema)(Row(2, 10L))
+
+ val result = processor.findAffectedRowsFromAuxiliaryTable(
+ rawAuxiliaryTableDf = aux,
+ perKeyMinimumSequenceInMicrobatchDf = minSeq,
+ batchId = 100L
+ )
+
+ assert(result.collect().isEmpty)
+ }
+
+ test("findAffectedRowsFromAuxiliaryTable excludes aux rows for keys not in
the microbatch") {
+ val processor = processorWithKeys(Seq("id"))
+ val keySchema = new StructType().add("id", IntegerType)
+ val userSchema = keySchema.add("value", StringType)
+
+ // Aux has rows for keys 1 and 2. Microbatch only mentions key=1, so
key=2's aux rows
+ // must be dropped (the inner join with minSeq strips them).
+ val aux = auxTableOf(userSchema)(
+ Row(1, "v1", 5L, null, Row(5L), null),
+ Row(2, "v2", 7L, null, Row(7L), null)
+ )
+ val minSeq = minSeqOf(keySchema)(Row(1, 10L))
+
+ val result = processor.findAffectedRowsFromAuxiliaryTable(
+ rawAuxiliaryTableDf = aux,
+ perKeyMinimumSequenceInMicrobatchDf = minSeq,
+ batchId = 100L
+ )
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(Row(1, "v1", 5L, null, Row(5L)))
+ )
+ }
+
+ // =============== findAffectedRowsFromTargetTable tests ===============
+
+ test("findAffectedRowsFromTargetTable includes both closed and active
affected rows") {
+ val processor = processorWithKeys(Seq("id"))
+ val keySchema = new StructType().add("id", IntegerType)
+ val userSchema = keySchema.add("value", StringType)
+
+ // Single key with four target rows:
+ // - row closed at endAt=5 -> < minSeq=10 -> excluded
+ // - row closed at endAt=10 -> = minSeq=10 -> included (>=)
+ // - row closed at endAt=15 -> > minSeq=10 -> included
+ // - row active (endAt=null) -> always included
+ val target = targetTableOf(userSchema)(
+ Row(1, "old", 1L, 5L, Row(1L)),
+ Row(1, "edge", 5L, 10L, Row(5L)),
+ Row(1, "recent", 10L, 15L, Row(10L)),
+ Row(1, "active", 15L, null, Row(15L))
+ )
+ val minSeq = minSeqOf(keySchema)(Row(1, 10L))
+
+ val result = processor.findAffectedRowsFromTargetTable(
+ targetTableDf = target,
+ perKeyMinimumSequenceInMicrobatchDf = minSeq
+ )
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "edge", 5L, 10L, Row(5L)),
+ Row(1, "recent", 10L, 15L, Row(10L)),
+ Row(1, "active", 15L, null, Row(15L))
+ )
+ )
+ }
+
+ test("findAffectedRowsFromTargetTable computes inclusion independently per
key") {
+ val processor = processorWithKeys(Seq("id"))
+ val keySchema = new StructType().add("id", IntegerType)
+ val userSchema = keySchema.add("value", StringType)
+
+ // Two keys with overlapping endAt ranges but different per-key minSeqs.
Each key is
+ // reconciled independently against its own minSeq.
+ val target = targetTableOf(userSchema)(
+ // Key 1: minSeq=10. "active" (null) and "recent" (15) are at/after 10.
+ Row(1, "k1.old", 1L, 5L, Row(1L)),
+ Row(1, "k1.recent", 5L, 15L, Row(5L)),
+ Row(1, "k1.active", 15L, null, Row(15L)),
+ // Key 2: minSeq=20. Only "active" (null) is at/after 20.
+ Row(2, "k2.old", 1L, 10L, Row(1L)),
+ Row(2, "k2.recent", 10L, 18L, Row(10L)),
+ Row(2, "k2.active", 18L, null, Row(18L))
+ )
+ val minSeq = minSeqOf(keySchema)(
+ Row(1, 10L),
+ Row(2, 20L)
+ )
+
+ val result = processor.findAffectedRowsFromTargetTable(
+ targetTableDf = target,
+ perKeyMinimumSequenceInMicrobatchDf = minSeq
+ )
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "k1.recent", 5L, 15L, Row(5L)),
+ Row(1, "k1.active", 15L, null, Row(15L)),
+ Row(2, "k2.active", 18L, null, Row(18L))
+ )
+ )
+ }
+
+ test("findAffectedRowsFromTargetTable respects composite keys") {
+ val keySchema = new StructType()
+ .add("region", StringType)
+ .add("customer_id", IntegerType)
+ val userSchema = keySchema.add("name", StringType)
+
+ val processor = processorWithKeys(Seq("region", "customer_id"))
+
+ // (US, 1) and (EU, 1) are distinct composite keys. (US, 1)'s active row
is included
+ // for minSeq=10; (EU, 1)'s active row is included for minSeq=12; (EU,
1)'s old closed
+ // row at endAt=5 is excluded (5 < 12). (US, 2) has no target rows.
+ val target = targetTableOf(userSchema)(
+ Row("US", 1, "us1", 1L, null, Row(1L)),
+ Row("EU", 1, "eu1.old", 1L, 5L, Row(1L)),
+ Row("EU", 1, "eu1", 5L, null, Row(5L))
+ )
+ val minSeq = minSeqOf(keySchema)(
+ Row("US", 1, 10L),
+ Row("EU", 1, 12L),
+ Row("US", 2, 100L)
+ )
+
+ val result = processor.findAffectedRowsFromTargetTable(
+ targetTableDf = target,
+ perKeyMinimumSequenceInMicrobatchDf = minSeq
+ )
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row("US", 1, "us1", 1L, null, Row(1L)),
+ Row("EU", 1, "eu1", 5L, null, Row(5L))
+ )
+ )
+ }
+
+ test("findAffectedRowsFromTargetTable returns an empty result when the
target table is empty") {
+ val processor = processorWithKeys(Seq("id"))
+ val keySchema = new StructType().add("id", IntegerType)
+ val userSchema = keySchema.add("value", StringType)
+
+ val target = targetTableOf(userSchema)()
+ val minSeq = minSeqOf(keySchema)(Row(1, 10L))
+
+ val result = processor.findAffectedRowsFromTargetTable(
+ targetTableDf = target,
+ perKeyMinimumSequenceInMicrobatchDf = minSeq
+ )
+
+ assert(result.collect().isEmpty)
+ }
+
+ test("findAffectedRowsFromTargetTable returns no rows for a microbatch key
that has " +
+ "no rows in the target table") {
+ val processor = processorWithKeys(Seq("id"))
+ val keySchema = new StructType().add("id", IntegerType)
+ val userSchema = keySchema.add("value", StringType)
+
+ // Target only has rows for key=1. Microbatch only sees key=2.
+ val target = targetTableOf(userSchema)(Row(1, "v", 1L, null, Row(1L)))
+ val minSeq = minSeqOf(keySchema)(Row(2, 10L))
+
+ val result = processor.findAffectedRowsFromTargetTable(
+ targetTableDf = target,
+ perKeyMinimumSequenceInMicrobatchDf = minSeq
+ )
+
+ assert(result.collect().isEmpty)
+ }
+
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
