Feature/sql array ref (#92)

* Don't omit fields before the sink to allow foreign keys to use omit fields

* Allow for deeply nested SQL references for arrays and objects, allow for unwrapping top level arrays for JSON output

Author: pflooky

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

@@ -966,6 +966,14 @@ case class FieldBuilder(field: Field = Field()) {
   def omit(omit: Boolean): FieldBuilder =
     this.modify(_.field.options).setTo(getGenBuilder.omit(omit).options)
 
+  /**
+    * Marks this field such that if it is the sole top-level field and is an array in a JSON task, the sink should
+    * output a bare JSON array (no enclosing object with the field name).
+    * Has no effect for non-JSON sinks or non-top-level contexts.
+    */
+  def unwrapTopLevelArray(enable: Boolean): FieldBuilder =
+    this.modify(_.field.options).setTo(getGenBuilder.unwrapTopLevel(enable).options)
+
   /**
    * Sets the primary key flag for the current field.
    *
@@ -1536,6 +1544,12 @@ case class GeneratorBuilder(options: Map[String, Any] = Map()) {
   def omit(omit: Boolean): GeneratorBuilder =
     this.modify(_.options)(_ ++ Map(OMIT -> omit.toString))
 
+  /**
+    * Instruct JSON sink to unwrap the top-level field if it is a single array field.
+    */
+  def unwrapTopLevel(enable: Boolean): GeneratorBuilder =
+    this.modify(_.options)(_ ++ Map(UNWRAP_TOP_LEVEL -> enable.toString))
+
   /**
    * Field is a primary key of the data source.
    *

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

@@ -996,9 +996,25 @@ case class FieldValidationBuilder(validationBuilder: ValidationBuilder = Validat
   private def removeTicksField: String = field.replaceAll("`", "")
 
   private def seqToString(seq: Seq[Any]): String = {
-    seq.head match {
-      case _: String => seq.mkString("'", "','", "'")
-      case _ => seq.mkString(",")
+    // If all values are numeric or numeric-looking strings, render as unquoted numbers to form ARRAY<DOUBLE/NUMERIC>
+    val allNumericLike = seq.forall {
+      case s: String => Try(s.trim.toDouble).isSuccess
+      case n: java.lang.Number => true
+      case n: Number => true
+      case _ => false
+    }
+    if (allNumericLike) {
+      seq.map {
+        case s: String => BigDecimal(s.trim).toString()
+        case n: java.lang.Number => n.toString
+        case n: Number => n.toString
+        case other => other.toString
+      }.mkString(",")
+    } else {
+      seq.head match {
+        case _: String => seq.mkString("'", "','", "'")
+        case _ => seq.mkString(",")
+      }
     }
   }
 }

api/src/main/scala/io/github/datacatering/datacaterer/api/model/Constants.scala

@@ -61,6 +61,9 @@ object Constants {
   lazy val DATA_CONTRACT_FILE = "dataContractFile"
   lazy val DATA_CONTRACT_SCHEMA = "dataContractSchema"
   lazy val ROWS_PER_SECOND = "rowsPerSecond"
+  // When applied on a top-level array field in a JSON task, instructs the sink to output a bare JSON array
+  // instead of an object { fieldName: [...] }.
+  lazy val UNWRAP_TOP_LEVEL = "unwrapTopLevel"
   lazy val HUDI_TABLE_NAME = "hoodie.table.name"
   lazy val ICEBERG_CATALOG_TYPE = "catalogType"
   lazy val ICEBERG_CATALOG_URI = "catalogUri"

app/src/main/scala/io/github/datacatering/datacaterer/core/sink/SinkFactory.scala

@@ -1,7 +1,7 @@
 package io.github.datacatering.datacaterer.core.sink
 
 import com.google.common.util.concurrent.RateLimiter
-import io.github.datacatering.datacaterer.api.model.Constants.{DELTA, DELTA_LAKE_SPARK_CONF, DRIVER, FORMAT, ICEBERG, ICEBERG_SPARK_CONF, JDBC, OMIT, PARTITIONS, PARTITION_BY, POSTGRES_DRIVER, RATE, ROWS_PER_SECOND, SAVE_MODE, TABLE}
+import io.github.datacatering.datacaterer.api.model.Constants.{DELTA, DELTA_LAKE_SPARK_CONF, DRIVER, FORMAT, ICEBERG, ICEBERG_SPARK_CONF, JDBC, JSON, OMIT, PARTITIONS, PARTITION_BY, PATH, POSTGRES_DRIVER, RATE, ROWS_PER_SECOND, SAVE_MODE, TABLE, UNWRAP_TOP_LEVEL}
 import io.github.datacatering.datacaterer.api.model.{FlagsConfig, FoldersConfig, MetadataConfig, SinkResult, Step}
 import io.github.datacatering.datacaterer.api.util.ConfigUtil
 import io.github.datacatering.datacaterer.core.exception.{FailedSaveDataDataFrameV2Exception, FailedSaveDataException}
@@ -91,6 +91,10 @@ class SinkFactory(
       .foreach(conf => df.sqlContext.setConf(conf._1, conf._2))
     val trySaveData = if (format == ICEBERG) {
       Try(tryPartitionAndSaveDfV2(df, saveMode, connectionConfig))
+    } else if (format == JSON) {
+      // Special-case: allow unwrapping top-level array to emit a bare JSON array file
+      val tryMaybeUnwrap = Try(trySaveJsonPossiblyUnwrapped(df, saveMode, connectionConfig))
+      tryMaybeUnwrap
     } else {
       val partitionedDf = partitionDf(df, connectionConfig)
       Try(partitionedDf
@@ -112,6 +116,36 @@ class SinkFactory(
     mapToSinkResult(dataSourceName, df, saveMode, connectionConfig, count, format, trySaveData.isSuccess, startTime, optException)
   }
 
+  private def trySaveJsonPossiblyUnwrapped(df: DataFrame, saveMode: SaveMode, connectionConfig: Map[String, String]): Unit = {
+    val shouldUnwrap = detectTopLevelArrayToUnwrap(df)
+    shouldUnwrap match {
+      case Some(arrayFieldName) =>
+        // Write a single file containing the JSON array string using Spark text writer
+        // We keep directory semantics consistent with other sinks
+        val path = connectionConfig.getOrElse(PATH, throw new IllegalArgumentException("Missing path for JSON sink"))
+        val jsonArrayDf = df.selectExpr(s"TO_JSON(`" + arrayFieldName + "`) AS value").coalesce(1)
+        jsonArrayDf.write.mode(saveMode).text(path)
+      case None =>
+        // Default JSON behavior
+        val partitionedDf = partitionDf(df, connectionConfig)
+        partitionedDf
+          .format(JSON)
+          .mode(saveMode)
+          .options(connectionConfig)
+          .save()
+    }
+  }
+
+  private def detectTopLevelArrayToUnwrap(df: DataFrame): Option[String] = {
+    val fields = df.schema.fields
+    if (fields.length == 1) {
+      val f = fields.head
+      val hasFlag = f.metadata.contains(UNWRAP_TOP_LEVEL) && f.metadata.getString(UNWRAP_TOP_LEVEL).equalsIgnoreCase("true")
+      val isArray = f.dataType.typeName == "array"
+      if (hasFlag && isArray) Some(f.name) else None
+    } else None
+  }
+
   private def partitionDf(df: DataFrame, stepOptions: Map[String, String]): DataFrameWriter[Row] = {
     val partitionDf = stepOptions.get(PARTITIONS)
       .map(partitionNum => df.repartition(partitionNum.toInt)).getOrElse(df)