Fix GroupBy.apply result assembly, UDF closure side effects, and empty-frame dtypes

python/cudf/benchmarks/internal/bench_fast_slow_proxy.py

@@ -0,0 +1,59 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Benchmarks of cudf.pandas proxy argument transformation."""
+
+import pytest
+
+from cudf.pandas.fast_slow_proxy import _transform_arg, make_final_proxy_type
+
+
+@pytest.fixture(scope="module")
+def proxy_object():
+    class Fast:
+        def __init__(self, x):
+            self.x = x
+
+        def to_slow(self):
+            return Slow(self.x)
+
+    class Slow:
+        def __init__(self, x):
+            self.x = x
+
+    Pxy = make_final_proxy_type(
+        "Pxy",
+        Fast,
+        Slow,
+        fast_to_slow=lambda fast: fast.to_slow(),
+        slow_to_fast=lambda slow: Fast(slow.x),
+    )
+    return Pxy(1)
+
+
+@pytest.mark.parametrize("size", [10, 10_000])
+def bench_transform_arg_unchanged_list(benchmark, size):
+    # No element needs transforming: the identity-scan returns the
+    # original container without rebuilding it.
+    arg = list(range(size))
+    benchmark(lambda: _transform_arg(arg, "_fsproxy_slow", set()))
+
+
+@pytest.mark.parametrize("size", [10, 10_000])
+def bench_transform_arg_unchanged_dict(benchmark, size):
+    arg = {i: i for i in range(size)}
+    benchmark(lambda: _transform_arg(arg, "_fsproxy_slow", set()))
+
+
+@pytest.mark.parametrize("size", [10, 10_000])
+def bench_transform_arg_list_with_proxy(benchmark, proxy_object, size):
+    # One proxy element forces the rebuild path.
+    arg = [*range(size - 1), proxy_object]
+    benchmark(lambda: _transform_arg(arg, "_fsproxy_slow", set()))
+
+
+@pytest.mark.parametrize("size", [10, 10_000])
+def bench_transform_arg_dict_with_proxy(benchmark, proxy_object, size):
+    arg = {i: i for i in range(size - 1)}
+    arg["proxy"] = proxy_object
+    benchmark(lambda: _transform_arg(arg, "_fsproxy_slow", set()))

python/cudf/cudf/core/column/column.py

@@ -3,6 +3,7 @@
 
 from __future__ import annotations
 
+import datetime
 import pickle
 import warnings
 from collections.abc import (
@@ -3864,9 +3865,17 @@ def as_column(
                     length=length,
                 )
             elif (
-                isinstance(element, (pd.Timestamp, pd.Timedelta, pd.Interval))
+                isinstance(
+                    element,
+                    (datetime.datetime, datetime.timedelta, pd.Interval),
+                )
                 or element is pd.NaT
             ):
+                # datetime.datetime/timedelta cover their pd.Timestamp/
+                # pd.Timedelta subclasses; routing stdlib datetimes through
+                # pandas keeps mixed datetime+non-datetime inputs on the
+                # object-dtype path (MixedTypeError) instead of silently
+                # coercing them.
                 # TODO: Remove this after
                 # https://github.com/apache/arrow/issues/26492
                 # is fixed.

python/cudf/cudf/core/dataframe.py

@@ -924,11 +924,13 @@ def _mapping_to_column_accessor(
             if (
                 dtype is None
                 and len(column) == 0
-                and isinstance(value, (list, tuple, range))
+                and isinstance(value, (list, tuple, Iterator))
             ):
-                # pandas' DataFrame constructor coerces untyped empty
-                # sequences to float64 (unlike Series([]), which stays
-                # object).
+                # pandas' DataFrame constructor defaults untyped empty
+                # sequences (list/tuple/iterator) to float64 (numpy's
+                # default for np.array([])), unlike Series([]) which
+                # defaults to object. An empty range stays int64 like
+                # pandas (as_column already handles it via from_range).
                 column = column_empty(0, dtype=np.dtype(np.float64))
             value_lengths.add(len(column))
             col_data[key] = column
@@ -4387,9 +4389,20 @@ def rename(
             result.index = out_index
 
         if columns:
-            result._data = result._data.rename_levels(
-                mapper=columns, level=level
-            )
+            new_ca = result._data.rename_levels(mapper=columns, level=level)
+            # pandas' rename rebuilds the columns Index from the transformed
+            # labels (``Index(items, tupleize_cols=False)`` in
+            # ``_transform_index``), re-inferring dtypes rather than
+            # preserving the originals: renaming object-dtype columns to
+            # all-string labels yields ``str``, and MultiIndex level dtypes
+            # are likewise re-inferred.
+            if new_ca.multiindex:
+                new_ca._level_dtypes = None
+            else:
+                new_ca.label_dtype = pd.Index(
+                    new_ca.names, tupleize_cols=False
+                ).dtype
+            result._data = new_ca
 
         return result
 
@@ -6686,8 +6699,11 @@ def quantile(
                 include=[np.number], exclude=["datetime64", "timedelta64"]
             )
 
-        if columns is None:
-            columns = set(data_df._column_names)
+        if columns is not None:
+            requested = set(columns)
+            data_df = data_df[
+                [k for k in data_df._column_names if k in requested]
+            ]
 
         if isinstance(q, numbers.Number):
             q_is_number = True
@@ -6737,17 +6753,16 @@ def quantile(
             interpolation = interpolation or "linear"
             result = {}
             for k in data_df._column_names:
-                if k in columns:
-                    ser = data_df[k]
-                    res = ser.quantile(
-                        qs,
-                        interpolation=interpolation,
-                        exact=exact,
-                        quant_index=False,
-                    )._column
-                    if len(res) == 0:
-                        res = column_empty(row_count=len(qs), dtype=ser.dtype)
-                    result[k] = res
+                ser = data_df[k]
+                res = ser.quantile(
+                    qs,
+                    interpolation=interpolation,
+                    exact=exact,
+                    quant_index=False,
+                )._column
+                if len(res) == 0:
+                    res = column_empty(row_count=len(qs), dtype=ser.dtype)
+                result[k] = res
         result_ca = ColumnAccessor(
             result,
             multiindex=data_df._data.multiindex,

python/cudf/cudf/core/groupby/groupby.py

@@ -2445,6 +2445,16 @@ def _post_process_chunk_results(
 
         if not len(chunk_results):
             return self.obj.head(0)
+        if (
+            isinstance(self.obj, DataFrame)
+            and not isinstance(chunk_results, ColumnBase)
+            and all(res is None for res in chunk_results)
+        ):
+            # pandas GH9684/GH57775: an all-None DataFrameGroupBy.apply
+            # returns an empty frame keeping the (non-grouping) columns and
+            # dtypes. (An all-None SeriesGroupBy.apply stays in the scalar
+            # branch below: pandas returns an object Series of Nones.)
+            return grouped_values.head(0).reset_index(drop=True)
         if isinstance(chunk_results, ColumnBase) or is_scalar(
             chunk_results[0]
         ):
@@ -2473,41 +2483,49 @@ def _post_process_chunk_results(
                 result.columns = result.columns.set_names(
                     [chunk_results[0].name]
                 )
-            # When the UDF is like df.x + df.y, the result for each
-            # group is the same length as the original group
-            elif (total_rows := sum(len(chk) for chk in chunk_results)) in {
-                len(self.obj),
-                len(group_names),
-            }:
-                result = concat(chunk_results)
-                if total_rows == len(group_names):
-                    result.index = group_names
-                    # TODO: Is there a better way to determine what
-                    # the column name should be, especially if we applied
-                    # a nameless UDF.
-                    result = result.to_frame(
-                        name=grouped_values._column_names[0]
-                    )
-                else:
-                    index_data = group_keys._data.copy(deep=True)
-                    inner_name = grouped_values.index.name
-                    index_data[None] = grouped_values.index._column
-                    mi = MultiIndex._from_data(index_data)
-                    # ColumnAccessor keys must be unique, so the inner
-                    # level's name (which may duplicate a key name) is
-                    # restored after construction.
-                    mi.names = [*mi.names[:-1], inner_name]
-                    result.index = mi
-            elif len(chunk_results) == len(group_names):
-                result = concat(chunk_results, axis=1).T
+            # pandas stacks Series results that share an identical index
+            # into a DataFrame with one row per group and columns given by
+            # the common index (DataFrameGroupBy._wrap_applied_output_series)
+            elif all(
+                chunk_results[0].index.equals(chk.index)
+                for chk in chunk_results[1:]
+            ):
+                # a consistent Series name becomes the columns-axis name
+                # (pandas GH6124). Chunks are renamed positionally before
+                # the axis=1 concat because cuDF rejects duplicate column
+                # names.
+                names = {chk.name for chk in chunk_results}
+                result = concat(
+                    [chk.rename(i) for i, chk in enumerate(chunk_results)],
+                    axis=1,
+                ).T
                 result.index = group_names
                 result.index.names = self.grouping.names
+                if len(names) == 1:
+                    result._data._level_names = (names.pop(),)
             else:
-                raise TypeError(
-                    "Error handling Groupby apply output with input of "
-                    f"type {type(self.obj)} and output of "
-                    f"type {type(chunk_results[0])}"
+                # pandas GH8467: Series results with differing indexes are
+                # concatenated along axis 0 into a Series with the group
+                # keys prepended as the outer index level(s), each key
+                # repeated by its chunk's actual length and the UDF-returned
+                # index kept as the inner level
+                # (GroupBy._concat_objects with ``not_indexed_same=True``).
+                # This also covers transform-like UDFs: chunks indexed like
+                # their input concatenate back to the grouped input's index.
+                lengths = [len(chk) for chk in chunk_results]
+                result = concat(chunk_results)
+                gather = as_column(
+                    np.repeat(np.arange(len(group_names)), lengths)
                 )
+                index_data = {
+                    i: col.take(gather)
+                    for i, col in enumerate(group_names._columns)
+                }
+                inner_name = result.index.name
+                index_data[None] = result.index._column
+                mi = MultiIndex._from_data(index_data)
+                mi.names = [*self.grouping.names, inner_name]
+                result.index = mi
         else:
             result = concat(chunk_results)
             if self._group_keys:
@@ -2525,6 +2543,19 @@ def _post_process_chunk_results(
                 # construction.
                 mi.names = [*mi.names[:-1], inner_name]
                 result.index = mi
+            elif len(result) == len(grouped_values) and result.index.equals(
+                grouped_values.index
+            ):
+                # Every chunk result is indexed like its input chunk, i.e.
+                # the UDF acted as a transform. pandas restores the original
+                # row order in this case (GroupBy._concat_objects) regardless
+                # of ``sort``. The concatenated chunks are in key-sorted
+                # group order, so gather back through the inverse of the
+                # grouping permutation.
+                _, _, (positions,) = self._groups(
+                    [self._range_column_from_obj]
+                )
+                result = result.take(positions.argsort().values)
         return result
 
     @_performance_tracking
@@ -2560,8 +2591,8 @@ def apply(
           where possible and will fall back to the iterative algorithm if
           necessary.
         include_groups : bool, default False
-            When True, will attempt to apply ``func`` to the groupings in
-            the case that they are columns of the DataFrame.
+            Only ``False`` is accepted (matching pandas 3.0, where
+            ``include_groups=True`` raises a ``ValueError``).
         kwargs : dict
             Optional keyword arguments to pass to the function.
             Currently not supported
@@ -2641,6 +2672,9 @@ def mult(df):
         dtype: int64
 
         """
