szehon-ho commented on code in PR #57495: URL: https://github.com/apache/spark/pull/57495#discussion_r3649119307
########## sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandler.scala: ########## @@ -0,0 +1,94 @@ +/* + * 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. + * + * Idempotent under same-`batchId` replay. + */ + 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 + ) + + // All three share the canonical schema; findAffectedRowsFromAuxiliaryTable drops the aux-only + // deletedByBatchId column. + val microbatchAndAffectedRows = preprocessedBatchDf + .unionByName(affectedRowsFromAuxiliaryTable) + .unionByName(affectedRowsFromTargetTable) + + val decomposedDf = microbatchAndAffectedRows + .transform(batchProcessor.decomposeOutOfOrderRows) + .transform(batchProcessor.dropRedundantRowsPostDecomposition) + .transform(d => batchProcessor.assertWellFormedRowsPostDecomposition(d, batchId)) Review Comment: nit: `assertWellFormedRowsPostDecomposition` and `dropRedundantRowsPostDecomposition` both document their input as "the output of `decomposeOutOfOrderRows`", so only one of them can literally receive it. Asserting first satisfies both contracts and also guards `dropRedundantRowsPostDecomposition` itself, whose `effectiveRecordStartAt` fallback assumes decomposition tails are the only rows with a null `__RECORD_START_AT`. As written, an ill-formed row that happens to tie with a neighbour is dropped as redundant and never reaches the assertion. ########## sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandlerSuite.scala: ########## @@ -0,0 +1,949 @@ +/* + * 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 = { Review Comment: `assertReplayStable` only ever replays a batch that completed both merges. The `deletedByBatchId` mechanism exists specifically for a failure *between* the two merges (see the scaladoc on `deletedByBatchIdColName`), and that path has no direct coverage here: every test asserts the state after both merges have run. Since this suite is in the same package as the `private[autocdc]` methods, it can drive the halfway state directly: run the reconciliation and call `mergeRowsIntoAuxiliaryTable` on its own, then call `execute` with the same `batchId` and assert both tables converge to what a clean single run produces. Worth covering at least a demotion case and a tombstone case. ########## sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandlerSuite.scala: ########## @@ -0,0 +1,949 @@ +/* + * 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("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. Review Comment: nit: the stamp comes from batch 2's upsert, not batch 1 - the assertion below expects `deletedByBatchId = 2`. Maybe "Batches 1-2: a delete records a tombstone, then a late upsert logically deletes it, stamping deletedByBatchId=2." -- 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]
