This is an automated email from the ASF dual-hosted git repository.
szehon-ho pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new cebd378f2e17 [SPARK-57356][SDP] Implement SCD2 Batch Processor;
Cleanup Delete Encoding Rows Post-Reconciliation
cebd378f2e17 is described below
commit cebd378f2e1710b30918e854407b53b0071e13a6
Author: Andreas Neumann <[email protected]>
AuthorDate: Wed Jul 22 15:08:50 2026 -0700
[SPARK-57356][SDP] Implement SCD2 Batch Processor; Cleanup Delete Encoding
Rows Post-Reconciliation
### What changes were proposed in this pull request?
This PR adds two post-reconciliation cleanup stages to the SCD2 batch
processor
(`Scd2BatchProcessor`), used by Declarative Pipelines' AUTO CDC / SCD
Type 2 processing.
Both operate on the output of `reconcileStartAndEndAt`, which is where an
affected key's
rows get their final `startAt` / `endAt` intervals.
- **`dropLeftoverDeletesPostReconciliation`** — drops delete-encoded rows
(tombstones and
decomposition tails) that became redundant after reconciliation. A
delete-encoded row is
redundant when the immediately preceding row's reconciled `endAt`
already equals its own
sequence, i.e. the preceding upsert already encodes the delete
boundary. Redundancy is
computed per key using a `lag` over the chronological per-key window
(`__previous_end_at` /
`__is_redundant_delete_encoding` temporary columns, dropped before
returning).
- e.g. an open upsert `[10, null)` followed by a tombstone at `15`
reconciles to a closed
upsert `[10, 15)`, making the tombstone redundant.
- e.g. an existing closed `[10, 20)` bisected by an event at `15`
reconciles the event to
`[15, 20)`, making the `[null, 20)` decomposition tail redundant.
- **`promoteDecompositionTailsToTombstones`** — converts decomposition
tails that *survive*
the deletion cleanup above into tombstones. A surviving tail is an
unmatched delete
boundary; rewriting its `recordStartAt` and `startAt` to the tail's end
sequence lets
downstream auxiliary-table handling preserve it as a tombstone. Column
metadata and the
input schema are preserved.
Both methods are `private[autocdc]` building blocks with unit coverage;
they are not yet
wired into the top-level processing flow (a follow-up composes them into
the pipeline).
### Why are the changes needed?
After start/end reconciliation, some delete-encoded rows carry
information that the
reconciled upserts already express, and some decomposition tails remain
as standalone delete
boundaries. Without this cleanup, redundant tombstones/tails would be
routed to the auxiliary
table needlessly, and unmatched tails would not be represented in the
form downstream aux
handling expects. These two stages normalize the reconciled output so
that each key's history
is encoded once and correctly.
### Does this PR introduce _any_ user-facing change?
No, SCD2 is an unreleased feature.
### How was this patch tested?
Added 13 unit tests to `Scd2BatchProcessorSuite` covering the two new
methods (drop-vs-keep
for tombstones and decomposition tails, no-window-predecessor cases,
upserts never dropped,
per-key independence, combined redundant drops; and tail-to-tombstone
rewriting, pass-through
of tombstones/upserts, user-column and schema/metadata preservation,
per-key promotion).
### Was this patch authored or co-authored using generative AI tooling?
Co-authored with Claude Opus 4.7 and 4.8
Closes #57365 from anew/SPARK-57356-post-reconciliation-cleanup.
Lead-authored-by: Andreas Neumann <[email protected]>
Co-authored-by: Anish Mahto <[email protected]>
Signed-off-by: Szehon Ho <[email protected]>
(cherry picked from commit e350bb02d90a107c59a4997373008a83585a9007)
Signed-off-by: Szehon Ho <[email protected]>
---
.../sql/pipelines/autocdc/Scd2BatchProcessor.scala | 114 +++++++
.../autocdc/Scd2BatchProcessorSuite.scala | 379 +++++++++++++++++++++
2 files changed, 493 insertions(+)
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 27b9036523f1..6e5af84216b5 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
@@ -811,6 +811,110 @@ case class Scd2BatchProcessor(
staged.select(outputColumns.toImmutableArraySeq: _*)
}
+ /**
+ * Drop delete-encoded rows (tombstones and decomposition tails) that became
redundant after
+ * reconciliation.
+ *
+ * The rule is the same for both row kinds: a delete-encoded row encodes a
delete boundary in its
+ * [[endAtColName]], and it is redundant once the immediately preceding
reconciled upsert closes
+ * (its [[endAtColName]]) exactly on that same boundary. When they coincide,
the closed upsert
+ * already carries the boundary, so the standalone delete-encoded row can be
dropped rather than
+ * routed to aux. Only the preceding upsert's `endAt` matters here; its
`startAt` (identity) is
+ * irrelevant to the comparison.
+ *
+ * Both examples reduce to that single check - the preceding upsert's
reconciled `endAt` equals
+ * the delete-encoded row's boundary:
+ * - Tombstone: an open upsert `[startAt=10, endAt=null)` followed by a
tombstone at `15`
+ * reconciles into a closed upsert `[10, 15)`. Its `endAt=15` matches
the tombstone's
+ * boundary `15`, so the tombstone is redundant.
+ * - Decomposition tail: an existing closed `[10, 20)` is bisected by an
out-of-order event at
+ * `15`. Decomposition first splits the original row into an open head
`[10, null)` plus a
+ * decomposition tail `[null, 20)` - a synthetic row with a null
recordStartAt that carries
+ * the original right boundary `20` so it can be re-closed later. The
bisecting event then
+ * reconciles into `[15, 20)`, whose `endAt=20` matches the tail's
boundary `20`, so the tail
+ * is redundant. (A tail that does *not* coincide with a preceding
upsert survives and is
+ * later turned into a tombstone by
[[promoteDecompositionTailsToTombstones]].)
+ */
+ private[autocdc] def dropLeftoverDeletesPostReconciliation(
+ reconciledDf: DataFrame): DataFrame = {
+ val recordStartAt =
+
Scd2BatchProcessor.recordStartAtOf(F.col(AutoCdcReservedNames.cdcMetadataColName))
+ val startAt = F.col(Scd2BatchProcessor.startAtColName)
+ val endAt = F.col(Scd2BatchProcessor.endAtColName)
+
+ // Both tombstones and decomposition tails encode a delete boundary in
their `endAt`. Either
+ // becomes redundant when the immediately preceding upsert was reconciled
to close exactly on
+ // that boundary, since the resulting closed upsert already carries it.
+ val row = Scd2IntervalColumns(recordStartAt, startAt, endAt)
+ val isTombstone = RowClassifier.isTombstone(row)
+ val isDecompositionTail = RowClassifier.isDecompositionTail(row)
+ val isDeleteEncodedRow = isTombstone || isDecompositionTail
+
+ // The immediately preceding chronologically-ordered row for the same key.
+ val previous = row.lagBy(1, orderChronologicallyPerKeyWindow)
+
+ val withWindowCols = reconciledDf
+ .withColumn(
+ Scd2BatchProcessor.isRedundantDeleteEncodingColName,
+ isDeleteEncodedRow &&
+ // Defensive: the predecessor should always be an upsert, because
upstream cleanup
+ // prevents adjacent delete-encoded rows sharing the same boundary
from reaching here.
+ // Asserting it directly keeps this method's redundancy rule
matching its documented
+ // "immediately preceding upsert" invariant, rather than dropping a
delete-encoded row
+ // that merely happens to abut another delete-encoded row on the
same boundary.
+ RowClassifier.isUpsertRepresentingRow(previous) &&
+ (previous.endAt <=> endAt)
+ )
+
+ withWindowCols
+ .filter(!F.col(Scd2BatchProcessor.isRedundantDeleteEncodingColName))
+ .drop(Scd2BatchProcessor.isRedundantDeleteEncodingColName)
+ }
+
+ /**
+ * Convert surviving decomposition tails into tombstones.
+ *
+ * A decomposition tail that survives deletion cleanup is an unmatched
delete boundary; setting
+ * [[recordStartAtFieldName]], and [[startAtColName]] to the tail's end
sequence lets downstream
+ * aux handling preserve it as a tombstone.
+ *
+ * For example, if an existing closed upsert `[startAt=10, endAt=20)` is
bisected by an event at
+ * `15`, decomposition first produces an open head `[10, null)` plus a tail
`[null, 20)`;
+ * reconciliation may close the head as `[10, 15)`, leaving the tail's
boundary at `20` to be
+ * promoted into a tombstone.
+ */
+ private[autocdc] def promoteDecompositionTailsToTombstones(
+ reconciledDf: DataFrame): DataFrame = {
+ val recordStartAt =
+
Scd2BatchProcessor.recordStartAtOf(F.col(AutoCdcReservedNames.cdcMetadataColName))
+ val startAt = F.col(Scd2BatchProcessor.startAtColName)
+ val endAt = F.col(Scd2BatchProcessor.endAtColName)
+ val isDecompositionTail =
+ RowClassifier.isDecompositionTail(Scd2IntervalColumns(recordStartAt,
startAt, endAt))
+
+ val outputColumns = reconciledDf.columns.map {
+ case c if c == AutoCdcReservedNames.cdcMetadataColName =>
+ val metadata = reconciledDf.schema(c).metadata
+ F.when(
+ isDecompositionTail,
+ Scd2BatchProcessor.constructCdcMetadataCol(
+ recordStartAt = endAt,
+ sequencingType = resolvedSequencingType
+ )
+ ).otherwise(F.col(c)).as(c, metadata)
+ case c if c == Scd2BatchProcessor.startAtColName =>
+ val metadata = reconciledDf.schema(c).metadata
+ F.when(isDecompositionTail, endAt).otherwise(startAt).as(c, metadata)
+ // The other cases match framework columns whose names are known and
identifier-safe; only
+ // this default case handles arbitrary user-column names, which may
contain dots or other
+ // special characters, so it is the only one that needs quoting.
+ case c =>
+ F.col(QuotingUtils.quoteIdentifier(c))
+ }
+
+ reconciledDf.select(outputColumns.toImmutableArraySeq: _*)
+ }
+
/**
* Return the schema field names of columns selected for history-tracking on
`df`:
* the eligible user-data columns (those not in [[ChangeArgs.keys]] or the
framework
@@ -1044,6 +1148,16 @@ object Scd2BatchProcessor {
private val finalEndAtColName: String =
s"${AutoCdcReservedNames.prefix}final_end_at"
+ /**
+ * Name of the temporary column used by
[[dropLeftoverDeletesPostReconciliation]] to flag
+ * delete-encoded rows (tombstones and decomposition tails) already encoded
by the immediately
+ * preceding upsert row's [[endAtColName]].
+ *
+ * Temporary in that the column has no observable side effect or persistence
across microbatches.
+ */
+ private val isRedundantDeleteEncodingColName: String =
+ s"${AutoCdcReservedNames.prefix}is_redundant_delete_encoding"
+
/**
* 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
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 649126222a24..0e671721bacc 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
@@ -2661,4 +2661,383 @@ class Scd2BatchProcessorSuite extends QueryTest with
SharedSparkSession {
assert(result.collect().isEmpty)
assert(result.columns.toSeq == df.columns.toSeq)
}
+
+ // =============== dropLeftoverDeletesPostReconciliation tests
===============
+
+ test("dropLeftoverDeletesPostReconciliation drops a tombstone whose sequence
the preceding " +
+ "upsert was reconciled to close on") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ // A closed upsert [10, 15) immediately followed by a tombstone at 15. The
upsert's reconciled
+ // endAt already encodes the delete boundary, so the standalone tombstone
is redundant.
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", 10L, 15L, Row(10L)),
+ Row(1, "alice", 15L, 15L, Row(15L))
+ )
+
+ val result = processor.dropLeftoverDeletesPostReconciliation(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", 10L, 15L, Row(10L))
+ )
+ )
+ }
+
+ test("dropLeftoverDeletesPostReconciliation drops a decomposition tail whose
boundary the " +
+ "preceding upsert was reconciled to close on") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ // Reconciled state of an existing closed [10, 20) bisected by an event at
15: the head closes
+ // at 15, the event closes at the tail's boundary 20, leaving the [null,
20) tail redundant
+ // because the [15, 20) upsert already encodes the boundary.
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", 10L, 15L, Row(10L)),
+ Row(1, "bob", 15L, 20L, Row(15L)),
+ Row(1, "alice", null, 20L, Row(null))
+ )
+
+ val result = processor.dropLeftoverDeletesPostReconciliation(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", 10L, 15L, Row(10L)),
+ Row(1, "bob", 15L, 20L, Row(15L))
+ )
+ )
+ }
+
+ test("dropLeftoverDeletesPostReconciliation keeps a tombstone whose sequence
differs from " +
+ "the preceding upsert's reconciled endAt") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ // The closed upsert ends at 15 but the tombstone is at 20, so the delete
boundary is not
+ // encoded by the preceding row and the tombstone must survive.
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", 5L, 15L, Row(5L)),
+ Row(1, "alice", 20L, 20L, Row(20L))
+ )
+
+ val result = processor.dropLeftoverDeletesPostReconciliation(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", 5L, 15L, Row(5L)),
+ Row(1, "alice", 20L, 20L, Row(20L))
+ )
+ )
+ }
+
+ test("dropLeftoverDeletesPostReconciliation keeps a decomposition tail whose
boundary the " +
+ "preceding row does not close on") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ // The preceding closed upsert ends at 12, strictly before the tail's
boundary 20, so the
+ // tail is an unmatched delete boundary that must survive for promotion to
a tombstone.
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", 5L, 12L, Row(5L)),
+ Row(1, "alice", null, 20L, Row(null))
+ )
+
+ val result = processor.dropLeftoverDeletesPostReconciliation(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", 5L, 12L, Row(5L)),
+ Row(1, "alice", null, 20L, Row(null))
+ )
+ )
+ }
+
+ test("dropLeftoverDeletesPostReconciliation keeps a delete-encoded row that
has no window " +
+ "predecessor") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ // With no predecessor, previousEndAt is null and `null <=> endAt` is
false, so a leading
+ // tombstone (key 1) and a leading decomposition tail (key 2) both survive.
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", 15L, 15L, Row(15L)),
+ Row(2, "bob", null, 20L, Row(null))
+ )
+
+ val result = processor.dropLeftoverDeletesPostReconciliation(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", 15L, 15L, Row(15L)),
+ Row(2, "bob", null, 20L, Row(null))
+ )
+ )
+ }
+
+ test("dropLeftoverDeletesPostReconciliation never drops an upsert even when
its endAt " +
+ "equals the preceding row's endAt") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ // Two closed upserts whose endAts coincide at 20. The `<=>` predicate
matches, but neither
+ // row is delete-encoded, so the `isDeleteEncodedRow` guard must keep
both. The overlapping
+ // intervals are synthetic here purely to isolate the guard.
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", 5L, 20L, Row(5L)),
+ Row(1, "bob", 10L, 20L, Row(10L))
+ )
+
+ val result = processor.dropLeftoverDeletesPostReconciliation(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", 5L, 20L, Row(5L)),
+ Row(1, "bob", 10L, 20L, Row(10L))
+ )
+ )
+ }
+
+ test("dropLeftoverDeletesPostReconciliation evaluates redundancy
independently per key") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ // key 1: a closed upsert [10, 15) followed by a redundant tombstone at 15
(dropped).
+ // key 2: a tombstone at 15 that is the first row in its window (kept) -
the key-1 upsert
+ // must not be treated as its predecessor.
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", 10L, 15L, Row(10L)),
+ Row(1, "alice", 15L, 15L, Row(15L)),
+ Row(2, "bob", 15L, 15L, Row(15L))
+ )
+
+ val result = processor.dropLeftoverDeletesPostReconciliation(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", 10L, 15L, Row(10L)),
+ Row(2, "bob", 15L, 15L, Row(15L))
+ )
+ )
+ }
+
+ test("dropLeftoverDeletesPostReconciliation drops a redundant tombstone and
a redundant " +
+ "decomposition tail in a single pass") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ // [5, 10) closed upsert then a redundant tombstone at 10, then [15, 20)
closed upsert then a
+ // redundant tail at 20. Both delete-encoded rows are encoded by their
immediate predecessors
+ // and drop together in one window pass.
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", 5L, 10L, Row(5L)),
+ Row(1, "alice", 10L, 10L, Row(10L)),
+ Row(1, "bob", 15L, 20L, Row(15L)),
+ Row(1, "alice", null, 20L, Row(null))
+ )
+
+ val result = processor.dropLeftoverDeletesPostReconciliation(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", 5L, 10L, Row(5L)),
+ Row(1, "bob", 15L, 20L, Row(15L))
+ )
+ )
+ }
+
+ test("dropLeftoverDeletesPostReconciliation keeps a delete-encoded row whose
predecessor is " +
+ "itself a delete-encoded row on the same boundary") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ // A closed upsert [5, 15) followed by two tombstones at 15. The first
tombstone is redundant
+ // (its predecessor is the upsert closing on 15) and drops. The second
tombstone's predecessor
+ // is the *first tombstone*, not an upsert, so - although their endAts
coincide at 15 - it must
+ // survive: only a preceding upsert makes a delete boundary redundant.
This guards the
+ // documented "immediately preceding upsert" invariant against adjacent
delete-encoded rows.
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", 5L, 15L, Row(5L)),
+ Row(1, "alice", 15L, 15L, Row(15L)),
+ Row(1, "alice", 15L, 15L, Row(15L))
+ )
+
+ val result = processor.dropLeftoverDeletesPostReconciliation(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", 5L, 15L, Row(5L)),
+ Row(1, "alice", 15L, 15L, Row(15L))
+ )
+ )
+ }
+
+ // =============== promoteDecompositionTailsToTombstones tests
===============
+
+ test("promoteDecompositionTailsToTombstones rewrites a decomposition tail
into a tombstone") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ // The [null, 20) tail becomes a tombstone [20, 20] with recordStartAt =
20; the closed
+ // upsert is untouched.
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", 10L, 20L, Row(10L)),
+ Row(1, "alice", null, 20L, Row(null))
+ )
+
+ val result = processor.promoteDecompositionTailsToTombstones(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", 10L, 20L, Row(10L)),
+ Row(1, "alice", 20L, 20L, Row(20L))
+ )
+ )
+ }
+
+ test("promoteDecompositionTailsToTombstones leaves tombstones and upserts
unchanged") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ // No decomposition tails present: a tombstone, an open upsert, and a
closed upsert must all
+ // pass through identically.
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", 15L, 15L, Row(15L)),
+ Row(2, "bob", 5L, null, Row(5L)),
+ Row(3, "carol", 5L, 15L, Row(5L))
+ )
+
+ val result = processor.promoteDecompositionTailsToTombstones(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", 15L, 15L, Row(15L)),
+ Row(2, "bob", 5L, null, Row(5L)),
+ Row(3, "carol", 5L, 15L, Row(5L))
+ )
+ )
+ }
+
+ test("promoteDecompositionTailsToTombstones preserves the tail's inherited
user columns") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType()
+ .add("id", IntegerType)
+ .add("name", StringType)
+ .add("status", StringType)
+
+ // Only the framework columns (startAt and the cdc-metadata recordStartAt)
are rewritten to
+ // the tail's boundary; the inherited user columns must survive verbatim.
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", "active", null, 20L, Row(null))
+ )
+
+ val result = processor.promoteDecompositionTailsToTombstones(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", "active", 20L, 20L, Row(20L))
+ )
+ )
+ }
+
+ test("promoteDecompositionTailsToTombstones preserves the input schema,
column metadata, " +
+ "and row count") {
+ val processor = processorWithKeys(Seq("id"))
+
+ def commentMetadata(comment: String): Metadata =
+ new MetadataBuilder().putString("comment", comment).build()
+
+ val 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,
+ Scd2BatchProcessor.cdcMetadataColSchema(LongType), nullable = false,
+ metadata = commentMetadata("framework _cdc_metadata"))
+
+ // A tail (rewritten) plus a non-tail (passed through) so both projection
paths are exercised.
+ val df = microbatchOf(schema)(
+ Row(1, "alice", null, 20L, Row(null)),
+ Row(1, "alice", 5L, 20L, Row(5L))
+ )
+
+ val result = processor.promoteDecompositionTailsToTombstones(df)
+
+ schema.fields.zip(result.schema.fields).foreach { case (in, out) =>
+ assert(in.name == out.name)
+ assert(in.dataType == out.dataType)
+ assert(in.nullable == out.nullable)
+ assert(in.metadata == out.metadata)
+ }
+ assert(result.count() == df.count())
+ }
+
+ test("promoteDecompositionTailsToTombstones promotes tails independently
across keys") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType().add("id", IntegerType).add("value",
StringType)
+
+ // Each key carries a closed upsert plus a decomposition tail; only the
tails are rewritten.
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", 5L, 20L, Row(5L)),
+ Row(1, "alice", null, 20L, Row(null)),
+ Row(2, "bob", 8L, 30L, Row(8L)),
+ Row(2, "bob", null, 30L, Row(null))
+ )
+
+ val result = processor.promoteDecompositionTailsToTombstones(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", 5L, 20L, Row(5L)),
+ Row(1, "alice", 20L, 20L, Row(20L)),
+ Row(2, "bob", 8L, 30L, Row(8L)),
+ Row(2, "bob", 30L, 30L, Row(30L))
+ )
+ )
+ }
+
+ test("promoteDecompositionTailsToTombstones quotes pass-through user columns
whose names " +
+ "contain a dot") {
+ val processor = processorWithKeys(Seq("id"))
+ val userSchema = new StructType()
+ .add("id", IntegerType)
+ .add("user.name", StringType)
+
+ // The `user.name` user column passes through the default branch
unchanged. Without quoting,
+ // `F.col("user.name")` would be parsed as a nested-field access (struct
`user`, field
+ // `name`) and fail to resolve, so the select would throw. A decomposition
tail (recordStartAt
+ // and startAt null, endAt=20) is promoted to a tombstone at 20 while the
dotted user column
+ // survives verbatim.
+ val df = targetTableOf(userSchema)(
+ Row(1, "alice", null, 20L, Row(null))
+ )
+
+ val result = processor.promoteDecompositionTailsToTombstones(df)
+
+ checkAnswer(
+ df = result,
+ expectedAnswer = Seq(
+ Row(1, "alice", 20L, 20L, Row(20L))
+ )
+ )
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
