wip

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

Author: tchow-zlai

.plugin-versions

@@ -1,5 +1,7 @@
 asdf-plugin-manager  https://github.com/asdf-community/asdf-plugin-manager.git  b5862c1
+gcloud               https://github.com/jthegedus/asdf-gcloud.git               00cdf06
 java                 https://github.com/halcyon/asdf-java.git                   0ec69b2
 python               https://github.com/danhper/asdf-python.git                 a3a0185
 sbt                  https://github.com/lerencao/asdf-sbt                       53c9f4b
 scala                https://github.com/asdf-community/asdf-scala.git           0533444
+thrift               https://github.com/alisaifee/asdf-thrift.git               fecdd6c

.tool-versions

@@ -5,3 +5,4 @@ sbt 1.8.2
 python
     3.7.17
     3.11.0
+gcloud 504.0.1

api/py/test/sample/group_bys/__init__.py

@@ -38,6 +38,10 @@ case class GcpFormatProvider(sparkSession: SparkSession) extends FormatProvider
   // Fixed to BigQuery for now.
   override def writeFormat(tableName: String): Format = {
 
+    val tableId = BigQueryUtil.parseTableId(tableName)
+    assert(Option(tableId.getProject).isDefined, s"project required for ${tableName}")
+    assert(Option(tableId.getDataset).isDefined, s"dataset required for ${tableName}")
+
     val tu = TableUtils(sparkSession)
     val partitionColumnOption =
       if (tu.tableReachable(tableName)) Map.empty else Map("partitionField" -> tu.partitionColumn)
@@ -49,7 +53,7 @@ case class GcpFormatProvider(sparkSession: SparkSession) extends FormatProvider
         ), // todo(tchow): No longer needed after https://github.com/GoogleCloudDataproc/spark-bigquery-connector/pull/1320
         "writeMethod" -> "indirect"
       ) ++ partitionColumnOption
-    BQuery(bqOptions.getProjectId, sparkOptions)
+    BQuery(tableId.getProject, sparkOptions)
   }
 
   private def format(tableName: String): Format = {
@@ -82,7 +86,7 @@ case class GcpFormatProvider(sparkSession: SparkSession) extends FormatProvider
           BQuery(table.getTableId.getProject, Map.empty)
         } else throw new IllegalStateException(s"Cannot support table of type: ${table.getDefinition}")
       })
-      .getOrElse(Hive)
+      .getOrElse(Option(btTableIdentifier.getProject).map(BQuery(_, Map.empty)).getOrElse(Hive))
 
     /**
       * Using federation