fix(spark): correct mod ANSI zero divisor and negative zero handling

In ANSI mode a floating-point zero divisor quietly returned NaN because
Arrow's rem only reports division by zero for integer and decimal types,
and a -0.0 divisor went unrecognised in both modes because Arrow's
floating-point comparisons order totally, so -0.0 is distinct from 0.0.

try_rem now detects zero divisors itself via an is_zero helper that
counts -0.0 as zero, raises for a zero divisor of any numeric type in
ANSI mode (masked by the validity of the dividend, since Spark's
remainder is null intolerant), and substitutes NULL for zero divisors
before calling Arrow's rem so the kernel never sees one.

Closes #23894.

Author: u70b3

datafusion/spark/src/function/math/modulus.rs

@@ -15,41 +15,77 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use arrow::array::{Scalar, new_null_array};
+use arrow::array::{ArrayRef, BooleanArray, Scalar, new_null_array};
 use arrow::compute::kernels::numeric::add;
 use arrow::compute::kernels::{
+    boolean::{and, is_not_null, or},
     cmp::{eq, lt},
-    numeric::rem,
+    numeric::{neg, rem},
     zip::zip,
 };
 use arrow::datatypes::DataType;
+use arrow::error::ArrowError;
 use datafusion_common::{Result, ScalarValue, assert_eq_or_internal_err};
 use datafusion_expr::{
     ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl, Signature, Volatility,
 };
 
+/// Returns a one element array holding negative zero, for the floating point
+/// types only.
+///
+/// Arrow's comparison kernels order floating point values totally, so `-0.0`
+/// compares as distinct from, and less than, `0.0`. Java, and therefore Spark,
+/// treats `-0.0` as equal to zero. The helper below uses this to restore the
+/// IEEE 754 answer.
+fn negative_zero(data_type: &DataType) -> Result<Option<ArrayRef>> {
+    match data_type {
+        DataType::Float16 | DataType::Float32 | DataType::Float64 => {
+            let zero = ScalarValue::new_zero(data_type)?.to_array()?;
+            Ok(Some(neg(zero.as_ref())?))
+        }
+        _ => Ok(None),
+    }
+}
+
+/// Rows of `values` that equal zero, counting `-0.0` as zero.
+fn is_zero(values: &ArrayRef) -> Result<BooleanArray> {
+    let zero = ScalarValue::new_zero(values.data_type())?.to_array()?;
+    let mask = eq(values, &Scalar::new(zero))?;
+    match negative_zero(values.data_type())? {
+        Some(negative_zero) => Ok(or(&mask, &eq(values, &Scalar::new(negative_zero))?)?),
+        None => Ok(mask),
+    }
+}
+
 /// Computes `rem(left, right)` with divide-by-zero handling.
-/// In ANSI mode, any zero divisor causes an error.
-/// In legacy mode (ANSI off), zero divisors are replaced with NULL before
-/// computing the remainder, so those positions return NULL while others
-/// compute normally.
+/// In ANSI mode, a zero divisor of any numeric type causes an error, with
+/// `-0.0` counting as zero; a row whose dividend is NULL never raises, to
+/// match Spark's null-intolerant remainder. In legacy mode (ANSI off), zero
+/// divisors are replaced with NULL before computing the remainder, so those
+/// positions return NULL while others compute normally.
 fn try_rem(
-    left: &arrow::array::ArrayRef,
-    right: &arrow::array::ArrayRef,
+    left: &ArrayRef,
+    right: &ArrayRef,
     enable_ansi_mode: bool,
-) -> Result<arrow::array::ArrayRef> {
+) -> Result<ArrayRef> {
+    let divisor_is_zero = is_zero(right)?;
+    // Null out zero divisors so that the remainder kernels never see one:
+    // division by zero then returns NULL instead of erroring (integers) or
+    // returning NaN (floats). ANSI mode reports the error itself below, so
+    // this substitution is harmless on rows that must raise.
+    let null = Scalar::new(new_null_array(right.data_type(), 1));
+    let safe_right = zip(&divisor_is_zero, &null, right)?;
     if enable_ansi_mode {
-        Ok(rem(left, right)?)
-    } else {
-        // In legacy mode, null out zero divisors so that division by zero
-        // returns NULL instead of erroring (integers) or returning NaN (floats).
-        let zero = ScalarValue::new_zero(right.data_type())?.to_array()?;
-        let zero = Scalar::new(zero);
-        let null = Scalar::new(new_null_array(right.data_type(), 1));
-        let is_zero = eq(right, &zero)?;
-        let safe_right = zip(&is_zero, &null, right)?;
-        Ok(rem(left, &safe_right)?)
+        // Spark's remainder expressions are null intolerant, so a row whose
+        // dividend is NULL evaluates to NULL and never raises, even when the
+        // divisor on that row is zero. Mask the check by the validity of the
+        // dividend to match.
+        let raises = and(&divisor_is_zero, &is_not_null(left.as_ref())?)?;
+        if raises.iter().flatten().any(|raises| raises) {
+            return Err(ArrowError::DivideByZero.into());
+        }
     }
+    Ok(rem(left, &safe_right)?)
 }
 
 /// Spark-compatible `mod` function
@@ -418,6 +454,101 @@ mod test {
         assert!(result.is_err());
     }
 
