From 9f9bcf093d0fcc3e019197076e993ed4755e1f7a Mon Sep 17 00:00:00 2001 From: Anish Mahto Date: Fri, 12 Jun 2026 17:59:22 +0000 Subject: [PATCH 1/6] implement foreachBatch callback --- .../autocdc/Scd2ForeachBatchHandler.scala | 92 ++ .../Scd2ForeachBatchHandlerSuite.scala | 874 ++++++++++++++++++ 2 files changed, 966 insertions(+) create mode 100644 sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandler.scala create mode 100644 sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandlerSuite.scala diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandler.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandler.scala new file mode 100644 index 0000000000000..a051f2314f3e9 --- /dev/null +++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandler.scala @@ -0,0 +1,92 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.sql.pipelines.autocdc + +import org.apache.spark.sql.catalyst.TableIdentifier +import org.apache.spark.sql.classic.DataFrame + +/** + * Exposes an API to execute one SCD Type 2 AutoCDC microbatch reconciliation on a + * foreachBatch streaming query. + */ +case class Scd2ForeachBatchHandler( + batchProcessor: Scd2BatchProcessor, + auxiliaryTableIdentifier: TableIdentifier, + targetTableIdentifier: TableIdentifier) { + + /** + * Process a single CDC microbatch and merge it into the auxiliary and target tables. + */ + def execute(batchDf: DataFrame, batchId: Long): Unit = { + ScdBatchValidator( + destinationIdentifier = targetTableIdentifier, + changeArgs = batchProcessor.changeArgs, + batchDf = batchDf, + batchId = batchId + ).validateMicrobatch() + + val preprocessedBatchDf = batchProcessor.preprocessMicrobatch(batchDf) + + val perKeyMinimumSequenceInMicrobatchDf = batchProcessor.computeMinimumSequencePerKey( + preprocessedBatchDf + ) + + val auxTableDf = batchDf.sparkSession.read.table(auxiliaryTableIdentifier.quotedString) + val affectedRowsFromAuxiliaryTable = batchProcessor.findAffectedRowsFromAuxiliaryTable( + rawAuxiliaryTableDf = auxTableDf, + perKeyMinimumSequenceInMicrobatchDf = perKeyMinimumSequenceInMicrobatchDf, + batchId = batchId + ) + + val targetTableDf = batchDf.sparkSession.read.table(targetTableIdentifier.quotedString) + val affectedRowsFromTargetTable = batchProcessor.findAffectedRowsFromTargetTable( + targetTableDf = targetTableDf, + perKeyMinimumSequenceInMicrobatchDf = perKeyMinimumSequenceInMicrobatchDf + ) + + val microbatchAndAffectedRows = preprocessedBatchDf + .unionByName(affectedRowsFromAuxiliaryTable) + .unionByName(affectedRowsFromTargetTable) + + val decomposedDf = microbatchAndAffectedRows + .transform(batchProcessor.decomposeOutOfOrderRows) + .transform(batchProcessor.dropRedundantRowsPostDecomposition) + + batchProcessor.assertWellFormedRowsPostDecomposition(decomposedDf, batchId) + + val reconciledDf = decomposedDf + .transform(batchProcessor.reconcileStartAndEndAt) + .transform(batchProcessor.dropLeftoverDeletesPostReconciliation) + .transform(batchProcessor.promoteDecompositionTailsToTombstones) + + val reconciledAndRoutedDf = batchProcessor.identifyAndTagAuxRows(reconciledDf) + + batchProcessor.mergeRowsIntoAuxiliaryTable( + reconciledDfWithAuxRowsTagged = reconciledAndRoutedDf, + originalAffectedRowsFromAuxiliaryTable = affectedRowsFromAuxiliaryTable, + auxiliaryTableIdentifier = auxiliaryTableIdentifier, + batchId = batchId + ) + + batchProcessor.mergeRowsIntoTargetTable( + reconciledDfWithAuxRowsTagged = reconciledAndRoutedDf, + affectedRowsFromTargetTable = affectedRowsFromTargetTable, + targetTableIdentifier = targetTableIdentifier + ) + } +} diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandlerSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandlerSuite.scala new file mode 100644 index 0000000000000..972aecb430f5d --- /dev/null +++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandlerSuite.scala @@ -0,0 +1,874 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.sql.pipelines.autocdc + +import org.scalatest.BeforeAndAfter + +import org.apache.spark.sql.{functions => F, AnalysisException, QueryTest, Row} +import org.apache.spark.sql.classic.DataFrame +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.test.SharedSparkSession +import org.apache.spark.sql.types._ + +/** + * End-to-end unit tests for [[Scd2ForeachBatchHandler]]. Unlike the focused unit suites that + * exercise individual [[Scd2BatchProcessor]] stages in isolation, these drive the entire + * microbatch reconciliation pipeline - validation, preprocessing, affected-row pull-in from both + * side tables, decomposition, start/end reconciliation, aux routing, and the two `MERGE INTO` + * writes - through the public `execute` entrypoint against an in-memory v2 catalog. + * + * They are the first place the cross-microbatch stateful behaviors (out-of-order arrivals, no-op + * run coalescing across batches, tombstone-driven suppression, and auxiliary-table garbage + * collection) can be observed against materialized target and auxiliary tables, so the idempotency + * / GC / cross-batch scenarios are emphasized here. + * + * The default flow tracks every persisted user column (`value`) under key `id`, sequences by + * `seq`, and treats `is_delete = true` rows as deletes. + */ +class Scd2ForeachBatchHandlerSuite + extends QueryTest + with SharedSparkSession + with BeforeAndAfter + with AutoCdcCatalogExecutionTestBase { + + private val sourceSchema = new StructType() + .add("id", IntegerType) + .add("value", StringType) + .add("seq", LongType) + .add("is_delete", BooleanType) + + /** The SCD2 cdc-metadata struct carries a single `recordStartAt` field (unlike SCD1's two). */ + private val scd2MetadataSchema: StructType = Scd2BatchProcessor.cdcMetadataColSchema(LongType) + + /** Canonical SCD2 row schema: persisted user columns + framework start/end + cdc metadata. */ + private val canonicalSchema = new StructType() + .add("id", IntegerType) + .add("value", StringType) + .add(Scd2BatchProcessor.startAtColName, LongType, nullable = true) + .add(Scd2BatchProcessor.endAtColName, LongType, nullable = true) + .add(AutoCdcReservedNames.cdcMetadataColName, scd2MetadataSchema, nullable = false) + + /** Auxiliary table schema: canonical schema plus the aux-only logical-delete marker column. */ + private val auxSchema = canonicalSchema + .add(Scd2BatchProcessor.deletedByBatchIdColName, LongType, nullable = true) + + /** Target table schema is exactly the canonical schema. */ + private val targetSchema = canonicalSchema + + private val processor = Scd2BatchProcessor( + changeArgs = ChangeArgs( + keys = Seq(UnqualifiedColumnName("id")), + sequencing = F.col("seq"), + storedAsScdType = ScdType.Type2, + deleteCondition = Some(F.col("is_delete")), + // Persist only id + value; seq / is_delete are control columns and must not be stored. + columnSelection = Some( + ColumnSelection.ExcludeColumns( + Seq(UnqualifiedColumnName("seq"), UnqualifiedColumnName("is_delete")) + ) + ) + ), + resolvedSequencingType = LongType + ) + + private def createAuxTable(seedRows: Row*): Unit = + createTable(defaultAuxIdent, defaultAuxTableIdentifier, auxSchema, seedRows: _*) + + private def createTargetTable(seedRows: Row*): Unit = + createTable(defaultTargetIdent, defaultTargetTableIdentifier, targetSchema, seedRows: _*) + + private def auxTable: DataFrame = spark.read.table(defaultAuxTableIdentifier.quotedString) + + private def targetTable: DataFrame = spark.read.table(defaultTargetTableIdentifier.quotedString) + + private def execWith(p: Scd2BatchProcessor): Scd2ForeachBatchHandler = Scd2ForeachBatchHandler( + batchProcessor = p, + auxiliaryTableIdentifier = defaultAuxTableIdentifier, + targetTableIdentifier = defaultTargetTableIdentifier + ) + + private def exec: Scd2ForeachBatchHandler = execWith(processor) + + /** A source UPSERT event: `(id, value, seq, is_delete = false)`. */ + private def upsert(id: Int, value: String, seq: Long): Row = Row(id, value, seq, false) + + /** A source DELETE event: `(id, null, seq, is_delete = true)`. */ + private def del(id: Int, seq: Long): Row = Row(id, null, seq, true) + + /** The cdc-metadata struct value for a given `recordStartAt`. */ + private def meta(recordStartAt: Long): Row = Row(recordStartAt) + + /** A canonical target row `(id, value, startAt, endAt, meta(recordStartAt))`. */ + private def targetRow( + id: Int, + value: String, + startAt: java.lang.Long, + endAt: java.lang.Long, + recordStartAt: Long): Row = + Row(id, value, startAt, endAt, meta(recordStartAt)) + + /** A canonical aux row `(id, value, startAt, endAt, meta(recordStartAt), deletedByBatchId)`. */ + private def auxRow( + id: Int, + value: String, + startAt: java.lang.Long, + endAt: java.lang.Long, + recordStartAt: Long, + deletedByBatchId: java.lang.Long): Row = + Row(id, value, startAt, endAt, meta(recordStartAt), deletedByBatchId) + + /** Run a microbatch of source rows through the default handler. */ + private def runBatch(batchId: Long)(rows: Row*): Unit = + exec.execute(microbatchOf(sourceSchema)(rows: _*), batchId) + + /** + * Run `rows` as batch `batchId`, capture both tables, then replay the identical batch under the + * same `batchId` and assert both tables are byte-for-byte unchanged. Models a crash/redelivery + * where a committed microbatch is reprocessed. + */ + private def assertReplayStable(batchId: Long)(rows: Row*): Unit = { + runBatch(batchId)(rows: _*) + val targetAfterFirst = targetTable.collect().toSeq + val auxAfterFirst = auxTable.collect().toSeq + + runBatch(batchId)(rows: _*) + checkAnswer(targetTable, targetAfterFirst) + checkAnswer(auxTable, auxAfterFirst) + } + + test("a record with a null sequencing value fails the microbatch without applying any changes") { + createAuxTable() + createTargetTable(targetRow(1, "old", 10L, null, 10L)) + + val batch = microbatchOf(sourceSchema)(Row(1, "bad", null, false)) + + checkError( + exception = intercept[AnalysisException] { + exec.execute(batch, batchId = 77L) + }, + condition = "AUTOCDC_MICROBATCH_VALIDATION.NULL_SEQUENCE", + sqlState = "22000", + parameters = Map( + "tableName" -> defaultTargetTableIdentifier.quotedString, + "batchId" -> "77", + "nullCount" -> "1" + ) + ) + + assert(auxTable.collect().isEmpty) + checkAnswer(targetTable, targetRow(1, "old", 10L, null, 10L)) + } + + test("inserting a new key creates an open current record") { + createAuxTable() + createTargetTable() + + runBatch(1L)(upsert(1, "a", 10L)) + + // Open interval [10, null); nothing routed to the aux table. + checkAnswer(targetTable, targetRow(1, "a", 10L, null, 10L)) + assert(auxTable.collect().isEmpty) + } + + test("two updates to a key in one batch produce a closed record followed by the open record") { + createAuxTable() + createTargetTable() + + runBatch(1L)(upsert(1, "a", 10L), upsert(1, "b", 20L)) + + // a closes at b's start; b stays open. No hidden rows (every event changed the value). + checkAnswer( + targetTable, + Seq( + targetRow(1, "a", 10L, 20L, 10L), + targetRow(1, "b", 20L, null, 20L) + ) + ) + assert(auxTable.collect().isEmpty) + } + + test("an insert and a later delete in the same batch leave a single closed record") { + createAuxTable() + createTargetTable() + + runBatch(1L)(upsert(1, "a", 10L), del(1, 20L)) + + // The closed interval [10, 20) already encodes the deletion boundary at 20, so the delete's + // tombstone is redundant and dropped during reconciliation - nothing lands in the aux table. + checkAnswer(targetTable, targetRow(1, "a", 10L, 20L, 10L)) + assert(auxTable.collect().isEmpty) + } + + test("an insert, update, delete, and re-insert for one key in a batch build the full history") { + createAuxTable() + createTargetTable() + + // Unlike SCD1 - which would collapse these to the single latest state for the key - SCD2 keeps + // every event: each distinct value gets its own interval, the delete ends the active record, + // and the re-insert opens a fresh record after the deletion gap. + runBatch(1L)( + upsert(1, "a", 10L), + upsert(1, "b", 20L), + del(1, 30L), + upsert(1, "c", 40L) + ) + + // a [10, 20), b [20, 30) (closed by the delete), a deletion gap over [30, 40), then c [40, ..). + // The delete leaves no tombstone: b's closed interval already carries the boundary at 30. + checkAnswer( + targetTable, + Seq( + targetRow(1, "a", 10L, 20L, 10L), + targetRow(1, "b", 20L, 30L, 20L), + targetRow(1, "c", 40L, null, 40L) + ) + ) + assert(auxTable.collect().isEmpty) + } + + test("repeating a key's value keeps one current record effective from its first occurrence") { + createAuxTable() + createTargetTable() + + runBatch(1L)(upsert(1, "a", 10L), upsert(1, "a", 20L)) + + // The run [10, 20] coalesces: the visible tail carries the run-head START_AT (10) but the + // tail's own recordStartAt (20). The head becomes a hidden no-op row in the aux table. + checkAnswer(targetTable, targetRow(1, "a", 10L, null, 20L)) + checkAnswer(auxTable, auxRow(1, "a", 10L, null, 10L, null)) + } + + test("deleting a key that has no current record leaves the dimension table empty") { + createAuxTable() + createTargetTable() + + runBatch(1L)(del(1, 5L)) + + // No preceding upsert closes on the boundary, so the tombstone survives as aux side state. + assert(targetTable.collect().isEmpty) + checkAnswer(auxTable, auxRow(1, null, 5L, 5L, 5L, null)) + } + + test("updating an existing key closes its current record and opens a new one") { + createAuxTable() + createTargetTable(targetRow(1, "a", 10L, null, 10L)) + + runBatch(2L)(upsert(1, "b", 20L)) + + checkAnswer( + targetTable, + Seq( + targetRow(1, "a", 10L, 20L, 10L), + targetRow(1, "b", 20L, null, 20L) + ) + ) + assert(auxTable.collect().isEmpty) + } + + test("deleting an existing key closes its current record with no open record remaining") { + createAuxTable() + createTargetTable(targetRow(1, "a", 10L, null, 10L)) + + runBatch(2L)(del(1, 20L)) + + // The resulting closed interval carries the deletion boundary; no tombstone needed. + checkAnswer(targetTable, targetRow(1, "a", 10L, 20L, 10L)) + assert(auxTable.collect().isEmpty) + } + + test("an update preserves already-closed historical records") { + createAuxTable() + createTargetTable( + targetRow(1, "a", 5L, 10L, 5L), // closed and settled well before the incoming event + targetRow(1, "b", 10L, null, 10L) // currently active + ) + + runBatch(3L)(upsert(1, "c", 20L)) + + // Only the active interval is pulled in and closed; the settled [5, 10) row is never touched. + checkAnswer( + targetTable, + Seq( + targetRow(1, "a", 5L, 10L, 5L), + targetRow(1, "b", 10L, 20L, 10L), + targetRow(1, "c", 20L, null, 20L) + ) + ) + assert(auxTable.collect().isEmpty) + } + + test("a late event older than all existing history is inserted as the earliest record") { + createAuxTable() + createTargetTable(targetRow(1, "a", 10L, null, 10L)) + + // b arrives late with seq=5, strictly before the seeded interval's start. + runBatch(2L)(upsert(1, "b", 5L)) + + checkAnswer( + targetTable, + Seq( + targetRow(1, "b", 5L, 10L, 5L), + targetRow(1, "a", 10L, null, 10L) + ) + ) + assert(auxTable.collect().isEmpty) + } + + test("a late update landing inside an existing record splits it around the new value") { + createAuxTable() + createTargetTable( + targetRow(1, "a", 10L, 20L, 10L), + targetRow(1, "c", 20L, null, 20L) + ) + + // b arrives late at seq=15, inside the closed [10, 20) interval. + runBatch(3L)(upsert(1, "b", 15L)) + + checkAnswer( + targetTable, + Seq( + targetRow(1, "a", 10L, 15L, 10L), + targetRow(1, "b", 15L, 20L, 15L), + targetRow(1, "c", 20L, null, 20L) + ) + ) + assert(auxTable.collect().isEmpty) + } + + test("a late delete landing inside an existing record shortens it to end at the deletion") { + createAuxTable() + createTargetTable( + targetRow(1, "a", 10L, 30L, 10L), + targetRow(1, "b", 30L, null, 30L) + ) + + // Delete arrives late at seq=20, inside the closed [10, 30) interval. + runBatch(4L)(del(1, 20L)) + + // a is decomposed and re-closed at the delete boundary (20); b is unaffected. The delete is + // covered by the new closed interval [10, 20), so it leaves no aux tombstone. + checkAnswer( + targetTable, + Seq( + targetRow(1, "a", 10L, 20L, 10L), + targetRow(1, "b", 30L, null, 30L) + ) + ) + assert(auxTable.collect().isEmpty) + } + + test("re-inserting a key after it was deleted opens a new current record") { + createAuxTable(auxRow(1, null, 20L, 20L, 20L, null)) + createTargetTable() + + // Revival strictly after the recorded deletion at 20. + runBatch(5L)(upsert(1, "x", 30L)) + + // The revival opens a fresh interval; the deletion boundary at 20 stays in the aux table since + // no visible interval closes on it (there is a real gap [20, 30) where the key was absent). + checkAnswer(targetTable, targetRow(1, "x", 30L, null, 30L)) + checkAnswer(auxTable, auxRow(1, null, 20L, 20L, 20L, null)) + } + + test("a value repeated across batches stays one record until a later change closes it") { + createAuxTable() + createTargetTable() + + // Batch 1: establish the run head. + runBatch(1L)(upsert(1, "a", 10L)) + checkAnswer(targetTable, targetRow(1, "a", 10L, null, 10L)) + assert(auxTable.collect().isEmpty) + + // Batch 2: a same-value upsert extends the run. The previously-visible head is demoted to the + // aux table and the new tail becomes the visible row (START_AT pinned to the run head, 10). + runBatch(2L)(upsert(1, "a", 20L)) + checkAnswer(targetTable, targetRow(1, "a", 10L, null, 20L)) + checkAnswer(auxTable, auxRow(1, "a", 10L, null, 10L, null)) + + // Batch 3: a real value change closes the "a" run and opens "b". The hidden head is retained + // as aux side state for any future bisecting event. + runBatch(3L)(upsert(1, "b", 30L)) + checkAnswer( + targetTable, + Seq( + targetRow(1, "a", 10L, 30L, 20L), + targetRow(1, "b", 30L, null, 30L) + ) + ) + checkAnswer(auxTable, auxRow(1, "a", 10L, null, 10L, null)) + } + + test("a late event arriving within an unchanged period splits the surrounding history") { + createAuxTable() + createTargetTable() + + // Build the Alice run [5, 10, 15] then Charlie at 20. + runBatch(1L)(upsert(1, "Alice", 5L), upsert(1, "Alice", 10L), upsert(1, "Alice", 15L)) + runBatch(2L)(upsert(1, "Charlie", 20L)) + + // Alice's run is [5, 20); the visible tail is the latest Alice event (15) with START_AT=5. + checkAnswer( + targetTable, + Seq( + targetRow(1, "Alice", 5L, 20L, 15L), + targetRow(1, "Charlie", 20L, null, 20L) + ) + ) + checkAnswer( + auxTable, + Seq( + auxRow(1, "Alice", 5L, null, 5L, null), + auxRow(1, "Alice", 5L, null, 10L, null) + ) + ) + + // Late Bob at 12 splits the Alice run: Alice [5, 12) (tail now the 10 event), Bob [12, 15), + // Alice [15, 20) (a fresh size-1 run). + runBatch(3L)(upsert(1, "Bob", 12L)) + checkAnswer( + targetTable, + Seq( + targetRow(1, "Alice", 5L, 12L, 10L), + targetRow(1, "Bob", 12L, 15L, 12L), + targetRow(1, "Alice", 15L, 20L, 15L), + targetRow(1, "Charlie", 20L, null, 20L) + ) + ) + // The hidden run head (recordStartAt=5) survives as side state. The other previously-hidden + // no-op (recordStartAt=10) is promoted to the visible tail of [5, 12); it leaves the aux table + // logically (stamped with this batch's id), to be physically garbage-collected by a later + // unrelated batch. + checkAnswer( + auxTable, + Seq( + auxRow(1, "Alice", 5L, null, 5L, null), + auxRow(1, "Alice", 5L, null, 10L, 3L) + ) + ) + } + + test("reprocessing an update microbatch is idempotent") { + createAuxTable() + createTargetTable(targetRow(1, "a", 10L, null, 10L)) + + assertReplayStable(2L)(upsert(1, "b", 20L)) + + checkAnswer( + targetTable, + Seq( + targetRow(1, "a", 10L, 20L, 10L), + targetRow(1, "b", 20L, null, 20L) + ) + ) + assert(auxTable.collect().isEmpty) + } + + test("reprocessing a delete microbatch is idempotent") { + createAuxTable() + createTargetTable(targetRow(1, "a", 10L, null, 10L)) + + assertReplayStable(2L)(del(1, 20L)) + + checkAnswer(targetTable, targetRow(1, "a", 10L, 20L, 10L)) + assert(auxTable.collect().isEmpty) + } + + test("reprocessing a microbatch of repeated values is idempotent") { + createAuxTable() + createTargetTable() + + assertReplayStable(1L)(upsert(1, "a", 10L), upsert(1, "a", 20L), upsert(1, "a", 30L)) + + // A single run of same-value events at sequences 10, 20, 30; the latest (30) is the visible + // tail (open from startAt 10), the earlier two are hidden. + checkAnswer(targetTable, targetRow(1, "a", 10L, null, 30L)) + checkAnswer( + auxTable, + Seq( + auxRow(1, "a", 10L, null, 10L, null), + auxRow(1, "a", 10L, null, 20L, null) + ) + ) + } + + test("reprocessing a delete of an unknown key is idempotent") { + createAuxTable() + createTargetTable() + + assertReplayStable(7L)(del(1, 5L)) + + assert(targetTable.collect().isEmpty) + checkAnswer(auxTable, auxRow(1, null, 5L, 5L, 5L, null)) + } + + test("byte-identical duplicate events in one microbatch collapse to a single record") { + createAuxTable() + createTargetTable() + + // Two fully identical events (same key, value, and sequence). Preprocessing keeps both 1:1; + // because they share a recordStartAt, reconciliation collapses them to one. The result is a + // single open record - and notably no hidden aux row, unlike a run of same-value events at + // *distinct* sequences (where the non-tail members are retained as side state). + runBatch(1L)(upsert(1, "a", 10L), upsert(1, "a", 10L)) + + checkAnswer(targetTable, targetRow(1, "a", 10L, null, 10L)) + assert(auxTable.collect().isEmpty) + } + + test("redelivering the same event in a later microbatch leaves both tables unchanged") { + createAuxTable() + createTargetTable() + + runBatch(1L)(upsert(1, "a", 10L)) + val targetAfterFirst = targetTable.collect().toSeq + val auxAfterFirst = auxTable.collect().toSeq + + // A genuinely new microbatch (different batch id) carries a duplicate of an already-processed + // event - same key and sequence. It collides with the persisted record at recordStartAt=10 and + // is absorbed, so the dimension is identical to having seen the event exactly once. + runBatch(2L)(upsert(1, "a", 10L)) + + checkAnswer(targetTable, targetAfterFirst) + checkAnswer(auxTable, auxAfterFirst) + checkAnswer(targetTable, targetRow(1, "a", 10L, null, 10L)) + assert(auxTable.collect().isEmpty) + } + + test("a late event predating a recorded deletion becomes a record ending at the deletion") { + // Batch 1 records a standalone tombstone for a never-seen key. + createAuxTable() + createTargetTable() + runBatch(1L)(del(1, 20L)) + checkAnswer(auxTable, auxRow(1, null, 20L, 20L, 20L, null)) + assert(targetTable.collect().isEmpty) + + // Batch 2: an upsert strictly before the delete. It materializes as the closed interval + // [10, 20) in the target, which now carries the deletion boundary, so the tombstone is no + // longer needed and is logically deleted (stamped with batch id 2). + runBatch(2L)(upsert(1, "x", 10L)) + checkAnswer(targetTable, targetRow(1, "x", 10L, 20L, 10L)) + checkAnswer(auxTable, auxRow(1, null, 20L, 20L, 20L, 2L)) + } + + test("a logically-deleted tombstone is physically garbage-collected by a later unrelated batch") { + createAuxTable() + createTargetTable() + runBatch(1L)(del(1, 20L)) + + // Batch 2's late upsert reconciles to the closed interval [10, 20), which itself ends exactly + // on the deletion boundary at 20. That closed upsert now encodes the deletion, making the + // standalone tombstone redundant - so the tombstone is logically deleted (stamped deletedBy=2). + // It is not physically removed yet: the marker must survive so a replay of batch 2 reproduces + // the same state, and reconciliation keeps excluding it. Physical removal is deferred. + runBatch(2L)(upsert(1, "x", 10L)) + checkAnswer(auxTable, auxRow(1, null, 20L, 20L, 20L, 2L)) + + // Batch 3 touches a different key. The tombstone is now matched by neither the source nor the + // current batch id (deletedBy=2 != 3), so the aux GC sweep finally hard-deletes it. + runBatch(3L)(upsert(2, "y", 30L)) + + assert(auxTable.collect().isEmpty) + checkAnswer( + targetTable, + Seq( + targetRow(1, "x", 10L, 20L, 10L), + targetRow(2, "y", 30L, null, 30L) + ) + ) + } + + test("updates, deletes, and inserts for different keys in one batch reconcile independently") { + createAuxTable() + createTargetTable( + targetRow(1, "a", 10L, null, 10L), + targetRow(2, "p", 10L, null, 10L) + ) + + // key 1: value change; key 2: delete; key 3: brand new insert - all in one microbatch. + runBatch(4L)(upsert(1, "b", 20L), del(2, 20L), upsert(3, "z", 20L)) + + checkAnswer( + targetTable, + Seq( + targetRow(1, "a", 10L, 20L, 10L), + targetRow(1, "b", 20L, null, 20L), + targetRow(2, "p", 10L, 20L, 10L), + targetRow(3, "z", 20L, null, 20L) + ) + ) + assert(auxTable.collect().isEmpty) + } + + // Source/target carry id + name + score, but only `name` is tracked: a change in the untracked + // `score` alone is a no-op run continuation, while a change in `name` opens a new interval. + private val trackedSourceSchema = new StructType() + .add("id", IntegerType) + .add("name", StringType) + .add("score", IntegerType) + .add("seq", LongType) + .add("is_delete", BooleanType) + + private val trackedCanonicalSchema = new StructType() + .add("id", IntegerType) + .add("name", StringType) + .add("score", IntegerType) + .add(Scd2BatchProcessor.startAtColName, LongType, nullable = true) + .add(Scd2BatchProcessor.endAtColName, LongType, nullable = true) + .add(AutoCdcReservedNames.cdcMetadataColName, scd2MetadataSchema, nullable = false) + + private val trackedAuxSchema = trackedCanonicalSchema + .add(Scd2BatchProcessor.deletedByBatchIdColName, LongType, nullable = true) + + private val trackedProcessor = Scd2BatchProcessor( + changeArgs = ChangeArgs( + keys = Seq(UnqualifiedColumnName("id")), + sequencing = F.col("seq"), + storedAsScdType = ScdType.Type2, + deleteCondition = Some(F.col("is_delete")), + columnSelection = Some( + ColumnSelection.ExcludeColumns( + Seq(UnqualifiedColumnName("seq"), UnqualifiedColumnName("is_delete")) + ) + ), + trackHistorySelection = Some( + ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("name"))) + ) + ), + resolvedSequencingType = LongType + ) + + test("changing only an untracked column updates the current record without adding history") { + createTable(defaultAuxIdent, defaultAuxTableIdentifier, trackedAuxSchema) + createTable(defaultTargetIdent, defaultTargetTableIdentifier, trackedCanonicalSchema) + + execWith(trackedProcessor).execute( + microbatchOf(trackedSourceSchema)( + Row(1, "alice", 100, 10L, false), + Row(1, "alice", 200, 20L, false) // only score changed -> no-op run continuation + ), + batchId = 1L + ) + + // Visible tail reflects the latest values (score=200) with the run-head START_AT (10). + checkAnswer(targetTable, Row(1, "alice", 200, 10L, null, meta(20L))) + checkAnswer(auxTable, Row(1, "alice", 100, 10L, null, meta(10L), null)) + + // A second untracked-only change in a separate microbatch keeps extending the same + // record: it still starts at 10 and now reflects score=300, with no new history opened. + // The previous tail (score=200) joins the original head as hidden side state. + execWith(trackedProcessor).execute( + microbatchOf(trackedSourceSchema)( + Row(1, "alice", 300, 30L, false) // still name=alice, only score changed + ), + batchId = 2L + ) + + checkAnswer(targetTable, Row(1, "alice", 300, 10L, null, meta(30L))) + checkAnswer( + auxTable, + Seq( + Row(1, "alice", 100, 10L, null, meta(10L), null), + Row(1, "alice", 200, 10L, null, meta(20L), null) + ) + ) + } + + test("changing a tracked column opens a new historical record") { + createTable(defaultAuxIdent, defaultAuxTableIdentifier, trackedAuxSchema) + createTable(defaultTargetIdent, defaultTargetTableIdentifier, trackedCanonicalSchema) + + execWith(trackedProcessor).execute( + microbatchOf(trackedSourceSchema)( + Row(1, "alice", 100, 10L, false), + Row(1, "bob", 100, 20L, false) // tracked name changed -> new interval + ), + batchId = 1L + ) + + checkAnswer( + targetTable, + Seq( + Row(1, "alice", 100, 10L, 20L, meta(10L)), + Row(1, "bob", 100, 20L, null, meta(20L)) + ) + ) + assert(auxTable.collect().isEmpty) + + // A second tracked change in a separate microbatch closes the now-current record at the new + // event and opens another, leaving the earlier history (alice) untouched. + execWith(trackedProcessor).execute( + microbatchOf(trackedSourceSchema)( + Row(1, "carol", 100, 30L, false) // tracked name changed again -> another new interval + ), + batchId = 2L + ) + + checkAnswer( + targetTable, + Seq( + Row(1, "alice", 100, 10L, 20L, meta(10L)), + Row(1, "bob", 100, 20L, 30L, meta(20L)), + Row(1, "carol", 100, 30L, null, meta(30L)) + ) + ) + assert(auxTable.collect().isEmpty) + } + + test("multiple new keys in one batch each build their own history independently") { + createAuxTable() + createTargetTable() + + runBatch(1L)( + upsert(1, "a", 10L), + upsert(2, "p", 10L), + upsert(1, "b", 20L), // key 1 changes value + upsert(2, "p", 20L) // key 2 no-op run + ) + + checkAnswer( + targetTable, + Seq( + targetRow(1, "a", 10L, 20L, 10L), + targetRow(1, "b", 20L, null, 20L), + targetRow(2, "p", 10L, null, 20L) + ) + ) + checkAnswer(auxTable, auxRow(2, "p", 10L, null, 10L, null)) + } + + test("a composite key distinguishes rows that share a single key component") { + val compositeSourceSchema = new StructType() + .add("country", StringType) + .add("city", StringType) + .add("population", StringType) + .add("seq", LongType) + .add("is_delete", BooleanType) + val compositeCanonicalSchema = new StructType() + .add("country", StringType) + .add("city", StringType) + .add("population", StringType) + .add(Scd2BatchProcessor.startAtColName, LongType, nullable = true) + .add(Scd2BatchProcessor.endAtColName, LongType, nullable = true) + .add(AutoCdcReservedNames.cdcMetadataColName, scd2MetadataSchema, nullable = false) + val compositeAuxSchema = compositeCanonicalSchema + .add(Scd2BatchProcessor.deletedByBatchIdColName, LongType, nullable = true) + + val compositeProcessor = Scd2BatchProcessor( + changeArgs = ChangeArgs( + keys = Seq(UnqualifiedColumnName("country"), UnqualifiedColumnName("city")), + sequencing = F.col("seq"), + storedAsScdType = ScdType.Type2, + deleteCondition = Some(F.col("is_delete")), + columnSelection = Some( + ColumnSelection.ExcludeColumns( + Seq(UnqualifiedColumnName("seq"), UnqualifiedColumnName("is_delete")) + ) + ) + ), + resolvedSequencingType = LongType + ) + + createTable(defaultAuxIdent, defaultAuxTableIdentifier, compositeAuxSchema) + createTable(defaultTargetIdent, defaultTargetTableIdentifier, compositeCanonicalSchema) + + execWith(compositeProcessor).execute( + microbatchOf(compositeSourceSchema)( + // Same city name, different country: distinct identities, never coalesced. + Row("US", "Springfield", "100", 10L, false), + Row("CA", "Springfield", "200", 10L, false), + Row("US", "Springfield", "150", 20L, false) // updates only the US identity + ), + batchId = 1L + ) + + checkAnswer( + targetTable, + Seq( + Row("US", "Springfield", "100", 10L, 20L, meta(10L)), + Row("US", "Springfield", "150", 20L, null, meta(20L)), + Row("CA", "Springfield", "200", 10L, null, meta(10L)) + ) + ) + assert(auxTable.collect().isEmpty) + } + + test("a key referenced with different casing resolves under case-insensitive analysis") { + withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") { + val caseProcessor = Scd2BatchProcessor( + changeArgs = ChangeArgs( + keys = Seq(UnqualifiedColumnName("ID")), // upper-case reference to lower-case `id` + sequencing = F.col("seq"), + storedAsScdType = ScdType.Type2, + deleteCondition = Some(F.col("is_delete")), + columnSelection = Some( + ColumnSelection.ExcludeColumns( + Seq(UnqualifiedColumnName("seq"), UnqualifiedColumnName("is_delete")) + ) + ) + ), + resolvedSequencingType = LongType + ) + + createAuxTable() + createTargetTable() + + execWith(caseProcessor).execute( + microbatchOf(sourceSchema)(upsert(1, "a", 10L), upsert(1, "b", 20L)), + batchId = 1L + ) + + checkAnswer( + targetTable, + Seq( + targetRow(1, "a", 10L, 20L, 10L), + targetRow(1, "b", 20L, null, 20L) + ) + ) + assert(auxTable.collect().isEmpty) + } + } + + test("a key referenced with non-matching casing fails under case-sensitive analysis") { + withSQLConf(SQLConf.CASE_SENSITIVE.key -> "true") { + val caseProcessor = Scd2BatchProcessor( + changeArgs = ChangeArgs( + keys = Seq(UnqualifiedColumnName("ID")), // does not match lower-case `id` + sequencing = F.col("seq"), + storedAsScdType = ScdType.Type2, + deleteCondition = Some(F.col("is_delete")), + columnSelection = Some( + ColumnSelection.ExcludeColumns( + Seq(UnqualifiedColumnName("seq"), UnqualifiedColumnName("is_delete")) + ) + ) + ), + resolvedSequencingType = LongType + ) + + createAuxTable() + createTargetTable() + + intercept[AnalysisException] { + execWith(caseProcessor).execute( + microbatchOf(sourceSchema)(upsert(1, "a", 10L)), + batchId = 1L + ) + } + } + } +} From ddc95e6cf4ef8d7a2128afa59aa470b0c251ef62 Mon Sep 17 00:00:00 2001 From: Anish Mahto Date: Fri, 12 Jun 2026 18:09:02 +0000 Subject: [PATCH 2/6] self-review --- .../autocdc/Scd2ForeachBatchHandler.scala | 12 +-- .../Scd2ForeachBatchHandlerSuite.scala | 83 ++++++++++++++++++- 2 files changed, 86 insertions(+), 9 deletions(-) diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandler.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandler.scala index a051f2314f3e9..19b6e27fb96a7 100644 --- a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandler.scala +++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandler.scala @@ -31,6 +31,8 @@ case class Scd2ForeachBatchHandler( /** * Process a single CDC microbatch and merge it into the auxiliary and target tables. + * + * Idempotent under same-`batchId` replay. */ def execute(batchDf: DataFrame, batchId: Long): Unit = { ScdBatchValidator( @@ -59,6 +61,8 @@ case class Scd2ForeachBatchHandler( perKeyMinimumSequenceInMicrobatchDf = perKeyMinimumSequenceInMicrobatchDf ) + // All three share the canonical schema; findAffectedRowsFromAuxiliaryTable drops the aux-only + // deletedByBatchId column. val microbatchAndAffectedRows = preprocessedBatchDf .unionByName(affectedRowsFromAuxiliaryTable) .unionByName(affectedRowsFromTargetTable) @@ -66,15 +70,13 @@ case class Scd2ForeachBatchHandler( val decomposedDf = microbatchAndAffectedRows .transform(batchProcessor.decomposeOutOfOrderRows) .transform(batchProcessor.dropRedundantRowsPostDecomposition) - - batchProcessor.assertWellFormedRowsPostDecomposition(decomposedDf, batchId) + .transform(d => batchProcessor.assertWellFormedRowsPostDecomposition(d, batchId)) - val reconciledDf = decomposedDf + val reconciledAndRoutedDf = decomposedDf .transform(batchProcessor.reconcileStartAndEndAt) .transform(batchProcessor.dropLeftoverDeletesPostReconciliation) .transform(batchProcessor.promoteDecompositionTailsToTombstones) - - val reconciledAndRoutedDf = batchProcessor.identifyAndTagAuxRows(reconciledDf) + .transform(batchProcessor.identifyAndTagAuxRows) batchProcessor.mergeRowsIntoAuxiliaryTable( reconciledDfWithAuxRowsTagged = reconciledAndRoutedDf, diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandlerSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandlerSuite.scala index 972aecb430f5d..cd6422c7f9bdc 100644 --- a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandlerSuite.scala +++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandlerSuite.scala @@ -174,6 +174,55 @@ class Scd2ForeachBatchHandlerSuite checkAnswer(targetTable, targetRow(1, "old", 10L, null, 10L)) } + test("a record with a null key fails the microbatch without applying any changes") { + createAuxTable() + createTargetTable(targetRow(1, "old", 10L, null, 10L)) + + val batch = microbatchOf(sourceSchema)(Row(null, "bad", 10L, false)) + + checkError( + exception = intercept[AnalysisException] { + exec.execute(batch, batchId = 7L) + }, + condition = "AUTOCDC_MICROBATCH_VALIDATION.NULL_KEY", + sqlState = "22000", + parameters = Map( + "tableName" -> defaultTargetTableIdentifier.quotedString, + "batchId" -> "7", + "nullKeyCounts" -> "`id`=1" + ) + ) + + assert(auxTable.collect().isEmpty) + checkAnswer(targetTable, targetRow(1, "old", 10L, null, 10L)) + } + + test("an empty microbatch leaves both tables unchanged") { + createAuxTable() + createTargetTable(targetRow(1, "a", 10L, null, 10L)) + + runBatch(2L)() // zero source rows + + checkAnswer(targetTable, targetRow(1, "a", 10L, null, 10L)) + assert(auxTable.collect().isEmpty) + } + + test("an empty microbatch garbage-collects a stale aux row from a prior batch") { + // Batch 1: a late upsert logically deletes a tombstone, stamping deletedByBatchId=1. + createAuxTable() + createTargetTable() + runBatch(1L)(del(1, 20L)) + runBatch(2L)(upsert(1, "x", 10L)) + checkAnswer(auxTable, auxRow(1, null, 20L, 20L, 20L, 2L)) // tombstone stamped, not yet GC'd + + // Batch 3: empty microbatch - no new work, but the GC clause still sweeps the aux table. + // The tombstone (deletedByBatchId=2, not equal to current batchId=3) is physically removed. + runBatch(3L)() + + assert(auxTable.collect().isEmpty) + checkAnswer(targetTable, targetRow(1, "x", 10L, 20L, 10L)) + } + test("inserting a new key creates an open current record") { createAuxTable() createTargetTable() @@ -372,6 +421,32 @@ class Scd2ForeachBatchHandlerSuite assert(auxTable.collect().isEmpty) } + test("two late events in one batch each bisect a distinct closed target row") { + createAuxTable() + createTargetTable( + targetRow(1, "a", 1L, 5L, 1L), + targetRow(1, "b", 5L, 10L, 5L), + targetRow(1, "c", 10L, 20L, 10L), + targetRow(1, "d", 20L, null, 20L) + ) + + // Late x at seq=7 bisects [5,10); late y at seq=15 bisects [10,20) — both in the same batch. + runBatch(5L)(upsert(1, "x", 7L), upsert(1, "y", 15L)) + + checkAnswer( + targetTable, + Seq( + targetRow(1, "a", 1L, 5L, 1L), + targetRow(1, "b", 5L, 7L, 5L), + targetRow(1, "x", 7L, 10L, 7L), + targetRow(1, "c", 10L, 15L, 10L), + targetRow(1, "y", 15L, 20L, 15L), + targetRow(1, "d", 20L, null, 20L) + ) + ) + assert(auxTable.collect().isEmpty) + } + test("re-inserting a key after it was deleted opens a new current record") { createAuxTable(auxRow(1, null, 20L, 20L, 20L, null)) createTargetTable() @@ -462,7 +537,7 @@ class Scd2ForeachBatchHandlerSuite ) } - test("reprocessing an update microbatch is idempotent") { + test("reprocessing an update microbatch leaves both tables unchanged") { createAuxTable() createTargetTable(targetRow(1, "a", 10L, null, 10L)) @@ -478,7 +553,7 @@ class Scd2ForeachBatchHandlerSuite assert(auxTable.collect().isEmpty) } - test("reprocessing a delete microbatch is idempotent") { + test("reprocessing a delete microbatch leaves both tables unchanged") { createAuxTable() createTargetTable(targetRow(1, "a", 10L, null, 10L)) @@ -488,7 +563,7 @@ class Scd2ForeachBatchHandlerSuite assert(auxTable.collect().isEmpty) } - test("reprocessing a microbatch of repeated values is idempotent") { + test("reprocessing a microbatch of repeated values leaves both tables unchanged") { createAuxTable() createTargetTable() @@ -506,7 +581,7 @@ class Scd2ForeachBatchHandlerSuite ) } - test("reprocessing a delete of an unknown key is idempotent") { + test("reprocessing a delete of an unknown key leaves both tables unchanged") { createAuxTable() createTargetTable() From 770394deceb8716f638999a8f8896383438ec601 Mon Sep 17 00:00:00 2001 From: andreas-neumann_data Date: Fri, 24 Jul 2026 06:21:40 +0000 Subject: [PATCH 3/6] [SPARK-57395][SDP] Replace non-ASCII em-dash in Scd2ForeachBatchHandlerSuite comment scalastyle's nonascii.message check failed on an em-dash (U+2014) in a test comment. Replace it with an ASCII '--' to fix the Scala linter. Co-authored-by: Isaac --- .../sql/pipelines/autocdc/Scd2ForeachBatchHandlerSuite.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandlerSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandlerSuite.scala index cd6422c7f9bdc..3d67bf7b9d105 100644 --- a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandlerSuite.scala +++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandlerSuite.scala @@ -430,7 +430,7 @@ class Scd2ForeachBatchHandlerSuite targetRow(1, "d", 20L, null, 20L) ) - // Late x at seq=7 bisects [5,10); late y at seq=15 bisects [10,20) — both in the same batch. + // Late x at seq=7 bisects [5,10); late y at seq=15 bisects [10,20) -- both in the same batch. runBatch(5L)(upsert(1, "x", 7L), upsert(1, "y", 15L)) checkAnswer( From 266417c96eff2172874e8ecb88c1b0d4f46649c0 Mon Sep 17 00:00:00 2001 From: andreas-neumann_data Date: Sat, 25 Jul 2026 01:16:48 +0000 Subject: [PATCH 4/6] [SPARK-57395][SDP] Address review: clarify empty-batch and same-event test coverage Per review feedback on the handler suite: - Reword the column-selection comment to "need not be included" rather than "must not be stored", which was confusing about API guarantees. - Split the empty-microbatch test into an explicit both-tables-empty case (initial processing) and a both-tables-non-empty case (live target + live aux row survive untouched, aux not GC'd), since empty batches are expected in general. - Pin down the "same event" collapse condition, which is (key, recordStartAt), not byte-identity: rename the misleading "byte-identical" test and add within-batch and across-batch cases where two events share key and sequence but differ in value and still collapse/absorb to a single record. Same-sequence collisions are undefined per the uniqueness contract, so these assert the structural outcome (one open record, no aux row, value is one of the inputs) rather than which value wins. Co-authored-by: Opus 4.8 --- .../Scd2ForeachBatchHandlerSuite.scala | 74 +++++++++++++++++-- 1 file changed, 66 insertions(+), 8 deletions(-) diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandlerSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandlerSuite.scala index 3d67bf7b9d105..4e145c76630dd 100644 --- a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandlerSuite.scala +++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandlerSuite.scala @@ -76,7 +76,7 @@ class Scd2ForeachBatchHandlerSuite sequencing = F.col("seq"), storedAsScdType = ScdType.Type2, deleteCondition = Some(F.col("is_delete")), - // Persist only id + value; seq / is_delete are control columns and must not be stored. + // Persist only id + value; seq / is_delete are control columns that need not be included. columnSelection = Some( ColumnSelection.ExcludeColumns( Seq(UnqualifiedColumnName("seq"), UnqualifiedColumnName("is_delete")) @@ -197,14 +197,29 @@ class Scd2ForeachBatchHandlerSuite checkAnswer(targetTable, targetRow(1, "old", 10L, null, 10L)) } - test("an empty microbatch leaves both tables unchanged") { + test("an empty microbatch with both tables empty leaves both empty (initial processing)") { + // The first batch of a stream may be empty before any data arrives; nothing should be + // written to either table. createAuxTable() + createTargetTable() + + runBatch(1L)() // zero source rows + + assert(targetTable.collect().isEmpty) + assert(auxTable.collect().isEmpty) + } + + test("an empty microbatch with both tables non-empty leaves both unchanged") { + // A live target row plus a live (non-deletable) aux row must both survive an empty batch + // untouched: no spurious writes, and the aux row is not GC'd (it was not deleted by a prior + // batch, so its deletedByBatchId is null). + createAuxTable(auxRow(1, "hidden", 5L, null, 5L, null)) createTargetTable(targetRow(1, "a", 10L, null, 10L)) runBatch(2L)() // zero source rows checkAnswer(targetTable, targetRow(1, "a", 10L, null, 10L)) - assert(auxTable.collect().isEmpty) + checkAnswer(auxTable, auxRow(1, "hidden", 5L, null, 5L, null)) } test("an empty microbatch garbage-collects a stale aux row from a prior batch") { @@ -591,20 +606,41 @@ class Scd2ForeachBatchHandlerSuite checkAnswer(auxTable, auxRow(1, null, 5L, 5L, 5L, null)) } - test("byte-identical duplicate events in one microbatch collapse to a single record") { + test("duplicate events at the same key and sequence in one microbatch collapse to one record") { createAuxTable() createTargetTable() - // Two fully identical events (same key, value, and sequence). Preprocessing keeps both 1:1; - // because they share a recordStartAt, reconciliation collapses them to one. The result is a - // single open record - and notably no hidden aux row, unlike a run of same-value events at - // *distinct* sequences (where the non-tail members are retained as side state). + // Two fully identical events (same key, value, and sequence). The collapse condition is + // (key, recordStartAt), so they merge to a single open record - and notably no hidden aux + // row, unlike a run of same-value events at *distinct* sequences (where the non-tail members + // are retained as side state). runBatch(1L)(upsert(1, "a", 10L), upsert(1, "a", 10L)) checkAnswer(targetTable, targetRow(1, "a", 10L, null, 10L)) assert(auxTable.collect().isEmpty) } + test("events sharing key and sequence but differing in value still collapse to one record") { + createAuxTable() + createTargetTable() + + // The "same event" condition is (key, recordStartAt), NOT byte-identity: two events at the + // same key and sequence collapse even when their values differ. Emitting two events at the + // same sequence for a key violates the documented uniqueness contract, so *which* value wins + // is undefined; we assert only the well-defined structural outcome -- exactly one open record + // for the key, with no hidden aux row -- and that the surviving value is one of the inputs. + runBatch(1L)(upsert(1, "a", 10L), upsert(1, "b", 10L)) + + val rows = targetTable.collect() + assert(rows.length == 1, s"expected a single collapsed record, got ${rows.toSeq}") + val row = rows.head + assert(row.getAs[Int]("id") == 1) + assert(Set("a", "b").contains(row.getAs[String]("value"))) + assert(row.getAs[Long](Scd2BatchProcessor.startAtColName) == 10L) + assert(row.isNullAt(row.fieldIndex(Scd2BatchProcessor.endAtColName)), "record should be open") + assert(auxTable.collect().isEmpty) + } + test("redelivering the same event in a later microbatch leaves both tables unchanged") { createAuxTable() createTargetTable() @@ -624,6 +660,28 @@ class Scd2ForeachBatchHandlerSuite assert(auxTable.collect().isEmpty) } + test("a later microbatch event sharing key and sequence but differing in value is absorbed") { + // Confirms the across-batch "same event" condition is the same (key, recordStartAt) match used + // within a batch: a later event at the already-persisted key and sequence collides with the + // stored record and is absorbed, leaving a single open record rather than adding history. + // (Same-sequence collisions are undefined per the uniqueness contract, so we assert the + // structural outcome and that the value is one of the two inputs, not which one wins.) + createAuxTable() + createTargetTable() + + runBatch(1L)(upsert(1, "a", 10L)) + runBatch(2L)(upsert(1, "b", 10L)) + + val rows = targetTable.collect() + assert(rows.length == 1, s"expected a single absorbed record, got ${rows.toSeq}") + val row = rows.head + assert(row.getAs[Int]("id") == 1) + assert(Set("a", "b").contains(row.getAs[String]("value"))) + assert(row.getAs[Long](Scd2BatchProcessor.startAtColName) == 10L) + assert(row.isNullAt(row.fieldIndex(Scd2BatchProcessor.endAtColName)), "record should be open") + assert(auxTable.collect().isEmpty) + } + test("a late event predating a recorded deletion becomes a record ending at the deletion") { // Batch 1 records a standalone tombstone for a never-seen key. createAuxTable() From be009175f1097c4c9e0a676a4a1d2a1d5a6d12a6 Mon Sep 17 00:00:00 2001 From: andreas-neumann_data Date: Sat, 25 Jul 2026 03:11:42 +0000 Subject: [PATCH 5/6] [SPARK-57395][SDP] Address review: assert-before-drop ordering, crash-recovery coverage, comment fix - Run assertWellFormedRowsPostDecomposition before dropRedundantRowsPostDecomposition in the handler. Both transforms document their input as the output of decomposeOutOfOrderRows, so asserting first honors both contracts; it also guards dropRedundantRowsPostDecomposition, whose effectiveRecordStartAt fallback assumes decomposition tails are the only rows with a null recordStartAt -- an ill-formed row that ties with a neighbour was previously dropped as redundant before the assertion could catch it. - Add direct coverage for the mid-batch crash idempotency path. assertReplayStable only replays batches that completed both merges, but deletedByBatchId exists for a failure *between* the aux and target merges. A new runBatchAuxMergeOnly helper drives that halfway state, and two tests (tombstone and demotion) assert that rerunning the same batchId converges to a clean single run. - Fix a stale comment: the garbage-collected tombstone is stamped deletedByBatchId=2 (batch 2's upsert), not batch 1. Co-authored-by: Opus 4.8 --- .../autocdc/Scd2ForeachBatchHandler.scala | 7 +- .../Scd2ForeachBatchHandlerSuite.scala | 82 ++++++++++++++++++- 2 files changed, 87 insertions(+), 2 deletions(-) diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandler.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandler.scala index 19b6e27fb96a7..444a00a15e940 100644 --- a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandler.scala +++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandler.scala @@ -69,8 +69,13 @@ case class Scd2ForeachBatchHandler( val decomposedDf = microbatchAndAffectedRows .transform(batchProcessor.decomposeOutOfOrderRows) - .transform(batchProcessor.dropRedundantRowsPostDecomposition) + // Assert well-formedness before dropping redundant rows: both transforms document their + // input as the output of decomposeOutOfOrderRows, and dropRedundantRowsPostDecomposition + // assumes decomposition tails are the only rows with a null recordStartAt. Checking first + // both honors that contract and prevents an ill-formed row from being silently dropped as + // redundant (when it ties with a neighbour) before the assertion can catch it. .transform(d => batchProcessor.assertWellFormedRowsPostDecomposition(d, batchId)) + .transform(batchProcessor.dropRedundantRowsPostDecomposition) val reconciledAndRoutedDf = decomposedDf .transform(batchProcessor.reconcileStartAndEndAt) diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandlerSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandlerSuite.scala index 4e145c76630dd..e7024c960e409 100644 --- a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandlerSuite.scala +++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandlerSuite.scala @@ -151,6 +151,55 @@ class Scd2ForeachBatchHandlerSuite checkAnswer(auxTable, auxAfterFirst) } + /** + * Replicate [[Scd2ForeachBatchHandler.execute]] but stop after the auxiliary-table merge, + * skipping the target-table merge. Models a crash *between* the two merges: the aux table has + * committed this `batchId`'s logical deletes / inserts, but the target table has not yet been + * updated. On recovery Structured Streaming reruns the same `batchId`, which + * [[Scd2BatchProcessor.deletedByBatchIdColName]] is designed to make idempotent. + */ + private def runBatchAuxMergeOnly(batchId: Long)(rows: Row*): Unit = { + val batchDf = microbatchOf(sourceSchema)(rows: _*) + ScdBatchValidator( + destinationIdentifier = defaultTargetTableIdentifier, + changeArgs = processor.changeArgs, + batchDf = batchDf, + batchId = batchId + ).validateMicrobatch() + + val preprocessed = processor.preprocessMicrobatch(batchDf) + val perKeyMin = processor.computeMinimumSequencePerKey(preprocessed) + + val affectedAux = processor.findAffectedRowsFromAuxiliaryTable( + rawAuxiliaryTableDf = auxTable, + perKeyMinimumSequenceInMicrobatchDf = perKeyMin, + batchId = batchId + ) + val affectedTarget = processor.findAffectedRowsFromTargetTable( + targetTableDf = targetTable, + perKeyMinimumSequenceInMicrobatchDf = perKeyMin + ) + + val reconciledAndRouted = preprocessed + .unionByName(affectedAux) + .unionByName(affectedTarget) + .transform(processor.decomposeOutOfOrderRows) + .transform(d => processor.assertWellFormedRowsPostDecomposition(d, batchId)) + .transform(processor.dropRedundantRowsPostDecomposition) + .transform(processor.reconcileStartAndEndAt) + .transform(processor.dropLeftoverDeletesPostReconciliation) + .transform(processor.promoteDecompositionTailsToTombstones) + .transform(processor.identifyAndTagAuxRows) + + // Only the aux merge runs; the target merge is skipped to model the mid-batch crash. + processor.mergeRowsIntoAuxiliaryTable( + reconciledDfWithAuxRowsTagged = reconciledAndRouted, + originalAffectedRowsFromAuxiliaryTable = affectedAux, + auxiliaryTableIdentifier = defaultAuxTableIdentifier, + batchId = batchId + ) + } + test("a record with a null sequencing value fails the microbatch without applying any changes") { createAuxTable() createTargetTable(targetRow(1, "old", 10L, null, 10L)) @@ -223,7 +272,8 @@ class Scd2ForeachBatchHandlerSuite } test("an empty microbatch garbage-collects a stale aux row from a prior batch") { - // Batch 1: a late upsert logically deletes a tombstone, stamping deletedByBatchId=1. + // Batches 1-2: a delete records a tombstone, then a late upsert logically deletes it, + // stamping deletedByBatchId=2. createAuxTable() createTargetTable() runBatch(1L)(del(1, 20L)) @@ -606,6 +656,36 @@ class Scd2ForeachBatchHandlerSuite checkAnswer(auxTable, auxRow(1, null, 5L, 5L, 5L, null)) } + test("recovering after a crash between the aux and target merges converges (tombstone)") { + // A standalone delete records a tombstone in the aux table. Simulate a crash right after the + // aux merge commits but before the target merge, then rerun the same batchId end to end. + // deletedByBatchId keeps the batch's aux writes visible to the replay so it re-derives the + // same output, and the result must match a clean single run. + createAuxTable() + createTargetTable() + + runBatchAuxMergeOnly(1L)(del(1, 5L)) // crash: aux merged, target not + runBatch(1L)(del(1, 5L)) // recovery: same batchId reruns fully + + assert(targetTable.collect().isEmpty) + checkAnswer(auxTable, auxRow(1, null, 5L, 5L, 5L, null)) + } + + test("recovering after a crash between the aux and target merges converges (demotion)") { + // A run of same-value events coalesces: the visible tail lands in the target and the run head + // is demoted to a hidden no-op row in the aux table. Simulate a crash after the aux merge but + // before the target merge, then rerun the same batchId; the recovered state must match a clean + // single run (visible tail in target, hidden head in aux). + createAuxTable() + createTargetTable() + + runBatchAuxMergeOnly(1L)(upsert(1, "a", 10L), upsert(1, "a", 20L)) // crash: aux merged only + runBatch(1L)(upsert(1, "a", 10L), upsert(1, "a", 20L)) // recovery: same batchId reruns fully + + checkAnswer(targetTable, targetRow(1, "a", 10L, null, 20L)) + checkAnswer(auxTable, auxRow(1, "a", 10L, null, 10L, null)) + } + test("duplicate events at the same key and sequence in one microbatch collapse to one record") { createAuxTable() createTargetTable() From 73c0a7bf8bdc8bcbae2ebb19d95a49cd5a012f6e Mon Sep 17 00:00:00 2001 From: andreas-neumann_data Date: Mon, 27 Jul 2026 19:43:04 +0000 Subject: [PATCH 6/6] [SPARK-57395][SDP] Address review: shared reconcile chain + deeper coverage Address szehon-ho's follow-up review comments: - Extract the reconciliation chain (validate -> preprocess -> find-affected -> union -> decompose -> reconcile -> route) out of Scd2ForeachBatchHandler.execute into a private[autocdc] reconcileMicrobatch returning an Scd2ReconciliationResult. Both execute and the crash-recovery test helper now call it, so the helper can no longer silently desynchronize from the real pipeline as transforms are reordered. - Add a crash-recovery test that actually exercises the deletedByBatchId re-inclusion clause: the crashed attempt logically deletes a pre-existing aux tombstone (stamping deletedByBatchId = batchId) rather than only inserting fresh rows, and the replay must re-observe it to preserve the deletion boundary. Also assert the modeled halfway aux state in the existing crash tests. - Add a decomposition-tail promotion test: a late delete bisecting a trailing closed record (no successor) leaves a surviving tail that promoteDecompositionTailsToTombstones promotes to an aux tombstone -- a path no other test in this suite reached. Paired with an assertReplayStable variant. - Add an upsert-vs-delete same-sequence tie-break test: unlike the undefined upsert/upsert case, this tie-break is defined (orderUpsertRepresentingRowsFirst), so assert exactly that the delete wins and survives as a tombstone. Co-authored-by: Opus 4.8 --- .../autocdc/Scd2ForeachBatchHandler.scala | 55 +++++++-- .../Scd2ForeachBatchHandlerSuite.scala | 113 ++++++++++++------ 2 files changed, 122 insertions(+), 46 deletions(-) diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandler.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandler.scala index 444a00a15e940..3e59fa09d25a5 100644 --- a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandler.scala +++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandler.scala @@ -35,6 +35,32 @@ case class Scd2ForeachBatchHandler( * Idempotent under same-`batchId` replay. */ def execute(batchDf: DataFrame, batchId: Long): Unit = { + val reconciled = reconcileMicrobatch(batchDf, batchId) + + batchProcessor.mergeRowsIntoAuxiliaryTable( + reconciledDfWithAuxRowsTagged = reconciled.reconciledAndRoutedDf, + originalAffectedRowsFromAuxiliaryTable = reconciled.affectedRowsFromAuxiliaryTable, + auxiliaryTableIdentifier = auxiliaryTableIdentifier, + batchId = batchId + ) + + batchProcessor.mergeRowsIntoTargetTable( + reconciledDfWithAuxRowsTagged = reconciled.reconciledAndRoutedDf, + affectedRowsFromTargetTable = reconciled.affectedRowsFromTargetTable, + targetTableIdentifier = targetTableIdentifier + ) + } + + /** + * Validate and reconcile a single CDC microbatch against the current auxiliary- and target-table + * state, producing the tagged post-reconciliation rows plus the affected-row sets the two merges + * consume. Performs no writes: this is the entire pipeline up to (but excluding) the aux/target + * merges, factored out of [[execute]] so that both [[execute]] and tests exercise the exact same + * transform chain (they cannot silently desynchronize). + */ + private[autocdc] def reconcileMicrobatch( + batchDf: DataFrame, + batchId: Long): Scd2ReconciliationResult = { ScdBatchValidator( destinationIdentifier = targetTableIdentifier, changeArgs = batchProcessor.changeArgs, @@ -83,17 +109,24 @@ case class Scd2ForeachBatchHandler( .transform(batchProcessor.promoteDecompositionTailsToTombstones) .transform(batchProcessor.identifyAndTagAuxRows) - batchProcessor.mergeRowsIntoAuxiliaryTable( - reconciledDfWithAuxRowsTagged = reconciledAndRoutedDf, - originalAffectedRowsFromAuxiliaryTable = affectedRowsFromAuxiliaryTable, - auxiliaryTableIdentifier = auxiliaryTableIdentifier, - batchId = batchId - ) - - batchProcessor.mergeRowsIntoTargetTable( - reconciledDfWithAuxRowsTagged = reconciledAndRoutedDf, - affectedRowsFromTargetTable = affectedRowsFromTargetTable, - targetTableIdentifier = targetTableIdentifier + Scd2ReconciliationResult( + reconciledAndRoutedDf = reconciledAndRoutedDf, + affectedRowsFromAuxiliaryTable = affectedRowsFromAuxiliaryTable, + affectedRowsFromTargetTable = affectedRowsFromTargetTable ) } } + +/** + * The products of [[Scd2ForeachBatchHandler.reconcileMicrobatch]] that the auxiliary- and + * target-table merges consume. + * + * @param reconciledAndRoutedDf the post-reconciliation rows tagged with aux-vs-target routing. + * @param affectedRowsFromAuxiliaryTable the aux rows pulled in for this microbatch (canonical SCD2 + * row schema, aux-only deletedByBatchId dropped). + * @param affectedRowsFromTargetTable the target rows pulled in for this microbatch. + */ +private[autocdc] case class Scd2ReconciliationResult( + reconciledAndRoutedDf: DataFrame, + affectedRowsFromAuxiliaryTable: DataFrame, + affectedRowsFromTargetTable: DataFrame) diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandlerSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandlerSuite.scala index e7024c960e409..287447cc06642 100644 --- a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandlerSuite.scala +++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandlerSuite.scala @@ -159,42 +159,12 @@ class Scd2ForeachBatchHandlerSuite * [[Scd2BatchProcessor.deletedByBatchIdColName]] is designed to make idempotent. */ private def runBatchAuxMergeOnly(batchId: Long)(rows: Row*): Unit = { - val batchDf = microbatchOf(sourceSchema)(rows: _*) - ScdBatchValidator( - destinationIdentifier = defaultTargetTableIdentifier, - changeArgs = processor.changeArgs, - batchDf = batchDf, - batchId = batchId - ).validateMicrobatch() - - val preprocessed = processor.preprocessMicrobatch(batchDf) - val perKeyMin = processor.computeMinimumSequencePerKey(preprocessed) - - val affectedAux = processor.findAffectedRowsFromAuxiliaryTable( - rawAuxiliaryTableDf = auxTable, - perKeyMinimumSequenceInMicrobatchDf = perKeyMin, - batchId = batchId - ) - val affectedTarget = processor.findAffectedRowsFromTargetTable( - targetTableDf = targetTable, - perKeyMinimumSequenceInMicrobatchDf = perKeyMin - ) - - val reconciledAndRouted = preprocessed - .unionByName(affectedAux) - .unionByName(affectedTarget) - .transform(processor.decomposeOutOfOrderRows) - .transform(d => processor.assertWellFormedRowsPostDecomposition(d, batchId)) - .transform(processor.dropRedundantRowsPostDecomposition) - .transform(processor.reconcileStartAndEndAt) - .transform(processor.dropLeftoverDeletesPostReconciliation) - .transform(processor.promoteDecompositionTailsToTombstones) - .transform(processor.identifyAndTagAuxRows) - - // Only the aux merge runs; the target merge is skipped to model the mid-batch crash. + // Reuse the handler's own reconciliation chain so this helper cannot drift from execute(), + // then run only the aux merge (skipping the target merge) to model the mid-batch crash. + val reconciled = exec.reconcileMicrobatch(microbatchOf(sourceSchema)(rows: _*), batchId) processor.mergeRowsIntoAuxiliaryTable( - reconciledDfWithAuxRowsTagged = reconciledAndRouted, - originalAffectedRowsFromAuxiliaryTable = affectedAux, + reconciledDfWithAuxRowsTagged = reconciled.reconciledAndRoutedDf, + originalAffectedRowsFromAuxiliaryTable = reconciled.affectedRowsFromAuxiliaryTable, auxiliaryTableIdentifier = defaultAuxTableIdentifier, batchId = batchId ) @@ -486,6 +456,34 @@ class Scd2ForeachBatchHandlerSuite assert(auxTable.collect().isEmpty) } + test("a late delete bisecting a trailing closed record promotes the surviving tail to a " + + "tombstone") { + // Unlike the bisection cases above, the bisected record is the LAST record for the key -- there + // is no successor event tying with the decomposition tail at the original boundary. So the tail + // is not dropped as redundant; it survives reconciliation and is promoted to an aux tombstone + // by promoteDecompositionTailsToTombstones (a transform the other tests never actually reach). + createAuxTable() + createTargetTable(targetRow(1, "a", 10L, 20L, 10L)) // closed, nothing after it + + runBatch(2L)(del(1, 15L)) + + // `a` is shortened to [10, 15); the original boundary at 20 survives as a promoted tombstone. + checkAnswer(targetTable, targetRow(1, "a", 10L, 15L, 10L)) + checkAnswer(auxTable, auxRow(1, "a", 20L, 20L, 20L, null)) + } + + test("a late delete bisecting a trailing closed record is replay-stable") { + // Same scenario as above, exercised through a same-batchId replay (no replay test elsewhere + // covers a decomposition/promotion path). + createAuxTable() + createTargetTable(targetRow(1, "a", 10L, 20L, 10L)) + + assertReplayStable(2L)(del(1, 15L)) + + checkAnswer(targetTable, targetRow(1, "a", 10L, 15L, 10L)) + checkAnswer(auxTable, auxRow(1, "a", 20L, 20L, 20L, null)) + } + test("two late events in one batch each bisect a distinct closed target row") { createAuxTable() createTargetTable( @@ -665,6 +663,9 @@ class Scd2ForeachBatchHandlerSuite createTargetTable() runBatchAuxMergeOnly(1L)(del(1, 5L)) // crash: aux merged, target not + // Halfway state: the tombstone is a fresh insert, so it is alive (deletedByBatchId = null). + checkAnswer(auxTable, auxRow(1, null, 5L, 5L, 5L, null)) + runBatch(1L)(del(1, 5L)) // recovery: same batchId reruns fully assert(targetTable.collect().isEmpty) @@ -680,12 +681,40 @@ class Scd2ForeachBatchHandlerSuite createTargetTable() runBatchAuxMergeOnly(1L)(upsert(1, "a", 10L), upsert(1, "a", 20L)) // crash: aux merged only + // Halfway state: the demoted run head is a fresh insert, alive (deletedByBatchId = null). + checkAnswer(auxTable, auxRow(1, "a", 10L, null, 10L, null)) + runBatch(1L)(upsert(1, "a", 10L), upsert(1, "a", 20L)) // recovery: same batchId reruns fully checkAnswer(targetTable, targetRow(1, "a", 10L, null, 20L)) checkAnswer(auxTable, auxRow(1, "a", 10L, null, 10L, null)) } + test("recovering after a crash that logically deleted a pre-existing aux row converges") { + // This is the case the deletedByBatchId re-inclusion clause exists for: unlike the crash tests + // above (fresh inserts, deletedByBatchId = null), here the crashed attempt logically DELETES a + // pre-existing aux row, stamping it with the batch id. The replay must re-observe that row via + // the `deletedByBatchId === batchId` clause in findAffectedRowsFromAuxiliaryTable; otherwise it + // sees an empty affected-aux set, loses the deletion boundary, and writes an open record. + createAuxTable() + createTargetTable() + + runBatch(1L)(del(1, 20L)) // batch 1: a live tombstone [20, 20) in the aux table + checkAnswer(auxTable, auxRow(1, null, 20L, 20L, 20L, null)) + + // Crash during batch 2: the late upsert promotes the tombstone out of the aux table, stamping + // it deletedByBatchId = 2, but the target merge never runs. + runBatchAuxMergeOnly(2L)(upsert(1, "x", 10L)) + checkAnswer(auxTable, auxRow(1, null, 20L, 20L, 20L, 2L)) + + // Replay of batch 2: re-observes the stamped-by-2 aux row, re-derives the same output. + runBatch(2L)(upsert(1, "x", 10L)) + + // The upsert lands as a record closed at the recorded deletion boundary (20), not left open. + checkAnswer(targetTable, targetRow(1, "x", 10L, 20L, 10L)) + checkAnswer(auxTable, auxRow(1, null, 20L, 20L, 20L, 2L)) + } + test("duplicate events at the same key and sequence in one microbatch collapse to one record") { createAuxTable() createTargetTable() @@ -721,6 +750,20 @@ class Scd2ForeachBatchHandlerSuite assert(auxTable.collect().isEmpty) } + test("an upsert and a delete at the same sequence resolve in favor of the delete") { + createAuxTable() + createTargetTable() + + // Unlike the upsert/upsert same-sequence case (undefined winner), the upsert-vs-delete + // tie-break at the same sequence is DEFINED: orderUpsertRepresentingRowsFirst sorts the upsert + // ahead so it detects the coincident delete via LEAD(1), and the delete wins -- surviving as a + // tombstone in the aux table with nothing left visible in the target. + runBatch(1L)(upsert(1, "a", 10L), del(1, 10L)) + + assert(targetTable.collect().isEmpty) + checkAnswer(auxTable, auxRow(1, null, 10L, 10L, 10L, null)) + } + test("redelivering the same event in a later microbatch leaves both tables unchanged") { createAuxTable() createTargetTable()