FEAT-#7459: Add methods to get and set backend. (#7460)

Add `get_backend()` to get the backend for a dataframe or series. Add `set_backend()`, and its alias `move_to()`, to set the backend of a dataframe or series.

To implement `set_backend()`, extend `FactoryDispatcher` so that it can dispatch I/O operations to the backend that the user chooses instead of always using `modin.config.Backend`. `set_backend()` can then use `FactoryDispatcher.from_pandas(backend=new_backend)` to get a query compiler with the given backend.

This commit also updates the documentation for "native" execution mode to reflect the updated guidance of using `Backend` to control execution. It also adds examples of using `get_backend()` and `set_backend()`.

Signed-off-by: sfc-gh-mvashishtha <[email protected]>

Author: sfc-gh-mvashishtha

.github/actions/run-core-tests/group_2/action.yml

@@ -18,6 +18,7 @@ runs:
                                                       modin/tests/pandas/dataframe/test_udf.py \
                                                       modin/tests/pandas/dataframe/test_window.py \
                                                       modin/tests/pandas/dataframe/test_pickle.py \
-                                                      modin/tests/pandas/test_repartition.py
+                                                      modin/tests/pandas/test_repartition.py \
+                                                      modin/tests/pandas/test_backend.py
           echo "::endgroup::"
         shell: bash -l {0}

docs/usage_guide/advanced_usage/index.rst

@@ -39,6 +39,9 @@ Additional APIs
 Modin also supports these additional APIs on top of pandas to improve user experience.
 
 - :py:meth:`~modin.pandas.DataFrame.modin.to_pandas` -- convert a Modin DataFrame/Series to a pandas DataFrame/Series.
+- :py:meth:`~modin.pandas.DataFrame.get_backend` -- Get the ``Backend`` :doc:`configuration variable </flow/modin/config>` of this ``DataFrame``.
+- :py:meth:`~modin.pandas.DataFrame.move_to` -- Move data and execution for this ``DataFrame`` to the given ``Backend`` :doc:`configuration variable </flow/modin/config>`. This method is an alias for ``DataFrame.set_backend``.
+- :py:meth:`~modin.pandas.DataFrame.set_backend` -- Move data and execution for this ``DataFrame`` to the given ``Backend`` :doc:`configuration variable </flow/modin/config>`. This method is an alias for ``DatFrame.move_to``.
 - :py:func:`~modin.pandas.io.from_pandas` -- convert a pandas DataFrame to a Modin DataFrame.
 - :py:meth:`~modin.pandas.DataFrame.modin.to_ray` -- convert a Modin DataFrame/Series to a Ray Dataset.
 - :py:func:`~modin.pandas.io.from_ray` -- convert a Ray Dataset to a Modin DataFrame.

docs/usage_guide/optimization_notes/index.rst

@@ -314,7 +314,7 @@ Copy-pastable example, showing how mixing pandas and Modin DataFrames in a singl
   # Possible output: TypeError
 
 
