feat: Enhance SQL generation with seeded randomness and improved determinism

- Introduced new methods for generating SQL expressions with seeded randomness in various data generators, ensuring consistent and varied outputs.
- Updated `DataGeneratorFactory` to utilize new random expression methods for weight calculations.
- Refactored SQL generation logic in `RandomDataGenerator`, `OneOfDataGenerator`, and `RegexNode` to support indexed random values.
- Added tests to verify deterministic behavior of seeded generators, ensuring expected outputs across multiple runs.
- Enhanced `DataGeneratorDeterminismTest` to validate the consistency and variability of generated values with seeded configurations.

These changes improve the reliability of data generation processes, particularly in scenarios requiring reproducible results.

Author: pflooky

app/src/integrationTest/scala/io/github/datacatering/datacaterer/core/generator/EnhancedForeignKeyIntegrationTest.scala

@@ -461,14 +461,14 @@ class EnhancedForeignKeyIntegrationTest extends SparkSuite with Matchers with Be
   class RatioCardinalityTestPlan(customersPath: String, ordersPath: String) extends PlanRun {
     val customers = csv("customers", customersPath, Map("saveMode" -> "overwrite", "header" -> "true"))
       .fields(
-        field.name("customer_id").regex("CUST[0-9]{6}"),
+        field.name("customer_id").regex("CUST[0-9]{10}"),
         field.name("name").expression("#{Name.name}")
       )
       .count(count.records(50))
 
     val orders = csv("orders", ordersPath, Map("saveMode" -> "overwrite", "header" -> "true"))
       .fields(
-        field.name("order_id").regex("ORD[0-9]{8}"),
+        field.name("order_id").regex("ORD[0-9]{10}"),
         field.name("customer_id"),
         field.name("amount").`type`(DoubleType).min(10.0).max(1000.0)
       )

app/src/main/scala/io/github/datacatering/datacaterer/core/generator/DataGeneratorFactory.scala

@@ -130,7 +130,9 @@ class DataGeneratorFactory(faker: Faker, enableFastGeneration: Boolean = false)(
     val namedStruct = dataGenerators.map(dg => s"'${dg.structField.name}', CAST(${dg.generateSqlExpressionWrapper} AS ${dg.structField.dataType.sql})").mkString(",")
     //if it is using a weighted oneOf generator, it will have a weight column
     val countGeneratorSql = countGenerator.generateSqlExpressionWrapper
-    val optWeightCol = if (countGeneratorSql.contains(RECORD_COUNT_GENERATOR_WEIGHT_FIELD)) Array(s"RAND() AS $RECORD_COUNT_GENERATOR_WEIGHT_FIELD") else Array[String]()
+    val optWeightCol = if (countGeneratorSql.contains(RECORD_COUNT_GENERATOR_WEIGHT_FIELD)) {
+      Array(s"${countGenerator.sqlRandomWithOffset(1)} AS $RECORD_COUNT_GENERATOR_WEIGHT_FIELD")
+    } else Array[String]()
     val perCountGeneratedExpr = df.columns ++ optWeightCol ++ Array(
       s"CAST($countGeneratorSql AS INT) AS $PER_FIELD_COUNT_GENERATED_NUM",
       s"CASE WHEN $PER_FIELD_COUNT_GENERATED_NUM = 0 THEN ARRAY() ELSE SEQUENCE(1, $PER_FIELD_COUNT_GENERATED_NUM) END AS $PER_FIELD_COUNT_GENERATED"

app/src/main/scala/io/github/datacatering/datacaterer/core/generator/provider/DataGenerator.scala

@@ -21,6 +21,12 @@ trait DataGenerator[T] extends BaseGenerator[T] with Serializable {
 
   lazy val optRandomSeed: Option[Long] = if (structField.metadata.contains(RANDOM_SEED)) Some(structField.metadata.getString(RANDOM_SEED).toLong) else None
   lazy val sqlRandom: String = optRandomSeed.map(seed => s"RAND($seed)").getOrElse("RAND()")
+  def sqlRandomWithOffset(offset: Long): String = optRandomSeed.map(seed => s"RAND(${seed + offset})").getOrElse("RAND()")
+  def sqlRandomWithIndex(indexExpr: String): String = optRandomSeed match {
+    case Some(seed) =>
+      s"(CAST((xxhash64(monotonically_increasing_id(), $seed, $indexExpr) & ${Long.MaxValue}) AS DOUBLE) / ${Long.MaxValue.toDouble})"
+    case None => "RAND()"
+  }
   lazy val random: Random = if (structField.metadata.contains(RANDOM_SEED)) new Random(structField.metadata.getString(RANDOM_SEED).toLong) else new Random()
   lazy val enabledNull: Boolean = if (structField.metadata.contains(ENABLED_NULL)) structField.metadata.getString(ENABLED_NULL).toBoolean else false
   lazy val enabledEdgeCases: Boolean = if (structField.metadata.contains(ENABLED_EDGE_CASE)) structField.metadata.getString(ENABLED_EDGE_CASE).toBoolean else false

app/src/main/scala/io/github/datacatering/datacaterer/core/generator/provider/FastDataGenerator.scala

@@ -102,7 +102,7 @@ object FastDataGenerator {
       parsedPattern match {
         case scala.util.Success(node) =>
           // Successfully parsed - use pure SQL generation
-          node.toSql
+          node.toSql(sqlRandom, sqlRandomWithIndex)
 
         case scala.util.Failure(_) =>
           // Parser failed - fall back to UDF (slower but correct)

app/src/main/scala/io/github/datacatering/datacaterer/core/generator/provider/OneOfDataGenerator.scala

@@ -27,7 +27,7 @@ object OneOfDataGenerator {
 
     override def generateSqlExpression: String = {
       val oneOfValuesString = oneOfValues.mkString("||")
-      s"CAST(SPLIT('$oneOfValuesString', '\\\\|\\\\|')[CAST(RAND() * $oneOfArrayLength AS INT)] AS ${structField.dataType.sql})"
+      s"CAST(SPLIT('$oneOfValuesString', '\\\\|\\\\|')[CAST($sqlRandom * $oneOfArrayLength AS INT)] AS ${structField.dataType.sql})"
     }
   }
 

app/src/main/scala/io/github/datacatering/datacaterer/core/generator/provider/RandomDataGenerator.scala

@@ -220,9 +220,9 @@ object RandomDataGenerator {
         val numCompleteWeeks = s"CAST($remainingDays / 7 AS INT)"
 
         // Generate random week (0 to numCompleteWeeks-1) and weekday (0-4)
-        // Use separate RAND() calls for week and weekday to ensure proper independence
+        // Use separate seeded RAND calls for week and weekday to ensure proper independence
         val randWeek = s"CAST($sqlRandom * $numCompleteWeeks AS INT)"
-        val randWeekday = s"CAST(RAND() * 5 AS INT)"
+        val randWeekday = s"CAST(${sqlRandomWithOffset(1)} * 5 AS INT)"
 
         s"DATE_ADD($firstMonday, $randWeek * 7 + $randWeekday)"
       } else {
@@ -291,7 +291,7 @@ object RandomDataGenerator {
     }
 
     override def generateSqlExpression: String = {
-      s"TO_BINARY(ARRAY_JOIN(TRANSFORM(ARRAY_REPEAT(1, CAST($sqlRandom * ${maxLength - minLength} + $minLength AS INT)), i -> CHAR(ROUND($sqlRandom * 94 + 32, 0))), ''), 'utf-8')"
+      s"TO_BINARY(ARRAY_JOIN(TRANSFORM(ARRAY_REPEAT(1, CAST($sqlRandom * ${maxLength - minLength} + $minLength AS INT)), i -> CHAR(ROUND(${sqlRandomWithIndex("i")} * 94 + 32, 0))), ''), 'utf-8')"
     }
   }
 
@@ -303,7 +303,7 @@ object RandomDataGenerator {
     }
 
     override def generateSqlExpression: String = {
-      s"TO_BINARY(CHAR(ROUND($sqlRandom * 94 + 32, 0)))"
+      s"TO_BINARY(CHAR(ROUND($sqlRandom * 94 + 32, 0)), 'utf-8')"
     }
   }
 
@@ -385,7 +385,7 @@ object RandomDataGenerator {
       val arraySize = valuesStr.split(",").length
       val randomIndexExpr = s"CAST($sqlRandom * $arraySize + 1 AS INT)"
       val sizeExpr = s"CAST($sqlRandom * ${arrayMaxSize - arrayMinSize} + $arrayMinSize AS INT)"
-      val arrayExpr = s"TRANSFORM(ARRAY_REPEAT(1, $sizeExpr), i -> ELEMENT_AT($valuesArray, CAST(RAND() * $arraySize + 1 AS INT)))"
+      val arrayExpr = s"TRANSFORM(ARRAY_REPEAT(1, $sizeExpr), i -> ELEMENT_AT($valuesArray, CAST(${sqlRandomWithIndex("i")} * $arraySize + 1 AS INT)))"
 
       applyEmptyProbability(arrayExpr)
     }
@@ -406,22 +406,25 @@ object RandomDataGenerator {
 
       // Calculate cumulative weights
       val totalWeight = weightedPairs.map(_._2).sum
+      require(totalWeight > 0,
+        s"Invalid weights in field '${structField.name}': total weight must be greater than 0.")
       val normalizedWeights = weightedPairs.map { case (v, w) => (v, w / totalWeight) }
-
-      // Build CASE WHEN statement for weighted selection
-      // Use RAND() directly in each WHEN clause to avoid nested subquery
-      var cumulativeProb = 0.0
-      val caseStatements = normalizedWeights.zipWithIndex.map { case ((value, weight), idx) =>
-        val prevCumulativeProb = cumulativeProb
-        cumulativeProb += weight
-        if (idx == normalizedWeights.length - 1) {
-          s"ELSE $value"
-        } else {
-          s"WHEN RAND() < $cumulativeProb THEN $value"
-        }
-      }.mkString(" ")
-
-      val weightedSelectExpr = s"(CASE $caseStatements END)"
+      val cumulativeThresholds = normalizedWeights.scanLeft(0.0)(_ + _._2).tail
+
+      val valuesArrayExpr = s"ARRAY(${normalizedWeights.map(_._1).mkString(",")})"
+      val thresholdsArrayExpr = s"ARRAY(${cumulativeThresholds.map(_.toString).mkString(",")})"
+      val zippedExpr = s"ZIP_WITH($valuesArrayExpr, $thresholdsArrayExpr, (v, t) -> named_struct('value', v, 'threshold', t))"
+
+      // Use a single RAND() per element to compare against cumulative thresholds
+      val weightedSelectExpr =
+        s"""AGGREGATE(
+           |  $zippedExpr,
+           |  named_struct('r', ${sqlRandomWithIndex("i")}, 'picked', CAST(NULL AS ${dataType.sql})),
+           |  (acc, x) -> IF(acc.picked IS NOT NULL, acc,
+           |    IF(acc.r < x.threshold, named_struct('r', acc.r, 'picked', x.value), acc)
+           |  ),
+           |  acc -> acc.picked
+           |)""".stripMargin
       val sizeExpr = s"CAST($sqlRandom * ${arrayMaxSize - arrayMinSize} + $arrayMinSize AS INT)"
       val arrayExpr = s"TRANSFORM(ARRAY_REPEAT(1, $sizeExpr), i -> $weightedSelectExpr)"
 
@@ -442,17 +445,28 @@ object RandomDataGenerator {
     assert(mapMinSize <= mapMaxSize, s"mapMinSize has to be less than or equal to mapMaxSize, " +
       s"field-name=${structField.name}, mapMinSize=$mapMinSize, mapMaxSize=$mapMaxSize")
 
-    override def keyGenerator: DataGenerator[T] = getGeneratorForDataType(keyDataType).asInstanceOf[DataGenerator[T]]
+    private def seededMetadata: Metadata = optRandomSeed match {
+      case Some(seed) => new MetadataBuilder().putString(RANDOM_SEED, seed.toString).build()
+      case None => Metadata.empty
+    }
+
+    override def keyGenerator: DataGenerator[T] =
+      getGeneratorForStructField(StructField(structField.name, keyDataType, nullable = true, seededMetadata), faker)
+        .asInstanceOf[DataGenerator[T]]
 
-    override def valueGenerator: DataGenerator[K] = getGeneratorForDataType(valueDataType).asInstanceOf[DataGenerator[K]]
+    override def valueGenerator: DataGenerator[K] =
+      getGeneratorForStructField(StructField(structField.name, valueDataType, nullable = true, seededMetadata), faker)
+        .asInstanceOf[DataGenerator[K]]
 
     //how to make it empty map when size is 0
     override def generateSqlExpression: String = {
-      val keyDataGenerator = getGeneratorForDataType(keyDataType)
-      val valueDataGenerator = getGeneratorForDataType(valueDataType)
+      val keyDataGenerator = getGeneratorForStructField(StructField(structField.name, keyDataType, nullable = true, seededMetadata), faker)
+      val valueDataGenerator = getGeneratorForStructField(StructField(structField.name, valueDataType, nullable = true, seededMetadata), faker)
       val keySql = keyDataGenerator.generateSqlExpressionWrapper
       val valueSql = valueDataGenerator.generateSqlExpressionWrapper
-      s"STR_TO_MAP(CONCAT_WS(',', TRANSFORM(ARRAY_REPEAT(1, CAST($sqlRandom * ${mapMaxSize - mapMinSize} + $mapMinSize AS INT)), i -> CONCAT($keySql, '->', $valueSql))), '->', ',')"
+      val keySqlWithIndex = optRandomSeed.map(_ => keySql.replace(sqlRandom, sqlRandomWithIndex("i"))).getOrElse(keySql)
+      val valueSqlWithIndex = optRandomSeed.map(_ => valueSql.replace(sqlRandom, sqlRandomWithIndex("i"))).getOrElse(valueSql)
+      s"STR_TO_MAP(CONCAT_WS(',', TRANSFORM(ARRAY_REPEAT(1, CAST($sqlRandom * ${mapMaxSize - mapMinSize} + $mapMinSize AS INT)), i -> CONCAT($keySqlWithIndex, '->', $valueSqlWithIndex))), '->', ',')"
     }
   }