Groupby dictionary aggregation works incorrect in cases when 'by' and columns to aggregate has intersection

[gshimansky]

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:

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)

Describe the problem

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

        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

[dchigarev]

@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:

modin/modin/data_management/functions/groupby_function.py

Lines 194 to 196 in e48ac35

	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)

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:

Changes

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"]

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):

modin/modin/data_management/functions/groupby_function.py

Lines 211 to 213 in e48ac35

	if not as_index:
	insert_levels = partition_idx == 0 and (drop or method == "size")
	result.reset_index(drop=not insert_levels, inplace=True)

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 method, so after #3298 will be merged we could handle as_index=False cases as well.

[dchigarev]

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

[dchigarev]

Another reproducer from just closed #2542

Reproducer

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)

[gshimansky]

@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.

[gshimansky]

One more code example with inconsistent behavior between Modin and Pandas:

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)

[gshimansky]

Is this issue unblocked now so it is possible to fix it?

[dchigarev]

Is this issue unblocked now so it is possible to fix it?

Yes, I'm working on it