Create singleton view for pipeline options during pipeline construction.

.github/trigger_files/beam_PreCommit_Python_Dill.json

@@ -1,4 +1,4 @@
 {
     "comment": "Modify this file in a trivial way to cause this test suite to run",
-    "revision": 2
+    "revision": 3
 }

sdks/python/apache_beam/coders/coders.py

@@ -927,16 +927,16 @@ def _create_impl(self):
 
 class DeterministicFastPrimitivesCoderV2(FastCoder):
   """Throws runtime errors when encoding non-deterministic values."""
-  def __init__(self, coder, step_label, update_compatibility_version=None):
+  def __init__(self, coder, step_label):
     self._underlying_coder = coder
     self._step_label = step_label
     self._use_relative_filepaths = True
     self._version_tag = "v2_69"
-    from apache_beam.transforms.util import is_v1_prior_to_v2
 
     # Versions prior to 2.69.0 did not use relative filepaths.
-    if update_compatibility_version and is_v1_prior_to_v2(
-        v1=update_compatibility_version, v2="2.69.0"):
+    from apache_beam.options.pipeline_construction_options import pipeline_construction_options
+    opts = pipeline_construction_options.options
+    if opts and opts.is_compat_version_prior_to("2.69.0"):
       self._version_tag = ""
       self._use_relative_filepaths = False
 
@@ -1005,20 +1005,20 @@ def to_type_hint(self):
     return Any
 
 
-def _should_force_use_dill(registry):
+def _should_force_use_dill():
+  from apache_beam.options.pipeline_construction_options import pipeline_construction_options
+
   # force_dill_deterministic_coders is for testing purposes. If there is a
   # DeterministicFastPrimitivesCoder in the pipeline graph but the dill
-  # encoding path is not really triggered dill does not have to be installed.
+  # encoding path is not really triggered dill does not have to be installed
   # and this check can be skipped.
-  if getattr(registry, 'force_dill_deterministic_coders', False):
+  if getattr(pipeline_construction_options,
+             'force_dill_deterministic_coders',
+             False):
     return True
 
-  from apache_beam.transforms.util import is_v1_prior_to_v2
-  update_compat_version = registry.update_compatibility_version
-  if not update_compat_version:
-    return False
-
-  if not is_v1_prior_to_v2(v1=update_compat_version, v2="2.68.0"):
+  opts = pipeline_construction_options.options
+  if opts is None or not opts.is_compat_version_prior_to("2.68.0"):
     return False
 
   try:
@@ -1043,12 +1043,9 @@ def _update_compatible_deterministic_fast_primitives_coder(coder, step_label):
    - In SDK version 2.69.0 cloudpickle is used to encode "special types" with
    relative filepaths in code objects and dynamic functions.
   """
-  from apache_beam.coders import typecoders
-
-  if _should_force_use_dill(typecoders.registry):
+  if _should_force_use_dill():
     return DeterministicFastPrimitivesCoder(coder, step_label)
-  return DeterministicFastPrimitivesCoderV2(
-      coder, step_label, typecoders.registry.update_compatibility_version)
+  return DeterministicFastPrimitivesCoderV2(coder, step_label)
 
 
 class FastPrimitivesCoder(FastCoder):

sdks/python/apache_beam/coders/coders_test_common.py

@@ -42,6 +42,8 @@
 from apache_beam.coders import proto2_coder_test_messages_pb2 as test_message
 from apache_beam.coders import typecoders
 from apache_beam.internal import pickler
+from apache_beam.options.pipeline_construction_options import pipeline_construction_options
+from apache_beam.options.pipeline_options import PipelineOptions
 from apache_beam.runners import pipeline_context
 from apache_beam.transforms import userstate
 from apache_beam.transforms import window
@@ -203,7 +205,7 @@ def tearDownClass(cls):
     assert not cls.seen_nested - standard, str(cls.seen_nested - standard)
 
   def tearDown(self):
-    typecoders.registry.update_compatibility_version = None
+    pipeline_construction_options.options = None
 
   @classmethod
   def _observe(cls, coder):
@@ -275,7 +277,8 @@ def test_deterministic_coder(self, compat_version):
     with relative filepaths in code objects and dynamic functions.
     """
 
-    typecoders.registry.update_compatibility_version = compat_version
+    pipeline_construction_options.options = PipelineOptions(
+        update_compatibility_version=compat_version)
     coder = coders.FastPrimitivesCoder()
     if not dill and compat_version == "2.67.0":
       with self.assertRaises(RuntimeError):
