This is an automated email from the ASF dual-hosted git repository.
riteshghorse pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 1273d22c7b5 [Python] get options outside of DoFn context in
bigquery(#27662)
1273d22c7b5 is described below
commit 1273d22c7b51e4ae6fab6d5ee9f4737b3b231730
Author: Ahmed Abualsaud <[email protected]>
AuthorDate: Tue Jul 25 20:08:06 2023 +0000
[Python] get options outside of DoFn context in bigquery(#27662)
---
sdks/python/apache_beam/io/gcp/bigquery_read_internal.py | 4 +++-
1 file changed, 3 insertions(+), 1 deletion(-)
diff --git a/sdks/python/apache_beam/io/gcp/bigquery_read_internal.py
b/sdks/python/apache_beam/io/gcp/bigquery_read_internal.py
index 6841de1e26f..5a59b409883 100644
--- a/sdks/python/apache_beam/io/gcp/bigquery_read_internal.py
+++ b/sdks/python/apache_beam/io/gcp/bigquery_read_internal.py
@@ -144,6 +144,8 @@ class _PassThroughThenCleanupTempDatasets(PTransform):
self.side_input = side_input
def expand(self, input):
+ pipeline_options = input.pipeline.options
+
class PassThrough(beam.DoFn):
def process(self, element):
yield element
@@ -151,7 +153,7 @@ class _PassThroughThenCleanupTempDatasets(PTransform):
class CleanUpProjects(beam.DoFn):
def process(self, unused_element, unused_signal, pipeline_details):
bq = bigquery_tools.BigQueryWrapper.from_pipeline_options(
- input.pipeline.options)
+ pipeline_options)
pipeline_details = pipeline_details[0]
if 'temp_table_ref' in pipeline_details.keys():
temp_table_ref = pipeline_details['temp_table_ref']
