hack

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

Author: tchow-zlai

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

@@ -1,12 +1,12 @@
 package ai.chronon.integrations.cloud_gcp
 
-import com.google.cloud.bigquery.{
+import com.google.cloud.spark.bigquery.repackaged.com.google.cloud.bigquery.{
   BigQuery,
   BigQueryOptions,
   ExternalTableDefinition,
   StandardTableDefinition,
   TableDefinition,
-  TableId
+  TableId,
 }
 import com.google.cloud.spark.bigquery.BigQueryCatalog
 import org.apache.iceberg.gcp.bigquery.BigQueryMetastoreCatalog
@@ -19,6 +19,7 @@ import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat
 import org.apache.spark.sql.execution.datasources.v2.parquet.ParquetTable
 import org.apache.spark.sql.types.StructType
 import org.apache.spark.sql.util.CaseInsensitiveStringMap
+import com.google.cloud.spark.bigquery.{SchemaConverters, SchemaConvertersConfiguration}
 
 import java.util
 import scala.jdk.CollectionConverters._
@@ -122,22 +123,22 @@ class DelegatingBigQueryMetastoreCatalog extends TableCatalog with SupportsNames
               fileBasedTable
             }
             case stTable: StandardTableDefinition => {
+              import com.google.cloud.spark.bigquery.repackaged.com.google.inject.Injector
+              import com.google.cloud.spark.bigquery.v2.Spark35BigQueryTable
               //todo(tchow): Support partitioning
 
               // Hack because there's a bug in the BigQueryCatalog where they ignore the projectId.
               // See: https://github.com/GoogleCloudDataproc/spark-bigquery-connector/pull/1340
               // ideally it should be the below:
               // val connectorTable = connectorCatalog.loadTable(ident)
+              // So instead, we read the bigqueryTable ourselves to get access to the schema and pass it through.
               val nativeTable = connectorCatalog.loadTable(Identifier.of(Array(tId.getDataset), tId.getTable))
-              logger.info(s"Table: ${nativeTable}")
-              logger.info(s"Table name: ${nativeTable.name()}")
-              logger.info(s"Table properties: ${nativeTable.properties()}")
-              logger.info(s"Table partitioning: ${nativeTable.partitioning()}")
-              logger.info("Table schema:")
-              logger.info(f"${nativeTable.schema()}")
-              logger.info("Table columns:")
-              logger.info(f"${nativeTable.columns()}")
-              nativeTable
+              val injector = nativeTable.getClass.getDeclaredField("injector")
+              injector.setAccessible(true)
+              val value = injector.get(nativeTable).asInstanceOf[Injector]
+              val sc = SchemaConverters.from(SchemaConvertersConfiguration.createDefault())
+              val sparkSchema = sc.toSpark(stTable.getSchema)
+              new Spark35BigQueryTable(value, () => sparkSchema)
             }
             case _ => throw new IllegalStateException(s"Cannot support table of type: ${table.getDefinition}")
           }