wrap with config

Co-authored-by: Thomas Chow <[email protected]>

Author: tchow-zlai

spark/src/main/scala/ai/chronon/spark/TableUtils.scala

@@ -482,7 +482,10 @@ class TableUtils(@transient val sparkSession: SparkSession) extends Serializable
     wrapWithCache(s"repartition & write to $tableName", df) {
       logger.info("Repartitioning before writing...")
       val dataPointer = DataPointer.from(tableName, sparkSession)
-      val repartitioned = repartitionInternal(df, tableName, stats, sortByCols)
+      val repartitioned =
+        if (sparkSession.conf.get("spark.chronon.write.repartition", "false").toBoolean)
+          repartitionInternal(df, tableName, stats, sortByCols)
+        else df
       repartitioned
         .select(repartitioned.columns.map {
           case c if c == partitionColumn && dataPointer.writeFormat.map(_.toUpperCase).exists("BIGQUERY".equals) =>