refactor tableutils into its own module catalog (#679)

## Summary

- Creates a `catalog` module to replace the `format` one we had. 
- Add a new bazel lib target for the `catalog` module
- Move TableUtils there, it should live alongside the formatprovider
stuff.
- Rename imports
- Followup is to remove the dependency on the main `spark` lib. This
requires decoupling with `Extensions`. Then we can drastically slim the
deps down in cloud gcp.

## Checklist
- [ ] Added Unit Tests
- [ ] Covered by existing CI
- [ ] Integration tested
- [ ] Documentation update


<!-- av pr metadata
This information is embedded by the av CLI when creating PRs to track
the status of stacks when using Aviator. Please do not delete or edit
this section of the PR.
```
{"parent":"main","parentHead":"","trunk":""}
```
-->

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

Author: tchow-zlai

cloud_aws/src/test/scala/ai/chronon/integrations/aws/GlueCatalogTest.scala

@@ -2,7 +2,7 @@ package ai.chronon.integrations.aws
 
 import ai.chronon.spark.ChrononHudiKryoRegistrator
 import ai.chronon.spark.SparkSessionBuilder
-import ai.chronon.spark.TableUtils
+import ai.chronon.spark.catalog.TableUtils
 import org.apache.spark.sql.SaveMode
 import org.apache.spark.sql.SparkSession
 import org.junit.Assert.assertEquals

cloud_aws/src/test/scala/ai/chronon/integrations/aws/HudiTableUtilsTest.scala

@@ -2,7 +2,7 @@ package ai.chronon.integrations.aws
 
 import ai.chronon.spark.ChrononHudiKryoRegistrator
 import ai.chronon.spark.SparkSessionBuilder
-import ai.chronon.spark.TableUtils
+import ai.chronon.spark.catalog.TableUtils
 import org.apache.spark.sql.SparkSession
 import org.junit.Assert.assertEquals
 import org.junit.Assert.assertTrue

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

@@ -1,6 +1,6 @@
 package ai.chronon.integrations.cloud_gcp
 
-import ai.chronon.spark.format.Format
+import ai.chronon.spark.catalog.Format
 import com.google.cloud.bigquery._
 import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
 import org.apache.spark.sql.execution.FileSourceScanExec

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

@@ -1,7 +1,7 @@
 package ai.chronon.integrations.cloud_gcp
 
-import ai.chronon.spark.TableUtils
-import ai.chronon.spark.format.Format
+import ai.chronon.spark.catalog.TableUtils
+import ai.chronon.spark.catalog.Format
 import com.google.cloud.bigquery.BigQueryOptions
 import com.google.cloud.spark.bigquery.v2.Spark35BigQueryTableProvider
 import org.apache.spark.sql.{DataFrame, SparkSession}

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

@@ -1,5 +1,6 @@
 package ai.chronon.integrations.cloud_gcp
-import ai.chronon.spark.format.{DefaultFormatProvider, Format, Iceberg}
+
+import ai.chronon.spark.catalog.{DefaultFormatProvider, Format, Iceberg}
 import com.google.cloud.bigquery._
 import com.google.cloud.spark.bigquery.v2.Spark31BigQueryTable
 import org.apache.iceberg.gcp.bigquery.BigQueryMetastoreCatalog

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

@@ -6,7 +6,7 @@ import ai.chronon.api.MetaData
 import ai.chronon.integrations.cloud_gcp.BigTableKVStore.ColumnFamilyQualifierString
 import ai.chronon.integrations.cloud_gcp.BigTableKVStore.ColumnFamilyString
 import ai.chronon.spark.SparkSessionBuilder
-import ai.chronon.spark.TableUtils
+import ai.chronon.spark.catalog.TableUtils
 import org.apache.spark.sql.functions
 import org.apache.spark.sql.functions.udf
 import org.rogach.scallop.ScallopConf

cloud_gcp/src/test/scala/ai/chronon/integrations/cloud_gcp/BigQueryCatalogTest.scala

@@ -1,7 +1,7 @@
 package ai.chronon.integrations.cloud_gcp
 
-import ai.chronon.spark.format.{FormatProvider, Iceberg}
-import ai.chronon.spark.{SparkSessionBuilder, TableUtils}
+import ai.chronon.spark.SparkSessionBuilder
+import ai.chronon.spark.catalog.{FormatProvider, Iceberg, TableUtils}
 import com.esotericsoftware.kryo.Kryo
 import com.esotericsoftware.kryo.io.{Input, Output}
 import com.google.cloud.hadoop.fs.gcs.{

spark/BUILD.bazel

@@ -49,6 +49,27 @@ scala_library(
     ],
 )
 
+scala_library(
+    name = "catalog_lib",
+    srcs = glob(["src/main/scala/ai/chronon/spark/catalog/*.scala"]),
+    format = select({
+        "//tools/config:scala_2_13": False,
+        "//conditions:default": True,
+    }),
+    visibility = ["//visibility:public"],
+    deps = [
+        ":lib",
+        "//api:lib",
+        "//api:thrift_java",
+        "//tools/build_rules/spark:spark-exec",
+        maven_artifact("org.slf4j:slf4j-api"),
+        maven_artifact("org.apache.thrift:libthrift"),
+        maven_artifact("org.apache.logging.log4j:log4j-api"),
+        maven_artifact("org.apache.logging.log4j:log4j-core"),
+        maven_artifact_with_suffix("io.delta:delta-spark"),
+    ],
+)
+
 scala_library(
     name = "submission_lib",
     srcs = glob(["src/main/scala/ai/chronon/spark/submission/*.scala"]),

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

@@ -33,6 +33,7 @@ import org.apache.spark.sql.functions._
 import org.apache.spark.sql.{DataFrame, Row, types}
 import org.apache.spark.sql.types.{StringType, StructType}
 import org.slf4j.{Logger, LoggerFactory}
+import ai.chronon.spark.catalog.TableUtils
 
 import scala.collection.{Seq, immutable, mutable}
 import scala.collection.mutable.ListBuffer

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

@@ -26,6 +26,7 @@ import org.apache.spark.sql.Row
 import org.apache.spark.sql.functions.expr
 import org.apache.spark.sql.types.{StringType, StructType}
 import org.slf4j.{Logger, LoggerFactory}
+import ai.chronon.spark.catalog.TableUtils
 
 import scala.collection.{Seq, immutable, mutable}
 import scala.util.Try

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

@@ -26,7 +26,7 @@ import ai.chronon.online.fetcher.{ConfPathOrName, FetchContext, FetcherMain, Met
 import ai.chronon.online.{Api, MetadataDirWalker, MetadataEndPoint, TopicChecker}
 import ai.chronon.orchestration.{JoinMergeNode, JoinPartNode}
 import ai.chronon.spark.batch._
-import ai.chronon.spark.format.Format
+import ai.chronon.spark.catalog.{Format, TableUtils}
 import ai.chronon.spark.stats.drift.{Summarizer, SummaryPacker, SummaryUploader}
 import ai.chronon.spark.stats.{CompareBaseJob, CompareJob, ConsistencyJob}
 import ai.chronon.spark.streaming.JoinSourceRunner

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

@@ -28,6 +28,7 @@ import org.apache.spark.sql.functions._
 import org.apache.spark.sql.types.{LongType, StructType}
 import org.apache.spark.util.sketch.BloomFilter
 import org.slf4j.{Logger, LoggerFactory}
+import ai.chronon.spark.catalog.TableUtils
 
 import java.util
 import scala.collection.Seq

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

@@ -31,6 +31,7 @@ import ai.chronon.api.{
   TsUtils,
   TimeRange
 }
+import ai.chronon.spark.catalog.TableUtils
 import ai.chronon.api.DataModel.ENTITIES
 import ai.chronon.api.DataModel.EVENTS
 import ai.chronon.api.Extensions._

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

@@ -21,6 +21,7 @@ import ai.chronon.aggregator.windowing.FiveMinuteResolution
 import ai.chronon.aggregator.windowing.Resolution
 import ai.chronon.aggregator.windowing.SawtoothOnlineAggregator
 import ai.chronon.api
+import ai.chronon.spark.catalog.TableUtils
 import ai.chronon.api.Accuracy
 import ai.chronon.api.Constants
 import ai.chronon.api.DataModel

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

@@ -21,6 +21,7 @@ import ai.chronon.api.DataModel.ENTITIES
 import ai.chronon.api.Extensions._
 import ai.chronon.api.ScalaJavaConversions._
 import ai.chronon.api._
+import ai.chronon.spark.catalog.TableUtils
 import ai.chronon.online.serde.SparkConversions
 import ai.chronon.orchestration.{JoinBootstrapNode, JoinPartNode}
 import ai.chronon.spark.Extensions._

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

@@ -20,6 +20,7 @@ import ai.chronon.api
 import ai.chronon.api.DataModel.ENTITIES
 import ai.chronon.api.Extensions._
 import ai.chronon.api.ScalaJavaConversions._
+import ai.chronon.spark.catalog.TableUtils
 import ai.chronon.api.{Accuracy, Constants, DateRange, JoinPart, PartitionRange, PartitionSpec}
 import ai.chronon.online.metrics.Metrics
 import ai.chronon.orchestration.JoinBootstrapNode

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

@@ -3,6 +3,7 @@ package ai.chronon.spark
 import ai.chronon.api.Extensions._
 import ai.chronon.api.ScalaJavaConversions.ListOps
 import ai.chronon.api.DateRange
+import ai.chronon.spark.catalog.TableUtils
 import ai.chronon.orchestration.JoinDerivationNode
 import ai.chronon.spark.Extensions._
 import org.apache.spark.sql.functions.{coalesce, col, expr}

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

@@ -23,6 +23,7 @@ import ai.chronon.api.ScalaJavaConversions._
 import ai.chronon.api._
 import ai.chronon.spark.Extensions._
 import com.google.gson.Gson
+import ai.chronon.spark.catalog.TableUtils
 import org.apache.spark.sql
 import org.apache.spark.sql.DataFrame
 import org.apache.spark.sql.expressions.UserDefinedFunction

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

@@ -21,6 +21,7 @@ import ai.chronon.api.{Builders, Constants, JoinPart, PartitionSpec, TimeUnit, W
 import ai.chronon.api.DataModel.ENTITIES
 import ai.chronon.api.DataModel.EVENTS
 import ai.chronon.api.Extensions._
+import ai.chronon.spark.catalog.TableUtils
 import ai.chronon.api.PartitionRange
 import ai.chronon.online.metrics.Metrics
 import ai.chronon.spark.Extensions._

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

@@ -19,7 +19,7 @@ package ai.chronon.spark
 import com.google.common.io.Files
 import org.apache.commons.io.FileUtils
 import org.apache.spark.sql.SaveMode
-
+import ai.chronon.spark.catalog.TableUtils
 import java.io.File
 
 object LocalTableExporter {

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

@@ -25,6 +25,7 @@ import ai.chronon.online._
 import ai.chronon.online.metrics._
 import ai.chronon.online.serde._
 import ai.chronon.spark.Extensions.{StructTypeOps, _}
+import ai.chronon.spark.catalog.TableUtils
 import org.apache.spark.rdd.RDD
 import org.apache.spark.sql.catalyst.expressions.GenericRow
 import org.apache.spark.sql.functions.col

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

@@ -18,18 +18,14 @@ package ai.chronon.spark
 
 import ai.chronon.api
 import ai.chronon.api.ThriftJsonCodec
+import ai.chronon.spark.catalog.TableUtils
 import com.fasterxml.jackson.databind.ObjectMapper
 import com.fasterxml.jackson.module.scala.DefaultScalaModule
 import org.apache.commons.lang.exception.ExceptionUtils
-import org.slf4j.Logger
-import org.slf4j.LoggerFactory
+import org.slf4j.{Logger, LoggerFactory}
 
-import java.io.BufferedWriter
-import java.io.File
-import java.io.FileWriter
-import java.nio.file.Files
-import java.nio.file.Paths
-import scala.collection.immutable.Map
+import java.io.{BufferedWriter, File, FileWriter}
+import java.nio.file.{Files, Paths}
 
 object MetadataExporter {
   @transient lazy val logger: Logger = LoggerFactory.getLogger(getClass)

spark/src/main/scala/ai/chronon/spark/batch/BootstrapJob.scala

@@ -7,11 +7,12 @@ import ai.chronon.online.serde.SparkConversions
 import ai.chronon.orchestration.JoinBootstrapNode
 import ai.chronon.spark.Extensions._
 import ai.chronon.spark.JoinUtils.{coalescedJoin, set_add}
-import ai.chronon.spark.{BootstrapInfo, JoinUtils, TableUtils}
+import ai.chronon.spark.{BootstrapInfo, JoinUtils}
 import org.apache.spark.sql
 import org.apache.spark.sql.DataFrame
 import org.apache.spark.sql.functions.{coalesce, col, lit, typedLit}
 import org.slf4j.{Logger, LoggerFactory}
+import ai.chronon.spark.catalog.TableUtils
 
 import scala.collection.Seq
 

spark/src/main/scala/ai/chronon/spark/batch/JoinPartJob.scala

@@ -1,16 +1,14 @@
 package ai.chronon.spark.batch
 
-import ai.chronon.api
-import ai.chronon.api.{Accuracy, Constants, DateRange, JoinPart, PartitionRange, PartitionSpec}
-import ai.chronon.api.DataModel
 import ai.chronon.api.DataModel.{ENTITIES, EVENTS}
 import ai.chronon.api.Extensions.{DateRangeOps, DerivationOps, GroupByOps, JoinPartOps, MetadataOps}
 import ai.chronon.api.PartitionRange.toTimeRange
 import ai.chronon.api.{Accuracy, Builders, Constants, DateRange, JoinPart, PartitionRange}
 import ai.chronon.online.metrics.Metrics
 import ai.chronon.orchestration.JoinPartNode
 import ai.chronon.spark.Extensions._
-import ai.chronon.spark.{GroupBy, JoinUtils, TableUtils}
+import ai.chronon.spark.catalog.TableUtils
+import ai.chronon.spark.{GroupBy, JoinUtils}
 import org.apache.spark.sql.DataFrame
 import org.apache.spark.sql.functions.{col, date_format}
 import org.apache.spark.util.sketch.BloomFilter

spark/src/main/scala/ai/chronon/spark/batch/LabelJoinV2.scala

@@ -6,9 +6,9 @@ import ai.chronon.api.PartitionRange.toTimeRange
 import ai.chronon.api._
 import ai.chronon.online.metrics.Metrics
 import ai.chronon.online.serde.SparkConversions
-import ai.chronon.spark.Analyzer
 import ai.chronon.spark.Extensions._
-import ai.chronon.spark.{GroupBy, JoinUtils, TableUtils}
+import ai.chronon.spark.GroupBy
+import ai.chronon.spark.catalog.TableUtils
 import org.apache.spark.sql.DataFrame
 import org.apache.spark.sql.functions.{col, lit}
 import org.apache.spark.sql.types.{DataType, StructType}

spark/src/main/scala/ai/chronon/spark/batch/MergeJob.scala

@@ -7,7 +7,8 @@ import ai.chronon.api.{Accuracy, Constants, DateRange, JoinPart, PartitionRange,
 import ai.chronon.orchestration.JoinMergeNode
 import ai.chronon.spark.Extensions._
 import ai.chronon.spark.JoinUtils.coalescedJoin
-import ai.chronon.spark.{JoinUtils, TableUtils}
+import ai.chronon.spark.JoinUtils
+import ai.chronon.spark.catalog.TableUtils
 import org.apache.spark.sql.DataFrame
 import org.apache.spark.sql.functions.{col, date_add, date_format, to_date}
 import org.slf4j.{Logger, LoggerFactory}

spark/src/main/scala/ai/chronon/spark/batch/SourceJob.scala

@@ -6,7 +6,7 @@ import ai.chronon.api.ScalaJavaConversions.JListOps
 import ai.chronon.orchestration.SourceWithFilterNode
 import ai.chronon.spark.Extensions._
 import ai.chronon.spark.JoinUtils.parseSkewKeys
-import ai.chronon.spark.TableUtils
+import ai.chronon.spark.catalog.TableUtils
 
 import scala.collection.{Map, Seq}
 import scala.jdk.CollectionConverters._

spark/src/main/scala/ai/chronon/spark/batch/StagingQuery.scala

@@ -5,9 +5,10 @@ import ai.chronon.api.ScalaJavaConversions._
 import ai.chronon.api.thrift.TBase
 import ai.chronon.api.{EngineType, ParametricMacro, PartitionRange, ThriftJsonCodec}
 import ai.chronon.spark.Extensions._
-import ai.chronon.spark.{SparkSessionBuilder, TableUtils}
+import ai.chronon.spark.SparkSessionBuilder
 import org.rogach.scallop.{ScallopConf, ScallopOption}
 import org.slf4j.{Logger, LoggerFactory}
+import ai.chronon.spark.catalog.TableUtils
 
 import scala.collection.mutable
 import scala.reflect.ClassTag

spark/src/main/scala/ai/chronon/spark/format/CreationUtils.scala renamed to spark/src/main/scala/ai/chronon/spark/catalog/CreationUtils.scala

@@ -1,4 +1,4 @@
-package ai.chronon.spark.format
+package ai.chronon.spark.catalog
 
 import org.apache.spark.sql.types.StructType
 

spark/src/main/scala/ai/chronon/spark/format/DefaultFormatProvider.scala renamed to spark/src/main/scala/ai/chronon/spark/catalog/DefaultFormatProvider.scala

@@ -1,4 +1,4 @@
-package ai.chronon.spark.format
+package ai.chronon.spark.catalog
 
 import org.apache.spark.sql.SparkSession
 import org.slf4j.{Logger, LoggerFactory}

spark/src/main/scala/ai/chronon/spark/format/DeltaLake.scala renamed to spark/src/main/scala/ai/chronon/spark/catalog/DeltaLake.scala

@@ -1,4 +1,4 @@
-package ai.chronon.spark.format
+package ai.chronon.spark.catalog
 
 import org.apache.spark.sql.SparkSession
 import org.apache.spark.sql.delta.DeltaLog

spark/src/main/scala/ai/chronon/spark/format/Format.scala renamed to spark/src/main/scala/ai/chronon/spark/catalog/Format.scala

@@ -1,4 +1,4 @@
-package ai.chronon.spark.format
+package ai.chronon.spark.catalog
 
 import org.apache.spark.sql.SparkSession
 import org.slf4j.{Logger, LoggerFactory}

spark/src/main/scala/ai/chronon/spark/format/FormatProvider.scala renamed to spark/src/main/scala/ai/chronon/spark/catalog/FormatProvider.scala

@@ -1,4 +1,4 @@
-package ai.chronon.spark.format
+package ai.chronon.spark.catalog
 
 import org.apache.spark.sql.SparkSession
 

spark/src/main/scala/ai/chronon/spark/format/Hive.scala renamed to spark/src/main/scala/ai/chronon/spark/catalog/Hive.scala

@@ -1,4 +1,4 @@
-package ai.chronon.spark.format
+package ai.chronon.spark.catalog
 
 import org.apache.spark.sql.SparkSession
 

spark/src/main/scala/ai/chronon/spark/format/Iceberg.scala renamed to spark/src/main/scala/ai/chronon/spark/catalog/Iceberg.scala

@@ -1,6 +1,5 @@
-package ai.chronon.spark.format
+package ai.chronon.spark.catalog
 
-import ai.chronon.spark.TableUtils
 import org.apache.spark.sql.SparkSession
 import org.apache.spark.sql.functions.{col, date_format}
 import org.apache.spark.sql.types.StructType

spark/src/main/scala/ai/chronon/spark/TableUtils.scala renamed to spark/src/main/scala/ai/chronon/spark/catalog/TableUtils.scala

@@ -14,30 +14,28 @@
  *    limitations under the License.
  */
 
-package ai.chronon.spark
+package ai.chronon.spark.catalog
 
 import ai.chronon.api.ColorPrinter.ColorString
 import ai.chronon.api.Extensions._
 import ai.chronon.api.ScalaJavaConversions._
 import ai.chronon.api.{Constants, PartitionRange, PartitionSpec, Query, QueryUtils, TsUtils}
 import ai.chronon.spark.Extensions._
-import ai.chronon.spark.format.CreationUtils.alterTablePropertiesSql
-import ai.chronon.spark.format.{CreationUtils, FormatProvider, Iceberg}
 import org.apache.hadoop.hive.metastore.api.AlreadyExistsException
-import org.apache.spark.sql.{AnalysisException, DataFrame, SaveMode, SparkSession}
 import org.apache.spark.sql.catalyst.analysis.TableAlreadyExistsException
 import org.apache.spark.sql.catalyst.plans.logical.{Filter, Project}
+import org.apache.spark.sql.catalyst.util.QuotingUtils
 import org.apache.spark.sql.functions._
 import org.apache.spark.sql.types._
+import org.apache.spark.sql.{AnalysisException, DataFrame, SaveMode, SparkSession}
 import org.apache.spark.storage.StorageLevel
 import org.slf4j.{Logger, LoggerFactory}
 
 import java.io.{PrintWriter, StringWriter}
-import java.time.{Instant, ZoneId}
 import java.time.format.DateTimeFormatter
+import java.time.{Instant, ZoneId}
 import scala.collection.{Seq, mutable}
 import scala.util.{Failure, Success, Try}
-import org.apache.spark.sql.catalyst.util.QuotingUtils
 
 /** Trait to track the table format in use by a Chronon dataset and some utility methods to help
   * retrieve metadata / configure it appropriately at creation time
@@ -274,7 +272,7 @@ class TableUtils(@transient val sparkSession: SparkSession) extends Serializable
 
     // Run tableProperties
     Option(tableProperties).filter(_.nonEmpty).foreach { props =>
-      sql(alterTablePropertiesSql(tableName, props))
+      sql(CreationUtils.alterTablePropertiesSql(tableName, props))
     }
 
     val finalizedDf = if (autoExpand) {
@@ -564,7 +562,7 @@ class TableUtils(@transient val sparkSession: SparkSession) extends Serializable
       sql(expandTableQueryOpt.get)
 
       // set a flag in table props to indicate that this is a dynamic table
-      sql(alterTablePropertiesSql(tableName, Map(Constants.ChrononDynamicTable -> true.toString)))
+      sql(CreationUtils.alterTablePropertiesSql(tableName, Map(Constants.ChrononDynamicTable -> true.toString)))
     }
   }