Add post-processing steps to `dask_cudf.groupby.CudfSeriesGroupby.agg…

…regate` (#8694)

Closes #8655 

Adds some post-processing steps to Dask-cuDF's series groupby when using the optimized codepaths for aggregations, to match [those done by Dask](https://github.com/dask/dask/blob/8601b540f8e7eac95fa739a5ca28f1d707299ed0/dask/dataframe/groupby.py#L1917-L1921). These ensure that a `dask_cudf.Series` is always returned for the groupby operation, which was a problem observed in #8655.

Authors:
  - Charles Blackmon-Luca (https://github.com/charlesbluca)

Approvers:
  - Richard (Rick) Zamora (https://github.com/rjzamora)

URL: #8694

python/dask_cudf/dask_cudf/groupby.py

@@ -143,7 +143,7 @@ def aggregate(self, arg, split_every=None, split_out=1):
                 sep=self.sep,
                 sort=self.sort,
                 as_index=self.as_index,
-            )
+            )[self._slice]
 
         return super().aggregate(
             arg, split_every=split_every, split_out=split_out

python/dask_cudf/dask_cudf/tests/test_groupby.py

@@ -48,10 +48,7 @@ def test_groupby_basic_aggs(aggregation):
     "func",
     [
         lambda df: df.groupby("x").agg({"y": "max"}),
-        pytest.param(
-            lambda df: df.groupby("x").y.agg(["sum", "max"]),
-            marks=pytest.mark.skip,
-        ),
+        lambda df: df.groupby("x").y.agg(["sum", "max"]),
     ],
 )
 def test_groupby_agg(func):
@@ -98,7 +95,6 @@ def test_groupby_agg_empty_partition(tmpdir, split_out):
     dd.assert_eq(gb.compute().sort_index(), expect)
 
 
-@pytest.mark.xfail(reason="cudf issues")
 @pytest.mark.parametrize(
     "func",
     [lambda df: df.groupby("x").std(), lambda df: df.groupby("x").y.std()],
@@ -115,23 +111,17 @@ def test_groupby_std(func):
 
     ddf = dask_cudf.from_cudf(gdf, npartitions=5)
 
-    a = func(gdf.to_pandas())
+    a = func(gdf).to_pandas()
     b = func(ddf).compute().to_pandas()
 
-    a.index.name = None
-    a.name = None
-    b.index.name = None
-
     dd.assert_eq(a, b)
 
 
 @pytest.mark.parametrize(
     "func",
     [
         lambda df: df.groupby("x").agg({"y": "collect"}),
-        pytest.param(
-            lambda df: df.groupby("x").y.agg("collect"), marks=pytest.mark.skip
-        ),
+        lambda df: df.groupby("x").y.agg("collect"),
     ],
 )
 def test_groupby_collect(func):