Refactor Frame reductions

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

@@ -172,11 +172,36 @@ def equals(self, other: ColumnBase, check_dtypes: bool = False) -> bool:
     def _null_equals(self, other: ColumnBase) -> ColumnBase:
         return self.binary_operator("NULL_EQUALS", other)
 
-    def all(self) -> bool:
-        return bool(libcudf.reduce.reduce("all", self, dtype=np.bool_))
+    def all(self, skipna: bool = True) -> bool:
+        # If all entries are null the result is True, including when the column
+        # is empty.
+        if self.null_count == self.size:
+            return True
+
+        # We don't want to call _process_for_reduction if skipna is False
+        # because all is not a reduction where the final output is also
+        # nullified by any nulls in the the input.
+        result_col = self._process_for_reduction(True) if skipna else self
+        if isinstance(result_col, ColumnBase):
+            return libcudf.reduce.reduce("all", result_col, dtype=np.bool_)
+        else:
+            return result_col
 
-    def any(self) -> bool:
-        return bool(libcudf.reduce.reduce("any", self, dtype=np.bool_))
+    def any(self, skipna: bool = True) -> bool:
+        # Early exit for fast cases.
+        if not skipna and self.has_nulls:
+            return True
+        elif skipna and self.null_count == self.size:
+            return False
+
+        # We don't want to call _process_for_reduction if skipna is False
+        # because any is not a reduction where the final output is also
+        # nullified by any nulls in the the input.
+        result_col = self._process_for_reduction(True) if skipna else self
+        if isinstance(result_col, ColumnBase):
+            return libcudf.reduce.reduce("any", result_col, dtype=np.bool_)
+        else:
+            return result_col
 
     def __sizeof__(self) -> int:
         n = 0
@@ -911,9 +936,9 @@ def astype(self, dtype: Dtype, **kwargs) -> ColumnBase:
             return self.as_interval_column(dtype, **kwargs)
         elif is_decimal_dtype(dtype):
             return self.as_decimal_column(dtype, **kwargs)
-        elif np.issubdtype(dtype, np.datetime64):
+        elif np.issubdtype(cast(Any, dtype), np.datetime64):
             return self.as_datetime_column(dtype, **kwargs)
-        elif np.issubdtype(dtype, np.timedelta64):
+        elif np.issubdtype(cast(Any, dtype), np.timedelta64):
             return self.as_timedelta_column(dtype, **kwargs)
         else:
             return self.as_numerical_column(dtype, **kwargs)