[SPARK-59093][PS][TEST][FOLLOWUP] Skip floor division overflow tests on old NumPy

python/pyspark/pandas/tests/data_type_ops/test_num_mul_div.py

@@ -15,11 +15,13 @@
 # limitations under the License.
 #
 
+import unittest
 
 import numpy as np
 import pandas as pd
 
 from pyspark import pandas as ps
+from pyspark.loose_version import LooseVersion
 from pyspark.pandas.tests.data_type_ops.testing_utils import OpsTestBase
 from pyspark.testing.pandasutils import PandasOnSparkTestCase
 from pyspark.testing.utils import is_ansi_mode_test
@@ -158,7 +160,11 @@ def test_floordiv(self):
             np.signbit(pser // 3.0).tolist(), np.signbit((psser // 3.0).to_pandas()).tolist()
         )
 
-        # The only quotient that does not fit in a long, where pandas wraps around.
+    @unittest.skipIf(
+        LooseVersion(np.__version__) < LooseVersion("1.24.0"),
+        "NumPy < 1.24 leaves integer floor division overflow undefined",
+    )
+    def test_floordiv_integer_overflow(self):
         pser = pd.Series([-(2**63)])
         psser = ps.from_pandas(pser)
         self.assert_eq((pser // -1).astype(float), psser // -1)

python/pyspark/pandas/tests/test_numpy_compat.py

@@ -702,23 +702,35 @@ def floor_divided(pdf):
         # low bits: -9007199254740993 // 2 is -4503599627370497, not -4503599627370496.
         pdf = pd.DataFrame(
             {
-                "x1": [9007199254740993, -9007199254740993, 4611686018427387905, 7, -7],
-                "x2": [1, 2, 3, 3, 3],
+                "x1": [9007199254740993, -9007199254740993, 4611686018427387905, 7, -7, -(2**63)],
+                "x2": [1, 2, 3, 3, 3, 2],
             }
         )
         self.assert_eq(floor_divided(pdf), np.floor_divide(pdf.x1, pdf.x2).astype("float64"))
 
-        # The most negative long divided by -1, whose quotient a long cannot hold. NumPy wraps
-        # around, while Spark's integer division raises.
-        pdf = pd.DataFrame({"x1": [-(2**63), -(2**63)], "x2": [-1, 2]})
-        self.assert_eq(floor_divided(pdf), np.floor_divide(pdf.x1, pdf.x2).astype("float64"))
-
         # Finite operands whose quotient overflows to an infinity, which is its own floor.
         pdf = pd.DataFrame(
             {"x1": [1e300, -1e300, 1e300, -1e300], "x2": [1e-300, 1e-300, -1e-300, -1e-300]}
         )
         self.assert_eq(floor_divided(pdf), np.floor_divide(pdf.x1, pdf.x2))
 
+    @unittest.skipIf(
+        LooseVersion(np.__version__) < LooseVersion("1.24.0"),
+        "NumPy < 1.24 leaves integer floor division overflow undefined",
+    )
+    def test_floor_divide_func_integer_overflow(self):
+        from pyspark.pandas.utils import _floor_divide_func
+
+        pdf = pd.DataFrame({"x1": [-(2**63)], "x2": [-1]})
+        psdf = ps.from_pandas(pdf)
+        result = (
+            psdf.spark.frame()
+            .select(_floor_divide_func(F.col("x1"), F.col("x2")).alias("result"))
+            .toPandas()["result"]
+            .rename(None)
+        )
+        self.assert_eq(result, np.floor_divide(pdf.x1, pdf.x2).astype("float64"))
+
     def test_np_logaddexp(self):
         for pdf in (
             pd.DataFrame(

python/pyspark/pandas/utils.py

@@ -1089,7 +1089,7 @@ def _floor_divide_integral(c1: Column, c2: Column) -> Column:
     truncated = F.call_function("div", c1_long, c2_long)
     remainder = F.try_mod(c1_long, c2_long)
     return F.when(
-        # The one quotient a long cannot hold, where NumPy wraps around and `div` would raise.
+        # Match NumPy 1.24+ for the one quotient a long cannot hold; `div` would raise.
         (c1_long == F.lit(-(2**63))) & (c2_long == F.lit(-1)),
         F.lit(float(-(2**63))),
     ).otherwise(