@@ -364,7 +367,8 @@ def test_deterministic_map_coder_is_update_compatible(self, compat_version):
     - In SDK version >=2.69.0 cloudpickle is used to encode "special types"
     with relative file.
     """
-    typecoders.registry.update_compatibility_version = compat_version
+    pipeline_construction_options.options = PipelineOptions(
+        update_compatibility_version=compat_version)
     values = [{
         MyTypedNamedTuple(i, 'a'): MyTypedNamedTuple('a', i)
         for i in range(10)
@@ -738,7 +742,8 @@ def test_cross_process_encoding_of_special_types_is_deterministic(
 
     if sys.executable is None:
       self.skipTest('No Python interpreter found')
-    typecoders.registry.update_compatibility_version = compat_version
+    pipeline_construction_options.options = PipelineOptions(
+        update_compatibility_version=compat_version)
 
     # pylint: disable=line-too-long
     script = textwrap.dedent(
@@ -750,7 +755,8 @@ def test_cross_process_encoding_of_special_types_is_deterministic(
         import logging
 
         from apache_beam.coders import coders
-        from apache_beam.coders import typecoders
+        from apache_beam.options.pipeline_construction_options import pipeline_construction_options
+        from apache_beam.options.pipeline_options import PipelineOptions
         from apache_beam.coders.coders_test_common import MyNamedTuple
         from apache_beam.coders.coders_test_common import MyTypedNamedTuple
         from apache_beam.coders.coders_test_common import MyEnum
@@ -802,7 +808,8 @@ def test_cross_process_encoding_of_special_types_is_deterministic(
         ])
 
         compat_version = {'"'+ compat_version +'"' if compat_version else None}
-        typecoders.registry.update_compatibility_version = compat_version
+        pipeline_construction_options.options = PipelineOptions(
+        update_compatibility_version=compat_version)
         coder = coders.FastPrimitivesCoder()
         deterministic_coder = coder.as_deterministic_coder("step")
 

sdks/python/apache_beam/coders/typecoders.py

@@ -84,7 +84,6 @@ def __init__(self, fallback_coder=None):
     self._coders: Dict[Any, Type[coders.Coder]] = {}
     self.custom_types: List[Any] = []
     self.register_standard_coders(fallback_coder)
-    self.update_compatibility_version = None
 
   def register_standard_coders(self, fallback_coder):
     """Register coders for all basic and composite types."""

sdks/python/apache_beam/io/gcp/bigquery_file_loads.py

@@ -1120,8 +1120,8 @@ def _load_data(
          of the load jobs would fail but not other. If any of them fails, then
          copy jobs are not triggered.
     """
-    self.reshuffle_before_load = not util.is_compat_version_prior_to(
-        p.options, "2.65.0")
+    self.reshuffle_before_load = not p.options.is_compat_version_prior_to(
+        "2.65.0")
     if self.reshuffle_before_load:
       # Ensure that TriggerLoadJob retry inputs are deterministic by breaking
       # fusion for inputs.

sdks/python/apache_beam/io/gcp/bigquery_file_loads_test.py

@@ -46,6 +46,7 @@
 from apache_beam.io.gcp.tests.bigquery_matcher import BigqueryFullResultMatcher
 from apache_beam.io.gcp.tests.bigquery_matcher import BigqueryFullResultStreamingMatcher
 from apache_beam.metrics.metric import Lineage
+from apache_beam.options.pipeline_construction_options import pipeline_construction_options
 from apache_beam.options.pipeline_options import PipelineOptions
 from apache_beam.options.pipeline_options import StandardOptions
 from apache_beam.runners.dataflow.test_dataflow_runner import TestDataflowRunner
@@ -486,8 +487,7 @@ def test_records_traverse_transform_with_mocks(self):
       param(compat_version="2.64.0"),
   ])
   def test_reshuffle_before_load(self, compat_version):
-    from apache_beam.coders import typecoders
-    typecoders.registry.force_dill_deterministic_coders = True
+    pipeline_construction_options.force_dill_deterministic_coders = True
     destination = 'project1:dataset1.table1'
 
     job_reference = bigquery_api.JobReference()
@@ -513,17 +513,14 @@ def test_reshuffle_before_load(self, compat_version):
         validate=False,
         temp_file_format=bigquery_tools.FileFormat.JSON)
 
-    options = PipelineOptions(
-        update_compatibility_version=compat_version,
-        # Disable unrelated compatibility change.
-        force_cloudpickle_deterministic_coders=True)
+    options = PipelineOptions(update_compatibility_version=compat_version)
     # Need to test this with the DirectRunner to avoid serializing mocks
     with TestPipeline('DirectRunner', options=options) as p:
       _ = p | beam.Create(_ELEMENTS) | transform
 
     reshuffle_before_load = compat_version is None
     assert transform.reshuffle_before_load == reshuffle_before_load
