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 0710f871e815 [SPARK-58553][PS][FOLLOWUP] Match fmax/fmin signed-zero
tie-break to installed NumPy
0710f871e815 is described below
commit 0710f871e815e2d7faa72cd5203a2fe3c5f8a3f8
Author: Spenser Sun <[email protected]>
AuthorDate: Thu Aug 13 15:35:21 2026 +0800
[SPARK-58553][PS][FOLLOWUP] Match fmax/fmin signed-zero tie-break to
installed NumPy
### What changes were proposed in this pull request?
SPARK-58553 replaced the `pandas_udf`-based `np.fmax` / `np.fmin`
implementations in the pandas API on Spark with native `F.greatest` / `F.least`
expressions. When the two operands are equal (for example `+0.0` and `-0.0`),
the native mapping breaks the tie by returning the **first** operand.
NumPy changed this signed-zero tie-break at **2.3.0**: `>= 2.3.0` returns
the first operand, while older versions return the **second**. The native
mapping therefore matches NumPy `>= 2.3.0` but disagrees with older versions on
the sign of a `±0.0` result.
This PR selects the tie operand based on the installed NumPy version so the
result matches `np.fmax` / `np.fmin` on that NumPy: return the first operand on
`>= 2.3.0`, the second on older versions. The implementation stays fully native
(`F.greatest` / `F.least`); only which operand is returned on a signed-zero tie
differs by version.
**Why not restore the original `pandas_udf` fallback for old NumPy?** The
original UDF matched the installed NumPy automatically (it calls `np.fmax` in a
Python worker), so restoring it for `< 2.3.0` would also be correct. But that
reintroduces the per-batch JVM <-> Python round trip that SPARK-58553 removed,
losing the performance and optimizer benefits for those users. Since the
signed-zero tie is the **only** cross-version difference (verified exhaustively
over every combination of [...]
**Scope:** `fmax` / `fmin` is the only affected function. The full
`test_numpy_compat.py` suite (18 tests, including the generic mapping sweeps
and every other SPARK-58532 conversion — `fmod`, `ldexp`, `heaviside`,
`reciprocal`, `float_power`, bitwise shifts, `signbit`, etc.) passes on the
minimum dependencies; the signed-zero tie in `fmax` / `fmin` is the only
version-sensitive behavior.
### Why are the changes needed?
The scheduled "Build / Python-only (Minimum dependencies of PySpark)" build
(NumPy 1.23.2) fails `pyspark.pandas.tests.test_numpy_compat
NumPyCompatTests.test_np_fmax_fmin`. The test asserts the sign bit of the
result via `np.signbit`, and on the two `±0.0` tie rows the native mapping
(first operand) disagrees with the reference computed from the installed NumPy
(second operand on 1.23.2). Regular CI runs a newer NumPy (`>= 2.3.0`), where
the native choice matches, which is why the or [...]
### Does this PR introduce _any_ user-facing change?
No. There is no change relative to any released Spark version (the released
implementation used the `pandas_udf`, which already matched the installed
NumPy). This aligns the unreleased native implementation from SPARK-58553 with
`np.fmax` / `np.fmin` on NumPy `< 2.3.0`. The numeric value is unchanged in all
cases (`+0.0` and `-0.0` are numerically equal); only the sign bit of a zero
result on a `±0.0` tie is corrected to match the installed NumPy.
### How was this patch tested?
-
`pyspark.pandas.tests.test_numpy_compat.NumPyCompatTests.test_np_fmax_fmin` and
the full `NumPyCompatTests` suite (18 tests) pass on NumPy 2.4.1 and in a
minimum-dependency environment (NumPy 1.23.2, pandas 2.2.0, pyarrow 18.0.0)
that reproduces the scheduled build.
- Confirmed the failure reproduces on NumPy 1.23.2 without this change and
is resolved with it.
- Verified across installed NumPy wheels (1.23.2 through 2.4.1) that the
signed-zero `±0.0` tie is the only `fmax` / `fmin` behavior that differs
between versions, and that the tie-break flips at exactly 2.3.0.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 4.8)
Closes #57978 from Spenserrrr/numpy-fmax-fmin-version-gate.
Authored-by: Spenser Sun <[email protected]>
Signed-off-by: Ruifeng Zheng <[email protected]>
(cherry picked from commit 614deed46805d0cb1cb6a415b7b6acc86ac5fbc3)
Signed-off-by: Ruifeng Zheng <[email protected]>
---
python/pyspark/pandas/numpy_compat.py | 32 ++++++++++++++++++++++++++------
1 file changed, 26 insertions(+), 6 deletions(-)
diff --git a/python/pyspark/pandas/numpy_compat.py
b/python/pyspark/pandas/numpy_compat.py
index 6611abb3b44c..765ba4453444 100644
--- a/python/pyspark/pandas/numpy_compat.py
+++ b/python/pyspark/pandas/numpy_compat.py
@@ -18,6 +18,7 @@ from typing import Any, Callable, no_type_check
import numpy as np
+from pyspark.loose_version import LooseVersion
from pyspark.sql import Column, functions as F
from pyspark.sql.pandas.functions import pandas_udf
from pyspark.sql.types import DoubleType, BooleanType
@@ -203,6 +204,29 @@ def _floor_divide_func(c1: Column, c2: Column) -> Column:
)
+# NumPy 2.3.0 changed how fmax/fmin break a signed-zero tie: for equal operands
+# (for example +0.0 and -0.0) it returns the first operand, while older
versions
+# returned the second. Track the installed NumPy so the result keeps the
matching
+# sign of zero.
+_tie_returns_first_operand = LooseVersion(np.__version__) >=
LooseVersion("2.3.0")
+
+
+def _fmax_func(c1: Column, c2: Column) -> Column:
+ tie = c1 if _tie_returns_first_operand else c2
+ return (
+ F.when(F.isnan(c1.cast("double")), c2)
+ .when(F.isnan(c2.cast("double")), c1)
+ .when(c1 == c2, tie)
+ .otherwise(F.greatest(c1, c2))
+ .cast("double")
+ )
+
+
+def _fmin_func(c1: Column, c2: Column) -> Column:
+ tie = c1 if _tie_returns_first_operand else c2
+ return F.when(c1 == c2, tie).otherwise(F.least(c1, c2)).cast("double")
+
+
binary_np_spark_mappings = {
"arctan2": F.atan2,
"bitwise_and": lambda c1, c2: c1.bitwiseAND(c2),
@@ -215,12 +239,8 @@ binary_np_spark_mappings = {
# np.floor_divide dispatches to the pandas-on-Spark floordiv dunder
operation
# before this registry is consulted, so this mapping is not used for that
case.
"floor_divide": _floor_divide_func,
- "fmax": lambda c1, c2: F.when(F.isnan(c1.cast("double")), c2)
- .when(F.isnan(c2.cast("double")), c1)
- .when(c1 == c2, c1)
- .otherwise(F.greatest(c1, c2))
- .cast("double"),
- "fmin": lambda c1, c2: F.when(c1 == c2, c1).otherwise(F.least(c1,
c2)).cast("double"),
+ "fmax": _fmax_func,
+ "fmin": _fmin_func,
"fmod": _fmod_func,
"gcd": pandas_udf(lambda s1, s2: np.gcd(s1, s2), DoubleType()), # type:
ignore[call-overload]
"heaviside": lambda c1, c2: F.when(
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
