[beam] branch release-2.25.0 updated: Passing project properly in BQSource

[ro...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

robinyqiu pushed a commit to branch release-2.25.0
in repository https://gitbox.apache.org/repos/asf/beam.git


The following commit(s) were added to refs/heads/release-2.25.0 by this push:
     new 9a1e7bf  Passing project properly in BQSource
     new 78167f0  Merge pull request #12991 from pabloem/cp-BEAM-10950
9a1e7bf is described below

commit 9a1e7bf4cd399177e5de7fe28bad741cade452a5
Author: Pablo Estrada <pa...@apache.org>
AuthorDate: Tue Sep 29 12:08:40 2020 -0700

    Passing project properly in BQSource
---
 sdks/python/apache_beam/io/gcp/bigquery.py       | 7 ++++++-
 sdks/python/apache_beam/io/gcp/bigquery_tools.py | 1 +
 2 files changed, 7 insertions(+), 1 deletion(-)

diff --git a/sdks/python/apache_beam/io/gcp/bigquery.py b/sdks/python/apache_beam/io/gcp/bigquery.py
index 4c42ed3..de68364 100644
--- a/sdks/python/apache_beam/io/gcp/bigquery.py
+++ b/sdks/python/apache_beam/io/gcp/bigquery.py
@@ -743,11 +743,13 @@ class _CustomBigQuerySource(BoundedSource):
       if (isinstance(self.table_reference, vp.ValueProvider) and
           self.table_reference.is_accessible()):
         table_ref = bigquery_tools.parse_table_reference(
-            self.table_reference.get(), project=self.project)
+            table_ref, project=self._get_project())
       elif isinstance(self.table_reference, vp.ValueProvider):
         # Size estimation is best effort. We return None as we have
         # no access to the table that we're querying.
         return None
+      if not table_ref.projectId:
+        table_ref.projectId = self._get_project()
       table = bq.get_table(
           table_ref.projectId, table_ref.datasetId, table_ref.tableId)
       return int(table.numBytes)
@@ -804,6 +806,9 @@ class _CustomBigQuerySource(BoundedSource):
         self._setup_temporary_dataset(bq)
         self.table_reference = self._execute_query(bq)
 
+      if not self.table_reference.projectId:
+        self.table_reference.projectId = self._get_project()
+
       schema, metadata_list = self._export_files(bq)
       self.split_result = [
           self._create_source(metadata.path, schema)
diff --git a/sdks/python/apache_beam/io/gcp/bigquery_tools.py b/sdks/python/apache_beam/io/gcp/bigquery_tools.py
index fde2986..f1fb5cd 100644
--- a/sdks/python/apache_beam/io/gcp/bigquery_tools.py
+++ b/sdks/python/apache_beam/io/gcp/bigquery_tools.py
@@ -784,6 +784,7 @@ class BigQueryWrapper(object):
             ),
             jobReference=job_reference,
         ))
+    logging.info('Performing BigQuery extract job: %s', request)
     response = self.client.jobs.Insert(request)
     return response.jobReference