diff --git a/native/spark-expr/src/conversion_funcs/numeric.rs b/native/spark-expr/src/conversion_funcs/numeric.rs index 029a38e3e7f..21be9a49fe5 100644 --- a/native/spark-expr/src/conversion_funcs/numeric.rs +++ b/native/spark-expr/src/conversion_funcs/numeric.rs @@ -305,6 +305,20 @@ macro_rules! cast_int_to_int_macro { }}; } +/// Spark's ANSI range check for casting a floating point value to a 32-bit or 64-bit integer, +/// from `FloatExactNumeric`/`DoubleExactNumeric.toInt`/`toLong` in Spark's `numerics.scala`: +/// `Math.floor(x) <= MaxValue && Math.ceil(x) >= MinValue`. The JVM widens a float source to +/// double for `Math.floor`/`Math.ceil` and converts the integer bounds to double for the +/// comparison, so the check is evaluated in `f64` here as well. A value that passes it is then +/// truncated towards zero with saturation by `as`, exactly like the JVM's `d2i`/`d2l`, so the +/// exactly representable bounds are valid inputs; note that `i64::MAX as f64 == 2^63`, so `±2^63` +/// passes the check and saturates to `i64::MAX`/`i64::MIN`, as it does in Spark. NaN fails both +/// comparisons and the infinities fail one of them. +fn spark_float_fits_integral>(value: F, min: f64, max: f64) -> bool { + let value: f64 = value.into(); + value.floor() <= max && value.ceil() >= min +} + // When Spark casts to Byte/Short Types, it does not cast directly to Byte/Short. // It casts to Int first and then to Byte/Short. Because of potential overflows in the Int cast, // this can cause unexpected Short/Byte cast results. Replicate this behavior. @@ -330,8 +344,10 @@ macro_rules! cast_float_to_int16_down { .iter() .map(|value| match value { Some(value) => { - let is_overflow = value.is_nan() || value.abs() as i32 == i32::MAX; - if is_overflow { + // Spark's ANSI cast converts to Int with the exact 32-bit range check + // first and then requires the truncated Int to round-trip through the + // narrower type. + if !spark_float_fits_integral(value, i32::MIN as f64, i32::MAX as f64) { return Err(cast_overflow( &format!($format_str, value).replace("e", "E"), $src_type_str, @@ -379,7 +395,6 @@ macro_rules! cast_float_to_int32_up { $rust_dest_type:ty, $src_type_str:expr, $dest_type_str:expr, - $max_dest_val:expr, $format_str:expr ) => {{ let cast_array = $array @@ -392,15 +407,18 @@ macro_rules! cast_float_to_int32_up { .iter() .map(|value| match value { Some(value) => { - let is_overflow = - value.is_nan() || value.abs() as $rust_dest_type == $max_dest_val; - if is_overflow { + if !spark_float_fits_integral( + value, + <$rust_dest_type>::MIN as f64, + <$rust_dest_type>::MAX as f64, + ) { return Err(cast_overflow( &format!($format_str, value).replace("e", "E"), $src_type_str, $dest_type_str, )); } + // `as` truncates towards zero and saturates like the JVM's d2i/d2l Ok(Some(value as $rust_dest_type)) } None => Ok(None), @@ -1056,7 +1074,6 @@ pub(crate) fn spark_cast_nonintegral_numeric_to_integral( i32, "FLOAT", "INT", - i32::MAX, "{:e}" ), (DataType::Float32, DataType::Int64) => cast_float_to_int32_up!( @@ -1068,7 +1085,6 @@ pub(crate) fn spark_cast_nonintegral_numeric_to_integral( i64, "FLOAT", "BIGINT", - i64::MAX, "{:e}" ), (DataType::Float64, DataType::Int8) => cast_float_to_int16_down!( @@ -1102,7 +1118,6 @@ pub(crate) fn spark_cast_nonintegral_numeric_to_integral( i32, "DOUBLE", "INT", - i32::MAX, "{:e}D" ), (DataType::Float64, DataType::Int64) => cast_float_to_int32_up!( @@ -1114,7 +1129,6 @@ pub(crate) fn spark_cast_nonintegral_numeric_to_integral( i64, "DOUBLE", "BIGINT", - i64::MAX, "{:e}D" ), (DataType::Decimal128(precision, scale), DataType::Int8) => { @@ -1261,6 +1275,333 @@ mod tests { assert!(result.is_err()); } + /// Casts `values` from the floating point type `S` to the integral type `D` with + /// `spark_cast_nonintegral_numeric_to_integral` and returns the resulting values. + fn cast_float_to_integral( + values: Vec, + eval_mode: EvalMode, + ) -> SparkResult> { + let array = PrimitiveArray::::from_iter_values(values); + let result = spark_cast_nonintegral_numeric_to_integral( + &array, + eval_mode, + &S::DATA_TYPE, + &D::DATA_TYPE, + )?; + Ok(result.as_primitive::().values().to_vec()) + } + + /// Asserts that each value overflows on its own in ANSI mode with Spark's `CAST_OVERFLOW` + /// error naming the source and target types. + fn assert_ansi_cast_overflow( + values: &[S::Native], + from_type: &str, + to_type: &str, + ) { + for value in values { + match cast_float_to_integral::(vec![*value], EvalMode::Ansi) { + Err(SparkError::CastOverFlow { + from_type: from, + to_type: to, + .. + }) => assert_eq!( + (from.as_str(), to.as_str()), + (from_type, to_type), + "unexpected types in the overflow error for {value:?}" + ), + other => panic!("expected CAST_OVERFLOW for {value:?}, got {other:?}"), + } + } + } + + #[test] + fn test_cast_double_to_int_ansi_boundaries() { + // Spark accepts x when floor(x) <= i32::MAX && ceil(x) >= i32::MIN and truncates it + // towards zero, so the exactly representable bounds and the fractional values inside + // them are valid inputs + assert_eq!( + cast_float_to_integral::( + vec![ + 2147483647.0, + -2147483648.0, + 2147483647.5, + -2147483648.5, + 2147483648.0f64.next_down(), + (-2147483649.0f64).next_up(), + ], + EvalMode::Ansi, + ) + .unwrap(), + vec![i32::MAX, i32::MIN, i32::MAX, i32::MIN, i32::MAX, i32::MIN] + ); + assert_ansi_cast_overflow::( + &[ + 2147483648.0, + -2147483649.0, + 2147483648.5, + -2147483649.5, + 1e10, + f64::MAX, + f64::MIN, + f64::NAN, + f64::INFINITY, + f64::NEG_INFINITY, + ], + "DOUBLE", + "INT", + ); + // the offending value is rendered like Spark's Double.toString plus the D suffix + match cast_float_to_integral::(vec![2147483648.0], EvalMode::Ansi) { + Err(SparkError::CastOverFlow { value, .. }) => assert_eq!(value, "2.147483648E9D"), + other => panic!("expected CAST_OVERFLOW, got {other:?}"), + } + } + + #[test] + fn test_cast_double_to_long_ansi_boundaries() { + // i64::MAX converted to f64 rounds up to 2^63, which Spark therefore accepts and + // saturates to i64::MAX, exactly like the JVM's d2l + let two_pow_63 = 9223372036854775808.0f64; + assert_eq!(i64::MAX as f64, two_pow_63); + assert_eq!( + cast_float_to_integral::( + vec![ + two_pow_63, + -two_pow_63, + two_pow_63.next_down(), + (-two_pow_63).next_up(), + ], + EvalMode::Ansi, + ) + .unwrap(), + vec![ + i64::MAX, + i64::MIN, + 9223372036854774784, + -9223372036854774784 + ] + ); + assert_ansi_cast_overflow::( + &[ + two_pow_63.next_up(), + (-two_pow_63).next_down(), + 1e19, + -1e19, + f64::MAX, + f64::NAN, + f64::INFINITY, + f64::NEG_INFINITY, + ], + "DOUBLE", + "BIGINT", + ); + } + + #[test] + fn test_cast_float_to_int_ansi_boundaries() { + // The bounds are compared in double precision, as in Spark: i32::MAX is not + // representable as f32 and rounds up to 2^31, which must overflow, while the largest + // f32 below 2^31 (2147483520) and -2^31 itself are accepted + let two_pow_31 = 2147483648.0f32; + assert_eq!(i32::MAX as f32, two_pow_31); + assert_eq!( + cast_float_to_integral::( + vec![two_pow_31.next_down(), -two_pow_31, (-two_pow_31).next_up()], + EvalMode::Ansi, + ) + .unwrap(), + vec![2147483520, i32::MIN, -2147483520] + ); + assert_ansi_cast_overflow::( + &[ + two_pow_31, + (-two_pow_31).next_down(), + 1e10, + f32::MAX, + f32::MIN, + f32::NAN, + f32::INFINITY, + f32::NEG_INFINITY, + ], + "FLOAT", + "INT", + ); + } + + #[test] + fn test_cast_float_to_long_ansi_boundaries() { + let two_pow_63 = 9223372036854775808.0f32; + assert_eq!(i64::MAX as f32, two_pow_63); + assert_eq!( + cast_float_to_integral::( + vec![ + two_pow_63, + -two_pow_63, + two_pow_63.next_down(), + (-two_pow_63).next_up(), + ], + EvalMode::Ansi, + ) + .unwrap(), + vec![ + i64::MAX, + i64::MIN, + 9223371487098961920, + -9223371487098961920 + ] + ); + assert_ansi_cast_overflow::( + &[ + two_pow_63.next_up(), + (-two_pow_63).next_down(), + 1e19, + f32::MAX, + f32::NAN, + f32::INFINITY, + f32::NEG_INFINITY, + ], + "FLOAT", + "BIGINT", + ); + } + + #[test] + fn test_cast_float_to_short_and_byte_ansi_boundaries() { + // Spark converts to INT first and then requires the truncated INT to round-trip through + // the narrower type, so 127.9 is a valid TINYINT while 128.0 is not, and a value that + // passes the INT range check can still overflow the narrower type + assert_eq!( + cast_float_to_integral::( + vec![127.0, -128.0, 127.9, -128.9], + EvalMode::Ansi + ) + .unwrap(), + vec![i8::MAX, i8::MIN, i8::MAX, i8::MIN] + ); + assert_ansi_cast_overflow::( + &[ + 128.0, + -129.0, + 2147483647.5, + 2147483648.0, + f64::NAN, + f64::INFINITY, + f64::NEG_INFINITY, + ], + "DOUBLE", + "TINYINT", + ); + assert_eq!( + cast_float_to_integral::( + vec![32767.0, -32768.0, 32767.9, -32768.9], + EvalMode::Ansi + ) + .unwrap(), + vec![i16::MAX, i16::MIN, i16::MAX, i16::MIN] + ); + assert_ansi_cast_overflow::( + &[ + 32768.0, + -32769.0, + 2147483647.5, + 2147483648.0, + f64::NAN, + f64::INFINITY, + f64::NEG_INFINITY, + ], + "DOUBLE", + "SMALLINT", + ); + assert_eq!( + cast_float_to_integral::( + vec![127.0, -128.0, 127.9, -128.9], + EvalMode::Ansi + ) + .unwrap(), + vec![i8::MAX, i8::MIN, i8::MAX, i8::MIN] + ); + assert_ansi_cast_overflow::( + &[ + 128.0, + -129.0, + 2147483520.0, + 2147483648.0, + f32::NAN, + f32::INFINITY, + f32::NEG_INFINITY, + ], + "FLOAT", + "TINYINT", + ); + assert_eq!( + cast_float_to_integral::( + vec![32767.0, -32768.0, 32767.9, -32768.9], + EvalMode::Ansi + ) + .unwrap(), + vec![i16::MAX, i16::MIN, i16::MAX, i16::MIN] + ); + assert_ansi_cast_overflow::( + &[ + 32768.0, + -32769.0, + 2147483520.0, + 2147483648.0, + f32::NAN, + f32::INFINITY, + f32::NEG_INFINITY, + ], + "FLOAT", + "SMALLINT", + ); + } + + #[test] + fn test_cast_float_to_integral_legacy_saturates() { + // LEGACY mode keeps truncating towards zero with saturation and mapping NaN to 0 + assert_eq!( + cast_float_to_integral::( + vec![ + 2147483648.0, + -2147483649.0, + 1e19, + -1e19, + f64::NAN, + f64::INFINITY, + f64::NEG_INFINITY, + ], + EvalMode::Legacy, + ) + .unwrap(), + vec![ + i32::MAX, + i32::MIN, + i32::MAX, + i32::MIN, + 0, + i32::MAX, + i32::MIN + ] + ); + assert_eq!( + cast_float_to_integral::( + vec![f32::MAX, f32::MIN, f32::NAN], + EvalMode::Legacy + ) + .unwrap(), + vec![i64::MAX, i64::MIN, 0] + ); + // narrowing casts go through INT first, so the INT result wraps like Spark's toShort + assert_eq!( + cast_float_to_integral::( + vec![32768.0, 2147483648.0], + EvalMode::Legacy + ) + .unwrap(), + vec![i16::MIN, -1] + ); + } + #[test] fn test_spark_cast_float_min_value_to_string() { let float_array: ArrayRef = Arc::new(Float32Array::from(vec![ diff --git a/spark/src/test/scala/org/apache/comet/CometNativeCastSuite.scala b/spark/src/test/scala/org/apache/comet/CometNativeCastSuite.scala index 9bfa8f63774..233cae38430 100644 --- a/spark/src/test/scala/org/apache/comet/CometNativeCastSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometNativeCastSuite.scala @@ -537,6 +537,93 @@ class CometNativeCastSuite extends CometTestBase with AdaptiveSparkPlanHelper { castTest(generateFloats(), DataTypes.LongType) } + // Boundary values of the ANSI casts from floating point to integral types + // (https://github.com/apache/datafusion-comet/issues/5673). Spark accepts a value when + // `floor(x) <= MaxValue && ceil(x) >= MinValue` holds in double precision and then truncates it + // towards zero with saturation, so the exactly representable bounds are valid inputs and, since + // Long.MaxValue.toDouble == 2^63, so is ±2^63 for BIGINT. Casts to TINYINT and SMALLINT first + // apply the INT range check and then require the truncated INT to fit the narrower type. + // + // Values that must overflow are cast one at a time so that each of them is checked in both + // engines. Comet renders the offending value in the error message with Rust's `{:e}` format, + // which only agrees with Java's Double.toString/Float.toString for large values with several + // significant digits, so the overflowing values are chosen from that range; the exact bounds of + // the narrower targets (such as 128.0 for TINYINT or 2^31 as a FLOAT for INT) are covered by the + // native unit tests in numeric.rs. + + test("cast FloatType to IntegerType - ANSI boundary values") { + // 2147483520.0f is the largest float below 2^31 (Int.MaxValue.toFloat rounds up to 2^31) + castTest( + withNulls(Seq(2147483520.0f, Int.MinValue.toFloat, -2147483520.0f)).toDF("a"), + DataTypes.IntegerType, + useDataFrameDiff = true) + Seq(2.5e9f, -2.5e9f, Float.NaN).foreach { v => + castTest( + Seq(v).toDF("a"), + DataTypes.IntegerType, + expectAnsiFailure = true, + useDataFrameDiff = true) + } + } + + test("cast FloatType to LongType - ANSI boundary values") { + castTest( + withNulls(Seq(Math.nextDown(Long.MaxValue.toFloat), Long.MinValue.toFloat)).toDF("a"), + DataTypes.LongType, + useDataFrameDiff = true) + // Long.MaxValue.toFloat == 2^63 is accepted by Spark and saturates to Long.MaxValue. + // TODO: try_cast is not compared for this value because the native TRY path goes through + // Arrow's cast, which returns NULL for 2^63 where Spark returns Long.MaxValue + castTest( + withNulls(Seq(Long.MaxValue.toFloat)).toDF("a"), + DataTypes.LongType, + testTry = false, + useDataFrameDiff = true) + Seq( + Math.nextUp(Long.MaxValue.toFloat), + Math.nextDown(Long.MinValue.toFloat), + 1.5e19f, + -1.5e19f, + Float.NaN).foreach { v => + castTest( + Seq(v).toDF("a"), + DataTypes.LongType, + expectAnsiFailure = true, + useDataFrameDiff = true) + } + } + + test("cast FloatType to ShortType - ANSI boundary values") { + castTest( + withNulls(Seq(Short.MaxValue.toFloat, Short.MinValue.toFloat, 32767.9f, -32768.9f)) + .toDF("a"), + DataTypes.ShortType, + useDataFrameDiff = true) + // 1.5e7 passes the INT range check but does not fit a SMALLINT; 2.5e9 fails the INT check + Seq(1.5e7f, -1.5e7f, 2.5e9f, -2.5e9f, Float.NaN).foreach { v => + castTest( + Seq(v).toDF("a"), + DataTypes.ShortType, + expectAnsiFailure = true, + useDataFrameDiff = true) + } + } + + test("cast FloatType to ByteType - ANSI boundary values") { + castTest( + withNulls(Seq(Byte.MaxValue.toFloat, Byte.MinValue.toFloat, 127.9f, -128.9f)).toDF("a"), + DataTypes.ByteType, + useDataFrameDiff = true) + // 1.5e7 passes the INT range check but does not fit a TINYINT; 2.5e9 fails the INT check + Seq(1.5e7f, -1.5e7f, 2.5e9f, -2.5e9f, Float.NaN).foreach { v => + castTest( + Seq(v).toDF("a"), + DataTypes.ByteType, + expectAnsiFailure = true, + useDataFrameDiff = true) + } + } + test("cast FloatType to DoubleType") { castTest(generateFloats(), DataTypes.DoubleType) } @@ -614,6 +701,80 @@ class CometNativeCastSuite extends CometTestBase with AdaptiveSparkPlanHelper { castTest(generateDoubles(), DataTypes.LongType) } + test("cast DoubleType to IntegerType - ANSI boundary values") { + castTest( + withNulls(Seq(Int.MaxValue.toDouble, Int.MinValue.toDouble, 2147483647.5d, -2147483648.5d)) + .toDF("a"), + DataTypes.IntegerType, + useDataFrameDiff = true) + Seq(2147483648.0d, -2147483649.0d, 2147483648.5d, -2147483649.5d).foreach { v => + castTest( + Seq(v).toDF("a"), + DataTypes.IntegerType, + expectAnsiFailure = true, + useDataFrameDiff = true) + } + } + + test("cast DoubleType to LongType - ANSI boundary values") { + castTest( + withNulls(Seq(Math.nextDown(Long.MaxValue.toDouble), Long.MinValue.toDouble)).toDF("a"), + DataTypes.LongType, + useDataFrameDiff = true) + // Long.MaxValue.toDouble == 2^63 is accepted by Spark and saturates to Long.MaxValue. + // TODO: try_cast is not compared for this value because the native TRY path goes through + // Arrow's cast, which returns NULL for 2^63 where Spark returns Long.MaxValue + castTest( + withNulls(Seq(Long.MaxValue.toDouble)).toDF("a"), + DataTypes.LongType, + testTry = false, + useDataFrameDiff = true) + Seq( + Math.nextUp(Long.MaxValue.toDouble), + Math.nextDown(Long.MinValue.toDouble), + 1.5e19d, + -1.5e19d).foreach { v => + castTest( + Seq(v).toDF("a"), + DataTypes.LongType, + expectAnsiFailure = true, + useDataFrameDiff = true) + } + } + + test("cast DoubleType to ShortType - ANSI boundary values") { + castTest( + withNulls(Seq(Short.MaxValue.toDouble, Short.MinValue.toDouble, 32767.9d, -32768.9d)) + .toDF("a"), + DataTypes.ShortType, + useDataFrameDiff = true) + // 2147483647.5 passes the INT range check but the truncated INT does not fit a SMALLINT; + // 2147483648.0 fails the INT range check itself + Seq(2147483647.5d, -2147483648.5d, 2147483648.0d, -2147483649.0d).foreach { v => + castTest( + Seq(v).toDF("a"), + DataTypes.ShortType, + expectAnsiFailure = true, + useDataFrameDiff = true) + } + } + + test("cast DoubleType to ByteType - ANSI boundary values") { + castTest( + withNulls(Seq(Byte.MaxValue.toDouble, Byte.MinValue.toDouble, 127.9d, -128.9d)).toDF("a"), + DataTypes.ByteType, + useDataFrameDiff = true) + // 2147483647.5 passes the INT range check but the truncated INT does not fit a TINYINT; + // 2147483648.0 fails the INT range check itself + Seq(2147483647.5d, -2147483648.5d, 2147483648.0d, -2147483649.0d).foreach { v => + castTest( + Seq(v).toDF("a"), + DataTypes.ByteType, + expectAnsiFailure = true, + useDataFrameDiff = true) + } + } + test("cast DoubleType to FloatType") { castTest(generateDoubles(), DataTypes.FloatType) }