BUG: Reverse operators on integer-NA series and numpy scalars resulti…

…ng in object dtype (pandas-dev#22024)

pandas/core/ops.py

@@ -1088,35 +1088,43 @@ def dispatch_to_extension_op(op, left, right):
     # we need to listify to avoid ndarray, or non-same-type extension array
     # dispatching
 
+    new_type, left_type, right_type = None, None, None
     if is_extension_array_dtype(left):
-
+        left_type = left.dtype.type
         new_left = left.values
         if isinstance(right, np.ndarray):
 
             # handle numpy scalars, this is a PITA
             # TODO(jreback)
             new_right = lib.item_from_zerodim(right)
+            right_type = new_right.dtype
             if is_scalar(new_right):
                 new_right = [new_right]
             new_right = list(new_right)
         elif is_extension_array_dtype(right) and type(left) != type(right):
+            right_type = new_right.dtype.type
             new_right = list(new_right)
         else:
             new_right = right
-
+            right_type = type(right)
     else:
 
         new_left = list(left.values)
         new_right = right
 
     res_values = op(new_left, new_right)
     res_name = get_op_result_name(left, right)
-
+    if right_type and left_type:
+        new_type = find_common_type([right_type, left_type])
     if op.__name__ == 'divmod':
         return _construct_divmod_result(
             left, res_values, left.index, res_name)
 
-    return _construct_result(left, res_values, left.index, res_name)
+    result = _construct_result(left, res_values, left.index, res_name)
+    if result.dtype == "object":
+        result = _construct_result(left, res_values, left.index, res_name,
+                                   new_type)
+    return result
 
 
 def _arith_method_SERIES(cls, op, special):
@@ -1143,7 +1151,6 @@ def na_op(x, y):
                 result[mask] = op(x[mask], com.values_from_object(y[mask]))
             else:
                 assert isinstance(x, np.ndarray)
-                assert is_scalar(y)
                 result = np.empty(len(x), dtype=x.dtype)
                 mask = notna(x)
                 result[mask] = op(x[mask], y)
@@ -1190,7 +1197,6 @@ def wrapper(left, right):
 
         elif (is_extension_array_dtype(left) or
                 is_extension_array_dtype(right)):
-            # TODO: should this include `not is_scalar(right)`?
             return dispatch_to_extension_op(op, left, right)
 
         elif is_datetime64_dtype(left) or is_datetime64tz_dtype(left):
@@ -1280,11 +1286,13 @@ def na_op(x, y):
         # should have guarantess on what x, y can be type-wise
         # Extension Dtypes are not called here
 
-        # Checking that cases that were once handled here are no longer
-        # reachable.
-        assert not (is_categorical_dtype(y) and not is_scalar(y))
+        # dispatch to the categorical if we have a categorical
+        # in either operand
+        if is_categorical_dtype(y) and not is_scalar(y):
+            # The `not is_scalar(y)` check excludes the string "category"
+            return op(y, x)
 
-        if is_object_dtype(x.dtype):
+        elif is_object_dtype(x.dtype):
             result = _comp_method_OBJECT_ARRAY(op, x, y)
 
         elif is_datetimelike_v_numeric(x, y):
@@ -1342,7 +1350,7 @@ def wrapper(self, other, axis=None):
             return self._constructor(res_values, index=self.index,
                                      name=res_name)
 
-        elif is_datetime64_dtype(self) or is_datetime64tz_dtype(self):
+        if is_datetime64_dtype(self) or is_datetime64tz_dtype(self):
             # Dispatch to DatetimeIndex to ensure identical
             # Series/Index behavior
             if (isinstance(other, datetime.date) and
@@ -1384,9 +1392,8 @@ def wrapper(self, other, axis=None):
                                      name=res_name)
 
         elif (is_extension_array_dtype(self) or
-              (is_extension_array_dtype(other) and not is_scalar(other))):
-            # Note: the `not is_scalar(other)` condition rules out
-            # e.g. other == "category"
+              (is_extension_array_dtype(other) and
+               not is_scalar(other))):
             return dispatch_to_extension_op(op, self, other)
 
         elif isinstance(other, ABCSeries):
@@ -1409,6 +1416,13 @@ def wrapper(self, other, axis=None):
             # is not.
             return result.__finalize__(self).rename(res_name)
 
+        elif isinstance(other, pd.Categorical):
+            # ordering of checks matters; by this point we know
+            # that not is_categorical_dtype(self)
+            res_values = op(self.values, other)
+            return self._constructor(res_values, index=self.index,
+                                     name=res_name)
+
         elif is_scalar(other) and isna(other):
             # numpy does not like comparisons vs None
             if op is operator.ne:
@@ -1538,41 +1552,6 @@ def flex_wrapper(self, other, level=None, fill_value=None, axis=0):
 # -----------------------------------------------------------------------------
 # DataFrame
 
-def dispatch_to_series(left, right, func):
-    """
-    Evaluate the frame operation func(left, right) by evaluating
-    column-by-column, dispatching to the Series implementation.
-
-    Parameters
-    ----------
-    left : DataFrame
-    right : scalar or DataFrame
-    func : arithmetic or comparison operator
-
-    Returns
-    -------
-    DataFrame
-    """
-    # Note: we use iloc to access columns for compat with cases
-    #       with non-unique columns.
-    if lib.is_scalar(right):
-        new_data = {i: func(left.iloc[:, i], right)
-                    for i in range(len(left.columns))}
-    elif isinstance(right, ABCDataFrame):
-        assert right._indexed_same(left)
-        new_data = {i: func(left.iloc[:, i], right.iloc[:, i])
-                    for i in range(len(left.columns))}
-    else:
-        # Remaining cases have less-obvious dispatch rules
-        raise NotImplementedError
-
-    result = left._constructor(new_data, index=left.index, copy=False)
-    # Pin columns instead of passing to constructor for compat with
-    # non-unique columns case
-    result.columns = left.columns
-    return result
-
-
 def _combine_series_frame(self, other, func, fill_value=None, axis=None,
                           level=None, try_cast=True):
     """