diff --git a/datafusion/spark/src/function/math/modulus.rs b/datafusion/spark/src/function/math/modulus.rs index 97f59c2cbb0cb..95cf89926d717 100644 --- a/datafusion/spark/src/function/math/modulus.rs +++ b/datafusion/spark/src/function/math/modulus.rs @@ -15,14 +15,18 @@ // specific language governing permissions and limitations // under the License. -use arrow::array::{Scalar, new_null_array}; -use arrow::compute::kernels::numeric::add; +use std::sync::Arc; + +use arrow::array::{ArrayRef, BooleanArray, Scalar, new_null_array}; +use arrow::compute::cast; use arrow::compute::kernels::{ - cmp::{eq, lt}, - numeric::rem, + boolean::{and, is_not_null, or}, + cmp::{eq, lt, neq}, + numeric::{add_wrapping, 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, @@ -34,10 +38,10 @@ use datafusion_expr::{ /// 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 { +) -> Result { if enable_ansi_mode { Ok(rem(left, right)?) } else { @@ -65,9 +69,58 @@ pub fn spark_mod( Ok(ColumnarValue::Array(result)) } +/// 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 and as not less than zero. The helpers below +/// use this to restore the IEEE 754 answer. +fn negative_zero(data_type: &DataType) -> Result> { + 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 { + 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), + } +} + +/// Rows of `values` that are less than zero, with `-0.0` counting as not +/// negative. +fn is_negative(values: &ArrayRef) -> Result { + let zero = ScalarValue::new_zero(values.data_type())?.to_array()?; + let mask = lt(values, &Scalar::new(zero))?; + match negative_zero(values.data_type())? { + Some(negative_zero) => { + Ok(and(&mask, &neq(values, &Scalar::new(negative_zero))?)?) + } + None => Ok(mask), + } +} + /// Spark-compatible `pmod` function +/// +/// Spark evaluates `val r = a % n; if (r < 0) (r + n) % n else r`, so the +/// adjustment only applies where the plain remainder is negative. +/// /// In ANSI mode, division by zero throws an error. /// In legacy mode, division by zero returns NULL (Spark behavior). +/// +/// `pmod` does not share [`try_rem`] with `mod` because it has to treat `-0.0` +/// as a zero divisor and has to raise on a zero divisor of any numeric type in +/// ANSI mode, including floating point, where Arrow's `rem` follows IEEE 754 +/// and quietly returns `NaN`. pub fn spark_pmod( args: &[ColumnarValue], enable_ansi_mode: bool, @@ -76,12 +129,53 @@ pub fn spark_pmod( let args = ColumnarValue::values_to_arrays(args)?; let left = &args[0]; let right = &args[1]; - let zero = ScalarValue::new_zero(left.data_type())?.to_array_of_size(left.len())?; - let result = try_rem(left, right, enable_ansi_mode)?; - let neg = lt(&result, &zero)?; - let plus = zip(&neg, right, &zero)?; - let result = add(&plus, &result)?; - let result = try_rem(&result, right, enable_ansi_mode)?; + + let divisor_is_zero = is_zero(right)?; + let divisor = if enable_ansi_mode { + // Spark's `pmod` is 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()); + } + Arc::clone(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 null = Scalar::new(new_null_array(right.data_type(), 1)); + zip(&divisor_is_zero, &null, right)? + }; + + let remainder = rem(left, &divisor)?; + + // Spark evaluates `(r + n) % n` with Java arithmetic. Java promotes `byte` + // and `short` operands to `int`, so those widths never overflow, while + // `int` and `long` are evaluated at their own width and wrap around. Widen + // the narrow integer types so the addition reproduces Java's promotion, and + // use the wrapping addition elsewhere so that, for example, + // `pmod(-1, -2147483648)` yields `2147483647` the way Spark does instead of + // reporting an overflow. + let widen = matches!(remainder.data_type(), DataType::Int8 | DataType::Int16); + let (wide_remainder, wide_divisor) = if widen { + ( + cast(&remainder, &DataType::Int32)?, + cast(&divisor, &DataType::Int32)?, + ) + } else { + (Arc::clone(&remainder), Arc::clone(&divisor)) + }; + let shifted = add_wrapping(&wide_remainder, &wide_divisor)?; + let shifted = rem(&shifted, &wide_divisor)?; + let shifted = if shifted.data_type() == remainder.data_type() { + shifted + } else { + cast(&shifted, remainder.data_type())? + }; + + // Select `(r + n) % n` only where `r < 0`. Adding zero to the other rows + // instead would turn Spark's `-0.0` results into `0.0`. + let result = zip(&is_negative(&remainder)?, &shifted, &remainder)?; Ok(ColumnarValue::Array(result)) } @@ -667,6 +761,182 @@ mod test { } } + /// Spark keeps the sign of a negative zero remainder, because `r < 0` is + /// false for `-0.0` in Java. The `.slt` runner renders `-0.0` and `0.0` + /// identically, so the sign can only be asserted here. + #[test] + fn test_pmod_negative_zero_result() { + let left = Float64Array::from(vec![ + Some(-5.0), + Some(-3.0), + Some(-0.0), + Some(5.0), + Some(0.0), + ]); + let right = Float64Array::from(vec![ + Some(2.5), + Some(3.0), + Some(3.0), + Some(2.5), + Some(3.0), + ]); + + let result = spark_pmod( + &[ + ColumnarValue::Array(Arc::new(left)), + ColumnarValue::Array(Arc::new(right)), + ], + false, + ) + .unwrap(); + + if let ColumnarValue::Array(result_array) = result { + let result = result_array + .as_any() + .downcast_ref::() + .unwrap(); + // -5.0 pmod 2.5 = -0.0 + assert_eq!(result.value(0), 0.0); + assert!(result.value(0).is_sign_negative()); + // -3.0 pmod 3.0 = -0.0 + assert!(result.value(1).is_sign_negative()); + // -0.0 pmod 3.0 = -0.0 + assert!(result.value(2).is_sign_negative()); + // 5.0 pmod 2.5 = 0.0 + assert!(result.value(3).is_sign_positive()); + // 0.0 pmod 3.0 = 0.0 + assert!(result.value(4).is_sign_positive()); + } else { + panic!("Expected array result"); + } + } + + /// Spark evaluates `(r + n) % n` with Java arithmetic, which wraps around + /// for `int` and `long` and promotes `byte` and `short` to `int`. + #[test] + fn test_pmod_integer_boundaries() { + let left = + Int32Array::from(vec![Some(-1), Some(-2), Some(i32::MAX), Some(i32::MIN)]); + let right = Int32Array::from(vec![ + Some(i32::MIN), + Some(i32::MIN), + Some(i32::MIN), + Some(-1), + ]); + let result = spark_pmod( + &[ + ColumnarValue::Array(Arc::new(left)), + ColumnarValue::Array(Arc::new(right)), + ], + false, + ) + .unwrap(); + if let ColumnarValue::Array(result_array) = result { + let result = result_array.as_any().downcast_ref::().unwrap(); + assert_eq!(result.value(0), i32::MAX); + assert_eq!(result.value(1), 2147483646); + assert_eq!(result.value(2), i32::MAX); + assert_eq!(result.value(3), 0); + } else { + panic!("Expected array result"); + } + + let left = Int64Array::from(vec![Some(-1), Some(i64::MIN)]); + let right = Int64Array::from(vec![Some(i64::MIN), Some(-1)]); + let result = spark_pmod( + &[ + ColumnarValue::Array(Arc::new(left)), + ColumnarValue::Array(Arc::new(right)), + ], + false, + ) + .unwrap(); + if let ColumnarValue::Array(result_array) = result { + let result = result_array.as_any().downcast_ref::().unwrap(); + assert_eq!(result.value(0), i64::MAX); + assert_eq!(result.value(1), 0); + } else { + panic!("Expected array result"); + } + + // Java widens byte and short to int, so `-1 pmod -128` stays `-1` + // rather than wrapping to `127`. + let left = Int8Array::from(vec![Some(-1), Some(i8::MIN)]); + let right = Int8Array::from(vec![Some(i8::MIN), Some(-1)]); + let result = spark_pmod( + &[ + ColumnarValue::Array(Arc::new(left)), + ColumnarValue::Array(Arc::new(right)), + ], + false, + ) + .unwrap(); + if let ColumnarValue::Array(result_array) = result { + let result = result_array.as_any().downcast_ref::().unwrap(); + assert_eq!(result.value(0), -1); + assert_eq!(result.value(1), 0); + } else { + panic!("Expected array result"); + } + + let left = Int16Array::from(vec![Some(-1), Some(i16::MIN), Some(128)]); + let right = Int16Array::from(vec![Some(i16::MIN), Some(-1), Some(i16::MIN)]); + let result = spark_pmod( + &[ + ColumnarValue::Array(Arc::new(left)), + ColumnarValue::Array(Arc::new(right)), + ], + false, + ) + .unwrap(); + if let ColumnarValue::Array(result_array) = result { + let result = result_array.as_any().downcast_ref::().unwrap(); + assert_eq!(result.value(0), -1); + assert_eq!(result.value(1), 0); + assert_eq!(result.value(2), 128); + } else { + panic!("Expected array result"); + } + } + + /// Spark raises on a zero divisor of any numeric type in ANSI mode, and + /// treats `-0.0` as a zero divisor. + #[test] + fn test_pmod_zero_divisor_by_type() { + let float_args = |divisor: f64| { + [ + ColumnarValue::Array(Arc::new(Float64Array::from(vec![Some(10.5)]))), + ColumnarValue::Array(Arc::new(Float64Array::from(vec![Some(divisor)]))), + ] + }; + + for divisor in [0.0, -0.0] { + assert!(spark_pmod(&float_args(divisor), true).is_err()); + + let result = spark_pmod(&float_args(divisor), false).unwrap(); + if let ColumnarValue::Array(result_array) = result { + assert!(result_array.is_null(0)); + } else { + panic!("Expected array result"); + } + } + + // A NULL dividend short circuits to NULL before the divisor is + // validated, so ANSI mode does not raise here. + let args = [ + ColumnarValue::Array(Arc::new(Int32Array::from(vec![None, Some(-7)]))), + ColumnarValue::Array(Arc::new(Int32Array::from(vec![Some(0), Some(3)]))), + ]; + let result = spark_pmod(&args, true).unwrap(); + if let ColumnarValue::Array(result_array) = result { + let result = result_array.as_any().downcast_ref::().unwrap(); + assert!(result.is_null(0)); + assert_eq!(result.value(1), 2); + } else { + panic!("Expected array result"); + } + } + #[test] fn test_pmod_edge_cases() { // Test edge cases for PMOD diff --git a/datafusion/sqllogictest/test_files/spark/math/pmod.slt b/datafusion/sqllogictest/test_files/spark/math/pmod.slt index aa4a197ba470f..320aa6d7b8144 100644 --- a/datafusion/sqllogictest/test_files/spark/math/pmod.slt +++ b/datafusion/sqllogictest/test_files/spark/math/pmod.slt @@ -21,6 +21,10 @@ # For more information, please see: # https://github.com/apache/datafusion/issues/15914 +# Audited against Spark 4.2.0 on 2026-07-25. +# Expected values verified with pyspark==4.2.0. +# Known divergences are captured as commented-out queries below. + # Basic PMOD tests with positive integers query I SELECT pmod(10::int, 3::int) as pmod_1; @@ -74,16 +78,82 @@ SELECT pmod(-7::int, 0::int) as pmod_zero_3; ---- NULL -# Division by zero errors in ANSI mode +# Division by zero errors in ANSI mode. +# Spark 4.2.0 raises REMAINDER_BY_ZERO (SQLSTATE 22012) here. DataFusion raises +# Arrow's divide-by-zero error instead: it does not model Spark's error classes +# or SQLSTATE values, so only the fact that an error is raised is asserted. +# https://github.com/apache/datafusion/issues/23897 +# Spark 3.5.8 and 4.0.4 raise DIVIDE_BY_ZERO for the same input, and Spark 4.1.3 +# renamed it to REMAINDER_BY_ZERO. The expectations below assert 4.2.0. +# Version-specific expectations are tracked by +# https://github.com/apache/datafusion/issues/23887 statement ok set datafusion.execution.enable_ansi_mode = true; -statement error DataFusion error: Arrow error: Divide by zero error +query error DataFusion error: Arrow error: Divide by zero error SELECT pmod(10::int, 0::int); -statement error DataFusion error: Arrow error: Divide by zero error +query error DataFusion error: Arrow error: Divide by zero error SELECT pmod(-7::int, 0::int); +# Spark raises for a zero divisor of any numeric type under ANSI mode, including +# floating point, where IEEE 754 remainder would otherwise yield NaN. +query error DataFusion error: Arrow error: Divide by zero error +SELECT pmod(10.5::float8, 0.0::float8); + +query error DataFusion error: Arrow error: Divide by zero error +SELECT pmod(10.5::float8, -0.0::float8); + +query error DataFusion error: Arrow error: Divide by zero error +SELECT pmod('NaN'::float8, 0.0::float8); + +query error DataFusion error: Arrow error: Divide by zero error +SELECT pmod(10.5::float4, 0.0::float4); + +query error DataFusion error: Arrow error: Divide by zero error +SELECT pmod(3.13::decimal(10,2), 0.0::decimal(10,2)); + +# pmod is null intolerant, so a NULL dividend short circuits to NULL before the +# zero divisor is validated, even under ANSI mode. +query I +SELECT pmod(NULL::int, 0::int); +---- +NULL + +query R +SELECT pmod(NULL::float8, 0.0::float8); +---- +NULL + +query R +SELECT pmod(NULL::decimal(10,2), 0.0::decimal(10,2)); +---- +NULL + +# The same short circuit applies per row on array inputs +query III +SELECT a, b, pmod(a, b) FROM (VALUES (NULL::int, 0::int), (NULL::int, 0::int)) AS t(a, b); +---- +NULL 0 NULL +NULL 0 NULL + +query III +SELECT a, b, pmod(a, b) FROM (VALUES (-7::int, 3::int), (NULL::int, 0::int)) AS t(a, b); +---- +-7 3 2 +NULL 0 NULL + +# ANSI mode does not change the result of a well defined pmod +query I +SELECT pmod(-7::int, 3::int); +---- +2 + +query I +SELECT pmod(arrow_cast(-1, 'Int32'), arrow_cast(-2147483648, 'Int32')); +---- +2147483647 + statement ok set datafusion.execution.enable_ansi_mode = false; @@ -280,6 +350,17 @@ SELECT pmod(-10.0::decimal(3,1), 3.0::decimal(2,1)) as pmod_decimal_4; ---- 2 +# Spark derives a narrower decimal result type than DataFusion. The values +# agree, only the declared precision differs. +# https://github.com/apache/datafusion/issues/23895 +# Spark 4.2.0: SELECT typeof(pmod(CAST(2.5 AS DECIMAL(3,1)), CAST(1.2 AS DECIMAL(2,1)))) +# -> decimal(2,1) +# DataFusion: Decimal128(3, 1) +# query T +# SELECT arrow_typeof(pmod(2.5::decimal(3,1), 1.2::decimal(2,1))); +# ---- +# Decimal128(2, 1) + # PMOD tests with different integer types query I SELECT pmod(10::int8, 3::int8) as pmod_int8_1; @@ -311,6 +392,9 @@ SELECT pmod(arrow_cast(-10, 'Int64'), arrow_cast(3, 'Int64')) as pmod_int64_2; ---- 2 +# Spark has no unsigned integer types, so the cases below have no Spark +# equivalent. `pmod` never adjusts an unsigned remainder, because it is never +# negative. # PMOD tests with unsigned integers query I SELECT pmod(arrow_cast(10, 'UInt8'), arrow_cast(3, 'UInt8')) as pmod_uint8_1; @@ -352,3 +436,154 @@ query R SELECT pmod(-7.2, 3.0) as pmod_scalar_4; ---- 1.8 + +# PMOD tests on array inputs, exercising every ColumnarValue shape combination + +# (Array, Scalar) +query II +SELECT a, pmod(a, 3::int) FROM (VALUES (10::int), (-10::int), (NULL::int), (0::int)) AS t(a); +---- +10 1 +-10 2 +NULL NULL +0 0 + +# (Scalar, Array) +query II +SELECT b, pmod(10::int, b) FROM (VALUES (3::int), (-3::int), (NULL::int), (0::int)) AS t(b); +---- +3 1 +-3 1 +NULL NULL +0 NULL + +# (Array, Array) +query III +SELECT a, b, pmod(a, b) FROM (VALUES + (10::int, 3::int), + (-10::int, 3::int), + (-10::int, -3::int), + (NULL::int, 3::int), + (10::int, NULL::int), + (10::int, 0::int) +) AS t(a, b); +---- +10 3 1 +-10 3 2 +-10 -3 -1 +NULL 3 NULL +10 NULL NULL +10 0 NULL + +# (Array, Array) on floating point, including the special values +query RRR +SELECT a, b, pmod(a, b) FROM (VALUES + (10.5::float8, 3.0::float8), + (-7.2::float8, 3.0::float8), + (-7.2::float8, -3.0::float8), + ('NaN'::float8, 2.0::float8), + ('Infinity'::float8, 2.0::float8), + (5.0::float8, 'Infinity'::float8), + (-5.0::float8, 'Infinity'::float8), + (5.0::float8, '-Infinity'::float8), + (10.5::float8, 0.0::float8) +) AS t(a, b); +---- +10.5 3 1.5 +-7.2 3 1.8 +-7.2 -3 -1.2 +NaN 2 NaN +Infinity 2 NaN +5 Infinity 5 +-5 Infinity NaN +5 -Infinity 5 +10.5 0 NULL + +# PMOD tests at the signed integer boundaries. +# Spark evaluates `(r + n) % n` with Java arithmetic: `byte` and `short` are +# promoted to `int` and never overflow, while `int` and `long` wrap around. +query I +SELECT pmod(arrow_cast(-1, 'Int32'), arrow_cast(-2147483648, 'Int32')); +---- +2147483647 + +query I +SELECT pmod(arrow_cast(-2, 'Int32'), arrow_cast(-2147483648, 'Int32')); +---- +2147483646 + +query I +SELECT pmod(arrow_cast(2147483647, 'Int32'), arrow_cast(-2147483648, 'Int32')); +---- +2147483647 + +query I +SELECT pmod(arrow_cast(-2147483648, 'Int32'), arrow_cast(-1, 'Int32')); +---- +0 + +query I +SELECT pmod(arrow_cast(-2147483648, 'Int32'), arrow_cast(3, 'Int32')); +---- +1 + +query I +SELECT pmod(arrow_cast(-1, 'Int64'), arrow_cast(-9223372036854775808, 'Int64')); +---- +9223372036854775807 + +query I +SELECT pmod(arrow_cast(-9223372036854775808, 'Int64'), arrow_cast(-1, 'Int64')); +---- +0 + +query I +SELECT pmod(arrow_cast(2, 'Int64'), arrow_cast(9223372036854775807, 'Int64')); +---- +2 + +query I +SELECT pmod(arrow_cast(-1, 'Int8'), arrow_cast(-128, 'Int8')); +---- +-1 + +query I +SELECT pmod(arrow_cast(-128, 'Int8'), arrow_cast(-1, 'Int8')); +---- +0 + +query I +SELECT pmod(arrow_cast(-1, 'Int16'), arrow_cast(-32768, 'Int16')); +---- +-1 + +query I +SELECT pmod(arrow_cast(-32768, 'Int16'), arrow_cast(-1, 'Int16')); +---- +0 + +query I +SELECT pmod(arrow_cast(128, 'Int16'), arrow_cast(-32768, 'Int16')); +---- +128 + +# The same boundary cases on array inputs +query II +SELECT a, pmod(a, arrow_cast(-2147483648, 'Int32')) FROM (VALUES + (arrow_cast(-1, 'Int32')), + (arrow_cast(-2, 'Int32')), + (arrow_cast(2147483647, 'Int32')), + (arrow_cast(NULL, 'Int32')) +) AS t(a); +---- +-1 2147483647 +-2 2147483646 +2147483647 2147483647 +NULL NULL + +# Spark accepts a string argument by implicitly casting it to a numeric type. +# DataFusion's `Signature::numeric` rejects it at planning time. +# https://github.com/apache/datafusion/issues/23896 +# Spark 4.2.0: SELECT pmod('10', CAST(3 AS INT)) -> 1 +# query I +# SELECT pmod('10'::string, 3::int);