From 2527669a00bb06cf069fe1d4ac958d076e9e090c Mon Sep 17 00:00:00 2001 From: ChengjieLi Date: Tue, 24 Oct 2023 13:40:44 +0800 Subject: [PATCH 1/3] fix --- python/xorbits/_mars/dataframe/groupby/sort.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/python/xorbits/_mars/dataframe/groupby/sort.py b/python/xorbits/_mars/dataframe/groupby/sort.py index 6e31500e5..762b7e0b6 100644 --- a/python/xorbits/_mars/dataframe/groupby/sort.py +++ b/python/xorbits/_mars/dataframe/groupby/sort.py @@ -117,8 +117,12 @@ def output_limit(self): def _execute_map(cls, ctx, op: "DataFrameGroupbySortShuffle"): df, pivots = [ctx[c.key] for c in op.inputs] out = op.outputs[0] + xdf = cudf if op.gpu else pd def _get_out_df(p_index, in_df): + if isinstance(in_df.index, xdf.MultiIndex): + # TODO: performance issue + in_df = in_df.sort_index() if p_index == 0: out_df = in_df.loc[: pivots[p_index]] elif p_index == op.n_partition - 1: From 22b96ea06a9a72752cd4bc231c0cd79626eb9deb Mon Sep 17 00:00:00 2001 From: ChengjieLi Date: Wed, 25 Oct 2023 13:32:24 +0800 Subject: [PATCH 2/3] fix --- python/xorbits/_mars/dataframe/groupby/aggregation.py | 10 ++++++++++ python/xorbits/_mars/dataframe/groupby/sort.py | 4 ---- 2 files changed, 10 insertions(+), 4 deletions(-) diff --git a/python/xorbits/_mars/dataframe/groupby/aggregation.py b/python/xorbits/_mars/dataframe/groupby/aggregation.py index 1d038944b..56531ee35 100644 --- a/python/xorbits/_mars/dataframe/groupby/aggregation.py +++ b/python/xorbits/_mars/dataframe/groupby/aggregation.py @@ -1150,6 +1150,9 @@ def _wrapped_func(col): size_recorder = ctx.get_remote_object(op.size_recorder_name) size_recorder.record(raw_size, agg_size) + for ad in agg_dfs: + if not ad.index.is_monotonic_increasing: + print(f"==============map: {ad}") ctx[op.outputs[0].key] = tuple(agg_dfs) @classmethod @@ -1186,6 +1189,13 @@ def _execute_combine(cls, ctx, op: "DataFrameGroupByAgg"): combines.append( cls._do_predefined_agg(input_obj, agg_func_name, gpu=op.gpu, **kwds) ) + for i, combine in enumerate(combines): + if ( + isinstance(combine.index, xdf.MultiIndex) + and op.groupby_params["sort"] is True + and not combine.index.is_monotonic_increasing + ): + combines[i] = combine.sort_index() ctx[op.outputs[0].key] = tuple(combines) @classmethod diff --git a/python/xorbits/_mars/dataframe/groupby/sort.py b/python/xorbits/_mars/dataframe/groupby/sort.py index 762b7e0b6..6e31500e5 100644 --- a/python/xorbits/_mars/dataframe/groupby/sort.py +++ b/python/xorbits/_mars/dataframe/groupby/sort.py @@ -117,12 +117,8 @@ def output_limit(self): def _execute_map(cls, ctx, op: "DataFrameGroupbySortShuffle"): df, pivots = [ctx[c.key] for c in op.inputs] out = op.outputs[0] - xdf = cudf if op.gpu else pd def _get_out_df(p_index, in_df): - if isinstance(in_df.index, xdf.MultiIndex): - # TODO: performance issue - in_df = in_df.sort_index() if p_index == 0: out_df = in_df.loc[: pivots[p_index]] elif p_index == op.n_partition - 1: From 1870c357d421537e96a853c73c583502655158af Mon Sep 17 00:00:00 2001 From: ChengjieLi Date: Wed, 25 Oct 2023 13:33:43 +0800 Subject: [PATCH 3/3] fix --- python/xorbits/_mars/dataframe/groupby/aggregation.py | 3 --- 1 file changed, 3 deletions(-) diff --git a/python/xorbits/_mars/dataframe/groupby/aggregation.py b/python/xorbits/_mars/dataframe/groupby/aggregation.py index 56531ee35..1363831ae 100644 --- a/python/xorbits/_mars/dataframe/groupby/aggregation.py +++ b/python/xorbits/_mars/dataframe/groupby/aggregation.py @@ -1150,9 +1150,6 @@ def _wrapped_func(col): size_recorder = ctx.get_remote_object(op.size_recorder_name) size_recorder.record(raw_size, agg_size) - for ad in agg_dfs: - if not ad.index.is_monotonic_increasing: - print(f"==============map: {ad}") ctx[op.outputs[0].key] = tuple(agg_dfs) @classmethod