-Using pandas to execute queries in Modin's ``"native"`` execution mode
+Using pandas to execute queries with Modin's ``"Pandas"`` backend
 """"""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""""
 
 By default, Modin distributes the data in a dataframe (or series) and attempts
@@ -323,39 +323,62 @@ to process data for different partitions in parallel.
 However, for certain scenarios, such as handling small datasets, Modin's
 parallel execution may introduce unnecessary overhead. In such cases, it's more
 efficient to use serial execution with a single, unpartitioned pandas dataframe.
-You can enable this kind of "native" execution by setting Modin's
-``StorageFormat`` and ``Engine``
-:doc:`configuration variables </flow/modin/config>` to ``"Native"``.
+You can enable this kind of local pandas execution by setting Modin's
+``Backend``
+:doc:`configuration variable </flow/modin/config>` to ``"Pandas"``.
 
-DataFrames created while Modin's global execution mode is set to ``"Native"``
-will continue to use native execution even if you switch the execution mode
+DataFrames created while Modin's global backend is set to ``"Pandas"``
+will continue to use native execution even if you switch the global backend
 later. Modin supports interoperability between distributed Modin DataFrames
-and those using native execution.
+and those using the pandas backend.
 
-Here is an example of using native execution:
+Here is an example of using the pandas backend.
 
 .. code-block:: python
 
   import modin.pandas as pd
   from modin import set_execution
-  from modin.config import StorageFormat, Engine
+  from modin.config import Backend
 
   # This dataframe will use Modin's default, distributed execution.
-  df_distributed_1 = pd.DataFrame([0])
-  assert df_distributed_1._query_compiler.engine != "Native"
+  original_backend = Backend.get()
+  assert original_backend != "Pandas"
+  distributed_df_1 = pd.DataFrame([0])
 
-  # Set execution to "Native" for native execution.
-  original_engine, original_storage_format = set_execution(
-    engine="Native",
-    storage_format="Native"
-  )
-  native_df = pd.DataFrame([1])
-  assert native_df._query_compiler.engine == "Native"
+  # Set backend to "Pandas" for local pandas execution.
+  Backend.put("Pandas")
+  modin_on_pandas_df = pd.DataFrame([1])
+  assert modin_on_pandas_df.get_backend() == "Pandas"
 
   # Revert to default settings for distributed execution
-  set_execution(engine=original_engine, storage_format=original_storage_format)
-  df_distributed_2 = pd.DataFrame([2])
-  assert df_distributed_2._query_compiler.engine == original_engine
+  Backend.put(original_backend)
+  distributed_df_2 = pd.DataFrame([2])
+  assert distributed_df_2.get_backend() == original_backend
+
+You can also use the pandas backend for some dataframes while using different
+backends for other dataframes. You can switch the backend of an individual
+dataframe or series with ``set_backend()`` or its synonym ``move_to()``.
+Here's an example of switching the backend for an individual dataframe.
+
+.. code-block:: python
+
+  import modin.pandas as pd
+
+  # This dataframe will use Modin's default, distributed execution.
+  original_backend = Backend.get()
+  assert original_backend != "Pandas"
+  distributed_df_1 = pd.DataFrame([0])
+
+  pandas_df_1 = distributed_df_1.move_to("Pandas")
+  assert pandas_df_1.get_backend() == "Pandas"
+  pandas_df_1 = pandas_df_1.sort_values(0)
+  assert pandas_df_1.get_backend() == "Pandas"
+
+  new_df = pandas_df_1.move_to(original_backend)
+  assert new_df.get_backend() == original_backend
+
+  new_df.set_backend("Pandas", inplace=True)
+  assert new_df.get_backend() == "Pandas"
 
 Operation-specific optimizations
 """"""""""""""""""""""""""""""""

environment-dev.yml

@@ -12,6 +12,8 @@ dependencies:
   - psutil>=5.8.0
 
   # optional dependencies
+  # NOTE Keep the ray and dask dependencies in sync with the Linux and Windows
+  # Unidist environment dependencies.
   # ray==2.5.0 broken: https://github.com/conda-forge/ray-packages-feedstock/issues/100
   - ray-core>=2.1.0,!=2.5.0
   - pyarrow>=10.0.1

modin/config/envvars.py

@@ -429,12 +429,7 @@ def put(cls, value: str) -> None:
         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]
+        execution = cls.get_execution_for_backend(value)
         set_execution(execution.engine, execution.storage_format)
 
     @classmethod
@@ -532,12 +527,24 @@ def get_execution_for_backend(cls, backend: str) -> Execution:
         execution : Execution
             The execution for the given backend
         """
-        if backend not in cls._BACKEND_TO_EXECUTION:
+        if not isinstance(backend, str):
+            raise TypeError(
+                "Backend value should be a string, but instead it is "
+                + f"{repr(backend)} of type {type(backend)}."
+            )
+        normalized_value = cls.normalize(backend)
+        if normalized_value not in cls.choices:
+            backend_choice_string = ", ".join(f"'{choice}'" for choice in cls.choices)
+            raise ValueError(
+                f"Unknown backend '{backend}'. Available backends are: "
+                + backend_choice_string
+            )
+        if normalized_value not in cls._BACKEND_TO_EXECUTION:
             raise ValueError(
                 f"Backend '{backend}' has no known execution. Please "
                 + "register an execution for it with Backend.register_backend()."
             )
-        return cls._BACKEND_TO_EXECUTION[backend]
+        return cls._BACKEND_TO_EXECUTION[normalized_value]
 
     @classmethod
     def get(cls) -> str:

modin/core/execution/dispatching/factories/dispatcher.py

