Repository: modin-project/modin

Issue:
Add "max-cost" concept to BackendCostCalculator
We need a way to express that there is a maximum cost to migrating data.


Discussion:

Patch:
diff --git a/docs/development/architecture.rst b/docs/development/architecture.rst
index 38d8df41..4959564f 100644
--- a/docs/development/architecture.rst
+++ b/docs/development/architecture.rst
@@ -92,8 +92,9 @@ pandas API, but cuts out a large majority of the repetition.
 QueryCompilers which are derived from QueryCompilerCaster can participate in automatic casting when
 different query compilers, representing different underlying engines, are used together in a
 function. A relative "cost" of casting is used to determine which query compiler everything should
-be moved to. Each query compiler must implement the function, 'qc_engine_switch_cost' to provide
-information and query costs.
+be moved to. Each query compiler must implement the function, `qc_engine_switch_cost` to provide
+information and query costs, as well as a `qc_engine_switch_max_cost` to provide the maximum "cost"
+allowed by the implementaton.
 
 Core Modin Dataframe
 """"""""""""""""""""
diff --git a/modin/core/storage_formats/base/query_compiler.py b/modin/core/storage_formats/base/query_compiler.py
index 3a1fe6d7..c0ce5580 100644
--- a/modin/core/storage_formats/base/query_compiler.py
+++ b/modin/core/storage_formats/base/query_compiler.py
@@ -331,6 +331,18 @@ class BaseQueryCompiler(
             return QCCoercionCost.COST_ZERO
         return None
 
+    @disable_logging
+    def qc_engine_switch_max_cost(self) -> int:
+        """
+        Return the max coercion cost allowed for switching to this engine.
+
+        Returns
+        -------
+        int
+            Max cost allowed for migrating the data to this qc.
+        """
+        return QCCoercionCost.COST_IMPOSSIBLE
+
     # Abstract Methods and Fields: Must implement in children classes
     # In some cases, there you may be able to use the same implementation for
     # some of these abstract methods, but for the sake of generality they are
diff --git a/modin/core/storage_formats/base/query_compiler_calculator.py b/modin/core/storage_formats/base/query_compiler_calculator.py
index dd039f82..58ff5e1c 100644
--- a/modin/core/storage_formats/base/query_compiler_calculator.py
+++ b/modin/core/storage_formats/base/query_compiler_calculator.py
@@ -42,6 +42,7 @@ class AggregatedBackendData:
         self.backend = backend
         self.qc_cls = type(query_compiler)
         self.cost = 0
+        self.max_cost = query_compiler.qc_engine_switch_max_cost()
 
 
 class BackendCostCalculator:
@@ -113,10 +114,17 @@ class BackendCostCalculator:
 
         min_value = None
         for k, v in self._backend_data.items():
+            if v.cost > v.max_cost:
+                continue
             if min_value is None or min_value > v.cost:
                 min_value = v.cost
                 self._result_backend = k
 
+        if self._result_backend is None:
+            raise ValueError(
+                f"Cannot cast to any of the available backends, as the estimated cost is too high. Tried these backends: [{','.join(self._backend_data.keys())}]"
+            )
+
         return self._result_backend
 
     def _add_cost_data(self, backend, cost):
diff --git a/modin/core/storage_formats/pandas/query_compiler_caster.py b/modin/core/storage_formats/pandas/query_compiler_caster.py
index d7d7352e..2f456bdd 100644
--- a/modin/core/storage_formats/pandas/query_compiler_caster.py
+++ b/modin/core/storage_formats/pandas/query_compiler_caster.py
@@ -118,6 +118,7 @@ def apply_argument_cast(obj: Fn) -> Fn:
         all_attrs.pop("get_backend")
         all_attrs.pop("__init__")
         all_attrs.pop("qc_engine_switch_cost")
+        all_attrs.pop("qc_engine_switch_max_cost")
         all_attrs.pop("from_pandas")
         for attr_name, attr_value in all_attrs.items():
             if isinstance(


Test patch:
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 7fa4b3cc..c2751a96 100644
--- a/modin/tests/pandas/native_df_interoperability/test_compiler_caster.py
+++ b/modin/tests/pandas/native_df_interoperability/test_compiler_caster.py
@@ -32,6 +32,9 @@ class CloudQC(NativeQueryCompiler):
     def get_backend(self):
         return "cloud"
 
+    def qc_engine_switch_max_cost(self):
+        return QCCoercionCost.COST_IMPOSSIBLE
+
     def qc_engine_switch_cost(self, other_qc_cls):
         return {
             CloudQC: QCCoercionCost.COST_ZERO,
@@ -48,6 +51,9 @@ class ClusterQC(NativeQueryCompiler):
     def get_backend(self):
         return "cluster"
 
+    def qc_engine_switch_max_cost(self):
+        return QCCoercionCost.COST_HIGH
+
     def qc_engine_switch_cost(self, other_qc_cls):
         return {
             CloudQC: QCCoercionCost.COST_MEDIUM,
@@ -64,6 +70,9 @@ class LocalMachineQC(NativeQueryCompiler):
     def get_backend(self):
         return "local_machine"
 
+    def qc_engine_switch_max_cost(self):
+        return QCCoercionCost.COST_MEDIUM
+
     def qc_engine_switch_cost(self, other_qc_cls):
         return {
             CloudQC: QCCoercionCost.COST_MEDIUM,
@@ -79,6 +88,9 @@ class PicoQC(NativeQueryCompiler):
     def get_backend(self):
         return "pico"
 
+    def qc_engine_switch_max_cost(self):
+        return QCCoercionCost.COST_LOW
+
     def qc_engine_switch_cost(self, other_qc_cls):
         return {
             CloudQC: QCCoercionCost.COST_LOW,
@@ -191,6 +203,11 @@ def test_two_qc_types_lhs(pico_df, cluster_df):
     assert type(df3) is type(cluster_df)  # should move to cluster
 
 
+def test_no_solution(pico_df, local_df, cluster_df, cloud_df):
+    with pytest.raises(ValueError, match=r"pico,local_machine,cluster,cloud"):
+        pico_df.concat(axis=1, other=[local_df, cluster_df, cloud_df])
+
+
 @pytest.mark.parametrize(
     "df1, df2, df3, df4, result_type",
     [
@@ -198,12 +215,13 @@ def test_two_qc_types_lhs(pico_df, cluster_df):
         ("cloud_df", "cloud_df", "cloud_df", "cloud_df", CloudQC),
         # moving all dfs to cloud is 1250, moving to cluster is 1000
         # regardless of how they are ordered
-        ("pico_df", "local_df", "cluster_df", "cloud_df", ClusterQC),
-        ("cloud_df", "local_df", "cluster_df", "pico_df", ClusterQC),
-        ("cloud_df", "cluster_df", "local_df", "pico_df", ClusterQC),
+        ("pico_df", "local_df", "cluster_df", "cloud_df", None),
+        ("cloud_df", "local_df", "cluster_df", "pico_df", None),
+        ("cloud_df", "cluster_df", "local_df", "pico_df", None),
         ("cloud_df", "cloud_df", "local_df", "pico_df", CloudQC),
         # Still move everything to cloud
         ("pico_df", "pico_df", "pico_df", "cloud_df", CloudQC),
+        ("pico_df", "pico_df", "local_df", "cloud_df", CloudQC),
     ],
 )
 def test_mixed_dfs(df1, df2, df3, df4, result_type, request):
@@ -211,19 +229,14 @@ def test_mixed_dfs(df1, df2, df3, df4, result_type, request):
     df2 = request.getfixturevalue(df2)
     df3 = request.getfixturevalue(df3)
     df4 = request.getfixturevalue(df4)
+    if result_type is None:
+        with pytest.raises(ValueError):
+            df1.concat(axis=1, other=[df2, df3, df4])
+        return
     result = df1.concat(axis=1, other=[df2, df3, df4])
     assert type(result) is result_type
 
 
-# This currently passes because we have no "max cost" associated
-# with a particular QC, so we would move all data to the PicoQC
-# As soon as we can represent "max-cost" the result of this operation
-# should be to move all dfs to the CloudQC
-def test_extreme_pico(pico_df, cloud_df):
-    result = cloud_df.concat(axis=1, other=[pico_df] * 7)
-    assert type(result) is PicoQC
-
-
 def test_call_on_non_qc(pico_df, cloud_df):
     pico_df1 = pd.DataFrame(query_compiler=pico_df)
     cloud_df1 = pd.DataFrame(query_compiler=cloud_df)