Repository: modin-project/modin
Pull request: FEAT-#7477: Move the query compiler calculator so it can be used in more places
Move the query casting calculator out to the base directory so it can be used by the API layer as well.
Change the calculator to use backend strings instead of query compilers directly, and change the
caster to use the FactoryDispatch methods to create a new calculator, negating the need to retain
the data frame from each query compiler.

- [x] first commit message and PR title follow format outlined [here](https://modin.readthedocs.io/en/latest/development/contributing.html#commit-message-formatting)
  > **_NOTE:_**  If you edit the PR title to match this format, you need to add another commit (even if it's empty) or amend your last commit for the CI job that checks the PR title to pick up the new PR title.
- [x] passes `flake8 modin/ asv_bench/benchmarks scripts/doc_checker.py`
- [x] passes `black --check modin/ asv_bench/benchmarks scripts/doc_checker.py`
- [x] signed commit with `git commit -s` <!-- you can amend your commit with a signature via `git commit -amend -s` -->
- [x] Resolves #7477 ? <!-- issue must be created for each patch -->
- [x] tests added and passing
- [x] module layout described at `docs/development/architecture.rst` is up-to-date <!-- if you have added, renamed or removed files or directories please update the documentation accordingly -->


Problem statement:
Move the costing calculator out so it can be used and refactored more easily
We need to move the query costing calculator out so it can be used by other approaches to casting and engine-switch; such as those which are implemented in the api layer.

Hints from issue discussion:

Commit message: lint

Patch to review:
diff --git a/modin/core/storage_formats/base/query_compiler.py b/modin/core/storage_formats/base/query_compiler.py
index 259be854caa..3a1fe6d7d94 100644
--- a/modin/core/storage_formats/base/query_compiler.py
+++ b/modin/core/storage_formats/base/query_compiler.py
@@ -31,7 +31,7 @@
 from pandas._typing import DtypeBackend, IndexLabel, Suffixes
 from pandas.core.dtypes.common import is_number, is_scalar
 
-from modin.config import Backend, Execution
+from modin.config.envvars import Backend, Execution
 from modin.core.dataframe.algebra.default2pandas import (
     BinaryDefault,
     CatDefault,
@@ -308,6 +308,7 @@ def default_to_pandas(self, pandas_op, *args, **kwargs) -> Self:
             return [self.__wrap_in_qc(obj) for obj in result]
         return self.__wrap_in_qc(result)
 
+    @disable_logging
     def qc_engine_switch_cost(self, other_qc_type: type) -> int:
         """
         Return the coercion costs of this qc to other_qc type.
diff --git a/modin/core/storage_formats/base/query_compiler_calculator.py b/modin/core/storage_formats/base/query_compiler_calculator.py
new file mode 100644
index 00000000000..13d481ceccf
--- /dev/null
+++ b/modin/core/storage_formats/base/query_compiler_calculator.py
@@ -0,0 +1,135 @@
+# Licensed to Modin Development Team under one or more contributor license agreements.
+# See the NOTICE file distributed with this work for additional information regarding
+# copyright ownership.  The Modin Development Team 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.
+
+"""
+Module contains ``QueryCompilerCostCalculator`` class.
+
+``QueryCompilerCostCalculator`` is used to determine the casting cost
+between a set of different backends. It aggregates the cost across
+all query compilers to determine the best query compiler to use.
+"""
+
+from typing import Union
+
+from modin.core.storage_formats.base.query_compiler import (
+    BaseQueryCompiler,
+    QCCoercionCost,
+)
+
+
+class AggregatedBackendData:
+    """
+    Contains information on Backends considered for computation.
+
+    Parameters
+    ----------
+    backend : str
+        String representing the backend name.
+    query_compiler : QueryCompiler
+    """
+
+    def __init__(self, backend: str, query_compiler: BaseQueryCompiler):
+        self.backend = backend
+        self.qc_cls = type(query_compiler)
+        self.cost = 0
+
+
+class BackendCostCalculator:
+    """
+    Calculate which Backend should be used for an operation.
+
+    Given a set of QueryCompilers containing various data, determine
+    which query compiler's backend would minimize the cost of casting
+    or coercion. Use the aggregate sum of coercion to determine overall
+    cost.
+    """
+
+    def __init__(self):
+        self._backend_data = {}
+        self._qc_list = []
+        self._default_qc = None
+        self._result_backend = None
+
+    def add_query_compiler(self, query_compiler):
+        """
+        Add a query compiler to be considered for casting.
+
+        Parameters
+        ----------
+        query_compiler : QueryCompiler
+        """
+        if isinstance(query_compiler, type):
+            # class, no data
+            qc_type = query_compiler
+        else:
+            # instance
+            qc_type = type(query_compiler)
+            self._qc_list.append(query_compiler)
+            backend = query_compiler.get_backend()
+            backend_data = AggregatedBackendData(backend, query_compiler)
+            self._backend_data[backend] = backend_data
+        # TODO: Handle the case where we have a class method and an instance argument of different
+        # backends.
+        self._default_qc = qc_type
+
+    def calculate(self) -> Union[str, type[BaseQueryCompiler]]:
+        """
+        Calculate which query compiler we should cast to.
+
+        Returns
+        -------
+        type
+            QueryCompiler class which should be used for the operation.
+        """
+        if self._result_backend is not None:
+            return self._result_backend
+        if self._default_qc is None:
+            raise ValueError("No query compilers registered")
+        if len(self._backend_data) == 0:
+            return self._default_qc
+
+        # instance selection
+        for qc_from in self._qc_list:
+            qc_to_cls_costed = set()
+            for qc_to in self._qc_list:
+                qc_cls_to = type(qc_to)
+                if qc_cls_to not in qc_to_cls_costed:
+                    qc_to_cls_costed.add(qc_cls_to)
+                    backend_to = qc_to.get_backend()
+                    cost = qc_from.qc_engine_switch_cost(qc_cls_to)
+                    if cost is not None:
+                        self._add_cost_data(backend_to, cost)
+
+        min_value = None
+        for k, v in self._backend_data.items():
+            if min_value is None or min_value > v.cost:
+                min_value = v.cost
+                self._result_backend = k
+
+        return self._result_backend
+
+    def _add_cost_data(self, backend, cost):
+        """
+        Add the cost data to the calculator.
+
+        Parameters
+        ----------
+        backend : str
+            String representing the backend for this engine.
+        cost : dict
+            Dictionary of query compiler classes to costs.
+        """
+        # We can assume that if we call this method, backend
+        # exists in the backend_data map
+        QCCoercionCost.validate_coersion_cost(cost)
+        self._backend_data[backend].cost += cost
diff --git a/modin/core/storage_formats/pandas/native_query_compiler.py b/modin/core/storage_formats/pandas/native_query_compiler.py
index 41978df8e2a..450b2682caf 100644
--- a/modin/core/storage_formats/pandas/native_query_compiler.py
+++ b/modin/core/storage_formats/pandas/native_query_compiler.py
@@ -28,9 +28,7 @@
 )
 from modin.core.storage_formats.base.query_compiler import BaseQueryCompiler
 from modin.core.storage_formats.pandas.query_compiler_caster import QueryCompilerCaster
-from modin.utils import (
-    _inherit_docstrings,
-)
+from modin.utils import _inherit_docstrings
 
 _NO_REPARTITION_ON_NATIVE_EXECUTION_EXCEPTION_MESSAGE = (
     "Modin dataframes and series using native execution do not have partitions."
diff --git a/modin/core/storage_formats/pandas/query_compiler_caster.py b/modin/core/storage_formats/pandas/query_compiler_caster.py
index 67d73d9f55e..50242a6d1b2 100644
--- a/modin/core/storage_formats/pandas/query_compiler_caster.py
+++ b/modin/core/storage_formats/pandas/query_compiler_caster.py
@@ -26,116 +26,14 @@
 
 from pandas.core.indexes.frozen import FrozenList
 
-from modin.core.storage_formats.base.query_compiler import (
-    BaseQueryCompiler,
-    QCCoercionCost,
+from modin.core.storage_formats.base.query_compiler import BaseQueryCompiler
+from modin.core.storage_formats.base.query_compiler_calculator import (
+    BackendCostCalculator,
 )
 
 Fn = TypeVar("Fn", bound=Any)
 
 
-class QueryCompilerCasterCalculator:
-    """
-    Calculate which QueryCompiler should be used for an operation.
-
-    Given a set of QueryCompilers containing various data, determine
-    which query compiler's backend would minimize the cost of casting
-    or coercion. Use the aggregate sum of coercion to determine overall
-    cost.
-    """
-
-    def __init__(self):
-        self._compiler_class_to_cost = {}
-        self._compiler_class_to_data_class = {}
-        self._qc_list = []
-        self._qc_cls_set = set()
-        self._result_type = None
-
-    def add_query_compiler(self, query_compiler):
-        """
-        Add a query compiler to be considered for casting.
-
-        Parameters
-        ----------
-        query_compiler : QueryCompiler
-        """
-        if isinstance(query_compiler, type):
-            # class
-            qc_type = query_compiler
-        else:
-            # instance
-            qc_type = type(query_compiler)
-            self._qc_list.append(query_compiler)
-            self._compiler_class_to_data_class[qc_type] = type(
-                query_compiler._modin_frame
-            )
-        self._qc_cls_set.add(qc_type)
-
-    def calculate(self):
-        """
-        Calculate which query compiler we should cast to.
-
-        Returns
-        -------
-        type
-            QueryCompiler class which should be used for the operation.
-        """
-        if self._result_type is not None:
-            return self._result_type
-        if len(self._qc_cls_set) == 1:
-            return list(self._qc_cls_set)[0]
-        if len(self._qc_cls_set) == 0:
-            raise ValueError("No query compilers registered")
-
-        for qc_from in self._qc_list:
-            for qc_cls_to in self._qc_cls_set:
-                cost = qc_from.qc_engine_switch_cost(qc_cls_to)
-                if cost is not None:
-                    self._add_cost_data({qc_cls_to: cost})
-            self._add_cost_data({type(qc_from): QCCoercionCost.COST_ZERO})
-        min_value = min(self._compiler_class_to_cost.values())
-        for key, value in self._compiler_class_to_cost.items():
-            if min_value == value:
-                self._result_type = key
-                break
-        return self._result_type
-
-    def _add_cost_data(self, costs: dict):
-        """
-        Add the cost data to the calculator.
-
-        Parameters
-        ----------
-        costs : dict
-            Dictionary of query compiler classes to costs.
-        """
-        for k, v in costs.items():
-            # filter out any extranious query compilers not in this operation
-            if k in self._qc_cls_set:
-                QCCoercionCost.validate_coersion_cost(v)
-                # Adds the costs associated with all coercions to a type, k
-                self._compiler_class_to_cost[k] = (
-                    v + self._compiler_class_to_cost[k]
-                    if k in self._compiler_class_to_cost
-                    else v
-                )
-
-    def result_data_cls(self):
-        """
-        Return the data frame associated with the calculated query compiler.
-
-        Returns
-        -------
-        DataFrame object
-            DataFrame object associated with the preferred query compiler.
-        """
-        qc_type = self.calculate()
-        if qc_type in self._compiler_class_to_data_class:
-            return self._compiler_class_to_data_class[qc_type]
-        else:
-            return None
-
-
 class QueryCompilerCaster:
     """Cast all query compiler arguments of the member function to current query compiler."""
 
@@ -217,6 +115,7 @@ def apply_argument_cast(obj: Fn) -> Fn:
         for key in current_class_attrs:
             all_attrs[key] = current_class_attrs[key]
         all_attrs.pop("__abstractmethods__")
+        all_attrs.pop("get_backend")
         all_attrs.pop("__init__")
         all_attrs.pop("qc_engine_switch_cost")
         all_attrs.pop("from_pandas")
@@ -250,8 +149,11 @@ def cast_args(*args: Tuple, **kwargs: Dict) -> Any:
         """
         if len(args) == 0 and len(kwargs) == 0:
             return
+
+        # Import moved inside the function to avoid cyclic import
+
         current_qc = args[0]
-        calculator = QueryCompilerCasterCalculator()
+        calculator = BackendCostCalculator()
         calculator.add_query_compiler(current_qc)
 
         def arg_needs_casting(arg):
@@ -272,9 +174,13 @@ def cast_to_qc(arg):
             qc_type = calculator.calculate()
             if qc_type is None or qc_type is type(arg):
                 return arg
-            # TODO: Should use the factory dispatcher here to switch backends
-            return qc_type.from_pandas(
-                arg.to_pandas(), data_cls=calculator.result_data_cls()
+
+            from modin.core.execution.dispatching.factories.dispatcher import (
+                FactoryDispatcher,
+            )
+
+            return FactoryDispatcher.from_pandas(
+                arg.to_pandas(), calculator.calculate()
             )
 
         if isinstance(current_qc, BaseQueryCompiler):
@@ -284,24 +190,43 @@ def cast_to_qc(arg):
             args = visit_nested_args(args, cast_to_qc)
             kwargs = visit_nested_args(kwargs, cast_to_qc)
 
-        result_qc_type = calculator.calculate()
+        result_backend = calculator.calculate()
+        obj_or_cls = args[0]
+        if isinstance(obj_or_cls, type):
+            # Currently we are executing on a class method
+            if isinstance(result_backend, str):
+                raise TypeError(
+                    "Result backend is a string, but we expected a class"
+                )  # pragma: no cover
+            # The preferred class has not changed;
+            if obj_or_cls == result_backend:
+                return obj(*args, **kwargs)
+            # The preferred class is different; get the replacement function
+            # There are no instances where this occurs, but if we add more class
+            # methods on DataFrame we can use the following lines to get the new
+            # function:
+            # > obj_new = getattr(obj_or_cls, obj.__name__)
+            # > return obj_new(*args, **kwargs)
+            raise NotImplementedError("Class methods on objects not supported")
+        else:
+            if isinstance(result_backend, type):
+                raise TypeError(
+                    "Result backend is a class, but we expected a string"
+                )  # pragma: no cover
+
+            current_backend = obj_or_cls.get_backend()
 
-        if result_qc_type is None or result_qc_type is type(current_qc):
-            return obj(*args, **kwargs)
+            if result_backend == current_backend:
+                return obj(*args, **kwargs)
 
-        # we need to cast current_qc to a new query compiler,
-        # then lookup the same function on the new query compiler
-        # we need to also drop the first argument to the original
-        # args since it references self on the original argument
-        # call
-        if result_qc_type != current_qc:
-            # TODO: Should use the factory dispatcher here to switch backends
-            new_qc = result_qc_type.from_pandas(
-                current_qc.to_pandas(), data_cls=calculator.result_data_cls()
+            from modin.core.execution.dispatching.factories.dispatcher import (
+                FactoryDispatcher,
+            )
+
+            new_qc = FactoryDispatcher.from_pandas(
+                current_qc.to_pandas(), result_backend
             )
             obj_new = getattr(new_qc, obj.__name__)
             return obj_new(*args[1:], **kwargs)
 
-        return obj(*args, **kwargs)
-
     return cast_args
diff --git a/modin/tests/pandas/native_df_interoperability/test_compiler_caster.py b/modin/tests/pandas/native_df_interoperability/test_compiler_caster.py
index b7b3bb2860e..7fa4b3ccd3a 100644
--- a/modin/tests/pandas/native_df_interoperability/test_compiler_caster.py
+++ b/modin/tests/pandas/native_df_interoperability/test_compiler_caster.py
@@ -15,16 +15,23 @@
 import pytest
 
 import modin.pandas as pd
+from modin.config.envvars import Backend, Engine, Execution
+from modin.core.execution.dispatching.factories import factories
+from modin.core.execution.dispatching.factories.factories import BaseFactory
+from modin.core.io.io import BaseIO
 from modin.core.storage_formats.base.query_compiler import QCCoercionCost
-from modin.core.storage_formats.pandas.native_query_compiler import NativeQueryCompiler
-from modin.core.storage_formats.pandas.query_compiler_caster import (
-    QueryCompilerCasterCalculator,
+from modin.core.storage_formats.base.query_compiler_calculator import (
+    BackendCostCalculator,
 )
+from modin.core.storage_formats.pandas.native_query_compiler import NativeQueryCompiler
 
 
 class CloudQC(NativeQueryCompiler):
     "Represents a cloud-hosted query compiler"
 
+    def get_backend(self):
+        return "cloud"
+
     def qc_engine_switch_cost(self, other_qc_cls):
         return {
             CloudQC: QCCoercionCost.COST_ZERO,
@@ -38,6 +45,9 @@ def qc_engine_switch_cost(self, other_qc_cls):
 class ClusterQC(NativeQueryCompiler):
     "Represents a local network cluster query compiler"
 
+    def get_backend(self):
+        return "cluster"
+
     def qc_engine_switch_cost(self, other_qc_cls):
         return {
             CloudQC: QCCoercionCost.COST_MEDIUM,
@@ -51,6 +61,9 @@ def qc_engine_switch_cost(self, other_qc_cls):
 class LocalMachineQC(NativeQueryCompiler):
     "Represents a local machine query compiler"
 
+    def get_backend(self):
+        return "local_machine"
+
     def qc_engine_switch_cost(self, other_qc_cls):
         return {
             CloudQC: QCCoercionCost.COST_MEDIUM,
@@ -63,6 +76,9 @@ def qc_engine_switch_cost(self, other_qc_cls):
 class PicoQC(NativeQueryCompiler):
     "Represents a query compiler with very few resources"
 
+    def get_backend(self):
+        return "pico"
+
     def qc_engine_switch_cost(self, other_qc_cls):
         return {
             CloudQC: QCCoercionCost.COST_LOW,
@@ -75,6 +91,9 @@ def qc_engine_switch_cost(self, other_qc_cls):
 class AdversarialQC(NativeQueryCompiler):
     "Represents a query compiler which returns non-sensical costs"
 
+    def get_backend(self):
+        return "adversarial"
+
     def qc_engine_switch_cost(self, other_qc_cls):
         return {
             CloudQC: -1000,
@@ -86,10 +105,41 @@ def qc_engine_switch_cost(self, other_qc_cls):
 class DefaultQC(NativeQueryCompiler):
     "Represents a query compiler with no costing information"
 
+    def get_backend(self):
+        return "test_casting_default"
+
 
 class DefaultQC2(NativeQueryCompiler):
     "Represents a query compiler with no costing information, but different."
 
+    def get_backend(self):
+        return "test_casting_default_2"
+
+
+def register_backend(name, qc):
+    class TestCasterIO(BaseIO):
+        _should_warn_on_default_to_pandas: bool = False
+        query_compiler_cls = qc
+
+    class TestCasterFactory(BaseFactory):
+        @classmethod
+        def prepare(cls):
+            cls.io_cls = TestCasterIO
+
+    factory_name = f"{name}OnNativeFactory"
+    setattr(factories, factory_name, TestCasterFactory)
+    Engine.add_option(name)
+    Backend.register_backend(name, Execution(name, "Native"))
+
+
+register_backend("pico", PicoQC)
+register_backend("cluster", ClusterQC)
+register_backend("cloud", CloudQC)
+register_backend("local_machine", LocalMachineQC)
+register_backend("adversarial", AdversarialQC)
+register_backend("test_casting_default", DefaultQC)
+register_backend("test_casting_default_2", DefaultQC2)
+
 
 @pytest.fixture()
 def cloud_df():
@@ -229,15 +279,14 @@ def test_default_to_caller(default_df, default2_df):
 
 
 def test_no_qc_data_to_calculate():
-    calculator = QueryCompilerCasterCalculator()
+    calculator = BackendCostCalculator()
     calculator.add_query_compiler(ClusterQC)
     result = calculator.calculate()
     assert result is ClusterQC
-    assert calculator.result_data_cls() is None
 
 
 def test_no_qc_to_calculate():
-    calculator = QueryCompilerCasterCalculator()
+    calculator = BackendCostCalculator()
     with pytest.raises(ValueError):
         calculator.calculate()
 


Code review comments:
diff_hunk: @@ -28,9 +28,7 @@
 )
 from modin.core.storage_formats.base.query_compiler import BaseQueryCompiler
 from modin.core.storage_formats.pandas.query_compiler_caster import QueryCompilerCaster
-from modin.utils import (
-    _inherit_docstrings,
-)
+from modin.utils import _inherit_docstrings
path: modin/core/storage_formats/pandas/native_query_compiler.py
text: @user1:
```suggestion
from modin.utils import (
    _inherit_docstrings,
)
```

diff_hunk: @@ -0,0 +1,135 @@
+# Licensed to Modin Development Team under one or more contributor license agreements.
+# See the NOTICE file distributed with this work for additional information regarding
+# copyright ownership.  The Modin Development Team 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.
+
+"""
+Module contains ``QueryCompilerCostCalculator`` class.
+
+``QueryCompilerCostCalculator`` is used to determine the casting cost
path: modin/core/storage_formats/base/query_compiler_calculator.py
text: @user1:
```suggestion
Module contains ``BackendCostCalculator`` class.

``BackendCostCalculator`` is used to determine the casting cost
```