Groupby benchmark fails after PR #2317 merge

[gshimansky]

System information

• OS Platform and Distribution (e.g., Linux Ubuntu 16.04):

Ubuntu 20.04.1 LTS

• Modin version (modin.__version__):

0.8.2.1+16.g3e32d02

• Python version:

Python 3.8.6

• Code we can use to reproduce:

python -u -m pytest --log-cli-level=info --benchmark-sort=name --benchmark-verbose --benchmark-json=results.json -r A -s --tb=native 'ci/benchmarks/test_benchmarks.py::test_groupby_sum[agg(sum)-no_indexing-10000x10000-int-modin]'

Benchmark fails with exception:

Traceback (most recent call last):
  File "/localdisk/gashiman/modin/ci/benchmarks/test_benchmarks.py", line 206, in test_groupby_sum
    result = benchmark.pedantic(
  File "/nfs/site/home/gashiman/.local/lib/python3.8/site-packages/pytest_benchmark/fixture.py", line 139, in pedantic
    return self._raw_pedantic(target, args=args, kwargs=kwargs, setup=setup, rounds=rounds,
  File "/nfs/site/home/gashiman/.local/lib/python3.8/site-packages/pytest_benchmark/fixture.py", line 213, in _raw_pedantic
    runner(loops_range)
  File "/nfs/site/home/gashiman/.local/lib/python3.8/site-packages/pytest_benchmark/fixture.py", line 87, in runner
    sys.settrace(None)
  File "/localdisk/gashiman/modin/ci/benchmarks/test_benchmarks.py", line 170, in benchmark_groupby_agg_sum_function
    result = gb.agg(sum)
  File "/localdisk/gashiman/modin/modin/pandas/groupby.py", line 399, in aggregate
    result = self._apply_agg_function(
  File "/localdisk/gashiman/modin/modin/pandas/groupby.py", line 887, in _apply_agg_function
    new_manager = groupby_qc.groupby_agg(
  File "/localdisk/gashiman/modin/modin/backends/pandas/query_compiler.py", line 2633, in groupby_agg
    new_modin_frame = self._modin_frame._apply_full_axis(
  File "/localdisk/gashiman/modin/modin/engines/base/frame/data.py", line 1301, in _apply_full_axis
    return self.broadcast_apply_full_axis(
  File "/localdisk/gashiman/modin/modin/engines/base/frame/data.py", line 1673, in broadcast_apply_full_axis
    new_axes = [
  File "/localdisk/gashiman/modin/modin/engines/base/frame/data.py", line 1674, in <listcomp>
    self._compute_axis_labels(i, new_partitions)
  File "/localdisk/gashiman/modin/modin/engines/base/frame/data.py", line 289, in _compute_axis_labels
    return self._frame_mgr_cls.get_indices(
  File "/localdisk/gashiman/modin/modin/engines/ray/pandas_on_ray/frame/partition_manager.py", line 99, in get_indices
    new_idx = ray.get(new_idx)
  File "/localdisk/gashiman/miniconda3/lib/python3.8/site-packages/ray/worker.py", line 1452, in get
    raise value.as_instanceof_cause()
ray.exceptions.RayTaskError(TypeError): ray::deploy_ray_func() (pid=4067967, ip=10.241.129.55)
  File "python/ray/_raylet.pyx", line 446, in ray._raylet.execute_task
  File "python/ray/_raylet.pyx", line 468, in ray._raylet.execute_task
ray.exceptions.RayTaskError: ray::deploy_ray_func() (pid=4067969, ip=10.241.129.55)
  File "python/ray/_raylet.pyx", line 482, in ray._raylet.execute_task
  File "/localdisk/gashiman/modin/modin/engines/ray/pandas_on_ray/frame/axis_partition.py", line 105, in deploy_ray_func
    result = func(*args)
  File "/localdisk/gashiman/modin/modin/engines/base/frame/axis_partition.py", line 224, in deploy_axis_func
    result = func(dataframe, **kwargs)
  File "/localdisk/gashiman/modin/modin/engines/base/frame/data.py", line 1036, in _map_reduce_func
    series_result = func(df, *args, **kwargs)
  File "/localdisk/gashiman/modin/modin/backends/pandas/query_compiler.py", line 2634, in <lambda>
    axis, lambda df: groupby_agg_builder(df)
  File "/localdisk/gashiman/modin/modin/backends/pandas/query_compiler.py", line 2627, in groupby_agg_builder
    return compute_groupby(df)
  File "/localdisk/gashiman/modin/modin/backends/pandas/query_compiler.py", line 2618, in compute_groupby
    result = agg_func(grouped_df, **agg_kwargs)
  File "/localdisk/gashiman/modin/modin/utils.py", line 123, in wrapper
    result = func(*args, **kwargs)
TypeError: unsupported operand type(s) for +: 'int' and 'tuple'

[gshimansky]

There is a problem with fixing this bug. When as_index=False Pandas moves by column to the first place in result DataFrame. This rearrangement is done for all .agg calls of DataFrameGroupBy object. But this rearrangement doesn't happen for .apply and .dtypes calls of DataFrameGroupBy object.
Currently we implement all three mentioned and some other calls using the same function _apply_agg_function which has no way to distinguish between .agg and .apply inside of it. I implemented necessary rearrangement in my last commit gshimansky@70ff9e2 but then it breaks .apply and .dtypes tests.
So the question are

1. What is the correct way to do this rearrangement where should it be done? Inside of Pandas backend only in groupby_agg or should it be done on API level in _apply_agg_function? On API level we may not have some information needed to make a decision.
2. How should we distinguish when rearrangement happens and when it doesn't happen? Should we add another flag that .apply and .dtypes. pass to _apply_agg_function?

Here is a test script that you can run on Pandas and Modin (my branch with fixed exception) and Modin before PR #2317 was merged. Maybe also @dchigarev with multi-column aggregation which does significant changes to groupby but still has this bug.

import modin.pandas as pd
#import pandas as pd

df = pd.DataFrame({
    "col1": [ 1,  2,  1,  2,  1,  3,  3,  2,  3],
    "col2": [11, 12, 13, 14, 15, 16, 17, 18, 19],
    "col3": [ 4,  4,  4,  4,  5,  5,  5,  5,  5],
    "col4": [ 1,  1,  1,  1,  1,  1,  1,  1,  1]
})
print(df)

gb0 = df.groupby('col3', as_index=False)
df0 = gb0.agg(sum)
print("agg func:\n", df0)

gb1 = df.groupby('col3', as_index=False)
df1 = gb1.agg('sum')
print("agg string:\n", df1)

gb2 = df.groupby('col3', as_index=False)
df2 = gb2.apply(sum)
print("apply:\n", df2)

[dchigarev]

@gshimansky
I'm not sure that I understood the problem, as far as I know, pandas do the following logic in groupby:

if not `apply` then:
    drop 'by' columns
else:
    do not drop 'by' columns

result = do_groupby()

if `as_index` is False:
    insert 'by' columns at the beginning of the `result` in case, these columns not in the resulted frame [1]

if any by is categorical: [2]
    drop all 'by' columns from the resulted frame;
    insert all 'by' columns at the beginning of `result`

[1] [2]

We have tests for all that case that passes in test_simple_row_groupby (apply + as_index=False + categorical 'by')

About reproducer:
Only the first case seems to be failing, it happens because we're wrapping built-in function into wrap_udf_function inside groupby_agg which we shouldn't do. The simple fix is to translate built-in function into its name at the API level of aggregate.

if callable(func) and isinstance(func, types.BuiltinFunctionType):
    func = func.__name__

These two lines added at the beginning of modin.pandas.DataFrameGroupBy.aggregate fixes that issue, and makes output of all three reproducers similar to pandas

[gshimansky]

Additional extension to the above test program is by adding user lambdas so that not only builtin functions are tested:

gb3 = df.groupby('col3', as_index=False)
df3 = gb3.agg(lambda df: df.sum())
print("agg lambda func:\n", df3)

gb4 = df.groupby('col3', as_index=False)
df4 = gb4.apply(lambda df: df.sum())
print("apply lambda:\n", df4)