Repository: modin-project/modin

Issue:
Poor Performance on TPC-H Queries
In a benchmark project ([link](https://github.com/hesamshahrokhi/modin_tpch_benchmark)), I tried to compare the performance of Pandas and Modin (on Ray and Dask) on simple TPC-H queries (Q1 and Q6) written in Pandas. The results ([link](https://github.com/hesamshahrokhi/modin_tpch_benchmark/blob/main/Results_TPCH_SF1.txt)) show that the ```read_table``` API is 2x faster than its Pandas equivalent on Ray. However, the same API on Dask, and also the individual query run times on both backends are much slower than Pandas. Moreover, the results of Q1 are not correct.
 
Could you please help me to understand why this is the case? I tried to use the suggested configurations and imported the data using ```read_table``` which is officially supported by Modin.


Discussion:
Hi @hesamshahrokhi! Thanks for the question!

The method you use to initialize dask is not intended for Modin:
https://github.com/hesamshahrokhi/modin_tpch_benchmark/blob/e7cf88cf5d6152c32593193a94c837f5fde4f298/tpch_dask.py#L12C17-L12C32. The `processes` parameter should be `True`. Could I also ask why you initialize the dask yourself rather than relying on the initialization that Modin does implicitly?

> Moreover, the results of Q1 are not correct.

Let's see what's wrong. Could you please provide more information?
Hi @anmyachev
Thanks for your quick response!

I did the explicit initialization of Dask and Ray because of a runtime warning that asked me to do so. Now, based on your suggestion, I entirely removed the explicit initializations for [Ray](https://github.com/hesamshahrokhi/modin_tpch_benchmark/blob/main/tpch_ray.py#L14-L15) and [Dask](https://github.com/hesamshahrokhi/modin_tpch_benchmark/blob/main/tpch_dask.py#L11-L12). But, after re-running the benchmark, I still see poor performance from Modin (on both backends). Here are the updated benchmark [results](https://github.com/hesamshahrokhi/modin_tpch_benchmark/blob/main/Results_TPCH_SF1.txt) and the mentioned warnings for [Ray](https://github.com/hesamshahrokhi/modin_tpch_benchmark/blob/main/Results_TPCH_SF1.txt#L22-L26) and [Dask](https://github.com/hesamshahrokhi/modin_tpch_benchmark/blob/main/Results_TPCH_SF1.txt#L48-L53).

Regarding the correctness issue in Q1, that still exists, I would refer you to the correct answer by [Pandas](https://github.com/hesamshahrokhi/modin_tpch_benchmark/blob/main/Results_TPCH_SF1.txt#L5-L10) and the incorrect answers by Modin on [Ray](https://github.com/hesamshahrokhi/modin_tpch_benchmark/blob/main/Results_TPCH_SF1.txt#L31-L36) and [Dask](https://github.com/hesamshahrokhi/modin_tpch_benchmark/blob/main/Results_TPCH_SF1.txt#L57-L62). In the Modin outputs, the titles for these columns ```sum_disc_price, sum_charge, avg_disc, and count_order``` are not match to their relevant values.
>  of a runtime warning that asked me to do so.

This was not the most obvious warning, which in fact simply showed that Modin initializes the engine itself. We have already removed it in Modin 0.27.0. 

What version of Modin are you using? Can you use version 0.27.0? Perhaps the bug has already been fixed and the performance is better.

> Regarding the correctness issue in Q1, that still exists, I would refer you to the correct answer by [Pandas](https://github.com/hesamshahrokhi/modin_tpch_benchmark/blob/main/Results_TPCH_SF1.txt#L5-L10) and the incorrect answers by Modin on [Ray](https://github.com/hesamshahrokhi/modin_tpch_benchmark/blob/main/Results_TPCH_SF1.txt#L31-L36) and [Dask](https://github.com/hesamshahrokhi/modin_tpch_benchmark/blob/main/Results_TPCH_SF1.txt#L57-L62). In the Modin outputs, the titles for these columns sum_disc_price, sum_charge, avg_disc, and count_order are not match to their relevant values.

We'll look at this problem.
I'm able to reproduce the problem with the following reproducer (for Modin 0.27.0):
```python
import numpy as np
import modin.pandas as pd
from modin.pandas.test.utils import df_equals

np.random.seed(42)

columns = ['l_returnflag', 'l_linestatus', 'l_discount', 'l_extendedprice', 'l_quantity', 'b', 'a'] # , 'l_extendedprice'] first fail
columns = ['l_returnflag', 'l_linestatus', 'l_discount', 'l_extendedprice', 'l_quantity']
df = pd.DataFrame(np.random.randint(0, 100, size=(1000, len(columns))), columns=columns)
df["a"] = ((df.l_extendedprice) * (1 - (df.l_discount)))
df['b'] = ((df.l_extendedprice) * (1 - (df.l_discount))) * 1.94

agg_df = df \
    .groupby(['l_returnflag', 'l_linestatus']) \
    .agg(
        sum_qty=("l_quantity", "sum"),
        sum_base_price=("l_extendedprice", "sum"),
        sum_disc_price=("a", "sum"),
        sum_charge=("b", "sum"),
        avg_qty=("l_quantity", "mean"),
        avg_price=("l_extendedprice", "mean"),
        avg_disc=("l_discount", "mean"),
        count_order=("l_returnflag", "count"),
    ).reset_index()


agg_df_pd = df._to_pandas() \
    .groupby(['l_returnflag', 'l_linestatus']) \
    .agg(
        sum_qty=("l_quantity", "sum"),
        sum_base_price=("l_extendedprice", "sum"),
        sum_disc_price=("a", "sum"),
        sum_charge=("b", "sum"),
        avg_qty=("l_quantity", "mean"),
        avg_price=("l_extendedprice", "mean"),
        avg_disc=("l_discount", "mean"),
        count_order=("l_returnflag", "count"),
    ).reset_index()

df_equals(agg_df, agg_df_pd)

```
> > of a runtime warning that asked me to do so.
> 
> This was not the most obvious warning, which in fact simply showed that Modin initializes the engine itself. We have already removed it in Modin 0.27.0.
> 
> What version of Modin are you using? Can you use version 0.27.0? Perhaps the bug has already been fixed and the performance is better.


I changed my modin version to 0.27.0. The warnings are gone but the performance and correctness issues still exist.


> 
> > Regarding the correctness issue in Q1, that still exists, I would refer you to the correct answer by [Pandas](https://github.com/hesamshahrokhi/modin_tpch_benchmark/blob/main/Results_TPCH_SF1.txt#L5-L10) and the incorrect answers by Modin on [Ray](https://github.com/hesamshahrokhi/modin_tpch_benchmark/blob/main/Results_TPCH_SF1.txt#L31-L36) and [Dask](https://github.com/hesamshahrokhi/modin_tpch_benchmark/blob/main/Results_TPCH_SF1.txt#L57-L62). In the Modin outputs, the titles for these columns sum_disc_price, sum_charge, avg_disc, and count_order are not match to their relevant values.
> 
> We'll look at this problem.

Thanks :)

Patch:
diff --git a/modin/pandas/groupby.py b/modin/pandas/groupby.py
--- a/modin/pandas/groupby.py
+++ b/modin/pandas/groupby.py
@@ -22,6 +22,7 @@
 import pandas.core.common as com
 import pandas.core.groupby
 from pandas._libs import lib
+from pandas.api.types import is_scalar
 from pandas.core.apply import reconstruct_func
 from pandas.core.dtypes.common import (
     is_datetime64_any_dtype,
@@ -904,19 +905,40 @@ def aggregate(self, func=None, *args, engine=None, engine_kwargs=None, **kwargs)
 
         do_relabel = None
         if isinstance(func, dict) or func is None:
-            relabeling_required, func_dict, new_columns, order = reconstruct_func(
+            # the order from `reconstruct_func` cannot be used correctly if there
+            # is more than one columnar partition, since for correct use all columns
+            # must be available within one partition.
+            old_kwargs = dict(kwargs)
+            relabeling_required, func_dict, new_columns, _ = reconstruct_func(
                 func, **kwargs
             )
 
             if relabeling_required:
 
                 def do_relabel(obj_to_relabel):  # noqa: F811
-                    new_order, new_columns_idx = order, pandas.Index(new_columns)
+                    # unwrap nested labels into one level tuple
+                    result_labels = [None] * len(old_kwargs)
+                    for idx, labels in enumerate(old_kwargs.values()):
+                        if is_scalar(labels) or callable(labels):
+                            result_labels[idx] = (
+                                labels if not callable(labels) else labels.__name__
+                            )
+                            continue
+                        new_elem = []
+                        for label in labels:
+                            if is_scalar(label) or callable(label):
+                                new_elem.append(
+                                    label if not callable(label) else label.__name__
+                                )
+                            else:
+                                new_elem.extend(label)
+                        result_labels[idx] = tuple(new_elem)
+
+                    new_order = obj_to_relabel.columns.get_indexer(result_labels)
+                    new_columns_idx = pandas.Index(new_columns)
                     if not self._as_index:
                         nby_cols = len(obj_to_relabel.columns) - len(new_columns_idx)
-                        new_order = np.concatenate(
-                            [np.arange(nby_cols), new_order + nby_cols]
-                        )
+                        new_order = np.concatenate([np.arange(nby_cols), new_order])
                         by_cols = obj_to_relabel.columns[:nby_cols]
                         if by_cols.nlevels != new_columns_idx.nlevels:
                             by_cols = by_cols.remove_unused_levels()


Test patch:
diff --git a/modin/pandas/test/test_groupby.py b/modin/pandas/test/test_groupby.py
--- a/modin/pandas/test/test_groupby.py
+++ b/modin/pandas/test/test_groupby.py
@@ -3087,6 +3087,60 @@ def test_groupby_named_aggregation():
     )
 
 
+def test_groupby_several_column_partitions():
+    # see details in #6948
+    columns = [
+        "l_returnflag",
+        "l_linestatus",
+        "l_discount",
+        "l_extendedprice",
+        "l_quantity",
+    ]
+    modin_df, pandas_df = create_test_dfs(
+        np.random.randint(0, 100, size=(1000, len(columns))), columns=columns
+    )
+
+    pandas_df["a"] = (pandas_df.l_extendedprice) * (1 - (pandas_df.l_discount))
+    # to create another column partition
+    modin_df["a"] = (modin_df.l_extendedprice) * (1 - (modin_df.l_discount))
+
+    eval_general(
+        modin_df,
+        pandas_df,
+        lambda df: df.groupby(["l_returnflag", "l_linestatus"])
+        .agg(
+            sum_qty=("l_quantity", "sum"),
+            sum_base_price=("l_extendedprice", "sum"),
+            sum_disc_price=("a", "sum"),
+            # sum_charge=("b", "sum"),
+            avg_qty=("l_quantity", "mean"),
+            avg_price=("l_extendedprice", "mean"),
+            avg_disc=("l_discount", "mean"),
+            count_order=("l_returnflag", "count"),
+        )
+        .reset_index(),
+    )
+
+
+def test_groupby_named_agg():
+    # from pandas docs
+
+    data = {
+        "A": [1, 1, 2, 2],
+        "B": [1, 2, 3, 4],
+        "C": [0.362838, 0.227877, 1.267767, -0.562860],
+    }
+    modin_df, pandas_df = create_test_dfs(data)
+    eval_general(
+        modin_df,
+        pandas_df,
+        lambda df: df.groupby("A").agg(
+            b_min=pd.NamedAgg(column="B", aggfunc="min"),
+            c_sum=pd.NamedAgg(column="C", aggfunc="sum"),
+        ),
+    )
+
+
 ### TEST GROUPBY WARNINGS ###