Use strict cast in try_cast_to to error on overflow instead of silent NULL

Author: sgrebnov

datafusion-federation/src/schema_cast/record_convert.rs

@@ -1,6 +1,6 @@
 use datafusion::arrow::{
     array::{Array, RecordBatch, RecordBatchOptions},
-    compute::cast,
+    compute::{cast_with_options, CastOptions},
     datatypes::{DataType, IntervalUnit, SchemaRef},
 };
 use std::sync::Arc;
@@ -153,8 +153,18 @@ pub fn try_cast_to(record_batch: RecordBatch, expected_schema: SchemaRef) -> Res
                     DataType::Interval(IntervalUnit::DayTime),
                 ) => cast_interval_monthdaynano_to_daytime(&Arc::clone(record_batch_col))
                     .map_err(make_err),
-                _ => cast(&Arc::clone(record_batch_col), expected_field.data_type())
-                    .map_err(make_err),
+                _ => {
+                    let options = CastOptions {
+                        safe: false,
+                        ..CastOptions::default()
+                    };
+                    cast_with_options(
+                        &Arc::clone(record_batch_col),
+                        expected_field.data_type(),
+                        &options,
+                    )
+                    .map_err(make_err)
+                }
             }
         })
         .collect::<Result<Vec<Arc<dyn Array>>>>()
@@ -295,4 +305,71 @@ mod test {
         let expected = ["++", "++", "++"];
         assert_batches_eq!(expected, &[result]);
     }
+
+    /// Casting Decimal128(38,9) → Decimal128(38,27) must return an error when
+    /// the upscale would overflow, instead of silently producing NULL.
+    #[test]
+    fn test_try_cast_to_decimal_overflow_returns_error() {
+        // Value with 12 integer digits: 110_367_043_872.497010000
+        // Internal i128 at scale 9 = 110367043872497010000
+        let value_i128: i128 = 110_367_043_872_497_010_000;
+
+        let source_schema = Arc::new(Schema::new(vec![Field::new(
+            "sum_charge",
+            DataType::Decimal128(38, 9),
+            true,
+        )]));
+
+        let source_array = Decimal128Array::from(vec![Some(value_i128)])
+            .with_precision_and_scale(38, 9)
+            .expect("valid Decimal128(38,9)");
+
+        let batch =
+            RecordBatch::try_new(source_schema, vec![Arc::new(source_array)]).expect("valid batch");
+
+        // Target schema with wider scale (38,27) — only allows 11 integer digits
+        let target_schema = Arc::new(Schema::new(vec![Field::new(
+            "sum_charge",
+            DataType::Decimal128(38, 27),
+            true,
+        )]));
+
+        let result = try_cast_to(batch, target_schema);
+        assert!(
+            result.is_err(),
+            "Decimal overflow should return an error, not silently produce NULL"
+        );
+    }
+
+    /// Casting Decimal128 with values that fit should succeed.
+    #[test]
+    fn test_try_cast_to_decimal_no_overflow_succeeds() {
+        // Value with 11 integer digits: 99_999_999_999.000000000 (fits in 38-27=11 digits)
+        let value_i128: i128 = 99_999_999_999_000_000_000;
+
+        let source_schema = Arc::new(Schema::new(vec![Field::new(
+            "amount",
+            DataType::Decimal128(38, 9),
+            true,
+        )]));
+
+        let source_array = Decimal128Array::from(vec![Some(value_i128)])
+            .with_precision_and_scale(38, 9)
+            .expect("valid Decimal128(38,9)");
+
+        let batch =
+            RecordBatch::try_new(source_schema, vec![Arc::new(source_array)]).expect("valid batch");
+
+        let target_schema = Arc::new(Schema::new(vec![Field::new(
+            "amount",
+            DataType::Decimal128(38, 27),
+            true,
+        )]));
+
+        let result = try_cast_to(batch, target_schema);
+        assert!(
+            result.is_ok(),
+            "Decimal cast should succeed when value fits: {result:?}"
+        );
+    }
 }