@@ -17,9 +17,14 @@
 Dispatcher routes the work to execution-specific functions.
 """
 
-from modin.config import Engine, IsExperimental, StorageFormat
+from typing import Union
+
+from pandas._libs.lib import NoDefault, no_default
+
+from modin.config import Backend, Engine, IsExperimental, StorageFormat
 from modin.core.execution.dispatching.factories import factories
-from modin.utils import _inherit_docstrings, get_current_execution
+from modin.core.storage_formats.base import BaseQueryCompiler
+from modin.utils import _inherit_docstrings
 
 
 class FactoryNotFoundError(AttributeError):
@@ -110,28 +115,39 @@ class FactoryDispatcher(object):
     def get_factory(cls) -> factories.BaseFactory:
         """Get current factory."""
         if cls.__factory is None:
-            from modin.pandas import _update_engine
 
-            Engine.subscribe(_update_engine)
-            Engine.subscribe(cls._update_factory)
-            StorageFormat.subscribe(cls._update_factory)
-        return cls.__factory
+            from modin.pandas import _initialize_engine
+
+            Engine.subscribe(
+                lambda engine_parameter: _initialize_engine(engine_parameter.get())
+            )
+            Backend.subscribe(cls._update_factory)
+        return_value = cls.__factory
+        return return_value
 
     @classmethod
-    def _update_factory(cls, *args):
+    def _get_prepared_factory_for_backend(cls, backend) -> factories.BaseFactory:
         """
-        Update and prepare factory with a new one specified via Modin config.
+        Get factory for the specified backend.
 
         Parameters
         ----------
