[pull] master from apache:master

.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'

.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'

.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: |

.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'

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
 

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

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=

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,