Repository: modin-project/modin
Pull request: FEAT-#6014: Stop defaulting to pandas in groupby frontend for fill-like methods
<!--
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?

Taking certain progress done by Ponder guys into upstream.

<!-- 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 #6014, #5428
- [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 -->


**Note**: marked as draft because some tests still fail.

Problem statement:
Stop defaulting to pandas in groupby frontend for fill-like methods
Instead of calling `self._default_to_pandas()` in `modin/pandas/groupby.py`, we should call the query compiler and let it default to pandas if something is not implemented.

This allows to implement such functions in non-pandas backends like HDK and other non-pandas engines.

This issue is covering fill-like methods and a few others, namely:
 - ffill
 - sem
 - nth
 - first
 - bfill
 - last
 - pad
 - get_group
 - ngroup
 - head
 - cumcount
 - tail
 - nlargest
 - nsmallest
 - unique

Hints from issue discussion:

Commit message: Fix lint issues

Patch to review:
diff --git a/modin/core/dataframe/algebra/default2pandas/__init__.py b/modin/core/dataframe/algebra/default2pandas/__init__.py
index 876b1e6a8af..2ad4064ca00 100644
--- a/modin/core/dataframe/algebra/default2pandas/__init__.py
+++ b/modin/core/dataframe/algebra/default2pandas/__init__.py
@@ -22,7 +22,7 @@
 from .rolling import RollingDefault
 from .default import DefaultMethod
 from .cat import CatDefault
-from .groupby import GroupByDefault
+from .groupby import GroupByDefault, SeriesGroupByDefault
 
 __all__ = [
     "DataFrameDefault",
@@ -35,4 +35,5 @@
     "DefaultMethod",
     "CatDefault",
     "GroupByDefault",
+    "SeriesGroupByDefault",
 ]
diff --git a/modin/core/dataframe/algebra/default2pandas/groupby.py b/modin/core/dataframe/algebra/default2pandas/groupby.py
index 82d591af167..d91a668b4b6 100644
--- a/modin/core/dataframe/algebra/default2pandas/groupby.py
+++ b/modin/core/dataframe/algebra/default2pandas/groupby.py
@@ -33,6 +33,11 @@ class GroupBy:
         pandas.core.groupby.DataFrameGroupBy.aggregate,
     ]
 
+    @classmethod
+    def _call_groupby(cls, df, *args, **kwargs):  # noqa: PR01
+        """Call .groupby() on passed `df`."""
+        return df.groupby(*args, **kwargs)
+
     @classmethod
     def validate_by(cls, by):
         """
@@ -158,7 +163,7 @@ def fn(
             """Group DataFrame and apply aggregation function to each group."""
             by = cls.validate_by(by)
 
-            grp = df.groupby(by, axis=axis, **groupby_kwargs)
+            grp = cls._call_groupby(df, by, axis=axis, **groupby_kwargs)
             agg_func = cls.get_func(key, **kwargs)
             result = agg_func(grp, *agg_args, **agg_kwargs)
 
@@ -189,7 +194,7 @@ def fn(
             """Group DataFrame and apply aggregation function to each group."""
             if not isinstance(by, (pandas.Series, pandas.DataFrame)):
                 by = cls.validate_by(by)
-                grp = df.groupby(by, axis=axis, **groupby_kwargs)
+                grp = cls._call_groupby(df, by, axis=axis, **groupby_kwargs)
                 grp_agg_func = cls.get_func(agg_func, **kwargs)
                 return grp_agg_func(
                     grp,
@@ -214,13 +219,13 @@ def fn(
             as_index = groupby_kwargs.pop("as_index", True)
             groupby_kwargs["as_index"] = True
 
-            grp = df.groupby(by, axis=axis, **groupby_kwargs)
+            grp = cls._call_groupby(df, by, axis=axis, **groupby_kwargs)
             func = cls.get_func(agg_func, **kwargs)
             result = func(grp, *agg_args, **agg_kwargs)
             method = kwargs.get("method")
 
             if isinstance(result, pandas.Series):
-                result = result.to_frame(MODIN_UNNAMED_SERIES_LABEL)
+                result = result.to_frame(result.name or MODIN_UNNAMED_SERIES_LABEL)
 
             if not as_index:
                 if isinstance(by, pandas.Series):
@@ -279,8 +284,9 @@ def build_groupby(cls, func):
             return cls.build_aggregate_method(func)
         return cls.build_groupby_reduce_method(func)
 
-    @staticmethod
+    @classmethod
     def handle_as_index_for_dataframe(
+        cls,
         result,
         internal_by_cols,
         by_cols_dtypes=None,
@@ -331,7 +337,7 @@ def handle_as_index_for_dataframe(
         if not inplace:
             result = result.copy()
 
-        reset_index, drop, lvls_to_drop, cols_to_drop = GroupBy.handle_as_index(
+        reset_index, drop, lvls_to_drop, cols_to_drop = cls.handle_as_index(
             result_cols=result.columns,
             result_index_names=result.index.names,
             internal_by_cols=internal_by_cols,
@@ -517,9 +523,29 @@ def handle_as_index(
         return reset_index, drop, lvls_to_drop, cols_to_drop
 
 
+class SeriesGroupBy(GroupBy):
+    """Builder for GroupBy aggregation functions for Series."""
+
+    @classmethod
+    def _call_groupby(cls, df, *args, **kwargs):  # noqa: PR01
+        """Call .groupby() on passed `df` squeezed to Series."""
+        # We can end up here by two means - either by "true" call
+        # like Series().groupby() or by df.groupby()[item].
+
+        if len(df.columns) == 1:
+            # Series().groupby() case
+            return df.squeeze(axis=1).groupby(*args, **kwargs)
+        # In second case surrounding logic will supplement grouping columns,
+        # so we need to drop them after grouping is over; our originally
+        # selected column is always the first, so use it
+        return df.groupby(*args, **kwargs)[df.columns[0]]
+
+
 class GroupByDefault(DefaultMethod):
     """Builder for default-to-pandas GroupBy aggregation functions."""
 
+    _groupby_cls = GroupBy
+
     OBJECT_TYPE = "GroupBy"
 
     @classmethod
@@ -542,7 +568,7 @@ def register(cls, func, **kwargs):
             aggregation.
         """
         return super().register(
-            GroupBy.build_groupby(func), fn_name=func.__name__, **kwargs
+            cls._groupby_cls.build_groupby(func), fn_name=func.__name__, **kwargs
         )
 
     # This specifies a `pandas.DataFrameGroupBy` method to pass the `agg_func` to,
@@ -551,7 +577,7 @@ def register(cls, func, **kwargs):
     #   2. `.apply(func)` applies func to a DataFrames, holding a whole group (group-wise).
     #   3. `.transform(func)` is the same as `.apply()` but also broadcast the `func`
     #      result to the group's original shape.