+        if include_groups:
+            # matches pandas 3.0
+            raise ValueError("include_groups=True is no longer allowed.")
         if kwargs:
             raise NotImplementedError(
                 "Passing kwargs to func is currently not supported."
@@ -2721,8 +2755,10 @@ def mult(df):
         else:
             raise ValueError(f"Unsupported engine '{engine}'")
 
-        if self._sort:
-            result = result.sort_index()
+        # No final sort: group-keyed results are already produced in
+        # sorted group-key order, and pandas preserves the UDF's
+        # within-group row order (and a transform's original row order)
+        # regardless of ``sort`` (pandas GH52444).
         if self._as_index is False:
             result = result.reset_index()
         return result

python/cudf/cudf/core/indexed_frame.py

@@ -4842,11 +4842,24 @@ def _reset_index(
             new_column_data[name] = col
         # This is to match pandas where the new data columns are always
         # inserted to the left of existing data columns.
+        label_dtype = None
+        if not self._data.multiindex:
+            # pandas computes the result columns by Index.insert into the
+            # existing columns Index, which preserves its dtype (e.g.
+            # Index([None], dtype=object).insert(0, "a") stays object);
+            # rebuilding from the merged labels would re-infer (pandas 3.0
+            # infers "str" for all-string labels). Emulate the insert
+            # provenance.
+            pd_columns = self._data.to_pandas_index
+            for new_name in reversed(list(new_column_data)):
+                pd_columns = pd_columns.insert(0, new_name)
+            label_dtype = pd_columns.dtype
         return (
             ColumnAccessor(
                 {**new_column_data, **self._data},
                 self._data.multiindex,
                 self._data._level_names,
+                label_dtype=label_dtype,
             ),
             index,
         )

python/cudf/cudf/core/series.py

@@ -1093,16 +1093,22 @@ def reset_index(
             raise TypeError(
                 "Cannot reset_index inplace on a Series to create a DataFrame"
             )
-        data, index = self._reset_index(
-            level=level, drop=drop, allow_duplicates=allow_duplicates
-        )
         if not drop:
+            # pandas semantics are ``self.to_frame(name).reset_index()``:
+            # resolve ``name`` first so the columns-dtype provenance in
+            # ``_reset_index`` sees the final value-column label.
             if name is no_default:
                 name = 0 if self.name is None else self.name
-            data[name] = data.pop(self.name)
+            frame = self._to_frame(name, index=self.index)
+            data, index = frame._reset_index(
+                level=level, drop=drop, allow_duplicates=allow_duplicates
+            )
             return self._constructor_expanddim._from_data(
                 data, index, attrs=self.attrs
             )
+        data, index = self._reset_index(
+            level=level, drop=drop, allow_duplicates=allow_duplicates
+        )
         # For ``name`` behavior, see:
         # https://github.com/pandas-dev/pandas/issues/44575
         # ``name`` has to be ignored when `drop=True`