Ensuring that BigQuery client will have the proper project id

connector/src/main/scala/com/google/cloud/spark/bigquery/BigQueryRelationProvider.scala

@@ -120,7 +120,7 @@ class BigQueryRelationProvider(
   }
 
   private def getOrCreateBigQuery(options: SparkBigQueryOptions) =
-    getBigQuery().getOrElse(BigQueryRelationProvider.createBigQuery(options))
+    getBigQuery().getOrElse(BigQueryUtil.createBigQuery(options))
 
   def createSparkBigQueryOptions(sqlContext: SQLContext,
                                  parameters: Map[String, String],
@@ -136,25 +136,6 @@ class BigQueryRelationProvider(
   override def shortName: String = "bigquery"
 }
 
-object BigQueryRelationProvider {
-
-  def createBigQuery(options: SparkBigQueryOptions): BigQuery =
-    options.createCredentials.fold(
-      BigQueryOptions.getDefaultInstance.getService
-    )(bigQueryWithCredentials(options.parentProject, _))
-
-  private def bigQueryWithCredentials(parentProject: String,
-                                      credentials: Credentials): BigQuery = {
-    BigQueryOptions
-      .newBuilder()
-      .setProjectId(parentProject)
-      .setCredentials(credentials)
-      .build()
-      .getService
-  }
-
-}
-
 // DefaultSource is required for spark.read.format("com.google.cloud.spark.bigquery")
 // BigQueryRelationProvider is more consistent with the internal providers
 class DefaultSource extends BigQueryRelationProvider

connector/src/main/scala/com/google/cloud/spark/bigquery/BigQueryUtil.scala

@@ -17,19 +17,21 @@ package com.google.cloud.spark.bigquery
 
 import java.util.Properties
 
-import com.google.cloud.bigquery.{BigQueryError, BigQueryException, TableId}
+import com.google.cloud.bigquery.{BigQuery, BigQueryError, BigQueryException, BigQueryOptions, TableId}
 import com.google.cloud.http.BaseHttpServiceException.UNKNOWN_CODE
 
 import scala.util.matching.Regex
 import scala.collection.JavaConverters._
 import io.grpc.StatusRuntimeException
 import com.google.api.gax.rpc.StatusCode
+import com.google.auth.Credentials
 import io.grpc.Status
+import org.apache.spark.internal.Logging
 
 /**
  * Static helpers for working with BigQuery.
  */
-object BigQueryUtil {
+object BigQueryUtil extends Logging{
 
   private val PROJECT_PATTERN = """\S+"""
   private val DATASET_PATTERN = """\w+"""
@@ -129,4 +131,19 @@ object BigQueryUtil {
   def toSeq[T](list: java.util.List[T]): Seq[T] = list.asScala.toSeq
 
   def toJavaIterator[T](it: Iterator[T]): java.util.Iterator[T] = it.asJava
+
+  def createBigQuery(options: SparkBigQueryOptions): BigQuery = {
+    val credentials = options.createCredentials
+      .getOrElse(BigQueryOptions.getDefaultInstance.getCredentials)
+    val parentProject = options.parentProject
+    logInfo(
+      s"BigQuery client project id is [$parentProject}], derived from teh parentProject option")
+    BigQueryOptions
+      .newBuilder()
+      .setProjectId(parentProject)
+      .setCredentials(credentials)
+      .build()
+      .getService
+  }
+
 }

connector/src/main/scala/com/google/cloud/spark/bigquery/direct/DirectBigQueryRelation.scala

@@ -43,7 +43,7 @@ private[bigquery] class DirectBigQueryRelation(
     getClient: SparkBigQueryOptions => BigQueryReadClient =
          DirectBigQueryRelation.createReadClient,
     bigQueryClient: SparkBigQueryOptions => BigQuery =
-         DirectBigQueryRelation.createBigQueryClient)
+         BigQueryUtil.createBigQuery)
     (@transient override val sqlContext: SQLContext)
     extends BigQueryRelation(options, table)(sqlContext)
         with TableScan with PrunedScan with PrunedFilteredScan {
@@ -356,14 +356,6 @@ object DirectBigQueryRelation {
     BigQueryReadClient.create(clientSettings.build)
   }
 
-  def createBigQueryClient(options: SparkBigQueryOptions): BigQuery = {
-    val BigQueryOptionsBuilder = BigQueryOptions.newBuilder()
-      .setHeaderProvider(headerProvider)
-    // set credentials of provided
-    options.createCredentials.foreach(BigQueryOptionsBuilder.setCredentials)
-    BigQueryOptionsBuilder.build.getService
-  }
-
   private def headerProvider =
     FixedHeaderProvider.create("user-agent",
       new SparkBigQueryConnectorUserAgentProvider("v1").getUserAgent)