Repository: modin-project/modin (python)

Issue:
Guard in QueryCompilerCaster to allow a QC to override casting behavior
Sometimes we may want to control how the QueryCompilerCaster behaves on a QC by QC basis. The QueryCompilerCaster is only used for the NativePandas QC; but Snowflake would like to add it to the Snowflake QC as well.

We also want to make sure that users do not accidentally download their entire warehouse, so I would like to propose the following approach:

1. Implement a function, BaseQueryCompiler::allow_coercion_to(OtherQueryCompiler) which will return true by default and false if coercion should not occur.
2. In the QueryCompilerCaster we gate the coercion based on the _argument's_ QC
3. If the argument cannot be coerced ( or casted ) then we emit an error
4. If the argument can be coerced we perform the cast as normal

This *should* allow for the same behavior for existing OSS engines, but allow for some more discretion for engines which want prevent downloading an entire warehouse.

Thoughts @anmyachev @devin-petersohn @mvashishtha 

I can start work on a PR for this if this seems pretty reasonable.

Pull request description:
FEAT-#7448: Allow QueryCompilerCaster to apply cost-optimization on automatic casting
<!--
Thank you for your contribution!
Please review the contributing docs: https://modin.readthedocs.io/en/latest/development/contributing.html
if you have questions about contributing.
-->

## What do these changes do?
Currently a QueryCompiler derived from QueryCompilerCaster will cast all parameters, regardless of type to the query compiler we are calling an operation on. This CL introduces the ability to symmetrically cast to different query compilers based on a cost estimation.

Each query compiler implements a new function which can return a cost estimate of data migration from that specific compiler's perspective. On each operation involving mixed types  we calculate the aggregate cost of migration across all query compilers and pick the query compiler class with the least cost.

The implementation of the cost-estimate, from each query compiler, can be anything, but it is constrained to a set of values between 0-1000.

### Max Cost Not Implemented
This implementation still can face an issue where a workload is moved to the wrong engine, because there is no threshold on the cost of migration implemented. This can be implemented the future.

<!-- Please give a short brief about these changes. -->

