Repository: modin-project/modin
Pull request #7456 (2025-03-06 22:37:11)

FEAT-#7455: Add Backend config variable as an alias for execution.


Patch:
diff --git a/modin/__init__.py b/modin/__init__.py
index 4e1d99cd..e85988a3 100644
--- a/modin/__init__.py
+++ b/modin/__init__.py
@@ -42,7 +42,7 @@ def set_execution(engine: Any = None, storage_format: Any = None) -> Tuple[Any,
 
     The method returns pair of old values, so it is easy to return back.
     """
-    from .config import Engine, StorageFormat
+    from .config import Backend, Engine, Execution, StorageFormat
 
     old_engine, old_storage_format = None, None
     # defer callbacks until both entities are set
@@ -55,6 +55,13 @@ def set_execution(engine: Any = None, storage_format: Any = None) -> Tuple[Any,
         Engine._check_callbacks(old_engine)
     if old_storage_format is not None:
         StorageFormat._check_callbacks(old_storage_format)
+    old_backend = Backend.get()
+    Backend._put_nocallback(
+        Backend.get_backend_for_execution(
+            Execution(engine=Engine.get(), storage_format=StorageFormat.get())
+        )
+    )
+    Backend._check_callbacks(old_backend)
 
     return old_engine, old_storage_format
 
diff --git a/modin/config/__init__.py b/modin/config/__init__.py
index 9863ba86..fe7d5af2 100644
--- a/modin/config/__init__.py
+++ b/modin/config/__init__.py
@@ -17,6 +17,7 @@ from modin.config.envvars import (
     AsvDataSizeConfig,
     AsvImplementation,
     AsyncReadMode,
+    Backend,
     BenchmarkMode,
     CIAWSAccessKeyID,
     CIAWSSecretAccessKey,
@@ -26,6 +27,7 @@ from modin.config.envvars import (
     DynamicPartitioning,
     Engine,
     EnvironmentVariable,
+    Execution,
     GithubCI,
     GpuCount,
     IsDebug,
@@ -70,6 +72,8 @@ __all__ = [
     "CpuCount",
     "GpuCount",
     "Memory",
+    "Backend",
+    "Execution",
     # Ray specific
     "IsRayCluster",
     "RayRedisAddress",
diff --git a/modin/config/envvars.py b/modin/config/envvars.py
index 5d137527..a48fbd0f 100644
--- a/modin/config/envvars.py
+++ b/modin/config/envvars.py
@@ -17,12 +17,14 @@ import os
 import secrets
 import sys
 import warnings
+from collections import namedtuple
 from textwrap import dedent
-from typing import Any, Optional
+from typing import Any, NoReturn, Optional
 
 from packaging import version
 from pandas.util._decorators import doc  # type: ignore[attr-defined]
 
+from modin import set_execution
 from modin.config.pubsub import (
     _TYPE_PARAMS,
     _UNSET,
@@ -251,15 +253,181 @@ class Engine(EnvironmentVariable, type=str):
         cls.has_custom_engine = True
         return choice
 
+    @classmethod
+    def put(cls, value: str) -> None:
+        """
+        Set the engine value.
+
+        Parameters
+        ----------
+        value : str
+            Engine value to set.
+        """
+        value = cls.normalize(value)
+        # Backend.put() will set Engine.
+        Backend.put(
+            Backend.get_backend_for_execution(
+                Execution(engine=value, storage_format=StorageFormat.get())
+            )
+        )
+
 
 class StorageFormat(EnvironmentVariable, type=str):
     """Engine to run on a single node of distribution."""
 
+    @classmethod
+    def put(cls, value: str) -> None:
+        """
+        Set the storage format value.
+
+        Parameters
+        ----------
+        value : str
+            Storage format value to set.
+        """
+        value = cls.normalize(value)
+        # Backend.put() will set StorageFormat.
+        Backend.put(
+            Backend.get_backend_for_execution(
+                Execution(engine=Engine.get(), storage_format=value)
+            )
+        )
+
     varname = "MODIN_STORAGE_FORMAT"
     default = "Pandas"
     choices = ("Pandas", "Native")
 
 
+Execution = namedtuple("Execution", ["storage_format", "engine"])
+
+
+class Backend(EnvironmentVariable, type=str):
+    """
+    An alias for execution, i.e. the combination of StorageFormat and Engine.
+
+    Setting backend may change StorageFormat and/or Engine to the corresponding
+    respective values, and setting Engine or StorageFormat may change Backend.
+
+    Modin's built-in backends include:
+        - "Ray" <-> (StorageFormat="Pandas", Engine="Ray")
+        - "Dask" <-> (StorageFormat="Pandas", Engine="Dask")
+        - "Python_Test" <-> (StorageFormat="Pandas", Engine="Python")
+            - This execution mode is meant for testing only.
+        - "Unidist" <-> (StorageFormat="Pandas", Engine="Unidist")
+        - "Pandas" <-> (StorageFormat="Native", Engine="Native")
+    """
+
+    _BACKEND_TO_EXECUTION: dict[str, Execution] = {}
+    _EXECUTION_TO_BACKEND: dict[Execution, str] = {}
+    varname: str = "MODIN_BACKEND"
+    choices: tuple[str, ...] = ("Ray", "Dask", "Python_Test", "Unidist", "Pandas")
+
+    @classmethod
+    def put(cls, value: str) -> None:
+        """
+        Set the backend value.
+
+        Parameters
+        ----------
+        value : str
+            Backend value to set.
+        """
+        value = cls.normalize(value)
+        if value not in cls.choices:
+            raise ValueError(
+                f"Unknown backend '{value}'. Please register the backend with Backend.register_backend()"
+            )
+        execution = cls._BACKEND_TO_EXECUTION[value]
+        set_execution(execution.engine, execution.storage_format)
+
+    @classmethod
+    def _get_default(cls) -> str:
+        """
+        Get the default backend value.
+
+        Returns
+        -------
+        str
+            Default backend value.
+        """
+        return cls._EXECUTION_TO_BACKEND[
+            Execution(StorageFormat._get_default(), Engine._get_default())
+        ]
+
+    @classmethod
+    def register_backend(cls: type["Backend"], name: str, execution: Execution) -> None:
+        """
+        Register a new backend.
+
+        Parameters
+        ----------
+        name : str
+            Backend name.
+        execution : Execution
+            Execution that corresponds to the backend.
+        """
+        name = cls.normalize(name)
+        super().add_option(name)
+        if name in cls._BACKEND_TO_EXECUTION:
+            raise ValueError(
+                f"Backend '{name}' is already registered with the execution {cls._BACKEND_TO_EXECUTION[name]}."
+            )
+        if execution in cls._EXECUTION_TO_BACKEND:
+            raise ValueError(
+                f"{execution} is already registered with the backend {cls._EXECUTION_TO_BACKEND[execution]}."
+            )
+        cls._BACKEND_TO_EXECUTION[name] = execution
+        cls._EXECUTION_TO_BACKEND[execution] = name
+
+    @classmethod
+    def add_option(cls, choice: str) -> NoReturn:
+        """
+        Raise an exception for trying to add an option to Backend directly.
+
+        Parameters
+        ----------
+        choice : str
+            Choice to add. Unused.
+
+        Raises
+        ------
+        ValueError
+            Always.
+        """
+        raise ValueError(
+            "Cannot add an option to Backend directly. Use Backend.register_backend instead."
+        )
+
+    @classmethod
+    def get_backend_for_execution(cls, execution: Execution) -> str:
+        """
+        Get the backend for the execution.
+
+        Parameters
+        ----------
+        execution : Execution
+            Execution to get the backend for.
+
+        Returns
+        -------
+        str
+            Backend for the execution.
+        """
+        if execution not in cls._EXECUTION_TO_BACKEND:
+            raise ValueError(
+                f"{execution} has no known backend. Please register a "
+                + "backend for it with Backend.register_backend()"
+            )
+        return cls._EXECUTION_TO_BACKEND[execution]
+
+
+Backend.register_backend("Ray", Execution("Pandas", "Ray"))
+Backend.register_backend("Dask", Execution("Pandas", "Dask"))
+Backend.register_backend("Python_Test", Execution("Pandas", "Python"))
+Backend.register_backend("Unidist", Execution("Pandas", "Unidist"))
+Backend.register_backend("Pandas", Execution("Native", "Native"))
+
+
 class IsExperimental(EnvironmentVariable, type=bool):
     """Whether to Turn on experimental features."""
 
diff --git a/modin/config/pubsub.py b/modin/config/pubsub.py
index 5e1ce2c5..f3af1700 100644
--- a/modin/config/pubsub.py
+++ b/modin/config/pubsub.py
@@ -367,6 +367,23 @@ class Parameter(object):
         cls._check_callbacks(cls._put_nocallback(value))
         cls._value_source = ValueSource.SET_BY_USER
 
+    @classmethod
+    def normalize(cls, value: Any) -> Any:
+        """
+        Normalize config value.
+
+        Parameters
+        ----------
+        value : Any
+            Config value to normalize.
+
+        Returns
+        -------
+        Any
+            Normalized config value.
+        """
+        return _TYPE_PARAMS[cls.type].normalize(value)
+
     @classmethod
     def once(cls, onvalue: Any, callback: Callable) -> None:
         """
@@ -382,7 +399,7 @@ class Parameter(object):
         callback : callable
             Callable that should be executed if config value matches `onvalue`.
         """
-        onvalue = _TYPE_PARAMS[cls.type].normalize(onvalue)
+        onvalue = cls.normalize(onvalue)
         if onvalue == cls.get():
             callback(cls)
         else:
@@ -405,7 +422,7 @@ class Parameter(object):
         """
         if not _TYPE_PARAMS[cls.type].verify(value):
             raise ValueError(f"Unsupported value: {value}")
-        value = _TYPE_PARAMS[cls.type].normalize(value)
+        value = cls.normalize(value)
         oldvalue, cls._value = cls.get(), value
         return oldvalue
 
@@ -444,7 +461,7 @@ class Parameter(object):
         if cls.choices is not None:
             if not _TYPE_PARAMS[cls.type].verify(choice):
                 raise ValueError(f"Unsupported choice value: {choice}")
-            choice = _TYPE_PARAMS[cls.type].normalize(choice)
+            choice = cls.normalize(choice)
             if choice not in cls.choices:
                 cls.choices += (choice,)
             return choice


Test patch:
diff --git a/modin/conftest.py b/modin/conftest.py
index cc7f87f3..c1939fc7 100644
--- a/modin/conftest.py
+++ b/modin/conftest.py
@@ -30,6 +30,8 @@ import requests
 import s3fs
 from pandas.util._decorators import doc
 
+from modin.config import Backend, Execution
+
 assert (
     "modin.utils" not in sys.modules
 ), "Do not import modin.utils before patching, or tests could fail"
@@ -217,6 +219,13 @@ class BaseOnPythonFactory(factories.BaseFactory):
 
 def set_base_execution(name=BASE_EXECUTION_NAME):
     setattr(factories, f"{name}Factory", BaseOnPythonFactory)
+    Backend.register_backend(
+        "BaseOnPython",
+        Execution(
+            engine="Python",
+            storage_format="Base",
+        ),
+    )
     modin.set_execution(engine="python", storage_format=name.split("On")[0])
 
 
@@ -242,6 +251,9 @@ def get_unique_base_execution():
 
     # Setting up the new execution
     setattr(factories, f"{execution_name}Factory", base_factory)
+    Backend.register_backend(
+        "execution_name", Execution(engine=engine_name, storage_format=format_name)
+    )
     old_engine, old_format = modin.set_execution(
         engine=engine_name, storage_format=format_name
     )
diff --git a/modin/tests/config/test_envvars.py b/modin/tests/config/test_envvars.py
index 17ac0b58..c0c9c0d7 100644
--- a/modin/tests/config/test_envvars.py
+++ b/modin/tests/config/test_envvars.py
@@ -12,16 +12,40 @@
 # governing permissions and limitations under the License.
 
 import os
+import re
 import sys
+from unittest.mock import Mock, patch
 
 import pandas
 import pytest
+from pytest import param
 
 import modin.config as cfg
 import modin.pandas as pd
 from modin.config.envvars import _check_vars
 from modin.config.pubsub import _UNSET, ExactStr
 from modin.pandas.base import BasePandasDataset
+from modin.tests.pandas.utils import switch_execution
+
+################# WARNING #####################################################
+# Test cases in this file affect global state, e.g. by setting environment
+# variables. The test cases may produce unexpected results when repeated on run
+# out of the order they are defined in. Be careful when running the test
+# locally or when adding new test cases. In particular, note:
+#   - test_ray_cluster_resources() causes us to permanently attach the
+#     `_update_engine` subscriber to Engine: https://github.com/modin-project/modin/blob/6252ebde19935bd1f6a6850209bf8a1f5e5ecfb7/modin/core/execution/dispatching/factories/dispatcher.py#L115
+#     Changing to any engine after that test runs will cause Modin to try to
+#     initialize the engine.
+#   - In CI, we only run these tests with Ray execution, in the
+#     `test-internal` job.
+#   - test_wrong_values() permanently messes up some config variables. For more
+#     details see https://github.com/modin-project/modin/issues/7454
+################# WARNING ######################
+
+UNIDIST_SKIP_REASON = (
+    "Switching to unidist causes an error since we have to execute unidist "
+    + "tests differently, with `mpiexec` instead of just `pytest`"
+)
 
 
 def reset_vars(*vars: tuple[cfg.Parameter]):
@@ -217,6 +241,259 @@ def test_context_manager_update_config(modify_config):
     assert cfg.LazyExecution.get() == "Auto"
 
 
+class TestBackend:
+
+    @pytest.mark.parametrize(
+        "engine, storage_format, expected_backend",
+        [
+            ("Python", "Pandas", "Python_Test"),
+            ("Ray", "Pandas", "Ray"),
+            param(
+                "Unidist",
+                "Pandas",
+                "Unidist",
+                marks=pytest.mark.skip(reason=UNIDIST_SKIP_REASON),
+            ),
+            ("Dask", "Pandas", "Dask"),
+            ("Native", "Native", "Pandas"),
+        ],
+    )
+    def test_setting_execution_changes_backend(
+        self, engine, storage_format, expected_backend
+    ):
+        previous_backend = cfg.Backend.get()
+        with switch_execution(engine, storage_format):
+            assert cfg.Backend.get() == expected_backend
+        assert cfg.Backend.get() == previous_backend
+
+    def test_subscribing_to_backend_triggers_callback(self):
+        backend_subscriber = Mock()
+        cfg.Backend.subscribe(backend_subscriber)
+        backend_subscriber.assert_called_once_with(cfg.Backend)
+
+    def test_setting_backend_triggers_all_callbacks(self):
+        # Start with a known backend (rather than the one that we start the
+        # test with).
+        with cfg.context(Backend="Pandas"):
+            backend_subscriber = Mock()
+            cfg.Backend.subscribe(backend_subscriber)
+            backend_subscriber.reset_mock()
+
+            storage_format_subscriber = Mock()
+            cfg.StorageFormat.subscribe(storage_format_subscriber)
+            storage_format_subscriber.reset_mock()
+
+            engine_subscriber = Mock()
+            cfg.Engine.subscribe(engine_subscriber)
+            engine_subscriber.reset_mock()
+
+            with cfg.context(Backend="Python_Test"):
+                backend_subscriber.assert_called_once_with(cfg.Backend)
+                storage_format_subscriber.assert_called_once_with(cfg.StorageFormat)
+                engine_subscriber.assert_called_once_with(cfg.Engine)
+
+    @pytest.mark.parametrize(
+        "backend, expected_engine, expected_storage_format",
+        [
+            ("Python_test", "Python", "Pandas"),
+            ("PYTHON_test", "Python", "Pandas"),
+            ("python_TEST", "Python", "Pandas"),
+            ("Ray", "Ray", "Pandas"),
+            param(
+                "Unidist",
+                "Unidist",
+                "Pandas",
+                marks=pytest.mark.skip(reason=UNIDIST_SKIP_REASON),
+            ),
+            ("Dask", "Dask", "Pandas"),
+            ("Pandas", "Native", "Native"),
+        ],
+    )
+    def test_setting_backend_changes_execution(
+        self, backend, expected_engine, expected_storage_format
+    ):
+        previous_engine = cfg.Engine.get()
+        previous_storage_format = cfg.StorageFormat.get()
+        with cfg.context(Backend=backend):
+            assert cfg.Engine.get() == expected_engine
+            assert cfg.StorageFormat.get() == expected_storage_format
+        assert cfg.Engine.get() == previous_engine
+        assert cfg.StorageFormat.get() == previous_storage_format
+
+    def test_setting_engine_alone_changes_backend(self):
+        # Start with a known backend (rather than the one that we start the
+        # test with).
+        with switch_execution(storage_format="Pandas", engine="Ray"):
+            current_backend = cfg.Backend.get()
+            assert current_backend == "Ray"
+            with cfg.context(Engine="Python"):
+                assert cfg.Backend.get() == "Python_Test"
+            assert cfg.Backend.get() == current_backend
+
+    def test_setting_engine_triggers_callbacks(self):
+        # Start with a known backend (rather than the one that we start the
+        # test with).
+        with switch_execution(storage_format="Pandas", engine="Ray"):
+            engine_subscriber = Mock()
+            cfg.Engine.subscribe(engine_subscriber)
+            engine_subscriber.reset_mock()
+
+            backend_subscriber = Mock()
+            cfg.Backend.subscribe(backend_subscriber)
+            backend_subscriber.reset_mock()
+
+            storage_format_subscriber = Mock()
+            cfg.StorageFormat.subscribe(storage_format_subscriber)
+            storage_format_subscriber.reset_mock()
+
+            with cfg.context(Engine="Dask"):
+                engine_subscriber.assert_called_once_with(cfg.Engine)
+                backend_subscriber.assert_called_once_with(cfg.Backend)
+                # StorageFormat stayed the same, so we don't call its callback.
+                storage_format_subscriber.assert_not_called()
+
+    def test_setting_storage_format_triggers_callbacks(self):
+        # There's only one built-in storage format, pandas, so we add a new one
+        # here.
+        cfg.StorageFormat.add_option("Pandasduplicate")
+        from modin.core.execution.dispatching.factories import factories
+
+        factories.PandasduplicateOnRayFactory = factories.PandasOnRayFactory
+        cfg.Backend.register_backend(
+            "NewBackend",
+            cfg.Execution(
+                storage_format="Pandasduplicate",
+                engine="Ray",
+            ),
+        )
+
+        with switch_execution(storage_format="Pandas", engine="Ray"):
+            engine_subscriber = Mock()
+            cfg.Engine.subscribe(engine_subscriber)
+            engine_subscriber.reset_mock()
+            backend_subscriber = Mock()
+            cfg.Backend.subscribe(backend_subscriber)
+            backend_subscriber.reset_mock()
+            storage_format_subscriber = Mock()
+            cfg.StorageFormat.subscribe(storage_format_subscriber)
+            storage_format_subscriber.reset_mock()
+            with cfg.context(StorageFormat="PANDASDUPLICATE"):
+                storage_format_subscriber.assert_called_once_with(cfg.StorageFormat)
+                backend_subscriber.assert_called_once_with(cfg.Backend)
+                # Engine stayed the same, so we don't call its callback.
+                engine_subscriber.assert_not_called()
+
+    @pytest.mark.parametrize("name", ["Python_Test", "python_Test"])
+    def test_register_existing_backend(self, name):
+        with pytest.raises(
+            ValueError,
+            match=re.escape(
+                "Backend 'Python_Test' is already registered with the execution "
+                + "Execution(storage_format='Pandas', engine='Python')"
+            ),
+        ):
+            cfg.Backend.register_backend(
+                name,
+                cfg.Execution(
+                    storage_format="Pandas",
+                    engine="Python",
+                ),
+            )
+
+    def test_register_existing_execution(self):
+        with pytest.raises(
+            ValueError,
+            match=re.escape(
+                "Execution(storage_format='Pandas', engine='Python') is already registered with the backend Python_Test."
+            ),
+        ):
+            cfg.Backend.register_backend(
+                "NewBackend2",
+                cfg.Execution(
+                    storage_format="Pandas",
+                    engine="Python",
+                ),
+            )
+
+    def test_set_invalid_backend(self):
+        with pytest.raises(ValueError, match=re.escape("Unknown backend 'Unknown'")):
+            cfg.Backend.put("Unknown")
+
+    def test_switch_to_unregistered_backend_with_switch_execution(self):
+        cfg.StorageFormat.add_option("Pandas2")
+        from modin.core.execution.dispatching.factories import factories
+
+        factories.Pandas2OnRayFactory = factories.PandasOnRayFactory
+        with pytest.raises(
+            ValueError,
+            match=re.escape(
+                "Execution(storage_format='Pandas2', engine='Ray') "
+                + "has no known backend. Please register a backend for it with "
+                + "Backend.register_backend()"
+            ),
+        ), switch_execution(engine="Ray", storage_format="Pandas2"):
+            pass
+
+    def test_switch_to_unregistered_backend_with_switch_storage_format(self):
+        cfg.StorageFormat.add_option("Pandas3")
+        from modin.core.execution.dispatching.factories import factories
+
+        factories.Pandas2OnRayFactory = factories.PandasOnPythonFactory
+        with cfg.context(StorageFormat="Pandas", Engine="Python"):
+            with pytest.raises(
+                ValueError,
+                match=re.escape(
+                    "Execution(storage_format='Pandas3', engine='Python') "
+                    + "has no known backend. Please register a backend for it with "
+                    + "Backend.register_backend()"
+                ),
+            ):
+                cfg.StorageFormat.put("Pandas3")
+
+    def test_switch_to_unregistered_backend_with_switch_engine(self):
+        cfg.Engine.add_option("Python2")
+        from modin.core.execution.dispatching.factories import factories
+
+        factories.PandasOnPython2Factory = factories.PandasOnPythonFactory
+        with cfg.context(StorageFormat="Pandas", Engine="Python"):
+            with pytest.raises(
+                ValueError,
+                match=re.escape(
+                    "Execution(storage_format='Pandas', engine='Python2') "
+                    + "has no known backend. Please register a backend for it with "
+                    + "Backend.register_backend()"
+                ),
+            ):
+                cfg.Engine.put("Python2")
+
+    # The default engine and storage format, and hence the default backend,
+    # will depend on which engines are available in the current environment.
+    # For simplicity, patch the defaults.
+    @patch(
+        target="modin.config.StorageFormat._get_default",
+    )
+    @patch(
+        target="modin.config.Engine._get_default",
+    )
+    def test_backend_default(
+        self,
+        mocked_get_default,
+        mocked_get_default2,
+    ):
+        mocked_get_default.return_value = "Native"
+        mocked_get_default2.return_value = "Native"
+        assert cfg.Backend._get_default() == "Pandas"
+
+    def test_add_backend_option(self):
+        with pytest.raises(
+            ValueError,
+            match=re.escape(
+                "Cannot add an option to Backend directly. Use Backend.register_backend instead."
+            ),
+        ):
+            cfg.Backend.add_option("NewBackend")
+
+
 @pytest.mark.parametrize(
     "config_name",
     [
diff --git a/modin/tests/core/test_dispatcher.py b/modin/tests/core/test_dispatcher.py
index a044e1aa..69a2d3a6 100644
--- a/modin/tests/core/test_dispatcher.py
+++ b/modin/tests/core/test_dispatcher.py
@@ -16,8 +16,7 @@ from contextlib import contextmanager
 import pytest
 
 import modin.pandas as pd
-from modin import set_execution
-from modin.config import Engine, Parameter, StorageFormat
+from modin.config import Backend, Engine, Execution, Parameter, StorageFormat
 from modin.core.execution.dispatching.factories import factories
 from modin.core.execution.dispatching.factories.dispatcher import (
     FactoryDispatcher,
@@ -27,15 +26,7 @@ from modin.core.execution.python.implementations.pandas_on_python.io import (
     PandasOnPythonIO,
 )
 from modin.core.storage_formats.pandas.query_compiler import PandasQueryCompiler
-
-
-@contextmanager
-def _switch_execution(engine: str, storage_format: str):
-    old_engine, old_storage = set_execution(engine, storage_format)
-    try:
-        yield
-    finally:
-        set_execution(old_engine, old_storage)
+from modin.tests.pandas.utils import switch_execution
 
 
 @contextmanager
@@ -91,6 +82,36 @@ factories.PandasOnTestFactory = PandasOnTestFactory
 factories.TestOnPythonFactory = TestOnPythonFactory
 factories.FooOnBarFactory = FooOnBarFactory
 
+Backend.register_backend(
+    "Test1",
+    Execution(
+        engine="Test",
+        storage_format="Pandas",
+    ),
+)
+
+Backend.register_backend(
+    "Test2",
+    Execution(
+        engine="Python",
+        storage_format="Test",
+    ),
+)
+Backend.register_backend(
+    "Test3",
+    Execution(
+        engine="Bar",
+        storage_format="Foo",
+    ),
+)
+Backend.register_backend(
+    "Test4",
+    Execution(
+        engine="Dask",
+        storage_format="Pyarrow",
+    ),
+)
+
 # register them as known "no init" engines for modin.pandas
 Engine.NOINIT_ENGINES |= {"Test", "Bar"}
 
@@ -101,7 +122,7 @@ def test_default_factory():
 
 
 def test_factory_switch():
-    with _switch_execution("Python", "Pandas"):
+    with switch_execution("Python", "Pandas"):
         with _switch_value(Engine, "Test"):
             assert FactoryDispatcher.get_factory() == PandasOnTestFactory
             assert FactoryDispatcher.get_factory().io_cls == "Foo"
@@ -119,7 +140,7 @@ def test_engine_wrong_factory():
 
 
 def test_set_execution():
-    with _switch_execution("Bar", "Foo"):
+    with switch_execution("Bar", "Foo"):
         assert FactoryDispatcher.get_factory() == FooOnBarFactory
 
 
@@ -132,7 +153,14 @@ def test_add_option():
     factories.StorageOnExecFactory = DifferentlyNamedFactory
     StorageFormat.add_option("sToragE")
     Engine.add_option("Exec")
-
-    with _switch_execution("Exec", "Storage"):
+    Backend.register_backend(
+        name="Test5",
+        execution=Execution(
+            engine="Exec",
+            storage_format="Storage",
+        ),
+    )
+
+    with switch_execution("Exec", "Storage"):
         df = pd.DataFrame([[1, 2, 3], [3, 4, 5], [5, 6, 7]])
         assert isinstance(df._query_compiler, PandasQueryCompiler)
diff --git a/modin/tests/pandas/utils.py b/modin/tests/pandas/utils.py
index 1d80720b..067953e6 100644
--- a/modin/tests/pandas/utils.py
+++ b/modin/tests/pandas/utils.py
@@ -19,6 +19,7 @@ import itertools
 import math
 import os
 import re
+from contextlib import contextmanager
 from io import BytesIO
 from pathlib import Path
 from string import ascii_letters
@@ -39,6 +40,7 @@ from pandas.core.dtypes.common import (
 )
 
 import modin.pandas as pd
+from modin import set_execution
 from modin.config import (
     Engine,
     MinColumnPartitionSize,
@@ -1681,3 +1683,14 @@ def dict_equals(dict1, dict2):
     for key1, key2 in itertools.zip_longest(sorted(dict1), sorted(dict2)):
         value_equals(key1, key2)
         value_equals(dict1[key1], dict2[key2])
+
+
+@contextmanager
+def switch_execution(engine: str, storage_format: str):
+    old_engine = Engine.get()
+    old_storage = StorageFormat.get()
+    try:
+        set_execution(engine, storage_format)
+        yield
+    finally:
+        set_execution(old_engine, old_storage)