feat: refactor for alter tables

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

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

Author: tchow-zlai

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

@@ -18,6 +18,11 @@ case class BigQueryFormat(project: String, bqClient: BigQuery, override val opti
     extends Format {
   override def name: String = "bigquery"
 
+  override def alterTableProperties(tableName: String,
+                                    tableProperties: Map[String, String]): (String => Unit) => Unit = {
+    throw new NotImplementedError("alterTableProperties not yet supported for BigQuery")
+  }
+
   override def primaryPartitions(tableName: String, partitionColumn: String, subPartitionsFilter: Map[String, String])(
       implicit sparkSession: SparkSession): Seq[String] =
     super.primaryPartitions(tableName, partitionColumn, subPartitionsFilter)

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

@@ -311,7 +311,8 @@ class TableUtils(@transient val sparkSession: SparkSession) extends Serializable
     //  https://app.asana.com/0/1208949807589885/1209111629687568/f
     if (writeFormat.name.toUpperCase != "BIGQUERY") {
       if (tableProperties != null && tableProperties.nonEmpty) {
-        sql(alterTablePropertiesSql(tableName, tableProperties))
+        val alterTblPropsOperation = writeFormat.alterTableProperties(tableName, tableProperties)
+        alterTblPropsOperation(sql)
       }
       if (autoExpand) {
         expandTable(tableName, df.schema)

spark/src/main/scala/ai/chronon/spark/format/Format.scala

@@ -1,5 +1,6 @@
 package ai.chronon.spark.format
 
+import ai.chronon.spark.format.CreationUtils.alterTablePropertiesSql
 import ai.chronon.spark.format.CreationUtils.createTableSql
 import org.apache.spark.sql.DataFrame
 import org.apache.spark.sql.SparkSession
@@ -73,6 +74,20 @@ trait Format {
 
   }
 
+  def alterTableProperties(tableName: String, tableProperties: Map[String, String]): (String => Unit) => Unit = {
+
+    def inner(tableName: String, tableProperties: Map[String, String])(sqlEvaluator: String => Unit) = {
+      val creationSql =
+        alterTablePropertiesSql(tableName, tableProperties)
+
+      sqlEvaluator(creationSql)
+
+    }
+
+    inner(tableName, tableProperties)
+
+  }
+
   // Help specify the appropriate table type to use in the Spark create table DDL query
   def createTableTypeString: String