REF: ignore_failures in BlockManager.reduce

pandas/core/frame.py

@@ -8598,6 +8598,8 @@ def _reduce(
             cols = self.columns[~dtype_is_dt]
             self = self[cols]
 
+        any_object = self.dtypes.apply(is_object_dtype).any()
+
         if axis is None and filter_type == "bool":
             labels = None
             constructor = None
@@ -8630,7 +8632,9 @@ def _get_data(axis_matters):
                 raise NotImplementedError(msg)
             return data
 
-        if numeric_only is not None and axis in [0, 1]:
+        if (numeric_only is not None and axis in [0, 1]) or (
+            numeric_only is None and axis == 0 and not any_object
+        ):
             df = self
             if numeric_only is True:
                 df = _get_data(axis_matters=True)
@@ -8639,6 +8643,7 @@ def _get_data(axis_matters):
                 axis = 0
 
             out_dtype = "bool" if filter_type == "bool" else None
+            ignore_failures = numeric_only is None
 
             def blk_func(values):
                 if isinstance(values, ExtensionArray):
@@ -8648,12 +8653,15 @@ def blk_func(values):
 
             # After possibly _get_data and transposing, we are now in the
             #  simple case where we can use BlockManager.reduce
-            res = df._mgr.reduce(blk_func)
-            out = df._constructor(res,).iloc[0].rename(None)
+            res, indexer = df._mgr.reduce(blk_func, ignore_failures=ignore_failures)
+            out = df._constructor(res).iloc[0]
+
             if out_dtype is not None:
                 out = out.astype(out_dtype)
             if axis == 0 and is_object_dtype(out.dtype):
-                out[:] = coerce_to_dtypes(out.values, df.dtypes)
+                # GH#35865 careful to cast explicitly to object
+                nvs = coerce_to_dtypes(out.values, df.dtypes.iloc[np.sort(indexer)])
+                out[:] = np.array(nvs, dtype=object)
             return out
 
         if not self._is_homogeneous_type or self._mgr.any_extension_types:

pandas/core/internals/managers.py

@@ -3,6 +3,7 @@
 import operator
 import re
 from typing import (
+    Callable,
     DefaultDict,
     Dict,
     List,
@@ -330,18 +331,46 @@ def _verify_integrity(self) -> None:
                 f"tot_items: {tot_items}"
             )
 
-    def reduce(self: T, func) -> T:
+    def reduce(
+        self: T, func: Callable, ignore_failures: bool = False
+    ) -> Tuple[T, np.ndarray]:
+        """
+        Apply reduction function blockwise, returning a single-row BlockManager.
+
+        Parameters
+        ----------
+        func : reduction function
+        ignore_failures : bool, default False
+            Whether to drop blocks where func raises TypeError.
+
+        Returns
+        -------
+        BlockManager
+        np.ndarray
+            Indexer of mgr_locs that are retained.
+        """
         # If 2D, we assume that we're operating column-wise
         assert self.ndim == 2
 
-        res_blocks = []
-        for blk in self.blocks:
-            nbs = blk.reduce(func)
+        res_blocks: List[Block] = []
+        for i, blk in enumerate(self.blocks):
+            try:
+                nbs = blk.reduce(func)
+            except TypeError:
+                if ignore_failures:
+                    continue
+                raise
             res_blocks.extend(nbs)
 
-        index = Index([0])  # placeholder
-        new_mgr = BlockManager.from_blocks(res_blocks, [self.items, index])
-        return new_mgr
+        index = Index([None])  # placeholder
+        if res_blocks:
+            indexer = np.concatenate([blk.mgr_locs.as_array for blk in res_blocks])
+            new_mgr = self._combine(res_blocks, copy=False, index=index)
+        else:
+            indexer = []
+            new_mgr = type(self).from_blocks([], [Index([]), index])
+
+        return new_mgr, indexer
 
     def operate_blockwise(self, other: "BlockManager", array_op) -> "BlockManager":
         """
@@ -721,7 +750,9 @@ def get_numeric_data(self, copy: bool = False) -> "BlockManager":
         self._consolidate_inplace()
         return self._combine([b for b in self.blocks if b.is_numeric], copy)
 
-    def _combine(self, blocks: List[Block], copy: bool = True) -> "BlockManager":
+    def _combine(
+        self: T, blocks: List[Block], copy: bool = True, index: Optional[Index] = None
+    ) -> T:
         """ return a new manager with the blocks """
         if len(blocks) == 0:
             return self.make_empty()
@@ -737,6 +768,8 @@ def _combine(self, blocks: List[Block], copy: bool = True) -> "BlockManager":
             new_blocks.append(b)
 
         axes = list(self.axes)
+        if index is not None:
+            axes[-1] = index
         axes[0] = self.items.take(indexer)
 
         return type(self).from_blocks(new_blocks, axes)