fix: handle partition overwrite

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

Author: tchow-zlai

cloud_gcp/src/main/scala/ai/chronon/integrations/cloud_gcp/BigQueryFormat.scala

@@ -41,15 +41,20 @@ case class GcpFormatProvider(sparkSession: SparkSession) extends FormatProvider
   override def writeFormat(tableName: String): Format = {
 
     val tu = TableUtils(sparkSession)
+    val isTableReachable = tu.tableReachable(tableName)
     val sparkOptions: Map[String, String] =
       Map(
-        "partitionField" -> tu.partitionColumn,
         "temporaryGcsBucket" -> sparkSession.conf.get(
           "spark.chronon.table.gcs.temporary_gcs_bucket"
         ), // todo(tchow): No longer needed after https://github.com/GoogleCloudDataproc/spark-bigquery-connector/pull/1320
         "writeMethod" -> "indirect"
       )
-    BQuery(bqOptions.getProjectId, sparkOptions)
+
+    if (isTableReachable) {
+      BQuery(bqOptions.getProjectId, sparkOptions + ("partitionField" -> tu.partitionColumn))
+    } else {
+      BQuery(bqOptions.getProjectId, sparkOptions)
+    }
   }
 
   private def format(tableName: String): Format = {