This is an automated email from the ASF dual-hosted git repository.
szehon-ho pushed a commit to branch branch-4.3
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.3 by this push:
new f68e4e8cb8de [SPARK-58572][SDP] AutoCDC Cross SCD Convergence Suite
f68e4e8cb8de is described below
commit f68e4e8cb8def2f19c4feef7e6e7f266cd2faac9
Author: AnishMahto <[email protected]>
AuthorDate: Thu Aug 27 15:29:20 2026 -0700
[SPARK-58572][SDP] AutoCDC Cross SCD Convergence Suite
### What changes were proposed in this pull request?
Given the same CDF (but can be shuffled in arbitrary order), every key's
output row in SCD1 should be exactly equal (in data columns, not necessarily
CDC metadata) to the last row per key in SCD2. In other words, SCD1 and SCD2
should converge.
Using this principle and reusing the existing AutoCDC out-of-order
convergence test harness, we should setup a cross-SCD convergence test so that
one implementation can keep the other in check for correctness, and detect
regressions made to one SCD's implementation but not the other.
Concretely, two things are accomplished in this PR:
1. Extract shared convergence-testing infra from
`AutoCdcOutOfOrderConvergenceSuite` into a new `AutoCdcRandomCdcTestMixin` trait
2. Implement the cross-SCD convergence test in
`AutoCdcCrossScdConvergenceSuite`
### Why are the changes needed?
Correctness and regression testing of AutoCDC SCD1 and SCD2 implementations.
### Does this PR introduce _any_ user-facing change?
No, test-only change.
### How was this patch tested?
Added `AutoCdcCrossScdConvergenceSuite`. Test-only change.
### Was this patch authored or co-authored using generative AI tooling?
Co-authored with Claude Sonnet 5 and GPT-5.6 Sol.
Closes #58055 from AnishMahto/SPARK-58572-cross-scd-fuzz-testing.
Authored-by: AnishMahto <[email protected]>
Signed-off-by: Szehon Ho <[email protected]>
(cherry picked from commit 3d0c5e64d72a091e649ae3095e8a63e61a0e6ca6)
Signed-off-by: Szehon Ho <[email protected]>
---
.../graph/AutoCdcCrossScdConvergenceSuite.scala | 108 ++++++++
.../graph/AutoCdcOutOfOrderConvergenceSuite.scala | 226 +++-------------
.../graph/AutoCdcRandomCdcTestMixin.scala | 295 +++++++++++++++++++++
3 files changed, 445 insertions(+), 184 deletions(-)
diff --git
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcCrossScdConvergenceSuite.scala
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcCrossScdConvergenceSuite.scala
new file mode 100644
index 000000000000..64170f48c0b7
--- /dev/null
+++
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcCrossScdConvergenceSuite.scala
@@ -0,0 +1,108 @@
+/*
+ * 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)
+ }
+
+ private val crossScdConvergenceTestName =
+ "SCD1 current rows match SCD2 open rows for the same shuffled CDC stream"
+
+ test(crossScdConvergenceTestName) {
+ val numDistinctKeys = resolveNumDistinctKeys()
+ val maxUniqueEventsPerKey = resolveMaxUniqueEventsPerKey()
+ val numBatches = resolveNumBatches()
+
+ forEachConvergenceSeed(crossScdConvergenceTestName) { (seed, seedIndex) =>
+ val rand = new Random(seed)
+ val sortedEventStream = generateRandomCdcEventStream(rand)
+ val shuffledEventStream = rand.shuffle(sortedEventStream)
+ val expectedLiveKeyCount = sortedEventStream
+ .groupBy(_.key)
+ .values
+ .count(events => !events.maxBy(_.sequence).isDelete)
+
+ // Avoid dumping thousands of events into every clue string (ScalaTest
evaluates clues
+ // eagerly).
+ withClue(
+ s"\ncross-SCD convergence testName=$crossScdConvergenceTestName " +
+ s"seedIndex=$seedIndex seed=$seed " +
+ s"(rerun this test with -D$convergenceReproSeedSystemProperty=$seed to
reproduce)\n" +
+ s"keys=$numDistinctKeys maxEventsPerKey=$maxUniqueEventsPerKey " +
+ s"numBatches=$numBatches expectedLiveKeys=$expectedLiveKeyCount " +
+ s"events=${sortedEventStream.size}\n"
+ ) {
+ val scd1Table = s"cross_scd1_$seedIndex"
+ val scd2Table = s"cross_scd2_$seedIndex"
+ runRandomCdcPipeline(scd1Table, ScdType.Type1, shuffledEventStream,
numBatches)
+ runRandomCdcPipeline(scd2Table, ScdType.Type2, shuffledEventStream,
numBatches)
+ assertCrossScdAgreement(scd1Table, scd2Table, expectedLiveKeyCount)
+ }
+ }
+ }
+}
diff --git
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcOutOfOrderConvergenceSuite.scala
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcOutOfOrderConvergenceSuite.scala
index ef4ba62052fb..e8e4eb5fdfce 100644
---
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcOutOfOrderConvergenceSuite.scala
+++
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcOutOfOrderConvergenceSuite.scala
@@ -17,37 +17,12 @@
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.AutoCdcOutOfOrderConvergenceSuite.SourceRow
-import org.apache.spark.sql.pipelines.utils.{ExecutionTest,
TestGraphRegistrationContext}
+import org.apache.spark.sql.pipelines.autocdc.ScdType
+import org.apache.spark.sql.pipelines.utils.ExecutionTest
import org.apache.spark.sql.test.SharedSparkSession
-object AutoCdcOutOfOrderConvergenceSuite {
- /**
- * A single CDC event in the 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)
-}
-
/**
* Differential test for the AutoCDC merge's order-invariance property, for
both SCD Type 1 and
* SCD Type 2: feeding the same randomly-generated CDC event stream as a
single sorted micro-batch
@@ -56,120 +31,8 @@ object AutoCdcOutOfOrderConvergenceSuite {
class AutoCdcOutOfOrderConvergenceSuite
extends ExecutionTest
with SharedSparkSession
- with AutoCdcGraphExecutionTestMixin {
-
- import testImplicits._
-
- // Distinct keys in the generated event stream.
- private val numDistinctKeys: Int = 5
- // Upper bound on unique events (one per sequence) generated per key, before
intentionally
- // duplicating some events.
- private val maxUniqueEventsPerKey: Int = 80
- // Probability an event is a delete; (1 - this) is the upsert probability.
- private val deleteEventProbability: Double = 0.20
- // Probability an event is immediately re-emitted with the same sequence and
payload.
- private val duplicateEventProbability: Double = 0.15
- // Probability an optional payload column is non-null; (1 - this) is the
null probability.
- private val nonNullProbability: Double = 0.75
- // Number of microbatches the out-of-order pipeline splits the shuffled
events across.
- private val numOutOfOrderBatches: Int = 8
-
- // System property used to pin the test seed for reproduction. If unset, the
suite generates a
- // fresh seed on each run and reports it in the failure message so a failing
seed can be replayed
- // by setting this property. Mirrors the convention used by
`RandomDataGenerator` and other Spark
- // suites that expose tunables via `spark.sql.test.<feature>` system
properties.
- private val seedSystemProperty: String =
- "spark.sql.test.autocdc.outOfOrderConvergenceSeed"
-
- private def resolveTestSeed(): Long = {
-
Option(System.getProperty(seedSystemProperty)).map(_.toLong).getOrElse(Random.nextLong())
- }
-
- private val keyColumn: String = "key"
- private val nameColumn: String = "name"
- private val amountColumn: String = "amount"
- private val activeColumn: String = "active"
- private val sequenceColumn: String = "sequence"
- private val isDeleteColumn: String = "is_delete"
-
- private val sourceColumnNames: Seq[String] =
- Seq(keyColumn, nameColumn, amountColumn, activeColumn, sequenceColumn,
isDeleteColumn)
-
- 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
- )
- }
-
- private def generateRandomCdcEventStream(rand: Random): Seq[SourceRow] = {
- var nextSequence: Long = 0L
- val events = ArrayBuffer.empty[SourceRow]
- (0 until numDistinctKeys).foreach { key =>
- val numUniqueEventsForKey = rand.between(1, maxUniqueEventsPerKey + 1)
- (0 until numUniqueEventsForKey).foreach { _ =>
- val isDelete = rand.nextDouble() < deleteEventProbability
- val event = randomUpsertOrDelete(rand, key, nextSequence, isDelete)
- nextSequence += 1
- events += event
- if (rand.nextDouble() < duplicateEventProbability) {
- events += event
- }
- }
- }
- events.sortBy(_.sequence).toSeq
- }
-
- /** Build a pipeline context with a single AutoCDC flow of `scdType` reading
from `stream`. */
- private def buildPipelineContext(
- targetTable: String,
- stream: MemoryStream[SourceRow],
- scdType: ScdType): TestGraphRegistrationContext = {
- new TestGraphRegistrationContext(spark) {
- registerTable(targetTable, catalog = Some(catalog), database =
Some(namespace))
- registerFlow(autoCdcFlow(
- name = s"${targetTable}_flow",
- target = targetTable,
- query = dfFlowFunc(stream.toDF().toDF(sourceColumnNames: _*)),
- keys = Seq(keyColumn),
- sequencing = functions.col(sequenceColumn),
- deleteCondition = Some(functions.col(isDeleteColumn) === true),
- columnSelection = Some(ColumnSelection.ExcludeColumns(
- Seq(UnqualifiedColumnName(isDeleteColumn))
- )),
- scdType = scdType
- ))
- }
- }
-
- /**
- * DDL fragment for the SCD-type-specific reserved columns a target table
carries after the
- * user-selected data columns: the CDC metadata column for SCD1, and the
interval bounds plus
- * metadata column for SCD2. The sequencing type is BIGINT here.
- */
- private def reservedColumnsDdl(scdType: ScdType): String = scdType match {
- case ScdType.Type1 => scd1MetadataDdl
- case ScdType.Type2 => scd2MetadataDdl
- }
-
- private def createTargetTable(targetTable: String, scdType: ScdType): Unit =
{
- spark.sql(
- s"CREATE TABLE $catalog.$namespace.$targetTable (" +
- s"`$keyColumn` INT NOT NULL, " +
- s"`$nameColumn` STRING, " +
- s"`$amountColumn` INT, " +
- s"`$activeColumn` BOOLEAN, " +
- s"`$sequenceColumn` BIGINT NOT NULL, " +
- s"${reservedColumnsDdl(scdType)})"
- )
- }
+ with AutoCdcGraphExecutionTestMixin
+ with AutoCdcRandomCdcTestMixin {
private def assertTargetsConverge(inOrderTable: String, outOfOrderTable:
String): Unit = {
checkAnswer(
@@ -178,54 +41,49 @@ class AutoCdcOutOfOrderConvergenceSuite
)
}
- private def runConvergenceTest(seed: Long, scdType: ScdType): Unit = {
- val rand = new Random(seed)
- val sortedEventStream = generateRandomCdcEventStream(rand)
- val shuffledEventStream = rand.shuffle(sortedEventStream)
-
- withClue(
- s"\nscdType=${scdType.label} seed=$seed " +
- s"(rerun with -D$seedSystemProperty=$seed to reproduce)\n" +
- s"events (${sortedEventStream.size} total, sorted by sequence):\n" +
- sortedEventStream.map(r => s" $r").mkString("\n") + "\n"
- ) {
- // Table names are scd-type-suffixed purely for readability: the SCD1
and SCD2 tests run as
- // separate test cases and the mixin's afterEach resets the catalog
between them, so they
- // could not collide even with identical names; the suffix just makes a
failing run's tables
- // self-identifying.
- val suffix = scdType.label.toLowerCase(java.util.Locale.ROOT)
- val inOrderTable = s"inorder_target_$suffix"
- val outOfOrderTable = s"outoforder_target_$suffix"
- createTargetTable(inOrderTable, scdType)
- createTargetTable(outOfOrderTable, scdType)
-
- val inOrderStream = MemoryStream[SourceRow]
- val inOrderCtx = buildPipelineContext(inOrderTable, inOrderStream,
scdType)
- inOrderStream.addData(sortedEventStream: _*)
- runPipeline(inOrderCtx)
-
- val outOfOrderStream = MemoryStream[SourceRow]
- val outOfOrderCtx = buildPipelineContext(outOfOrderTable,
outOfOrderStream, scdType)
- val totalEvents = shuffledEventStream.size
- (0 until numOutOfOrderBatches).foreach { batchIndex =>
- val batchStart = batchIndex * totalEvents / numOutOfOrderBatches
- val batchEnd = (batchIndex + 1) * totalEvents / numOutOfOrderBatches
- outOfOrderStream.addData(shuffledEventStream.slice(batchStart,
batchEnd): _*)
- runPipeline(outOfOrderCtx)
+ private def runConvergenceTest(scdType: ScdType, testName: String): Unit = {
+ val numDistinctKeys = resolveNumDistinctKeys()
+ val maxUniqueEventsPerKey = resolveMaxUniqueEventsPerKey()
+ val numBatches = resolveNumBatches()
+
+ forEachConvergenceSeed(testName) { (seed, seedIndex) =>
+ val rand = new Random(seed)
+ val sortedEventStream = generateRandomCdcEventStream(rand)
+ val shuffledEventStream = rand.shuffle(sortedEventStream)
+
+ withClue(
+ s"\nout-of-order convergence scdType=${scdType.label}
testName=$testName " +
+ s"seedIndex=$seedIndex seed=$seed " +
+ s"(rerun this test with -D$convergenceReproSeedSystemProperty=$seed to
reproduce)\n" +
+ s"keys=$numDistinctKeys maxEventsPerKey=$maxUniqueEventsPerKey " +
+ s"numBatches=$numBatches events=${sortedEventStream.size}\n"
+ ) {
+ val inOrderTable = s"inorder_target_$seedIndex"
+ val outOfOrderTable = s"outoforder_target_$seedIndex"
+
+ // In-order baseline: one microbatch with the sequence-sorted stream.
+ runRandomCdcPipeline(inOrderTable, scdType, sortedEventStream,
numBatches = 1)
+ // Out-of-order: same events shuffled across the configured number of
microbatches.
+ runRandomCdcPipeline(outOfOrderTable, scdType, shuffledEventStream,
numBatches)
+
+ // Only the user-visible target must converge. The auxiliary tables
legitimately differ by
+ // arrival order (e.g. deletedByBatchId stamps and cross-batch GC
depend on how events are
+ // batched), so they are not compared.
+ assertTargetsConverge(inOrderTable, outOfOrderTable)
}
-
- // Only the user-visible target must converge. The auxiliary tables
legitimately differ by
- // arrival order (e.g. deletedByBatchId stamps and cross-batch GC depend
on how events are
- // batched), so they are not compared.
- assertTargetsConverge(inOrderTable, outOfOrderTable)
}
}
- test("SCD1 merge converges across micro-batch shuffling for randomly
generated CDC events") {
- runConvergenceTest(resolveTestSeed(), ScdType.Type1)
+ private val scd1OutOfOrderTestName =
+ "SCD1 merge converges across micro-batch shuffling for randomly generated
CDC events"
+ private val scd2OutOfOrderTestName =
+ "SCD2 merge converges across micro-batch shuffling for randomly generated
CDC events"
+
+ test(scd1OutOfOrderTestName) {
+ runConvergenceTest(ScdType.Type1, scd1OutOfOrderTestName)
}
- test("SCD2 merge converges across micro-batch shuffling for randomly
generated CDC events") {
- runConvergenceTest(resolveTestSeed(), ScdType.Type2)
+ test(scd2OutOfOrderTestName) {
+ runConvergenceTest(ScdType.Type2, scd2OutOfOrderTestName)
}
}
diff --git
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcRandomCdcTestMixin.scala
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcRandomCdcTestMixin.scala
new file mode 100644
index 000000000000..df9746e8c093
--- /dev/null
+++
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcRandomCdcTestMixin.scala
@@ -0,0 +1,295 @@
+/*
+ * 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.internal.Logging
+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.convergenceMultiRunBaseSeed` (multi-run mode)
+ * - `spark.sql.test.autocdc.convergenceMultiRunNumSeeds` (multi-run mode)
+ * - `spark.sql.test.autocdc.convergenceReproSeed` (repro mode; overrides
multi-run)
+ * - `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.convergenceMultiRunNumSeeds=10 \
+ * -Dspark.sql.test.autocdc.convergenceNumKeys=100 \
+ * 'pipelines/testOnly *AutoCdcCrossScdConvergenceSuite'
+ * }}}
+ *
+ * Two execution modes:
+ * - Multi-run (default): each test generates `convergenceMultiRunNumSeeds`
random CDC streams
+ * from `convergenceMultiRunBaseSeed`. Used for CI and local stress
testing.
+ * - Repro: when `convergenceReproSeed` is set, each test runs a single
stream with that seed
+ * and ignores `convergenceMultiRunBaseSeed` and
`convergenceMultiRunNumSeeds`. Used to replay
+ * one failing case from the seed printed in a failure message.
+ */
+trait AutoCdcRandomCdcTestMixin extends Logging {
+ 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 multiRunBaseSeedSystemProperty: String =
+ "spark.sql.test.autocdc.convergenceMultiRunBaseSeed"
+ protected val multiRunNumSeedsSystemProperty: String =
+ "spark.sql.test.autocdc.convergenceMultiRunNumSeeds"
+ protected val convergenceReproSeedSystemProperty: String =
+ "spark.sql.test.autocdc.convergenceReproSeed"
+ 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
+ }
+
+ protected def configuredBaseSeed: Long =
+ Option(System.getProperty(multiRunBaseSeedSystemProperty))
+ .map(_.toLong)
+ .getOrElse(defaultBaseSeed)
+
+ private def configuredReproSeed: Option[Long] =
+
Option(System.getProperty(convergenceReproSeedSystemProperty)).map(_.toLong)
+
+ private def resolveNumSeeds(): Int =
+ positiveIntProp(multiRunNumSeedsSystemProperty, 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)` for each convergence iteration.
+ *
+ * When [[configuredReproSeed]] is set, invokes the callback once with that
seed (repro mode).
+ * Otherwise derives `numSeeds` iterations from [[configuredBaseSeed]] and
[[testName]]
+ * (multi-run mode).
+ */
+ protected def forEachConvergenceSeed(testName: String)(callback: (Long, Int)
=> Unit): Unit = {
+ configuredReproSeed match {
+ case Some(reproSeed) =>
+ logInfo(
+ s"AutoCDC convergence repro mode for test '$testName': " +
+ s"-D$convergenceReproSeedSystemProperty=$reproSeed " +
+ s"(-D$multiRunBaseSeedSystemProperty and " +
+ s"-D$multiRunNumSeedsSystemProperty are ignored)")
+ callback(reproSeed, 0)
+ case None =>
+ val perTestBaseSeed = configuredBaseSeed ^ testName.hashCode.toLong
+ val numSeeds = resolveNumSeeds()
+ val masterRand = new Random(perTestBaseSeed)
+ val seeds = perTestBaseSeed +: 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 {
+ case ScdType.Type1 => None
+ case ScdType.Type2 => Some(ColumnSelection.ExcludeColumns(
+ Seq(UnqualifiedColumnName(sequenceColumn))))
+ }
+ )
+ val totalEvents = events.size
+ (0 until numBatches).foreach { batchIndex =>
+ val batchStart = batchIndex * totalEvents / numBatches
+ val batchEnd = (batchIndex + 1) * totalEvents / numBatches
+ stream.addData(events.slice(batchStart, batchEnd): _*)
+ runPipeline(ctx)
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