-    typecoders.registry.force_dill_deterministic_coders = False
+    pipeline_construction_options.force_dill_deterministic_coders = False
 
   def test_load_job_id_used(self):
     job_reference = bigquery_api.JobReference()
@@ -1000,8 +997,7 @@ def dynamic_destination_resolver(element, *side_inputs):
   ])
   def test_triggering_frequency(
       self, is_streaming, with_auto_sharding, compat_version):
-    from apache_beam.coders import typecoders
-    typecoders.registry.force_dill_deterministic_coders = True
+    pipeline_construction_options.force_dill_deterministic_coders = True
 
     destination = 'project1:dataset1.table1'
 
@@ -1108,7 +1104,7 @@ def __call__(self):
           label='CheckDestinations')
       assert_that(jobs, equal_to(expected_jobs), label='CheckJobs')
 
-    typecoders.registry.force_dill_deterministic_coders = False
+    pipeline_construction_options.force_dill_deterministic_coders = False
 
 
 class BigQueryFileLoadsIT(unittest.TestCase):

sdks/python/apache_beam/options/pipeline_construction_options.py

@@ -0,0 +1,37 @@
+#
+# 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.
+#
+
+"""Singleton providing access to pipeline options during graph construction.
+
+This module provides a lightweight singleton that holds the full
+PipelineOptions set during Pipeline.__init__.
+"""
+
+from typing import Optional
+
+from apache_beam.options.pipeline_options import PipelineOptions
+
+
+class PipelineConstructionOptions:
+  """Holds the current pipeline's options during graph construction.
+
+  Set during Pipeline.__init__.
+  """
+  options: Optional[PipelineOptions] = None
+
+
+pipeline_construction_options = PipelineConstructionOptions()

sdks/python/apache_beam/options/pipeline_options.py

@@ -668,6 +668,25 @@ def view_as(self, cls: Type[PipelineOptionsT]) -> PipelineOptionsT:
     view._all_options = self._all_options
     return view
 
+  def is_compat_version_prior_to(self, breaking_change_version):
+    """Check if update_compatibility_version is prior to a breaking change.
+
+    Returns True if the pipeline should use old behavior (i.e., the
+    update_compatibility_version is set and is earlier than the given version).
+    Returns False if update_compatibility_version is not set or is >= the
+    breaking change version.
+
+    Args:
+      breaking_change_version: Version string (e.g., "2.72.0") at which
+        the breaking change was introduced.
+    """
+    v1 = self.view_as(StreamingOptions).update_compatibility_version
+    if v1 is None:
+      return False
+    v1_parts = (v1.split('.') + ['0', '0', '0'])[:3]
+    v2_parts = (breaking_change_version.split('.') + ['0', '0', '0'])[:3]
+    return tuple(map(int, v1_parts)) < tuple(map(int, v2_parts))
+
   def _visible_option_list(self) -> List[str]:
     return sorted(
         option for option in dir(self._visible_options) if option[0] != '_')
@@ -898,18 +917,6 @@ def _add_argparse_args(cls, parser):
         'their condition met. Some operations, such as GroupByKey, disallow '
         'this. This exists for cases where such loss is acceptable and for '
         'backwards compatibility. See BEAM-9487.')
-    parser.add_argument(
-        '--force_cloudpickle_deterministic_coders',
-        default=False,
-        action='store_true',
-        help=(
-            'Force the use of cloudpickle-based deterministic coders '
-            'instead of dill-based coders, even when '
-            'update_compatibility_version  would normally trigger dill usage '
-            'for backward compatibility. This flag overrides automatic coder '
-            'selection to always use the modern cloudpickle serialization '
-            ' path. Warning: May break pipeline update compatibility with '
-            ' SDK versions prior to 2.68.0.'))
 
   def validate(self, unused_validator):
     errors = []

sdks/python/apache_beam/options/pipeline_options_test.py

@@ -987,6 +987,72 @@ def test_comma_separated_dataflow_service_options(self):
                      options.get_all_options()['dataflow_service_options'])
 
 
