REFACTOR-modin-project#2011: move default_to_pandas in groupby to backend (modin-project#2041)

Signed-off-by: ienkovich <ilya.enkovich@intel.com>

Author: ienkovich

modin/data_management/functions/groupby_function.py

@@ -14,6 +14,7 @@
 import pandas
 
 from .mapreducefunction import MapReduceFunction
+from modin.pandas.utils import try_cast_to_pandas
 
 
 class GroupbyReduceFunction(MapReduceFunction):
@@ -29,9 +30,13 @@ def caller(
             numeric_only=True,
             drop=False,
         ):
-            assert isinstance(
-                by, type(query_compiler)
-            ), "Can only use groupby reduce with another Query Compiler"
+            if not isinstance(by, type(query_compiler)):
+                by = try_cast_to_pandas(by)
+                return query_compiler.default_to_pandas(
+                    lambda df: map_func(
+                        df.groupby(by=by, axis=axis, **groupby_args), **map_args
+                    )
+                )
             assert axis == 0, "Can only groupby reduce with axis=0"
 
             if numeric_only:

modin/pandas/groupby.py

@@ -455,17 +455,21 @@ def all(self, **kwargs):
         )
 
     def size(self):
-        if is_list_like(self._by) and any(isinstance(o, Series) for o in self._by):
-            # We don't have good way to handle this right now, fall back to Pandas.
-            return self._default_to_pandas(lambda df: df.size())
         if self._axis == 0:
             # Size always works in as_index=True mode so it is necessary to make a
             #  copy of _kwargs and change as_index in it
             kwargs = self._kwargs.copy()
             kwargs["as_index"] = True
+            # Series objects in 'by' mean we couldn't handle the case and transform
+            # 'by' to a query compiler. In this case we replace column names with
+            # actual columns to be able to apply goupby to a Series.
+            if is_list_like(self._by) and any(isinstance(o, Series) for o in self._by):
+                by = [self._df[o] if isinstance(o, str) else o for o in self._by]
+            else:
+                by = self._by
             work_object = SeriesGroupBy(
                 self._df[self._df.columns[0]],
-                self._by,
+                by,
                 self._axis,
                 drop=False,
                 idx_name=None,
@@ -653,7 +657,7 @@ def _wrap_aggregation(
         DataFrame or Series
             Returns the same type as `self._df`.
         """
-        if not isinstance(self._by, type(self._query_compiler)) or self._axis != 0:
+        if self._axis != 0:
             return self._default_to_pandas(default_func, **kwargs)
         # For aggregations, pandas behavior does this for the result.
         # For other operations it does not, so we wait until there is an aggregation to