fix: Make cast from float/double to decimal compatible with Spark

docs/source/user-guide/compatibility.md

@@ -174,13 +174,15 @@ The following cast operations are generally compatible with Spark except for the
 | float | integer |  |
 | float | long |  |
 | float | double |  |
+| float | decimal |  |
 | float | string | There can be differences in precision. For example, the input "1.4E-45" will produce 1.0E-45 instead of 1.4E-45 |
 | double | boolean |  |
 | double | byte |  |
 | double | short |  |
 | double | integer |  |
 | double | long |  |
 | double | float |  |
+| double | decimal |  |
 | double | string | There can be differences in precision. For example, the input "1.4E-45" will produce 1.0E-45 instead of 1.4E-45 |
 | decimal | byte |  |
 | decimal | short |  |
@@ -210,8 +212,6 @@ The following cast operations are not compatible with Spark for all inputs and a
 |-|-|-|
 | integer | decimal  | No overflow check |
 | long | decimal  | No overflow check |
-| float | decimal  | There can be rounding differences |
-| double | decimal  | There can be rounding differences |
 | string | float  | Does not support inputs ending with 'd' or 'f'. Does not support 'inf'. Does not support ANSI mode. |
 | string | double  | Does not support inputs ending with 'd' or 'f'. Does not support 'inf'. Does not support ANSI mode. |
 | string | decimal  | Does not support inputs ending with 'd' or 'f'. Does not support 'inf'. Does not support ANSI mode. Returns 0.0 instead of null if input contains no digits |

native/spark-expr/src/conversion_funcs/cast.rs

@@ -15,6 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
+use crate::conversion_funcs::schubfach;
 use crate::timezone;
 use crate::utils::array_with_timezone;
 use crate::{EvalMode, SparkError, SparkResult};
@@ -49,6 +50,7 @@ use num::{
     ToPrimitive,
 };
 use regex::Regex;
+use std::num::Saturating;
 use std::str::FromStr;
 use std::{
     any::Any,
@@ -1278,17 +1280,29 @@ where
     let input = array.as_any().downcast_ref::<PrimitiveArray<T>>().unwrap();
     let mut cast_array = PrimitiveArray::<Decimal128Type>::builder(input.len());
 
-    let mul = 10_f64.powi(scale as i32);
-
     for i in 0..input.len() {
         if input.is_null(i) {
             cast_array.append_null();
         } else {
             let input_value = input.value(i).as_();
-            let value = (input_value * mul).round().to_i128();
+            let value = schubfach::to_decimal(input_value);
 
             match value {
-                Some(v) => {
+                Some((significand, exponent)) => {
+                    let mut v = if input_value < 0. {
+                        -significand
+                    } else {
+                        significand
+                    } as i128;
+
+                    let k = exponent + scale as i32;
+                    if k > 0 {
+                        v = v.saturating_mul(Saturating(10_i128).pow(k as u32).0);
+                    } else if k < 0 {
+                        let dk = Saturating(10_i128).pow((-k) as u32).0;
+                        let (div, rem) = if dk < v { v.div_rem(&dk) } else { (0, v) };
+                        v = if rem * 2 >= dk { div + 1 } else { div };
+                    }
                     if Decimal128Type::validate_decimal_precision(v, precision).is_err() {
                         if eval_mode == EvalMode::Ansi {
                             return Err(SparkError::NumericValueOutOfRange {
@@ -1298,6 +1312,7 @@ where
                             });
                         } else {
                             cast_array.append_null();
+                            continue;
                         }
                     }
                     cast_array.append_value(v);

native/spark-expr/src/conversion_funcs/mod.rs

@@ -16,3 +16,5 @@
 // under the License.
 
 pub mod cast;
+
+mod schubfach;