szehon-ho commented on code in PR #58055: URL: https://github.com/apache/spark/pull/58055#discussion_r3808209970
########## sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcRandomCdcTestMixin.scala: ########## @@ -0,0 +1,236 @@ +/* + * 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, target DDL, microbatch feeding, for random-data Review Comment: Drop "target DDL" from this sentence (and the stray comma before "for random-data AutoCDC suites"). There is no DDL here - `runRandomCdcPipeline`'s own doc correctly says the tables come from pipeline materialization. ########## sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcRandomCdcTestMixin.scala: ########## @@ -0,0 +1,236 @@ +/* + * 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, target DDL, 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.convergenceMaxBatches` + * + * 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 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 defaultNumDistinctKeys: Int = 10 + protected val defaultMaxUniqueEventsPerKey: Int = 40 Review Comment: Optional, feel free to skip: consider keeping `defaultMaxUniqueEventsPerKey` closer to the previous 80. Keys went 5 -> 10 while per-key depth went 80 -> 40. Extra keys add independent parallel cases; per-key depth is what stresses SCD2 reconciliation - bisection, decomposition, run boundaries. Not blocking, and the current split is reasonable if it is what fits the runtime budget. ########## sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcCrossScdConvergenceSuite.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.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): 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): _*) + checkAnswer(scd1Data, scd2CurrentData) Review Comment: Also assert the SCD1 row count against the expected live keys. Derive them from `sortedEventStream` - per key take the max-sequence event, keep it unless it is a delete. As written, a bug that fed no events (say in `runRandomCdcPipeline`'s `batchStart`/`batchEnd` slicing) would leave both sides empty and pass. ########## sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcCrossScdConvergenceSuite.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.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): 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): _*) + checkAnswer(scd1Data, scd2CurrentData) + } + + test("SCD1 current rows match SCD2 open rows across independently shuffled CDC streams") { + val numDistinctKeys = resolveNumDistinctKeys() + val maxUniqueEventsPerKey = resolveMaxUniqueEventsPerKey() + val numOutOfOrderBatches = resolveNumOutOfOrderBatches() + + forEachConvergenceSeed { (seed, seedIndex) => + val rand = new Random(seed) + val sortedEventStream = generateRandomCdcEventStream(rand) + + // Independent shuffle / batching RNGs so each SCD side exercises a different arrival order + // while still being fully determined by `seed`. + val scd1Rand = new Random(rand.nextLong()) + val scd2Rand = new Random(rand.nextLong()) + val scd1Shuffled = scd1Rand.shuffle(sortedEventStream) + val scd2Shuffled = scd2Rand.shuffle(sortedEventStream) + val scd1Batches = 1 + scd1Rand.nextInt(numOutOfOrderBatches) + val scd2Batches = 1 + scd2Rand.nextInt(numOutOfOrderBatches) Review Comment: Consider shuffling both sides with the same order instead of independent RNGs. With independent shuffles, a failure could be a genuine SCD1/SCD2 divergence or an order-invariance bug on one side - the property `AutoCdcOutOfOrderConvergenceSuite` already covers. Was the extra state space per run the intent? ########## sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcRandomCdcTestMixin.scala: ########## @@ -0,0 +1,236 @@ +/* + * 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 +} Review Comment: Collapse to one line: `import org.apache.spark.sql.pipelines.autocdc.{ColumnSelection, ScdType, UnqualifiedColumnName}`. It is 94 characters, within the limit. ########## sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcRandomCdcTestMixin.scala: ########## @@ -0,0 +1,236 @@ +/* + * 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, target DDL, 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.convergenceMaxBatches` + * + * 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 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 defaultNumDistinctKeys: Int = 10 + protected val defaultMaxUniqueEventsPerKey: Int = 40 + protected val defaultNumOutOfOrderBatches: Int = 8 + protected val defaultNumSeedsPerRun: Int = 3 + + // 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.convergenceMaxBatches" + + 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(Random.nextLong()) + } + + private def resolveNumSeeds(): Int = + positiveIntProp(numSeedsSystemProperty, defaultNumSeedsPerRun) + + protected def resolveNumDistinctKeys(): Int = + positiveIntProp(numKeysSystemProperty, defaultNumDistinctKeys) + + protected def resolveMaxUniqueEventsPerKey(): Int = + positiveIntProp(maxEventsPerKeySystemProperty, defaultMaxUniqueEventsPerKey) + + protected def resolveNumOutOfOrderBatches(): Int = + positiveIntProp(numBatchesSystemProperty, defaultNumOutOfOrderBatches) + + /** + * 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 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 Review Comment: Worth a short comment here on why every event gets its own sequence value, so this is not changed by accident later. SCD1 and SCD2 deliberately disagree when an upsert and a delete share a sequence: SCD1 keeps the upsert (`incomingWinsDelete` requires a strict `>`), SCD2 keeps the delete (per `Scd2BatchProcessor`'s scaladoc). Both document that case as undefined, so a generator change that allowed colliding sequences would make `AutoCdcCrossScdConvergenceSuite` fail against correct code. ########## sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcRandomCdcTestMixin.scala: ########## @@ -0,0 +1,236 @@ +/* + * 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, target DDL, 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.convergenceMaxBatches` + * + * 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 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 defaultNumDistinctKeys: Int = 10 + protected val defaultMaxUniqueEventsPerKey: Int = 40 + protected val defaultNumOutOfOrderBatches: Int = 8 + protected val defaultNumSeedsPerRun: Int = 3 + + // 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.convergenceMaxBatches" Review Comment: Reconcile these names, or document that the value means two different things. `AutoCdcOutOfOrderConvergenceSuite` uses it as an exact batch count; `AutoCdcCrossScdConvergenceSuite` (lines 73-74) uses it as an upper bound via `1 + rand.nextInt(n)`. The property name says "Max", the accessor says "Num". ########## sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcRandomCdcTestMixin.scala: ########## @@ -0,0 +1,236 @@ +/* + * 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, target DDL, 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.convergenceMaxBatches` + * + * 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 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 defaultNumDistinctKeys: Int = 10 + protected val defaultMaxUniqueEventsPerKey: Int = 40 + protected val defaultNumOutOfOrderBatches: Int = 8 + protected val defaultNumSeedsPerRun: Int = 3 + + // 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.convergenceMaxBatches" + + 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(Random.nextLong()) + } + + private def resolveNumSeeds(): Int = + positiveIntProp(numSeedsSystemProperty, defaultNumSeedsPerRun) + + protected def resolveNumDistinctKeys(): Int = + positiveIntProp(numKeysSystemProperty, defaultNumDistinctKeys) + + protected def resolveMaxUniqueEventsPerKey(): Int = + positiveIntProp(maxEventsPerKeySystemProperty, defaultMaxUniqueEventsPerKey) + + protected def resolveNumOutOfOrderBatches(): Int = + positiveIntProp(numBatchesSystemProperty, defaultNumOutOfOrderBatches) + + /** + * 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() + Review Comment: Remove the trailing whitespace on this line. `WhitespaceEndOfLineChecker` rejects it, and it is why the "Linters, licenses, and dependencies" check is red on `f32f8fa`: `AutoCdcRandomCdcTestMixin.scala:184:0: Whitespace at end of line`. It is the only lint violation in the diff. ########## sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/AutoCdcRandomCdcTestMixin.scala: ########## @@ -0,0 +1,236 @@ +/* + * 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, target DDL, 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.convergenceMaxBatches` + * + * 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 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 defaultNumDistinctKeys: Int = 10 + protected val defaultMaxUniqueEventsPerKey: Int = 40 + protected val defaultNumOutOfOrderBatches: Int = 8 + protected val defaultNumSeedsPerRun: Int = 3 + + // 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.convergenceMaxBatches" + + 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(Random.nextLong()) + } + + private def resolveNumSeeds(): Int = + positiveIntProp(numSeedsSystemProperty, defaultNumSeedsPerRun) + + protected def resolveNumDistinctKeys(): Int = + positiveIntProp(numKeysSystemProperty, defaultNumDistinctKeys) + + protected def resolveMaxUniqueEventsPerKey(): Int = + positiveIntProp(maxEventsPerKeySystemProperty, defaultMaxUniqueEventsPerKey) + + protected def resolveNumOutOfOrderBatches(): Int = + positiveIntProp(numBatchesSystemProperty, defaultNumOutOfOrderBatches) + + /** + * 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 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 + } + + /** + * 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 + ) Review Comment: This fixture can never generate an SCD2 no-op run, so the run-coalescing half of `Scd2BatchProcessor` is unreachable from either convergence suite. `trackHistorySelection` is unset here, so `Scd2BatchProcessor.computeTrackedHistoryColumns` tracks every non-key, non-framework column - including `sequence`. After `dropRedundantRowsPostDecomposition`, no two rows in a key window share an effective `recordStartAt`, and for upsert-representing rows `recordStartAt` equals the `sequence` data column, so any two adjacent upserts differ in a tracked column and `RowClassifier.isNoOpUpsertContinuation` is never true. The `duplicateEventProbability` repeats do not help either: they reuse the same sequence and get dropped as redundant. So the aux table only ever holds tombstones, and hidden no-op upserts, `runStartAt` propagation, the `endAt`-clearing branch of `reconcileStartAndEndAt`, and demotion of a previously visible target row to a hidden aux row are never exercised. Suggest having the generator emit repeats that reuse the previous payload with a new sequence, and passing `trackHistorySelection = Some(ColumnSelection.ExcludeColumns(Seq(UnqualifiedColumnName(sequenceColumn))))` so those repeats actually register as no-ops. The generator change is the one that matters most - `[name, amount, active]` alone only match on consecutive events about 0.5% of the time, which is under one no-op run per generated stream. The cross-SCD property still holds after the change, since a run's visible row is its tail and carries the tail event's own data. -- 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]
