Fix issue where certain arguments were not defaulting to pandas (#890)

* Fix issue where certain arguments were not defaulting to pandas

* Resolves #876
* Convert all DataFrame and Series arguments to pandas when doing a
  pandas default implementation
* Add tests

* Fix outdated test

modin/pandas/base.py

@@ -193,8 +193,6 @@ def _validate_other(
     def _binary_op(self, op, other, axis=None, **kwargs):
         axis = self._get_axis_number(axis) if axis is not None else 1
         if kwargs.get("level", None) is not None:
-            if isinstance(other, BasePandasDataset):
-                other = other._to_pandas()
             return self._default_to_pandas(
                 getattr(getattr(pandas, self.__name__), op), other, axis=axis, **kwargs
             )
@@ -214,6 +212,11 @@ def _default_to_pandas(self, op, *args, **kwargs):
                 empty_self_str,
             )
         )
+        args = (a._to_pandas() if hasattr(a, "_to_pandas") else a for a in args)
+        kwargs = {
+            k: v._to_pandas() if hasattr(v, "_to_pandas") else v
+            for k, v in kwargs.items()
+        }
         if callable(op):
             result = op(self._to_pandas(), *args, **kwargs)
         elif isinstance(op, str):
@@ -382,8 +385,6 @@ def align(
         fill_axis=0,
         broadcast_axis=None,
     ):
-        if isinstance(other, BasePandasDataset):
-            other = other._to_pandas()
         return self._default_to_pandas(
             "align",
             other,
@@ -687,15 +688,11 @@ def clip_upper(self, threshold, axis=None, inplace=False):
         return self.clip(upper=threshold, axis=axis, inplace=inplace)
 
     def combine(self, other, func, fill_value=None, **kwargs):
-        if isinstance(other, type(self)):
-            other = other._to_pandas()
         return self._default_to_pandas(
             "combine", other, func, fill_value=fill_value, **kwargs
         )
 
     def combine_first(self, other):
-        if isinstance(other, type(self)):
-            other = other._to_pandas()
         return self._default_to_pandas("combine_first", other=other)
 
     def compound(self, axis=None, skipna=None, level=None):
@@ -899,7 +896,6 @@ def dot(self, other):
             if len(common) > len(self_labels) or len(common) > len(other.index):
                 raise ValueError("matrices are not aligned")
             if isinstance(self, DataFrame) and isinstance(other, DataFrame):
-                other = other._to_pandas()
                 return self._default_to_pandas("dot", other)
         else:
             other = np.asarray(other)
@@ -1193,7 +1189,7 @@ def fillna(
         if isinstance(value, BasePandasDataset):
             new_query_compiler = self._default_to_pandas(
                 "fillna",
-                value=value._to_pandas(),
+                value=value,
                 method=method,
                 axis=axis,
                 inplace=False,
@@ -1558,8 +1554,6 @@ def mask(
         errors="raise",
         try_cast=False,
     ):
-        if isinstance(other, BasePandasDataset):
-            other = other._to_pandas()
         return self._default_to_pandas(
             "mask",
             cond,
@@ -2065,8 +2059,6 @@ def reindex(
         return self._create_or_update_from_compiler(final_query_compiler, not copy)
 
     def reindex_like(self, other, method=None, copy=True, limit=None, tolerance=None):
-        if isinstance(other, BasePandasDataset):
-            other = other._to_pandas()
         return self._default_to_pandas(
             "reindex_like",
             other,
@@ -3236,8 +3228,6 @@ def __eq__(self, other):
         return self.eq(other)
 
     def __finalize__(self, other, method=None, **kwargs):
-        if isinstance(other, BasePandasDataset):
-            other = other._to_pandas()
         return self._default_to_pandas("__finalize__", other, method=method, **kwargs)
 
     def __ge__(self, other):

modin/pandas/test/test_dataframe.py

@@ -1180,8 +1180,12 @@ def test_asof(self):
 
     def test_assign(self):
         data = test_data_values[0]
+        modin_df = pd.DataFrame(data)
+        pandas_df = pandas.DataFrame(data)
         with pytest.warns(UserWarning):
-            pd.DataFrame(data).assign()
+            modin_result = modin_df.assign(new_column=pd.Series(modin_df.iloc[:, 0]))
+        pandas_result = pandas_df.assign(new_column=pd.Series(pandas_df.iloc[:, 0]))
+        df_equals(modin_result, pandas_result)
 
     def test_astype(self):
         td = TestData()

modin/pandas/test/test_series.py

@@ -964,8 +964,13 @@ def test_compound(data):
 
 @pytest.mark.parametrize("data", test_data_values, ids=test_data_keys)
 def test_compress(data):
-    modin_series, _ = create_test_series(data)  # noqa: F841
-    with pytest.raises(KeyError):
+    modin_series, pandas_series = create_test_series(data)  # noqa: F841
+    try:
+        pandas_series.compress(pandas_series > 30)
+    except Exception as e:
+        with pytest.raises(type(e)):
+            modin_series.compress(modin_series > 30)
+    else:
         modin_series.compress(modin_series > 30)