Introduce an enum CheckpointingImpl to simplify Pathways registration and to allow multiple options to be specified and resolved in order of priority.

PiperOrigin-RevId: 833441656

Author: cpgaffney1

checkpoint/CHANGELOG.md

@@ -17,6 +17,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Validate checkpoints before writing merged OCDBT database using in-memory
   state, avoiding additional I/O to re-read metadata.
 - add `support_format` to utils.to_shape_dtype_struct()
+- Moved `register_pathways_handlers` to `ocp.pathways.register_type_handlers`.
 
 ## [0.11.28] - 2025-11-06
 

checkpoint/orbax/__init__.py

@@ -32,6 +32,7 @@
 from orbax.checkpoint import msgpack_utils
 from orbax.checkpoint import options
 from orbax.checkpoint import path
+from orbax.checkpoint import pathways
 from orbax.checkpoint import serialization
 from orbax.checkpoint import transform_utils
 from orbax.checkpoint import tree

checkpoint/orbax/checkpoint/__init__.py

@@ -32,6 +32,7 @@
 from orbax.checkpoint import msgpack_utils
 from orbax.checkpoint import options
 from orbax.checkpoint import path
+from orbax.checkpoint import pathways
 from orbax.checkpoint import serialization
 from orbax.checkpoint import transform_utils
 from orbax.checkpoint import tree

checkpoint/orbax/checkpoint/_src/serialization/pathways_handler_registry.py

@@ -14,6 +14,9 @@
 
 """Registers the Pathways handlers with the given options."""
 
+from __future__ import annotations
+
+import enum
 import types
 
 from absl import logging
@@ -73,19 +76,57 @@ def _register_numpy_and_scalar_handlers():
   )
 
 
+class CheckpointingImpl(enum.Enum):
+  """The implementation to use for Pathways checkpointing."""
+
+  NO_DISPATCHER = enum.auto()
+  COLOCATED_PYTHON = enum.auto()
+
+  @classmethod
+  def from_options(
+      cls,
+      *,
+      use_colocated_python: bool = False,
+  ) -> CheckpointingImpl:
+    """Obtains a CheckpointingImpl from the given options.
+
+    More than one option can be set to True. Resolves in order of priority:
+      1. Colocated Python
+      4. No Dispatcher
+
+    Args:
+      use_colocated_python: Whether to use colocated Python. # BEGIN
+      use_remote_python: Whether to use remote Python.
+      use_persistence_array_handler: Whether to use the persistence array
+
+    Returns:
+      The CheckpointingImpl to use.
+    """
+    if use_colocated_python:
+      return cls.COLOCATED_PYTHON
+    else:
+      return cls.NO_DISPATCHER
+
+
 def get_pathways_array_handler(
     use_single_replica_array_handler: bool = False,
-    use_colocated_python: bool = True,
+    checkpointing_impl: CheckpointingImpl | None = None,
     **kwargs,
 ) -> type_handlers.ArrayHandler:
   """Returns the Pathways ArrayHandler with the given options."""
