Repository: pandas-dev/pandas
Pull request: BUG: `agg` with dictlike and  non-unique col will return wrong type
- [x] closes #51099 (Replace xxxx with the GitHub issue number)
- [x] [Tests added and passed](https://pandas.pydata.org/pandas-docs/dev/development/contributing_codebase.html#writing-tests) if fixing a bug or adding a new feature
- [ ] All [code checks passed](https://pandas.pydata.org/pandas-docs/dev/development/contributing_codebase.html#pre-commit).
- [ ] Added [type annotations](https://pandas.pydata.org/pandas-docs/dev/development/contributing_codebase.html#type-hints) to new arguments/methods/functions.
- [x] Added an entry in the latest `doc/source/whatsnew/vX.X.X.rst` file if fixing a bug or adding a new feature.


Problem statement:
BUG: apply/agg with dictlike and non-unique columns
```python
df = pd.DataFrame(
    {"A": [None, 2, 3], "B": [1.0, np.nan, 3.0], "C": ["foo", None, "bar"]}
)
df.columns = ["A", "A", "C"]

result = df.agg({"A": "count"})  # same with 'apply' instead of 'agg'
expected = df["A"].count()
tm.assert_series_equal(result, expected)
```

This goes through Apply.agg_dict_like, which does

```
results = {
    key: obj._gotitem(key, ndim=1).agg(how) for key, how in arg.items()
}
```

Which operates column-by-column on the relevant columns _if_ columns are unique.  But with a repeated column obj._gotitem returns a DataFrame, so the .agg returns a DataFrame.

No existing test cases get here with non-unique columns, or with Resample/GroupBy objects whose underlying object has non-unique columns.

cc @rhshadrach

Hints from issue discussion:
I think we can resolve this by taking the indices within columns for each key and iterating through these rather than relying on _gotitem.
take

Commit message: Merge branch 'main' into bug-agg-nonunique-col

Patch to review:
diff --git a/doc/source/whatsnew/v2.0.0.rst b/doc/source/whatsnew/v2.0.0.rst
index 193f837400ac0..a6767b89995b0 100644
--- a/doc/source/whatsnew/v2.0.0.rst
+++ b/doc/source/whatsnew/v2.0.0.rst
@@ -1371,6 +1371,7 @@ Reshaping
 - Bug in :meth:`DataFrame.explode` raising ``ValueError`` on multiple columns with ``NaN`` values or empty lists (:issue:`46084`)
 - Bug in :meth:`DataFrame.transpose` with ``IntervalDtype`` column with ``timedelta64[ns]`` endpoints (:issue:`44917`)
 - Bug in :meth:`DataFrame.agg` and :meth:`Series.agg` would ignore arguments when passed a list of functions (:issue:`50863`)
+- Bug in :meth:`DataFrame.agg` and :meth:`Series.agg` on non-unique columns would return incorrect type when dist-like argument passed in (:issue:`51099`)
 
 Sparse
 ^^^^^^
diff --git a/pandas/core/apply.py b/pandas/core/apply.py
index 2ffd49f674cfb..cbe1d77d4ff0b 100644
--- a/pandas/core/apply.py
+++ b/pandas/core/apply.py
@@ -410,17 +410,32 @@ def agg_dict_like(self) -> DataFrame | Series:
             context_manager = com.temp_setattr(obj, "as_index", True)
         else:
             context_manager = nullcontext()
+
+        if isinstance(selected_obj, ABCDataFrame):
+            is_non_unique_col = selected_obj.columns.duplicated().tolist()
+        else:
+            is_non_unique_col = [False]
+
         with context_manager:
             if selected_obj.ndim == 1:
                 # key only used for output
-                colg = obj._gotitem(selection, ndim=1)
-                results = {key: colg.agg(how) for key, how in arg.items()}
+                key_res = obj._gotitem(selection, ndim=1)
+                results = {key: key_res.agg(how) for key, how in arg.items()}
+            elif any(is_non_unique_col):
+                # GH#51099
+                # results is a dict of lists
+                results = {}
+                for key, how in arg.items():
+                    key_res = []
+                    for col_idx in selected_obj.columns.get_indexer_for([key]):
+                        col = selected_obj.iloc[:, col_idx]
+                        key_res.append(col.agg(how))
+                    results[key] = key_res
             else:
                 # key used for column selection and output
                 results = {
                     key: obj._gotitem(key, ndim=1).agg(how) for key, how in arg.items()
                 }
-
         # set the final keys
         keys = list(arg.keys())
 
@@ -455,7 +470,7 @@ def agg_dict_like(self) -> DataFrame | Series:
         else:
             from pandas import Series
 
-            # we have a dict of scalars
+            # we have a dict of scalars or a list of scalars
             # GH 36212 use name only if obj is a series
             if obj.ndim == 1:
                 obj = cast("Series", obj)
@@ -463,7 +478,17 @@ def agg_dict_like(self) -> DataFrame | Series:
             else:
                 name = None
 
-            result = Series(results, name=name)
+            if any(is_non_unique_col):
+                # Expand the scalar list and construct a series.
+                series_list = []
+                for key, value in results.items():
+                    assert isinstance(value, list)
+                    series_list.append(Series(value, index=[key] * len(value)))
+
+                result = concat(series_list, axis=0)
+                result.name = name
+            else:
+                result = Series(results, name=name)
 
         return result
 
diff --git a/pandas/tests/apply/test_frame_apply.py b/pandas/tests/apply/test_frame_apply.py
index c7eb8c0332e84..f6907c0cea09b 100644
--- a/pandas/tests/apply/test_frame_apply.py
+++ b/pandas/tests/apply/test_frame_apply.py
@@ -1665,3 +1665,15 @@ def foo2(x, b=2, c=0):
         columns=MultiIndex.from_tuples([("x", "foo1"), ("x", "foo2")]),
     )
     tm.assert_frame_equal(result, expected)
+
+
+def test_agg_dist_like_and_nonunique_columns():
+    # GH#51099
+    df = DataFrame(
+        {"A": [None, 2, 3], "B": [1.0, np.nan, 3.0], "C": ["foo", None, "bar"]}
+    )
+    df.columns = ["A", "A", "C"]
+
+    result = df.agg({"A": "count"})
+    expected = df["A"].count()
+    tm.assert_series_equal(result, expected)


Code review comments:
diff_hunk: @@ -410,17 +410,32 @@ def agg_dict_like(self) -> DataFrame | Series:
             context_manager = com.temp_setattr(obj, "as_index", True)
         else:
             context_manager = nullcontext()
+
+        if isinstance(selected_obj, ABCDataFrame):
path: pandas/core/apply.py
text: @user1:
Below can you just check `if selected_obj.ndim == 2 and selected_obj.columns.nunique() < len(selected_obj.columns)`?

diff_hunk: @@ -455,15 +470,25 @@ def agg_dict_like(self) -> DataFrame | Series:
         else:
             from pandas import Series
 
-            # we have a dict of scalars
+            # we have a dict of scalars or a list of scalars
             # GH 36212 use name only if obj is a series
             if obj.ndim == 1:
                 obj = cast("Series", obj)
                 name = obj.name
             else:
                 name = None
 
-            result = Series(results, name=name)
+            if any(is_non_unique_col):
+                # Expand the scalar list and construct a series.
+                series_list = []
+                for key, value in results.items():
+                    assert isinstance(value, list)
+                    series_list.append(Series(value, index=[key] * len(value)))
+
+                result = concat(series_list, axis=0)
+                result.name = name
+            else:
+                result = Series(results, name=name)
path: pandas/core/apply.py
text: @user1:
Instead of storing results in a dictionary, can you store them in two lists: result_index, result_data. Then you can do `Series(result_data, index=result_index)` and you don't need separate logic here.

diff_hunk: @@ -410,17 +410,32 @@ def agg_dict_like(self) -> DataFrame | Series:
             context_manager = com.temp_setattr(obj, "as_index", True)
         else:
             context_manager = nullcontext()
+
+        if isinstance(selected_obj, ABCDataFrame):
+            is_non_unique_col = selected_obj.columns.duplicated().tolist()
+        else:
+            is_non_unique_col = [False]
+
         with context_manager:
             if selected_obj.ndim == 1:
                 # key only used for output
-                colg = obj._gotitem(selection, ndim=1)
-                results = {key: colg.agg(how) for key, how in arg.items()}
+                key_res = obj._gotitem(selection, ndim=1)
+                results = {key: key_res.agg(how) for key, how in arg.items()}
+            elif any(is_non_unique_col):
+                # GH#51099
+                # results is a dict of lists
+                results = {}
+                for key, how in arg.items():
+                    key_res = []
+                    for col_idx in selected_obj.columns.get_indexer_for([key]):
path: pandas/core/apply.py
text: @user1:
get_indexer_for is expensive, I think we should call it once. Something like

```
indices = selected_obj.columns.get_indexer_for(keys)
labels = selected_obj.columns.take(indices)
label_to_indices = defaultdict(list)
for index, label in zip(indices, labels):
    label_to_indices[label].append(index)
```

Then you can iterate through label_to_indices.

diff_hunk: @@ -410,17 +410,32 @@ def agg_dict_like(self) -> DataFrame | Series:
             context_manager = com.temp_setattr(obj, "as_index", True)
         else:
             context_manager = nullcontext()
+
+        if isinstance(selected_obj, ABCDataFrame):
+            is_non_unique_col = selected_obj.columns.duplicated().tolist()
+        else:
+            is_non_unique_col = [False]
+
         with context_manager:
             if selected_obj.ndim == 1:
                 # key only used for output
-                colg = obj._gotitem(selection, ndim=1)
-                results = {key: colg.agg(how) for key, how in arg.items()}
+                key_res = obj._gotitem(selection, ndim=1)
+                results = {key: key_res.agg(how) for key, how in arg.items()}
+            elif any(is_non_unique_col):
+                # GH#51099
+                # results is a dict of lists
+                results = {}
+                for key, how in arg.items():
+                    key_res = []
+                    for col_idx in selected_obj.columns.get_indexer_for([key]):
+                        col = selected_obj.iloc[:, col_idx]
path: pandas/core/apply.py
text: @user1:
Use `._ix` instead of `.iloc` (~3x faster)