AnishMahto commented on code in PR #58055: URL: https://github.com/apache/spark/pull/58055#discussion_r3865479713
########## 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: Good idea, done. -- 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]