-
-  if use_colocated_python:
-    logging.info('Using ColocatedPythonDispatcher')
-    dispatcher = dispatchers.ColocatedPythonDispatcher()
-  else:
-    logging.info('Not using dispatcher')
-    dispatcher = None
+  # If not set, use whichever dispatcher implementation is available.
+  checkpointing_impl = checkpointing_impl or CheckpointingImpl.from_options(
+      use_colocated_python=True,
+  )
+  match checkpointing_impl:
+    case CheckpointingImpl.COLOCATED_PYTHON:
+      logging.info('Using ColocatedPythonDispatcher')
+      dispatcher = dispatchers.ColocatedPythonDispatcher()
+    case CheckpointingImpl.NO_DISPATCHER:
+      logging.info('Not using dispatcher')
+      dispatcher = None
+    case _:
+      raise ValueError(f'Unsupported CheckpointingImpl: {checkpointing_impl}')
 
   return _get_array_hander_with_dispatcher(
       dispatcher,
@@ -96,15 +137,15 @@ def get_pathways_array_handler(
 
 def register_pathways_handlers(
     use_single_replica_array_handler: bool = False,
-    use_colocated_python: bool = True,
+    checkpointing_impl: CheckpointingImpl | None = None,
     **kwargs,
 ):
   """Registers the Pathways handlers with the given options.
 
   Args:
     use_single_replica_array_handler: Whether to use the
       SingleReplicaArrayHandler.
-    use_colocated_python: Use ColocatedPythonDispatcher with jax array handler.
+    checkpointing_impl: The implementation to use for Pathways checkpointing.
     **kwargs: Keyword arguments to pass to the ArrayHandler.
   """
   _register_numpy_and_scalar_handlers()
@@ -113,7 +154,7 @@ def register_pathways_handlers(
       jax.Array,
       get_pathways_array_handler(
           use_single_replica_array_handler,
-          use_colocated_python,
+          checkpointing_impl=checkpointing_impl,
           **kwargs,
       ),
       override=True,

checkpoint/orbax/checkpoint/_src/testing/benchmarks/pytree_checkpoint_benchmark.py

@@ -124,11 +124,11 @@ def register_array_type_handler(self, options: PyTreeCheckpointOptions):
     else:
       if options.use_jax_array_handler:
         if options.use_persistence_array_handler:
-          ocp.type_handlers.register_pathways_handlers(
+          ocp.pathways.register_type_handlers(
               use_persistence_array_handler=options.use_persistence_array_handler,
           )
         else:
-          ocp.type_handlers.register_pathways_handlers(
+          ocp.pathways.register_type_handlers(
               use_colocated_python=options.use_colocated_python,
               use_replica_parallel=options.use_replica_parallel,
               enable_replica_parallel_separate_folder=options.enable_replica_parallel_separate_folder,

checkpoint/orbax/checkpoint/_src/testing/benchmarks/single_replica_benchmark.py

@@ -119,7 +119,9 @@ def test_fn(
 
     pathways_handler_registry.register_pathways_handlers(
         use_single_replica_array_handler=True,
-        use_colocated_python=options.use_colocated_python,
+        checkpointing_impl=pathways_handler_registry.CheckpointingImpl.from_options(
+            use_colocated_python=options.use_colocated_python,
+        ),
         replica_axis_index=options.replica_axis_index,
         primary_replica_id=options.primary_replica_id,
         use_replica_parallel=options.use_replica_parallel,

checkpoint/orbax/checkpoint/_src/testing/benchmarks/single_replica_benchmark_test.py

@@ -237,7 +237,9 @@ def test_test_fn_handler_options(
     mock_is_runtime_to_distributed_ids_initialized.assert_called_once()
     register_pathways_handlers.assert_called_once_with(
         use_single_replica_array_handler=True,
-        use_colocated_python=options.use_colocated_python,
+        checkpointing_impl=pathways_handler_registry.CheckpointingImpl.from_options(
+            use_colocated_python=options.use_colocated_python,
+        ),
         replica_axis_index=options.replica_axis_index,
         primary_replica_id=options.primary_replica_id,
         use_replica_parallel=options.use_replica_parallel,

checkpoint/orbax/checkpoint/pathways.py

@@ -0,0 +1,20 @@
+# Copyright 2025 The Orbax Authors.
+#
+# Licensed 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.
+
+"""Public symbols for Pathways-related serialization."""
+
+# pylint: disable=g-importing-member, unused-import, g-bad-import-order
+
+from orbax.checkpoint._src.serialization.pathways_handler_registry import CheckpointingImpl
+from orbax.checkpoint._src.serialization.pathways_handler_registry import register_pathways_handlers as register_type_handlers

checkpoint/orbax/checkpoint/type_handlers.py

@@ -34,7 +34,6 @@
 from orbax.checkpoint._src.serialization.type_handler_registry import get_type_handler
 from orbax.checkpoint._src.serialization.type_handler_registry import has_type_handler
 from orbax.checkpoint._src.serialization.type_handler_registry import register_standard_handlers_with_options
-from orbax.checkpoint._src.serialization.pathways_handler_registry import register_pathways_handlers
 from orbax.checkpoint._src.serialization.type_handler_registry import register_type_handler
 from orbax.checkpoint._src.serialization.type_handler_registry import supported_types