[pull] master from apache:master

.github/trigger_files/beam_PostCommit_Python_Xlang_Gcp_Direct.json

@@ -1,4 +1,4 @@
 {
   "comment": "Modify this file in a trivial way to cause this test suite to run",
-  "modification": 15
+  "modification": 16
 }

.github/workflows/build_wheels.yml

@@ -244,13 +244,13 @@ jobs:
     - name: Install Python
       uses: actions/setup-python@v5
       with:
-        python-version: '3.10'
+        python-version: '3.11'
     - uses: docker/setup-qemu-action@v3
       if: ${{matrix.os_python.arch == 'aarch64'}}
       name: Set up QEMU
     - name: Install cibuildwheel
       # note: sync cibuildwheel version with gradle task sdks:python:bdistPy* steps
-      run: pip install cibuildwheel==2.23.3 setuptools
+      run: pip install cibuildwheel==3.3.1 setuptools
     - name: Build wheel
       # Only build wheel if it is one of the target versions for this platform, otherwise no-op
       if: ${{ contains(matrix.os_python.python, matrix.py_version) }}

.github/workflows/run_rc_validation_java_quickstart.yml

@@ -88,7 +88,7 @@ jobs:
       - name: Run QuickStart Java Flink Runner
         uses: ./.github/actions/gradle-command-self-hosted-action
         with:
-          gradle-command: :runners:flink:1.20:runQuickstartJavaFlinkLocal
+          gradle-command: :runners:flink:2.0:runQuickstartJavaFlinkLocal
           arguments: |
             -Prepourl=${{ env.APACHE_REPO_URL }} \
             -Pver=${{ env.RELEASE_VERSION }}

release/build.gradle.kts

