(spark) branch master updated: [SPARK-58553][PS][FOLLOWUP] Match fmax/fmin signed-zero tie-break to installed NumPy

[ru...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

zhengruifeng pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git


The following commit(s) were added to refs/heads/master by this push:
     new 614deed46805 [SPARK-58553][PS][FOLLOWUP] Match fmax/fmin signed-zero tie-break to installed NumPy
614deed46805 is described below

commit 614deed46805d0cb1cb6a415b7b6acc86ac5fbc3
Author: Spenser Sun <hs...@gmail.com>
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 <hs...@gmail.com>
    Signed-off-by: Ruifeng Zheng <ru...@apache.org>
---
 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: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org