+    #[test]
+    fn test_mod_zero_division_ansi_float() {
+        // In ANSI mode a zero divisor of any numeric type must raise,
+        // including floating point, where Arrow's `rem` follows IEEE 754
+        // and quietly returns NaN (#23894)
+        let left = Float64Array::from(vec![Some(10.5), Some(7.2)]);
+        let right = Float64Array::from(vec![Some(0.0), Some(2.0)]);
+
+        let left_value = ColumnarValue::Array(Arc::new(left));
+        let right_value = ColumnarValue::Array(Arc::new(right));
+
+        let result = spark_mod(&[left_value, right_value], true);
+        assert!(result.is_err());
+    }
+
+    #[test]
+    fn test_mod_negative_zero_divisor_legacy() {
+        // `-0.0` counts as a zero divisor, so it returns NULL in legacy
+        // mode rather than NaN (#23894)
+        let left = Float64Array::from(vec![Some(10.5), Some(7.5)]);
+        let right = Float64Array::from(vec![Some(-0.0), Some(2.0)]);
+
+        let left_value = ColumnarValue::Array(Arc::new(left));
+        let right_value = ColumnarValue::Array(Arc::new(right));
+
+        let result = spark_mod(&[left_value, right_value], false).unwrap();
+
+        if let ColumnarValue::Array(result_array) = result {
+            let result_float64 = result_array
+                .as_any()
+                .downcast_ref::<Float64Array>()
+                .unwrap();
+            assert!(result_float64.is_null(0)); // 10.5 % -0.0 = NULL
+            assert_eq!(result_float64.value(1), 1.5); // 7.5 % 2.0 = 1.5
+        } else {
+            panic!("Expected array result");
+        }
+    }
+
+    #[test]
+    fn test_mod_negative_zero_divisor_ansi() {
+        // `-0.0` counts as a zero divisor, so it raises in ANSI mode (#23894)
+        let left = Float64Array::from(vec![Some(10.5)]);
+        let right = Float64Array::from(vec![Some(-0.0)]);
+
+        let left_value = ColumnarValue::Array(Arc::new(left));
+        let right_value = ColumnarValue::Array(Arc::new(right));
+
+        let result = spark_mod(&[left_value, right_value], true);
+        assert!(result.is_err());
+    }
+
+    #[test]
+    fn test_mod_zero_division_ansi_null_dividend() {
+        // Spark's remainder expressions are null intolerant: a NULL dividend
+        // short-circuits to NULL before the divisor is validated, so a zero
+        // divisor on such a row must not raise, even in ANSI mode (#23894)
+        let left = Int32Array::from(vec![None, Some(10)]);
+        let right = Int32Array::from(vec![Some(0), Some(3)]);
+
+        let left_value = ColumnarValue::Array(Arc::new(left));
+        let right_value = ColumnarValue::Array(Arc::new(right));
+
+        let result = spark_mod(&[left_value, right_value], true).unwrap();
+
+        if let ColumnarValue::Array(result_array) = result {
+            let result_int32 =
+                result_array.as_any().downcast_ref::<Int32Array>().unwrap();
+            assert!(result_int32.is_null(0)); // NULL % 0 = NULL (no error)
+            assert_eq!(result_int32.value(1), 1); // 10 % 3 = 1
+        } else {
+            panic!("Expected array result");
+        }
+
+        // Same for floating point
+        let left = Float64Array::from(vec![None, Some(10.5)]);
+        let right = Float64Array::from(vec![Some(0.0), Some(2.0)]);
+
+        let left_value = ColumnarValue::Array(Arc::new(left));
+        let right_value = ColumnarValue::Array(Arc::new(right));
+
+        let result = spark_mod(&[left_value, right_value], true).unwrap();
+
+        if let ColumnarValue::Array(result_array) = result {
+            let result_float64 = result_array
+                .as_any()
+                .downcast_ref::<Float64Array>()
+                .unwrap();
+            assert!(result_float64.is_null(0)); // NULL % 0.0 = NULL (no error)
+            assert!((result_float64.value(1) - 0.5).abs() < f64::EPSILON); // 10.5 % 2.0 = 0.5
+        } else {
+            panic!("Expected array result");
+        }
+    }
+
     // PMOD tests
     #[test]
     fn test_pmod_int32() {

datafusion/sqllogictest/test_files/spark/math/mod.slt

@@ -160,6 +160,18 @@ SELECT MOD(10.5::float8, 0.0::float8) as mod_div_zero_float;
 ----
 NULL
 
+# A negative zero divisor counts as zero and returns NULL in legacy mode
+query R
+SELECT MOD(10.5::float8, -0.0::float8) as mod_div_zero_neg_float;
+----
+NULL
+
+# A NULL dividend evaluates to NULL regardless of the divisor
+query I
+SELECT MOD(NULL::int, 0::int) as mod_null_dividend_legacy;
+----
+NULL
+
 # Division by zero errors in ANSI mode
 statement ok
 set datafusion.execution.enable_ansi_mode = true;
@@ -170,6 +182,25 @@ SELECT MOD(10::int, 0::int);
 statement error DataFusion error: Arrow error: Divide by zero error
 SELECT MOD(-7::int, 0::int);
 
+# A zero divisor of any numeric type raises, including floating point
+statement error DataFusion error: Arrow error: Divide by zero error
+SELECT MOD(10.5::float8, 0.0::float8);
+
+# A negative zero divisor counts as zero and raises
+statement error DataFusion error: Arrow error: Divide by zero error
+SELECT MOD(10.5::float8, -0.0::float8);
+
+# A NULL dividend short-circuits to NULL before the divisor is validated
+query I
+SELECT MOD(NULL::int, 0::int) as mod_null_dividend_ansi;
+----
+NULL
+
+query R
+SELECT MOD(NULL::float8, 0.0::float8) as mod_null_dividend_ansi_float;
+----
+NULL
+
 statement ok
 set datafusion.execution.enable_ansi_mode = false;