[beam] branch master updated: [BEAM-14039] Propagate ignore_unknown_columns parameter.

[pabloem]

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

pabloem 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 83c1670  [BEAM-14039] Propagate ignore_unknown_columns parameter.
     new fa8979c  Merge pull request #16999 from [BEAM-14039] Propagate 
ignore_unknown_columns parameter.
83c1670 is described below

commit 83c1670eff327fe4cd46e711e0bfa40335d08ae1
Author: Jakub Kukul <jakub.ku...@curbfood.com>
AuthorDate: Tue Mar 1 14:57:17 2022 +0200

    [BEAM-14039] Propagate ignore_unknown_columns parameter.
    
    Right now, the ignore_unknown_columns parameter is never
    passed from WriteToBigquery to _StreamToBigQuery
    and therefore never reaches the bq_client.insert_rows_json method.
---
 sdks/python/apache_beam/io/gcp/bigquery.py      | 4 ++++
 sdks/python/apache_beam/io/gcp/bigquery_test.py | 2 ++
 2 files changed, 6 insertions(+)

diff --git a/sdks/python/apache_beam/io/gcp/bigquery.py 
b/sdks/python/apache_beam/io/gcp/bigquery.py
index fb9a9ab..0503de8 100644
--- a/sdks/python/apache_beam/io/gcp/bigquery.py
+++ b/sdks/python/apache_beam/io/gcp/bigquery.py
@@ -1796,6 +1796,7 @@ class _StreamToBigQuery(PTransform):
       retry_strategy,
       additional_bq_parameters,
       ignore_insert_ids,
+      ignore_unknown_columns,
       with_auto_sharding,
       test_client=None):
     self.table_reference = table_reference
@@ -1811,6 +1812,7 @@ class _StreamToBigQuery(PTransform):
     self.test_client = test_client
     self.additional_bq_parameters = additional_bq_parameters
     self.ignore_insert_ids = ignore_insert_ids
+    self.ignore_unknown_columns = ignore_unknown_columns
     self.with_auto_sharding = with_auto_sharding
 
   class InsertIdPrefixFn(DoFn):
@@ -1836,6 +1838,7 @@ class _StreamToBigQuery(PTransform):
         test_client=self.test_client,
         additional_bq_parameters=self.additional_bq_parameters,
         ignore_insert_ids=self.ignore_insert_ids,
+        ignore_unknown_columns=self.ignore_unknown_columns,
         with_batched_input=self.with_auto_sharding)
 
     def _add_random_shard(element):
@@ -2168,6 +2171,7 @@ bigquery_v2_messages.TableSchema`. or a `ValueProvider` 
that has a JSON string,
           retry_strategy=self.insert_retry_strategy,
           additional_bq_parameters=self.additional_bq_parameters,
           ignore_insert_ids=self._ignore_insert_ids,
+          ignore_unknown_columns=self._ignore_unknown_columns,
           with_auto_sharding=self.with_auto_sharding,
           test_client=self.test_client)
 
diff --git a/sdks/python/apache_beam/io/gcp/bigquery_test.py 
b/sdks/python/apache_beam/io/gcp/bigquery_test.py
index 97ed721..a4fd505 100644
--- a/sdks/python/apache_beam/io/gcp/bigquery_test.py
+++ b/sdks/python/apache_beam/io/gcp/bigquery_test.py
@@ -1019,6 +1019,7 @@ class 
PipelineBasedStreamingInsertTest(_TestCaseWithTempDirCleanUp):
               retry_strategy=None,
               additional_bq_parameters=[],
               ignore_insert_ids=False,
+              ignore_unknown_columns=False,
               with_auto_sharding=False,
               test_client=client))
 
@@ -1078,6 +1079,7 @@ class 
PipelineBasedStreamingInsertTest(_TestCaseWithTempDirCleanUp):
               retry_strategy=None,
               additional_bq_parameters=[],
               ignore_insert_ids=False,
+              ignore_unknown_columns=False,
               with_auto_sharding=with_auto_sharding,
               test_client=client))