tests for merge in place

Author: sfc-gh-jkew

modin/config/__init__.py

@@ -19,6 +19,7 @@
     AsyncReadMode,
     AutoSwitchBackend,
     Backend,
+    BackendMergeCastInPlace,
     BenchmarkMode,
     CIAWSAccessKeyID,
     CIAWSSecretAccessKey,
@@ -78,6 +79,7 @@
     "GpuCount",
     "Memory",
     "Backend",
+    "BackendMergeCastInPlace",
     "Execution",
     "AutoSwitchBackend",
     "ShowBackendSwitchProgress",

modin/config/envvars.py

@@ -1385,6 +1385,31 @@ def disable(cls) -> None:
         cls.put(False)
 
 
+class BackendMergeCastInPlace(EnvironmentVariable, type=bool):
+    """
+    Whether to cast a DataFrame in-place when performing a merge when using hybrid mode.
+
+    This flag modifies the behavior of a cast performed on operations involving more
+    than one type of query compiler. If enabled the actual cast will be performed in-place
+    and the input DataFrame will have a new backend. If disabled the original DataFrame
+    will remain on the same underlying engine.
+
+    """
+
+    varname = "MODIN_BACKEND_MERGE_CAST_IN_PLACE"
+    default = True
+
+    @classmethod
+    def enable(cls) -> None:
+        """Enable casting in place when performing a merge operation betwen two different compilers."""
+        cls.put(True)
+
+    @classmethod
+    def disable(cls) -> None:
+        """Disable casting in place when performing a merge operation betwen two different compilers."""
+        cls.put(False)
+
+
 class DynamicPartitioning(EnvironmentVariable, type=bool):
     """
     Set to true to use Modin's dynamic-partitioning implementation where possible.

modin/core/storage_formats/pandas/query_compiler_caster.py

@@ -31,7 +31,7 @@
 from pandas.core.indexes.frozen import FrozenList
 from typing_extensions import Self
 
-from modin.config import AutoSwitchBackend, Backend
+from modin.config import AutoSwitchBackend, Backend, BackendMergeCastInPlace
 from modin.config import context as config_context
 from modin.core.storage_formats.base.query_compiler import (
     BaseQueryCompiler,
@@ -1120,12 +1120,18 @@ def cast_to_qc(arg):
                     and arg.get_backend() != result_backend
                 ):
                     return arg
-                arg.set_backend(
-                    result_backend,
-                    switch_operation=f"{_normalize_class_name(class_of_wrapped_fn)}.{name}",
-                    inplace=True,
-                )
-                cast = arg
+                if BackendMergeCastInPlace.get():
+                    arg.set_backend(
+                        result_backend,
+                        switch_operation=f"{_normalize_class_name(class_of_wrapped_fn)}.{name}",
+                        inplace=True,
+                    )
+                    cast = arg
+                else:
+                    cast = arg.set_backend(
+                        result_backend,
+                        switch_operation=f"{_normalize_class_name(class_of_wrapped_fn)}.{name}",
+                    )
                 inplace_update_trackers.append(
                     InplaceUpdateTracker(
                         input_castable=arg,
@@ -1157,7 +1163,9 @@ def cast_to_qc(arg):
             original_qc,
             new_castable,
         ) in inplace_update_trackers:
-            new_castable._copy_into(original_castable)
+            new_qc = new_castable._get_query_compiler()
+            if BackendMergeCastInPlace.get() or original_qc is not new_qc:
+                new_castable._copy_into(original_castable)
 
         return _maybe_switch_backend_post_op(
             result,

modin/tests/pandas/native_df_interoperability/test_compiler_caster.py

@@ -664,9 +664,13 @@ def test_merge_in_place(default_df, lazy_df, cloud_df):
     # Both arguments now have the same qc type
     assert type(lazy_df) is type(default_df)
 
-    df = cloud_df.merge(lazy_df)
-    assert type(df) is type(cloud_df)
-    assert type(lazy_df) is type(cloud_df)
+    with config_context(BackendMergeCastInPlace=False):
+        lazy_df = lazy_df.move_to("Lazy")
+        cloud_df = cloud_df.move_to("Cloud")
+        df = cloud_df.merge(lazy_df)
+        assert type(df) is type(cloud_df)
+        assert lazy_df.get_backend() == "Lazy"
+        assert cloud_df.get_backend() == "Cloud"
 
 
 def test_information_asymmetry(default_df, cloud_df, eager_df, lazy_df):