-    __aggregation_methods_dict = {
+    _aggregation_methods_dict = {
         "axis_wise": pandas.core.groupby.DataFrameGroupBy.aggregate,
         "group_wise": pandas.core.groupby.DataFrameGroupBy.apply,
         "transform": pandas.core.groupby.DataFrameGroupBy.transform,
@@ -575,4 +601,16 @@ def get_aggregation_method(cls, how):
         -----
         Visit ``BaseQueryCompiler.groupby_agg`` doc-string for more information about `how` parameter.
         """
-        return cls.__aggregation_methods_dict[how]
+        return cls._aggregation_methods_dict[how]
+
+
+class SeriesGroupByDefault(GroupByDefault):
+    """Builder for default-to-pandas GroupBy aggregation functions for Series."""
+
+    _groupby_cls = SeriesGroupBy
+
+    _aggregation_methods_dict = {
+        "axis_wise": pandas.core.groupby.SeriesGroupBy.aggregate,
+        "group_wise": pandas.core.groupby.SeriesGroupBy.apply,
+        "transform": pandas.core.groupby.SeriesGroupBy.transform,
+    }
diff --git a/modin/core/storage_formats/base/query_compiler.py b/modin/core/storage_formats/base/query_compiler.py
index 9469cd2e612..43af33f84d8 100644
--- a/modin/core/storage_formats/base/query_compiler.py
+++ b/modin/core/storage_formats/base/query_compiler.py
@@ -29,6 +29,7 @@
     RollingDefault,
     CatDefault,
     GroupByDefault,
+    SeriesGroupByDefault,
 )
 from modin.error_message import ErrorMessage
 from . import doc_utils
@@ -2762,6 +2763,7 @@ def groupby_agg(
         agg_kwargs,
         how="axis_wise",
         drop=False,
+        series_groupby=False,
     ):
         """
         Group QueryCompiler data and apply passed aggregation function.
@@ -2790,6 +2792,8 @@ def groupby_agg(
         drop : bool, default: False
             If `by` is a QueryCompiler indicates whether or not by-data came
             from the `self`.
+        series_groupby : bool, default: False
+            Whether we should treat `self` as Series when performing groupby.
 
         Returns
         -------
@@ -2802,7 +2806,8 @@ def groupby_agg(
         elif drop and isinstance(by, type(self)):
             by = list(by.columns)
 
-        return GroupByDefault.register(GroupByDefault.get_aggregation_method(how))(
+        defaulter = SeriesGroupByDefault if series_groupby else GroupByDefault
+        return defaulter.register(defaulter.get_aggregation_method(how))(
             self,
             by=by,
             agg_func=agg_func,
@@ -2857,6 +2862,30 @@ def groupby_skew(
             drop=drop,
         )
 
+    @doc_utils.doc_groupby_method(
+        action="compute cumulative count",
+        result="count of all the previous values",
+        refer_to="cumcount",
+    )
+    def groupby_cumcount(
+        self,
+        by,
+        axis,
+        groupby_kwargs,
+        agg_args,
+        agg_kwargs,
+        drop=False,
+    ):
+        return self.groupby_agg(
+            by=by,
+            agg_func="cumcount",
+            axis=axis,
+            groupby_kwargs=groupby_kwargs,
+            agg_args=agg_args,
+            agg_kwargs=agg_kwargs,
+            drop=drop,
+        )
+
     @doc_utils.doc_groupby_method(
         action="compute cumulative sum",
         result="sum of all the previous values",
@@ -2975,6 +3004,28 @@ def groupby_std(
             drop=drop,
         )
 
+    @doc_utils.doc_groupby_method(
+        action="compute standard error", result="standard error", refer_to="sem"
+    )
+    def groupby_sem(
+        self,
+        by,
+        axis,
+        groupby_kwargs,
+        agg_args,
+        agg_kwargs,
+        drop=False,
+    ):
+        return self.groupby_agg(
+            by=by,
+            agg_func="sem",
+            axis=axis,
+            groupby_kwargs=groupby_kwargs,
+            agg_args=agg_args,
+            agg_kwargs=agg_kwargs,
+            drop=drop,
+        )
+
     @doc_utils.doc_groupby_method(
         action="compute numerical rank", result="numerical rank", refer_to="rank"
     )
@@ -3135,6 +3186,30 @@ def groupby_dtypes(
             drop=drop,
         )
 
+    @doc_utils.doc_groupby_method(
+        action="construct DataFrame from group with provided name",
+        result="DataFrame for given group",
+        refer_to="get_group",
+    )
+    def groupby_get_group(
+        self,
+        by,
+        axis,
+        groupby_kwargs,
+        agg_args,
+        agg_kwargs,
+        drop=False,
+    ):
+        return self.groupby_agg(
+            by=by,
+            agg_func="get_group",
+            axis=axis,
+            groupby_kwargs=groupby_kwargs,
+            agg_args=agg_args,
+            agg_kwargs=agg_kwargs,
+            drop=drop,
+        )
+
     @doc_utils.doc_groupby_method(
         action="shift data with the specified settings",
         result="shifted value",
@@ -3159,6 +3234,225 @@ def groupby_shift(
             drop=drop,
         )
 
+    @doc_utils.doc_groupby_method(
+        action="get first value in group",
+        result="first value",
+        refer_to="first",
+    )
+    def groupby_first(
+        self,
+        by,
+        axis,
+        groupby_kwargs,
+        agg_args,
+        agg_kwargs,
+        drop=False,
+    ):
+        return self.groupby_agg(
+            by=by,
+            agg_func="first",
+            axis=axis,
+            groupby_kwargs=groupby_kwargs,
+            agg_args=agg_args,
+            agg_kwargs=agg_kwargs,
+            drop=drop,
+        )
+
+    @doc_utils.doc_groupby_method(
+        action="get last value in group",
+        result="last value",
+        refer_to="last",
+    )
+    def groupby_last(
+        self,
+        by,
+        axis,
+        groupby_kwargs,
+        agg_args,
+        agg_kwargs,
+        drop=False,
+    ):
+        return self.groupby_agg(
+            by=by,
+            agg_func="last",
+            axis=axis,
+            groupby_kwargs=groupby_kwargs,
+            agg_args=agg_args,
+            agg_kwargs=agg_kwargs,
+            drop=drop,
+        )
+
+    @doc_utils.doc_groupby_method(
+        action="get first n values of a group",
+        result="first n values of a group",
+        refer_to="head",
+    )
+    def groupby_head(
+        self,
+        by,
+        axis,
+        groupby_kwargs,
+        agg_args,
+        agg_kwargs,
+        drop=False,
+    ):
+        return self.groupby_agg(
+            by=by,
+            agg_func="head",
+            axis=axis,
+            groupby_kwargs=groupby_kwargs,
+            agg_args=agg_args,
+            agg_kwargs=agg_kwargs,
+            drop=drop,
+        )
+
+    @doc_utils.doc_groupby_method(
+        action="get last n values in group",
+        result="last n values",
+        refer_to="tail",
+    )
+    def groupby_tail(
+        self,
+        by,
+        axis,
+        groupby_kwargs,
+        agg_args,
+        agg_kwargs,
+        drop=False,
+    ):
+        return self.groupby_agg(
+            by=by,
+            agg_func="tail",
+            axis=axis,
+            groupby_kwargs=groupby_kwargs,
+            agg_args=agg_args,
+            agg_kwargs=agg_kwargs,
+            drop=drop,
+        )
+
+    @doc_utils.doc_groupby_method(
+        action="get nth value in group",
+        result="nth value",
+        refer_to="nth",
+    )
+    def groupby_nth(
+        self,
+        by,
+        axis,
+        groupby_kwargs,
+        agg_args,
+        agg_kwargs,
+        drop=False,
+    ):
+        return self.groupby_agg(
+            by=by,
+            agg_func="nth",
+            axis=axis,
+            groupby_kwargs=groupby_kwargs,
+            agg_args=agg_args,
+            agg_kwargs=agg_kwargs,
+            drop=drop,
+        )
+
+    @doc_utils.doc_groupby_method(
+        action="get group number of each value",
+        result="group number of each value",
+        refer_to="ngroup",
+    )
+    def groupby_ngroup(
+        self,
+        by,
+        axis,
+        groupby_kwargs,
+        agg_args,
+        agg_kwargs,
+        drop=False,
+    ):
+        return self.groupby_agg(
+            by=by,
+            agg_func="ngroup",
+            axis=axis,
+            groupby_kwargs=groupby_kwargs,
+            agg_args=agg_args,
+            agg_kwargs=agg_kwargs,
+            drop=drop,
+        )
+
+    @doc_utils.doc_groupby_method(
+        action="get n largest values in group",
+        result="n largest values",
+        refer_to="nlargest",
+    )
+    def groupby_nlargest(
+        self,
+        by,
+        axis,
+        groupby_kwargs,
+        agg_args,
+        agg_kwargs,
+        drop=False,
+    ):
+        return self.groupby_agg(
+            by=by,
+            agg_func="nlargest",
+            axis=axis,
+            groupby_kwargs=groupby_kwargs,
+            agg_args=agg_args,
+            agg_kwargs=agg_kwargs,
+            drop=drop,
+            series_groupby=True,
+        )
+
+    @doc_utils.doc_groupby_method(
+        action="get n nsmallest values in group",
+        result="n nsmallest values",
+        refer_to="nsmallest",
+    )
+    def groupby_nsmallest(
+        self,
+        by,
+        axis,
+        groupby_kwargs,
+        agg_args,
+        agg_kwargs,
+        drop=False,
+    ):
+        return self.groupby_agg(
+            by=by,
+            agg_func="nsmallest",
+            axis=axis,
+            groupby_kwargs=groupby_kwargs,
+            agg_args=agg_args,
+            agg_kwargs=agg_kwargs,
+            drop=drop,
+            series_groupby=True,
+        )
+
+    @doc_utils.doc_groupby_method(
+        action="get unique values in group",
+        result="unique values",
+        refer_to="unique",
+    )
+    def groupby_unique(
+        self,
+        by,
+        axis,
+        groupby_kwargs,
+        agg_args,
+        agg_kwargs,
+        drop=False,
+    ):
+        return self.groupby_agg(
+            by=by,
+            agg_func="unique",
+            axis=axis,
+            groupby_kwargs=groupby_kwargs,
+            agg_args=agg_args,
+            agg_kwargs=agg_kwargs,
+            drop=drop,
+            series_groupby=True,
+        )
+
     # END Manual Partitioning methods
 
     @doc_utils.add_refer_to("DataFrame.unstack")
diff --git a/modin/core/storage_formats/pandas/query_compiler.py b/modin/core/storage_formats/pandas/query_compiler.py
index 4eb622d280e..80b4c7b6ea3 100644
--- a/modin/core/storage_formats/pandas/query_compiler.py
+++ b/modin/core/storage_formats/pandas/query_compiler.py
@@ -59,7 +59,11 @@
     Binary,
     GroupByReduce,
 )
-from modin.core.dataframe.algebra.default2pandas.groupby import GroupBy, GroupByDefault
+from modin.core.dataframe.algebra.default2pandas.groupby import (
+    GroupBy,
+    GroupByDefault,
+    SeriesGroupByDefault,
+)
 from .utils import get_group_names, merge_partitioning
 from .groupby import GroupbyReduceImpl
 
@@ -3036,12 +3040,24 @@ def groupby_agg(
         agg_kwargs,
         how="axis_wise",
         drop=False,
+        series_groupby=False,
     ):
-        # Defaulting to pandas in case of an empty frame as we can't process it properly.
+        level = groupby_kwargs.get("level", None)
+        if is_list_like(level) and len(level) == 1:
+            level = level[0]
+        # We default to pandas in complex cases of level groupby, i.e. when level is given and
+        # it is not an integer in a multi-index axis or by is not None.
+        complex_level = level is not None and (
+            by is not None
+            or not isinstance(level, int)
+            or not self.has_multiindex(axis)
+        )
+
+        # Also defaulting to pandas in case of an empty frame as we can't process it properly.
         # Higher API level won't pass empty data here unless the frame has delayed
         # computations. So we apparently lose some laziness here (due to index access)
         # because of the inability to process empty groupby natively.
-        if len(self.columns) == 0 or len(self.index) == 0:
+        if complex_level or len(self.columns) == 0 or len(self.index) == 0:
             return super().groupby_agg(
                 by, agg_func, axis, groupby_kwargs, agg_args, agg_kwargs, how, drop
             )
@@ -3057,7 +3073,10 @@ def groupby_agg(
             ), f"Only 'axis_wise' aggregation is supported with dictionary functions, got: {how}"
         else:
             agg_func = functools.partial(
-                GroupByDefault.get_aggregation_method(how), func=agg_func
+                (
+                    SeriesGroupByDefault if series_groupby else GroupByDefault
+                ).get_aggregation_method(how),
+                func=agg_func,
             )
 
         # since we're going to modify `groupby_kwargs` dict in a `groupby_agg_builder`,
@@ -3066,6 +3085,8 @@ def groupby_agg(
         groupby_kwargs = groupby_kwargs.copy()
 
         as_index = groupby_kwargs.get("as_index", True)
+        # remove 'level' from kwargs as we're going to handle it separately
+        groupby_kwargs.pop("level", None)
         by, internal_by = self._groupby_internal_columns(by, drop)
 
         broadcastable_by = [o._modin_frame for o in by if isinstance(o, type(self))]
@@ -3135,12 +3156,24 @@ def groupby_agg_builder(df, by=None, drop=False, partition_idx=None):
                 by = []
 
             by += not_broadcastable_by
+            if level is not None:
+                assert not by, "When handling simple level groupby, `by` must be empty"
+                by = df.axes[axis].get_level_values(level)
 
             def compute_groupby(df, drop=False, partition_idx=0):
                 """Compute groupby aggregation for a single partition."""
-                grouped_df = df.groupby(by=by, axis=axis, **groupby_kwargs)
+                target_df = df.squeeze(axis=1) if series_groupby else df
+                grouped_df = target_df.groupby(by=by, axis=axis, **groupby_kwargs)
                 try:
                     result = partition_agg_func(grouped_df, *agg_args, **agg_kwargs)
+                except ValueError:
+                    # when we're aggregating a column which isn't present in given
+                    # partition, pandas can throw a ValueError
+                    if len(df.columns.difference(by)):
+                        raise  # this is not the expected case, report back
+                    # this should actually be an empty dataframe with correct
+                    # metadata, so use bogus aggregation instead
+                    result = grouped_df.agg("min", *agg_args, **agg_kwargs)
                 except DataError:
                     # This happens when the partition is filled with non-numeric data and a
                     # numeric operation is done. We need to build the index here to avoid
diff --git a/modin/pandas/groupby.py b/modin/pandas/groupby.py
index 037a1e27e65..41400723bca 100644
--- a/modin/pandas/groupby.py
+++ b/modin/pandas/groupby.py
@@ -20,7 +20,7 @@
 from pandas.core.apply import reconstruct_func
 from pandas.errors import SpecificationError
 import pandas.core.groupby
-from pandas.core.dtypes.common import is_list_like, is_numeric_dtype
+from pandas.core.dtypes.common import is_list_like, is_numeric_dtype, is_integer
 from pandas._libs.lib import no_default
 import pandas.core.common as com
 from types import BuiltinFunctionType
@@ -133,6 +133,19 @@ def __init__(
         self._squeeze = squeeze
         self._kwargs.update(kwargs)
 
+    def __override(self, **kwargs):
+        new_kw = dict(
+            df=self._df,
+            by=self._by,
+            axis=self._axis,
+            squeeze=self._squeeze,
+            idx_name=self._idx_name,
+            drop=self._drop,
+            **self._kwargs,
+        )
+        new_kw.update(kwargs)
+        return type(self)(**new_kw)
+
     def __getattr__(self, key):
         """
         Alter regular attribute access, looks up the name in the columns.
@@ -219,10 +232,14 @@ def skew(self, *args, **kwargs):
         )
 
     def ffill(self, limit=None):
-        return self._default_to_pandas(lambda df: df.ffill(limit=limit))
+        return self.fillna(limit=limit, method="ffill")
 
     def sem(self, ddof=1):
-        return self._default_to_pandas(lambda df: df.sem(ddof=ddof))
+        return self._wrap_aggregation(
+            type(self._query_compiler).groupby_sem,
+            agg_kwargs=dict(ddof=ddof),
+            numeric_only=True,
+        )
 
     def sample(self, n=None, frac=None, replace=False, weights=None, random_state=None):
         return self._default_to_pandas(
@@ -391,7 +408,28 @@ def _shift(data, periods, freq, axis, fill_value, is_set_nan_rows=True):
         return result
 
     def nth(self, n, dropna=None):
-        return self._default_to_pandas(lambda df: df.nth(n, dropna=dropna))
+        # TODO: what we really should do is create a GroupByNthSelector to mimic
+        # pandas behavior and then implement some of these methods there.
+        # Adapted error checking from pandas
+        if dropna:
+            if not is_integer(n):
+                raise ValueError("dropna option only supported for an integer argument")
+
+            if dropna not in ["any", "all"]:
+                # Note: when agg-ing picker doesn't raise this, just returns NaN
+                raise ValueError(
+                    "For a DataFrame or Series groupby.nth, dropna must be "
+                    + "either None, 'any' or 'all', "
+                    + f"(was passed {dropna})."
+                )
+
+        return self._check_index(
+            self._wrap_aggregation(
+                type(self._query_compiler).groupby_nth,
+                numeric_only=False,
+                agg_kwargs=dict(n=n, dropna=dropna),
+            )
+        )
 
     def cumsum(self, axis=0, *args, **kwargs):
         return self._check_index_name(
@@ -461,7 +499,11 @@ def dtypes(self):
         )
 
     def first(self, **kwargs):
-        return self._default_to_pandas(lambda df: df.first(**kwargs))
+        return self._wrap_aggregation(
+            type(self._query_compiler).groupby_first,
+            agg_kwargs=dict(**kwargs),
+            numeric_only=False,
+        )
 
     def backfill(self, limit=None):
         warnings.warn(
@@ -592,7 +634,7 @@ def cummin(self, axis=0, **kwargs):
         )
 
     def bfill(self, limit=None):
-        return self._default_to_pandas(lambda df: df.bfill(limit=limit))
+        return self.fillna(limit=limit, method="bfill")
 
     def idxmin(self):
         return self._default_to_pandas(lambda df: df.idxmin())
@@ -690,10 +732,16 @@ def do_relabel(obj_to_relabel):
                 kwargs = {}
             func = func_dict
         elif is_list_like(func):
-            return self._default_to_pandas(
-                lambda df, *args, **kwargs: df.aggregate(func, *args, **kwargs),
-                *args,
-                **kwargs,
+            # for list-list aggregation pandas always puts
+            # groups as index in the result, ignoring as_index,
+            # so we have to reset it to default value
+            return self.__override(as_index=True)._wrap_aggregation(
+                qc_method=type(self._query_compiler).groupby_agg,
+                numeric_only=False,
+                agg_func=func,
+                agg_args=args,
+                agg_kwargs=kwargs,
+                how="axis_wise",
             )
         elif callable(func):
             return self._check_index(
@@ -726,7 +774,11 @@ def do_relabel(obj_to_relabel):
     agg = aggregate
 
     def last(self, **kwargs):
-        return self._default_to_pandas(lambda df: df.last(**kwargs))
+        return self._wrap_aggregation(
+            type(self._query_compiler).groupby_last,
+            agg_kwargs=dict(**kwargs),
+            numeric_only=False,
+        )
 
     def mad(self, **kwargs):
         warnings.warn(
@@ -754,7 +806,7 @@ def corrwith(self):
         return self._default_to_pandas(lambda df: df.corrwith)
 
     def pad(self, limit=None):
-        return self._default_to_pandas(lambda df: df.pad(limit=limit))
+        return self.fillna(limit=limit, method="pad")
 
     def max(self, numeric_only=False, min_count=-1):
         return self._wrap_aggregation(
@@ -771,7 +823,15 @@ def var(self, ddof=1):
         )
 
     def get_group(self, name, obj=None):
-        return self._default_to_pandas(lambda df: df.get_group(name, obj=obj))
+        work_object = self.__override(df=obj if obj is not None else self._df)
+
+        return work_object._check_index(
+            work_object._wrap_aggregation(
+                qc_method=type(work_object._query_compiler).groupby_get_group,
+                numeric_only=False,
+                agg_kwargs=dict(name=name),
+            )
+        )
 
     def __len__(self):
         return len(self.indices)
@@ -860,7 +920,17 @@ def boxplot(
         )
 
     def ngroup(self, ascending=True):
-        return self._default_to_pandas(lambda df: df.ngroup(ascending))
+        result = self._wrap_aggregation(
+            type(self._query_compiler).groupby_ngroup,
+            numeric_only=False,
+            agg_kwargs=dict(ascending=ascending),
+        )
+        if not isinstance(result, Series):
+            # The result should always be a Series with name None and type int64
+            result = result.squeeze(axis=1)
+        # TODO: this might not hold in the future
+        result.name = None
+        return result
 
     def nunique(self, dropna=True):
         return self._check_index(
@@ -883,7 +953,16 @@ def median(self, numeric_only=None):
         )
 
     def head(self, n=5):
-        return self._default_to_pandas(lambda df: df.head(n))
+        # groupby().head()/.tail() ignore as_index, so override it to True
+        work_object = self.__override(as_index=True)
+
+        return work_object._check_index(
+            work_object._wrap_aggregation(
+                type(work_object._query_compiler).groupby_head,
+                agg_kwargs=dict(n=n),
+                numeric_only=False,
+            )
+        )
 
     def cumprod(self, axis=0, *args, **kwargs):
         return self._check_index_name(
@@ -951,13 +1030,27 @@ def pipe(self, func, *args, **kwargs):
         return com.pipe(self, func, *args, **kwargs)
 
     def cumcount(self, ascending=True):
-        result = self._default_to_pandas(lambda df: df.cumcount(ascending=ascending))
-        # pandas does not name the index on cumcount
-        result._query_compiler.set_index_name(None)
+        result = self._wrap_aggregation(
+            type(self._query_compiler).groupby_cumcount,
+            numeric_only=False,
+            agg_kwargs=dict(ascending=ascending),
+        )
+        if not isinstance(result, Series):
+            # The result should always be a Series with name None and type int64
+            result = result.squeeze(axis=1)
+            result.name = None
         return result
 
     def tail(self, n=5):
-        return self._default_to_pandas(lambda df: df.tail(n))
+        # groupby().head()/.tail() ignore as_index, so override it to True
+        work_object = self.__override(as_index=True)
+        return work_object._check_index(
+            work_object._wrap_aggregation(
+                type(work_object._query_compiler).groupby_tail,
+                agg_kwargs=dict(n=n),
+                numeric_only=False,
+            )
+        )
 
     # expanding and rolling are unique cases and need to likely be handled
     # separately. They do not appear to be commonly used.
@@ -971,6 +1064,8 @@ def hist(self):
         return self._default_to_pandas(lambda df: df.hist())
 
     def quantile(self, q=0.5, interpolation="linear"):
+        # TODO: pandas 1.5 now supports numeric_only as an argument
+        # TODO: handle list-like cases properly
         if is_list_like(q):
             return self._default_to_pandas(
                 lambda df: df.quantile(q=q, interpolation=interpolation)
@@ -1410,19 +1505,64 @@ def is_monotonic_decreasing(self):
     def is_monotonic_increasing(self):
         return self._default_to_pandas(lambda ser: ser.is_monotonic_increasing)
 
-    def nlargest(self, n: int = 5, keep: str = "first") -> Series:
-        return self._default_to_pandas(lambda ser: ser.nlargest(n=n, keep=keep))
-
-    def nsmallest(self, n: int = 5, keep: str = "first") -> Series:
-        return self._default_to_pandas(lambda ser: ser.nsmallest(n=n, keep=keep))
-
-    def unique(self):
-        return self._default_to_pandas(lambda ser: ser.unique())
-
     @property
     def dtype(self):
         return self._default_to_pandas(lambda ser: ser.dtype)
 
+    def unique(self):
+        return self._check_index(
+            self._wrap_aggregation(
+                type(self._query_compiler).groupby_unique,
+                numeric_only=False,
+            )
+        )
+
+    def nlargest(self, n=5, keep="first"):
+        return self._check_index(
+            self._wrap_aggregation(
+                type(self._query_compiler).groupby_nlargest,
+                agg_kwargs=dict(n=n, keep=keep),
+                numeric_only=True,
+            )
+        )
+
+    def nsmallest(self, n=5, keep="first"):
+        return self._check_index(
+            self._wrap_aggregation(
+                type(self._query_compiler).groupby_nsmallest,
+                agg_kwargs=dict(n=n, keep=keep),
+                numeric_only=True,
+            )
+        )
+
+    def aggregate(self, func=None, *args, **kwargs):
+        if isinstance(func, dict):
+            raise SpecificationError("nested renamer is not supported")
+        elif is_list_like(func):
+            from .dataframe import DataFrame
+
+            result = DataFrame(
+                query_compiler=self._query_compiler.groupby_agg(
+                    by=self._by,
+                    agg_func=func,
+                    axis=self._axis,
+                    groupby_kwargs=self._kwargs,
+                    agg_args=args,
+                    agg_kwargs=kwargs,
+                )
+            )
+            # query compiler always gives result a multiindex on the axis with the
+            # function names, but series always gets a regular index on the columns
+            # because there is no need to identify which original column's aggregation
+            # the new column represents. alternatively we could give the query compiler
+            # a hint that it's for a series, not a dataframe.
+            maybe_squeezed = result.squeeze() if self._squeeze else result
+            return maybe_squeezed.set_axis(labels=func, axis=1)
+        else:
+            return super().aggregate(func, *args, **kwargs)
+
+    agg = aggregate
+
 
 if IsExperimental.get():
     from modin.experimental.cloud.meta_magic import make_wrapped_class
diff --git a/modin/pandas/test/test_groupby.py b/modin/pandas/test/test_groupby.py
index c41c0e713ba..29d01480812 100644
--- a/modin/pandas/test/test_groupby.py
+++ b/modin/pandas/test/test_groupby.py
@@ -124,13 +124,28 @@ def test_mixed_dtypes_groupby(as_index):
             pandas_groupby = pandas_df.set_index(by[0]).groupby(
                 by=by[-1], as_index=as_index
             )
+            # difference in behaviour between .groupby().ffill() and
+            # .groupby.fillna(method='ffill') on duplicated indices
+            # caused by https://github.com/pandas-dev/pandas/issues/43412
+            # is hurting the tests, for now sort the frames
+            md_sorted_grpby = (
+                modin_df.set_index(by[0])
+                .sort_index()
+                .groupby(by=by[0], as_index=as_index)
+            )
+            pd_sorted_grpby = (
+                pandas_df.set_index(by[0])
+                .sort_index()
+                .groupby(by=by[0], as_index=as_index)
+            )
         else:
             modin_groupby = modin_df.groupby(by=by[0], as_index=as_index)
             pandas_groupby = pandas_df.groupby(by=by[-1], as_index=as_index)
+            md_sorted_grpby, pd_sorted_grpby = modin_groupby, pandas_groupby
 
         modin_groupby_equals_pandas(modin_groupby, pandas_groupby)
         eval_ngroups(modin_groupby, pandas_groupby)
-        eval_general(modin_groupby, pandas_groupby, lambda df: df.ffill())
+        eval_general(md_sorted_grpby, pd_sorted_grpby, lambda df: df.ffill())
         eval_general(
             modin_groupby,
             pandas_groupby,
@@ -163,9 +178,9 @@ def test_mixed_dtypes_groupby(as_index):
 
         eval_dtypes(modin_groupby, pandas_groupby)
         eval_general(modin_groupby, pandas_groupby, lambda df: df.first())
-        eval_general(modin_groupby, pandas_groupby, lambda df: df.backfill())
+        eval_general(md_sorted_grpby, pd_sorted_grpby, lambda df: df.backfill())
         eval_cummin(modin_groupby, pandas_groupby)
-        eval_general(modin_groupby, pandas_groupby, lambda df: df.bfill())
+        eval_general(md_sorted_grpby, pd_sorted_grpby, lambda df: df.bfill())
         eval_general(modin_groupby, pandas_groupby, lambda df: df.idxmin())
         eval_prod(modin_groupby, pandas_groupby)
         if as_index:


Code review comments:
diff_hunk: @@ -3135,12 +3156,24 @@ def groupby_agg_builder(df, by=None, drop=False, partition_idx=None):
                 by = []
 
             by += not_broadcastable_by
+            if level is not None:
+                assert not by, "When handling simple level groupby, `by` must be empty"
+                by = df.axes[axis].get_level_values(level)
path: modin/core/storage_formats/pandas/query_compiler.py
text: @user1:
we used to rely on that every partition has a complete index and so we could have continued passing an original `level` argument without retrieving actual level values, what has changed?

@author:
I'm not exactly sure how this could have worked, as `pandas.groupby()` now bails if `level` is non-`None` and `by` is non-`None` (even if `by` is an empty list).

Also, while technically we could pass the `level` argument instead of getting the list of values (and handle it like _not_ pass the `by` if `level` is non-`None`), the code downwards relies on `by` having a valid length.

@user1:
I've been able to make this work when I explicitly replaced an empty list with `None` in case of a `level` was passed:
<details><summary>git diff</summary>

```diff
diff --git a/modin/core/storage_formats/pandas/query_compiler.py b/modin/core/storage_formats/pandas/query_compiler.py
index 80b4c7b6..8188ea32 100644
--- a/modin/core/storage_formats/pandas/query_compiler.py
+++ b/modin/core/storage_formats/pandas/query_compiler.py
@@ -3085,8 +3085,6 @@ class PandasQueryCompiler(BaseQueryCompiler):
         groupby_kwargs = groupby_kwargs.copy()

         as_index = groupby_kwargs.get("as_index", True)
-        # remove 'level' from kwargs as we're going to handle it separately
-        groupby_kwargs.pop("level", None)
         by, internal_by = self._groupby_internal_columns(by, drop)

         broadcastable_by = [o._modin_frame for o in by if isinstance(o, type(self))]
@@ -3156,9 +3154,8 @@ class PandasQueryCompiler(BaseQueryCompiler):
                 by = []

             by += not_broadcastable_by
-            if level is not None:
-                assert not by, "When handling simple level groupby, `by` must be empty"
-                by = df.axes[axis].get_level_values(level)
+            if level is not None and not by:
+                by = None

             def compute_groupby(df, drop=False, partition_idx=0):
                 """Compute groupby aggregation for a single partition."""
@@ -3205,7 +3202,7 @@ class PandasQueryCompiler(BaseQueryCompiler):
                         result,
                         internal_by_cols,
                         by_cols_dtypes=df[internal_by_cols].dtypes.values,
-                        by_length=len(by),
+                        by_length=len(by) if by is not None else 0,
                         selection=selection,
                         partition_idx=partition_idx,
                         drop=drop,
```

</details>

My main complaint regarding your approach is that getting an index level may be quite expensive, as MultiIndex levels are usually dictionary encoded, meaning that `.get_level_values` requires materializing an actual index array:
```python
import pandas as pd
import numpy as np
import timeit

N = 10_000_000
df = pd.DataFrame({"a": np.arange(N)}, index=pd.MultiIndex.from_product([["a", "b"], np.arange(N // 2)]))
print(timeit.timeit(lambda: df.index.get_level_values(0), number=10)) # ~1.01s
```

@author:
This was my original intention, though I thought `by` is going to be used in more places 😄 

That said, when I remove defaulting to pandas on non-empty `level` at the top (what you didn't do in your diff BTW), one test fails for me (so I still have to fix that):
`pytest --execution=PandasOnPython "modin/pandas/test/test_groupby.py::test_mixed_dtypes_groupby[False]" --pdb`

This is being caused by the logic not resetting the index in this case, as we need valid (non-zero) `by_length` for which I do not see any valid way of computing the length aside from getting values using `.get_level_values()`...

This is my latest patch:
```diff
diff --git a/modin/core/storage_formats/pandas/query_compiler.py b/modin/core/storage_formats/pandas/query_compiler.py
index 8188ea32..775fa5fe 100644
--- a/modin/core/storage_formats/pandas/query_compiler.py
+++ b/modin/core/storage_formats/pandas/query_compiler.py
@@ -3042,22 +3042,11 @@ class PandasQueryCompiler(BaseQueryCompiler):
         drop=False,
         series_groupby=False,
     ):
-        level = groupby_kwargs.get("level", None)
-        if is_list_like(level) and len(level) == 1:
-            level = level[0]
-        # We default to pandas in complex cases of level groupby, i.e. when level is given and
-        # it is not an integer in a multi-index axis or by is not None.
-        complex_level = level is not None and (
-            by is not None
-            or not isinstance(level, int)
-            or not self.has_multiindex(axis)
-        )
-
-        # Also defaulting to pandas in case of an empty frame as we can't process it properly.
+        # Defaulting to pandas in case of an empty frame as we can't process it properly.
         # Higher API level won't pass empty data here unless the frame has delayed
         # computations. So we apparently lose some laziness here (due to index access)
         # because of the inability to process empty groupby natively.
-        if complex_level or len(self.columns) == 0 or len(self.index) == 0:
+        if len(self.columns) == 0 or len(self.index) == 0:
             return super().groupby_agg(
                 by, agg_func, axis, groupby_kwargs, agg_args, agg_kwargs, how, drop
             )
@@ -3154,7 +3143,7 @@ class PandasQueryCompiler(BaseQueryCompiler):
                 by = []

             by += not_broadcastable_by
-            if level is not None and not by:
+            if groupby_kwargs.get("level", None) is not None and not by:
                 by = None

             def compute_groupby(df, drop=False, partition_idx=0):
@@ -3202,7 +3191,7 @@ class PandasQueryCompiler(BaseQueryCompiler):
                         result,
                         internal_by_cols,
                         by_cols_dtypes=df[internal_by_cols].dtypes.values,
-                        by_length=len(by) if by is not None else 0,
+                        by_length=len(by) if by is not None else None,
                         selection=selection,
                         partition_idx=partition_idx,
                         drop=drop,
```

@user1:
So the way to compute valid `by_length` for using level parameter would be:
```python
level = groupby_kwargs.get("level", None)
if level is not None and not by:
    by = None
    by_length = len(level) if is_list_like(level) else 1
else:
    by_length = len(by)
```
Note that `by_length` has to indicate the number of "columns" you're grouping on, not the length of the grouper itself. 

You would additionally need to add a check for whether the kernel you're running is a transform-like aggregation, as pandas never reset the index of the transform results (I don't know how the tests used to work without this `is_transform` check, probably they were just defaulting to pandas).

So summarizing this, here are the final diffs that include all the mentioned changes. I've run groupby and rolling tests for them and they all have passed.

<details><summary>git diff</summary>

```diff
diff --git a/modin/core/dataframe/algebra/default2pandas/groupby.py b/modin/core/dataframe/algebra/default2pandas/groupby.py
index d91a668b..b5ae3a1e 100644
--- a/modin/core/dataframe/algebra/default2pandas/groupby.py
+++ b/modin/core/dataframe/algebra/default2pandas/groupby.py
@@ -433,7 +433,7 @@ class GroupBy:
         if by_length is None:
             by_length = len(internal_by_cols)

-        reset_index = by_length > 0 or selection is not None
+        reset_index = method != "transform" and (by_length > 0 or selection is not None)

         # If the method is "size" then the result contains only one unique named column
         # and we don't have to worry about any naming conflicts, so inserting all of
diff --git a/modin/core/storage_formats/pandas/query_compiler.py b/modin/core/storage_formats/pandas/query_compiler.py
index 80b4c7b6..839c6210 100644
--- a/modin/core/storage_formats/pandas/query_compiler.py
+++ b/modin/core/storage_formats/pandas/query_compiler.py
@@ -39,6 +39,7 @@ from collections.abc import Iterable
 from typing import List, Hashable
 import warnings
 import hashlib
+from pandas.core.groupby.base import transformation_kernels

 from modin.core.storage_formats.base.query_compiler import BaseQueryCompiler
 from modin.config import Engine
@@ -3042,22 +3043,11 @@ class PandasQueryCompiler(BaseQueryCompiler):
         drop=False,
         series_groupby=False,
     ):
-        level = groupby_kwargs.get("level", None)
-        if is_list_like(level) and len(level) == 1:
-            level = level[0]
-        # We default to pandas in complex cases of level groupby, i.e. when level is given and
-        # it is not an integer in a multi-index axis or by is not None.
-        complex_level = level is not None and (
-            by is not None
-            or not isinstance(level, int)
-            or not self.has_multiindex(axis)
-        )
-
         # Also defaulting to pandas in case of an empty frame as we can't process it properly.
         # Higher API level won't pass empty data here unless the frame has delayed
         # computations. So we apparently lose some laziness here (due to index access)
         # because of the inability to process empty groupby natively.
-        if complex_level or len(self.columns) == 0 or len(self.index) == 0:
+        if len(self.columns) == 0 or len(self.index) == 0:
             return super().groupby_agg(
                 by, agg_func, axis, groupby_kwargs, agg_args, agg_kwargs, how, drop
             )
@@ -3067,6 +3057,8 @@ class PandasQueryCompiler(BaseQueryCompiler):
                 by, agg_func, axis, groupby_kwargs, agg_args, agg_kwargs, drop
             )

+        is_transform_method = how == "transform" or (isinstance(agg_func, str) and agg_func in transformation_kernels)
+
         if isinstance(agg_func, dict):
             assert (
                 how == "axis_wise"
@@ -3085,8 +3077,6 @@ class PandasQueryCompiler(BaseQueryCompiler):
         groupby_kwargs = groupby_kwargs.copy()

         as_index = groupby_kwargs.get("as_index", True)
-        # remove 'level' from kwargs as we're going to handle it separately
-        groupby_kwargs.pop("level", None)
         by, internal_by = self._groupby_internal_columns(by, drop)

         broadcastable_by = [o._modin_frame for o in by if isinstance(o, type(self))]
@@ -3156,9 +3146,12 @@ class PandasQueryCompiler(BaseQueryCompiler):
                 by = []

             by += not_broadcastable_by
-            if level is not None:
-                assert not by, "When handling simple level groupby, `by` must be empty"
-                by = df.axes[axis].get_level_values(level)
+            level = groupby_kwargs.get("level", None)
+            if level is not None and not by:
+                by = None
+                by_length = len(level) if is_list_like(level) else 1
+            else:
+                by_length = len(by)

             def compute_groupby(df, drop=False, partition_idx=0):
                 """Compute groupby aggregation for a single partition."""
@@ -3205,11 +3198,12 @@ class PandasQueryCompiler(BaseQueryCompiler):
                         result,
                         internal_by_cols,
                         by_cols_dtypes=df[internal_by_cols].dtypes.values,
-                        by_length=len(by),
+                        by_length=by_length,
                         selection=selection,
                         partition_idx=partition_idx,
                         drop=drop,
                         inplace=True,
+                        method="transform" if is_transform_method else None
                     )
                 else:
                     new_index_names = tuple(
```

</details>

@author:
Thanks @user1, this helped a lot!

diff_hunk: @@ -391,7 +408,28 @@ def _shift(data, periods, freq, axis, fill_value, is_set_nan_rows=True):
         return result
 
     def nth(self, n, dropna=None):
-        return self._default_to_pandas(lambda df: df.nth(n, dropna=dropna))
+        # TODO: what we really should do is create a GroupByNthSelector to mimic
+        # pandas behavior and then implement some of these methods there.
+        # Adapted error checking from pandas
+        if dropna:
+            if not is_integer(n):
+                raise ValueError("dropna option only supported for an integer argument")
+
+            if dropna not in ["any", "all"]:
path: modin/pandas/groupby.py
text: @user1:
as you taught us "a tuple is slightly more performant than a list" :D
```suggestion
            if dropna not in ("any", "all"):
```

diff_hunk: @@ -3036,12 +3040,24 @@ def groupby_agg(
         agg_kwargs,
         how="axis_wise",
         drop=False,
+        series_groupby=False,
     ):
-        # Defaulting to pandas in case of an empty frame as we can't process it properly.
+        level = groupby_kwargs.get("level", None)
+        if is_list_like(level) and len(level) == 1:
+            level = level[0]
+        # We default to pandas in complex cases of level groupby, i.e. when level is given and
+        # it is not an integer in a multi-index axis or by is not None.
+        complex_level = level is not None and (
+            by is not None
+            or not isinstance(level, int)
+            or not self.has_multiindex(axis)
+        )
path: modin/core/storage_formats/pandas/query_compiler.py
text: @user1:
Why? It used to work fine before. Maybe we can make this condition more specific and only include new cases that are now failing?

@author:
The way it "worked" before, if I understand the codebase correctly, was by defaulting to pandas at frontend and not even reaching this point in code altogether 😄

diff_hunk: @@ -3135,12 +3156,24 @@ def groupby_agg_builder(df, by=None, drop=False, partition_idx=None):
                 by = []
 
             by += not_broadcastable_by
+            if level is not None:
+                assert not by, "When handling simple level groupby, `by` must be empty"
+                by = df.axes[axis].get_level_values(level)
 
             def compute_groupby(df, drop=False, partition_idx=0):
                 """Compute groupby aggregation for a single partition."""
-                grouped_df = df.groupby(by=by, axis=axis, **groupby_kwargs)
+                target_df = df.squeeze(axis=1) if series_groupby else df
+                grouped_df = target_df.groupby(by=by, axis=axis, **groupby_kwargs)
                 try:
                     result = partition_agg_func(grouped_df, *agg_args, **agg_kwargs)
+                except ValueError:
+                    # when we're aggregating a column which isn't present in given
+                    # partition, pandas can throw a ValueError
path: modin/core/storage_formats/pandas/query_compiler.py
text: @user1:
well, we generally shouldn't allow this, we do a dictionary filtering for that purpose:
https://github.com/modin-project/modin/blob/ab90115e5493b36a173539cfa2bbc68086dbfb52/modin/core/storage_formats/pandas/query_compiler.py#L3135-L3138
so when does the described case happen?

@author:
Super-simple example would be something like
```python
import modin.pandas as pd
print(pd.DataFrame({'a':[0,1]}).groupby('a',as_index=False)['a'].agg(['mean']))
```

<details><summary>This errors out like shown below unless I fix it the way you've asked about</summary>

```
Traceback (most recent call last):
  File "c:\Vass\Ponder\modin\modin\core\storage_formats\pandas\query_compiler.py", line 3222, in groupby_agg_builder
    return compute_groupby(df, drop, partition_idx)
  File "c:\Vass\Ponder\modin\modin\core\storage_formats\pandas\query_compiler.py", line 3168, in compute_groupby
    result = partition_agg_func(grouped_df, *agg_args, **agg_kwargs)
  File "C:\Users\Vass\.conda\envs\modin\lib\site-packages\pandas\core\groupby\generic.py", line 895, in aggregate
    result = op.agg()
  File "C:\Users\Vass\.conda\envs\modin\lib\site-packages\pandas\core\apply.py", line 175, in agg
    return self.agg_list_like()
  File "C:\Users\Vass\.conda\envs\modin\lib\site-packages\pandas\core\apply.py", line 438, in agg_list_like
    raise ValueError("no results")
ValueError: no results

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "<string>", line 1, in <module>
  File "c:\Vass\Ponder\modin\modin\logging\logger_decorator.py", line 128, in run_and_log
    return obj(*args, **kwargs)
  File "c:\Vass\Ponder\modin\modin\pandas\groupby.py", line 738, in aggregate
    return self.__override(as_index=True)._wrap_aggregation(
  File "c:\Vass\Ponder\modin\modin\logging\logger_decorator.py", line 128, in run_and_log
    return obj(*args, **kwargs)
  File "c:\Vass\Ponder\modin\modin\pandas\groupby.py", line 1318, in _wrap_aggregation
    query_compiler=qc_method(
  File "c:\Vass\Ponder\modin\modin\logging\logger_decorator.py", line 128, in run_and_log
    return obj(*args, **kwargs)
  File "c:\Vass\Ponder\modin\modin\core\storage_formats\pandas\query_compiler.py", line 3230, in groupby_agg
    new_modin_frame = self._modin_frame.broadcast_apply_full_axis(
  File "c:\Vass\Ponder\modin\modin\logging\logger_decorator.py", line 128, in run_and_log
    return obj(*args, **kwargs)
  File "c:\Vass\Ponder\modin\modin\core\dataframe\pandas\dataframe\utils.py", line 375, in run_f_on_minimally_updated_metadata
    result = f(self, *args, **kwargs)
  File "c:\Vass\Ponder\modin\modin\core\dataframe\pandas\dataframe\dataframe.py", line 2914, in broadcast_apply_full_axis
    new_partitions = self._partition_mgr_cls.broadcast_axis_partitions(
  File "c:\Vass\Ponder\modin\modin\logging\logger_decorator.py", line 128, in run_and_log
    return obj(*args, **kwargs)
  File "c:\Vass\Ponder\modin\modin\core\dataframe\pandas\partitioning\partition_manager.py", line 58, in wait
    result = func(cls, *args, **kwargs)
  File "c:\Vass\Ponder\modin\modin\core\dataframe\pandas\partitioning\partition_manager.py", line 482, in broadcast_axis_partitions
    [
  File "c:\Vass\Ponder\modin\modin\core\dataframe\pandas\partitioning\partition_manager.py", line 483, in <listcomp>
    left_partitions[i].apply(
  File "c:\Vass\Ponder\modin\modin\core\dataframe\pandas\partitioning\axis_partition.py", line 105, in apply
    self.deploy_func_between_two_axis_partitions(
  File "c:\Vass\Ponder\modin\modin\core\dataframe\pandas\partitioning\axis_partition.py", line 260, in deploy_func_between_two_axis_partitions
    result = func(lt_frame, rt_frame, *f_args, **f_kwargs)
  File "c:\Vass\Ponder\modin\modin\core\dataframe\pandas\dataframe\dataframe.py", line 1678, in _tree_reduce_func
    series_result = func(df, *args, **kwargs)
  File "c:\Vass\Ponder\modin\modin\core\storage_formats\pandas\query_compiler.py", line 3232, in <lambda>
    func=lambda df, by=None, partition_idx=None: groupby_agg_builder(
  File "c:\Vass\Ponder\modin\modin\core\storage_formats\pandas\query_compiler.py", line 3226, in groupby_agg_builder
    return compute_groupby(df.copy(), drop, partition_idx)
  File "c:\Vass\Ponder\modin\modin\core\storage_formats\pandas\query_compiler.py", line 3168, in compute_groupby
    result = partition_agg_func(grouped_df, *agg_args, **agg_kwargs)
  File "C:\Users\Vass\.conda\envs\modin\lib\site-packages\pandas\core\groupby\generic.py", line 895, in aggregate
    result = op.agg()
  File "C:\Users\Vass\.conda\envs\modin\lib\site-packages\pandas\core\apply.py", line 175, in agg
    return self.agg_list_like()
  File "C:\Users\Vass\.conda\envs\modin\lib\site-packages\pandas\core\apply.py", line 438, in agg_list_like
    raise ValueError("no results")
ValueError: no results
```
</details>

@user1:
well, it seems that your approach doesn't match pandas as well:

<details><summary>example</summary>

```python
# modin on your branch
>>> print(pd.DataFrame({'a':[0,1]}).groupby('a',as_index=False)['a'].agg(['mean']))
Empty DataFrame
Columns: []
Index: [0, 1]
# pandas 1.5.3
>>> print(pd.DataFrame({'a':[0,1]}).groupby('a',as_index=False)['a'].agg(['mean']))
   mean
a
0   0.0
1   1.0
# pandas 2.0
>>> print(pd.DataFrame({'a':[0,1]}).groupby('a',as_index=False)['a'].agg(['mean']))
   a  mean
0  0   0.0
1  1   1.0
```

</details>

The pandas error you're referencing seems to only happen when one applies a list aggregation when grouped on all of the columns of the frame (so there are no columns to aggregate):
```python
>>> import pandas as pd # the following valid for both pandas 1.5.3 and 2.0
>>> df = pd.DataFrame({"a": [1, 1], "b": [3, 4]})
>>> df[["a"]].groupby("a").agg(["mean"])
ValueError: no results
>>> df.groupby(["a", "b"]).agg(["mean"])
ValueError: no results
```

Given that we have a warning claiming that we're not supporting the intersection of the grouping columns and the ones you can take via `.__getitem__`, and that even your workaround doesn't provide a result matching pandas, maybe it's worth converting this warning into the default-to-pandas call? What do you think?
https://github.com/modin-project/modin/blob/0276f1d0474928f73cb0351eeecf0a1b46cdb131/modin/pandas/groupby.py#L552-L562

@author:
That code was the illustration of what was failing... I was fixing a test, one of examples is `"modin\pandas\test\test_groupby.py::test_multi_column_groupby_different_partitions[False-2-False-renaming_aggs_at_same_partition]` failing.
<details><summary>Traceback</summary>

```
modin\pandas\test\test_groupby.py:1934:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
modin\pandas\test\test_groupby.py:1207: in eval___getitem__
    eval_general(
modin\pandas\test\utils.py:833: in eval_general
    values = execute_callable(
modin\pandas\test\utils.py:812: in execute_callable
    md_result = fn(modin_df, **md_kwargs)
modin\pandas\test\test_groupby.py:1197: in test
    res = grp[item].agg(fns)
modin\logging\logger_decorator.py:128: in run_and_log
    return obj(*args, **kwargs)
modin\pandas\groupby.py:738: in aggregate
    return self.__override(as_index=True)._wrap_aggregation(
modin\logging\logger_decorator.py:128: in run_and_log
    return obj(*args, **kwargs)
modin\pandas\groupby.py:1318: in _wrap_aggregation
    query_compiler=qc_method(
modin\logging\logger_decorator.py:128: in run_and_log
    return obj(*args, **kwargs)
modin\core\storage_formats\pandas\query_compiler.py:3230: in groupby_agg
    new_modin_frame = self._modin_frame.broadcast_apply_full_axis(
modin\logging\logger_decorator.py:128: in run_and_log
    return obj(*args, **kwargs)
modin\core\dataframe\pandas\dataframe\utils.py:375: in run_f_on_minimally_updated_metadata
    result = f(self, *args, **kwargs)
modin\core\dataframe\pandas\dataframe\dataframe.py:2914: in broadcast_apply_full_axis
    new_partitions = self._partition_mgr_cls.broadcast_axis_partitions(
modin\logging\logger_decorator.py:128: in run_and_log
    return obj(*args, **kwargs)
modin\core\dataframe\pandas\partitioning\partition_manager.py:58: in wait
    result = func(cls, *args, **kwargs)
modin\core\dataframe\pandas\partitioning\partition_manager.py:482: in broadcast_axis_partitions
    [
modin\core\dataframe\pandas\partitioning\partition_manager.py:483: in <listcomp>
    left_partitions[i].apply(
modin\core\dataframe\pandas\partitioning\axis_partition.py:105: in apply
    self.deploy_func_between_two_axis_partitions(
modin\core\dataframe\pandas\partitioning\axis_partition.py:260: in deploy_func_between_two_axis_partitions
    result = func(lt_frame, rt_frame, *f_args, **f_kwargs)
modin\core\dataframe\pandas\dataframe\dataframe.py:1678: in _tree_reduce_func
    series_result = func(df, *args, **kwargs)
modin\core\storage_formats\pandas\query_compiler.py:3232: in <lambda>
    func=lambda df, by=None, partition_idx=None: groupby_agg_builder(
modin\core\storage_formats\pandas\query_compiler.py:3226: in groupby_agg_builder
    return compute_groupby(df.copy(), drop, partition_idx)
modin\core\storage_formats\pandas\query_compiler.py:3168: in compute_groupby
    result = partition_agg_func(grouped_df, *agg_args, **agg_kwargs)
C:\Users\Vass\.conda\envs\modin\lib\site-packages\pandas\core\groupby\generic.py:895: in aggregate
    result = op.agg()
C:\Users\Vass\.conda\envs\modin\lib\site-packages\pandas\core\apply.py:175: in agg
    return self.agg_list_like()
```
</details>

@user1:
Okay, so this seems to only happen when the kernel is submitted to a partition consisting of columns to group only. Meaning that we can filter out these columns using the `apply_indices` parameter so we wouldn't submit extra kernels that we don't really need.

I've run groupby test with the following diffs and they all have passed:

```diff
diff --git a/modin/core/storage_formats/pandas/query_compiler.py b/modin/core/storage_formats/pandas/query_compiler.py
index 839c6210..56dfc62d 100644
--- a/modin/core/storage_formats/pandas/query_compiler.py
+++ b/modin/core/storage_formats/pandas/query_compiler.py
@@ -3059,6 +3059,8 @@ class PandasQueryCompiler(BaseQueryCompiler):

         is_transform_method = how == "transform" or (isinstance(agg_func, str) and agg_func in transformation_kernels)

+        original_agg_func = agg_func
+
         if isinstance(agg_func, dict):
             assert (
                 how == "axis_wise"
@@ -3159,14 +3161,6 @@ class PandasQueryCompiler(BaseQueryCompiler):
                 grouped_df = target_df.groupby(by=by, axis=axis, **groupby_kwargs)
                 try:
                     result = partition_agg_func(grouped_df, *agg_args, **agg_kwargs)
-                except ValueError:
-                    # when we're aggregating a column which isn't present in given
-                    # partition, pandas can throw a ValueError
-                    if len(df.columns.difference(by)):
-                        raise  # this is not the expected case, report back
-                    # this should actually be an empty dataframe with correct
-                    # metadata, so use bogus aggregation instead
-                    result = grouped_df.agg("min", *agg_args, **agg_kwargs)
                 except DataError:
                     # This happens when the partition is filled with non-numeric data and a
                     # numeric operation is done. We need to build the index here to avoid
@@ -3227,7 +3221,12 @@ class PandasQueryCompiler(BaseQueryCompiler):
                 # all the time, so this will try to fast-path the code first.
                 return compute_groupby(df.copy(), drop, partition_idx)

-        apply_indices = list(agg_func.keys()) if isinstance(agg_func, dict) else None
+        if isinstance(original_agg_func, dict):
+            apply_indices = list(agg_func.keys())
+        elif isinstance(original_agg_func, list):
+            apply_indices = self.columns.difference(internal_by).tolist()
+        else:
+            apply_indices = None

         new_modin_frame = self._modin_frame.broadcast_apply_full_axis(
             axis=axis,
```

@author:
Thanks @user1, this helped a lot!