BUG: named agg result depended on the output name matching the column name (GH#63743)

@@ -1886,7 +1886,7 @@ def transform(self):   def reconstruct_func(-    func: AggFuncType | None, **kwargs+    func: AggFuncType | None, allow_skip_normalization: bool = False, /, **kwargs ) -> tuple[bool, AggFuncType, tuple[str, ...] | None, npt.NDArray[np.intp] | None]:     """     This is the internal function to reconstruct func given if there is relabeling@@ -1902,10 +1902,22 @@ def reconstruct_func(     names, and the reconstructed order of columns.     If relabeling is False, the columns and order will be None. +    Named aggregation is the one exception: when ``allow_skip_normalization`` is+    True, every output name equals its source column name, and every aggfunc+    reduces to a scalar, relabeling is reported as False (and columns/order as+    None) even though named aggregation was used, because the caller can consume+    the un-normalized func directly.+     Parameters     ----------     func: agg function (e.g. 'min' or Callable) or list of agg functions         (e.g. ['min', np.max]) or dictionary (e.g. {'A': ['min', np.max]}).+    allow_skip_normalization: bool, default False+        Whether the caller can handle the un-normalized ``{column: aggfunc}`` form+        that named aggregation reduces to when every output name equals its source+        column name and every aggfunc is a scalar reduction. Callers that rely on+        ``columns``/``order`` being returned whenever named aggregation was used+        must leave this False.     **kwargs: dict, kwargs used in is_multi_agg_with_relabel and         normalize_keyword_aggregation function for relabelling @@ -1924,6 +1936,9 @@ def reconstruct_func(     >>> reconstruct_func("min")     (False, 'min', None, None)     """+    # deferred: pandas.core.groupby.generic imports from this module at import+    #  time, so a top-level import here would be circular+    from pandas.core.groupby.base import reduction_kernels     from pandas.core.groupby.generic import NamedAgg      relabeling = func is None and (@@ -1946,7 +1961,7 @@ def reconstruct_func(             raise TypeError("Must provide 'func' or tuples of '(column, aggfunc).")      if relabeling:-        normalization_needed = False+        normalization_needed = not allow_skip_normalization         # error: Incompatible types in assignment (expression has type         # "MutableMapping[Hashable, list[Callable[..., Any] | str]]", variable has type         # "Callable[..., Any] | str | list[Callable[..., Any] | str] |@@ -1964,7 +1979,17 @@ def reconstruct_func(             else:                 column, aggfunc = val -            if column != key:+            # The un-normalized {column: aggfunc} form only matches the normalized+            #  one when the aggfunc reduces each group to a single scalar. A+            #  list-like aggfunc, or a string naming a non-reduction groupby method+            #  (e.g. "describe"/"ohlc", which give a frame per group), would widen+            #  the result past the one output column per keyword named aggregation+            #  promises.+            if (+                column != key+                or is_list_like(aggfunc)+                or (isinstance(aggfunc, str) and aggfunc not in reduction_kernels)+            ):                 normalization_needed = True             converted_kwargs[key] = column, aggfunc 

@@ -2264,7 +2264,18 @@ def aggregate(         1   1.0         2   3.0         """-        relabeling, func, columns, order = reconstruct_func(func, **kwargs)+        # This method can consume the un-normalized {column: aggfunc} form, but only+        #  when the columns are unique: dict aggregation fans a single key out to+        #  every matching column, while named aggregation must produce exactly one+        #  output per keyword.+        #  `func is None` is a precondition for relabeling at all, and short-circuits+        #  materializing _obj_with_exclusions on the far more common plain-agg path.+        allow_skip_normalization = (+            func is None and self._obj_with_exclusions.columns.is_unique+        )+        relabeling, func, columns, order = reconstruct_func(+            func, allow_skip_normalization, **kwargs+        )         func = maybe_mangle_lambdas(func)          if maybe_use_numba(engine):

@@ -1,4 +1,5 @@ import numpy as np+import pytest  import pandas as pd import pandas._testing as tm@@ -103,3 +104,39 @@ def test_reconstruct_func():     result = pd.core.apply.reconstruct_func("min")     expected = (False, "min", None, None)     tm.assert_equal(result, expected)+++def test_reconstruct_func_allow_skip_normalization():+    # GH#63743 the fast path is observationally equivalent by design, so only+    #  reconstruct_func itself can pin that it still fires+    result = pd.core.apply.reconstruct_func(None, True, B=("B", "sum"))+    assert result == (False, {"B": "sum"}, None, None)+++def test_agg_relabel_output_name_matches_column():+    # GH#63743 named aggregation returns a DataFrame indexed by the output+    #  names even when those names match the source column names+    df = pd.DataFrame({"A": [1, 2, 3], "B": [4, 5, 6]})++    result = df.agg(A=("A", "min"))+    expected = pd.DataFrame({"A": [1]}, index=pd.Index(["A"]))+    tm.assert_frame_equal(result, expected)++    result = df.agg(A=("A", "min"), B=("B", "max"))+    expected = pd.DataFrame(+        {"A": [1.0, np.nan], "B": [np.nan, 6.0]}, index=pd.Index(["A", "B"])+    )+    tm.assert_frame_equal(result, expected)+++def test_agg_relabel_output_name_matches_column_axis_1():+    # GH#63743 named aggregation with axis=1 raises regardless of whether the+    #  output name matches the column name+    df = pd.DataFrame({"A": [1, 2, 3], "B": [4, 5, 6]}, index=["A", "B", "C"])+    msg = "Named aggregation is not supported when axis=1."++    with pytest.raises(NotImplementedError, match=msg):+        df.agg(A=("A", "min"), axis=1)++    with pytest.raises(NotImplementedError, match=msg):+        df.agg(x=("A", "min"), axis=1)

@@ -2095,6 +2095,56 @@ def test_agg_relabel_with_name_match_and_namedagg():     tm.assert_frame_equal(result, expected)  +def test_agg_relabel_with_name_match_listlike_aggfunc():+    # GH#63743 a list-like aggfunc takes the same path whether or not the output+    #  name matches the column name+    df = pd.DataFrame({"A": [0, 0, 1, 1], "B": [1, 2, 3, 4]})++    with pytest.raises(TypeError, match="unhashable"):+        df.groupby("A").agg(B=("B", ["sum", "max"]))++    with pytest.raises(TypeError, match="unhashable"):+        df.groupby("A").agg(x=("B", ["sum", "max"]))++    # a tuple aggfunc is read as a single (name, func) pair, not two aggfuncs+    result = df.groupby("A").agg(B=("B", ("sum", "max")))+    expected = df.groupby("A").agg(x=("B", ("sum", "max")))+    expected.columns = ["B"]+    tm.assert_frame_equal(result, expected)+++@pytest.mark.parametrize("aggfunc", ["describe", "ohlc"])+def test_agg_relabel_with_name_match_non_reduction(aggfunc):+    # GH#63743 a string aggfunc that is not a reduction gives a frame per group,+    #  so it must still produce one output column per keyword when the output+    #  name matches the column name+    df = pd.DataFrame({"A": [0, 0, 1, 1], "B": [1, 2, 3, 4]})++    result = df.groupby("A").agg(B=("B", aggfunc))+    expected = df.groupby("A").agg(z=("B", aggfunc))+    expected.columns = ["B"]+    tm.assert_frame_equal(result, expected)++    result = df.groupby("A").agg(B=pd.NamedAgg("B", aggfunc))+    tm.assert_frame_equal(result, expected)+++def test_agg_relabel_with_name_match_duplicate_columns():+    # GH#63743 named aggregation gives one output column per keyword even when+    #  the source label is duplicated and the output name matches it+    df = pd.DataFrame(+        [[0, 1, 2], [0, 3, 4], [1, 5, 6], [1, 7, 8]], columns=["A", "B", "B"]+    )++    result = df.groupby("A").agg(B=("B", "sum"))+    expected = df.groupby("A").agg(x=("B", "sum"))+    expected.columns = ["B"]+    tm.assert_frame_equal(result, expected)++    result = df.groupby("A").agg(B=pd.NamedAgg("B", "sum"))+    tm.assert_frame_equal(result, expected)++ def test_multiple_partial_functions_same_name():     # GH#28570     quant50 = partial(np.percentile, q=50)