-        *args : iterable
-            This parameters serves the compatibility purpose.
-            Does not affect the result.
+        backend : str
+            Backend name.
+
+        Returns
+        -------
+        factories.BaseFactory
+            Factory for the specified backend.
         """
-        factory_name = get_current_execution() + "Factory"
+        execution = Backend.get_execution_for_backend(backend)
+        from modin.pandas import _initialize_engine
+
+        _initialize_engine(execution.engine)
+        factory_name = f"{execution.storage_format}On{execution.engine}Factory"
         experimental_factory_name = "Experimental" + factory_name
         try:
-            cls.__factory = getattr(factories, factory_name, None) or getattr(
+            factory = getattr(factories, factory_name, None) or getattr(
                 factories, experimental_factory_name
             )
         except AttributeError:
@@ -145,26 +161,54 @@ def _update_factory(cls, *args):
                 raise FactoryNotFoundError(
                     msg.format(factory_name, experimental_factory_name)
                 )
-            cls.__factory = StubFactory.set_failing_name(factory_name)
+            factory = StubFactory.set_failing_name(factory_name)
         else:
             try:
-                cls.__factory.prepare()
+                factory.prepare()
             except ModuleNotFoundError as err:
-                # incorrectly initialized, should be reset to None again
-                # so that an unobvious error does not appear in the following code:
-                # "AttributeError: 'NoneType' object has no attribute 'from_non_pandas'"
-                cls.__factory = None
                 raise ModuleNotFoundError(
                     f"Make sure all required packages are installed: {str(err)}"
                 ) from err
-            except BaseException:
-                cls.__factory = None
-                raise
+        return factory
 
     @classmethod
-    @_inherit_docstrings(factories.BaseFactory._from_pandas)
-    def from_pandas(cls, df):
-        return cls.get_factory()._from_pandas(df)
+    def _update_factory(cls, *args):
+        """
+        Update and prepare factory with a new one specified via Modin config.
+
+        Parameters
+        ----------
+        *args : iterable
+            This parameters serves the compatibility purpose.
+            Does not affect the result.
+        """
+        cls.__factory = cls._get_prepared_factory_for_backend(Backend.get())
+
+    @classmethod
+    def from_pandas(
+        cls, df, backend: Union[str, NoDefault] = no_default
+    ) -> BaseQueryCompiler:
+        """
+        Create a Modin query compiler from a pandas DataFrame.
+
+        Parameters
+        ----------
+        df : pandas.DataFrame
+            The pandas DataFrame to convert.
+        backend : str or NoDefault, default: NoDefault
+            The backend to use for the resulting query compiler. If NoDefault,
+            use the current global default ``Backend`` from the Modin config.
+
+        Returns
+        -------
+        BaseQueryCompiler
+            A Modin query compiler that wraps the input pandas DataFrame.
+        """
+        return (
+            cls.get_factory()
+            if backend is no_default
+            else cls._get_prepared_factory_for_backend(backend)
+        )._from_pandas(df)
 
     @classmethod
     @_inherit_docstrings(factories.BaseFactory._from_arrow)

modin/pandas/__init__.py

@@ -101,10 +101,10 @@
 
 from modin.config import Parameter
 
-_is_first_update = {}
+_engine_initialized = {}
 
 
-def _update_engine(publisher: Parameter):
+def _initialize_engine(engine_string: str):
     from modin.config import (
         CpuCount,
         Engine,
@@ -116,25 +116,25 @@ def _update_engine(publisher: Parameter):
     # Set this so that Pandas doesn't try to multithread by itself
     os.environ["OMP_NUM_THREADS"] = "1"
 
-    if publisher.get() == "Ray":
-        if _is_first_update.get("Ray", True):
+    if engine_string == "Ray":
+        if not _engine_initialized.get("Ray", False):
             from modin.core.execution.ray.common import initialize_ray
 
             initialize_ray()
-    elif publisher.get() == "Dask":
-        if _is_first_update.get("Dask", True):
+    elif engine_string == "Dask":
+        if not _engine_initialized.get("Dask", False):
             from modin.core.execution.dask.common import initialize_dask
 
             initialize_dask()
-    elif publisher.get() == "Unidist":
-        if _is_first_update.get("Unidist", True):
+    elif engine_string == "Unidist":
+        if not _engine_initialized.get("Unidist", False):
             from modin.core.execution.unidist.common import initialize_unidist
 
             initialize_unidist()
-    elif publisher.get() not in Engine.NOINIT_ENGINES:
-        raise ImportError("Unrecognized execution engine: {}.".format(publisher.get()))
+    elif engine_string not in Engine.NOINIT_ENGINES:
+        raise ImportError("Unrecognized execution engine: {}.".format(engine_string))
 
-    _is_first_update[publisher.get()] = False
+    _engine_initialized[engine_string] = True
 
 
 from modin.pandas import arrays, errors

modin/pandas/base.py

@@ -67,17 +67,19 @@
 )
 from pandas.core.indexes.api import ensure_index
 from pandas.core.methods.describe import _refine_percentiles
+from pandas.util._decorators import doc
 from pandas.util._validators import (
     validate_ascending,
     validate_bool_kwarg,
     validate_percentile,
 )
 
 from modin import pandas as pd
+from modin.config import Backend, Execution
 from modin.error_message import ErrorMessage
 from modin.logging import ClassLogger, disable_logging
 from modin.pandas.accessor import CachedAccessor, ModinAPI
-from modin.pandas.utils import is_scalar
+from modin.pandas.utils import GET_BACKEND_DOC, SET_BACKEND_DOC, is_scalar
 from modin.utils import _inherit_docstrings, expanduser_path_arg, try_cast_to_pandas
 
 from .utils import _doc_binary_op, is_full_grab_slice
@@ -4388,3 +4390,31 @@ def __array_ufunc__(
 
     # namespace for additional Modin functions that are not available in Pandas
     modin: ModinAPI = CachedAccessor("modin", ModinAPI)
+
+    @doc(SET_BACKEND_DOC, class_name=__qualname__)
+    def set_backend(self, backend: str, inplace: bool = False) -> Optional[Self]:
+        # TODO(https://github.com/modin-project/modin/issues/7467): refactor
+        # to avoid this cyclic import in most places we do I/O, e.g. in
+        # modin/pandas/io.py
+        from modin.core.execution.dispatching.factories.dispatcher import (
+            FactoryDispatcher,
+        )
+
+        pandas_self = self._query_compiler.to_pandas()
+        query_compiler = FactoryDispatcher.from_pandas(df=pandas_self, backend=backend)
+        if inplace:
+            self._update_inplace(query_compiler)
+            return None
+        else:
+            return self.__constructor__(query_compiler=query_compiler)
+
+    move_to = set_backend
+
+    @doc(GET_BACKEND_DOC, class_name=__qualname__)
+    def get_backend(self) -> str:
+        return Backend.get_backend_for_execution(
+            Execution(
+                engine=self._query_compiler.engine,
+                storage_format=self._query_compiler.storage_format,
+            )
+        )