Repository: modin-project/modin
Pull request: FIX-#3376: fix handling of groupby-dict aggregation of 'by' cols
Signed-off-by: Dmitry Chigarev <dmitry.chigarev@intel.com>

<!--
Thank you for your contribution!
Please review the contributing docs: https://modin.readthedocs.io/en/latest/CONTRIBUTING.html
if you have questions about contributing.
-->

## What do these changes do?

Previously, there was no ability to select  the same columns that the frame was grouped on in a dict aggregation because the following code was dropping all of the intersections of 'by' and the aggregated columns in order to avoid naming conflicts:
https://github.com/modin-project/modin/blob/ddd4afe487d1aee19426cd454e4412a17f1a555c/modin/core/dataframe/algebra/groupby_function.py#L196-L197
This PR changes this logic, by not dropping the columns that were explicitly selected for aggregation and also introducing a common logic of handling the `as_index=False` parameter and all of the derived naming conflicts for any groupby implementation (`GroupBy.handle_as_index` method).

Please visit the original issue (#3376) for a more precise problem description.

- [x] commit message follows format outlined [here](https://modin.readthedocs.io/en/latest/contributing.html)
- [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 #3376 <!-- issue must be created for each patch -->
- [ ] tests added and passing
- [x] module layout described at `docs/developer/architecture.rst` is up-to-date <!-- if you have added, renamed or removed files or directories please update the documentation accordingly -->


Problem statement:
Groupby dictionary aggregation works incorrect in cases when 'by' and columns to aggregate has intersection
### System information
- **OS Platform and Distribution (e.g., Linux Ubuntu 16.04)**:

Ubuntu 20.04.3 LTS

- **Modin version** (`modin.__version__`):

0.10.2+2.g253ecd98.dirty

- **Python version**:

Python 3.8.6

- **Code we can use to reproduce**:

```python
import modin.pandas as pd

df = pd.DataFrame({
    "col0": [ 1,  2,  1,  2],
    "col1": [11, 12, 13, 14],
    "col2": [ 4,  4,  4,  4],
    "col3": [ 1,  1,  1,  1]
})

print(df.shape)
gb = df.groupby("col0")
result = gb.agg({"col0": "count"})
string = "%s" % repr(result)
print(string)
```

<!--
You can obtain the Modin version with

python -c "import modin; print(modin.__version__)"
-->

### Describe the problem
<!-- Describe the problem clearly here. -->

Code above works on Pandas but produces a `ValueError: No objects to concatenate` exception on Modin on Python and Ray engines (those that I tried). I traced the problem down to these lines
```python
        if drop and len(df.columns.intersection(by_part)) > 0:
            df.drop(columns=by_part, errors="ignore", inplace=True)
```
that are located inside of `reduce` function. As a result of these lines `df` that contains only one column `col0` gets empty because this column is dropped. I am not sure whether it is intentional, it doesn't look right for me. @dchigarev please comment.

### Source code / logs
<!-- Include any logs or source code that would be helpful to diagnose the problem. If including tracebacks, please include the full traceback. Large logs and files should be attached. Try to provide a reproducible test case that is the bare minimum necessary to generate the problem. -->

Hints from issue discussion:
@gshimansky at the reduce phase "by" columns are supposed to be presented only as index levels of the `df`, however, there are some corner cases, that produces the `df` frame which has "by" located at the `df` columns and at the index levels at the same time. The following code cleans up the frame from such duplications:
https://github.com/modin-project/modin/blob/e48ac35d4b632d21a5b297b09d9140ed0a5820a1/modin/data_management/functions/groupby_function.py#L194-L196
but as it's revealed from your report there are some cases, where this code doesn't work well. When we explicitly want to use some of the "by" columns in aggregation we shouldn't drop them at this piece of code.

With the following changes, your reproducer as well as all of the existed tests are working:

<details><summary>Changes</summary>

```diff
diff --git a/modin/data_management/functions/groupby_function.py b/modin/data_management/functions/groupby_function.py
index c4046cc0..84811fd6 100644
--- a/modin/data_management/functions/groupby_function.py
+++ b/modin/data_management/functions/groupby_function.py
@@ -120,6 +120,7 @@ class GroupbyReduceFunction(MapReduceFunction):
         # right index and placing columns in the correct order.
         groupby_args["as_index"] = True
         groupby_args["observed"] = True
+        apply_func = cls.try_filter_dict(map_func, df)
         if other is not None:
             # Other is a broadcasted partition that represents 'by' columns
             # Concatenate it with 'df' to group on its columns names
@@ -134,7 +135,6 @@ class GroupbyReduceFunction(MapReduceFunction):
         else:
             by_part = by

-        apply_func = cls.try_filter_dict(map_func, df)
         result = apply_func(
             df.groupby(by=by_part, axis=axis, **groupby_args), **map_args
         )
@@ -192,8 +192,12 @@ class GroupbyReduceFunction(MapReduceFunction):
         # there is a bug in pandas with intersection that forces us to do so:
         # https://github.com/pandas-dev/pandas/issues/39699
         by_part = pandas.Index(df.index.names)
-        if drop and len(df.columns.intersection(by_part)) > 0:
-            df.drop(columns=by_part, errors="ignore", inplace=True)
+        if drop:
+            to_drop = df.columns.intersection(by_part)
+            if isinstance(reduce_func, dict):
+                to_drop = to_drop.difference(reduce_func.keys())
+            if len(to_drop) > 0:
+                df.drop(columns=by_part, errors="ignore", inplace=True)

         groupby_args = groupby_args.copy()
         as_index = groupby_args["as_index"]
```

</details>

This patch should fix all of the inconcordances with pandas of handling intersections of "by" and cols to aggregate, **but for `as_index=True` cases only!**

When `as_index=False` groupby just insert all of the "by" columns to the frame by resetting its index (which contains "by" cols):
https://github.com/modin-project/modin/blob/e48ac35d4b632d21a5b297b09d9140ed0a5820a1/modin/data_management/functions/groupby_function.py#L211-L213
However, if "by" cols weren't dropped because they had been picked for aggregation, then we'll get a naming conflict here and so another exception. To fix this we should introduce some logic of proper handling of the naming conflicts by detecting and dropping duplicated index levels before resetting them. Luckily, this logic is introduced by #3298 in [`handle_as_index`](https://github.com/YarShev/modin/blob/0c2f4a1bb76b48c3092a8372cc877ba4ee9416b9/modin/data_management/functions/groupby_function.py#L538) method, so after #3298 will be merged we could handle `as_index=False` cases as well.
by the way, we already had an issue for this bug (#2542), I'm closing the old one since this issue appears to be much more descriptive
Another reproducer from just closed #2542
<details><summary>Reproducer</summary>

```python
import pandas
import modin.pandas as pd
from modin.pandas.test.utils import test_data, df_equals, create_test_dfs
import numpy as np
import pandas

md_df, pd_df = create_test_dfs(test_data["int_data"])

# 'by' and columns to aggregate has intersection (`md_df.columns[-1]`)
by = [md_df.columns[0], md_df.columns[-1]]
agg_dict = {
    md_df.columns[1]: [("new_sum", "sum"), ("new_prod", "prod")],
    md_df.columns[-1]: np.sum,
}

for as_index in [True, False]:
    md_res = md_df.groupby(by, as_index=as_index).agg(agg_dict)
    pd_res = pd_df.groupby(by, as_index=as_index).agg(agg_dict)
    try:
        df_equals(md_res, pd_res)
    except AssertionError as e:
        print(f"\nas_index-{as_index}:\n{e}\n")

```

Output:
```
as_index-True:
DataFrame.columns are different

DataFrame.columns levels are different
[left]:  1, Index([('col34', 'new_sum'), ('col34', 'new_prod'), 'index'], dtype='object')
[right]: 2, MultiIndex([('col34',  'new_sum'),
            ('col34', 'new_prod'),
            ('index',      'sum')],
           )


as_index-False:
DataFrame are different

DataFrame shape mismatch
[left]:  (253, 5)
[right]: (253, 4)

```


</details>
@dchigarev Thank you for your description of the problem. I think there is a small mistake in your comment, the last `as_index=False` should actually be `as_index=True`.

One more code example with inconsistent behavior between Modin and Pandas:
```python
df = pd.DataFrame({"a": [1, 2, 3], "b": [4, 5, 6], "c": [7, 8, 9], "d": [10, 11, 12]})
result = df.groupby("a")["a", "b", "c"].agg({"b": "count", "c": "sum", "a": "sum"})
print(result)
```
Is this issue unblocked now so it is possible to fix it?
> Is this issue unblocked now so it is possible to fix it?

Yes, I'm working on it
The PR (#3592) linked to this issue will fix processing intersection cases only for non-renaming aggregations. The fix for renaming aggregations is a bit more complicated, I've created a separate issue for this: #3602

Commit message: Clarify in-code comments

Signed-off-by: Dmitry Chigarev <dmitry.chigarev@intel.com>

Patch to review:
diff --git a/modin/core/dataframe/algebra/default2pandas/groupby.py b/modin/core/dataframe/algebra/default2pandas/groupby.py
index 4999283e18f..a4d943794c0 100644
--- a/modin/core/dataframe/algebra/default2pandas/groupby.py
+++ b/modin/core/dataframe/algebra/default2pandas/groupby.py
@@ -16,6 +16,7 @@
 from .default import DefaultMethod
 
 import pandas
+from pandas.core.dtypes.common import is_list_like
 
 
 # FIXME: there is no sence of keeping `GroupBy` and `GroupByDefault` logic in a different
@@ -207,7 +208,8 @@ def fn(
             if numeric_only:
                 df = df.select_dtypes(include="number")
 
-            by = by.squeeze(axis=1)
+            if isinstance(by, pandas.DataFrame):
+                by = by.squeeze(axis=1)
             if (
                 drop
                 and isinstance(by, pandas.Series)
@@ -230,13 +232,34 @@ def fn(
                 result = result.to_frame()
 
             if not as_index:
-                if (
-                    len(result.index.names) == 1 and result.index.names[0] is None
-                ) or all([name in result.columns for name in result.index.names]):
-                    drop = False
-                elif kwargs.get("method") == "size":
-                    drop = True
-                result = result.reset_index(drop=not drop)
+                method = kwargs.get("method")
+                if isinstance(by, pandas.Series):
+                    # 1. If `drop` is True then 'by' Series represents a column from the
+                    #    source frame and so the 'by' is internal.
+                    # 2. If method is 'size' then any 'by' considered to be internal
+                    #    (this is a hacky legacy from the ``groupby_size`` implementation)
+                    internal_by = (by.name,) if drop or method == "size" else tuple()
+                else:
+                    internal_by = by
+
+                drop, lvls_to_drop, cols_to_drop = cls.handle_as_index(
+                    result.columns,
+                    result.index.names,
+                    internal_by,
+                    by_cols_dtypes=(
+                        df.index.dtypes.values
+                        if isinstance(df.index, pandas.MultiIndex)
+                        else (df.index.dtype,)
+                    ),
+                    is_multi_by=len(internal_by) > 1,
+                    drop=drop,
+                    method=method,
+                )
+                if len(lvls_to_drop) > 0:
+                    result.index = result.index.droplevel(lvls_to_drop)
+                if len(cols_to_drop) > 0:
+                    result.drop(columns=cols_to_drop, inplace=True)
+                result = result.reset_index(drop=drop)
 
             if result.index.name == "__reduced__":
                 result.index.name = None
@@ -270,6 +293,172 @@ def build_groupby(cls, func):
             return cls.build_aggregate_method(func)
         return cls.build_groupby_reduce_method(func)
 
+    @staticmethod
+    def handle_as_index(
+        result_cols,
+        result_index_names,
+        internal_by_cols,
+        by_cols_dtypes=None,
+        is_multi_by=None,
+        selection=None,
+        partition_idx=0,
+        drop=True,
+        method=None,
+    ):
+        """
+        Help to handle ``as_index=False`` parameter for the GroupBy result.
+
+        This function resolve naming conflicts of the index levels to insert and the column labels
+        for the GroupBy result. The logic of this function assumes that the initial GroupBy result
+        was computed as ``as_index=True``.
+
+        Parameters
+        ----------
+        result_cols : pandas.Index
+            Columns of the GroupBy result.
+        result_index_names : list-like
+            Index names of the GroupBy result.
+        internal_by_cols : list-like
+            Internal 'by' columns.
+        by_cols_dtypes : list-like, optional
+            Data types of the internal 'by' columns. Required to do special casing
+            in case of categorical 'by'. If not specified, assume that there is no
+            categorical data in 'by'.
+        is_multi_by : bool, optional
+            Whether grouping on multiple items. If not specified, assume ``True`` if
+            the amount of `internal_by_cols` is greater than 1, ``False`` otherwise.
+        selection : label or list of labels, optional
+            Set of columns to which aggregation was applied. If not specified
+            assuming that aggregation was applied to all of the available columns.
+        partition_idx : int, default: 0
+            Positional index of the current partition.
+        drop : bool, default: True
+            Indicates whether or not any of the `by` data came from the same frame.
+        method : str, optional
+            Name of the groupby function. This is a hint to be able to do special casing.
+            Note: this parameter is a legacy from the ``groupby_size`` implementation,
+            it's a hacky one and probably will be removed in the future.
+
+        Returns
+        -------
+        drop_index : bool
+            Indicates whether to drop all index levels (True) or insert them into the
+            resulting columns (False).
+        lvls_to_drop : list of ints
+            Contains numeric indices of the levels of the result index to drop as intersected.
+        cols_to_drop : list of labels
+            Contains labels of the columns to drop from the result as intersected.
+
+        Examples
+        --------
+        >>> groupby_result = compute_groupby_without_processing_as_index_parameter()
+        >>> if not as_index:
+        >>>     drop, lvls_to_drop, cols_to_drop = handle_as_index(**extract_required_params(groupby_result))
+        >>>     if len(lvls_to_drop) > 0:
+        >>>         groupby_result.index = groupby_result.index.droplevel(lvls_to_drop)
+        >>>     if len(cols_to_drop) > 0:
+        >>>         groupby_result = groupby_result.drop(columns=cols_to_drop)
+        >>>     groupby_result_with_processed_as_index_parameter = groupby_result.reset_index(drop=drop)
+        >>> else:
+        >>>     groupby_result_with_processed_as_index_parameter = groupby_result
+        """
+        # 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
+        # the "by" into the result (just a fast-path)
+        if method == "size":
+            return False, [], []
+
+        if is_multi_by is None:
+            is_multi_by = len(internal_by_cols) > 1
+
+        # Pandas logic of resolving naming conflicts is the following:
+        #   1. If any categorical is in 'by' and 'by' is multi-column, then the categorical
+        #      index is prioritized: drop intersected columns and insert all of the 'by' index
+        #      levels to the frame as columns.
+        #   2. Otherwise, aggregation result is prioritized: drop intersected index levels and
+        #      insert the filtered ones to the frame as columns.
+        if by_cols_dtypes is not None:
+            keep_index_levels = (
+                is_multi_by
+                and selection is None
+                and any(isinstance(x, pandas.CategoricalDtype) for x in by_cols_dtypes)
+            )
+        else:
+            keep_index_levels = False
+
+        # 1. We insert 'by'-columns to the result at the beginning of the frame and so only to the
+        #    first partition, if partition_idx != 0 we just drop the index. If there are no columns
+        #    that are required to drop (keep_index_levels is True) then we can exit here.
+        # 2. We don't insert 'by'-columns to the result if 'by'-data came from a different
+        #    frame (drop is False), there's only one exception for this rule: if the `method` is "size",
+        #    so if (drop is False) and method is not "size" we just drop the index and so can exit here.
+        if (not keep_index_levels and partition_idx != 0) or (
+            not drop and method != "size"
+        ):
+            return True, [], []
+
+        if not isinstance(internal_by_cols, pandas.Index):
+            if not is_list_like(internal_by_cols):
+                internal_by_cols = [internal_by_cols]
+            internal_by_cols = pandas.Index(internal_by_cols)
+
+        internal_by_cols = (
+            internal_by_cols[~internal_by_cols.str.match(r"__reduced__.*", na=False)]
+            if hasattr(internal_by_cols, "str")
+            else internal_by_cols
+        )
+
+        if selection is not None:
+            selection = (
+                selection
+                if isinstance(selection, pandas.Index)
+                else pandas.Index(selection)
+            )
+
+        lvls_to_drop = []
+        cols_to_drop = []
+
+        if not keep_index_levels:
+            # We want to insert only these internal-by-cols that are not presented
+            # in the result in order to not create naming conflicts
+            cols_to_insert = (
+                internal_by_cols.copy()
+                if selection is None and internal_by_cols.nlevels != result_cols.nlevels
+                else frozenset(
+                    (
+                        col
+                        for col in internal_by_cols
+                        if col not in (result_cols if selection is None else selection)
+                    )
+                )
+            )
+        else:
+            cols_to_insert = internal_by_cols
+            # We want to drop such internal-by-cols that are presented
+            # in the result in order to not create naming conflicts
+            cols_to_drop = (
+                pandas.Index([])
+                if internal_by_cols.nlevels != result_cols.nlevels
+                else frozenset((col for col in cols_to_insert if col in result_cols))
+            )
+
+        lvls_to_drop = (
+            result_index_names
+            if partition_idx != 0
+            else [
+                i
+                for i, name in enumerate(result_index_names)
+                if name not in cols_to_insert
+            ]
+        )
+
+        drop = False
+        if len(lvls_to_drop) == len(result_index_names):
+            drop = True
+            lvls_to_drop = []
+
+        return drop, lvls_to_drop, cols_to_drop
+
 
 class GroupByDefault(DefaultMethod):
     """Builder for default-to-pandas GroupBy aggregation functions."""
diff --git a/modin/core/dataframe/algebra/groupby.py b/modin/core/dataframe/algebra/groupby.py
index b2694981223..515410d6317 100644
--- a/modin/core/dataframe/algebra/groupby.py
+++ b/modin/core/dataframe/algebra/groupby.py
@@ -121,6 +121,7 @@ def map(
         # right index and placing columns in the correct order.
         groupby_args["as_index"] = True
         groupby_args["observed"] = True
+        apply_func = cls.try_filter_dict(map_func, df)
         if other is not None:
             # Other is a broadcasted partition that represents 'by' columns
             # Concatenate it with 'df' to group on its columns names
@@ -135,7 +136,6 @@ def map(
         else:
             by_part = by
 
-        apply_func = cls.try_filter_dict(map_func, df)
         result = apply_func(
             df.groupby(by=by_part, axis=axis, **groupby_args), **map_args
         )
@@ -193,8 +193,12 @@ def reduce(
         # there is a bug in pandas with intersection that forces us to do so:
         # https://github.com/pandas-dev/pandas/issues/39699
         by_part = pandas.Index(df.index.names)
-        if drop and len(df.columns.intersection(by_part)) > 0:
-            df.drop(columns=by_part, errors="ignore", inplace=True)
+        if drop:
+            to_drop = df.columns.intersection(by_part)
+            if isinstance(reduce_func, dict):
+                to_drop = to_drop.difference(reduce_func.keys())
+            if len(to_drop) > 0:
+                df.drop(columns=by_part, errors="ignore", inplace=True)
 
         groupby_args = groupby_args.copy()
         as_index = groupby_args["as_index"]
@@ -210,8 +214,26 @@ def reduce(
         result = apply_func(df.groupby(axis=axis, **groupby_args), **reduce_args)
 
         if not as_index:
-            insert_levels = partition_idx == 0 and (drop or method == "size")
-            result.reset_index(drop=not insert_levels, inplace=True)
+            drop, lvls_to_drop, cols_to_drop = GroupBy.handle_as_index(
+                result.columns,
+                result.index.names,
+                by_part,
+                by_cols_dtypes=(
+                    df.index.dtypes.values
+                    if isinstance(df.index, pandas.MultiIndex)
+                    else (df.index.dtype,)
+                ),
+                is_multi_by=len(by_part) > 1,
+                selection=reduce_func.keys() if isinstance(reduce_func, dict) else None,
+                partition_idx=partition_idx,
+                drop=drop,
+                method=method,
+            )
+            if len(lvls_to_drop) > 0:
+                result.index = result.index.droplevel(lvls_to_drop)
+            if len(cols_to_drop) > 0:
+                result.drop(columns=cols_to_drop, inplace=True)
+            result = result.reset_index(drop=drop)
         # Result could not always be a frame, so wrapping it into DataFrame
         return pandas.DataFrame(result)
 
diff --git a/modin/core/storage_formats/pandas/query_compiler.py b/modin/core/storage_formats/pandas/query_compiler.py
index c65c4cf8caa..8bcd7c0ab04 100644
--- a/modin/core/storage_formats/pandas/query_compiler.py
+++ b/modin/core/storage_formats/pandas/query_compiler.py
@@ -51,6 +51,7 @@
     GroupByReduce,
     groupby_reduce_functions,
 )
+from modin.core.dataframe.algebra.default2pandas.groupby import GroupBy
 
 
 def _get_axis(axis):
@@ -2756,69 +2757,31 @@ def compute_groupby(df, drop=False, partition_idx=0):
                 result.drop(columns=missmatched_cols, inplace=True, errors="ignore")
 
                 if not as_index:
-                    keep_index_levels = len(by) > 1 and any(
-                        isinstance(x, pandas.CategoricalDtype)
-                        for x in df[internal_by_cols].dtypes
+                    drop, lvls_to_drop, cols_to_drop = GroupBy.handle_as_index(
+                        result_cols,
+                        result.index.names,
+                        internal_by_cols,
+                        by_cols_dtypes=df[internal_by_cols].dtypes.values,
+                        is_multi_by=is_multi_by,
+                        selection=(
+                            agg_func.keys() if isinstance(agg_func, dict) else None
+                        ),
+                        partition_idx=partition_idx,
+                        drop=drop,
                     )
-
-                    if internal_by_cols.nlevels != result_cols.nlevels:
-                        cols_to_insert = (
-                            pandas.Index([])
-                            if keep_index_levels
-                            else internal_by_cols.copy()
-                        )
-                    else:
-                        cols_to_insert = (
-                            internal_by_cols.intersection(result_cols)
-                            if keep_index_levels
-                            else internal_by_cols.difference(result_cols)
-                        )
-
-                    if keep_index_levels:
-                        result.drop(
-                            columns=cols_to_insert, inplace=True, errors="ignore"
-                        )
-
-                    drop = True
-                    if partition_idx == 0:
-                        drop = False
-                        if not keep_index_levels:
-                            lvls_to_drop = [
-                                i
-                                for i, name in enumerate(result.index.names)
-                                if name not in cols_to_insert
-                            ]
-                            if len(lvls_to_drop) == result.index.nlevels:
-                                drop = True
-                            else:
-                                result.index = result.index.droplevel(lvls_to_drop)
-
-                    if (
-                        not isinstance(result.index, pandas.MultiIndex)
-                        and result.index.name is None
-                    ):
-                        drop = True
-
+                    if len(lvls_to_drop) > 0:
+                        result.index = result.index.droplevel(lvls_to_drop)
+                    if len(cols_to_drop) > 0:
+                        result.drop(columns=cols_to_drop, inplace=True)
                     result.reset_index(drop=drop, inplace=True)
-
-                new_index_names = [
-                    None
-                    if isinstance(name, str) and name.startswith("__reduced__")
-                    else name
-                    for name in result.index.names
-                ]
-
-                cols_to_drop = (
-                    result.columns[result.columns.str.match(r"__reduced__.*", na=False)]
-                    if hasattr(result.columns, "str")
-                    else []
-                )
-
-                result.index.names = new_index_names
-
-                # Not dropping columns if result is Series
-                if len(result.columns) > 1:
-                    result.drop(columns=cols_to_drop, inplace=True)
+                else:
+                    new_index_names = tuple(
+                        None
+                        if isinstance(name, str) and name.startswith("__reduced__")
+                        else name
+                        for name in result.index.names
+                    )
+                    result.index.names = new_index_names
 
                 return result
 
diff --git a/modin/pandas/groupby.py b/modin/pandas/groupby.py
index c341599d221..a5d32503c80 100644
--- a/modin/pandas/groupby.py
+++ b/modin/pandas/groupby.py
@@ -451,6 +451,22 @@ def try_get_str_func(fn):
                 func, **kwargs
             )
             func_dict = {col: try_get_str_func(fn) for col, fn in func_dict.items()}
+            if (
+                relabeling_required
+                and not self._as_index
+                and any(col in func_dict for col in self._internal_by)
+            ):
+                ErrorMessage.missmatch_with_pandas(
+                    operation="GroupBy.aggregate(**dictionary_renaming_aggregation)",
+                    message=(
+                        "intersection of the columns to aggregate and 'by' is not yet supported when 'as_index=False', "
+                        "columns with group names of the intersection will not be presented in the result. "
+                        "To achieve the desired result rewrite the original code from:\n"
+                        "df.groupby('by_column', as_index=False).agg(agg_func=('by_column', agg_func))\n"
+                        "to the:\n"
+                        "df.groupby('by_column').agg(agg_func=('by_column', agg_func)).reset_index()"
+                    ),
+                )
 
             if any(i not in self._df.columns for i in func_dict.keys()):
                 from pandas.core.base import SpecificationError
diff --git a/modin/pandas/test/test_groupby.py b/modin/pandas/test/test_groupby.py
index 2e116c78ca4..7d1c5a91655 100644
--- a/modin/pandas/test/test_groupby.py
+++ b/modin/pandas/test/test_groupby.py
@@ -16,6 +16,7 @@
 import numpy as np
 import modin.pandas as pd
 from modin.utils import try_cast_to_pandas, get_current_execution, hashable
+from modin.core.dataframe.algebra.default2pandas.groupby import GroupBy
 from modin.pandas.utils import from_pandas, is_scalar
 from .utils import (
     df_equals,
@@ -163,6 +164,10 @@ def test_mixed_dtypes_groupby(as_index):
             {"col2": sum},
             {"col2": "max", "col4": "sum", "col5": "min"},
             {"col2": max, "col4": sum, "col5": "min"},
+            # Intersection of 'by' and agg cols for MapReduce impl
+            {"col0": "count", "col1": "count", "col2": "count"},
+            # Intersection of 'by' and agg cols for FullAxis impl
+            {"col0": "nunique", "col1": "nunique", "col2": "nunique"},
         ]
         for func in agg_functions:
             eval_agg(modin_groupby, pandas_groupby, func)
@@ -363,10 +368,34 @@ def maybe_get_columns(df, by):
         eval_var(modin_groupby, pandas_groupby)
         eval_skew(modin_groupby, pandas_groupby)
 
-    agg_functions = [lambda df: df.sum(), "min", "max", min, sum]
+    agg_functions = [
+        lambda df: df.sum(),
+        "min",
+        "max",
+        min,
+        sum,
+        # Intersection of 'by' and agg cols for MapReduce impl
+        {"col1": "count", "col2": "count"},
+        # Intersection of 'by' and agg cols for FullAxis impl
+        {"col1": "nunique", "col2": "nunique"},
+    ]
     for func in agg_functions:
-        eval_agg(modin_groupby, pandas_groupby, func)
-        eval_aggregate(modin_groupby, pandas_groupby, func)
+        # Pandas raises an exception when 'by' contains categorical key and `as_index=False`
+        # because of this bug: https://github.com/pandas-dev/pandas/issues/36698
+        # Modin correctly processes the result, that's why `check_exception_type=None` in some cases
+        is_pandas_bug_case = not as_index and col1_category and isinstance(func, dict)
+        eval_general(
+            modin_groupby,
+            pandas_groupby,
+            lambda grp: grp.agg(func),
+            check_exception_type=None if is_pandas_bug_case else True,
+        )
+        eval_general(
+            modin_groupby,
+            pandas_groupby,
+            lambda grp: grp.aggregate(func),
+            check_exception_type=None if is_pandas_bug_case else True,
+        )
 
     eval_general(modin_groupby, pandas_groupby, lambda df: df.last(), is_default=True)
     eval_general(
@@ -1366,7 +1395,7 @@ def test_shift_freq(groupby_axis, shift_axis):
                     "min": (list(test_data["int_data"].keys())[-1], min),
                 },
             },
-            marks=pytest.mark.skip("See Modin issue #2542"),
+            marks=pytest.mark.skip("See Modin issue #3602"),
         ),
     ],
 )
@@ -1401,14 +1430,24 @@ def test_agg_func_None_rename(by_and_agg_dict, as_index):
     [["sum", "min", "max"], ["mean", "quantile"]],
     ids=["reduction", "aggregation"],
 )
-def test_dict_agg_rename_mi_columns(as_index, by_length, agg_fns):
+@pytest.mark.parametrize(
+    "intersection_with_by_cols",
+    [pytest.param(True, marks=pytest.mark.skip("See Modin issue #3602")), False],
+)
+def test_dict_agg_rename_mi_columns(
+    as_index, by_length, agg_fns, intersection_with_by_cols
+):
     md_df, pd_df = create_test_dfs(test_data["int_data"])
     mi_columns = generate_multiindex(len(md_df.columns), nlevels=4)
 
     md_df.columns, pd_df.columns = mi_columns, mi_columns
 
     by = list(md_df.columns[:by_length])
-    agg_cols = list(md_df.columns[by_length : by_length + 3])
+    agg_cols = (
+        list(md_df.columns[by_length - 1 : by_length + 2])
+        if intersection_with_by_cols
+        else list(md_df.columns[by_length : by_length + 3])
+    )
 
     agg_dict = {
         f"custom-{i}" + str(agg_fns[i % len(agg_fns)]): (col, agg_fns[i % len(agg_fns)])
@@ -1655,7 +1694,20 @@ def get_columns(df):
                     list(test_data_values[0].keys())[-1]: (sum, min, max),
                 }
             ),
-            marks=pytest.mark.skip("See modin issue #2542"),
+            id="Agg and 'by' intersection (MapReduce implementation)",
+        ),
+        pytest.param(
+            lambda grp: grp.agg(
+                {
+                    list(test_data_values[0].keys())[1]: (max, "mean", "nunique"),
+                    list(test_data_values[0].keys())[-1]: (sum, min, max),
+                }
+            ),
+            id="Agg and 'by' intersection (FullAxis implementation)",
+        ),
+        pytest.param(
+            lambda grp: grp.agg({list(test_data_values[0].keys())[0]: "count"}),
+            id="Agg and 'by' intersection (modin-project/modin#3376)",
         ),
     ],
 )
@@ -1714,3 +1766,110 @@ def test_not_str_by(by, as_index):
     eval_general(md_grp, pd_grp, lambda grp: grp.agg(lambda df: df.mean()))
     eval_general(md_grp, pd_grp, lambda grp: grp.dtypes)
     eval_general(md_grp, pd_grp, lambda grp: grp.first())
+
+
+@pytest.mark.parametrize("internal_by_length", [0, 1, 2])
+@pytest.mark.parametrize("external_by_length", [0, 1, 2])
+@pytest.mark.parametrize("has_categorical_by", [True, False])
+@pytest.mark.parametrize(
+    "agg_func",
+    [
+        pytest.param(
+            lambda grp: grp.apply(lambda df: df.dtypes), id="modin_dtype_impl"
+        ),
+        pytest.param(lambda grp: grp.apply(lambda df: df.sum()), id="apply(sum)"),
+        pytest.param(lambda grp: grp.count(), id="count"),
+        pytest.param(lambda grp: grp.nunique(), id="nunique"),
+        pytest.param(
+            {"c": "sum", "d": "nunique"}, id="dict_agg_no_intersection_with_by"
+        ),
+        pytest.param(
+            {"b": "mean", "c": "sum", "d": "nunique"},
+            id="dict_agg_has_intersection_with_by",
+        ),
+        pytest.param(
+            {"a": "nunique", "c": "sum", "d": "nunique"},
+            id="dict_agg_has_intersection_with_categorical_by",
+        ),
+    ],
+)
+def test_handle_as_index(
+    internal_by_length, external_by_length, has_categorical_by, agg_func, request
+):
+    """
+    Test ``GroupBy.handle_as_index``.
+
+    The role of the ``handle_as_index`` method is to build a groupby result considering
+    ``as_index=False`` from the result that was computed with ``as_index=True``.
+
+    So the testing flow is the following:
+        1. Compute GroupBy result with the ``as_index=True`` parameter via Modin.
+        2. Build ``as_index=False`` result from the ``as_index=True`` using ``handle_as_index`` method.
+        3. Compute GroupBy result with the ``as_index=False`` parameter via Pandas (is the reference result).
+        4. Compare the result from the second step with the reference.
+    """
+    by_length = sum((internal_by_length, external_by_length))
+    if by_length == 0:
+        return
+
+    if (
+        has_categorical_by
+        and by_length > 1
+        and (
+            isinstance(agg_func, dict)
+            or ("nunique" in request.node.callspec.id.split("-"))
+        )
+    ):
+        pytest.skip(
+            "The linked bug makes pandas raise an exception when 'by' is categorical: "
+            "https://github.com/pandas-dev/pandas/issues/36698"
+        )
+
+    data = {
+        "a": [1, 1, 2, 2],
+        "b": [2, 2, 3, 3],
+        "c": [1, 2, 3, 4],
+        "d": [5, 6, 7, 8],
+    }
+
+    if isinstance(agg_func, dict):
+        selection = list(agg_func.keys())
+        agg_dict = agg_func
+        agg_func = lambda grp: grp.agg(agg_dict)  # noqa: E731 (lambda assignment)
+    else:
+        selection = None
+
+    df = pandas.DataFrame(data)
+    external_by_cols = GroupBy.validate_by(df.add_prefix("external_"))
+
+    if has_categorical_by:
+        df = df.astype({"a": "category"})
+
+    internal_by = df.columns[:internal_by_length].tolist()
+    external_by = external_by_cols[:external_by_length]
+
+    by = internal_by + external_by
+
+    grp_res = pd.DataFrame(df).groupby(by, as_index=True)
+    grp_ref = df.groupby(by, as_index=False)
+
+    res = agg_func(grp_res)
+    ref = agg_func(grp_ref)
+
+    drop, lvls_to_drop, cols_to_drop = GroupBy.handle_as_index(
+        result_cols=res.columns,
+        result_index_names=res.index.names,
+        internal_by_cols=internal_by,
+        by_cols_dtypes=df[internal_by].dtypes.values,
+        is_multi_by=len(by) > 1,
+        selection=selection,
+        drop=len(internal_by) != 0,
+    )
+
+    if len(lvls_to_drop) > 0:
+        res.index = res.index.droplevel(lvls_to_drop)
+    if len(cols_to_drop) > 0:
+        res = res.drop(columns=cols_to_drop)
+    res = res.reset_index(drop=drop)
+
+    df_equals(res, ref)


Code review comments:
diff_hunk: @@ -270,6 +293,172 @@ def build_groupby(cls, func):
             return cls.build_aggregate_method(func)
         return cls.build_groupby_reduce_method(func)
 
+    @staticmethod
+    def handle_as_index(
+        result_cols,
+        result_index_names,
+        internal_by_cols,
+        by_cols_dtypes=None,
+        is_multi_by=None,
+        selection=None,
+        partition_idx=0,
+        drop=True,
+        method=None,
+    ):
+        """
+        Help to handle ``as_index=False`` parameter for the GroupBy result.
+
+        This function resolve naming conflicts of the index levels to insert and the column labels
+        for the GroupBy result. The logic of this function assumes that the initial GroupBy result
+        was computed as ``as_index=True``.
+
+        Parameters
+        ----------
+        result_cols : pandas.Index
+            Columns of the GroupBy result.
+        result_index_names : list-like
+            Index names of the GroupBy result.
+        internal_by_cols : list-like
+            Internal 'by' columns.
+        by_cols_dtypes : list-like, optional
+            Data types of the internal 'by' columns. Required to do special casing
+            in case of categorical 'by'. If not specified, assume that there is no
+            categorical data in 'by'.
+        is_multi_by : bool, optional
+            Whether grouping on multiple items. If not specified, assume ``True`` if
+            the amount of `internal_by_cols` is greater than 1, ``False`` otherwise.
+        selection : label or list of labels, optional
+            Set of columns to which aggregation was applied. If not specified
+            assuming that aggregation was applied to all of the available columns.
+        partition_idx : int, default: 0
+            Positional index of the current partition.
+        drop : bool, default: True
+            Indicates whether or not any of the `by` data came from the same frame.
+        method : str, optional
+            Name of the groupby function. This is a hint to be able to do special casing.
+            Note: this parameter is a legacy from the ``groupby_size`` implementation,
+            it's a hacky one and probably will be removed in the future.
+
+        Returns
+        -------
+        drop_index : bool
+            Indicates whether to drop all index levels (True) or insert them into the
+            resulting columns (False).
+        lvls_to_drop : list of ints
+            Contains numeric indices of the levels of the result index to drop as intersected.
+        cols_to_drop : list of labels
+            Contains labels of the columns to drop from the result as intersected.
+
+        Examples
+        --------
+        >>> groupby_result = compute_groupby_without_processing_as_index_parameter()
+        >>> if not as_index:
+        >>>     drop, lvls_to_drop, cols_to_drop = handle_as_index(**extract_required_params(groupby_result))
+        >>>     if len(lvls_to_drop) > 0:
+        >>>         groupby_result.index = groupby_result.index.droplevel(lvls_to_drop)
+        >>>     if len(cols_to_drop) > 0:
+        >>>         groupby_result = groupby_result.drop(columns=cols_to_drop)
+        >>>     groupby_result_with_processed_as_index_parameter = groupby_result.reset_index(drop=drop)
+        >>> else:
+        >>>     groupby_result_with_processed_as_index_parameter = groupby_result
+        """
+        # 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
+        # the "by" into the result (just a fast-path)
+        if method == "size":
+            return False, [], []
+
+        if is_multi_by is None:
+            is_multi_by = len(internal_by_cols) > 1
+
+        # Pandas logic of resolving naming conflicts is the following:
+        #   1. If any categorical is in 'by' and 'by' is multi-column, then the categorical
+        #      index is prioritized: drop intersected columns and insert all of the 'by' index
+        #      levels to the frame as columns.
+        #   2. Otherwise, aggregation result is prioritized: drop intersected index levels and
+        #      insert the filtered ones to the frame as columns.
+        if by_cols_dtypes is not None:
+            keep_index_levels = (
+                is_multi_by
+                and selection is None
+                and any(isinstance(x, pandas.CategoricalDtype) for x in by_cols_dtypes)
+            )
+        else:
+            keep_index_levels = False
+
+        # 1. We insert 'by'-columns to the result at the beginning of the frame and so only to the
+        #    first partition, if partition_idx != 0 we just drop the index. If there are no columns
+        #    that are required to drop (keep_index_levels is True) then we can exit here.
+        # 2. We don't insert 'by'-columns to the result if 'by'-data came from a different
+        #    frame (drop is False), there's only one exception for this rule: if the `method` is "size",
+        #    so if (drop is False) and method is not "size" we just drop the index and so can exit here.
+        if (not keep_index_levels and partition_idx != 0) or (
+            not drop and method != "size"
+        ):
+            return True, [], []
+
+        if not isinstance(internal_by_cols, pandas.Index):
+            if not is_list_like(internal_by_cols):
+                internal_by_cols = [internal_by_cols]
+            internal_by_cols = pandas.Index(internal_by_cols)
+
+        internal_by_cols = (
+            internal_by_cols[~internal_by_cols.str.match(r"__reduced__.*", na=False)]
+            if hasattr(internal_by_cols, "str")
+            else internal_by_cols
+        )
+
+        if selection is not None:
+            selection = (
+                selection
+                if isinstance(selection, pandas.Index)
+                else pandas.Index(selection)
+            )
path: modin/core/dataframe/algebra/default2pandas/groupby.py
text: @user1:
```suggestion
        if selection is not None and not isinstance(selection, pandas.Index):
            selection = pandas.Index(selection)
```

diff_hunk: @@ -270,6 +293,172 @@ def build_groupby(cls, func):
             return cls.build_aggregate_method(func)
         return cls.build_groupby_reduce_method(func)
 
+    @staticmethod
+    def handle_as_index(
+        result_cols,
+        result_index_names,
+        internal_by_cols,
+        by_cols_dtypes=None,
+        is_multi_by=None,
+        selection=None,
+        partition_idx=0,
+        drop=True,
+        method=None,
+    ):
+        """
+        Help to handle ``as_index=False`` parameter for the GroupBy result.
+
+        This function resolve naming conflicts of the index levels to insert and the column labels
+        for the GroupBy result. The logic of this function assumes that the initial GroupBy result
+        was computed as ``as_index=True``.
+
+        Parameters
+        ----------
+        result_cols : pandas.Index
+            Columns of the GroupBy result.
+        result_index_names : list-like
+            Index names of the GroupBy result.
+        internal_by_cols : list-like
+            Internal 'by' columns.
+        by_cols_dtypes : list-like, optional
+            Data types of the internal 'by' columns. Required to do special casing
+            in case of categorical 'by'. If not specified, assume that there is no
+            categorical data in 'by'.
+        is_multi_by : bool, optional
+            Whether grouping on multiple items. If not specified, assume ``True`` if
+            the amount of `internal_by_cols` is greater than 1, ``False`` otherwise.
+        selection : label or list of labels, optional
+            Set of columns to which aggregation was applied. If not specified
+            assuming that aggregation was applied to all of the available columns.
+        partition_idx : int, default: 0
+            Positional index of the current partition.
+        drop : bool, default: True
+            Indicates whether or not any of the `by` data came from the same frame.
+        method : str, optional
+            Name of the groupby function. This is a hint to be able to do special casing.
+            Note: this parameter is a legacy from the ``groupby_size`` implementation,
+            it's a hacky one and probably will be removed in the future.
+
+        Returns
+        -------
+        drop_index : bool
+            Indicates whether to drop all index levels (True) or insert them into the
+            resulting columns (False).
+        lvls_to_drop : list of ints
+            Contains numeric indices of the levels of the result index to drop as intersected.
+        cols_to_drop : list of labels
+            Contains labels of the columns to drop from the result as intersected.
+
+        Examples
+        --------
+        >>> groupby_result = compute_groupby_without_processing_as_index_parameter()
+        >>> if not as_index:
+        >>>     drop, lvls_to_drop, cols_to_drop = handle_as_index(**extract_required_params(groupby_result))
+        >>>     if len(lvls_to_drop) > 0:
+        >>>         groupby_result.index = groupby_result.index.droplevel(lvls_to_drop)
+        >>>     if len(cols_to_drop) > 0:
+        >>>         groupby_result = groupby_result.drop(columns=cols_to_drop)
+        >>>     groupby_result_with_processed_as_index_parameter = groupby_result.reset_index(drop=drop)
+        >>> else:
+        >>>     groupby_result_with_processed_as_index_parameter = groupby_result
+        """
+        # 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
+        # the "by" into the result (just a fast-path)
+        if method == "size":
+            return False, [], []
+
+        if is_multi_by is None:
+            is_multi_by = len(internal_by_cols) > 1
+
+        # Pandas logic of resolving naming conflicts is the following:
+        #   1. If any categorical is in 'by' and 'by' is multi-column, then the categorical
+        #      index is prioritized: drop intersected columns and insert all of the 'by' index
+        #      levels to the frame as columns.
+        #   2. Otherwise, aggregation result is prioritized: drop intersected index levels and
+        #      insert the filtered ones to the frame as columns.
+        if by_cols_dtypes is not None:
+            keep_index_levels = (
+                is_multi_by
+                and selection is None
+                and any(isinstance(x, pandas.CategoricalDtype) for x in by_cols_dtypes)
+            )
+        else:
+            keep_index_levels = False
+
+        # 1. We insert 'by'-columns to the result at the beginning of the frame and so only to the
+        #    first partition, if partition_idx != 0 we just drop the index. If there are no columns
+        #    that are required to drop (keep_index_levels is True) then we can exit here.
+        # 2. We don't insert 'by'-columns to the result if 'by'-data came from a different
+        #    frame (drop is False), there's only one exception for this rule: if the `method` is "size",
+        #    so if (drop is False) and method is not "size" we just drop the index and so can exit here.
+        if (not keep_index_levels and partition_idx != 0) or (
+            not drop and method != "size"
+        ):
+            return True, [], []
+
+        if not isinstance(internal_by_cols, pandas.Index):
+            if not is_list_like(internal_by_cols):
+                internal_by_cols = [internal_by_cols]
+            internal_by_cols = pandas.Index(internal_by_cols)
+
+        internal_by_cols = (
+            internal_by_cols[~internal_by_cols.str.match(r"__reduced__.*", na=False)]
+            if hasattr(internal_by_cols, "str")
+            else internal_by_cols
+        )
+
+        if selection is not None:
+            selection = (
+                selection
+                if isinstance(selection, pandas.Index)
+                else pandas.Index(selection)
+            )
+
+        lvls_to_drop = []
+        cols_to_drop = []
+
+        if not keep_index_levels:
+            # We want to insert only these internal-by-cols that are not presented
+            # in the result in order to not create naming conflicts
+            cols_to_insert = (
+                internal_by_cols.copy()
+                if selection is None and internal_by_cols.nlevels != result_cols.nlevels
+                else frozenset(
+                    (
+                        col
+                        for col in internal_by_cols
+                        if col not in (result_cols if selection is None else selection)
+                    )
+                )
+            )
+        else:
+            cols_to_insert = internal_by_cols
+            # We want to drop such internal-by-cols that are presented
+            # in the result in order to not create naming conflicts
+            cols_to_drop = (
+                pandas.Index([])
+                if internal_by_cols.nlevels != result_cols.nlevels
+                else frozenset((col for col in cols_to_insert if col in result_cols))
path: modin/core/dataframe/algebra/default2pandas/groupby.py
text: @user1:
```suggestion
                else (frozenset(cols_to_insert) & frozenset(result_cols))
```

diff_hunk: @@ -270,6 +293,172 @@ def build_groupby(cls, func):
             return cls.build_aggregate_method(func)
         return cls.build_groupby_reduce_method(func)
 
+    @staticmethod
+    def handle_as_index(
+        result_cols,
+        result_index_names,
+        internal_by_cols,
+        by_cols_dtypes=None,
+        is_multi_by=None,
+        selection=None,
+        partition_idx=0,
+        drop=True,
+        method=None,
+    ):
+        """
+        Help to handle ``as_index=False`` parameter for the GroupBy result.
+
+        This function resolve naming conflicts of the index levels to insert and the column labels
+        for the GroupBy result. The logic of this function assumes that the initial GroupBy result
+        was computed as ``as_index=True``.
+
+        Parameters
+        ----------
+        result_cols : pandas.Index
+            Columns of the GroupBy result.
+        result_index_names : list-like
+            Index names of the GroupBy result.
+        internal_by_cols : list-like
+            Internal 'by' columns.
+        by_cols_dtypes : list-like, optional
+            Data types of the internal 'by' columns. Required to do special casing
+            in case of categorical 'by'. If not specified, assume that there is no
+            categorical data in 'by'.
+        is_multi_by : bool, optional
+            Whether grouping on multiple items. If not specified, assume ``True`` if
+            the amount of `internal_by_cols` is greater than 1, ``False`` otherwise.
+        selection : label or list of labels, optional
+            Set of columns to which aggregation was applied. If not specified
+            assuming that aggregation was applied to all of the available columns.
+        partition_idx : int, default: 0
+            Positional index of the current partition.
+        drop : bool, default: True
+            Indicates whether or not any of the `by` data came from the same frame.
+        method : str, optional
+            Name of the groupby function. This is a hint to be able to do special casing.
+            Note: this parameter is a legacy from the ``groupby_size`` implementation,
+            it's a hacky one and probably will be removed in the future.
+
+        Returns
+        -------
+        drop_index : bool
+            Indicates whether to drop all index levels (True) or insert them into the
+            resulting columns (False).
+        lvls_to_drop : list of ints
+            Contains numeric indices of the levels of the result index to drop as intersected.
+        cols_to_drop : list of labels
+            Contains labels of the columns to drop from the result as intersected.
+
+        Examples
+        --------
+        >>> groupby_result = compute_groupby_without_processing_as_index_parameter()
+        >>> if not as_index:
+        >>>     drop, lvls_to_drop, cols_to_drop = handle_as_index(**extract_required_params(groupby_result))
+        >>>     if len(lvls_to_drop) > 0:
+        >>>         groupby_result.index = groupby_result.index.droplevel(lvls_to_drop)
+        >>>     if len(cols_to_drop) > 0:
+        >>>         groupby_result = groupby_result.drop(columns=cols_to_drop)
+        >>>     groupby_result_with_processed_as_index_parameter = groupby_result.reset_index(drop=drop)
+        >>> else:
+        >>>     groupby_result_with_processed_as_index_parameter = groupby_result
+        """
+        # 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
+        # the "by" into the result (just a fast-path)
+        if method == "size":
+            return False, [], []
+
+        if is_multi_by is None:
+            is_multi_by = len(internal_by_cols) > 1
+
+        # Pandas logic of resolving naming conflicts is the following:
+        #   1. If any categorical is in 'by' and 'by' is multi-column, then the categorical
+        #      index is prioritized: drop intersected columns and insert all of the 'by' index
+        #      levels to the frame as columns.
+        #   2. Otherwise, aggregation result is prioritized: drop intersected index levels and
+        #      insert the filtered ones to the frame as columns.
+        if by_cols_dtypes is not None:
+            keep_index_levels = (
+                is_multi_by
+                and selection is None
+                and any(isinstance(x, pandas.CategoricalDtype) for x in by_cols_dtypes)
+            )
+        else:
+            keep_index_levels = False
+
+        # 1. We insert 'by'-columns to the result at the beginning of the frame and so only to the
+        #    first partition, if partition_idx != 0 we just drop the index. If there are no columns
+        #    that are required to drop (keep_index_levels is True) then we can exit here.
+        # 2. We don't insert 'by'-columns to the result if 'by'-data came from a different
+        #    frame (drop is False), there's only one exception for this rule: if the `method` is "size",
+        #    so if (drop is False) and method is not "size" we just drop the index and so can exit here.
+        if (not keep_index_levels and partition_idx != 0) or (
+            not drop and method != "size"
+        ):
+            return True, [], []
+
+        if not isinstance(internal_by_cols, pandas.Index):
+            if not is_list_like(internal_by_cols):
+                internal_by_cols = [internal_by_cols]
+            internal_by_cols = pandas.Index(internal_by_cols)
+
+        internal_by_cols = (
+            internal_by_cols[~internal_by_cols.str.match(r"__reduced__.*", na=False)]
path: modin/core/dataframe/algebra/default2pandas/groupby.py
text: @user1:
```suggestion
            internal_by_cols[~internal_by_cols.str.startswith("__reduced__", na=False)]
```
or something like that. Do not use regexps unless really necessary, in this case for performance reasons (however, some measurements would be good to prove me right or wrong tho)

@author:
`startswith` is slightly faster indeed:
```
startswith:
        Total time of 10 runs: 6.242122787982225; Mean time: 0.6242122787982225
regexp:
        Total time of 10 runs: 7.681726414011791; Mean time: 0.7681726414011791
```

<details><summary>reproduced</summary>

```python
import pandas
import timeit
import random

random.seed(42)
n_elements = 1_000_000
n_runs = 10

reduced_columns_percentage = 0.1
almost_reduced_columns_percentage = 0.2
regular_columns_percentage = 1 - (
    reduced_columns_percentage + almost_reduced_columns_percentage
)

reduced = [
    f"__reduced__someth{i}ngsomething"
    for i in range(int(n_elements * reduced_columns_percentage))
]
almost_reduced = [
    f"__reduced_someth{i}ngsomething"
    for i in range(int(n_elements * almost_reduced_columns_percentage))
]
regular = [
    f"someth{i}ngsomething" for i in range(int(n_elements * regular_columns_percentage))
]

index = reduced + almost_reduced + regular
random.shuffle(index)

index = pandas.Index(index)

startswith_time = timeit.timeit(
    lambda: index.str.startswith("__reduced__", na=False), number=n_runs
)
regexp_time = timeit.timeit(
    lambda: index.str.match(r"__reduced__.*", na=False), number=n_runs
)

print(
    f"startswith:\n\tTotal time of {n_runs} runs: {startswith_time}; Mean time: {startswith_time / n_runs}"
)
print(f"regexp:\n\tTotal time of {n_runs} runs: {regexp_time}; Mean time: {regexp_time / n_runs}")
```

</details>

P.S. Timings of half a second for this filtering may scare, but note, that in the repro I filter a million values when in the actual code the filtered object is an Index, containing column names to group on, its length is unlikely to be big.

diff_hunk: @@ -1714,3 +1766,110 @@ def test_not_str_by(by, as_index):
     eval_general(md_grp, pd_grp, lambda grp: grp.agg(lambda df: df.mean()))
     eval_general(md_grp, pd_grp, lambda grp: grp.dtypes)
     eval_general(md_grp, pd_grp, lambda grp: grp.first())
+
+
+@pytest.mark.parametrize("internal_by_length", [0, 1, 2])
+@pytest.mark.parametrize("external_by_length", [0, 1, 2])
+@pytest.mark.parametrize("has_categorical_by", [True, False])
+@pytest.mark.parametrize(
+    "agg_func",
+    [
+        pytest.param(
+            lambda grp: grp.apply(lambda df: df.dtypes), id="modin_dtype_impl"
+        ),
+        pytest.param(lambda grp: grp.apply(lambda df: df.sum()), id="apply(sum)"),
+        pytest.param(lambda grp: grp.count(), id="count"),
+        pytest.param(lambda grp: grp.nunique(), id="nunique"),
+        pytest.param(
+            {"c": "sum", "d": "nunique"}, id="dict_agg_no_intersection_with_by"
+        ),
+        pytest.param(
+            {"b": "mean", "c": "sum", "d": "nunique"},
+            id="dict_agg_has_intersection_with_by",
+        ),
+        pytest.param(
+            {"a": "nunique", "c": "sum", "d": "nunique"},
+            id="dict_agg_has_intersection_with_categorical_by",
+        ),
+    ],
+)
+def test_handle_as_index(
+    internal_by_length, external_by_length, has_categorical_by, agg_func, request
+):
+    """
+    Test ``GroupBy.handle_as_index``.
+
+    The role of the ``handle_as_index`` method is to build a groupby result considering
+    ``as_index=False`` from the result that was computed with ``as_index=True``.
+
+    So the testing flow is the following:
+        1. Compute GroupBy result with the ``as_index=True`` parameter via Modin.
+        2. Build ``as_index=False`` result from the ``as_index=True`` using ``handle_as_index`` method.
+        3. Compute GroupBy result with the ``as_index=False`` parameter via Pandas (is the reference result).
path: modin/pandas/test/test_groupby.py
text: @user1:
```suggestion
        3. Compute GroupBy result with the ``as_index=False`` parameter via pandas as the reference result.
```

diff_hunk: @@ -270,6 +293,172 @@ def build_groupby(cls, func):
             return cls.build_aggregate_method(func)
         return cls.build_groupby_reduce_method(func)
 
+    @staticmethod
+    def handle_as_index(
+        result_cols,
+        result_index_names,
+        internal_by_cols,
+        by_cols_dtypes=None,
+        is_multi_by=None,
+        selection=None,
+        partition_idx=0,
+        drop=True,
+        method=None,
+    ):
+        """
+        Help to handle ``as_index=False`` parameter for the GroupBy result.
+
+        This function resolve naming conflicts of the index levels to insert and the column labels
+        for the GroupBy result. The logic of this function assumes that the initial GroupBy result
+        was computed as ``as_index=True``.
+
+        Parameters
+        ----------
+        result_cols : pandas.Index
+            Columns of the GroupBy result.
+        result_index_names : list-like
+            Index names of the GroupBy result.
+        internal_by_cols : list-like
+            Internal 'by' columns.
+        by_cols_dtypes : list-like, optional
+            Data types of the internal 'by' columns. Required to do special casing
+            in case of categorical 'by'. If not specified, assume that there is no
+            categorical data in 'by'.
+        is_multi_by : bool, optional
+            Whether grouping on multiple items. If not specified, assume ``True`` if
+            the amount of `internal_by_cols` is greater than 1, ``False`` otherwise.
+        selection : label or list of labels, optional
+            Set of columns to which aggregation was applied. If not specified
+            assuming that aggregation was applied to all of the available columns.
+        partition_idx : int, default: 0
+            Positional index of the current partition.
+        drop : bool, default: True
+            Indicates whether or not any of the `by` data came from the same frame.
+        method : str, optional
+            Name of the groupby function. This is a hint to be able to do special casing.
+            Note: this parameter is a legacy from the ``groupby_size`` implementation,
+            it's a hacky one and probably will be removed in the future.
+
+        Returns
+        -------
+        drop_index : bool
+            Indicates whether to drop all index levels (True) or insert them into the
+            resulting columns (False).
+        lvls_to_drop : list of ints
+            Contains numeric indices of the levels of the result index to drop as intersected.
+        cols_to_drop : list of labels
+            Contains labels of the columns to drop from the result as intersected.
+
+        Examples
+        --------
+        >>> groupby_result = compute_groupby_without_processing_as_index_parameter()
+        >>> if not as_index:
+        >>>     drop, lvls_to_drop, cols_to_drop = handle_as_index(**extract_required_params(groupby_result))
+        >>>     if len(lvls_to_drop) > 0:
+        >>>         groupby_result.index = groupby_result.index.droplevel(lvls_to_drop)
+        >>>     if len(cols_to_drop) > 0:
+        >>>         groupby_result = groupby_result.drop(columns=cols_to_drop)
+        >>>     groupby_result_with_processed_as_index_parameter = groupby_result.reset_index(drop=drop)
+        >>> else:
+        >>>     groupby_result_with_processed_as_index_parameter = groupby_result
+        """
+        # 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
+        # the "by" into the result (just a fast-path)
+        if method == "size":
+            return False, [], []
+
+        if is_multi_by is None:
+            is_multi_by = len(internal_by_cols) > 1
+
+        # Pandas logic of resolving naming conflicts is the following:
+        #   1. If any categorical is in 'by' and 'by' is multi-column, then the categorical
+        #      index is prioritized: drop intersected columns and insert all of the 'by' index
+        #      levels to the frame as columns.
+        #   2. Otherwise, aggregation result is prioritized: drop intersected index levels and
+        #      insert the filtered ones to the frame as columns.
+        if by_cols_dtypes is not None:
+            keep_index_levels = (
+                is_multi_by
+                and selection is None
+                and any(isinstance(x, pandas.CategoricalDtype) for x in by_cols_dtypes)
+            )
+        else:
+            keep_index_levels = False
+
+        # 1. We insert 'by'-columns to the result at the beginning of the frame and so only to the
+        #    first partition, if partition_idx != 0 we just drop the index. If there are no columns
+        #    that are required to drop (keep_index_levels is True) then we can exit here.
+        # 2. We don't insert 'by'-columns to the result if 'by'-data came from a different
+        #    frame (drop is False), there's only one exception for this rule: if the `method` is "size",
+        #    so if (drop is False) and method is not "size" we just drop the index and so can exit here.
+        if (not keep_index_levels and partition_idx != 0) or (
+            not drop and method != "size"
+        ):
+            return True, [], []
+
+        if not isinstance(internal_by_cols, pandas.Index):
+            if not is_list_like(internal_by_cols):
+                internal_by_cols = [internal_by_cols]
+            internal_by_cols = pandas.Index(internal_by_cols)
+
+        internal_by_cols = (
+            internal_by_cols[~internal_by_cols.str.match(r"__reduced__.*", na=False)]
+            if hasattr(internal_by_cols, "str")
+            else internal_by_cols
+        )
+
+        if selection is not None:
+            selection = (
+                selection
+                if isinstance(selection, pandas.Index)
+                else pandas.Index(selection)
+            )
+
+        lvls_to_drop = []
+        cols_to_drop = []
+
+        if not keep_index_levels:
+            # We want to insert only these internal-by-cols that are not presented
+            # in the result in order to not create naming conflicts
+            cols_to_insert = (
+                internal_by_cols.copy()
+                if selection is None and internal_by_cols.nlevels != result_cols.nlevels
+                else frozenset(
+                    (
+                        col
+                        for col in internal_by_cols
+                        if col not in (result_cols if selection is None else selection)
+                    )
+                )
+            )
+        else:
+            cols_to_insert = internal_by_cols
+            # We want to drop such internal-by-cols that are presented
+            # in the result in order to not create naming conflicts
+            cols_to_drop = (
+                pandas.Index([])
+                if internal_by_cols.nlevels != result_cols.nlevels
+                else frozenset((col for col in cols_to_insert if col in result_cols))
+            )
+
+        lvls_to_drop = (
+            result_index_names
+            if partition_idx != 0
+            else [
+                i
+                for i, name in enumerate(result_index_names)
+                if name not in cols_to_insert
+            ]
+        )
path: modin/core/dataframe/algebra/default2pandas/groupby.py
text: @user1:
again, I suggest rewriting this to a simpler outer if-else form

diff_hunk: @@ -270,6 +293,172 @@ def build_groupby(cls, func):
             return cls.build_aggregate_method(func)
         return cls.build_groupby_reduce_method(func)
 
+    @staticmethod
+    def handle_as_index(
+        result_cols,
+        result_index_names,
+        internal_by_cols,
+        by_cols_dtypes=None,
+        is_multi_by=None,
+        selection=None,
+        partition_idx=0,
+        drop=True,
+        method=None,
+    ):
+        """
+        Help to handle ``as_index=False`` parameter for the GroupBy result.
+
+        This function resolve naming conflicts of the index levels to insert and the column labels
path: modin/core/dataframe/algebra/default2pandas/groupby.py
text: @user1:
```suggestion
        This function resolves naming conflicts of the index levels to insert and the column labels
```

diff_hunk: @@ -1714,3 +1766,110 @@ def test_not_str_by(by, as_index):
     eval_general(md_grp, pd_grp, lambda grp: grp.agg(lambda df: df.mean()))
     eval_general(md_grp, pd_grp, lambda grp: grp.dtypes)
     eval_general(md_grp, pd_grp, lambda grp: grp.first())
+
+
+@pytest.mark.parametrize("internal_by_length", [0, 1, 2])
+@pytest.mark.parametrize("external_by_length", [0, 1, 2])
+@pytest.mark.parametrize("has_categorical_by", [True, False])
+@pytest.mark.parametrize(
+    "agg_func",
+    [
+        pytest.param(
+            lambda grp: grp.apply(lambda df: df.dtypes), id="modin_dtype_impl"
+        ),
+        pytest.param(lambda grp: grp.apply(lambda df: df.sum()), id="apply(sum)"),
+        pytest.param(lambda grp: grp.count(), id="count"),
+        pytest.param(lambda grp: grp.nunique(), id="nunique"),
+        pytest.param(
+            {"c": "sum", "d": "nunique"}, id="dict_agg_no_intersection_with_by"
+        ),
+        pytest.param(
+            {"b": "mean", "c": "sum", "d": "nunique"},
+            id="dict_agg_has_intersection_with_by",
+        ),
+        pytest.param(
+            {"a": "nunique", "c": "sum", "d": "nunique"},
+            id="dict_agg_has_intersection_with_categorical_by",
+        ),
+    ],
+)
+def test_handle_as_index(
+    internal_by_length, external_by_length, has_categorical_by, agg_func, request
+):
+    """
+    Test ``GroupBy.handle_as_index``.
+
+    The role of the ``handle_as_index`` method is to build a groupby result considering
+    ``as_index=False`` from the result that was computed with ``as_index=True``.
+
+    So the testing flow is the following:
+        1. Compute GroupBy result with the ``as_index=True`` parameter via Modin.
+        2. Build ``as_index=False`` result from the ``as_index=True`` using ``handle_as_index`` method.
+        3. Compute GroupBy result with the ``as_index=False`` parameter via Pandas (is the reference result).
+        4. Compare the result from the second step with the reference.
+    """
+    by_length = sum((internal_by_length, external_by_length))
+    if by_length == 0:
+        return
path: modin/pandas/test/test_groupby.py
text: @user1:
shouldn't this be an assertion that the function isn't actually testing anything (called uselessly)?

@author:
you mean `pytest.skip` or something similar?

added this for now:
```python
    if by_length == 0:
        pytest.skip("No keys to group on was passed, skipping the test.")
```

diff_hunk: @@ -270,6 +293,172 @@ def build_groupby(cls, func):
             return cls.build_aggregate_method(func)
         return cls.build_groupby_reduce_method(func)
 
+    @staticmethod
+    def handle_as_index(
+        result_cols,
+        result_index_names,
+        internal_by_cols,
+        by_cols_dtypes=None,
+        is_multi_by=None,
+        selection=None,
+        partition_idx=0,
+        drop=True,
+        method=None,
+    ):
+        """
+        Help to handle ``as_index=False`` parameter for the GroupBy result.
+
+        This function resolve naming conflicts of the index levels to insert and the column labels
+        for the GroupBy result. The logic of this function assumes that the initial GroupBy result
+        was computed as ``as_index=True``.
+
+        Parameters
+        ----------
+        result_cols : pandas.Index
+            Columns of the GroupBy result.
+        result_index_names : list-like
+            Index names of the GroupBy result.
+        internal_by_cols : list-like
+            Internal 'by' columns.
+        by_cols_dtypes : list-like, optional
+            Data types of the internal 'by' columns. Required to do special casing
+            in case of categorical 'by'. If not specified, assume that there is no
+            categorical data in 'by'.
+        is_multi_by : bool, optional
+            Whether grouping on multiple items. If not specified, assume ``True`` if
+            the amount of `internal_by_cols` is greater than 1, ``False`` otherwise.
+        selection : label or list of labels, optional
+            Set of columns to which aggregation was applied. If not specified
+            assuming that aggregation was applied to all of the available columns.
+        partition_idx : int, default: 0
+            Positional index of the current partition.
+        drop : bool, default: True
+            Indicates whether or not any of the `by` data came from the same frame.
+        method : str, optional
+            Name of the groupby function. This is a hint to be able to do special casing.
+            Note: this parameter is a legacy from the ``groupby_size`` implementation,
+            it's a hacky one and probably will be removed in the future.
+
+        Returns
+        -------
+        drop_index : bool
+            Indicates whether to drop all index levels (True) or insert them into the
+            resulting columns (False).
+        lvls_to_drop : list of ints
+            Contains numeric indices of the levels of the result index to drop as intersected.
+        cols_to_drop : list of labels
+            Contains labels of the columns to drop from the result as intersected.
+
+        Examples
+        --------
+        >>> groupby_result = compute_groupby_without_processing_as_index_parameter()
+        >>> if not as_index:
+        >>>     drop, lvls_to_drop, cols_to_drop = handle_as_index(**extract_required_params(groupby_result))
+        >>>     if len(lvls_to_drop) > 0:
+        >>>         groupby_result.index = groupby_result.index.droplevel(lvls_to_drop)
+        >>>     if len(cols_to_drop) > 0:
+        >>>         groupby_result = groupby_result.drop(columns=cols_to_drop)
+        >>>     groupby_result_with_processed_as_index_parameter = groupby_result.reset_index(drop=drop)
+        >>> else:
+        >>>     groupby_result_with_processed_as_index_parameter = groupby_result
+        """
+        # 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
+        # the "by" into the result (just a fast-path)
+        if method == "size":
+            return False, [], []
+
+        if is_multi_by is None:
+            is_multi_by = len(internal_by_cols) > 1
+
+        # Pandas logic of resolving naming conflicts is the following:
+        #   1. If any categorical is in 'by' and 'by' is multi-column, then the categorical
+        #      index is prioritized: drop intersected columns and insert all of the 'by' index
+        #      levels to the frame as columns.
+        #   2. Otherwise, aggregation result is prioritized: drop intersected index levels and
+        #      insert the filtered ones to the frame as columns.
+        if by_cols_dtypes is not None:
+            keep_index_levels = (
+                is_multi_by
+                and selection is None
+                and any(isinstance(x, pandas.CategoricalDtype) for x in by_cols_dtypes)
+            )
+        else:
+            keep_index_levels = False
+
+        # 1. We insert 'by'-columns to the result at the beginning of the frame and so only to the
+        #    first partition, if partition_idx != 0 we just drop the index. If there are no columns
+        #    that are required to drop (keep_index_levels is True) then we can exit here.
+        # 2. We don't insert 'by'-columns to the result if 'by'-data came from a different
+        #    frame (drop is False), there's only one exception for this rule: if the `method` is "size",
+        #    so if (drop is False) and method is not "size" we just drop the index and so can exit here.
+        if (not keep_index_levels and partition_idx != 0) or (
+            not drop and method != "size"
+        ):
+            return True, [], []
+
+        if not isinstance(internal_by_cols, pandas.Index):
+            if not is_list_like(internal_by_cols):
+                internal_by_cols = [internal_by_cols]
+            internal_by_cols = pandas.Index(internal_by_cols)
+
+        internal_by_cols = (
+            internal_by_cols[~internal_by_cols.str.match(r"__reduced__.*", na=False)]
+            if hasattr(internal_by_cols, "str")
+            else internal_by_cols
+        )
+
+        if selection is not None:
+            selection = (
+                selection
+                if isinstance(selection, pandas.Index)
+                else pandas.Index(selection)
+            )
+
+        lvls_to_drop = []
+        cols_to_drop = []
+
+        if not keep_index_levels:
+            # We want to insert only these internal-by-cols that are not presented
+            # in the result in order to not create naming conflicts
+            cols_to_insert = (
+                internal_by_cols.copy()
+                if selection is None and internal_by_cols.nlevels != result_cols.nlevels
+                else frozenset(
+                    (
+                        col
+                        for col in internal_by_cols
+                        if col not in (result_cols if selection is None else selection)
+                    )
+                )
+            )
path: modin/core/dataframe/algebra/default2pandas/groupby.py
text: @user1:
I suggest to rewrite this in a normal `if..else` way, current one looks overcomplicated.

Don't try to fit everything in a single statement, it makes things harder to understand in a lot of cases.

```python
if selection is None:
    if internal_by_cols.nlevels != result_cols.nlevels:
        cols_to_insert = internal_by_cols.copy()
    else:
        cols_to_insert = frozenset(internal_by_cols) - frozenset(result_cols)
else:
    cols_to_insert = frozenset(internal_by_cols) - frozenset(selection)
```
or something along this lines

diff_hunk: @@ -1714,3 +1766,110 @@ def test_not_str_by(by, as_index):
     eval_general(md_grp, pd_grp, lambda grp: grp.agg(lambda df: df.mean()))
     eval_general(md_grp, pd_grp, lambda grp: grp.dtypes)
     eval_general(md_grp, pd_grp, lambda grp: grp.first())
+
+
+@pytest.mark.parametrize("internal_by_length", [0, 1, 2])
+@pytest.mark.parametrize("external_by_length", [0, 1, 2])
+@pytest.mark.parametrize("has_categorical_by", [True, False])
+@pytest.mark.parametrize(
+    "agg_func",
+    [
+        pytest.param(
+            lambda grp: grp.apply(lambda df: df.dtypes), id="modin_dtype_impl"
+        ),
+        pytest.param(lambda grp: grp.apply(lambda df: df.sum()), id="apply(sum)"),
+        pytest.param(lambda grp: grp.count(), id="count"),
+        pytest.param(lambda grp: grp.nunique(), id="nunique"),
+        pytest.param(
+            {"c": "sum", "d": "nunique"}, id="dict_agg_no_intersection_with_by"
+        ),
+        pytest.param(
+            {"b": "mean", "c": "sum", "d": "nunique"},
+            id="dict_agg_has_intersection_with_by",
+        ),
+        pytest.param(
+            {"a": "nunique", "c": "sum", "d": "nunique"},
+            id="dict_agg_has_intersection_with_categorical_by",
+        ),
+    ],
+)
+def test_handle_as_index(
+    internal_by_length, external_by_length, has_categorical_by, agg_func, request
+):
+    """
+    Test ``GroupBy.handle_as_index``.
+
+    The role of the ``handle_as_index`` method is to build a groupby result considering
+    ``as_index=False`` from the result that was computed with ``as_index=True``.
+
+    So the testing flow is the following:
+        1. Compute GroupBy result with the ``as_index=True`` parameter via Modin.
+        2. Build ``as_index=False`` result from the ``as_index=True`` using ``handle_as_index`` method.
+        3. Compute GroupBy result with the ``as_index=False`` parameter via Pandas (is the reference result).
+        4. Compare the result from the second step with the reference.
+    """
+    by_length = sum((internal_by_length, external_by_length))
+    if by_length == 0:
+        return
+
+    if (
+        has_categorical_by
+        and by_length > 1
+        and (
+            isinstance(agg_func, dict)
+            or ("nunique" in request.node.callspec.id.split("-"))
+        )
+    ):
+        pytest.skip(
+            "The linked bug makes pandas raise an exception when 'by' is categorical: "
+            "https://github.com/pandas-dev/pandas/issues/36698"
+        )
+
+    data = {
+        "a": [1, 1, 2, 2],
+        "b": [2, 2, 3, 3],
+        "c": [1, 2, 3, 4],
+        "d": [5, 6, 7, 8],
+    }
+
+    if isinstance(agg_func, dict):
+        selection = list(agg_func.keys())
+        agg_dict = agg_func
+        agg_func = lambda grp: grp.agg(agg_dict)  # noqa: E731 (lambda assignment)
+    else:
+        selection = None
+
+    df = pandas.DataFrame(data)
+    external_by_cols = GroupBy.validate_by(df.add_prefix("external_"))
+
+    if has_categorical_by:
+        df = df.astype({"a": "category"})
+
+    internal_by = df.columns[:internal_by_length].tolist()
+    external_by = external_by_cols[:external_by_length]
+
+    by = internal_by + external_by
+
+    grp_res = pd.DataFrame(df).groupby(by, as_index=True)
+    grp_ref = df.groupby(by, as_index=False)
path: modin/pandas/test/test_groupby.py
text: @user1:
please don't save a few typed letters and spell the words in the full - I suggest replacing `res` with `result` and `ref` with `reference` or smth similar (here and in other places of this function)

diff_hunk: @@ -1714,3 +1766,110 @@ def test_not_str_by(by, as_index):
     eval_general(md_grp, pd_grp, lambda grp: grp.agg(lambda df: df.mean()))
     eval_general(md_grp, pd_grp, lambda grp: grp.dtypes)
     eval_general(md_grp, pd_grp, lambda grp: grp.first())
+
+
+@pytest.mark.parametrize("internal_by_length", [0, 1, 2])
+@pytest.mark.parametrize("external_by_length", [0, 1, 2])
+@pytest.mark.parametrize("has_categorical_by", [True, False])
+@pytest.mark.parametrize(
+    "agg_func",
+    [
+        pytest.param(
+            lambda grp: grp.apply(lambda df: df.dtypes), id="modin_dtype_impl"
+        ),
+        pytest.param(lambda grp: grp.apply(lambda df: df.sum()), id="apply(sum)"),
+        pytest.param(lambda grp: grp.count(), id="count"),
+        pytest.param(lambda grp: grp.nunique(), id="nunique"),
+        pytest.param(
+            {"c": "sum", "d": "nunique"}, id="dict_agg_no_intersection_with_by"
+        ),
+        pytest.param(
+            {"b": "mean", "c": "sum", "d": "nunique"},
+            id="dict_agg_has_intersection_with_by",
+        ),
+        pytest.param(
+            {"a": "nunique", "c": "sum", "d": "nunique"},
+            id="dict_agg_has_intersection_with_categorical_by",
+        ),
+    ],
+)
+def test_handle_as_index(
+    internal_by_length, external_by_length, has_categorical_by, agg_func, request
+):
+    """
+    Test ``GroupBy.handle_as_index``.
+
+    The role of the ``handle_as_index`` method is to build a groupby result considering
+    ``as_index=False`` from the result that was computed with ``as_index=True``.
+
+    So the testing flow is the following:
+        1. Compute GroupBy result with the ``as_index=True`` parameter via Modin.
+        2. Build ``as_index=False`` result from the ``as_index=True`` using ``handle_as_index`` method.
+        3. Compute GroupBy result with the ``as_index=False`` parameter via Pandas (is the reference result).
+        4. Compare the result from the second step with the reference.
+    """
+    by_length = sum((internal_by_length, external_by_length))
+    if by_length == 0:
+        return
+
+    if (
+        has_categorical_by
+        and by_length > 1
+        and (
+            isinstance(agg_func, dict)
+            or ("nunique" in request.node.callspec.id.split("-"))
+        )
+    ):
+        pytest.skip(
+            "The linked bug makes pandas raise an exception when 'by' is categorical: "
+            "https://github.com/pandas-dev/pandas/issues/36698"
+        )
+
+    data = {
+        "a": [1, 1, 2, 2],
+        "b": [2, 2, 3, 3],
+        "c": [1, 2, 3, 4],
+        "d": [5, 6, 7, 8],
+    }
+
+    if isinstance(agg_func, dict):
+        selection = list(agg_func.keys())
+        agg_dict = agg_func
+        agg_func = lambda grp: grp.agg(agg_dict)  # noqa: E731 (lambda assignment)
+    else:
+        selection = None
+
+    df = pandas.DataFrame(data)
+    external_by_cols = GroupBy.validate_by(df.add_prefix("external_"))
+
+    if has_categorical_by:
+        df = df.astype({"a": "category"})
+
+    internal_by = df.columns[:internal_by_length].tolist()
+    external_by = external_by_cols[:external_by_length]
+
+    by = internal_by + external_by
+
+    grp_res = pd.DataFrame(df).groupby(by, as_index=True)
+    grp_ref = df.groupby(by, as_index=False)
+
+    res = agg_func(grp_res)
+    ref = agg_func(grp_ref)
+
+    drop, lvls_to_drop, cols_to_drop = GroupBy.handle_as_index(
+        result_cols=res.columns,
+        result_index_names=res.index.names,
+        internal_by_cols=internal_by,
+        by_cols_dtypes=df[internal_by].dtypes.values,
+        is_multi_by=len(by) > 1,
+        selection=selection,
+        drop=len(internal_by) != 0,
+    )
+
+    if len(lvls_to_drop) > 0:
+        res.index = res.index.droplevel(lvls_to_drop)
+    if len(cols_to_drop) > 0:
+        res = res.drop(columns=cols_to_drop)
+    res = res.reset_index(drop=drop)
path: modin/pandas/test/test_groupby.py
text: @user1:
I saw this pattern at least 4 or 5 times already... I feel that it should be a helper function or something.

@author:
wrapped it into the `handle_as_index_for_dataframe` function