Merge pull request #16999 from [BEAM-14039] Propagate ignore_unknown_columns parameter.

pabloem · web-flow · commit fa8979c87f64 · 2022-03-10T09:56:50.000-08:00
[BEAM-14039] Propagate ignore_unknown_columns parameter.
diff --git a/sdks/python/apache_beam/io/gcp/bigquery.py b/sdks/python/apache_beam/io/gcp/bigquery.py
@@ -1796,6 +1796,7 @@ def __init__(
       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 @@ def __init__(
     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 @@ def expand(self, input):
         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 @@ def expand(self, pcoll):
           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
@@ -1019,6 +1019,7 @@ def store_callback(table, **kwargs):
               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 @@ def store_callback(table, **kwargs):
               retry_strategy=None,
               additional_bq_parameters=[],
               ignore_insert_ids=False,
+              ignore_unknown_columns=False,
               with_auto_sharding=with_auto_sharding,
               test_client=client))
 