@@ -39,7 +39,7 @@ task("runJavaExamplesValidationTask") {
   dependsOn(":runners:direct-java:runQuickstartJavaDirect")
   dependsOn(":runners:google-cloud-dataflow-java:runQuickstartJavaDataflow")
   dependsOn(":runners:spark:3:runQuickstartJavaSpark")
-  dependsOn(":runners:flink:1.20:runQuickstartJavaFlinkLocal")
+  dependsOn(":runners:flink:2.0:runQuickstartJavaFlinkLocal")
   dependsOn(":runners:direct-java:runMobileGamingJavaDirect")
   if (project.hasProperty("ver") || !project.version.toString().endsWith("SNAPSHOT")) {
     // only run one variant of MobileGaming on Dataflow for nightly

sdks/python/apache_beam/ml/rag/enrichment/bigquery_vector_search.py

@@ -241,7 +241,7 @@ def format_query(self, items: List[EmbeddableItem]) -> str:
                 ARRAY_AGG(
                     STRUCT({"distance, " if self.include_distance else ""}\
  {base_columns_str})
-                ) as embeddable_items
+                ) as chunks
             FROM VECTOR_SEARCH(
                 (SELECT {columns_str}, {self.embedding_column}
                  FROM `{self.table_name}`

sdks/python/apache_beam/ml/rag/ingestion/bigquery.py

@@ -14,8 +14,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import warnings
 from collections.abc import Callable
-from dataclasses import dataclass
 from typing import Any
 from typing import Dict
 from typing import Optional
@@ -28,41 +28,56 @@
 from apache_beam.typehints.row_type import RowTypeConstraint
 
 EmbeddableToDictFn = Callable[[EmbeddableItem], Dict[str, any]]
+# Backward compatibility alias.
+ChunkToDictFn = EmbeddableToDictFn
 
 
-@dataclass
 class SchemaConfig:
-  """Configuration for custom BigQuery schema and row conversion.
-
-  Allows overriding the default schema and row conversion logic for BigQuery
-  vector storage. This enables custom table schemas beyond the default
-  id/embedding/content/metadata structure.
-
-  Attributes:
-      schema: BigQuery TableSchema dict defining the table structure.
-          Example:
-          >>> {
-          ...   'fields': [
-          ...     {'name': 'id', 'type': 'STRING'},
-          ...     {'name': 'embedding', 'type': 'FLOAT64', 'mode': 'REPEATED'},
-          ...     {'name': 'custom_field', 'type': 'STRING'}
-          ...   ]
-          ... }
-      embeddable_to_dict_fn: Function that converts an
-          EmbeddableItem to a dict matching the schema.
-          Takes an EmbeddableItem and returns
-          Dict[str, Any] with keys matching
-          schema fields.
-          Example:
-          >>> def embeddable_to_dict(item: EmbeddableItem) -> Dict[str, Any]:
-          ...     return {
-          ...         'id': item.id,
-          ...         'embedding': item.embedding.dense_embedding,
-          ...         'custom_field': item.metadata.get('custom_field')
-          ...     }
-  """
-  schema: Dict
-  embeddable_to_dict_fn: EmbeddableToDictFn
+  def __init__(
+      self,
+      schema: Dict,
+      embeddable_to_dict_fn: Optional[EmbeddableToDictFn] = None,
+      **kwargs):
+    """Configuration for custom BigQuery schema and row conversion.
+
+    Allows overriding the default schema and row conversion logic for BigQuery
+    vector storage. This enables custom table schemas beyond the default
+    id/embedding/content/metadata structure.
+
+    Args:
+        schema: BigQuery TableSchema dict defining the table structure.
+        embeddable_to_dict_fn: Function that converts an EmbeddableItem to a
+            dict matching the schema. Takes an EmbeddableItem and returns
+            Dict[str, Any] with keys matching schema fields.
+
+    Example with custom schema:
+      >>> schema_config = SchemaConfig(
+      ...   schema={
+      ...     'fields': [
+      ...       {'name': 'id', 'type': 'STRING'},
+      ...       {'name': 'embedding', 'type': 'FLOAT64', 'mode': 'REPEATED'},
+      ...       {'name': 'source_url', 'type': 'STRING'}
+      ...     ]
+      ...   },
+      ...   embeddable_to_dict_fn=lambda item: {
+      ...       'id': item.id,
+      ...       'embedding': item.embedding.dense_embedding,
+      ...       'source_url': item.metadata.get('url')
+      ...   }
+      ... )
+    """
+    self.schema = schema
+    if 'chunk_to_dict_fn' in kwargs:
+      warnings.warn(
+          "chunk_to_dict_fn is deprecated, use embeddable_to_dict_fn",
+          DeprecationWarning,
+          stacklevel=2)
+      embeddable_to_dict_fn = kwargs.pop('chunk_to_dict_fn')
+    if kwargs:
+      raise TypeError(f"Unexpected keyword arguments: {', '.join(kwargs)}")
+    if embeddable_to_dict_fn is None:
+      raise TypeError("SchemaConfig requires embeddable_to_dict_fn")
+    self.embeddable_to_dict_fn = embeddable_to_dict_fn
 
 
 class BigQueryVectorWriterConfig(VectorDatabaseWriteConfig):

sdks/python/apache_beam/typehints/typehints.py

@@ -1462,9 +1462,10 @@ def normalize(x, none_as_type=False):
   # Convert bare builtin types to correct type hints directly
   elif x in _KNOWN_PRIMITIVE_TYPES:
     return _KNOWN_PRIMITIVE_TYPES[x]
-  elif getattr(x, '__module__',
-               None) in ('typing', 'collections', 'collections.abc') or getattr(
-                   x, '__origin__', None) in _KNOWN_PRIMITIVE_TYPES:
+  elif isinstance(x, types.UnionType) or getattr(
+      x, '__module__',
+      None) in ('typing', 'collections', 'collections.abc') or getattr(
+          x, '__origin__', None) in _KNOWN_PRIMITIVE_TYPES:
     beam_type = native_type_compatibility.convert_to_beam_type(x)
     if beam_type != x:
       # We were able to do the conversion.

sdks/python/apache_beam/typehints/typehints_test.py

@@ -1596,6 +1596,22 @@ def test_hint_helper(self):
     self.assertFalse(is_consistent_with(Union[str, int], str))
     self.assertFalse(is_consistent_with(str, NonBuiltInGeneric[str]))
 
+  def test_hint_helper_pipe_union(self):
+    pipe_union = int | None  # pylint: disable=unsupported-binary-operation
+    typing_union = Union[int, None]
+    self.assertTrue(is_consistent_with(int, pipe_union))
+    self.assertTrue(is_consistent_with(type(None), pipe_union))
+    self.assertFalse(is_consistent_with(str, pipe_union))
+    self.assertTrue(
+        is_consistent_with(int, pipe_union) == is_consistent_with(
+            int, typing_union))
+    self.assertTrue(
+        is_consistent_with(str, pipe_union) == is_consistent_with(
+            str, typing_union))
+    pipe_union_2 = int | float  # pylint: disable=unsupported-binary-operation
+    self.assertTrue(is_consistent_with(int, pipe_union_2))
+    self.assertTrue(is_consistent_with(float, pipe_union_2))
+
   def test_positional_arg_hints(self):
     self.assertEqual(typehints.Any, _positional_arg_hints('x', {}))
     self.assertEqual(int, _positional_arg_hints('x', {'x': int}))
@@ -1934,6 +1950,14 @@ def test_pipe_operator_as_union(self):
         native_type_compatibility.convert_to_beam_type(type_a),
         native_type_compatibility.convert_to_beam_type(type_b))
 
+  def test_normalize_pipe_union(self):
+    pipe_union = int | None  # pylint: disable=unsupported-binary-operation
+    normalized = typehints.normalize(pipe_union)
+    self.assertIsInstance(normalized, typehints.UnionConstraint)
+    pipe_union_2 = int | float  # pylint: disable=unsupported-binary-operation
+    normalized_2 = typehints.normalize(pipe_union_2)
+    self.assertIsInstance(normalized_2, typehints.UnionConstraint)
+
 
 class TestNonBuiltInGenerics(unittest.TestCase):
   def test_no_error_thrown(self):

sdks/python/apache_beam/yaml/examples/testing/examples_test.py

@@ -563,8 +563,11 @@ def _wordcount_minimal_test_preprocessor(
   return _wordcount_random_shuffler(test_spec, all_words, env)
 
 
-@YamlExamplesTestSuite.register_test_preprocessor(
-    ['test_wordCountInclude_yaml', 'test_wordCountImport_yaml'])
+@YamlExamplesTestSuite.register_test_preprocessor([
+    'test_wordCountInclude_yaml',
+    'test_wordCountImport_yaml',
+    'test_wordCountInheritance_yaml'
+])
 def _wordcount_jinja_test_preprocessor(
     test_spec: dict, expected: List[str], env: TestEnvironment):
   """
@@ -679,6 +682,7 @@ def _kafka_test_preprocessor(
     'test_anomaly_scoring_yaml',
     'test_wordCountInclude_yaml',
     'test_wordCountImport_yaml',
+    'test_wordCountInheritance_yaml',
     'test_iceberg_to_alloydb_yaml'
 ])
 def _io_write_test_preprocessor(
@@ -1256,8 +1260,11 @@ def _batch_log_analysis_test_preprocessor(
   return test_spec
 
 
-@YamlExamplesTestSuite.register_test_preprocessor(
-    ['test_wordCountInclude_yaml', 'test_wordCountImport_yaml'])
+@YamlExamplesTestSuite.register_test_preprocessor([
+    'test_wordCountInclude_yaml',
+    'test_wordCountImport_yaml',
+    'test_wordCountInheritance_yaml'
+])
 def _jinja_preprocessor(raw_spec_string: str, test_name: str):
   """
   Preprocessor for Jinja-based YAML tests.

sdks/python/apache_beam/yaml/examples/testing/input_data.py

@@ -86,6 +86,11 @@ def word_count_jinja_template_data(test_name: str) -> list[str]:
         'apache_beam/yaml/examples/transforms/jinja/'
         'import/macros/wordCountMacros.yaml'
     ]
+  elif test_name == 'test_wordCountInheritance_yaml':
+    return [
+        'apache_beam/yaml/examples/transforms/jinja/'
+        'inheritance/base/base_pipeline.yaml'
+    ]
   return []
 
 

sdks/python/apache_beam/yaml/examples/transforms/jinja/inheritance/README.md

@@ -0,0 +1,77 @@
+<!--
+    Licensed to the Apache Software Foundation (ASF) under one
+    or more contributor license agreements.  See the NOTICE file
+    distributed with this work for additional information
+    regarding copyright ownership.  The ASF licenses this file
+    to you under the Apache License, Version 2.0 (the
+    "License"); you may not use this file except in compliance
+    with the License.  You may obtain a copy of the License at
+
+      http://www.apache.org/licenses/LICENSE-2.0
+
+    Unless required by applicable law or agreed to in writing,
+    software distributed under the License is distributed on an
+    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+    KIND, either express or implied.  See the License for the
+    specific language governing permissions and limitations
+    under the License.
+-->
+
+# Jinja Inheritance Example
+
+This folder contains an example of how to use Jinja2 inheritance in Beam YAML pipelines.
+
+## Files
+
+*   **base/base_pipeline.yaml**: A complete WordCount pipeline (Read -> Split -> Explode -> Combine -> MapToFields -> Write). It defines a block `extra_steps` between `Explode` and `MapToFields` to allow child pipelines to inject additional transforms.
+*   **wordCountInheritance.yaml**: Extends `base/base_pipeline.yaml` and injects a `Combine` transform into the `extra_steps` block to combine words.
+
+## Running the Example
+
+To run the child pipeline (which includes the inherited base pipeline logic + the new filter):
+
+General setup:
+```sh
+export PIPELINE_FILE=apache_beam/yaml/examples/transforms/jinja/inheritance/wordCountInheritance.yaml
+export KINGLEAR="gs://dataflow-samples/shakespeare/kinglear.txt"
+export TEMP_LOCATION="gs://MY-BUCKET/wordCounts/"
+export PROJECT="MY-PROJECT"
+export REGION="MY-REGION"
+
+cd <PATH_TO_BEAM_REPO>/beam/sdks/python
+```
+
+Multiline Run Example:
+```sh
+python -m apache_beam.yaml.main \
+  --project=${PROJECT} \
+  --region=${REGION} \
+  --yaml_pipeline_file="${PIPELINE_FILE}" \
+  --jinja_variables='{
+    "readFromTextTransform": {"path": "'"${KINGLEAR}"'"},
+    "mapToFieldsSplitConfig": {
+      "language": "python",
+      "fields": {
+        "value": "1"
+      }
+    },
+    "explodeTransform": {"fields": "word"},
+    "combineTransform": {
+      "group_by": "word",
+      "combine": {"value": "sum"}
+    },
+    "mapToFieldsCountConfig": {
+      "language": "python",
+      "fields": {"output": "word + \" - \" + str(value)"}
+    },
+    "writeToTextTransform": {"path": "'"${TEMP_LOCATION}"'"}
+  }'
+```
+
+Single Line Run Example:
+```sh
+python -m apache_beam.yaml.main --project=${PROJECT} --region=${REGION} \
+--yaml_pipeline_file="${PIPELINE_FILE}" --jinja_variables='{"readFromTextTransform":
+{"path": "'"${KINGLEAR}"'"}, "mapToFieldsSplitConfig": {"language": "python", "fields":{"value":"1"}}, "explodeTransform":{"fields":"word"}, "combineTransform":{"group_by":"word", "combine":{"value":"sum"}}, "mapToFieldsCountConfig":{"language": "python", "fields":{"output":"word + \" - \" + str(value)"}}, "writeToTextTransform":{"path":"'"${TEMP_LOCATION}"'"}}'
+```
+

sdks/python/apache_beam/yaml/examples/transforms/jinja/inheritance/base/base_pipeline.yaml

@@ -0,0 +1,55 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+pipeline:
+  type: chain
+  transforms:
+    - type: ReadFromText
+      config:
+        path: {{readFromTextTransform.path}}
+
+    - type: MapToFields
+      name: Split words
+      config:
+        language: python
+        fields:
+          word:
+            callable: |-
+              import re
+              def my_mapping(row):
+                return re.findall(r'[A-Za-z\']+', row.line.lower())
+          value: {{mapToFieldsSplitConfig.fields.value}}
+    - type: Explode
+      config:
+        fields:
+          - {{explodeTransform.fields}}
+
+    # Inheritance injection point: content added here by child pipelines will be executed
+    # after Explode and before MapToFields.
+{% block extra_steps %}
+{% endblock %}
+
+    - type: MapToFields
+      name: Format output
+      config:
+        language: {{mapToFieldsCountConfig.language}}
+        fields:
+          output: {{mapToFieldsCountConfig.fields.output}}
+    - name: Write to GCS
+      type: WriteToText
+      config:
+        path: {{writeToTextTransform.path}}