This is an automated email from the ASF dual-hosted git repository.
zhengruifeng pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 966e56784638 [SPARK-58581][PS][FOLLOWUP] Fix NumPy floor_divide
rounding and integer precision
966e56784638 is described below
commit 966e567846387bd5b5188a5d56c5ab150e53b463
Author: Spenser Sun <[email protected]>
AuthorDate: Fri Aug 28 15:56:28 2026 +0800
[SPARK-58581][PS][FOLLOWUP] Fix NumPy floor_divide rounding and integer
precision
### What changes were proposed in this pull request?
The native `floor_divide` mapping computes `floor(c1 / c2)`. That gives a
wrong answer in two cases, both fixed here.
- **Floating operands.** `1.0 // 0.1` returned 10, where pandas, NumPy and
Python all give 9. The exact quotient is a little under 10, and the division
rounds it up before the floor ever runs. NumPy works the quotient out from the
remainder instead, which is exact in IEEE arithmetic, so the floating branch
now does the same, including NumPy's sign correction and snap-to-nearest.
- **Integral operands.** They were cast to double before dividing, which
drops the low bits above 2^53, so `-9007199254740993 // 2` came back as
`-4503599627370496` instead of `-4503599627370497`. They now divide in integer
space, and only the result is cast.
A second commit handles `-2**63 // -1`, the one quotient a long cannot
hold. Spark's integer division raises `ARITHMETIC_OVERFLOW` there, so that pair
is guarded and returns NumPy's wrapped value, which is what this mapping's
original pandas UDF produced.
A third commit covers a quotient that overflows to an infinity, which the
old and new formulas both returned as `nan`. NumPy keeps the infinity, and an
infinite quotient is its own floor, so `1e300 // 1e-300` now skips the flooring
step.
### Why are the changes needed?
Both answers disagree with pandas, which pandas-on-Spark is meant to match.
The `1.0 // 0.1` case is also inconsistent inside Spark itself. `1.0 % 0.1`
gives `0.09999999999999995`, and a remainder that large only makes sense if the
quotient is 9: with a quotient of 10 there would be nothing left over.
### Does this PR introduce _any_ user-facing change?
No. `np.floor_divide` goes through the pandas-on-Spark `floordiv` dunder
operation before this registry is consulted, so the mapping is not reached
today, as the comment on the entry notes.
**A follow-up PR will cover the reachable path.** The same two bugs are
live in `data_type_ops/num_ops.py` (`IntegralOps.floordiv`,
`FractionalOps.floordiv` and both `rfloordiv`), which use `F.floor(lc / rc)`:
`ps.Series([1.0]) // 0.1` gives `10.0` where pandas gives `9.0`, and `psser //
2` is off by one above 2^53. That one does change what users see on a
long-standing path, so it gets its own JIRA and PR rather than riding along
here.
### How was this patch tested?
Added three cases to `test_floor_divide_func`: divisors that binary cannot
represent exactly, integral operands above 2^53, and `-2**63 // -1`. They
compare exactly rather than with `almost=True`, which would let an off-by-one
through at these magnitudes.
Ran `pyspark.pandas.tests.test_numpy_compat` and
`pyspark.pandas.tests.connect.test_parity_numpy_compat`; both pass.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 5)
Closes #58306 from Spenserrrr/numpy-floordiv-precision.
Authored-by: Spenser Sun <[email protected]>
Signed-off-by: Ruifeng Zheng <[email protected]>
(cherry picked from commit 670de9da885f22c21a6195de45013ea6f045af4c)
Signed-off-by: Ruifeng Zheng <[email protected]>
---
python/pyspark/pandas/numpy_compat.py | 111 ++++++++++++++++-------
python/pyspark/pandas/tests/test_numpy_compat.py | 50 ++++++++--
2 files changed, 121 insertions(+), 40 deletions(-)
diff --git a/python/pyspark/pandas/numpy_compat.py
b/python/pyspark/pandas/numpy_compat.py
index 865213080d77..aaa7d843cd75 100644
--- a/python/pyspark/pandas/numpy_compat.py
+++ b/python/pyspark/pandas/numpy_compat.py
@@ -187,52 +187,99 @@ def _logaddexp_func(c1: Column, c2: Column, base2: bool =
False) -> Column:
)
+def _floor_divide_floating(c1: Column, c2: Column) -> Column:
+ """Return floor(c1 / c2) for finite non-zero double operands, derived from
the remainder.
+
+ Flooring the quotient is wrong when the division rounds up across an
integer: 1.0 / 0.1
+ rounds to exactly 10.0, so its floor is 10 where NumPy, pandas and Python
return 9. A
+ remainder is exact, so NumPy's npy_divmod derives the quotient from it, as
this does.
+ """
+ remainder = F.try_mod(c1, c2)
+ # The remainder carries the dividend's sign, so this is the truncating
quotient.
+ truncated = (c1 - remainder) / c2
+ # Truncating and flooring differ by one on opposite signs with a remainder
left over.
+ quotient = F.when(
+ (remainder != 0) & ((remainder < 0) != (c2 < 0)), truncated -
F.lit(1.0)
+ ).otherwise(truncated)
+ # The quotient is whole in exact arithmetic, but the division can leave it
a few bits off, so
+ # round it back. F.floor cannot do this: it returns a bigint, which raises
on an infinity.
+ floor = quotient - F.pmod(quotient, F.lit(1.0))
+ return (
+ # An infinite quotient is its own floor, and has to be returned before
the line above
+ # is used, since pmod of an infinity is nan and leaves `floor` nan.
+ F.when(quotient.isin(float("inf"), float("-inf")), quotient)
+ # Flooring goes one too low when the division landed just under the
whole number.
+ .when(quotient - floor > F.lit(0.5), floor + F.lit(1.0))
+ .otherwise(floor)
+ )
+
+
+def _floor_divide_integral(c1: Column, c2: Column) -> Column:
+ """Return floor(c1 / c2) for integral operands, keeping the quotient in
integer space.
+
+ Casting an operand above 2**53 to double drops its low bits, turning
9007199254740993 into
+ 9007199254740992, and Spark's `/` always divides as double. The long casts
are no-ops for
+ the integral types the caller admits; they are there because `div` rejects
a double even in
+ a branch the guard turns off.
+ """
+ c1_long = c1.cast("long")
+ c2_long = c2.cast("long")
+ # `div` is integer division, truncating toward zero, so it needs the same
flooring
+ # correction as the floating helper. Integer arithmetic cannot round, so
nothing more.
+ 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.
+ (c1_long == F.lit(-(2**63))) & (c2_long == F.lit(-1)),
+ F.lit(float(-(2**63))),
+ ).otherwise(
+ F.when((remainder != 0) & ((remainder < 0) != (c2_long < 0)),
truncated - F.lit(1))
+ .otherwise(truncated)
+ .cast("double")
+ )
+
+
def _floor_divide_func(c1: Column, c2: Column) -> Column:
c1_double = c1.cast("double")
c2_double = c2.cast("double")
+ integral_types = ["tinyint", "smallint", "int", "bigint"]
- return F.when(
- F.typeof(c1).isin("float", "double") | F.typeof(c2).isin("float",
"double"),
+ return (
+ # Null, nan and a zero divisor are handled the same way for every
operand type.
F.when(c1.isNull() | F.isnan(c1), c1_double)
.when(c2.isNull() | F.isnan(c2), c2_double)
+ # pandas upcasts a zero divisor instead of raising. A negative zero
divisor negates the
+ # result, and no comparison can see that sign, so the string form is
used. A nullable Int64
+ # returns 0 instead, but arrives as bigint like a default int64, whose
answer this follows.
.when(
- c1_double.isin(float("-inf"), float("inf")),
- F.when(
- c2_double == 0,
- F.when(
- (c1_double < 0) != (c2_double.cast("string") == "-0.0"),
- F.lit(float("-inf")),
- ).otherwise(F.lit(float("inf"))),
- ).otherwise(F.lit(float("nan"))),
+ c2_double == 0,
+ F.when(c1_double == 0, F.lit(float("nan")))
+ .when((c1_double < 0) != (c2_double.cast("string") == "-0.0"),
F.lit(float("-inf")))
+ .otherwise(F.lit(float("inf"))),
)
+ # Integral operands divide as integers, so operands above 2**53 keep
their low bits.
.when(
- c2_double.isin(float("-inf"), float("inf")),
- F.when(c1_double == 0, c1_double / c2_double)
- .when((c1_double < 0) != (c2_double < 0), F.lit(-1.0))
- .otherwise(F.lit(0.0)),
+ F.typeof(c1).isin(integral_types) &
F.typeof(c2).isin(integral_types),
+ _floor_divide_integral(c1, c2),
)
+ # Only floating operands can be infinite or a negative zero, handled
before the division.
.when(
- c2_double == 0,
- F.when(c1_double == 0, F.lit(float("nan")))
+ F.typeof(c1).isin("float", "double") | F.typeof(c2).isin("float",
"double"),
+ # An infinite dividend has no remainder, so NumPy's quotient is
nan for any divisor.
+ F.when(c1_double.isin(float("-inf"), float("inf")),
F.lit(float("nan")))
+ # An infinite divisor gives a quotient between -1 and 1, so the
floor is 0 or -1.
.when(
- (c1_double < 0) != (c2_double.cast("string") == "-0.0"),
- F.lit(float("-inf")),
+ c2_double.isin(float("-inf"), float("inf")),
+ F.when(c1_double == 0, c1_double / c2_double)
+ .when((c1_double < 0) != (c2_double < 0), F.lit(-1.0))
+ .otherwise(F.lit(0.0)),
)
- .otherwise(F.lit(float("inf"))),
- )
- .when(c1_double == 0, c1_double / c2_double)
- .otherwise((c1_double / c2_double) - F.pmod(c1_double / c2_double,
F.lit(1.0))),
- ).otherwise(
- # np.floor_divide on pandas Series returns IEEE values for an integral
zero divisor.
- F.when(c1.isNull() | F.isnan(c1), c1_double)
- .when(c2.isNull() | F.isnan(c2), c2_double)
- .when(
- c2_double == 0,
- F.when(c1_double == 0, F.lit(float("nan")))
- .when(c1_double < 0, F.lit(float("-inf")))
- .otherwise(F.lit(float("inf"))),
+ # Dividing a zero dividend keeps its sign, so -0.0 // 3.0 is -0.0.
+ .when(c1_double == 0, c1_double / c2_double)
+ .otherwise(_floor_divide_floating(c1_double, c2_double)),
)
- .otherwise((c1_double / c2_double) - F.pmod(c1_double / c2_double,
F.lit(1.0)))
+ # A decimal cannot be matched by name: typeof reports its precision,
as in decimal(10,2).
+ .otherwise(_floor_divide_floating(c1_double, c2_double))
)
diff --git a/python/pyspark/pandas/tests/test_numpy_compat.py
b/python/pyspark/pandas/tests/test_numpy_compat.py
index 4b3f24427103..a733155b7199 100644
--- a/python/pyspark/pandas/tests/test_numpy_compat.py
+++ b/python/pyspark/pandas/tests/test_numpy_compat.py
@@ -365,6 +365,15 @@ class NumPyCompatTestsMixin:
def test_floor_divide_func(self):
from pyspark.pandas.numpy_compat import _floor_divide_func
+ def floor_divided(pdf):
+ psdf = ps.from_pandas(pdf)
+ return (
+ psdf.spark.frame()
+ .select(_floor_divide_func(F.col("x1"),
F.col("x2")).alias("result"))
+ .toPandas()["result"]
+ .rename(None)
+ )
+
for pdf in (
pd.DataFrame(
{
@@ -419,14 +428,39 @@ class NumPyCompatTestsMixin:
}
),
):
- 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),
almost=True)
+ self.assert_eq(floor_divided(pdf), np.floor_divide(pdf.x1,
pdf.x2), almost=True)
+
+ # Divisors binary cannot represent exactly, where the quotient rounds
up across an
+ # integer: 1.0 / 0.1 rounds to 10.0, so flooring it gives 10 instead
of 9. Compared
+ # exactly, since almost=True would accept an off-by-one on the large
values below.
+ pdf = pd.DataFrame(
+ {
+ "x1": [1.0, 10.0, 2.0, 0.5, 7.0, -1.0, -10.0, 3.0],
+ "x2": [0.1, 0.1, 0.2, 0.1, 0.7, 0.1, 0.1, 7.0],
+ }
+ )
+ self.assert_eq(floor_divided(pdf), np.floor_divide(pdf.x1, pdf.x2))
+
+ # Integral operands above 2**53, where casting an operand to double
would drop its
+ # low bits: -9007199254740993 // 2 is -4503599627370497, not
-4503599627370496.
+ pdf = pd.DataFrame(
+ {
+ "x1": [9007199254740993, -9007199254740993,
4611686018427387905, 7, -7],
+ "x2": [1, 2, 3, 3, 3],
+ }
+ )
+ 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))
def test_np_logaddexp(self):
for pdf in (
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
