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..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 @@ -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 649126222a24f..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 @@ -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)) + ) + ) + } }