- [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 #7448? <!-- 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 -->


Patch:
diff --git a/docs/development/architecture.rst b/docs/development/architecture.rst
index 6585b798..38d8df41 100644
--- a/docs/development/architecture.rst
+++ b/docs/development/architecture.rst
@@ -89,6 +89,12 @@ Dataframe.
 In the interest of reducing the pandas API, the Query Compiler layer closely follows the
 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.
+
 Core Modin Dataframe
 """"""""""""""""""""
 
diff --git a/modin/core/storage_formats/base/query_compiler.py b/modin/core/storage_formats/base/query_compiler.py
index 1410b148..a1e4595c 100644
--- a/modin/core/storage_formats/base/query_compiler.py
+++ b/modin/core/storage_formats/base/query_compiler.py
@@ -21,6 +21,7 @@ from __future__ import annotations
 
 import abc
 import warnings
+from enum import IntEnum
 from functools import cached_property
 from typing import TYPE_CHECKING, Hashable, List, Literal, Optional
 
@@ -107,6 +108,47 @@ def _set_axis(axis):
     return axis_setter
 
 
+class QCCoercionCost(IntEnum):  # noqa: PR01
+    """
+    Coercion costs between different Query Compiler backends.
+
+    Coercion costs between query compilers can be expressed
+    as integers in the range 0 to 1000, where 1000 is
+    considered impossible. Since coercion costs can be a
+    function of many variables ( dataset size, partitioning,
+    network throughput, and query time ) we define a set range
+    of cost values to simplify comparisons between two query
+    compilers / engines in a unified way.
+
+    COST_ZERO means there is no cost associated, or that the query compilers
+    are the same.
+
+    COST_IMPOSSIBLE means the coercion is effectively impossible, which can
+    occur if the target system is unable to store the data as a result
+    of the coercion. Currently this does not prevent coercion.
+    """
+
+    COST_ZERO = 0
+    COST_LOW = 250
+    COST_MEDIUM = 500
+    COST_HIGH = 750
+    COST_IMPOSSIBLE = 1000
+
+    @classmethod
+    def validate_coersion_cost(cls, cost: QCCoercionCost):
+        """
+        Validate that the coercion cost is within range.
+
+        Parameters
+        ----------
+        cost : QCCoercionCost
+        """
+        if int(cost) < int(QCCoercionCost.COST_ZERO) or int(cost) > int(
+            QCCoercionCost.COST_IMPOSSIBLE
+        ):
+            raise ValueError("Query compiler coercsion cost out of range")
+
+
 # FIXME: many of the BaseQueryCompiler methods are hiding actual arguments
 # by using *args and **kwargs. They should be spread into actual parameters.
 # Currently actual arguments are placed in the methods docstrings, but since they're
@@ -247,6 +289,28 @@ class BaseQueryCompiler(
             return [self.__wrap_in_qc(obj) for obj in result]
         return self.__wrap_in_qc(result)
 
+    def qc_engine_switch_cost(self, other_qc_type: type) -> int:
+        """
+        Return the coercion costs of this qc to other_qc type.
+
+        Values returned must be within the acceptable range of
+        QCCoercionCost
+
+        Parameters
+        ----------
+        other_qc_type : QueryCompiler Class
+            The query compiler class to which we should return the cost of switching.
+
+        Returns
+        -------
+        int
+            Cost of migrating the data from this qc to the other_qc or
+            None if the cost cannot be determined.
+        """
+        if isinstance(self, other_qc_type):
+            return QCCoercionCost.COST_ZERO
+        return None
+
     # 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/pandas/query_compiler_caster.py b/modin/core/storage_formats/pandas/query_compiler_caster.py
index 211860a8..67d73d9f 100644
--- a/modin/core/storage_formats/pandas/query_compiler_caster.py
+++ b/modin/core/storage_formats/pandas/query_compiler_caster.py
@@ -26,11 +26,116 @@ from typing import Any, Dict, Tuple, TypeVar
 
 from pandas.core.indexes.frozen import FrozenList
 
-from modin.core.storage_formats.base.query_compiler import BaseQueryCompiler
+from modin.core.storage_formats.base.query_compiler import (
+    BaseQueryCompiler,
+    QCCoercionCost,
+)
 
 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."""
 
@@ -55,48 +160,39 @@ class QueryCompilerCaster:
         apply_argument_cast(cls)
 
 
-def cast_nested_args_to_current_qc_type(arguments, current_qc):
+def visit_nested_args(arguments, fn: callable):
     """
-    Cast all arguments in nested fashion to current query compiler.
+    Visit each argument recursively, calling fn on each one.
 
     Parameters
     ----------
     arguments : tuple or dict
-    current_qc : BaseQueryCompiler
+    fn : Callable to apply to matching arguments
 
     Returns
     -------
     tuple or dict
         Returns args and kwargs with all query compilers casted to current_qc.
     """
-
-    def cast_arg_to_current_qc(arg):
-        current_qc_type = type(current_qc)
-        if isinstance(arg, BaseQueryCompiler) and not isinstance(arg, current_qc_type):
-            data_cls = current_qc._modin_frame
-            return current_qc_type.from_pandas(arg.to_pandas(), data_cls)
-        else:
-            return arg
-
     imutable_types = (FrozenList, tuple)
     if isinstance(arguments, imutable_types):
         args_type = type(arguments)
         arguments = list(arguments)
-        arguments = cast_nested_args_to_current_qc_type(arguments, current_qc)
+        arguments = visit_nested_args(arguments, fn)
 
         return args_type(arguments)
     if isinstance(arguments, list):
         for i in range(len(arguments)):
             if isinstance(arguments[i], (list, dict)):
-                cast_nested_args_to_current_qc_type(arguments[i], current_qc)
+                visit_nested_args(arguments[i], fn)
             else:
-                arguments[i] = cast_arg_to_current_qc(arguments[i])
+                arguments[i] = fn(arguments[i])
     elif isinstance(arguments, dict):
         for key in arguments:
             if isinstance(arguments[key], (list, dict)):
-                cast_nested_args_to_current_qc_type(arguments[key], current_qc)
+                visit_nested_args(arguments[key], fn)
             else:
-                arguments[key] = cast_arg_to_current_qc(arguments[key])
+                arguments[key] = fn(arguments[key])
     return arguments
 
 
@@ -115,13 +211,15 @@ def apply_argument_cast(obj: Fn) -> Fn:
     """
     if isinstance(obj, type):
         all_attrs = dict(inspect.getmembers(obj))
-        all_attrs.pop("__abstractmethods__")
 
         # This is required because inspect converts class methods to member functions
         current_class_attrs = vars(obj)
         for key in current_class_attrs:
             all_attrs[key] = current_class_attrs[key]
-
+        all_attrs.pop("__abstractmethods__")
+        all_attrs.pop("__init__")
+        all_attrs.pop("qc_engine_switch_cost")
+        all_attrs.pop("from_pandas")
         for attr_name, attr_value in all_attrs.items():
             if isinstance(
                 attr_value, (FunctionType, MethodType, classmethod, staticmethod)
@@ -150,10 +248,60 @@ def apply_argument_cast(obj: Fn) -> Fn:
         -------
         Any
         """
+        if len(args) == 0 and len(kwargs) == 0:
+            return
         current_qc = args[0]
+        calculator = QueryCompilerCasterCalculator()
+        calculator.add_query_compiler(current_qc)
+
+        def arg_needs_casting(arg):
+            current_qc_type = type(current_qc)
+            return isinstance(arg, BaseQueryCompiler) and not isinstance(
+                arg, current_qc_type
+            )
+
+        def register_query_compilers(arg):
+            if not arg_needs_casting(arg):
+                return arg
+            calculator.add_query_compiler(arg)
+            return arg
+
+        def cast_to_qc(arg):
+            if not arg_needs_casting(arg):
+                return 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()
+            )
+
         if isinstance(current_qc, BaseQueryCompiler):
-            kwargs = cast_nested_args_to_current_qc_type(kwargs, current_qc)
-            args = cast_nested_args_to_current_qc_type(args, current_qc)
+            visit_nested_args(kwargs, register_query_compilers)
+            visit_nested_args(args, register_query_compilers)
+
+            args = visit_nested_args(args, cast_to_qc)
+            kwargs = visit_nested_args(kwargs, cast_to_qc)
+
+        result_qc_type = calculator.calculate()
+
+        if result_qc_type is None or result_qc_type is type(current_qc):
+            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()
+            )
+            obj_new = getattr(new_qc, obj.__name__)
+            return obj_new(*args[1:], **kwargs)
+
         return obj(*args, **kwargs)
 
     return cast_args
diff --git a/modin/pandas/base.py b/modin/pandas/base.py
index 18a64cd7..b0b6cee4 100644
--- a/modin/pandas/base.py
+++ b/modin/pandas/base.py
@@ -4400,8 +4400,31 @@ class BasePandasDataset(ClassLogger):
             FactoryDispatcher,
         )
 
+        progress_split_count = 2
+        progress_iter = iter(range(progress_split_count))
+        if Backend.normalize(backend) != self.get_backend():
+            try:
+                from tqdm.auto import trange
+
+                progress_iter = iter(
+                    trange(
+                        progress_split_count,
+                        desc=f"Transferring data from {self.get_backend()} to {Backend.normalize(backend)} ...",
+                    )
+                )
+            except ImportError:
+                # Iterate over blank range(2) if tqdm is not installed
+                pass
+        # If tqdm is imported and a conversion is necessary, then display a progress bar.
+        next(progress_iter)
         pandas_self = self._query_compiler.to_pandas()
+        next(progress_iter)
         query_compiler = FactoryDispatcher.from_pandas(df=pandas_self, backend=backend)
+        try:
+            next(progress_iter)
+        except StopIteration:
+            # Last call to next informs tqdm that the operation is done
+            pass
         if inplace:
             self._update_inplace(query_compiler)
             return None
diff --git a/modin/pandas/series.py b/modin/pandas/series.py
index b03395e2..ebf190cd 100644
--- a/modin/pandas/series.py
+++ b/modin/pandas/series.py
@@ -754,7 +754,9 @@ class Series(BasePandasDataset):
         """
         Return int position of the largest value in the Series.
         """
-        result = self.idxmax(axis=axis, skipna=skipna, *args, **kwargs)
+        result = self.reset_index(drop=True).idxmax(
+            axis=axis, skipna=skipna, *args, **kwargs
+        )
         if np.isnan(result) or result is pandas.NA:
             result = -1
         return result
@@ -765,7 +767,9 @@ class Series(BasePandasDataset):
         """
         Return int position of the smallest value in the Series.
         """
-        result = self.idxmin(axis=axis, skipna=skipna, *args, **kwargs)
+        result = self.reset_index(drop=True).idxmin(
+            axis=axis, skipna=skipna, *args, **kwargs
+        )
         if np.isnan(result) or result is pandas.NA:
             result = -1
         return result
@@ -1710,6 +1714,29 @@ class Series(BasePandasDataset):
             mapper=mapper, index=index, axis=axis, copy=copy, inplace=inplace
         )
 
+    def _set_axis_name(self, name, axis=0, inplace=False) -> Union[Series, None]:
+        """
+        Alter the name of the axis.
+
+        Parameters
+        ----------
+        name : str
+            Name for the Series.
+        axis : str or int, default: 0
+            The axis to set the label.
+            Only 0 is valid for Series.
+        inplace : bool, default: False
+            Whether to modify `self` directly or return a copy.
+
+        Returns
+        -------
+        Series or None
+        """
+        self._get_axis_number(axis)  # raises ValueError if not 0
+        renamed = self if inplace else self.copy()
+        renamed.index = renamed.index.set_names(name)
+        return None if inplace else renamed
+
     def rename(
         self,
         index=None,


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
new file mode 100644
index 00000000..b7b3bb28
--- /dev/null
+++ b/modin/tests/pandas/native_df_interoperability/test_compiler_caster.py
@@ -0,0 +1,270 @@
+# 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.
+
+import pandas
+import pytest
+
+import modin.pandas as pd
+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,
+)
+
+
+class CloudQC(NativeQueryCompiler):
+    "Represents a cloud-hosted query compiler"
+
+    def qc_engine_switch_cost(self, other_qc_cls):
+        return {
+            CloudQC: QCCoercionCost.COST_ZERO,
+            ClusterQC: QCCoercionCost.COST_MEDIUM,
+            DefaultQC: QCCoercionCost.COST_MEDIUM,
+            LocalMachineQC: QCCoercionCost.COST_HIGH,
+            PicoQC: QCCoercionCost.COST_IMPOSSIBLE,
+        }[other_qc_cls]
+
+
+class ClusterQC(NativeQueryCompiler):
+    "Represents a local network cluster query compiler"
+
+    def qc_engine_switch_cost(self, other_qc_cls):
+        return {
+            CloudQC: QCCoercionCost.COST_MEDIUM,
+            ClusterQC: QCCoercionCost.COST_ZERO,
+            DefaultQC: None,  # cluster qc knows nothing about default qc
+            LocalMachineQC: QCCoercionCost.COST_MEDIUM,
+            PicoQC: QCCoercionCost.COST_HIGH,
+        }[other_qc_cls]
+
+
+class LocalMachineQC(NativeQueryCompiler):
+    "Represents a local machine query compiler"
+
+    def qc_engine_switch_cost(self, other_qc_cls):
+        return {
+            CloudQC: QCCoercionCost.COST_MEDIUM,
+            ClusterQC: QCCoercionCost.COST_LOW,
+            LocalMachineQC: QCCoercionCost.COST_ZERO,
+            PicoQC: QCCoercionCost.COST_MEDIUM,
+        }[other_qc_cls]
+
+
+class PicoQC(NativeQueryCompiler):
+    "Represents a query compiler with very few resources"
+
+    def qc_engine_switch_cost(self, other_qc_cls):
+        return {
+            CloudQC: QCCoercionCost.COST_LOW,
+            ClusterQC: QCCoercionCost.COST_LOW,
+            LocalMachineQC: QCCoercionCost.COST_LOW,
+            PicoQC: QCCoercionCost.COST_ZERO,
+        }[other_qc_cls]
+
+
+class AdversarialQC(NativeQueryCompiler):
+    "Represents a query compiler which returns non-sensical costs"
+
+    def qc_engine_switch_cost(self, other_qc_cls):
+        return {
+            CloudQC: -1000,
+            ClusterQC: 10000,
+            AdversarialQC: QCCoercionCost.COST_ZERO,
+        }[other_qc_cls]
+
+
+class DefaultQC(NativeQueryCompiler):
+    "Represents a query compiler with no costing information"
+
+
+class DefaultQC2(NativeQueryCompiler):
+    "Represents a query compiler with no costing information, but different."
+
+
+@pytest.fixture()
+def cloud_df():
+    return CloudQC(pandas.DataFrame([0, 1, 2]))
+
+
+@pytest.fixture()
+def cluster_df():
+    return ClusterQC(pandas.DataFrame([0, 1, 2]))
+
+
+@pytest.fixture()
+def local_df():
+    return LocalMachineQC(pandas.DataFrame([0, 1, 2]))
+
+
+@pytest.fixture()
+def pico_df():
+    return PicoQC(pandas.DataFrame([0, 1, 2]))
+
+
+@pytest.fixture()
+def adversarial_df():
+    return AdversarialQC(pandas.DataFrame([0, 1, 2]))
+
+
+@pytest.fixture()
+def default_df():
+    return DefaultQC(pandas.DataFrame([0, 1, 2]))
+
+
+@pytest.fixture()
+def default2_df():
+    return DefaultQC2(pandas.DataFrame([0, 1, 2]))
+
+
+def test_two_same_qc_types_noop(pico_df):
+    df3 = pico_df.concat(axis=1, other=pico_df)
+    assert type(df3) is type(pico_df)
+
+
+def test_two_qc_types_rhs(pico_df, cluster_df):
+    df3 = pico_df.concat(axis=1, other=cluster_df)
+    assert type(df3) is type(cluster_df)  # should move to cluster
+
+
+def test_two_qc_types_lhs(pico_df, cluster_df):
+    df3 = cluster_df.concat(axis=1, other=pico_df)
+    assert type(df3) is type(cluster_df)  # should move to cluster
+
+
+@pytest.mark.parametrize(
+    "df1, df2, df3, df4, result_type",
+    [
+        # no-op
+        ("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),
+        ("cloud_df", "cloud_df", "local_df", "pico_df", CloudQC),
+        # Still move everything to cloud
+        ("pico_df", "pico_df", "pico_df", "cloud_df", CloudQC),
+    ],
+)
+def test_mixed_dfs(df1, df2, df3, df4, result_type, request):
+    df1 = request.getfixturevalue(df1)
+    df2 = request.getfixturevalue(df2)
+    df3 = request.getfixturevalue(df3)
+    df4 = request.getfixturevalue(df4)
+    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)
+
+    df1 = pd.concat([pico_df1, cloud_df1])
+    assert type(df1._query_compiler) is CloudQC
+
+
+def test_adversarial_high(adversarial_df, cluster_df):
+    with pytest.raises(ValueError):
+        adversarial_df.concat(axis=1, other=cluster_df)
+
+
+def test_adversarial_low(adversarial_df, cloud_df):
+    with pytest.raises(ValueError):
+        adversarial_df.concat(axis=1, other=cloud_df)
+
+
+def test_two_two_qc_types_default_rhs(default_df, cluster_df):
+    # none of the query compilers know about each other here
+    # so we default to the caller
+    df3 = default_df.concat(axis=1, other=cluster_df)
+    assert type(df3) is type(default_df)  # should move to default
+
+
+def test_two_two_qc_types_default_lhs(default_df, cluster_df):
+    # none of the query compilers know about each other here
+    # so we default to the caller
+    df3 = cluster_df.concat(axis=1, other=default_df)
+    assert type(df3) is type(cluster_df)  # should move to cluster
+
+
+def test_two_two_qc_types_default_2_rhs(default_df, cloud_df):
+    # cloud knows a bit about costing; so we prefer moving to there
+    df3 = default_df.concat(axis=1, other=cloud_df)
+    assert type(df3) is type(cloud_df)  # should move to cloud
+
+
+def test_two_two_qc_types_default_2_lhs(default_df, cloud_df):
+    # cloud knows a bit about costing; so we prefer moving to there
+    df3 = cloud_df.concat(axis=1, other=default_df)
+    assert type(df3) is type(cloud_df)  # should move to cloud
+
+
+def test_default_to_caller(default_df, default2_df):
+    # No qc knows anything; default to caller
+    df3 = default_df.concat(axis=1, other=default2_df)
+    assert type(df3) is type(default_df)  # should stay on caller
+    df3 = default2_df.concat(axis=1, other=default_df)
+    assert type(df3) is type(default2_df)  # should stay on caller
+    df3 = default_df.concat(axis=1, other=default_df)
+    assert type(df3) is type(default_df)  # no change
+
+
+def test_no_qc_data_to_calculate():
+    calculator = QueryCompilerCasterCalculator()
+    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()
+    with pytest.raises(ValueError):
+        calculator.calculate()
+
+
+def test_qc_default_self_cost(default_df, default2_df):
+    assert default_df.qc_engine_switch_cost(type(default2_df)) is None
+    assert (
+        default_df.qc_engine_switch_cost(type(default_df)) is QCCoercionCost.COST_ZERO
+    )
+
+
+def test_qc_casting_changed_operation(pico_df, cloud_df):
+    pico_df1 = pd.DataFrame(query_compiler=pico_df)
+    cloud_df1 = pd.DataFrame(query_compiler=cloud_df)
+    native_cdf2 = cloud_df1._to_pandas()
+    native_pdf2 = pico_df1._to_pandas()
+    expected = native_cdf2 + native_pdf2
+    # test both directions
+    df_cast_to_rhs = pico_df1 + cloud_df1
+    df_cast_to_lhs = cloud_df1 + pico_df1
+    assert df_cast_to_rhs._to_pandas().equals(expected)
+    assert df_cast_to_lhs._to_pandas().equals(expected)
+
+
+def test_qc_mixed_loc(pico_df, cloud_df):
+    pico_df1 = pd.DataFrame(query_compiler=pico_df)
+    cloud_df1 = pd.DataFrame(query_compiler=cloud_df)
+    assert pico_df1[pico_df1[0][0]][cloud_df1[0][1]] == 1
+    assert pico_df1[cloud_df1[0][0]][pico_df1[0][1]] == 1
+    assert cloud_df1[pico_df1[0][0]][pico_df1[0][1]] == 1
diff --git a/modin/tests/pandas/test_backend.py b/modin/tests/pandas/test_backend.py
index 37a2978e..e0b3aef1 100644
--- a/modin/tests/pandas/test_backend.py
+++ b/modin/tests/pandas/test_backend.py
@@ -13,8 +13,10 @@
 
 import platform
 import re
+from unittest.mock import patch
 
 import pytest
+import tqdm.auto
 
 import modin.pandas as pd
 from modin.config import Backend
@@ -117,7 +119,10 @@ def test_set_valid_backend(
     data_class,
     expected_result_backend,
 ):
-    with config_context(Backend=starting_backend):
+    progress_iter_count = 2
+    with patch.object(
+        tqdm.auto, "trange", return_value=range(progress_iter_count)
+    ) as mock_trange, config_context(Backend=starting_backend):
         original_df = data_class([1])
         # convert to pandas for comparison while still on the `starting_backend`.
         original_df_as_pandas = original_df.modin.to_pandas()
@@ -135,6 +140,14 @@ def test_set_valid_backend(
         # The global Backend should remain the same even if we change the
         # backend for a single dataframe.
         assert Backend.get() == Backend.normalize(starting_backend)
+        if Backend.normalize(starting_backend) == Backend.normalize(
+            expected_result_backend
+        ):
+            mock_trange.assert_not_called()
+        else:
+            # trange constructor is only called once and the iterator is consumed
+            # progress_iter_count times, but we can't easily assert on the number of iterations
+            mock_trange.assert_called_once()
 
 
 def test_set_nonexistent_backend():
diff --git a/modin/tests/pandas/test_series.py b/modin/tests/pandas/test_series.py
index 706afeaf..828cc158 100644
--- a/modin/tests/pandas/test_series.py
+++ b/modin/tests/pandas/test_series.py
@@ -5188,3 +5188,30 @@ def test_logical_binary_with_list(op):
     result_md = getattr(series_md, op)(rhs)
     result_pd = getattr(series_pd, op)(rhs)
     df_equals(result_md, result_pd)
+
+
+@pytest.mark.parametrize("op", ["argmax", "argmin"])
+def test_argmax_argmin_7413(op):
+    # Ensures that argmin/argmax use positional index, not the actual index value
+    series_md, series_pd = create_test_series([1, 2, 3], index=["b", "a", "c"])
+    result_md = getattr(series_md, op)()
+    result_pd = getattr(series_pd, op)()
+    assert result_md == result_pd
+
+
+def test_rename_axis():
+    series_md, series_pd = create_test_series([0, 1, 2])
+    eval_general(series_md, series_pd, lambda ser: ser.rename_axis("name"))
+    eval_general(
+        series_md,
+        series_pd,
+        lambda ser: ser.rename_axis("new_name", inplace=True),
+        __inplace__=True,
+    )
+    # axis=1 is invalid for series
+    eval_general(
+        series_md,
+        series_pd,
+        lambda ser: ser.rename_axis("newer_name", axis=1),
+        expected_exception=ValueError("No axis named 1 for object type Series"),
+    )