From 787d705b6c7aba3c0e994a042941aaec7f68ee72 Mon Sep 17 00:00:00 2001 From: Anish Mahto Date: Wed, 10 Jun 2026 18:24:09 +0000 Subject: [PATCH 1/5] implement post-reconciliation cleanup (cherry picked from commit 75ce1702dbcbfe2bf92ffe9128a820378489b3dc) --- .../autocdc/Scd2BatchProcessor.scala | 100 ++++++ .../autocdc/Scd2BatchProcessorSuite.scala | 326 ++++++++++++++++++ 2 files changed, 426 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 27b9036523f12..0a2044596a6f6 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,93 @@ case class Scd2BatchProcessor( staged.select(outputColumns.toImmutableArraySeq: _*) } + /** + * Drop delete-encoded rows (tombstones and decomposition tails) that became redundant after + * reconciliation. + * + * A tombstone or decomposition tail is redundant when the immediately preceding row's reconciled + * [[endAtColName]] equals its own sequence: the preceding upsert already encodes the delete + * boundary, so the standalone delete-encoded row should no longer be routed to aux. + * + * For example, an open upsert `[startAt=10, endAt=null)` followed by a tombstone at `15` + * reconciles into a closed upsert `[10, 15)`, making the tombstone redundant. Likewise, an + * existing closed `[10, 20)` bisected by an event at `15` reconciles the event into `[15, 20)`, + * making the `[null, 20)` decomposition tail redundant. + */ + 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 + + val withWindowCols = reconciledDf + .withColumn( + Scd2BatchProcessor.previousEndAtColName, + F.lag(endAt, 1).over(orderChronologicallyPerKeyWindow) + ) + .withColumn( + Scd2BatchProcessor.isRedundantDeleteEncodingColName, + isDeleteEncodedRow && (F.col(Scd2BatchProcessor.previousEndAtColName) <=> endAt) + ) + + withWindowCols + .filter(!F.col(Scd2BatchProcessor.isRedundantDeleteEncodingColName)) + .drop( + Scd2BatchProcessor.previousEndAtColName, + Scd2BatchProcessor.isRedundantDeleteEncodingColName + ) + } + + /** + * Convert surviving decomposition tails into tombstones. + * + * A decomposition tail that survives deletion cleanup is an unmatched delete boundary; setting + * [[recordStartAtFieldName]], [[startAtColName]], and [[endAtColName]] 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) + case c => + F.col(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 +1131,19 @@ object Scd2BatchProcessor { private val finalEndAtColName: String = s"${AutoCdcReservedNames.prefix}final_end_at" + /** + * Names of temporary columns used by [[dropLeftoverDeletesPostReconciliation]] to identify + * delete-encoded rows (tombstones and decomposition tails) already encoded by the immediately + * preceding upsert row's [[endAtColName]]. + * + * Temporary in that the columns have no observable side effect or persistence across + * microbatches. + */ + private[autocdc] val previousEndAtColName: String = + s"${AutoCdcReservedNames.prefix}previous_end_at" + private[autocdc] 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 649126222a24f..91ebcff3f463d 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,330 @@ 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)) + ) + ) + } + + // =============== 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)) + ) + ) + } } From cbf50d94e2aa97582d9c3e24460e88abafe2972d Mon Sep 17 00:00:00 2001 From: Anish Mahto Date: Wed, 10 Jun 2026 18:51:22 +0000 Subject: [PATCH 2/5] cleanup (cherry picked from commit 0d5d7d3b0006849ab2f44f91ecc111ce3fd9cf1d) --- .../autocdc/Scd2BatchProcessor.scala | 8 +-- .../autocdc/Scd2BatchProcessorSuite.scala | 72 +++++++++---------- 2 files changed, 40 insertions(+), 40 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 0a2044596a6f6..a0984d1b389e2 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 @@ -861,8 +861,8 @@ case class Scd2BatchProcessor( * Convert surviving decomposition tails into tombstones. * * A decomposition tail that survives deletion cleanup is an unmatched delete boundary; setting - * [[recordStartAtFieldName]], [[startAtColName]], and [[endAtColName]] to the tail's end - * sequence lets downstream aux handling preserve it as a tombstone. + * [[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)`; @@ -1139,9 +1139,9 @@ object Scd2BatchProcessor { * Temporary in that the columns have no observable side effect or persistence across * microbatches. */ - private[autocdc] val previousEndAtColName: String = + private val previousEndAtColName: String = s"${AutoCdcReservedNames.prefix}previous_end_at" - private[autocdc] val isRedundantDeleteEncodingColName: String = + private val isRedundantDeleteEncodingColName: String = s"${AutoCdcReservedNames.prefix}is_redundant_delete_encoding" /** 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 91ebcff3f463d..fb6d555a94102 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 @@ -2695,8 +2695,8 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { // 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", 10L, 15L, Row(10L)), + Row(1, "bob", 15L, 20L, Row(15L)), Row(1, "alice", null, 20L, Row(null)) ) @@ -2706,7 +2706,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { df = result, expectedAnswer = Seq( Row(1, "alice", 10L, 15L, Row(10L)), - Row(1, "bob", 15L, 20L, Row(15L)) + Row(1, "bob", 15L, 20L, Row(15L)) ) ) } @@ -2719,7 +2719,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { // 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", 5L, 15L, Row(5L)), Row(1, "alice", 20L, 20L, Row(20L)) ) @@ -2728,7 +2728,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { checkAnswer( df = result, expectedAnswer = Seq( - Row(1, "alice", 5L, 15L, Row(5L)), + Row(1, "alice", 5L, 15L, Row(5L)), Row(1, "alice", 20L, 20L, Row(20L)) ) ) @@ -2742,7 +2742,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { // 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", 5L, 12L, Row(5L)), Row(1, "alice", null, 20L, Row(null)) ) @@ -2751,7 +2751,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { checkAnswer( df = result, expectedAnswer = Seq( - Row(1, "alice", 5L, 12L, Row(5L)), + Row(1, "alice", 5L, 12L, Row(5L)), Row(1, "alice", null, 20L, Row(null)) ) ) @@ -2765,8 +2765,8 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { // 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)) + Row(1, "alice", 15L, 15L, Row(15L)), + Row(2, "bob", null, 20L, Row(null)) ) val result = processor.dropLeftoverDeletesPostReconciliation(df) @@ -2774,8 +2774,8 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { checkAnswer( df = result, expectedAnswer = Seq( - Row(1, "alice", 15L, 15L, Row(15L)), - Row(2, "bob", null, 20L, Row(null)) + Row(1, "alice", 15L, 15L, Row(15L)), + Row(2, "bob", null, 20L, Row(null)) ) ) } @@ -2789,8 +2789,8 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { // 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)) + Row(1, "alice", 5L, 20L, Row(5L)), + Row(1, "bob", 10L, 20L, Row(10L)) ) val result = processor.dropLeftoverDeletesPostReconciliation(df) @@ -2798,8 +2798,8 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { checkAnswer( df = result, expectedAnswer = Seq( - Row(1, "alice", 5L, 20L, Row(5L)), - Row(1, "bob", 10L, 20L, Row(10L)) + Row(1, "alice", 5L, 20L, Row(5L)), + Row(1, "bob", 10L, 20L, Row(10L)) ) ) } @@ -2814,7 +2814,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { 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)) + Row(2, "bob", 15L, 15L, Row(15L)) ) val result = processor.dropLeftoverDeletesPostReconciliation(df) @@ -2823,7 +2823,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { df = result, expectedAnswer = Seq( Row(1, "alice", 10L, 15L, Row(10L)), - Row(2, "bob", 15L, 15L, Row(15L)) + Row(2, "bob", 15L, 15L, Row(15L)) ) ) } @@ -2837,9 +2837,9 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { // 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", 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)) ) @@ -2848,8 +2848,8 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { checkAnswer( df = result, expectedAnswer = Seq( - Row(1, "alice", 5L, 10L, Row(5L)), - Row(1, "bob", 15L, 20L, Row(15L)) + Row(1, "alice", 5L, 10L, Row(5L)), + Row(1, "bob", 15L, 20L, Row(15L)) ) ) } @@ -2863,7 +2863,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { // 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", 10L, 20L, Row(10L)), Row(1, "alice", null, 20L, Row(null)) ) @@ -2885,9 +2885,9 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { // 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)) + 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) @@ -2895,9 +2895,9 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { 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)) + Row(1, "alice", 15L, 15L, Row(15L)), + Row(2, "bob", 5L, null, Row(5L)), + Row(3, "carol", 5L, 15L, Row(5L)) ) ) } @@ -2949,7 +2949,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { // 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)) + Row(1, "alice", 5L, 20L, Row(5L)) ) val result = processor.promoteDecompositionTailsToTombstones(df) @@ -2969,10 +2969,10 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { // 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", 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)) + Row(2, "bob", 8L, 30L, Row(8L)), + Row(2, "bob", null, 30L, Row(null)) ) val result = processor.promoteDecompositionTailsToTombstones(df) @@ -2980,10 +2980,10 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { checkAnswer( df = result, expectedAnswer = Seq( - Row(1, "alice", 5L, 20L, Row(5L)), + 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)) + Row(2, "bob", 8L, 30L, Row(8L)), + Row(2, "bob", 30L, 30L, Row(30L)) ) ) } From 205a29fe1920c54cd8809f1ad7f299d4e3a607ab Mon Sep 17 00:00:00 2001 From: andreas-neumann_data Date: Mon, 20 Jul 2026 19:16:43 +0000 Subject: [PATCH 3/5] quote pass-through user columns in promoteDecompositionTailsToTombstones The default pass-through branch used F.col(c), which parses column names containing dots (e.g. `user.name`) as nested-field paths and fails to resolve them. Quote the name with QuotingUtils.quoteIdentifier, matching how the rest of the class handles user column names, and add a test with a dotted column. --- .../autocdc/Scd2BatchProcessor.scala | 2 +- .../autocdc/Scd2BatchProcessorSuite.scala | 26 +++++++++++++++++++ 2 files changed, 27 insertions(+), 1 deletion(-) diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala index a0984d1b389e2..47133515478c4 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 @@ -892,7 +892,7 @@ case class Scd2BatchProcessor( val metadata = reconciledDf.schema(c).metadata F.when(isDecompositionTail, endAt).otherwise(startAt).as(c, metadata) case c => - F.col(c) + F.col(QuotingUtils.quoteIdentifier(c)) } reconciledDf.select(outputColumns.toImmutableArraySeq: _*) 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 fb6d555a94102..7377ac1c72516 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 @@ -2987,4 +2987,30 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { ) ) } + + 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)) + ) + ) + } } From 024285c89f839d15ef553a0d18bc1a8d4387b8c2 Mon Sep 17 00:00:00 2001 From: andreas-neumann_data Date: Wed, 22 Jul 2026 04:17:38 +0000 Subject: [PATCH 4/5] address review: clarify redundancy example and quoting comment - Rewrite the dropLeftoverDeletesPostReconciliation scaladoc so both examples are stated as the same rule (a delete-encoded row is redundant when the immediately preceding reconciled upsert's endAt equals the row's boundary), and spell out where the [null, 20) decomposition tail in the second example comes from (the head + tail split produced when a closed row is bisected). - Add a comment in promoteDecompositionTailsToTombstones explaining that only the default (user-column) case needs quoting, since the other cases match framework columns with known, identifier-safe names. --- .../autocdc/Scd2BatchProcessor.scala | 28 ++++++++++++++----- 1 file changed, 21 insertions(+), 7 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 47133515478c4..56b198770de35 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 @@ -815,14 +815,25 @@ case class Scd2BatchProcessor( * Drop delete-encoded rows (tombstones and decomposition tails) that became redundant after * reconciliation. * - * A tombstone or decomposition tail is redundant when the immediately preceding row's reconciled - * [[endAtColName]] equals its own sequence: the preceding upsert already encodes the delete - * boundary, so the standalone delete-encoded row should no longer be routed to aux. + * 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. * - * For example, an open upsert `[startAt=10, endAt=null)` followed by a tombstone at `15` - * reconciles into a closed upsert `[10, 15)`, making the tombstone redundant. Likewise, an - * existing closed `[10, 20)` bisected by an event at `15` reconciles the event into `[15, 20)`, - * making the `[null, 20)` decomposition tail redundant. + * 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 = { @@ -891,6 +902,9 @@ case class Scd2BatchProcessor( 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)) } From 0803968a9984627c596863939808eabe5aa8dcb9 Mon Sep 17 00:00:00 2001 From: andreas-neumann_data Date: Wed, 22 Jul 2026 18:54:11 +0000 Subject: [PATCH 5/5] address review: assert delete-encoding predecessor is an upsert dropLeftoverDeletesPostReconciliation only checked that the immediately preceding row's endAt equals the delete-encoded row's boundary, but the method's documented rule is that the preceding *upsert* must close on that boundary. Add RowClassifier.isUpsertRepresentingRow(previous) to the redundancy predicate so the method enforces its invariant directly, rather than dropping a delete-encoded row that merely abuts another delete-encoded row on the same boundary. Build the lagged interval via row.lagBy(1, ...) and read previous.endAt from it, which removes the separate previousEndAtColName temp column. Add a test covering two adjacent tombstones on the same boundary: the first is redundant and drops, the second survives because its predecessor is not an upsert. Co-authored-by: Isaac --- .../autocdc/Scd2BatchProcessor.scala | 28 +++++++++---------- .../autocdc/Scd2BatchProcessorSuite.scala | 27 ++++++++++++++++++ 2 files changed, 41 insertions(+), 14 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 56b198770de35..6e5af84216b58 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 @@ -850,22 +850,25 @@ case class Scd2BatchProcessor( 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.previousEndAtColName, - F.lag(endAt, 1).over(orderChronologicallyPerKeyWindow) - ) .withColumn( Scd2BatchProcessor.isRedundantDeleteEncodingColName, - isDeleteEncodedRow && (F.col(Scd2BatchProcessor.previousEndAtColName) <=> endAt) + 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.previousEndAtColName, - Scd2BatchProcessor.isRedundantDeleteEncodingColName - ) + .drop(Scd2BatchProcessor.isRedundantDeleteEncodingColName) } /** @@ -1146,15 +1149,12 @@ object Scd2BatchProcessor { s"${AutoCdcReservedNames.prefix}final_end_at" /** - * Names of temporary columns used by [[dropLeftoverDeletesPostReconciliation]] to identify + * 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 columns have no observable side effect or persistence across - * microbatches. + * Temporary in that the column has no observable side effect or persistence across microbatches. */ - private val previousEndAtColName: String = - s"${AutoCdcReservedNames.prefix}previous_end_at" private val isRedundantDeleteEncodingColName: String = s"${AutoCdcReservedNames.prefix}is_redundant_delete_encoding" 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 7377ac1c72516..0e671721baccc 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 @@ -2854,6 +2854,33 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession { ) } + 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") {