feat: Enhance credit card generation and SQL expression handling

- Updated `creditCard` method in `FieldBuilder` to include specific card types in the generated expressions, improving accuracy for Visa, Mastercard, and Amex.
- Refactored `BatchDataProcessor` to ensure proper resource management with a lazy initialization of the `SinkFactory`, enhancing performance and reliability.
- Improved SQL generation in `RandomDataGenerator` to handle credit card patterns more effectively, ensuring correct regex expressions are generated based on card type.
- Added new utility methods for handling unique SQL generation for regex patterns, enhancing the flexibility of data generation.
- Updated tests to validate the new credit card generation logic and SQL expression handling, ensuring expected behavior across various scenarios.

These changes improve the robustness and accuracy of data generation processes, particularly for financial data.

Author: pflooky

api/src/main/scala/io/github/datacatering/datacaterer/api/TaskBuilder.scala

@@ -1376,10 +1376,11 @@ case class FieldBuilder(field: Field = Field()) {
    * }}}
    */
   def creditCard(cardType: String = "visa"): FieldBuilder = {
-    val expr = cardType.toLowerCase match {
-      case "visa" => "#{Finance.creditCard}"
-      case "mastercard" => "#{Finance.creditCard}"
-      case "amex" => "#{Finance.creditCard}"
+    val normalized = Option(cardType).map(_.trim.toLowerCase).getOrElse("")
+    val expr = normalized match {
+      case "visa" => "#{Finance.creditCard 'VISA'}"
+      case "mastercard" => "#{Finance.creditCard 'MASTERCARD'}"
+      case "amex" => "#{Finance.creditCard 'AMEX'}"
       case _ => "#{Finance.creditCard}"
     }
     this.modify(_.field.options).setTo(getGenBuilder.expression(expr).options)

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

@@ -24,36 +24,49 @@ class BatchDataProcessor(connectionConfigsByName: Map[String, Map[String, String
                          metadataConfig: MetadataConfig, flagsConfig: FlagsConfig, generationConfig: GenerationConfig, streamingConfig: StreamingConfig = StreamingConfig())(implicit sparkSession: SparkSession) {
 
   private val LOGGER = Logger.getLogger(getClass.getName)
-  private lazy val sinkFactory = new SinkFactory(flagsConfig, metadataConfig, foldersConfig, streamingConfig)
+  private var sinkFactoryOpt: Option[SinkFactory] = None
+  private def sinkFactory: SinkFactory = {
+    sinkFactoryOpt match {
+      case Some(existing) => existing
+      case None =>
+        val created = new SinkFactory(flagsConfig, metadataConfig, foldersConfig, streamingConfig)
+        sinkFactoryOpt = Some(created)
+        created
+    }
+  }
   private lazy val sinkRouter = new SinkRouter()
   private lazy val recordTrackingProcessor = new RecordTrackingProcessor(foldersConfig.recordTrackingFolderPath)
   private lazy val validationRecordTrackingProcessor = new RecordTrackingProcessor(foldersConfig.recordTrackingForValidationFolderPath)
 
   def splitAndProcess(plan: Plan, executableTasks: List[(TaskSummary, Task)], optValidations: Option[List[ValidationConfiguration]])
                      (implicit sparkSession: SparkSession): (List[DataSourceResult], Option[PerformanceMetrics]) = {
-    val faker = getDataFaker(plan)
-    val dataGeneratorFactory = new DataGeneratorFactory(faker, flagsConfig.enableFastGeneration)
-    val uniqueFieldUtil = new UniqueFieldsUtil(plan, executableTasks, flagsConfig.enableUniqueCheckOnlyInBatch, generationConfig)
-    
-    // Create StepDataCoordinator for data generation
-    val stepDataCoordinator = new StepDataCoordinator(dataGeneratorFactory, uniqueFieldUtil, connectionConfigsByName, flagsConfig)
-    
-    // Create execution strategy
-    val executionStrategy = ExecutionStrategyFactory.create(plan, executableTasks, generationConfig)
-
-    // Route to appropriate execution mode based on strategy
-    executionStrategy.getGenerationMode match {
-      case GenerationMode.Batched =>
-        LOGGER.info("Using batched generation mode")
-        executeBatchedGeneration(plan, executableTasks, optValidations, stepDataCoordinator, executionStrategy)
-      
-      case GenerationMode.AllUpfront =>
-        LOGGER.info("Using all-upfront generation mode for streaming execution")
-        executeAllUpfrontGeneration(plan, executableTasks, optValidations, stepDataCoordinator, executionStrategy)
-      
-      case GenerationMode.Progressive =>
-        LOGGER.warn("Progressive generation mode not yet implemented, falling back to batched mode")
-        executeBatchedGeneration(plan, executableTasks, optValidations, stepDataCoordinator, executionStrategy)
+    try {
+      val faker = getDataFaker(plan)
+      val dataGeneratorFactory = new DataGeneratorFactory(faker, flagsConfig.enableFastGeneration)
+      val uniqueFieldUtil = new UniqueFieldsUtil(plan, executableTasks, flagsConfig.enableUniqueCheckOnlyInBatch, generationConfig)
+
+      // Create StepDataCoordinator for data generation
+      val stepDataCoordinator = new StepDataCoordinator(dataGeneratorFactory, uniqueFieldUtil, connectionConfigsByName, flagsConfig)
+
+      // Create execution strategy
+      val executionStrategy = ExecutionStrategyFactory.create(plan, executableTasks, generationConfig)
+
+      // Route to appropriate execution mode based on strategy
+      executionStrategy.getGenerationMode match {
+        case GenerationMode.Batched =>
+          LOGGER.info("Using batched generation mode")
+          executeBatchedGeneration(plan, executableTasks, optValidations, stepDataCoordinator, executionStrategy)
+
+        case GenerationMode.AllUpfront =>
+          LOGGER.info("Using all-upfront generation mode for streaming execution")
+          executeAllUpfrontGeneration(plan, executableTasks, optValidations, stepDataCoordinator, executionStrategy)
+
+        case GenerationMode.Progressive =>
+          LOGGER.warn("Progressive generation mode not yet implemented, falling back to batched mode")
+          executeBatchedGeneration(plan, executableTasks, optValidations, stepDataCoordinator, executionStrategy)
+      }
+    } finally {
+      shutdownSinkFactory()
     }
   }
 
@@ -274,73 +287,77 @@ class BatchDataProcessor(connectionConfigsByName: Map[String, Map[String, String
     val dataSourcesUsedInValidation = getDataSourcesUsedInValidation(optValidations)
     val pekkoStreamingWriter = new PekkoStreamingSinkWriter(foldersConfig, streamingConfig)
 
-    generatedData.flatMap { case (dataSourceStepName, df) =>
-      val dataSourceName = dataSourceStepName.split("\\.").head
-      val (step, task) = stepAndTaskByDataSourceName(dataSourceStepName)
-      val dataSourceConfig = connectionConfigsByName.getOrElse(dataSourceName, Map())
-      
-      // Skip reference mode steps - they should not be saved to sinks
-      val isReferenceMode = step.options.get(ENABLE_REFERENCE_MODE).map(_.toBoolean).getOrElse(DEFAULT_ENABLE_REFERENCE_MODE)
-      if (isReferenceMode) {
-        LOGGER.debug(s"Skipping save for reference data source, data-source=$dataSourceName, step-name=${step.name}")
-        None
-      } else {
-        val stepWithConfig = step.copy(options = dataSourceConfig ++ step.options)
-        val format = stepWithConfig.options.getOrElse("format", throw new IllegalArgumentException(s"No format specified for $dataSourceName"))
-        
-        // Add rate information to options for router decision
-        val optionsWithRate = executionStrategy match {
-          case dbs: DurationBasedExecutionStrategy if dbs.getTargetRate.isDefined =>
-            stepWithConfig.options + ("hasRateControl" -> "true")
-          case _: PatternBasedExecutionStrategy =>
-            stepWithConfig.options + ("hasRateControl" -> "true")
-          case _ => stepWithConfig.options
-        }
-        
-        // Determine sink strategy using router
-        val sinkStrategy = sinkRouter.determineSinkStrategy(format, executionStrategy.getGenerationMode, optionsWithRate)
-        
-        LOGGER.info(s"Routing to sink, data-source=$dataSourceName, format=$format, strategy=$sinkStrategy")
-        
-        // Apply record tracking if enabled
-        if (flagsConfig.enableRecordTracking) {
-          recordTrackingProcessor.trackRecords(df, dataSourceName, plan.name, stepWithConfig)
-        }
-        if (dataSourcesUsedInValidation.contains(dataSourceName)) {
-          validationRecordTrackingProcessor.trackRecords(df, dataSourceName, plan.name, stepWithConfig)
-        }
-        
-        sinkStrategy match {
-          case SinkStrategy.BatchSink =>
-            // Use standard batch writer
-            val sinkResult = sinkFactory.pushToSink(df, dataSourceName, stepWithConfig, startTime, isMultiBatch = false, isLastBatch = true)
-            Some(DataSourceResult(dataSourceName, task, stepWithConfig, sinkResult, 1))
-
-          case SinkStrategy.StreamingSink =>
-            // Use Pekko streaming writer with rate control
-            val (rate, rateFunction, totalDuration) = executionStrategy match {
-              case dbs: DurationBasedExecutionStrategy =>
-                (dbs.getTargetRate.getOrElse(1), None, None)
-              case pbs: PatternBasedExecutionStrategy =>
-                // For pattern-based, pass rate function for dynamic control
-                val avgRate = pbs.getAverageRate
-                val rateFn = pbs.getRateFunction
-                val duration = pbs.getDurationSeconds
-                LOGGER.info(s"Using pattern-based streaming with dynamic rate control, data-source=$dataSourceName, average-rate=$avgRate/sec, duration=${duration}s")
-                (avgRate, Some(rateFn), Some(duration))
-              case _ => (1, None, None)
-            }
-            if (rateFunction.isEmpty) {
-              LOGGER.info(s"Using streaming sink with rate control, data-source=$dataSourceName, rate=$rate/sec")
-            }
-            val (sinkResult, metrics) = pekkoStreamingWriter.saveWithRateControl(
-              dataSourceName, df, format, dataSourceConfig, stepWithConfig, rate, startTime, rateFunction, totalDuration
-            )
-
-            // Create result with performance metrics
-            Some(DataSourceResult(dataSourceName, task, stepWithConfig, sinkResult, 1, metrics))
+    try {
+      generatedData.flatMap { case (dataSourceStepName, df) =>
+        val dataSourceName = dataSourceStepName.split("\\.").head
+        val (step, task) = stepAndTaskByDataSourceName(dataSourceStepName)
+        val dataSourceConfig = connectionConfigsByName.getOrElse(dataSourceName, Map())
+
+        // Skip reference mode steps - they should not be saved to sinks
+        val isReferenceMode = step.options.get(ENABLE_REFERENCE_MODE).map(_.toBoolean).getOrElse(DEFAULT_ENABLE_REFERENCE_MODE)
+        if (isReferenceMode) {
+          LOGGER.debug(s"Skipping save for reference data source, data-source=$dataSourceName, step-name=${step.name}")
+          None
+        } else {
+          val stepWithConfig = step.copy(options = dataSourceConfig ++ step.options)
+          val format = stepWithConfig.options.getOrElse("format", throw new IllegalArgumentException(s"No format specified for $dataSourceName"))
+
+          // Add rate information to options for router decision
+          val optionsWithRate = executionStrategy match {
+            case dbs: DurationBasedExecutionStrategy if dbs.getTargetRate.isDefined =>
+              stepWithConfig.options + ("hasRateControl" -> "true")
+            case _: PatternBasedExecutionStrategy =>
+              stepWithConfig.options + ("hasRateControl" -> "true")
+            case _ => stepWithConfig.options
+          }
+
+          // Determine sink strategy using router
+          val sinkStrategy = sinkRouter.determineSinkStrategy(format, executionStrategy.getGenerationMode, optionsWithRate)
+
+          LOGGER.info(s"Routing to sink, data-source=$dataSourceName, format=$format, strategy=$sinkStrategy")
+
+          // Apply record tracking if enabled
+          if (flagsConfig.enableRecordTracking) {
+            recordTrackingProcessor.trackRecords(df, dataSourceName, plan.name, stepWithConfig)
+          }
+          if (dataSourcesUsedInValidation.contains(dataSourceName)) {
+            validationRecordTrackingProcessor.trackRecords(df, dataSourceName, plan.name, stepWithConfig)
+          }
+
+          sinkStrategy match {
+            case SinkStrategy.BatchSink =>
+              // Use standard batch writer
+              val sinkResult = sinkFactory.pushToSink(df, dataSourceName, stepWithConfig, startTime, isMultiBatch = false, isLastBatch = true)
+              Some(DataSourceResult(dataSourceName, task, stepWithConfig, sinkResult, 1))
+
+            case SinkStrategy.StreamingSink =>
+              // Use Pekko streaming writer with rate control
+              val (rate, rateFunction, totalDuration) = executionStrategy match {
+                case dbs: DurationBasedExecutionStrategy =>
+                  (dbs.getTargetRate.getOrElse(1), None, None)
+                case pbs: PatternBasedExecutionStrategy =>
+                  // For pattern-based, pass rate function for dynamic control
+                  val avgRate = pbs.getAverageRate
+                  val rateFn = pbs.getRateFunction
+                  val duration = pbs.getDurationSeconds
+                  LOGGER.info(s"Using pattern-based streaming with dynamic rate control, data-source=$dataSourceName, average-rate=$avgRate/sec, duration=${duration}s")
+                  (avgRate, Some(rateFn), Some(duration))
+                case _ => (1, None, None)
+              }
+              if (rateFunction.isEmpty) {
+                LOGGER.info(s"Using streaming sink with rate control, data-source=$dataSourceName, rate=$rate/sec")
+              }
+              val (sinkResult, metrics) = pekkoStreamingWriter.saveWithRateControl(
+                dataSourceName, df, format, dataSourceConfig, stepWithConfig, rate, startTime, rateFunction, totalDuration
+              )
+
+              // Create result with performance metrics
+              Some(DataSourceResult(dataSourceName, task, stepWithConfig, sinkResult, 1, metrics))
+          }
         }
       }
+    } finally {
+      pekkoStreamingWriter.shutdown()
     }
   }
 
@@ -446,5 +463,10 @@ class BatchDataProcessor(connectionConfigsByName: Map[String, Map[String, String
       case _ => new Faker() with Serializable
     }
   }
+
+  private def shutdownSinkFactory(): Unit = {
+    sinkFactoryOpt.foreach(_.shutdown())
+    sinkFactoryOpt = None
+  }
 }
 

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

@@ -71,7 +71,8 @@ object FastDataGenerator {
    * Falls back to UDF-based generation for unsupported patterns.
    */
   class FastRegexDataGenerator(val structField: StructField, val faker: Faker = new Faker()) extends NullableDataGenerator[String] {
-    import io.github.datacatering.datacaterer.core.generator.provider.regex.RegexPatternParser
+    import io.github.datacatering.datacaterer.api.model.Constants.{INDEX_INC_FIELD, IS_UNIQUE}
+    import io.github.datacatering.datacaterer.core.generator.provider.regex.{RegexNode, CharacterClassNode, Digit, LiteralNode, RegexPatternParser, SequenceNode}
     import io.github.datacatering.datacaterer.core.model.Constants.GENERATE_REGEX_UDF
     import org.apache.log4j.Logger
 
@@ -81,6 +82,8 @@ object FastDataGenerator {
 
     // Parse the regex pattern once during initialization
     private val parsedPattern = RegexPatternParser.parse(regex)
+    private val isUniqueRegex = structField.metadata.contains(IS_UNIQUE) &&
+      structField.metadata.getString(IS_UNIQUE).toBoolean
 
     // Log parsing result for debugging
     parsedPattern match {
@@ -98,11 +101,23 @@ object FastDataGenerator {
       faker.regexify(regex)
     }
 
+    private def uniqueSqlForPattern(node: RegexNode): Option[String] = node match {
+      case SequenceNode(List(LiteralNode(prefix), CharacterClassNode(Digit, min, max)))
+        if min == max =>
+        Some(s"CONCAT('$prefix', LPAD(CAST($INDEX_INC_FIELD AS BIGINT), $min, '0'))")
+      case CharacterClassNode(Digit, min, max) if min == max =>
+        Some(s"LPAD(CAST($INDEX_INC_FIELD AS BIGINT), $min, '0')")
+      case _ => None
+    }
+
     override def generateSqlExpression: String = {
       parsedPattern match {
         case scala.util.Success(node) =>
-          // Successfully parsed - use pure SQL generation
-          node.toSql(sqlRandom, sqlRandomWithIndex)
+          if (isUniqueRegex) {
+            uniqueSqlForPattern(node).getOrElse(node.toSql(sqlRandom, sqlRandomWithIndex))
+          } else {
+            node.toSql(sqlRandom, sqlRandomWithIndex)
+          }
 
         case scala.util.Failure(_) =>
           // Parser failed - fall back to UDF (slower but correct)