You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by yi...@apache.org on 2026/08/25 08:24:57 UTC
(spark) branch master updated: [SPARK-58630][PS] Use native Spark expressions for NumPy copysign
This is an automated email from the ASF dual-hosted git repository.
Yicong-Huang 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 06017e301aad [SPARK-58630][PS] Use native Spark expressions for NumPy copysign
06017e301aad is described below
commit 06017e301aaddee53baffbe4d1631a8c440e7594
Author: Yicong Huang <17...@users.noreply.github.com>
AuthorDate: Tue Aug 25 08:24:36 2026 +0000
[SPARK-58630][PS] Use native Spark expressions for NumPy copysign
### What changes were proposed in this pull request?
Replace the `numpy.copysign` pandas UDF mapping in `python/pyspark/pandas/numpy_compat.py` with a native Spark SQL expression (umbrella: SPARK-58532).
`copysign(x, y)` returns `|x|` with the sign of `y`. The sign is taken from `y`'s IEEE-754 sign bit: `-0.0` counts as negative (detected via a string cast, as the existing `reciprocal` mapping does) rather than via `y < 0`, which misses `-0.0`. NaN falls through to a positive sign. Null handling is type-aware: a float `y` column carries missing values as NaN (surfaced as NULL), where `copysign(x, NaN)` is `|x|`; an integer `y` column's NULL is genuine and propagates.
### Why are the changes needed?
Native expressions avoid pandas UDF serialization/worker overhead and work uniformly on classic and Spark Connect.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Added `test_np_copysign` and `test_np_copysign_signed_zero` (also run via the Spark Connect parity suite), covering integer/float/nullable-integer inputs and asserting `np.signbit` for the signed-zero, NaN, null, and `+-inf` edge cases.
### Was this patch authored or co-authored using generative AI tooling?
No.
Closes #57836 from Yicong-Huang/native-numpy-copysign.
Authored-by: Yicong Huang <17...@users.noreply.github.com>
Signed-off-by: Yicong-Huang <17...@users.noreply.github.com>
---
python/pyspark/pandas/numpy_compat.py | 21 +++++++++--
python/pyspark/pandas/tests/test_numpy_compat.py | 44 ++++++++++++++++++++++++
2 files changed, 62 insertions(+), 3 deletions(-)
diff --git a/python/pyspark/pandas/numpy_compat.py b/python/pyspark/pandas/numpy_compat.py
index 55a12aae853d..8f0caea398e4 100644
--- a/python/pyspark/pandas/numpy_compat.py
+++ b/python/pyspark/pandas/numpy_compat.py
@@ -122,6 +122,23 @@ unary_np_spark_mappings = {
}
+def _copysign_func(c1: Column, c2: Column) -> Column:
+ # Sign of y is taken from its IEEE-754 sign bit, so -0.0 counts as negative.
+ # c2 < 0 misses -0.0, so detect it via the string cast, the same way the
+ # 'reciprocal' mapping distinguishes -0.0 from 0.0. NaN's sign bit is positive
+ # and c2 < 0 is already false for NaN, so it correctly falls through to +1.0.
+ sign = F.when((c2 < 0) | (c2.cast("string") == "-0.0"), F.lit(-1.0)).otherwise(F.lit(1.0))
+ # An integer y column's NULL is a genuine missing value and propagates. A
+ # float/double column instead stores its missing value as NaN (surfaced as a
+ # Spark NULL by pandas-on-Spark), for which copysign(x, NaN) returns |x|. A
+ # nullable Float64 <NA> collapses to that same Spark NULL, so it is likewise
+ # treated as NaN and returns |x| rather than propagating; this cannot be
+ # distinguished here and matches the prior pandas_udf behavior.
+ return F.when(
+ c2.isNull() & ~F.typeof(c2).isin("float", "double"), F.lit(None).cast("double")
+ ).otherwise(F.abs(c1.cast("double")) * sign)
+
+
def _fmod_func(c1: Column, c2: Column) -> Column:
c1_double = c1.cast("double")
c2_double = c2.cast("double")
@@ -237,9 +254,7 @@ binary_np_spark_mappings = {
"bitwise_and": lambda c1, c2: c1.bitwiseAND(c2),
"bitwise_or": lambda c1, c2: c1.bitwiseOR(c2),
"bitwise_xor": lambda c1, c2: c1.bitwiseXOR(c2),
- "copysign": pandas_udf( # type: ignore[call-overload]
- lambda s1, s2: np.copysign(s1, s2), DoubleType()
- ),
+ "copysign": _copysign_func,
"float_power": lambda c1, c2: F.pow(c1.cast("double"), c2.cast("double")),
# 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.
diff --git a/python/pyspark/pandas/tests/test_numpy_compat.py b/python/pyspark/pandas/tests/test_numpy_compat.py
index 4e9c082cb2c7..ccb6eb33f4c8 100644
--- a/python/pyspark/pandas/tests/test_numpy_compat.py
+++ b/python/pyspark/pandas/tests/test_numpy_compat.py
@@ -477,6 +477,50 @@ class NumPyCompatTestsMixin:
)
self.assert_eq(np.signbit(result.to_pandas()), expected_signbit)
+ def test_np_copysign(self):
+ for pdf in (
+ pd.DataFrame(
+ {
+ "x1": [-64, -2, -1, 0, 1, 2, 64],
+ "x2": [2, -3, -2, -3, 3, -1, 2],
+ }
+ ),
+ pd.DataFrame(
+ {
+ "x1": [-np.inf, -64.0, -2.0, -0.0, 0.0, 2.0, 64.0, np.inf, np.nan, 1.0],
+ "x2": [2.0, -3.0, -2.0, 0.0, -0.0, -1.0, np.inf, -np.inf, 2.0, np.nan],
+ }
+ ),
+ pd.DataFrame(
+ {
+ "x1": pd.array([1, -2, 3, None, None], dtype="Int64"),
+ "x2": pd.array([-2, 3, None, 2, None], dtype="Int64"),
+ }
+ ),
+ ):
+ psdf = ps.from_pandas(pdf)
+ result = np.copysign(psdf.x1, psdf.x2)
+ expected = np.copysign(pdf.x1, pdf.x2)
+ self.assert_eq(result, expected, almost=True)
+ # copysign only differs from |x| in the sign bit, so assert on signbit
+ # explicitly -- 0.0 == -0.0 numerically and would hide a wrong sign.
+ self.assert_eq(np.signbit(result.to_pandas()), np.signbit(expected))
+
+ def test_np_copysign_signed_zero(self):
+ # np.copysign takes the sign from y's IEEE-754 sign bit, not from y < 0:
+ # copysign(1.0, -0.0) == -1.0 and copysign(1.0, 0.0) == 1.0.
+ pdf = pd.DataFrame(
+ {
+ "x1": [1.0, 1.0, -0.0, -0.0, 3.0],
+ "x2": [0.0, -0.0, 0.0, -0.0, -0.0],
+ }
+ )
+ psdf = ps.from_pandas(pdf)
+ result = np.copysign(psdf.x1, psdf.x2).to_pandas()
+ expected = np.copysign(pdf.x1, pdf.x2)
+ self.assert_eq(result, expected)
+ self.assert_eq(np.signbit(result), np.signbit(expected))
+
def test_np_heaviside(self):
for pdf in (
pd.DataFrame({"x1": [-2, -1, 0, 1, 2], "x2": [-2, -1, 0, 1, 2]}),
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org