ENH/PERF: enable column-wise reductions for EA-backed columns

pandas/core/frame.py

@@ -8481,6 +8481,22 @@ def _count_level(self, level, axis=0, numeric_only=False):
     def _reduce(
         self, op, name, axis=0, skipna=True, numeric_only=None, filter_type=None, **kwds
     ):
+        """
+        Reduce DataFrame over axis with given operation.
+
+        Parameters
+        ----------
+        op : func
+            The reducing function to be called on the values.
+        name : str
+            The name of the reduction.
+        axis : int
+        numeric_only : bool, optional
+        filter_type : None or "bool"
+            Set to "bool" for ops that only work on boolean values.
+        skipna, **kwds : keywords to pass to the `op` function
+
+        """
 
         assert filter_type is None or filter_type == "bool", filter_type
 
@@ -8531,7 +8547,12 @@ def _get_data(axis_matters):
                 raise NotImplementedError(msg)
             return data
 
-        if numeric_only is not None and axis in [0, 1]:
+        # special case for block-wise
+        if (
+            not self._mgr.any_extension_types
+            and numeric_only is not None
+            and axis in [0, 1]
+        ):
             df = self
             if numeric_only is True:
                 df = _get_data(axis_matters=True)
@@ -8559,6 +8580,54 @@ def blk_func(values):
                 out[:] = coerce_to_dtypes(out.values, df.dtypes)
             return out
 
+        def array_func(values):
+            if isinstance(values, ExtensionArray):
+                return values._reduce(name, skipna=skipna, **kwds)
+            else:
+                return op(values, skipna=skipna, **kwds)
+
+        # all other options with axis=0 are done column-array-wise
+        if axis == 0:
+
+            def _constructor(df, result, index=None):
+                index = index if index is not None else df.columns
+                if len(result):
+                    return df._constructor_sliced(result, index=index)
+                else:
+                    # set correct dtype for empty result
+                    dtype = "bool" if filter_type == "bool" else "float64"
+                    return df._constructor_sliced(result, index=index, dtype=dtype)
+
+            def _reduce_columns(df, op):
+                result = [op(arr) for arr in df._iter_column_arrays()]
+                return _constructor(df, result)
+
+            df = self
+            if numeric_only is True:
+                df = _get_data(axis_matters=True)
+
+            if numeric_only is not None:
+                return _reduce_columns(df, array_func)
+            else:
+                # need to catch and ignore exceptions when numeric_only=None
+                try:
+                    return _reduce_columns(df, array_func)
+                except TypeError:
+                    # if column-wise fails and numeric_only was None, we try
+                    # again but removing those columns for which it fails
+                    result = []
+                    indices = []
+                    for i, arr in enumerate(df._iter_column_arrays()):
+                        try:
+                            res = array_func(arr)
+                        except Exception:
+                            pass
+                        else:
+                            result.append(res)
+                            indices.append(i)
+
+                    return _constructor(df, result, index=df.columns[indices])
+
         if not self._is_homogeneous_type:
             # try to avoid self.values call
 

pandas/core/generic.py

@@ -11397,7 +11397,7 @@ def stat_func(
         if level is not None:
             return self._agg_by_level(name, axis=axis, level=level, skipna=skipna)
         return self._reduce(
-            func, name=name, axis=axis, skipna=skipna, numeric_only=numeric_only
+            func, name=name, axis=axis, skipna=skipna, numeric_only=numeric_only,
         )
 
     return set_function_name(stat_func, name, cls)

pandas/core/internals/managers.py

@@ -327,24 +327,24 @@ def _verify_integrity(self) -> None:
                 f"tot_items: {tot_items}"
             )
 
-    def reduce(self, func, *args, **kwargs):
+    def reduce(self, func):
         # If 2D, we assume that we're operating column-wise
         if self.ndim == 1:
             # we'll be returning a scalar
             blk = self.blocks[0]
-            return func(blk.values, *args, **kwargs)
+            return func(blk.values)
 
         res = {}
         for blk in self.blocks:
-            bres = func(blk.values, *args, **kwargs)
+            bres = func(blk.values)
 
             if np.ndim(bres) == 0:
                 # EA
                 assert blk.shape[0] == 1
                 new_res = zip(blk.mgr_locs.as_array, [bres])
             else:
                 assert bres.ndim == 1, bres.shape
-                assert blk.shape[0] == len(bres), (blk.shape, bres.shape, args, kwargs)
+                assert blk.shape[0] == len(bres), (blk.shape, bres.shape)
                 new_res = zip(blk.mgr_locs.as_array, bres)
 
             nr = dict(new_res)

pandas/tests/frame/test_analytics.py

@@ -410,7 +410,7 @@ def test_stat_operators_attempt_obj_array(self, method):
         for df in [df1, df2]:
             assert df.values.dtype == np.object_
             result = getattr(df, method)(1)
-            expected = getattr(df.astype("f8"), method)(1)
+            expected = getattr(df, method)(1)
 
             if method in ["sum", "prod"]:
                 tm.assert_series_equal(result, expected)