diff --git a/.github/workflows/build_release_candidate.yml b/.github/workflows/build_release_candidate.yml index 978b546b8068..2f7ac8dfae2f 100644 --- a/.github/workflows/build_release_candidate.yml +++ b/.github/workflows/build_release_candidate.yml @@ -67,7 +67,7 @@ jobs: repository: apache/beam persist-credentials: true - name: Install Java 11 - uses: actions/setup-java@v5 + uses: actions/setup-java@v6 with: distribution: 'temurin' java-version: | @@ -126,7 +126,7 @@ jobs: GITHUB_EVENT_INPUTS_APACHE_ID: ${{ github.event.inputs.APACHE_ID }} GITHUB_EVENT_INPUTS_APACHE_PASSWORD: ${{ github.event.inputs.APACHE_PASSWORD }} - name: Install Java 11 - uses: actions/setup-java@v5 + uses: actions/setup-java@v6 with: distribution: 'temurin' java-version: '11' @@ -298,7 +298,7 @@ jobs: - name: Free Disk Space (Ubuntu) uses: jlumbroso/free-disk-space@v1.3.1 - name: Install Java - uses: actions/setup-java@v5 + uses: actions/setup-java@v6 with: distribution: 'temurin' java-version: | @@ -364,7 +364,7 @@ jobs: with: node-version: '16' - name: Install Java 21 - uses: actions/setup-java@v5 + uses: actions/setup-java@v6 with: distribution: 'temurin' java-version: '21' @@ -607,7 +607,7 @@ jobs: with: python-version: '3.10' - name: Install Java 17 - uses: actions/setup-java@v5 + uses: actions/setup-java@v6 with: distribution: 'temurin' java-version: '17' diff --git a/.github/workflows/code_completion_plugin_tests.yml b/.github/workflows/code_completion_plugin_tests.yml index 8e74ec1c684a..d4ad44def670 100644 --- a/.github/workflows/code_completion_plugin_tests.yml +++ b/.github/workflows/code_completion_plugin_tests.yml @@ -72,7 +72,7 @@ jobs: # Setup Java environment for the next steps - name: Setup Java - uses: actions/setup-java@v5 + uses: actions/setup-java@v6 with: distribution: 'temurin' java-version: '11' diff --git a/.github/workflows/republish_released_docker_containers.yml b/.github/workflows/republish_released_docker_containers.yml index 5ca9d52b1e5a..28ccbb7ec235 100644 --- a/.github/workflows/republish_released_docker_containers.yml +++ b/.github/workflows/republish_released_docker_containers.yml @@ -65,7 +65,7 @@ jobs: - name: Free Disk Space (Ubuntu) uses: jlumbroso/free-disk-space@v1.3.1 - name: Install Java 11 - uses: actions/setup-java@v5 + uses: actions/setup-java@v6 with: distribution: 'temurin' java-version: | diff --git a/.github/workflows/typescript_tests.yml b/.github/workflows/typescript_tests.yml index b34f7a5894c6..2e9290569c2e 100644 --- a/.github/workflows/typescript_tests.yml +++ b/.github/workflows/typescript_tests.yml @@ -112,7 +112,7 @@ jobs: with: python-version: '3.10' - name: Install Java - uses: actions/setup-java@v5 + uses: actions/setup-java@v6 with: distribution: 'temurin' java-version: '17' @@ -182,7 +182,7 @@ jobs: with: python-version: '3.10' - name: Install Java - uses: actions/setup-java@v5 + uses: actions/setup-java@v6 with: distribution: 'temurin' java-version: '17' diff --git a/CHANGES.md b/CHANGES.md index 151cbe14aaa2..16c2d3ee6ddb 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 @@ -204,6 +205,8 @@ * (Java) Projects using the Flink runner with Flink 2.1 or later alongside libraries requiring `org.lz4:lz4-java` (e.g., Kafka clients) may encounter a Gradle capability conflict, because Flink 2.1+ ships `at.yawk.lz4:lz4-java` which declares the same capability. To resolve, add a `capabilitiesResolution` rule to your `build.gradle` that selects `at.yawk.lz4:lz4-java` ([#38947](https://github.com/apache/beam/issues/38947)). * (Python) Long-running Python pipelines might experience memory growth and periodic OOMs ([#39406](https://github.com/apache/beam/issues/39406)). +* (Java) Pipelines with a moderate to heavy Cloud Storage read workload might experience a performance regression ([#39548](https://github.com/apache/beam/issues/39548)). +* (Java) Pipelines using the Dataflow Runner and Java versions 17+ may experience spiky memory caused by a JVM upgrade in the runner image ([#39897](https://github.com/apache/beam/issues/39897). # [2.74.0] - 2026-06-02 @@ -249,6 +252,8 @@ ## Known Issues * (Python) Long-running Python pipelines might experience memory growth and periodic OOMs ([#39406](https://github.com/apache/beam/issues/39406)). +* (Java) Pipelines with a moderate to heavy Cloud Storage read workload might experience a performance regression ([#39548](https://github.com/apache/beam/issues/39548)). +* (Java) Pipelines using the Dataflow Runner and Java versions 17+ may experience spiky memory caused by a JVM upgrade in the runner image ([#39897](https://github.com/apache/beam/issues/39897). # [2.73.0] - 2026-04-29 diff --git a/sdks/go.mod b/sdks/go.mod index c20f8689e7b2..2fad445f605c 100644 --- a/sdks/go.mod +++ b/sdks/go.mod @@ -70,7 +70,7 @@ require ( require ( github.com/avast/retry-go/v4 v4.7.0 - github.com/fsouza/fake-gcs-server v1.56.0 + github.com/fsouza/fake-gcs-server v1.56.1 github.com/golang-cz/devslog v0.0.17 github.com/moby/moby/api v1.55.0 github.com/moby/moby/client v0.5.1 @@ -82,7 +82,7 @@ require ( cloud.google.com/go/auth v0.23.0 // indirect cloud.google.com/go/auth/oauth2adapt v0.2.8 // indirect cloud.google.com/go/monitoring v1.29.0 // indirect - cloud.google.com/go/pubsub/v2 v2.6.1 // indirect + cloud.google.com/go/pubsub/v2 v2.6.2 // indirect dario.cat/mergo v1.0.2 // indirect filippo.io/edwards25519 v1.2.0 // indirect github.com/GoogleCloudPlatform/grpc-gcp-go/grpcgcp v1.6.0 // indirect diff --git a/sdks/go.sum b/sdks/go.sum index 4df6c2295d2e..05697f3385d2 100644 --- a/sdks/go.sum +++ b/sdks/go.sum @@ -87,8 +87,8 @@ cloud.google.com/go/pubsub v1.3.1/go.mod h1:i+ucay31+CNRpDW4Lu78I4xXG+O1r/MAHgjp cloud.google.com/go/pubsub v1.19.0/go.mod h1:/O9kmSe9bb9KRnIAWkzmqhPjHo6LtzGOBYd/kr06XSs= cloud.google.com/go/pubsub v1.51.1 h1:R3G1wCOxBO7jRpL8x2pdZMv1GAJDF6ax/m2zPOtvTNE= cloud.google.com/go/pubsub v1.51.1/go.mod h1:y2T0IKtW1iWwVvazYaRpqOAFO4gy2+O7dTDt9TWY/5U= -cloud.google.com/go/pubsub/v2 v2.6.1 h1:jX6gnC4n8BgYx6MOYICgbbaXZpr1vKeNOE3Bn17P5zg= -cloud.google.com/go/pubsub/v2 v2.6.1/go.mod h1:1y2lZnKfUFPZz0PU4YmXyk4lA11+xmYA42zbC32RkxQ= +cloud.google.com/go/pubsub/v2 v2.6.2 h1:YPeEXnf4LZz9eVdpbXqBMRlxdwk4breCUsoIp7uDet4= +cloud.google.com/go/pubsub/v2 v2.6.2/go.mod h1:JaFvWNVRk3Knoil/4M1ECeLOaI9D8drbmJWypQlK5aM= cloud.google.com/go/secretmanager v1.3.0/go.mod h1:+oLTkouyiYiabAQNugCeTS3PAArGiMJuBqvJnJsyH+U= cloud.google.com/go/spanner v1.94.0 h1:tve2XojeMa32SsrDNkp3J08TW4TZUVrv8TvTqGUif4A= cloud.google.com/go/spanner v1.94.0/go.mod h1:Z2+83J5oVDmd1n5ntVMmjEuiNoXOpAyNeG7y1tuEHk0= @@ -364,8 +364,8 @@ github.com/fogleman/gg v1.3.0/go.mod h1:R/bRT+9gY/C5z7JzPU0zXsXHKM4/ayA+zqcVNZzP github.com/form3tech-oss/jwt-go v3.2.2+incompatible/go.mod h1:pbq4aXjuKjdthFRnoDwaVPLA+WlJuPGy+QneDUgJi2k= github.com/fortytw2/leaktest v1.3.0/go.mod h1:jDsjWgpAGjm2CA7WthBh/CdZYEPF31XHquHwclZch5g= github.com/fsnotify/fsnotify v1.5.1/go.mod h1:T3375wBYaZdLLcVNkcVbzGHY7f1l/uK5T5Ai1i3InKU= -github.com/fsouza/fake-gcs-server v1.56.0 h1:0Y5MHIJYLSCXAnlYZPq3CA3k2ggsnVj3TsVWHnDFfmk= -github.com/fsouza/fake-gcs-server v1.56.0/go.mod h1:YWNj8Gseru6LXGCBRSqU7fHJq/ESitLrFHT49lYHvdc= +github.com/fsouza/fake-gcs-server v1.56.1 h1:K03sAvbLvDz4hAynpCCUqnNRp+ik9JFSvHbkD/wTPOU= +github.com/fsouza/fake-gcs-server v1.56.1/go.mod h1:rzibfBNKouMLeVYDkIDqUiCEcfgDyJWe+4PhG7uesmU= github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04= github.com/gin-contrib/sse v0.1.0/go.mod h1:RHrZQHXnP2xjPF+u1gW/2HnVO7nvIa9PG3Gm+fLHvGI= github.com/gin-gonic/gin v1.6.3/go.mod h1:75u5sXoLsGZoRN5Sgbi1eraJ4GU3++wFwWzhwvtwp4M= diff --git a/sdks/python/apache_beam/io/gcp/bigquery.py b/sdks/python/apache_beam/io/gcp/bigquery.py index 314effad5520..40f17bfe9b0e 100644 --- a/sdks/python/apache_beam/io/gcp/bigquery.py +++ b/sdks/python/apache_beam/io/gcp/bigquery.py @@ -366,6 +366,7 @@ def chain_after(result): import uuid import warnings from dataclasses import dataclass +from enum import Enum from typing import Optional from typing import Union @@ -442,6 +443,7 @@ def chain_after(result): __all__ = [ 'TableRowJsonCoder', 'BigQueryDisposition', + 'BigQuerySchemaUpdateOption', 'BigQuerySource', 'BigQuerySink', 'BigQueryQueryPriority', @@ -483,6 +485,35 @@ def chain_after(result): """ 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 @@ def validate_write(disposition): 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 @@ def __init__( 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 @@ def __init__( 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 @@ def __init__( 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 @@ def _compute_method(self, experiments, is_streaming_pipeline): 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 @@ def expand(self, pcoll): 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 @@ def find_in_nested_dict(schema): 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 @@ def serialize(side_inputs): '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 51d13d96b73a..dcadee7f6a1a 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 @@ def test_to_from_runner_api(self): 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 @@ def test_to_from_runner_api(self): 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):