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 4933473944cd [SPARK-57222][SDP] Implement SCD2 Batch Processor;
Decompose affected rows
4933473944cd is described below
commit 4933473944cdf4e2543e80e7346bffb584a26f9f
Author: AnishMahto <[email protected]>
AuthorDate: Mon Jul 6 09:45:42 2026 -0700
[SPARK-57222][SDP] Implement SCD2 Batch Processor; Decompose affected 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.
**Decompose affected rows:**
Given the set of affected rows in the current microbatch execution -
incoming rows in the microbatch, affected rows from aux table, affected rows
from target table - the first step in microbatch reconciliation is decomposing
closed historical rows that are being bisected by the microbatch.
A closed historical row is a row in the target table that has a non-null
start-at and end-at. It's possible an incoming upsert/delete in the microbatch
lands with a sequence in between an existing closed row's start/end at (i.e is
a late-arriving event), bisecting it.
Decomposing a closed row means exactly this - bisecting the closed interval
into a left and right end point, called the decomposed head and tail of the
original closed row respectively. The head represents some past upsert event,
the tail represents some past delete or upsert-overtake event.
Once a closed row is decomposed into its end points, it can either coalesce
with other endpoints/events from the full set of affected rows to form a new
historical row, or it can be demoted back to the aux table as a tombstone or
no-op upsert.
**Assert well formed rows post-decomposition**
Given the decomposed affected rows, assert each classifies into one of the
known row-types; tombstone, synthetic decomposition tail, or upsert
representing row (which can be further classified as a valid open or closed
upsert).
**Drop redundant rows post-decomposition:**
Decomposition transforms a single row into two synthetic children rows
(decomposition head and tail), and its possible the resulting decomposition
tail is made logically redundant by the incoming microbatch.
It's also possible that there are duplicate events by key+sequencing,
either both introduced in the same microbatch or across the incoming and a past
microbatch. Post-decomposition is a good time to reconcile these duplicates,
because by this point we have the maximal possible set of rows that will be
merged back into the target/aux tables.
We must drop redundant rows prior to reconciling the new start/end-ats for
this decomposed set of microbatch-affected rows, to:
1. Prevent zero width rows (start at == end at) from ever appearing in the
target table.
2. Ensure reconciliation of a row's start/end at is fully derived from at
most one other row in the decomposed set.
### Why are the changes needed?
Needed for core SCD2 reconciliation logic.
### Does this PR introduce _any_ user-facing change?
No, SCD2 is unreleased.
### How was this patch tested?
Unit tests added to `Scd2BatchProcessorSuite`.
### Was this patch authored or co-authored using generative AI tooling?
Co-authored with Claude Opus 4.7
Closes #56311 from AnishMahto/SPARK-57222-SCD2-decompose-affected-rows.
Authored-by: AnishMahto <[email protected]>
Signed-off-by: Jose Torres <[email protected]>
---
.../sql/pipelines/autocdc/Scd2BatchProcessor.scala | 420 ++++++++++++-
.../autocdc/Scd2BatchProcessorSuite.scala | 693 +++++++++++++++++++++
2 files changed, 1108 insertions(+), 5 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 2815420836e0..b0c511958ab7 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,9 +20,11 @@ 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.expressions.{CreateMap, If, Literal,
RaiseError}
import org.apache.spark.sql.catalyst.util.QuotingUtils
-import org.apache.spark.sql.classic.DataFrame
-import org.apache.spark.sql.types.{DataType, StructField, StructType}
+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.util.ArrayImplicits._
/**
@@ -49,6 +51,60 @@ case class Scd2BatchProcessor(
*/
private lazy val keysRaw: Seq[String] = changeArgs.keys.map(_.name)
+ /**
+ * WindowSpec that sorts CDC event rows in ascending order per key, by event
origination
+ * sequence time (i.e, record start at).
+ */
+ private[autocdc] val orderChronologicallyPerKeyWindow: WindowSpec = {
+ val recordStartAtCol = Scd2BatchProcessor.recordStartAtOf(
+ F.col(AutoCdcReservedNames.cdcMetadataColName)
+ )
+ val startAtCol = F.col(Scd2BatchProcessor.startAtColName)
+ val endAtCol = F.col(Scd2BatchProcessor.endAtColName)
+ val row = Scd2IntervalColumns(recordStartAtCol, startAtCol, endAtCol)
+
+ // Order by effective recordStartAt. The decomposition-tail fallback -
tails carry a null
+ // recordStartAt and order by their endAt instead - is encapsulated in
+ // Scd2IntervalColumns.effectiveRecordStartAt, so no tail special-casing
is needed here.
+ val effectiveRecordStartAt = row.effectiveRecordStartAt.asc
+
+ val orderDecompositionTailsFirst =
RowClassifier.isDecompositionTail(row).desc
+
+ val orderUpsertRepresentingRowsFirst =
RowClassifier.isUpsertRepresentingRow(row).desc
+
+ Window
+ .partitionBy(keysQuoted.map(F.col): _*)
+ .orderBy(
+ // Primary sort key: the source-CDC sequence time. Users are required
to guarantee
+ // events emitted by their source have a unique sequence per key;
violating this is
+ // publicly documented as undefined behavior.
+ //
+ // Decomposition tails are synthetic rows created during
reconciliation; they have a
+ // null recordStartAt and use their endAt as their _effective_
recordStartAt. Under
+ // a compliant source they are the only rows that may legitimately tie
with another
+ // row on effective recordStartAt: a tail inherits its endAt from its
parent closed
+ // row, and that endAt is by construction the recordStartAt of the
event that
+ // originally closed the run, so the tail ties with that closing event
whenever
+ // both appear in the same partition.
+ effectiveRecordStartAt,
+ // Among rows tied on effective recordStartAt, decomposition tails
sort first so a
+ // tail can detect its own redundancy via LEAD(1): if the next row is
a non-tail at
+ // the same instant, the synthetic close the tail encodes is already
represented by
+ // that event and the tail is dropped downstream.
+ //
+ // Any tiebreaking beyond this rule only meaningfully fires when the
user's source
+ // has emitted two or more events at the same sequence, violating the
uniqueness
+ // contract above. Behavior in that case is publicly undefined and the
remaining
+ // tiebreaker clauses exist as a best-effort to keep retries and
replays deterministic.
+ orderDecompositionTailsFirst,
+ // Upsert-representing rows sort before tombstones because rows detect
if they are being
+ // bisected by LEAD(1). This allows upserts to match against
same-sequence deletes, an
+ // arbitrary but deterministic convention. When this happens, the
delete event will survive
+ // and persist as a tombstone in the auxiliary table.
+ orderUpsertRepresentingRowsFirst
+ )
+ }
+
/**
* Reconcile a CDC microbatch into the canonical form the auxiliary- and
target-table merges
* consume.
@@ -340,6 +396,249 @@ case class Scd2BatchProcessor(
affectedRowsFromTargetTable
}
+
+ /**
+ * For every closed non-tombstone row in the input dataframe whose immediate
window-order
+ * successor (in the same per-key partition, per
[[orderChronologicallyPerKeyWindow]])
+ * has `recordStartAt` strictly less than its `endAt` (in other words the
row is being
+ * bisected by its neighbor), replace that row by a "head" + "tail" pair:
+ * - head: copies the parent row exactly, except [[endAtColName]] is set
to null.
+ * - tail: copies the parent row exactly, except [[startAtColName]] is set
to null and
+ * [[recordStartAtFieldName]] inside [[cdcMetadataColName]] is set to
null. All user
+ * data columns are inherited from the parent as-is.
+ *
+ * Decomposition tails are uniquely identified by [[recordStartAtFieldName]]
= null and
+ * are temporary: downstream reconciliation drops them when a coincident
non-tail row
+ * already represents the same closure, or promotes them to tombstones in
the aux table
+ * otherwise.
+ *
+ * All other input rows pass through unchanged ("no-op decompose"). This
includes:
+ * 1. Open rows ([[endAtColName]] = null): incoming upserts, no-op
continuations, etc.
+ * 2. Tombstones ([[startAtColName]] = [[endAtColName]]): protected from
decomposition
+ * even though they qualify as "closed" in the broader sense, because
their interval
+ * is degenerate.
+ * 3. Closed non-tombstone rows whose successor's
[[recordStartAtFieldName]] is `>=`
+ * this row's [[endAtColName]]: the closing event already coincides
with or follows
+ * the run boundary, so there is nothing to bisect.
+ *
+ * Bisection detection is implemented via `LEAD(1)` over
[[orderChronologicallyPerKeyWindow]]:
+ * because the window orders by effective recordStartAt ascending, examining
only the
+ * immediate successor is sufficient to decide whether any other row in the
partition has
+ * a recordStartAt within `[recordStartAt, endAt)`.
+ *
+ * Decomposing closed rows that are being bisected gives us that chance to
form new
+ * closed intervals using the incoming microbatch events, later in
reconciliation.
+ *
+ * @param rowsToDecomposePerKey
+ * a dataframe conforming to the canonical SCD2 row schema
+ * `[user_cols..., [[startAtColName]], [[endAtColName]],
[[cdcMetadataColName]]]`, where
+ * [[cdcMetadataColName]] conforms to [[cdcMetadataColSchema]].
Decomposition tails
+ * (rows with [[recordStartAtFieldName]] = null) MUST NOT be present on
input - they are
+ * produced exclusively by this function.
+ * @return
+ * a dataframe with the same schema as the input. Every closed
non-tombstone row that
+ * was bisected has been replaced by its head + tail pair; every other row
is carried
+ * through as-is. Each output row can be classified as one of:
{decomposition head,
+ * decomposition tail, instantaneous delete, open upsert,
closed-and-unbisected row}. It's
+ * possible that some of the returned decomposition tails are logically
redundant, as
+ * deletion markers that are immediately overtaken by a succeeding row.
+ */
+ private[autocdc] def decomposeOutOfOrderRows(rowsToDecomposePerKey:
DataFrame): DataFrame = {
+ val recordStartAtField =
+
Scd2BatchProcessor.recordStartAtOf(F.col(AutoCdcReservedNames.cdcMetadataColName))
+ val startAtCol = F.col(Scd2BatchProcessor.startAtColName)
+ val endAtCol = F.col(Scd2BatchProcessor.endAtColName)
+ val nextRecordStartAt = F.col(Scd2BatchProcessor.nextRecordStartAtColName)
+
+ // Track the next (in sorted order) row's recordStartAt in a temporary
column.
+ val rowsToDecomposeWithWindowCols = rowsToDecomposePerKey.withColumn(
+ Scd2BatchProcessor.nextRecordStartAtColName,
+ F.lead(recordStartAtField, 1).over(orderChronologicallyPerKeyWindow)
+ )
+
+ val isClosedUpsertRow = RowClassifier.isClosedUpsert(
+ Scd2IntervalColumns(recordStartAtField, startAtCol, endAtCol)
+ )
+ val nextRowBisectsCurrentRow =
+ nextRecordStartAt.isNotNull && nextRecordStartAt < endAtCol
+ val rowShouldDecompose = isClosedUpsertRow && nextRowBisectsCurrentRow
+
+ val originalCols: Seq[String] = rowsToDecomposePerKey.columns.toSeq
+ val originalSchema: StructType = rowsToDecomposePerKey.schema
+ def withOriginalSchemaPreserved(fieldName: String, expr: Column): Column =
{
+ val f = originalSchema(fieldName)
+ expr.cast(f.dataType).as(f.name, f.metadata)
+ }
+
+ // Constructs the head of a row post-decomposition.
+ def constructDecomposedRowHead: Column = {
+ val fields = originalCols.map {
+ case colName if colName == Scd2BatchProcessor.endAtColName =>
+ // End-at is opened (set to null); every other column is inherited
as-is from the
+ // original parent row.
+ withOriginalSchemaPreserved(colName, F.lit(null))
+ case colName =>
+ withOriginalSchemaPreserved(colName, F.col(colName))
+ }
+ F.struct(fields: _*)
+ }
+
+ // Constructs the tail of a row post-decomposition.
+ def constructDecomposedRowTail: Column = {
+ val fields = originalCols.map {
+ case colName if colName == Scd2BatchProcessor.startAtColName =>
+ // Start-at is opened (set to null), every other column is inherited
as-is from the
+ // original parent row.
+ withOriginalSchemaPreserved(colName, F.lit(null))
+ case colName if colName == AutoCdcReservedNames.cdcMetadataColName =>
+ withOriginalSchemaPreserved(
+ colName,
+ Scd2BatchProcessor.constructCdcMetadataCol(
+ recordStartAt = F.lit(null).cast(resolvedSequencingType),
+ sequencingType = resolvedSequencingType
+ )
+ )
+ case colName =>
+ withOriginalSchemaPreserved(colName, F.col(colName))
+ }
+ F.struct(fields: _*)
+ }
+
+ // No-op decomposition carries over the row exactly as-is.
+ def constructNoopDecomposedRow: Column = {
+ val fields = originalCols.map(colName =>
+ withOriginalSchemaPreserved(colName, F.col(colName))
+ )
+ F.struct(fields: _*)
+ }
+
+ // If a row is bisected by its window-order successor, decompose it into a
head + tail
+ // pair. Otherwise pass through as a single-element array so the explode
below is uniform.
+ val perRowDecompositionResults = F
+ .when(
+ rowShouldDecompose,
+ F.array(constructDecomposedRowHead, constructDecomposedRowTail)
+ )
+ .otherwise(F.array(constructNoopDecomposedRow))
+
+ rowsToDecomposeWithWindowCols
+ // The output schema matches the input schema exactly; no extra columns
are projected.
+ .withColumn(
+ Scd2BatchProcessor.decompositionExplodedColName,
+ F.explode(perRowDecompositionResults)
+ )
+ .select(F.col(s"${Scd2BatchProcessor.decompositionExplodedColName}.*"))
+ }
+
+ /**
+ * Asserts that every row in `decomposedRowsPerKey` conforms to one of the
four canonical
+ * post-decomposition shapes - instantaneous delete, open upsert, closed
upsert, or
+ * decomposition tail - and is otherwise a structural identity transform.
+ *
+ * @param decomposedRowsPerKey
+ * the output of [[decomposeOutOfOrderRows]]: a dataframe conforming to
the canonical
+ * SCD2 row schema `[user_cols..., [[startAtColName]], [[endAtColName]],
+ * [[cdcMetadataColName]]]`.
+ * @return
+ * a dataframe with the exact same schema and rows as the input. Failing
the
+ * well-formedness check is treated as an internal-invariant violation: at
execution
+ * time, the first ill-formed row encountered aborts the query with a
SparkRuntimeException
+ */
+ private[autocdc] def assertWellFormedRowsPostDecomposition(
+ decomposedRowsPerKey: DataFrame,
+ batchId: Long
+ ): DataFrame = {
+ val recordStartAtField =
+
Scd2BatchProcessor.recordStartAtOf(F.col(AutoCdcReservedNames.cdcMetadataColName))
+ val startAtCol = F.col(Scd2BatchProcessor.startAtColName)
+ val endAtCol = F.col(Scd2BatchProcessor.endAtColName)
+
+ val row = Scd2IntervalColumns(recordStartAtField, startAtCol, endAtCol)
+ val isWellFormedRow =
+ RowClassifier.isDecompositionTail(row) ||
+ RowClassifier.isDeleteRepresentingRow(row) ||
+ RowClassifier.isUpsertRepresentingRow(row)
+
+ def stringOrNullLit(c: Column): Column = F.coalesce(c.cast(StringType),
F.lit("null"))
+ val malformedRowDiagnostic = F.concat(
+ F.lit(
+ s"During SCD2 reconciliation of microbatch [id=${batchId}],
encountered a " +
+ "post-decomposition row of unexpected shape:"
+ ),
+ F.lit(s" ${Scd2BatchProcessor.recordStartAtFieldName}="),
+ stringOrNullLit(recordStartAtField),
+ F.lit(s", ${Scd2BatchProcessor.startAtColName}="),
+ stringOrNullLit(startAtCol),
+ F.lit(s", ${Scd2BatchProcessor.endAtColName}="),
+ stringOrNullLit(endAtCol),
+ F.lit(".")
+ )
+
+ val internalErrorOnMalformed = ExpressionUtils.column(
+ If(
+ predicate = ExpressionUtils.expression(isWellFormedRow),
+ trueValue = Literal(null, BooleanType),
+ falseValue = RaiseError(
+ Literal("INTERNAL_ERROR"),
+ CreateMap(Seq(
+ Literal("message"),
+ ExpressionUtils.expression(malformedRowDiagnostic)
+ )),
+ BooleanType
+ )
+ )
+ )
+ decomposedRowsPerKey.filter(internalErrorOnMalformed.isNull)
+ }
+
+ /**
+ * Drops rows that are redundant within the per-key chronological window
output by
+ * [[decomposeOutOfOrderRows]]. The redundancy criterion is a single rule:
+ *
+ * A row is redundant whenever it sorts before some other row with the
same effective
+ * recordStartAt under [[orderChronologicallyPerKeyWindow]]; equivalently,
only the
+ * trailing row in any such tie survives.
+ *
+ * The window's tiebreakers (see [[orderChronologicallyPerKeyWindow]]) put
the more
+ * informative row last in every meaningful collision: a decomposition tail
drops in
+ * favor of the coincident real event that already encodes the same closure,
and an
+ * upsert drops in favor of a same-sequence tombstone (delete wins over
upsert at the
+ * same instant).
+ *
+ * @param decomposedRowsPerKey
+ * the output of [[decomposeOutOfOrderRows]]: a dataframe conforming to
the canonical
+ * SCD2 row schema `[user_cols..., [[startAtColName]], [[endAtColName]],
+ * [[cdcMetadataColName]]]`.
+ * @return
+ * a dataframe conforming to the same schema as the input. Per key, no two
rows in the
+ * returned dataframe will share the same effective record-start-at.
+ */
+ private[autocdc] def dropRedundantRowsPostDecomposition(
+ decomposedRowsPerKey: DataFrame): DataFrame = {
+ val recordStartAtField =
+
Scd2BatchProcessor.recordStartAtOf(F.col(AutoCdcReservedNames.cdcMetadataColName))
+ val startAtCol = F.col(Scd2BatchProcessor.startAtColName)
+ val endAtCol = F.col(Scd2BatchProcessor.endAtColName)
+ val effectiveRecordStartAt =
+ Scd2IntervalColumns(recordStartAtField, startAtCol,
endAtCol).effectiveRecordStartAt
+
+ val withNextEffectiveRecordStartAt = decomposedRowsPerKey.withColumn(
+ Scd2BatchProcessor.nextEffectiveRecordStartAtColName,
+ F.lead(effectiveRecordStartAt, 1).over(orderChronologicallyPerKeyWindow)
+ )
+ val nextEffectiveRecordStartAt =
+
withNextEffectiveRecordStartAt.col(Scd2BatchProcessor.nextEffectiveRecordStartAtColName)
+
+ // A row is redundant whenever it ties with its immediate window-order
successor on
+ // effective recordStartAt. Window tiebreakers ensure the *leading* row in
any such tie
+ // is the one to drop.
+ val isRedundantAtSameEffectiveSequence =
+ effectiveRecordStartAt <=> nextEffectiveRecordStartAt
+
+ withNextEffectiveRecordStartAt
+ .filter(!isRedundantAtSameEffectiveSequence)
+ .drop(Scd2BatchProcessor.nextEffectiveRecordStartAtColName)
+ }
}
/**
@@ -386,7 +685,9 @@ case class Scd2BatchProcessor(
* 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.
+ * rows by [startAt, endAt) intervals. The target table also never persists a
zero-width
+ * upsert-representing row (a non-tombstone with `startAt == endAt`); such
rows are dropped
+ * during reconciliation.
*
* -------------
* Concept: aux table.
@@ -404,7 +705,7 @@ case class Scd2BatchProcessor(
* -------------
* Concept: decomposition tail.
*
- * A transient and synthetic row produced by the batch processor during
reconciliation (not
+ * A temporary and synthetic row produced by the batch processor during
reconciliation (not
* from the CDC source) when a previously-closed historical row [START_AT=X,
END_AT=Y] is
* bisected by a late-arriving event. The bisected row is split into a head
* [START_AT=X, END_AT=null] - inheriting the original row's data and
`__RECORD_START_AT` -
@@ -498,9 +799,40 @@ object Scd2BatchProcessor {
/**
* Name of temporary column projected onto microbatch to compute the min
sequencing value per
* key within the microbatch.
+ *
+ * Temporary in that the column has no observable side effect or persistence
across microbatches.
+ * Package level visiblity for unit-testing.
*/
private[autocdc] val minSequenceColName: String =
s"${AutoCdcReservedNames.prefix}min_sequence"
+ /**
+ * Name of temporary column projected onto intermediary dataframe during
decomposition to track
+ * the next row's record start at when in sorted in chronological order as
per
+ * [[orderChronologicallyPerKeyWindow]].
+ *
+ * Temporary in that the column has no observable side effect or persistence
across microbatches.
+ */
+ private val nextRecordStartAtColName: String =
+ s"${AutoCdcReservedNames.prefix}next_record_start_at"
+
+ /**
+ * Name of temporary column projected onto intermediary dataframe during
decomposition that
+ * stores the child rows that result from decomposing a parent row.
+ *
+ * Temporary in that the column has no observable side effect or persistence
across microbatches.
+ */
+ private val decompositionExplodedColName: String =
+ s"${AutoCdcReservedNames.prefix}decompose_output"
+
+ /**
+ * Name of temporary column used by [[dropRedundantRowsPostDecomposition]]
to reference the
+ * next row's effective recordStartAt in chronologically sorted order.
+ *
+ * Temporary in that the column has no observable side effect or persistence
across microbatches.
+ */
+ private[autocdc] val nextEffectiveRecordStartAtColName: String =
+ s"${AutoCdcReservedNames.prefix}next_effective_record_start_at"
+
/**
* Name of the temporary column used to identify the sequence associated
with the anchor
* row found in the auxiliary table for the incoming microbatch. Since
sequences must be unique
@@ -520,7 +852,7 @@ object Scd2BatchProcessor {
StructType(
Seq(
// The sequence value of the originating CDC event for this row.
Nullable because
- // decomposition tails, which are transient and synthetically
constructed during
+ // decomposition tails, which are temporarily and synthetically
constructed during
// reconciliation, have a null record start at.
StructField(recordStartAtFieldName, sequencingType, nullable = true)
)
@@ -547,3 +879,81 @@ object Scd2BatchProcessor {
F.struct(cdcMetadataFieldsInOrder.toImmutableArraySeq: _*)
}
}
+
+/**
+ * The three columns that locate a row on the SCD2 timeline: its source
record-start sequence
+ * (`recordStartAt`, null only for decomposition tails) and the bounds of its
visible interval
+ * [`startAt`, `endAt`). [[RowClassifier]] classifies a row purely from this
triple.
+ */
+private[autocdc] case class Scd2IntervalColumns(
+ recordStartAt: Column,
+ startAt: Column,
+ endAt: Column) {
+
+ /**
+ * The row's effective ordering position. Decomposition tails carry no
`recordStartAt` and
+ * fall back to their closing sequence (`endAt`), the same convention used by
+ * [[Scd2BatchProcessor.orderChronologicallyPerKeyWindow]].
+ */
+ def effectiveRecordStartAt: Column = F.coalesce(recordStartAt, endAt)
+}
+
+object RowClassifier {
+
+ /**
+ * Synthetic right boundary created by splitting a closed row, temporarily
present during
+ * microbatch reconciliation but never materializes in the target or aux
tables.
+ */
+ private[autocdc] def isDecompositionTail(row: Scd2IntervalColumns): Column =
+ row.recordStartAt.isNull && row.startAt.isNull && row.endAt.isNotNull
+
+ /**
+ * Upsert row that is currently open in the visible timeline. Hidden no-op
upserts are
+ * also open until reconciliation decides whether they should stay hidden.
+ */
+ private[autocdc] def isOpenUpsert(row: Scd2IntervalColumns): Column =
+ row.recordStartAt.isNotNull &&
+ row.startAt.isNotNull &&
+ row.endAt.isNull &&
+ // startAt < recordStartAt implies this row belongs to but is not the
head of some
+ // upsert-run, else this is the head of a run.
+ row.startAt <= row.recordStartAt
+
+ /**
+ * Upsert row whose visible interval has already been closed by a strictly
later event;
+ * the historical counterpart to [[isOpenUpsert]].
+ *
+ * Notably, a zero-width [startAt, endAt) interval is not considered a valid
closed upsert.
+ */
+ private[autocdc] def isClosedUpsert(row: Scd2IntervalColumns): Column =
+ row.recordStartAt.isNotNull &&
+ row.startAt.isNotNull &&
+ row.endAt.isNotNull &&
+ row.recordStartAt < row.endAt &&
+ row.startAt < row.endAt &&
+ // startAt <= recordStartAt covers both the run-head case (startAt ==
recordStartAt)
+ // and the no-op-continuation case (startAt < recordStartAt).
+ row.startAt <= row.recordStartAt
+
+ /**
+ * Any row that semantically encodes an upsert event.
+ */
+ private[autocdc] def isUpsertRepresentingRow(row: Scd2IntervalColumns):
Column =
+ isOpenUpsert(row) || isClosedUpsert(row)
+
+ /**
+ * Delete-representing row, encoded as an instantaneous (zero-width)
interval at `recordStartAt`
+ * (`startAt == endAt == recordStartAt`). Never materializes in the target
table.
+ *
+ * User-data column values on these rows are not part of the SCD2 contract:
they may reflect
+ * the originating delete event, the values of the upsert whose
closed-interval row was bisected
+ * (when promoted from a decomposition tail), or be null altogether.
Reconciliation does not
+ * consume these values for any semantic decision.
+ */
+ private[autocdc] def isDeleteRepresentingRow(row: Scd2IntervalColumns):
Column =
+ row.recordStartAt.isNotNull &&
+ row.startAt.isNotNull &&
+ row.endAt.isNotNull &&
+ row.startAt === row.recordStartAt &&
+ row.endAt === row.recordStartAt
+}
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 c2dee5f9f33a..c29aa50e9e72 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,6 +17,7 @@
package org.apache.spark.sql.pipelines.autocdc
+import org.apache.spark.{SparkException, SparkRuntimeException}
import org.apache.spark.sql.{functions => F, Column, QueryTest, Row}
import org.apache.spark.sql.classic.DataFrame
import org.apache.spark.sql.internal.SQLConf
@@ -108,6 +109,105 @@ class Scd2BatchProcessorSuite extends QueryTest with
SharedSparkSession {
resolvedSequencingType = LongType
)
+ // =============== orderChronologicallyPerKeyWindow tests ===============
+
+ test("orderChronologicallyPerKeyWindow sorts both decomposition tails and
non-tails by " +
+ "effective recordStartAt ascending") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ // Three rows at distinct effective recordStartAts. The decomposition tail
+ // (recordStartAt = null, endAt = 10) takes its endAt as its effective
ordering sequence,
+ // so the expected per-key window order is 5, tail(10), 15.
+ val df = targetTableOf(userSchema)(
+ Row(1, "v15", 15L, null, Row(15L)),
+ Row(1, "tail", null, 10L, Row(null)),
+ Row(1, "v5", 5L, null, Row(5L))
+ )
+
+ val withRn = df.withColumn(
+ "rn", F.row_number().over(processor.orderChronologicallyPerKeyWindow)
+ )
+
+ checkAnswer(
+ df = withRn,
+ expectedAnswer = Seq(
+ Row(1, "v5", 5L, null, Row(5L), 1),
+ Row(1, "tail", null, 10L, Row(null), 2),
+ Row(1, "v15", 15L, null, Row(15L), 3)
+ )
+ )
+ }
+
+ test("orderChronologicallyPerKeyWindow tiebreakers: tails before non-tails,
then " +
+ "upsert-representing rows before tombstones, when effective recordStartAt
ties") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ // Each key isolates one tiebreaker rule under test. All rows in a
partition tie at
+ // effective recordStartAt = 10, so only the tiebreakers determine order.
+ // key=1: tails-first - tail vs tombstone, tail must come first.
+ // key=2: upsert-representing-first (open variant) - open upsert vs
tombstone.
+ // key=3: upsert-representing-first (closed variant) - closed run head
vs tombstone.
+ val df = targetTableOf(userSchema)(
+ Row(1, "tomb", 10L, 10L, Row(10L)),
+ Row(1, "tail", null, 10L, Row(null)),
+ Row(2, "tomb", 10L, 10L, Row(10L)),
+ Row(2, "open", 10L, null, Row(10L)),
+ Row(3, "tomb", 10L, 10L, Row(10L)),
+ Row(3, "closed", 10L, 20L, Row(10L))
+ )
+
+ val withRn = df.withColumn(
+ "rn", F.row_number().over(processor.orderChronologicallyPerKeyWindow)
+ )
+
+ checkAnswer(
+ df = withRn,
+ expectedAnswer = Seq(
+ Row(1, "tail", null, 10L, Row(null), 1),
+ Row(1, "tomb", 10L, 10L, Row(10L), 2),
+ Row(2, "open", 10L, null, Row(10L), 1),
+ Row(2, "tomb", 10L, 10L, Row(10L), 2),
+ Row(3, "closed", 10L, 20L, Row(10L), 1),
+ Row(3, "tomb", 10L, 10L, Row(10L), 2)
+ )
+ )
+ }
+
+ test("orderChronologicallyPerKeyWindow orders rows independently per key") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ // Two keys with interleaved input order. Each key's window must be sorted
independently
+ // by effective recordStartAt - rows from key 2 must not influence
row_number positions
+ // of rows for key 1, and vice versa.
+ val df = targetTableOf(userSchema)(
+ Row(1, "k1-15", 15L, null, Row(15L)),
+ Row(2, "k2-7", 7L, null, Row(7L)),
+ Row(1, "k1-5", 5L, null, Row(5L)),
+ Row(2, "k2-3", 3L, null, Row(3L)),
+ Row(1, "k1-10", 10L, null, Row(10L)),
+ Row(2, "k2-20", 20L, null, Row(20L))
+ )
+
+ val withRn = df.withColumn(
+ "rn", F.row_number().over(processor.orderChronologicallyPerKeyWindow)
+ )
+
+ checkAnswer(
+ df = withRn,
+ expectedAnswer = Seq(
+ Row(1, "k1-5", 5L, null, Row(5L), 1),
+ Row(1, "k1-10", 10L, null, Row(10L), 2),
+ Row(1, "k1-15", 15L, null, Row(15L), 3),
+ Row(2, "k2-3", 3L, null, Row(3L), 1),
+ Row(2, "k2-7", 7L, null, Row(7L), 2),
+ Row(2, "k2-20", 20L, null, Row(20L), 3)
+ )
+ )
+ }
+
// =============== preprocessMicrobatch tests ===============
test("preprocessMicrobatch appends framework columns __START_AT, __END_AT, "
+
@@ -1217,4 +1317,597 @@ class Scd2BatchProcessorSuite extends QueryTest with
SharedSparkSession {
assert(result.collect().isEmpty)
}
+ // =============== decomposeOutOfOrderRows tests ===============
+
+ test("decomposeOutOfOrderRows passes through open rows, tombstones, and " +
+ "last-in-partition closed rows") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType()
+ .add("id", IntegerType)
+ .add("value", StringType)
+ .add("amount", IntegerType)
+
+ // None of these rows is eligible for decomposition:
+ // - "open": endAt is null, so it is not a closed row
+ // - "tomb": startAt == endAt, so it is excluded by the strict `<`
closed check
+ // - "last": closed, but is the last row in its window partition (no
successor)
+ val df = targetTableOf(userSchema)(
+ Row(1, "open", 100, 5L, null, Row(5L)),
+ Row(1, "tomb", 200, 10L, 10L, Row(10L)),
+ Row(1, "last", 300, 15L, 25L, Row(15L))
+ )
+
+ val result = processor.decomposeOutOfOrderRows(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "open", 100, 5L, null, Row(5L)),
+ Row(1, "tomb", 200, 10L, 10L, Row(10L)),
+ Row(1, "last", 300, 15L, 25L, Row(15L))
+ )
+ )
+ }
+
+ test("decomposeOutOfOrderRows leaves a closed row alone when the next event
arrives at " +
+ "exactly its endAt") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType()
+ .add("id", IntegerType)
+ .add("value", StringType)
+ .add("amount", IntegerType)
+
+ // Closed [5, 10] would decompose if the successor's recordStartAt were
strictly less
+ // than 10. Here the successor lands at exactly 10, which means it doesn't
actually
+ // bisect the closed row and therefore shouldn't decompose it.
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", 42, 5L, 10L, Row(5L)),
+ Row(1, "bob", 99, 10L, null, Row(10L))
+ )
+
+ val result = processor.decomposeOutOfOrderRows(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", 42, 5L, 10L, Row(5L)),
+ Row(1, "bob", 99, 10L, null, Row(10L))
+ )
+ )
+ }
+
+ test("decomposeOutOfOrderRows splits a bisected row into head + tail, " +
+ "both inheriting parent data") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType()
+ .add("id", IntegerType)
+ .add("value", StringType)
+ .add("amount", IntegerType)
+
+ // Closed [5, 30] is bisected by a successor at recordStartAt = 15 (15 <
30). Expected:
+ // - head: parent with endAt set to null
+ // - tail: parent with startAt set to null and recordStartAt (in
cdcMetadata) set to null
+ // - successor: passes through
+ // Both head and tail must carry the parent's data columns (value="alice",
amount=42)
+ // identically.
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", 42, 5L, 30L, Row(5L)),
+ Row(1, "bob", 99, 15L, null, Row(15L))
+ )
+
+ val result = processor.decomposeOutOfOrderRows(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", 42, 5L, null, Row(5L)), // head
+ Row(1, "alice", 42, null, 30L, Row(null)), // tail
+ Row(1, "bob", 99, 15L, null, Row(15L)) // bisecting successor
+ )
+ )
+ }
+
+ test("decomposeOutOfOrderRows triggers on any kind of bisecting successor") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType()
+ .add("id", IntegerType)
+ .add("value", StringType)
+ .add("amount", IntegerType)
+
+ // Three independent keys, each with a closed parent [5, 50] bisected by a
successor
+ // of a different kind. All three parents must decompose, regardless of
the successor's
+ // own row kind (the bisection check looks only at recordStartAt <
parent.endAt).
+ val df = targetTableOf(userSchema)(
+ // Key 1: bisected by an open upsert.
+ Row(1, "alice", 1, 5L, 50L, Row(5L)),
+ Row(1, "bob", 2, 10L, null, Row(10L)),
+
+ // Key 2: bisected by a tombstone.
+ Row(2, "carol", 3, 5L, 50L, Row(5L)),
+ Row(2, "dave", 4, 20L, 20L, Row(20L)),
+
+ // Key 3: bisected by another closed non-tombstone.
+ Row(3, "eve", 5, 5L, 50L, Row(5L)),
+ Row(3, "frank", 6, 30L, 40L, Row(30L))
+ )
+
+ val result = processor.decomposeOutOfOrderRows(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ // Key 1.
+ Row(1, "alice", 1, 5L, null, Row(5L)),
+ Row(1, "alice", 1, null, 50L, Row(null)),
+ Row(1, "bob", 2, 10L, null, Row(10L)),
+ // Key 2.
+ Row(2, "carol", 3, 5L, null, Row(5L)),
+ Row(2, "carol", 3, null, 50L, Row(null)),
+ Row(2, "dave", 4, 20L, 20L, Row(20L)),
+ // Key 3.
+ Row(3, "eve", 5, 5L, null, Row(5L)),
+ Row(3, "eve", 5, null, 50L, Row(null)),
+ Row(3, "frank", 6, 30L, 40L, Row(30L))
+ )
+ )
+ }
+
+ test("decomposeOutOfOrderRows uses chronological window order, not input
order") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType()
+ .add("id", IntegerType)
+ .add("value", StringType)
+ .add("amount", IntegerType)
+
+ // Same data as the basic split test but with the input rows shuffled into
a
+ // non-chronological order. The window orders rows by effective
recordStartAt, so the
+ // result must still recognize that [5, 30] is bisected by the row at
recordStartAt = 15.
+ val df = targetTableOf(userSchema)(
+ Row(1, "bob", 99, 15L, null, Row(15L)), // appears first in input
+ Row(1, "alice", 42, 5L, 30L, Row(5L)) // appears last in input but
lower in window
+ )
+
+ val result = processor.decomposeOutOfOrderRows(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", 42, 5L, null, Row(5L)),
+ Row(1, "alice", 42, null, 30L, Row(null)),
+ Row(1, "bob", 99, 15L, null, Row(15L))
+ )
+ )
+ }
+
+ test("decomposeOutOfOrderRows isolates per-key partitions") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType()
+ .add("id", IntegerType)
+ .add("value", StringType)
+ .add("amount", IntegerType)
+
+ // Key 1 has a bisecting pair. Key 2 only contains a single closed row
whose recordStartAt
+ // happens to coincide with key 1's parent. The window partitions by key,
so key 1's
+ // bisecting successor must NOT bleed into key 2's partition.
+ val df = targetTableOf(userSchema)(
+ // Key 1: closed [5, 30] bisected by recordStartAt = 15.
+ Row(1, "alice", 42, 5L, 30L, Row(5L)),
+ Row(1, "bob", 99, 15L, null, Row(15L)),
+
+ // Key 2: a single closed [5, 30] with no successor in its own partition.
+ Row(2, "carol", 7, 5L, 30L, Row(5L))
+ )
+
+ val result = processor.decomposeOutOfOrderRows(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ // Key 1 decomposes.
+ Row(1, "alice", 42, 5L, null, Row(5L)),
+ Row(1, "alice", 42, null, 30L, Row(null)),
+ Row(1, "bob", 99, 15L, null, Row(15L)),
+ // Key 2 passes through.
+ Row(2, "carol", 7, 5L, 30L, Row(5L))
+ )
+ )
+ }
+
+ test("decomposeOutOfOrderRows handles a cascade of consecutive bisected
rows") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType()
+ .add("id", IntegerType)
+ .add("value", StringType)
+ .add("amount", IntegerType)
+
+ // A nested cascade: each closed row (except the innermost) is bisected by
the next.
+ // Decomposition decisions are independent and based on each row's
immediate successor:
+ // [5, 30] bisected by [10, 25] -> decomposes
+ // [10, 25] bisected by [15, 20] -> decomposes
+ // [15, 20] is the last row -> passes through
+ val df = targetTableOf(userSchema)(
+ Row(1, "outer", 1, 5L, 30L, Row(5L)),
+ Row(1, "middle", 2, 10L, 25L, Row(10L)),
+ Row(1, "inner", 3, 15L, 20L, Row(15L))
+ )
+
+ val result = processor.decomposeOutOfOrderRows(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ // outer decomposes.
+ Row(1, "outer", 1, 5L, null, Row(5L)),
+ Row(1, "outer", 1, null, 30L, Row(null)),
+ // middle decomposes.
+ Row(1, "middle", 2, 10L, null, Row(10L)),
+ Row(1, "middle", 2, null, 25L, Row(null)),
+ // inner passes through.
+ Row(1, "inner", 3, 15L, 20L, Row(15L))
+ )
+ )
+ }
+
+ test("decomposeOutOfOrderRows returns empty for empty input") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType()
+ .add("id", IntegerType)
+ .add("value", StringType)
+ .add("amount", IntegerType)
+
+ val df = targetTableOf(userSchema)()
+
+ val result = processor.decomposeOutOfOrderRows(df)
+
+ assert(result.collect().isEmpty)
+ assert(result.columns.toSeq == df.columns.toSeq)
+ }
+
+ test("decomposeOutOfOrderRows preserves per-column schema attributes through
" +
+ "head + tail decomposition") {
+ val processor = processorWithKeys(Seq("id"))
+
+ def commentMetadata(comment: String): Metadata =
+ new MetadataBuilder().putString("comment", comment).build()
+
+ val cdcMetadataInnerSchema = new StructType().add(
+ Scd2BatchProcessor.recordStartAtFieldName,
+ LongType,
+ nullable = true,
+ metadata = commentMetadata("inner __RECORD_START_AT")
+ )
+
+ val schema = new StructType()
+ .add("id", IntegerType, nullable = false, metadata =
commentMetadata("user key"))
+ .add("value", StringType, nullable = true, metadata =
commentMetadata("user data"))
+ .add(
+ Scd2BatchProcessor.startAtColName, LongType, nullable = true,
+ metadata = commentMetadata("framework __START_AT"))
+ .add(
+ Scd2BatchProcessor.endAtColName, LongType, nullable = true,
+ metadata = commentMetadata("framework __END_AT"))
+ .add(
+ AutoCdcReservedNames.cdcMetadataColName, cdcMetadataInnerSchema,
nullable = false,
+ metadata = commentMetadata("framework _cdc_metadata"))
+
+ // Closed [5, 30] bisected by recordStartAt = 15.
+ val df = microbatchOf(schema)(
+ Row(1, "alice", 5L, 30L, Row(5L)),
+ Row(1, "bob", 15L, null, Row(15L))
+ )
+
+ val result = processor.decomposeOutOfOrderRows(df)
+
+ // Output schema must match the input field-for-field on name, dataType
(which carries
+ // inner-struct field metadata), nullable, and outer metadata.
+ schema.fields.zip(result.schema.fields).foreach { case (in, out) =>
+ assert(in.name == out.name)
+ assert(in.dataType == out.dataType)
+ assert(in.nullable == out.nullable)
+ assert(in.metadata == out.metadata)
+ }
+ }
+
+ // =============== assertWellFormedRowsPostDecomposition tests
===============
+
+ test("assertWellFormedRowsPostDecomposition passes through a dataframe whose
rows match all " +
+ "four canonical shapes") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ // One row of each canonical shape.
+ // tombstone: startAt = endAt = recordStartAt
+ // open upsert: endAt is null, startAt and recordStartAt non-null
+ // closed upsert: startAt < endAt, recordStartAt non-null
+ // decomposition tail: startAt and recordStartAt null, endAt non-null
+ val df = targetTableOf(userSchema)(
+ Row(1, "tomb", 10L, 10L, Row(10L)),
+ Row(2, "open", 20L, null, Row(20L)),
+ Row(3, "closed", 30L, 40L, Row(30L)),
+ Row(4, "tail", null, 50L, Row(null))
+ )
+
+ checkAnswer(
+ df = processor.assertWellFormedRowsPostDecomposition(df, batchId = 0),
+ expectedAnswer = Seq(
+ Row(1, "tomb", 10L, 10L, Row(10L)),
+ Row(2, "open", 20L, null, Row(20L)),
+ Row(3, "closed", 30L, 40L, Row(30L)),
+ Row(4, "tail", null, 50L, Row(null))
+ )
+ )
+ }
+
+ test("assertWellFormedRowsPostDecomposition throws on invalid row shape") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ // recordStartAt = null + startAt = 5 + endAt = 10 matches none of the
canonical shapes:
+ // the decomposition-tail kind rejects on startAt non-null, while the
tombstone, open
+ // upsert, and closed upsert kinds all reject on recordStartAt null.
+ val df = targetTableOf(userSchema)(
+ Row(1, "malformed", 5L, 10L, Row(null))
+ )
+
+ val wrapper = intercept[SparkException] {
+ processor.assertWellFormedRowsPostDecomposition(df, batchId =
0).collect()
+ }
+ assert(wrapper.getCause.isInstanceOf[SparkRuntimeException],
+ s"Expected SparkRuntimeException cause, got ${wrapper.getCause}")
+ val cause = wrapper.getCause.asInstanceOf[SparkRuntimeException]
+ assert(cause.getCondition == "INTERNAL_ERROR")
+ assert(cause.getMessage.contains("post-decomposition"))
+ assert(cause.getMessage.contains(Scd2BatchProcessor.startAtColName))
+ assert(cause.getMessage.contains(Scd2BatchProcessor.endAtColName))
+ }
+
+ test("assertWellFormedRowsPostDecomposition does not introduce extra
columns") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ val df = targetTableOf(userSchema)(
+ Row(1, "open", 10L, null, Row(10L))
+ )
+
+ val result = processor.assertWellFormedRowsPostDecomposition(df, batchId =
0)
+
+ assert(result.schema == df.schema)
+ }
+
+ // =============== dropRedundantRowsPostDecomposition tests ===============
+
+ test("dropRedundantRowsPostDecomposition passes through events with distinct
sequences") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ // Distinct effective recordStartAts within the dataframe, so no
redundancies - identity
+ // transformation expected.
+ val df = targetTableOf(userSchema)(
+ Row(1, "v5", 5L, null, Row(5L)),
+ Row(1, "v10", 10L, null, Row(10L)),
+ Row(1, "v15", 15L, 20L, Row(15L))
+ )
+
+ val result = processor.dropRedundantRowsPostDecomposition(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "v5", 5L, null, Row(5L)),
+ Row(1, "v10", 10L, null, Row(10L)),
+ Row(1, "v15", 15L, 20L, Row(15L))
+ )
+ )
+ }
+
+ test("dropRedundantRowsPostDecomposition drops a same-event duplicate") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ // Two open upserts tied on effective recordStartAt = 10. The
chronologically-leading
+ // copy is dropped; exactly one copy survives. We do not assert which
user-data variant
+ // survives because the window's tiebreaker among truly-identical rows is
intentionally
+ // undefined.
+ val df = targetTableOf(userSchema)(
+ Row(1, "first", 10L, null, Row(10L)),
+ Row(1, "second", 10L, null, Row(10L))
+ )
+
+ val result = processor.dropRedundantRowsPostDecomposition(df)
+ val survivors = result.collect()
+
+ assert(survivors.length == 1)
+ assert(Set("first", "second").contains(survivors.head.getString(1)))
+ }
+
+ test("dropRedundantRowsPostDecomposition drops an open upsert overtaken at
the same instant") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ // Open upsert at recordStartAt=10 followed by a tombstone at the same
instant
+ // (effectiveRecordStartAt=10). The upsert opens at an instant the
tombstone already
+ // overtakes, leaving it 0-width: it ties with its successor on effective
sequence
+ // and is dropped. The tombstone (last in partition) survives.
+ val df = targetTableOf(userSchema)(
+ Row(1, "open", 10L, null, Row(10L)),
+ Row(1, "tomb", 10L, 10L, Row(10L))
+ )
+
+ val result = processor.dropRedundantRowsPostDecomposition(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "tomb", 10L, 10L, Row(10L))
+ )
+ )
+ }
+
+ test("dropRedundantRowsPostDecomposition drops a decomposition tail
coincident with " +
+ "an incoming event") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ // Decomposition tail closing at endAt=30 followed by a non-tail event at
+ // recordStartAt=30. Both have effective recordStartAt = 30. The synthetic
delete the
+ // tail encodes is already represented by the coincident event, so the
leading tail is
+ // dropped. The event survives.
+ val df = targetTableOf(userSchema)(
+ Row(1, "tail", null, 30L, Row(null)),
+ Row(1, "event", 30L, null, Row(30L))
+ )
+
+ val result = processor.dropRedundantRowsPostDecomposition(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "event", 30L, null, Row(30L))
+ )
+ )
+ }
+
+ test("dropRedundantRowsPostDecomposition preserves rows when the next event
arrives at " +
+ "a strictly later instant") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ // Open upsert at recordStartAt=10 and decomposition tail closing at
endAt=30 are each
+ // followed by a row at a strictly later effective sequence. No row ties
with its
+ // successor on effective recordStartAt, so the redundancy filter doesn't
fire. Every
+ // row survives.
+ val df = targetTableOf(userSchema)(
+ Row(1, "open", 10L, null, Row(10L)),
+ Row(1, "next1", 15L, null, Row(15L)),
+ Row(1, "tail", null, 30L, Row(null)),
+ Row(1, "next2", 35L, null, Row(35L))
+ )
+
+ val result = processor.dropRedundantRowsPostDecomposition(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "open", 10L, null, Row(10L)),
+ Row(1, "next1", 15L, null, Row(15L)),
+ Row(1, "tail", null, 30L, Row(null)),
+ Row(1, "next2", 35L, null, Row(35L))
+ )
+ )
+ }
+
+ test("dropRedundantRowsPostDecomposition preserves the last row in every
per-key partition") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ // Each key has a single row; LEAD(1) is null at the end of every
partition, so the
+ // redundancy filter is vacuous. If per-key isolation were broken, key 1's
row would
+ // see key 2's row as its successor and tie on effective sequence; proper
partitioning
+ // preserves both.
+ val df = targetTableOf(userSchema)(
+ Row(1, "v10", 10L, null, Row(10L)),
+ Row(2, "v10", 10L, null, Row(10L))
+ )
+
+ val result = processor.dropRedundantRowsPostDecomposition(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "v10", 10L, null, Row(10L)),
+ Row(2, "v10", 10L, null, Row(10L))
+ )
+ )
+ }
+
+ test("dropRedundantRowsPostDecomposition cascades across consecutive
duplicate rows") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ // Three identical open upserts followed by a distinct event. Each
adjacent same-effective
+ // pair ties: the chronologically-leading two copies are dropped, leaving
exactly one
+ // duplicate plus the distinct event. We don't assert which user-data
variant of the
+ // duplicate survives.
+ val df = targetTableOf(userSchema)(
+ Row(1, "dup1", 5L, null, Row(5L)),
+ Row(1, "dup2", 5L, null, Row(5L)),
+ Row(1, "dup3", 5L, null, Row(5L)),
+ Row(1, "different", 10L, null, Row(10L))
+ )
+
+ val result = processor.dropRedundantRowsPostDecomposition(df)
+ val survivors = result.collect()
+
+ assert(survivors.length == 2)
+ val dupSurvivor = survivors.find(_.getLong(2) == 5L)
+ assert(dupSurvivor.isDefined)
+ assert(Set("dup1", "dup2", "dup3").contains(dupSurvivor.get.getString(1)))
+ assert(survivors.exists(r => r.getString(1) == "different"))
+ }
+
+ test("dropRedundantRowsPostDecomposition: same-class collisions reduce to
exactly one row") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ // Each key encodes one same-class collision at effective recordStartAt =
10. The window
+ // does not sub-order rows within a class, so which row survives is
undefined. Hence we only
+ // assert: (a) exactly one row survives per key, and (b) the survivor is
indeed one of the
+ // inputs.
+ // key=1: two open upsert rows
+ // key=2: open upsert + closed run head (both upsert-representing).
+ // key=3: two closed run heads (structurally impossible under SCD2
invariants but defensively
+ // covered).
+ // key=4: two tombstones.
+ val df = targetTableOf(userSchema)(
+ Row(1, "openA", 10L, null, Row(10L)),
+ Row(1, "openB", 10L, null, Row(10L)),
+ Row(2, "open", 10L, null, Row(10L)),
+ Row(2, "closed", 10L, 20L, Row(10L)),
+ Row(3, "closedA", 10L, 20L, Row(10L)),
+ Row(3, "closedB", 10L, 20L, Row(10L)),
+ Row(4, "tombA", 10L, 10L, Row(10L)),
+ Row(4, "tombB", 10L, 10L, Row(10L))
+ )
+
+ val expectedSurvivorsPerKey: Map[Int, Set[String]] = Map(
+ 1 -> Set("openA", "openB"),
+ 2 -> Set("open", "closed"),
+ 3 -> Set("closedA", "closedB"),
+ 4 -> Set("tombA", "tombB")
+ )
+
+ val survivors = processor.dropRedundantRowsPostDecomposition(df).collect()
+ assert(survivors.length == 4)
+
+ expectedSurvivorsPerKey.foreach { case (k, validValues) =>
+ val perKey = survivors.filter(_.getInt(0) == k)
+ assert(perKey.length == 1)
+ assert(validValues.contains(perKey.head.getString(1)))
+ }
+ }
+
+ test("dropRedundantRowsPostDecomposition collapses many rows tied on
effective " +
+ "recordStartAt to the trailing one") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ // All three rows tie at effectiveRecordStartAt=10. Window tiebreakers
order them as
+ // tail (tails-first), open upsert (upsert-representing-first), tombstone.
The
+ // redundancy filter then drops every row whose successor shares its
effective sequence,
+ // leaving only the trailing tombstone.
+ val df = targetTableOf(userSchema)(
+ Row(1, "tail", null, 10L, Row(null)),
+ Row(1, "open", 10L, null, Row(10L)),
+ Row(1, "tomb", 10L, 10L, Row(10L))
+ )
+
+ val result = processor.dropRedundantRowsPostDecomposition(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "tomb", 10L, 10L, Row(10L))
+ )
+ )
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