+class CompatVersionTest(unittest.TestCase):
+  def test_is_compat_version_prior_to(self):
+    test_cases = [
+        # Basic comparison cases
+        ("1.0.0", "2.0.0", True),  # v1 < v2 in major
+        ("2.0.0", "1.0.0", False),  # v1 > v2 in major
+        ("1.1.0", "1.2.0", True),  # v1 < v2 in minor
+        ("1.2.0", "1.1.0", False),  # v1 > v2 in minor
+        ("1.0.1", "1.0.2", True),  # v1 < v2 in patch
+        ("1.0.2", "1.0.1", False),  # v1 > v2 in patch
+
+        # Equal versions
+        ("1.0.0", "1.0.0", False),  # Identical
+        ("0.0.0", "0.0.0", False),  # Both zero
+
+        # Different lengths - shorter vs longer
+        ("1.0", "1.0.0", False),  # Should be equal (1.0 = 1.0.0)
+        ("1.0", "1.0.1", True),  # 1.0.0 < 1.0.1
+        ("1.2", "1.2.0", False),  # Should be equal (1.2 = 1.2.0)
+        ("1.2", "1.2.3", True),  # 1.2.0 < 1.2.3
+        ("2", "2.0.0", False),  # Should be equal (2 = 2.0.0)
+        ("2", "2.0.1", True),  # 2.0.0 < 2.0.1
+        ("1", "2.0", True),  # 1.0.0 < 2.0.0
+
+        # Different lengths - longer vs shorter
+        ("1.0.0", "1.0", False),  # Should be equal
+        ("1.0.1", "1.0", False),  # 1.0.1 > 1.0.0
+        ("1.2.0", "1.2", False),  # Should be equal
+        ("1.2.3", "1.2", False),  # 1.2.3 > 1.2.0
+        ("2.0.0", "2", False),  # Should be equal
+        ("2.0.1", "2", False),  # 2.0.1 > 2.0.0
+        ("2.0", "1", False),  # 2.0.0 > 1.0.0
+
+        # Mixed length comparisons
+        ("1.0", "2.0.0", True),  # 1.0.0 < 2.0.0
+        ("2.0", "1.0.0", False),  # 2.0.0 > 1.0.0
+        ("1", "1.0.1", True),  # 1.0.0 < 1.0.1
+        ("1.1", "1.0.9", False),  # 1.1.0 > 1.0.9
+
+        # Large numbers
+        ("1.9.9", "2.0.0", True),  # 1.9.9 < 2.0.0
+        ("10.0.0", "9.9.9", False),  # 10.0.0 > 9.9.9
+        ("1.10.0", "1.9.0", False),  # 1.10.0 > 1.9.0
+        ("1.2.10", "1.2.9", False),  # 1.2.10 > 1.2.9
+
+        # Sequential versions
+        ("1.0.0", "1.0.1", True),
+        ("1.0.1", "1.0.2", True),
+        ("1.0.9", "1.1.0", True),
+        ("1.9.9", "2.0.0", True),
+    ]
+
+    for v1, v2, expected in test_cases:
+      options = PipelineOptions(update_compatibility_version=v1)
+      self.assertEqual(
+          options.is_compat_version_prior_to(v2),
+          expected,
+          msg=f"Failed {v1} < {v2} == {expected}")
+
+    # None case: no update_compatibility_version set
+    options_no_compat = PipelineOptions()
+    self.assertFalse(
+        options_no_compat.is_compat_version_prior_to("1.0.0"),
+        msg="Should return False when update_compatibility_version is not set")
+
+
 if __name__ == '__main__':
   logging.getLogger().setLevel(logging.INFO)
   unittest.main()

sdks/python/apache_beam/pipeline.py

@@ -74,12 +74,12 @@
 from apache_beam.coders import typecoders
 from apache_beam.internal import pickler
 from apache_beam.io.filesystems import FileSystems
+from apache_beam.options.pipeline_construction_options import pipeline_construction_options
 from apache_beam.options.pipeline_options import CrossLanguageOptions
 from apache_beam.options.pipeline_options import DebugOptions
 from apache_beam.options.pipeline_options import PipelineOptions
 from apache_beam.options.pipeline_options import SetupOptions
 from apache_beam.options.pipeline_options import StandardOptions
-from apache_beam.options.pipeline_options import StreamingOptions
 from apache_beam.options.pipeline_options import TypeOptions
 from apache_beam.options.pipeline_options_validator import PipelineOptionsValidator
 from apache_beam.portability import common_urns
@@ -226,8 +226,7 @@ def __init__(
       raise ValueError(
           'Pipeline has validations errors: \n' + '\n'.join(errors))
 
-    typecoders.registry.update_compatibility_version = self._options.view_as(
-        StreamingOptions).update_compatibility_version
+    pipeline_construction_options.options = self._options
 
     # set default experiments for portable runners
     # (needs to occur prior to pipeline construction)