rebase

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

Author: tchow-zlai

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

@@ -23,7 +23,7 @@ case object BigQueryNative extends Format {
       dfw.load(bqFriendlyName)
     } else {
       dfw
-        .option("filter", partitionFilters.trim.stripPrefix("(").stripSuffix(")"))
+        .option("filter", partitionFilters)
         .load(bqFriendlyName)
     }
   }

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

@@ -114,6 +114,17 @@ class BigQueryCatalogTest extends AnyFlatSpec with MockitoSugar {
     println(allParts)
   }
 
+  it should "integration testing bigquery partition pushdown" ignore {
+    import spark.implicits._
+    val iceberg = "data.checkouts_native"
+
+    val singleFilter = tableUtils.loadTable(iceberg, List("ds = '2023-11-30'"))
+    val multiFilter = tableUtils.loadTable(iceberg, List("ds = '2023-11-30'", "ds = '2023-11-30'"))
+    assertEquals(
+      singleFilter.select("user_id", "ds").as[(String, String)].collect.toList,
+      multiFilter.select("user_id", "ds").as[(String, String)].collect.toList)
+  }
+
   it should "integration testing formats" ignore {
     val externalTable = "default_iceberg.data.checkouts_parquet"
     val externalFormat = FormatProvider.from(spark).readFormat(externalTable)

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

@@ -901,7 +901,7 @@ object Driver {
       // Create a JoinPartNode from the join part
       val joinPartNode = new JoinPartNode()
         .setJoinPart(joinPart)
-        .setLeftSourceTable(JoinUtils.computeLeftSourceTableName(join))
+        .setLeftSourceTable(JoinUtils.computeFullLeftSourceTableName(join))
         .setLeftDataModel(join.left.dataModel)
         .setSkewKeys(join.skewKeys)
 

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

@@ -1,8 +1,5 @@
 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

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

@@ -30,7 +30,7 @@ class MergeJob(node: JoinMergeNode, range: DateRange, joinParts: Seq[JoinPart])(
   private val leftInputTable = if (join.bootstrapParts != null || join.onlineExternalParts != null) {
     join.metaData.bootstrapTable
   } else {
-    JoinUtils.computeLeftSourceTableName(join)
+    JoinUtils.computeFullLeftSourceTableName(join)
   }
   // Use the node's Join's metadata for output table
   private val outputTable = node.metaData.outputTable