Move options override to before validation.

Author: Claude

sdks/python/apache_beam/pipeline.py

@@ -204,6 +204,17 @@ def __init__(
           'Runner %s is not a PipelineRunner object or the '
           'name of a registered runner.' % runner)
 
+    # Runner can specify the default runner to be used.
+    if (self._options.view_as(SetupOptions).pickle_library == 'default' and
+        runner.default_pickle_library_override() is not None):
+      logging.info(
+          "Runner defaulting to pickling library: %s.",
+          runner.default_pickle_library_override())
+      self._options.view_as(
+          SetupOptions).pickle_library = runner.default_pickle_library_override(
+          )
+    pickler.set_library(self._options.view_as(SetupOptions).pickle_library)
+
     # Validate pipeline options
     errors = PipelineOptionsValidator(self._options, runner).validate()
     if errors:
@@ -223,15 +234,6 @@ def __init__(
     # Default runner to be used.
     self.runner = runner
 
-    if (self._options.view_as(SetupOptions).pickle_library == 'default' and
-        self.runner.default_pickle_library_override() is not None):
-      logging.info(
-          "Default pickling library set to : %s.",
-          runner.default_pickle_library_override())
-      self._options.view_as(
-          SetupOptions).pickle_library = runner.default_pickle_library_override(
-          )
-    pickler.set_library(self._options.view_as(SetupOptions).pickle_library)
     # Stack of transforms generated by nested apply() calls. The stack will
     # contain a root node as an enclosing (parent) node for top transforms.
     self.transforms_stack = [

sdks/python/apache_beam/pipeline_test.py

@@ -41,6 +41,7 @@
 from apache_beam.portability.api import beam_runner_api_pb2
 from apache_beam.pvalue import AsSingleton
 from apache_beam.pvalue import TaggedOutput
+from apache_beam.runners.direct import direct_runner
 from apache_beam.testing.test_pipeline import TestPipeline
 from apache_beam.testing.util import assert_that
 from apache_beam.testing.util import equal_to
@@ -62,7 +63,6 @@
 from apache_beam.transforms.window import TimestampedValue
 from apache_beam.utils import windowed_value
 from apache_beam.utils.timestamp import MIN_TIMESTAMP
-from apache_beam.runners.direct import direct_runner
 
 
 class FakeUnboundedSource(SourceBase):
@@ -169,7 +169,8 @@ def test_runner_overrides_default_pickler(self, mock_info):
         from apache_beam.internal import pickler
         from apache_beam.internal import dill_pickler
         self.assertIs(pickler.desired_pickle_lib, dill_pickler)
-    mock_info.assert_any_call('Default pickling library set to : %s.', 'dill')
+    mock_info.assert_any_call(
+        'Runner defaulting to pickling library: %s.', 'dill')
 
   def test_flatmap_builtin(self):
     with TestPipeline() as pipeline: