This is an automated email from the ASF dual-hosted git repository.
claudevdm 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 01559cc9127 Added schema_update_options to Python BigQuery writes
(#39078)
01559cc9127 is described below
commit 01559cc91277fdcbcd45080ffe68a620790f8283
Author: Lalit Yadav <[email protected]>
AuthorDate: Thu Aug 27 08:39:34 2026 -0500
Added schema_update_options to Python BigQuery writes (#39078)
* Add schema_update_options to Python BigQuery writes
* Fix BigQuery schema update options formatting
* Validate BigQuery schema update options
* updated CHANGES.md
* changing the BigQuerySchemaUpdateOption to Enum
---------
Co-authored-by: claudevdm <[email protected]>
---
CHANGES.md | 1 +
sdks/python/apache_beam/io/gcp/bigquery.py | 90 ++++++++++++++-
sdks/python/apache_beam/io/gcp/bigquery_test.py | 144 +++++++++++++++++++++++-
3 files changed, 232 insertions(+), 3 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index 903f65b50a6..16c2d3ee6dd 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -65,6 +65,7 @@
## I/Os
* Support for X source added (Java/Python)
([#X](https://github.com/apache/beam/issues/X)).
+* Added `schema_update_options` to `WriteToBigQuery` file loads, allowing
BigQuery load jobs to add nullable fields or relax required fields when
appending data (Python) ([#21141](https://github.com/apache/beam/issues/21141)).
* BigQueryIO now supports reading BigQuery Lakehouse runtime catalog (BigLake
metastore) Iceberg tables with the Storage Read API, using 4-part
`project.catalog.namespace.table` identifiers (or a `TableReference` with a
composite `catalog.namespace` dataset id). Previously such references were
silently mis-parsed (Java)
([#39597](https://github.com/apache/beam/issues/39597)) .
## New Features / Improvements
diff --git a/sdks/python/apache_beam/io/gcp/bigquery.py
b/sdks/python/apache_beam/io/gcp/bigquery.py
index 314effad552..40f17bfe9b0 100644
--- a/sdks/python/apache_beam/io/gcp/bigquery.py
+++ b/sdks/python/apache_beam/io/gcp/bigquery.py
@@ -366,6 +366,7 @@ import time
import uuid
import warnings
from dataclasses import dataclass
+from enum import Enum
from typing import Optional
from typing import Union
@@ -442,6 +443,7 @@ except ImportError:
__all__ = [
'TableRowJsonCoder',
'BigQueryDisposition',
+ 'BigQuerySchemaUpdateOption',
'BigQuerySource',
'BigQuerySink',
'BigQueryQueryPriority',
@@ -483,6 +485,35 @@ overhead:
https://cloud.google.com/bigquery/quotas#streaming_inserts
"""
MAX_INSERT_PAYLOAD_SIZE = 9 << 20
+_SCHEMA_UPDATE_OPTIONS = 'schemaUpdateOptions'
+
+
+def _merge_schema_update_options(
+ additional_bq_parameters, schema_update_options):
+ additional_bq_parameters = dict(additional_bq_parameters or {})
+ if _SCHEMA_UPDATE_OPTIONS in additional_bq_parameters:
+ raise ValueError(
+ '%s can be set with either schema_update_options or '
+ 'additional_bq_parameters, but not both.' % _SCHEMA_UPDATE_OPTIONS)
+ additional_bq_parameters[_SCHEMA_UPDATE_OPTIONS] = schema_update_options
+ return additional_bq_parameters
+
+
+class _AdditionalBQParametersWithSchemaUpdateOptions(object):
+ def __init__(self, additional_bq_parameters, schema_update_options):
+ self.additional_bq_parameters = additional_bq_parameters
+ self.schema_update_options = schema_update_options
+
+ def __call__(self, destination):
+ if callable(self.additional_bq_parameters):
+ additional_bq_parameters = self.additional_bq_parameters(destination)
+ elif isinstance(self.additional_bq_parameters, vp.ValueProvider):
+ additional_bq_parameters = self.additional_bq_parameters.get()
+ else:
+ additional_bq_parameters = self.additional_bq_parameters
+ return _merge_schema_update_options(
+ additional_bq_parameters, self.schema_update_options)
+
@deprecated(since='2.11.0', current="bigquery_tools.parse_table_reference")
def _parse_table_reference(table, dataset=None, project=None):
@@ -580,6 +611,32 @@ class BigQueryDisposition(object):
return disposition
+class BigQuerySchemaUpdateOption(str, Enum):
+ """Enum holding standard strings used for schema update options."""
+
+ ALLOW_FIELD_ADDITION = 'ALLOW_FIELD_ADDITION'
+ ALLOW_FIELD_RELAXATION = 'ALLOW_FIELD_RELAXATION'
+
+ @staticmethod
+ def validate(options):
+ if options is None:
+ return None
+ if not isinstance(options, list):
+ raise ValueError(
+ 'schema_update_options must be a list. Received %s.' %
+ type(options).__name__)
+ values = tuple(option.value for option in BigQuerySchemaUpdateOption)
+ validated_options = []
+ for option in options:
+ try:
+ validated_options.append(BigQuerySchemaUpdateOption(option).value)
+ except ValueError:
+ raise ValueError(
+ 'Invalid schema update option %s. Expecting %s' %
+ (option, values)) from None
+ return validated_options
+
+
class BigQueryQueryPriority(object):
"""Class holding standard strings used for query priority."""
@@ -2007,7 +2064,8 @@ class WriteToBigQuery(PTransform):
primary_key: list[str] = None,
expansion_service=None,
big_lake_configuration=None,
- type_overrides=None):
+ type_overrides=None,
+ schema_update_options=None):
"""Initialize a WriteToBigQuery transform.
Args:
@@ -2108,6 +2166,14 @@ bigquery_v2_messages.TableSchema`. or a `ValueProvider`
that has a JSON string,
These can be 'timePartitioning', 'clustering', etc. They are passed
directly to the job load configuration. See
https://cloud.google.com/bigquery/docs/reference/rest/v2/Job#jobconfigurationload
+ schema_update_options (list): Allows the schema of the destination
+ table to be updated as a side effect of the load job. Each item may be
+ a :class:`BigQuerySchemaUpdateOption` member or its string value.
+ Supported values are
+ :attr:`BigQuerySchemaUpdateOption.ALLOW_FIELD_ADDITION` and
+ :attr:`BigQuerySchemaUpdateOption.ALLOW_FIELD_RELAXATION`. This option
+ is only valid for ``FILE_LOADS`` and cannot be specified together with
+ ``schemaUpdateOptions`` in ``additional_bq_parameters``.
table_side_inputs (tuple): A tuple with ``AsSideInput`` PCollections to
be
passed to the table callable (if one is provided).
schema_side_inputs: A tuple with ``AsSideInput`` PCollections to be
@@ -2222,6 +2288,8 @@ bigquery_v2_messages.TableSchema`. or a `ValueProvider`
that has a JSON string,
self._temp_file_format = temp_file_format or bigquery_tools.FileFormat.JSON
self.additional_bq_parameters = additional_bq_parameters or {}
+ self.schema_update_options = BigQuerySchemaUpdateOption.validate(
+ schema_update_options)
self.table_side_inputs = table_side_inputs or ()
self.schema_side_inputs = schema_side_inputs or ()
self._ignore_insert_ids = ignore_insert_ids
@@ -2252,6 +2320,16 @@ bigquery_v2_messages.TableSchema`. or a `ValueProvider`
that has a JSON string,
else:
return self.method
+ def _additional_bq_parameters_for_file_loads(self):
+ if self.schema_update_options is None:
+ return self.additional_bq_parameters
+ if (callable(self.additional_bq_parameters) or
+ isinstance(self.additional_bq_parameters, vp.ValueProvider)):
+ return _AdditionalBQParametersWithSchemaUpdateOptions(
+ self.additional_bq_parameters, self.schema_update_options)
+ return _merge_schema_update_options(
+ self.additional_bq_parameters, self.schema_update_options)
+
def expand(self, pcoll):
p = pcoll.pipeline
@@ -2270,6 +2348,12 @@ bigquery_v2_messages.TableSchema`. or a `ValueProvider`
that has a JSON string,
experiments = p.options.view_as(DebugOptions).experiments or []
method_to_use = self._compute_method(experiments, is_streaming_pipeline)
+ if (self.schema_update_options is not None and
+ method_to_use != WriteToBigQuery.Method.FILE_LOADS):
+ raise ValueError(
+ 'schema_update_options is only supported when writing to BigQuery '
+ 'with FILE_LOADS.')
+
if method_to_use == WriteToBigQuery.Method.STREAMING_INSERTS:
if self.schema == SCHEMA_AUTODETECT:
raise ValueError(
@@ -2369,7 +2453,8 @@ bigquery_v2_messages.TableSchema`. or a `ValueProvider`
that has a JSON string,
test_client=self.test_client,
table_side_inputs=self.table_side_inputs,
schema_side_inputs=self.schema_side_inputs,
- additional_bq_parameters=self.additional_bq_parameters,
+ additional_bq_parameters=(
+ self._additional_bq_parameters_for_file_loads()),
validate=self._validate,
is_streaming_pipeline=is_streaming_pipeline,
load_job_project_id=self.load_job_project_id)
@@ -2448,6 +2533,7 @@ bigquery_v2_messages.TableSchema`. or a `ValueProvider`
that has a JSON string,
'method': self.method,
'insert_retry_strategy': self.insert_retry_strategy,
'additional_bq_parameters': self.additional_bq_parameters,
+ 'schema_update_options': self.schema_update_options,
'table_side_inputs': table_side_inputs,
'schema_side_inputs': schema_side_inputs,
'triggering_frequency': self.triggering_frequency,
diff --git a/sdks/python/apache_beam/io/gcp/bigquery_test.py
b/sdks/python/apache_beam/io/gcp/bigquery_test.py
index 51d13d96b73..dcadee7f6a1 100644
--- a/sdks/python/apache_beam/io/gcp/bigquery_test.py
+++ b/sdks/python/apache_beam/io/gcp/bigquery_test.py
@@ -1028,7 +1028,10 @@ class TestWriteToBigQuery(unittest.TestCase):
original = WriteToBigQuery(
table=lambda _, side_input: side_input['table'],
table_side_inputs=(table_record_pcv, ),
- schema=schema)
+ schema=schema,
+ schema_update_options=[
+ beam_bq.BigQuerySchemaUpdateOption.ALLOW_FIELD_ADDITION
+ ])
# pylint: disable=expression-not-assigned
p | beam.Create([]) | 'MyWriteToBigQuery' >> original
@@ -1070,6 +1073,145 @@ class TestWriteToBigQuery(unittest.TestCase):
deserialized_side_input_data.window_mapping_fn)
self.assertEqual(
original_side_input_data.view_fn, deserialized_side_input_data.view_fn)
+ self.assertEqual(
+ original.schema_update_options, deserialized.schema_update_options)
+
+ def test_schema_update_options_added_to_file_load_parameters(self):
+ additional_bq_parameters = {'timePartitioning': {'type': 'DAY'}}
+ schema_update_options = [
+ beam_bq.BigQuerySchemaUpdateOption.ALLOW_FIELD_ADDITION
+ ]
+ transform = WriteToBigQuery(
+ table='dataset.table',
+ method=WriteToBigQuery.Method.FILE_LOADS,
+ additional_bq_parameters=additional_bq_parameters,
+ schema_update_options=schema_update_options)
+
+ self.assertEqual(['ALLOW_FIELD_ADDITION'], transform.schema_update_options)
+ self.assertIs(type(transform.schema_update_options[0]), str)
+ self.assertEqual({
+ 'timePartitioning': {
+ 'type': 'DAY'
+ },
+ 'schemaUpdateOptions': ['ALLOW_FIELD_ADDITION'],
+ },
+ transform._additional_bq_parameters_for_file_loads())
+ self.assertNotIn('schemaUpdateOptions', additional_bq_parameters)
+
+ def test_schema_update_options_keeps_additional_bq_parameters_path(self):
+ additional_bq_parameters = {
+ 'schemaUpdateOptions': [
+ beam_bq.BigQuerySchemaUpdateOption.ALLOW_FIELD_ADDITION
+ ]
+ }
+ transform = WriteToBigQuery(
+ table='dataset.table',
+ method=WriteToBigQuery.Method.FILE_LOADS,
+ additional_bq_parameters=additional_bq_parameters)
+
+ self.assertEqual(
+ additional_bq_parameters,
+ transform._additional_bq_parameters_for_file_loads())
+
+ def test_schema_update_options_rejects_duplicate_configuration(self):
+ transform = WriteToBigQuery(
+ table='dataset.table',
+ method=WriteToBigQuery.Method.FILE_LOADS,
+ additional_bq_parameters={
+ 'schemaUpdateOptions': [
+ beam_bq.BigQuerySchemaUpdateOption.ALLOW_FIELD_RELAXATION
+ ]
+ },
+ schema_update_options=[
+ beam_bq.BigQuerySchemaUpdateOption.ALLOW_FIELD_ADDITION
+ ])
+
+ with self.assertRaisesRegex(ValueError, 'schemaUpdateOptions'):
+ transform._additional_bq_parameters_for_file_loads()
+
+ def test_schema_update_options_rejects_non_list(self):
+ schema_update_option = (
+ beam_bq.BigQuerySchemaUpdateOption.ALLOW_FIELD_ADDITION)
+ with self.assertRaisesRegex(ValueError, 'must be a list'):
+ WriteToBigQuery(
+ table='dataset.table',
+ method=WriteToBigQuery.Method.FILE_LOADS,
+ schema_update_options=schema_update_option)
+
+ def test_schema_update_options_rejects_invalid_value(self):
+ with self.assertRaisesRegex(ValueError, 'Invalid schema update option'):
+ WriteToBigQuery(
+ table='dataset.table',
+ method=WriteToBigQuery.Method.FILE_LOADS,
+ schema_update_options=['INVALID_SCHEMA_UPDATE_OPTION'])
+
+ def test_schema_update_options_accepts_valid_string(self):
+ transform = WriteToBigQuery(
+ table='dataset.table',
+ method=WriteToBigQuery.Method.FILE_LOADS,
+ schema_update_options=['ALLOW_FIELD_RELAXATION'])
+
+ self.assertEqual(['ALLOW_FIELD_RELAXATION'],
+ transform.schema_update_options)
+
+ def test_schema_update_options_with_callable_additional_bq_parameters(self):
+ schema_update_options = [
+ beam_bq.BigQuerySchemaUpdateOption.ALLOW_FIELD_ADDITION
+ ]
+
+ def additional_bq_parameters(destination):
+ self.assertEqual('project:dataset.table', destination)
+ return {'clustering': {'fields': ['columnA']}}
+
+ transform = WriteToBigQuery(
+ table='dataset.table',
+ method=WriteToBigQuery.Method.FILE_LOADS,
+ additional_bq_parameters=additional_bq_parameters,
+ schema_update_options=schema_update_options)
+
+ additional_parameters =
transform._additional_bq_parameters_for_file_loads()
+ self.assertEqual({
+ 'clustering': {
+ 'fields': ['columnA']
+ },
+ 'schemaUpdateOptions': schema_update_options,
+ },
+ additional_parameters('project:dataset.table'))
+
+ def test_schema_update_options_with_value_provider_parameters(self):
+ schema_update_options = [
+ beam_bq.BigQuerySchemaUpdateOption.ALLOW_FIELD_ADDITION
+ ]
+ transform = WriteToBigQuery(
+ table='dataset.table',
+ method=WriteToBigQuery.Method.FILE_LOADS,
+ additional_bq_parameters=StaticValueProvider(
+ dict, {'timePartitioning': {
+ 'type': 'DAY'
+ }}),
+ schema_update_options=schema_update_options)
+
+ additional_parameters =
transform._additional_bq_parameters_for_file_loads()
+ self.assertEqual({
+ 'timePartitioning': {
+ 'type': 'DAY'
+ },
+ 'schemaUpdateOptions': schema_update_options,
+ },
+ additional_parameters('project:dataset.table'))
+
+ def test_schema_update_options_only_supported_for_file_loads(self):
+ p = TestPipeline()
+ pcoll = p | beam.Create([{'columnA': 'value'}])
+
+ with self.assertRaisesRegex(ValueError, 'FILE_LOADS'):
+ _ = pcoll | WriteToBigQuery(
+ table='dataset.table',
+ schema='columnA:STRING',
+ method=WriteToBigQuery.Method.STREAMING_INSERTS,
+ schema_update_options=[
+ beam_bq.BigQuerySchemaUpdateOption.ALLOW_FIELD_ADDITION
+ ])
def test_streaming_triggering_frequency_without_auto_sharding(self):
def noop(table, **kwargs):
