szehon-ho commented on code in PR #58055: URL: https://github.com/apache/spark/pull/58055#discussion_r3848940419
########## sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcCrossScdConvergenceSuite.scala: ########## @@ -0,0 +1,105 @@ +/* + * 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.graph + +import scala.util.Random + +import org.apache.spark.sql.functions +import org.apache.spark.sql.pipelines.autocdc.{Scd2BatchProcessor, ScdType} +import org.apache.spark.sql.pipelines.utils.ExecutionTest +import org.apache.spark.sql.test.SharedSparkSession + +/** + * Differential test for cross-SCD current-state agreement: given the same randomly-generated + * CDF, every live key's SCD Type 1 target row must equal (in user data columns) the current + * open SCD Type 2 row for that key - this is by definitions of an SCD1 and SCD2 transformation. + * + * By asserting the final outcome of SCD1 equals the final live rows of SCD2, each implementation + * is an effective verifier of the other, and catches regressions or behavior changes made to one + * implementation but not the other. + * + * CDC metadata and SCD2 interval bounds are not compared. + */ +class AutoCdcCrossScdConvergenceSuite + extends ExecutionTest + with SharedSparkSession + with AutoCdcGraphExecutionTestMixin + with AutoCdcRandomCdcTestMixin { + + /** + * Assert SCD1 live rows equal SCD2 current open rows (`__END_AT IS NULL`) on user data + * columns only. + */ + private def assertCrossScdAgreement( + scd1Table: String, + scd2Table: String, + expectedLiveKeyCount: Int): Unit = { + val scd1Data = spark.table(s"$catalog.$namespace.$scd1Table").select( + dataColumnNames.map(functions.col): _* + ) + val scd2CurrentData = spark.table(s"$catalog.$namespace.$scd2Table") + .where(functions.col(Scd2BatchProcessor.endAtColName).isNull) + .select(dataColumnNames.map(functions.col): _*) + + // Verify the number of live keys (i.e rows that haven't been fully deleted) are the same in + // both SCD1 and SCD2, after all events are applied. + val scd1LiveKeyCount = scd1Data.count() + val scd2LiveKeyCount = scd2CurrentData.count() + assert( + scd1LiveKeyCount == expectedLiveKeyCount, + s"Expected $expectedLiveKeyCount live SCD1 keys, found $scd1LiveKeyCount") + assert( + scd2LiveKeyCount == expectedLiveKeyCount, + s"Expected $expectedLiveKeyCount live SCD2 keys, found $scd2LiveKeyCount") + + checkAnswer(scd1Data, scd2CurrentData) Review Comment: Assert that no-op runs actually happen. Right now, if `noOpContinuationProbability` or the `sequence` exclusion is changed later, the suites still pass and you quietly go back to never exercising run-coalescing - the gap I flagged last round, which nothing would catch. After the SCD2 run, check the aux table has at least one hidden no-op row: ```scala assert( spark.table(auxTableNameFor(scd2Table)) .where(functions.col(Scd2BatchProcessor.endAtColName).isNull) .count() > 0) ``` Tombstones always have a non-null `__END_AT`, so a null one is a hidden no-op row. ########## sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcRandomCdcTestMixin.scala: ########## @@ -0,0 +1,272 @@ +/* + * 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.graph + +import scala.collection.mutable.ArrayBuffer +import scala.util.Random + +import org.apache.spark.sql.execution.streaming.runtime.MemoryStream +import org.apache.spark.sql.functions +import org.apache.spark.sql.pipelines.autocdc.{ColumnSelection, ScdType, UnqualifiedColumnName} +import org.apache.spark.sql.pipelines.graph.AutoCdcRandomCdcTestMixin.SourceRow +import org.apache.spark.sql.pipelines.utils.ExecutionTest +import org.apache.spark.sql.test.SharedSparkSession + +object AutoCdcRandomCdcTestMixin { + /** + * A single CDC event in a randomly-generated AutoCDC source stream. + * + * @param key Identity column (the AutoCDC `keys`). + * @param name Data column (nullable string). + * @param amount Data column (nullable int). + * @param active Data column (nullable boolean). + * @param sequence Sequencing value (the AutoCDC `sequencing` expression). + * @param isDelete Drives the AutoCDC `deleteCondition`; `true` marks the event as a delete, + * `false` as an upsert. Excluded from the target via `columnSelection`. + */ + case class SourceRow( + key: Int, + name: Option[String], + amount: Option[Int], + active: Option[Boolean], + sequence: Long, + isDelete: Boolean) +} + +/** + * Shared random-CDC fixture helpers for AutoCDC differential convergence suites + * ([[AutoCdcOutOfOrderConvergenceSuite]], [[AutoCdcCrossScdConvergenceSuite]]). + * + * Owns the common event schema, stream generator, and microbatch feeding for random-data AutoCDC + * suites. + * + * Exposed random data generation knobs (optional; defaults are CI-sized): + * - `spark.sql.test.autocdc.convergenceBaseSeed` + * - `spark.sql.test.autocdc.convergenceNumSeeds` + * - `spark.sql.test.autocdc.convergenceNumKeys` + * - `spark.sql.test.autocdc.convergenceMaxEventsPerKey` + * - `spark.sql.test.autocdc.convergenceNumBatches` + * + * Suites may override these defaults when they genuinely need a different baseline. For local + * stress testing, for example: + * {{{ + * build/sbt \ + * -Dspark.sql.test.autocdc.convergenceNumSeeds=10 \ + * -Dspark.sql.test.autocdc.convergenceNumKeys=100 \ + * 'pipelines/testOnly *AutoCdcCrossScdConvergenceSuite' + * }}} + * + * The shared base seed property supplies the first iteration's seed and deterministically derives + * any remaining per-iteration seeds from it. + */ +trait AutoCdcRandomCdcTestMixin { + self: ExecutionTest with SharedSparkSession with AutoCdcGraphExecutionTestMixin => + + // Probability an event is a delete; (1 - this) is the upsert probability. + protected val deleteEventProbability: Double = 0.20 + // Probability an event is immediately re-emitted with the same sequence and payload. + protected val duplicateEventProbability: Double = 0.15 + // Probability an upsert repeats the previous upsert's payload at a new sequence. In SCD2 this + // will produce a no-op upsert row, provided that sequence is excluded from track-history column + // selection. + protected val noOpContinuationProbability: Double = 0.15 + // Probability an optional payload column is non-null; (1 - this) is the null probability. + protected val nonNullProbability: Double = 0.75 + + // CI-sized defaults shared by every convergence suite. Override in a suite only when that + // suite genuinely needs a different baseline; prefer the shared system properties for + // local stress scaling so both suites stay aligned under normal CI. + protected val defaultBaseSeed: Long = 0x5EEDL + protected val defaultNumDistinctKeys: Int = 5 + protected val defaultMaxUniqueEventsPerKey: Int = 80 + protected val defaultNumBatches: Int = 8 + protected val defaultNumSeedsPerRun: Int = 1 + + // Exposed so suite failure clues can tell callers how to force a deterministic replay. + protected val baseSeedSystemProperty: String = + "spark.sql.test.autocdc.convergenceBaseSeed" + protected val numSeedsSystemProperty: String = + "spark.sql.test.autocdc.convergenceNumSeeds" + private val numKeysSystemProperty: String = + "spark.sql.test.autocdc.convergenceNumKeys" + private val maxEventsPerKeySystemProperty: String = + "spark.sql.test.autocdc.convergenceMaxEventsPerKey" + private val numBatchesSystemProperty: String = + "spark.sql.test.autocdc.convergenceNumBatches" + + private def positiveIntProp(name: String, default: Int): Int = { + val value = Option(System.getProperty(name)).map(_.toInt).getOrElse(default) + require(value > 0, s"$name must be positive, but got $value") + value + } + + private def resolveBaseSeed(): Long = { + Option(System.getProperty(baseSeedSystemProperty)) + .map(_.toLong) + .getOrElse(defaultBaseSeed) + } + + private def resolveNumSeeds(): Int = + positiveIntProp(numSeedsSystemProperty, defaultNumSeedsPerRun) + + protected def resolveNumDistinctKeys(): Int = + positiveIntProp(numKeysSystemProperty, defaultNumDistinctKeys) + + protected def resolveMaxUniqueEventsPerKey(): Int = + positiveIntProp(maxEventsPerKeySystemProperty, defaultMaxUniqueEventsPerKey) + + protected def resolveNumBatches(): Int = + positiveIntProp(numBatchesSystemProperty, defaultNumBatches) + + /** + * Invoke `callback(seed, seedIndex)` once per configured seed. The first iteration uses the + * base seed directly; any remaining iteration seeds are deterministically derived from it. + */ + protected def forEachConvergenceSeed(callback: (Long, Int) => Unit): Unit = { + val baseSeed = resolveBaseSeed() + val numSeeds = resolveNumSeeds() + val masterRand = new Random(baseSeed) + val seeds = baseSeed +: Seq.fill(numSeeds - 1)(masterRand.nextLong()) Review Comment: Give each test its own seed. Every test builds `new Random(0x5EED)`, so all three generate the same events and the same shuffle. The cross-SCD test replays exactly what the out-of-order tests already ran, which means a CI run explores one stream rather than three. Mix the test name into the seed, e.g. `baseSeed ^ testName.hashCode`, and keep printing the effective seed in the clue so it is still reproducible. ########## sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcRandomCdcTestMixin.scala: ########## @@ -0,0 +1,272 @@ +/* + * 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.graph + +import scala.collection.mutable.ArrayBuffer +import scala.util.Random + +import org.apache.spark.sql.execution.streaming.runtime.MemoryStream +import org.apache.spark.sql.functions +import org.apache.spark.sql.pipelines.autocdc.{ColumnSelection, ScdType, UnqualifiedColumnName} +import org.apache.spark.sql.pipelines.graph.AutoCdcRandomCdcTestMixin.SourceRow +import org.apache.spark.sql.pipelines.utils.ExecutionTest +import org.apache.spark.sql.test.SharedSparkSession + +object AutoCdcRandomCdcTestMixin { + /** + * A single CDC event in a randomly-generated AutoCDC source stream. + * + * @param key Identity column (the AutoCDC `keys`). + * @param name Data column (nullable string). + * @param amount Data column (nullable int). + * @param active Data column (nullable boolean). + * @param sequence Sequencing value (the AutoCDC `sequencing` expression). + * @param isDelete Drives the AutoCDC `deleteCondition`; `true` marks the event as a delete, + * `false` as an upsert. Excluded from the target via `columnSelection`. + */ + case class SourceRow( + key: Int, + name: Option[String], + amount: Option[Int], + active: Option[Boolean], + sequence: Long, + isDelete: Boolean) +} + +/** + * Shared random-CDC fixture helpers for AutoCDC differential convergence suites + * ([[AutoCdcOutOfOrderConvergenceSuite]], [[AutoCdcCrossScdConvergenceSuite]]). + * + * Owns the common event schema, stream generator, and microbatch feeding for random-data AutoCDC + * suites. + * + * Exposed random data generation knobs (optional; defaults are CI-sized): + * - `spark.sql.test.autocdc.convergenceBaseSeed` + * - `spark.sql.test.autocdc.convergenceNumSeeds` + * - `spark.sql.test.autocdc.convergenceNumKeys` + * - `spark.sql.test.autocdc.convergenceMaxEventsPerKey` + * - `spark.sql.test.autocdc.convergenceNumBatches` + * + * Suites may override these defaults when they genuinely need a different baseline. For local + * stress testing, for example: + * {{{ + * build/sbt \ + * -Dspark.sql.test.autocdc.convergenceNumSeeds=10 \ + * -Dspark.sql.test.autocdc.convergenceNumKeys=100 \ + * 'pipelines/testOnly *AutoCdcCrossScdConvergenceSuite' + * }}} + * + * The shared base seed property supplies the first iteration's seed and deterministically derives + * any remaining per-iteration seeds from it. + */ +trait AutoCdcRandomCdcTestMixin { + self: ExecutionTest with SharedSparkSession with AutoCdcGraphExecutionTestMixin => + + // Probability an event is a delete; (1 - this) is the upsert probability. + protected val deleteEventProbability: Double = 0.20 + // Probability an event is immediately re-emitted with the same sequence and payload. + protected val duplicateEventProbability: Double = 0.15 + // Probability an upsert repeats the previous upsert's payload at a new sequence. In SCD2 this + // will produce a no-op upsert row, provided that sequence is excluded from track-history column + // selection. + protected val noOpContinuationProbability: Double = 0.15 + // Probability an optional payload column is non-null; (1 - this) is the null probability. + protected val nonNullProbability: Double = 0.75 + + // CI-sized defaults shared by every convergence suite. Override in a suite only when that + // suite genuinely needs a different baseline; prefer the shared system properties for + // local stress scaling so both suites stay aligned under normal CI. + protected val defaultBaseSeed: Long = 0x5EEDL + protected val defaultNumDistinctKeys: Int = 5 + protected val defaultMaxUniqueEventsPerKey: Int = 80 + protected val defaultNumBatches: Int = 8 + protected val defaultNumSeedsPerRun: Int = 1 + + // Exposed so suite failure clues can tell callers how to force a deterministic replay. + protected val baseSeedSystemProperty: String = + "spark.sql.test.autocdc.convergenceBaseSeed" + protected val numSeedsSystemProperty: String = + "spark.sql.test.autocdc.convergenceNumSeeds" + private val numKeysSystemProperty: String = + "spark.sql.test.autocdc.convergenceNumKeys" + private val maxEventsPerKeySystemProperty: String = + "spark.sql.test.autocdc.convergenceMaxEventsPerKey" + private val numBatchesSystemProperty: String = + "spark.sql.test.autocdc.convergenceNumBatches" + + private def positiveIntProp(name: String, default: Int): Int = { + val value = Option(System.getProperty(name)).map(_.toInt).getOrElse(default) + require(value > 0, s"$name must be positive, but got $value") + value + } + + private def resolveBaseSeed(): Long = { + Option(System.getProperty(baseSeedSystemProperty)) + .map(_.toLong) + .getOrElse(defaultBaseSeed) + } + + private def resolveNumSeeds(): Int = + positiveIntProp(numSeedsSystemProperty, defaultNumSeedsPerRun) + + protected def resolveNumDistinctKeys(): Int = + positiveIntProp(numKeysSystemProperty, defaultNumDistinctKeys) + + protected def resolveMaxUniqueEventsPerKey(): Int = + positiveIntProp(maxEventsPerKeySystemProperty, defaultMaxUniqueEventsPerKey) + + protected def resolveNumBatches(): Int = + positiveIntProp(numBatchesSystemProperty, defaultNumBatches) + + /** + * Invoke `callback(seed, seedIndex)` once per configured seed. The first iteration uses the + * base seed directly; any remaining iteration seeds are deterministically derived from it. + */ + protected def forEachConvergenceSeed(callback: (Long, Int) => Unit): Unit = { + val baseSeed = resolveBaseSeed() + val numSeeds = resolveNumSeeds() + val masterRand = new Random(baseSeed) + val seeds = baseSeed +: Seq.fill(numSeeds - 1)(masterRand.nextLong()) + seeds.zipWithIndex.foreach { case (seed, seedIndex) => + callback(seed, seedIndex) + } + } + + // Forward declare key, sequence, and data columns, so that inheriting suites can reference them. + protected val keyColumn: String = "key" + protected val nameColumn: String = "name" + protected val amountColumn: String = "amount" + protected val activeColumn: String = "active" + protected val sequenceColumn: String = "sequence" + protected val isDeleteColumn: String = "is_delete" + + protected val sourceColumnNames: Seq[String] = + Seq(keyColumn, nameColumn, amountColumn, activeColumn, sequenceColumn, isDeleteColumn) + + /** User data columns on the target; excludes CDC metadata and SCD2 interval bounds. */ + protected val dataColumnNames: Seq[String] = + Seq(keyColumn, nameColumn, amountColumn, activeColumn, sequenceColumn) + + private def randomUpsertOrDelete( + rand: Random, key: Int, sequence: Long, isDelete: Boolean): SourceRow = { + val colorPalette = Seq("red", "blue", "green", "yellow") + SourceRow( + key = key, + name = Option.when(rand.nextDouble() < nonNullProbability)( + colorPalette(rand.nextInt(colorPalette.length))), + amount = Option.when(rand.nextDouble() < nonNullProbability)(rand.nextInt(100)), + active = Option.when(rand.nextDouble() < nonNullProbability)(rand.nextBoolean()), + sequence = sequence, + isDelete = isDelete + ) + } + + /** + * Generate a sequence-sorted CDC event stream. + */ + protected def generateRandomCdcEventStream(rand: Random): Seq[SourceRow] = { + val numDistinctKeys = resolveNumDistinctKeys() + val maxUniqueEventsPerKey = resolveMaxUniqueEventsPerKey() + + var nextSequence: Long = 0L + val allEvents = ArrayBuffer.empty[SourceRow] + (0 until numDistinctKeys).foreach { key => + val numUniqueEventsForKey = rand.between(1, maxUniqueEventsPerKey + 1) + val eventsForKey = ArrayBuffer.empty[SourceRow] + + (0 until numUniqueEventsForKey).foreach { _ => + val isDelete = rand.nextDouble() < deleteEventProbability + val event = if (isDelete) { + randomUpsertOrDelete(rand, key, nextSequence, isDelete = true) + } else { + val previousEventIfUpsertOpt = eventsForKey.lastOption.filterNot(_.isDelete) + val upsertToNoOpContinueOpt = previousEventIfUpsertOpt.filter( + _ => rand.nextDouble() < noOpContinuationProbability) + + upsertToNoOpContinueOpt match { + case Some(upsertToNoOpContinue) => + // If we're no-op continuing a previous upsert, reuse the same [tracked history] + // columns, incrementing only the sequence. This relies on sequence being the single + // non-track-history column in the AutoCDC configuration. + upsertToNoOpContinue.copy(sequence = nextSequence) + case _ => + // If we're not no-op continuing a previous upsert, create a new upsert event. + randomUpsertOrDelete(rand, key, nextSequence, isDelete = false) + } + } + + // By AutoCDC contract, only exact duplicate re-emissions (handled separately below) may + // reuse sequences for a particular key. Otherwise, the behavior for two unique events for + // the same key with the same sequence leads to undefined behavior. Each distinct event + // creation for this key should increment `nextSequence`. + nextSequence += 1 + eventsForKey += event + + if (rand.nextDouble() < duplicateEventProbability) { + // Full duplicate events are intentionally not counted against `numUniqueEventsForKey`. + // These differ from no-op upsert continuation events, as they share the same sequence as + // their preceding event too, in addition to all other columns. + eventsForKey += event + } + } + + allEvents.addAll(eventsForKey) + } + allEvents.sortBy(_.sequence).toSeq + } + + /** + * Feed `events` through an AutoCDC pipeline of `scdType` across `numBatches` microbatches + * (one pipeline run per microbatch). The target and auxiliary tables are created by pipeline + * materialization from the flow's inferred schema. + */ + protected def runRandomCdcPipeline( + targetTable: String, + scdType: ScdType, + events: Seq[SourceRow], + numBatches: Int): Unit = { + val session = spark + import session.implicits._ + + val stream = MemoryStream[SourceRow] + val ctx = singleAutoCdcFlowPipeline( + flowName = s"${targetTable}_flow", + target = targetTable, + sourceDf = stream.toDF().toDF(sourceColumnNames: _*), + keys = Seq(keyColumn), + sequencing = functions.col(sequenceColumn), + columnSelection = Some(ColumnSelection.ExcludeColumns( + Seq(UnqualifiedColumnName(isDeleteColumn)) + )), + deleteCondition = Some(functions.col(isDeleteColumn) === true), + scdType = scdType, + trackHistorySelection = scdType match { Review Comment: This removes fuzz coverage of SCD2's default track-history mode. On master, the out-of-order suite's SCD2 test ran with `trackHistorySelection` unset, so every column was tracked and runs never coalesced. `runRandomCdcPipeline` now always excludes `sequence` for SCD2, so no suite fuzzes that mode anymore - it survives only in the deterministic suites like `AutoCdcScd2SinglePipelineSuite`. The exclusion mode is what I asked for last round, but it should sit alongside the old mode rather than replace it. Make `trackHistorySelection` a parameter (defaulting to what you have now) so the two suites can cover different modes. One caveat: SPARK-58937 needed out-of-order *plus* coalescing, so if the out-of-order suite goes back to the default mode you lose the combination that found it. Running its SCD2 test over both modes keeps everything, at roughly double the cost for that test. -- 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]
