anew commented on code in PR #57495: URL: https://github.com/apache/spark/pull/57495#discussion_r3660845001
########## sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandlerSuite.scala: ########## @@ -0,0 +1,1087 @@ +/* + * 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 that need not be included. + 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) + } + + /** + * 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)) + + 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("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 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)) + checkAnswer(auxTable, auxRow(1, "hidden", 5L, null, 5L, null)) + } + + test("an empty microbatch garbage-collects a stale aux row from a prior batch") { + // 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)) + 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() + + 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") { Review Comment: Good point, added as anew test case. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
