From 6fa3035176104d3978cdb73ece9a71600bd88aad Mon Sep 17 00:00:00 2001
From: Yicong-Huang <17627829+Yicong-Huang@users.noreply.github.com>
Date: Fri, 21 Aug 2026 20:28:32 +0000
Subject: [PATCH] fix: match numpy reciprocal parity for decimal, boolean, and
 narrower integer columns

---
 python/pyspark/pandas/numpy_compat.py         | 30 ++++++++++++-------
 .../pyspark/pandas/tests/test_numpy_compat.py | 26 ++++++++++++++++
 2 files changed, 46 insertions(+), 10 deletions(-)

diff --git a/python/pyspark/pandas/numpy_compat.py b/python/pyspark/pandas/numpy_compat.py
index 33ec402a20128..05dbc7c1c7b44 100644
--- a/python/pyspark/pandas/numpy_compat.py
+++ b/python/pyspark/pandas/numpy_compat.py
@@ -70,21 +70,31 @@
     "rad2deg": F.degrees,
     "radians": F.radians,
     "reciprocal": lambda c: F.when(
-        F.typeof(c).isin("float", "double"),
+        # Floating-point and decimal inputs take a true reciprocal; numpy
+        # applies it element-wise to Decimal objects as well.
+        F.typeof(c).isin("float", "double") | F.typeof(c).startswith("decimal"),
         F.when(c.isNull(), c.cast("double"))
         .when(
-            c == 0,
+            # Cast to double so the zero check also analyzes for the integer and
+            # boolean columns that fall through to the otherwise branch (Spark
+            # type-checks every branch of the CASE, not just the taken one).
+            c.cast("double") == 0,
             F.when(c.cast("string") == "-0.0", F.lit(float("-inf"))).otherwise(F.lit(float("inf"))),
         )
-        .otherwise(F.lit(1.0) / c),
+        .otherwise(F.lit(1.0) / c.cast("double")),
     ).otherwise(
-        # Integer input: numpy does integer division (truncated toward zero),
-        # so casting the float quotient to long reproduces 1 -> 1, -1 -> -1,
-        # and every other magnitude -> 0. Dividing by 0 overflows to the int64
-        # minimum, matching numpy's behavior on integer arrays.
-        F.when(c == 0, F.lit(float(np.iinfo(np.int64).min))).otherwise(
-            (F.lit(1) / c).cast("long").cast("double")
-        )
+        # Integer and boolean inputs: numpy does integer division (truncated
+        # toward zero), so only +/-1 survive and every other magnitude -> 0.
+        # Dividing by 0 overflows to the width-specific integer minimum for int
+        # (int32) and bigint (int64), while narrower widths (tinyint, smallint,
+        # and boolean promoted to int8) return 0. Cast through long so boolean
+        # and narrower integers can take part in the division.
+        F.when(
+            c.cast("long") == 0,
+            F.when(F.typeof(c) == "int", F.lit(float(np.iinfo(np.int32).min)))
+            .when(F.typeof(c) == "bigint", F.lit(float(np.iinfo(np.int64).min)))
+            .otherwise(F.lit(0.0)),
+        ).otherwise((F.lit(1) / c.cast("long")).cast("long").cast("double"))
     ),
     "rint": lambda c: F.rint(c.cast("double")),
     "sign": F.signum,
diff --git a/python/pyspark/pandas/tests/test_numpy_compat.py b/python/pyspark/pandas/tests/test_numpy_compat.py
index a3f92ef720583..b211d5d91aaa9 100644
--- a/python/pyspark/pandas/tests/test_numpy_compat.py
+++ b/python/pyspark/pandas/tests/test_numpy_compat.py
@@ -17,6 +17,7 @@
 
 import platform
 import unittest
+from decimal import Decimal
 
 import numpy as np
 import pandas as pd
@@ -180,6 +181,31 @@ def test_np_reciprocal_integer(self):
 
                 self.assert_eq(np.reciprocal(psdf.a), np.reciprocal(pdf.a), almost=True)
 
+    @_skip_if_numpy_differs
+    def test_np_reciprocal_non_default_dtypes(self):
+        # The non-floating reciprocal branch also serves narrower integers,
+        # booleans, and decimals. numpy divides integers (and booleans, as
+        # int8) toward zero, so 0 overflows to the width-specific minimum
+        # (0 for int8/int16, int32 min for int32), while decimals take a true
+        # floating reciprocal. Lock in parity with pandas for each.
+        for dtype in ("int8", "int16", "int32"):
+            with self.subTest(dtype=dtype):
+                pdf = pd.DataFrame({"a": np.array([-2, -1, 0, 1, 2], dtype=dtype)})
+                psdf = ps.from_pandas(pdf)
+
+                self.assert_eq(np.reciprocal(psdf.a), np.reciprocal(pdf.a), almost=True)
+
+        # Boolean: numpy promotes to int8 (True -> 1, False -> 0).
+        pdf = pd.DataFrame({"a": [True, False, True]})
+        psdf = ps.from_pandas(pdf)
+        self.assert_eq(np.reciprocal(psdf.a), np.reciprocal(pdf.a), almost=True)
+
+        # Decimal: numpy takes a floating reciprocal. 0 is excluded because
+        # numpy raises DivisionByZero on Decimal('0').
+        pdf = pd.DataFrame({"a": [Decimal("2.5"), Decimal("-4"), Decimal("0.5")]})
+        psdf = ps.from_pandas(pdf)
+        self.assert_eq(np.reciprocal(psdf.a), np.reciprocal(pdf.a), almost=True)
+
     def test_np_bitwise_shift_functions(self):
         pdf = pd.DataFrame(
             {