This is an automated email from the ASF dual-hosted git repository.
jose-torres pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 4d3ca4b7c8c1 [SPARK-57150][SDP] SCD1 Out-of-order Convergence Suite
4d3ca4b7c8c1 is described below
commit 4d3ca4b7c8c193ff96fe034280aa1cfc17d89627
Author: AnishMahto <[email protected]>
AuthorDate: Tue Jun 23 08:37:27 2026 -0700
[SPARK-57150][SDP] SCD1 Out-of-order Convergence Suite
### What changes were proposed in this pull request?
A key feature of SDP's AutoCDC implementation is that it supports
reconciling out-of-order (by sequence) events. This support also adds
significant complexity to the reconciliation logic as it requires
cross-microbatch stateful tracking in the auxiliary table, and is prone to
breaking as the implementation evolves over time.
Introduce an A/B style test suite to execute the implementation on both a
sequence-sorted single-microbatch event stream and the same events on a
shuffled multi-microbatch event stream. If out-of-order processing is correct,
then the SCD1 implementation should produce the same target tables for both
runs.
Data is randomly generated, but with a constant seed for reproducibility.
### Why are the changes needed?
Test correctness of the AutoCDC SCD1 out-of-order reconciliation algorithm.
Prevents signal for existing implementation's correctness and helps prevent
regressions in future iterations.
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
Test only change.
### Was this patch authored or co-authored using generative AI tooling?
Co-authored with Claude Opus 4.7
Closes #56214 from AnishMahto/SPARK-57150-SCD1-OOO-convergence-suite.
Authored-by: AnishMahto <[email protected]>
Signed-off-by: Jose Torres <[email protected]>
(cherry picked from commit c87b8dca5775e64ec4b0e2626902d9a7d89a2b28)
Signed-off-by: Jose Torres <[email protected]>
---
.../AutoCdcScd1OutOfOrderConvergenceSuite.scala | 207 +++++++++++++++++++++
1 file changed, 207 insertions(+)
diff --git
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1OutOfOrderConvergenceSuite.scala
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1OutOfOrderConvergenceSuite.scala
new file mode 100644
index 000000000000..fb3a179f25c7
--- /dev/null
+++
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcScd1OutOfOrderConvergenceSuite.scala
@@ -0,0 +1,207 @@
+/*
+ * 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,
UnqualifiedColumnName}
+import
org.apache.spark.sql.pipelines.graph.AutoCdcScd1OutOfOrderConvergenceSuite.SourceRow
+import org.apache.spark.sql.pipelines.utils.{ExecutionTest,
TestGraphRegistrationContext}
+import org.apache.spark.sql.test.SharedSparkSession
+
+object AutoCdcScd1OutOfOrderConvergenceSuite {
+ /**
+ * 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 SCD1 AutoCDC merge's order-invariance property:
feeding the same
+ * randomly-generated CDC event stream as a single sorted micro-batch and as
several shuffled
+ * micro-batches must converge to the same target table contents.
+ */
+class AutoCdcScd1OutOfOrderConvergenceSuite
+ extends ExecutionTest
+ with SharedSparkSession
+ with AutoCdcGraphExecutionTestMixin {
+
+ // 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.scd1OutOfOrderConvergenceSeed"
+
+ 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 SCD1 AutoCDC flow reading from
`stream`. */
+ private def buildPipelineContext(
+ targetTable: String,
+ stream: MemoryStream[SourceRow]): 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))
+ ))
+ ))
+ }
+ }
+
+ private def createTargetTable(targetTable: String): 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"$cdcMetadataDdl)"
+ )
+ }
+
+ private def assertTargetsConverge(inOrderTable: String, outOfOrderTable:
String): Unit = {
+ checkAnswer(
+ spark.table(s"$catalog.$namespace.$outOfOrderTable"),
+ spark.table(s"$catalog.$namespace.$inOrderTable")
+ )
+ }
+
+ private def runConvergenceTest(seed: Long): Unit = {
+ val session = spark
+ import session.implicits._
+
+ val rand = new Random(seed)
+ val sortedEventStream = generateRandomCdcEventStream(rand)
+ val shuffledEventStream = rand.shuffle(sortedEventStream)
+
+ withClue(
+ s"\nseed=$seed (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"
+ ) {
+ val inOrderTable = "inorder_target"
+ val outOfOrderTable = "outoforder_target"
+ createTargetTable(inOrderTable)
+ createTargetTable(outOfOrderTable)
+
+ val inOrderStream = MemoryStream[SourceRow]
+ val inOrderCtx = buildPipelineContext(inOrderTable, inOrderStream)
+ inOrderStream.addData(sortedEventStream: _*)
+ runPipeline(inOrderCtx)
+
+ val outOfOrderStream = MemoryStream[SourceRow]
+ val outOfOrderCtx = buildPipelineContext(outOfOrderTable,
outOfOrderStream)
+ 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)
+ }
+
+ assertTargetsConverge(inOrderTable, outOfOrderTable)
+ }
+ }
+
+ test("SCD1 merge converges across micro-batch shuffling for randomly
generated CDC events") {
+ runConvergenceTest(resolveTestSeed())
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
