fix: correctly rounded decimal to double/float cast matching BigDecimal.doubleValue/floatValue

native/spark-expr/benches/cast_numeric.rs

@@ -15,13 +15,16 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use arrow::array::{builder::Int32Builder, RecordBatch};
+use arrow::array::{builder::Int32Builder, Decimal128Array, RecordBatch};
 use arrow::datatypes::{DataType, Field, Schema};
 use criterion::{criterion_group, criterion_main, Criterion};
 use datafusion::physical_expr::{expressions::Column, PhysicalExpr};
 use datafusion_comet_spark_expr::{Cast, EvalMode, SparkCastOptions};
+use std::hint::black_box;
 use std::sync::Arc;
 
+const NUM_ROWS: usize = 8192;
+
 fn criterion_benchmark(c: &mut Criterion) {
     let batch = create_int32_batch();
     let expr = Arc::new(Column::new("a", 0));
@@ -52,6 +55,59 @@ fn criterion_benchmark(c: &mut Criterion) {
     group.bench_function("cast_i32_to_i64", |b| {
         b.iter(|| cast_i32_to_i64.evaluate(&batch).unwrap());
     });
+    group.finish();
+
+    let decimal_cast = |data_type| {
+        Cast::new(
+            Arc::new(Column::new("a", 0)),
+            data_type,
+            SparkCastOptions::new_without_timezone(EvalMode::Legacy, false),
+            None,
+            None,
+        )
+    };
+    let decimal_to_f64 = decimal_cast(DataType::Float64);
+    let decimal_to_f32 = decimal_cast(DataType::Float32);
+    let cases = [
+        (
+            "decimal18_to_f64",
+            create_decimal128_batch(18, 0, 1_i128 << 53),
+            &decimal_to_f64,
+        ),
+        (
+            "decimal18_to_f64_nulls",
+            create_decimal128_batch(18, 5, 1_i128 << 53),
+            &decimal_to_f64,
+        ),
+        (
+            "decimal38_to_f64",
+            create_decimal128_batch(38, 0, 10_i128.pow(37)),
+            &decimal_to_f64,
+        ),
+        (
+            "decimal38_to_f64_nulls",
+            create_decimal128_batch(38, 5, 10_i128.pow(37)),
+            &decimal_to_f64,
+        ),
+        (
+            "decimal12_to_f32",
+            create_decimal128_batch(12, 0, 1_i128 << 24),
+            &decimal_to_f32,
+        ),
+        (
+            "decimal12_to_f32_nulls",
+            create_decimal128_batch(12, 5, 1_i128 << 24),
+            &decimal_to_f32,
+        ),
+    ];
+
+    let mut group = c.benchmark_group("cast_decimal_scale_zero");
+    for (name, batch, cast) in cases {
+        group.bench_function(name, |b| {
+            b.iter(|| black_box(cast.evaluate(black_box(&batch)).unwrap()))
+        });
+    }
+    group.finish();
 }
 
 fn create_int32_batch() -> RecordBatch {
@@ -69,6 +125,27 @@ fn create_int32_batch() -> RecordBatch {
     RecordBatch::try_new(schema.clone(), vec![Arc::new(array)]).unwrap()
 }
 
+fn create_decimal128_batch(precision: u8, null_every: usize, base: i128) -> RecordBatch {
+    let array: Decimal128Array = (0..NUM_ROWS)
+        .map(|i| {
+            if null_every != 0 && i % null_every == 0 {
+                None
+            } else {
+                let magnitude = base + i as i128;
+                Some(if i % 2 == 0 { magnitude } else { -magnitude })
+            }
+        })
+        .collect::<Decimal128Array>()
+        .with_precision_and_scale(precision, 0)
+        .unwrap();
+    let schema = Arc::new(Schema::new(vec![Field::new(
+        "a",
+        DataType::Decimal128(precision, 0),
+        true,
+    )]));
+    RecordBatch::try_new(schema, vec![Arc::new(array)]).unwrap()
+}
+
 fn config() -> Criterion {
     Criterion::default()
 }

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

@@ -19,12 +19,13 @@ use crate::conversion_funcs::boolean::{
     cast_boolean_to_timestamp, is_df_cast_from_bool_spark_compatible,
 };
 use crate::conversion_funcs::numeric::{
-    cast_decimal128_to_utf8, cast_decimal_to_timestamp, cast_float32_to_decimal128,
-    cast_float64_to_decimal128, cast_float_to_timestamp, cast_int_to_decimal128,
-    cast_int_to_timestamp, is_df_cast_from_decimal_spark_compatible,
-    is_df_cast_from_float_spark_compatible, is_df_cast_from_int_spark_compatible,
-    spark_cast_decimal_to_boolean, spark_cast_float32_to_utf8, spark_cast_float64_to_utf8,
-    spark_cast_int_to_int, spark_cast_nonintegral_numeric_to_integral,
+    cast_decimal128_to_float32, cast_decimal128_to_float64, cast_decimal128_to_utf8,
+    cast_decimal_to_timestamp, cast_float32_to_decimal128, cast_float64_to_decimal128,
+    cast_float_to_timestamp, cast_int_to_decimal128, cast_int_to_timestamp,
+    is_df_cast_from_decimal_spark_compatible, is_df_cast_from_float_spark_compatible,
+    is_df_cast_from_int_spark_compatible, spark_cast_decimal_to_boolean,
+    spark_cast_float32_to_utf8, spark_cast_float64_to_utf8, spark_cast_int_to_int,
+    spark_cast_nonintegral_numeric_to_integral,
 };
 use crate::conversion_funcs::string::{
     cast_string_to_date, cast_string_to_decimal, cast_string_to_float, cast_string_to_int,
@@ -327,6 +328,11 @@ pub(crate) fn cast_array(
             spark_cast_nonintegral_numeric_to_integral(&array, eval_mode, &from_type, to_type)
         }
         (Decimal128(_p, _s), Boolean) => spark_cast_decimal_to_boolean(&array),
+        // Spark rounds the exact decimal value once (BigDecimal.doubleValue / floatValue);
+        // DataFusion's `(unscaled as f64) / 10^scale` rounds twice and can be off by one ulp.
+        // The conversion cannot fail, so it is the same in every eval mode.
+        (Decimal128(_, scale), Float64) => cast_decimal128_to_float64(&array, *scale),
+        (Decimal128(_, scale), Float32) => cast_decimal128_to_float32(&array, *scale),
         // Spark LEGACY cast uses Java BigDecimal.toString() which produces scientific notation
         // when adjusted_exponent < -6 (e.g. "0E-18" for zero with scale=18).
         // TRY and ANSI use plain notation ("0.000000000000000000") so DataFusion handles those.