Return errors from decimal conversions

datafusion_iceberg/src/pruning_statistics.rs

@@ -291,7 +291,11 @@ fn any_iter_to_array(
                 ScalarValue::Decimal128(
                     opt.and_then(|value| {
                         let d = *value.downcast::<Decimal>().ok()?;
-                        (decimal_scale(&d) == scale as u32).then(|| decimal_mantissa(&d))
+                        if decimal_scale(&d) == scale as u32 {
+                            decimal_mantissa(&d).ok()
+                        } else {
+                            None
+                        }
                     }),
                     precision,
                     scale,
@@ -751,7 +755,7 @@ mod tests {
     #[test]
     fn any_iter_to_array_decimal128() {
         let iter = vec![
-            Some(Value::Decimal(decimal_from_i128_with_scale(12345, 2)).into_any()),
+            Some(Value::Decimal(decimal_from_i128_with_scale(12345, 2).unwrap()).into_any()),
             None,
         ]
         .into_iter();
@@ -767,7 +771,7 @@ mod tests {
     fn any_iter_to_array_decimal128_scale_mismatch_is_null() {
         // Stored scale (2) != column scale (4): emit null rather than misread the mantissa.
         let iter = std::iter::once(Some(
-            Value::Decimal(decimal_from_i128_with_scale(12345, 2)).into_any(),
+            Value::Decimal(decimal_from_i128_with_scale(12345, 2).unwrap()).into_any(),
         ));
         let array = any_iter_to_array(iter, &DataType::Decimal128(10, 4)).unwrap();
         let dec = array.as_any().downcast_ref::<Decimal128Array>().unwrap();

datafusion_iceberg/src/statistics.rs

@@ -152,7 +152,7 @@ fn convert_value_to_scalar_value(value: Value, field_type: &Type) -> Result<Scal
                 }
             };
             Ok(ScalarValue::Decimal128(
-                Some(decimal_mantissa(&decimal)),
+                Some(decimal_mantissa(&decimal)?),
                 precision,
                 scale,
             ))
@@ -224,7 +224,7 @@ mod tests {
         });
 
         let scalar = convert_value_to_scalar_value(
-            Value::Decimal(decimal_from_i128_with_scale(mantissa, 0)),
+            Value::Decimal(decimal_from_i128_with_scale(mantissa, 0).unwrap()),
             &field_type,
         )
         .unwrap();

datafusion_iceberg/tests/roundtrip_types.rs

@@ -290,7 +290,7 @@ fn parquet_stats_and_partition_value_decode_correctly() {
         .flatten()
         .expect("partition value should have been inferred from stats");
 
-    let amount = Value::Decimal(decimal_from_i128_with_scale(amount_val, 2));
+    let amount = Value::Decimal(decimal_from_i128_with_scale(amount_val, 2).unwrap());
     assert_eq!(partition_value, amount);
 
     let uuid_val = Value::UUID(Uuid::parse_str(uuid_str).unwrap());

iceberg-rust-spec/src/spec/decimal.rs

@@ -1,41 +1,32 @@
 //! Decimal helpers for Iceberg's maximum 38-digit precision.
 
-use fastnum::{decimal::Context, D128};
+use fastnum::{
+    decimal::{Context, Sign},
+    D128, U128,
+};
 
 use crate::error::Error;
 
 /// Decimal representation capable of storing every Iceberg decimal value.
 pub type Decimal = D128;
 
 /// Creates a decimal from an unscaled value and scale.
-#[must_use]
-pub fn decimal_from_i128_with_scale(mantissa: i128, scale: u32) -> Decimal {
-    if scale == 0 {
-        return D128::from_i128(mantissa).expect("i128 always fits in D128");
-    }
-
-    let is_negative = mantissa < 0;
-    let digits = mantissa.unsigned_abs().to_string();
-    let scale = scale as usize;
-    let value = if digits.len() <= scale {
-        format!(
-            "{}0.{}{}",
-            if is_negative { "-" } else { "" },
-            "0".repeat(scale - digits.len()),
-            digits
-        )
+pub fn decimal_from_i128_with_scale(mantissa: i128, scale: u32) -> Result<Decimal, Error> {
+    let sign = if mantissa < 0 {
+        Sign::Minus
     } else {
-        let decimal_point = digits.len() - scale;
-        format!(
-            "{}{}.{}",
-            if is_negative { "-" } else { "" },
-            &digits[..decimal_point],
-            &digits[decimal_point..]
-        )
+        Sign::Plus
     };
-
-    D128::from_str(&value, Context::default())
-        .expect("a decimal assembled from an i128 and scale is valid")
+    let exponent = -i32::try_from(scale)?;
+    let magnitude = U128::from_u128(mantissa.unsigned_abs())
+        .map_err(|_| Error::Conversion(mantissa.to_string(), "decimal mantissa".to_string()))?;
+
+    Ok(D128::from_parts(
+        magnitude,
+        exponent,
+        sign,
+        Context::default(),
+    ))
 }
 
 /// Parses an exact decimal value.
@@ -45,20 +36,53 @@ pub fn decimal_from_str_exact(value: &str) -> Result<Decimal, Error> {
 }
 
 /// Returns the signed unscaled value.
-#[must_use]
-pub fn decimal_mantissa(decimal: &Decimal) -> i128 {
+pub fn decimal_mantissa(decimal: &Decimal) -> Result<i128, Error> {
     let magnitude = decimal
         .digits()
         .to_u128()
-        .expect("an Iceberg decimal has at most 38 digits");
-    let magnitude = i128::try_from(magnitude).expect("38 decimal digits fit in i128");
+        .map_err(|_| Error::Conversion(decimal.to_string(), "i128 decimal mantissa".to_string()))?;
+
     if decimal.is_sign_negative() {
-        -magnitude
+        if magnitude == i128::MIN.unsigned_abs() {
+            Ok(i128::MIN)
+        } else {
+            Ok(-i128::try_from(magnitude)?)
+        }
     } else {
-        magnitude
+        Ok(i128::try_from(magnitude)?)
     }
 }
 
+/// Encodes a decimal using the minimum-length big-endian two's-complement form.
+#[must_use]
+pub(crate) fn decimal_to_be_bytes_min(decimal: &Decimal) -> Vec<u8> {
+    let mut bytes = decimal.digits().to_radix_be(256);
+    if bytes.is_empty() || bytes.iter().all(|byte| *byte == 0) {
+        return vec![0];
+    }
+
+    if decimal.is_sign_negative() {
+        bytes.insert(0, 0);
+        bytes.iter_mut().for_each(|byte| *byte = !*byte);
+
+        for byte in bytes.iter_mut().rev() {
+            let (value, carry) = byte.overflowing_add(1);
+            *byte = value;
+            if !carry {
+                break;
+            }
+        }
+
+        if bytes[0] == 0xff && bytes[1] & 0x80 != 0 {
+            bytes.remove(0);
+        }
+    } else if bytes[0] & 0x80 != 0 {
+        bytes.insert(0, 0);
+    }
+
+    bytes
+}
+
 /// Returns the number of digits after the decimal point.
 #[must_use]
 pub fn decimal_scale(decimal: &Decimal) -> u32 {
@@ -101,10 +125,26 @@ mod tests {
 
     #[test]
     fn mantissa_and_scale_round_trip() {
-        let mantissa = -99_999_999_999_999_999_999_999_999_999_999_999_999_i128;
-        let decimal = decimal_from_i128_with_scale(mantissa, 7);
-        assert_eq!(decimal_mantissa(&decimal), mantissa);
-        assert_eq!(decimal_scale(&decimal), 7);
+        for (mantissa, scale) in [
+            (99_999_999_999_999_999_999_999_999_999_999_999_999_i128, 0),
+            (-99_999_999_999_999_999_999_999_999_999_999_999_999_i128, 7),
+            (1, 38),
+        ] {
+            let decimal = decimal_from_i128_with_scale(mantissa, scale).unwrap();
+            assert_eq!(decimal_mantissa(&decimal).unwrap(), mantissa);
+            assert_eq!(decimal_scale(&decimal), scale);
+        }
+    }
+
+    #[test]
+    fn rejects_scale_outside_i32() {
+        assert!(decimal_from_i128_with_scale(1, i32::MAX as u32 + 1).is_err());
+    }
+
+    #[test]
+    fn rejects_mantissa_outside_i128() {
+        let decimal = decimal_from_str_exact("170141183460469231731687303715884105728").unwrap();
+        assert!(decimal_mantissa(&decimal).is_err());
     }
 
     #[test]
@@ -114,4 +154,33 @@ mod tests {
         assert_eq!(i128_to_be_bytes_min(-128), vec![0x80]);
         assert_eq!(i128_to_be_bytes_min(-129), vec![0xff, 0x7f]);
     }
+
+    #[test]
+    fn decimal_big_endian_encoding_does_not_require_i128() {
+        let positive = decimal_from_str_exact("170141183460469231731687303715884105728").unwrap();
+        let negative = decimal_from_str_exact("-170141183460469231731687303715884105729").unwrap();
+
+        assert_eq!(
+            decimal_to_be_bytes_min(&positive),
+            vec![0, 0x80, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]
+        );
+        assert_eq!(
+            decimal_to_be_bytes_min(&negative),
+            vec![
+                0xff, 0x7f, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff,
+                0xff, 0xff, 0xff,
+            ]
+        );
+    }
+
+    #[test]
+    fn decimal_big_endian_encoding_matches_i128_encoding() {
+        for value in [i128::MIN, -129, -128, -1, 0, 1, 127, 128, i128::MAX] {
+            let decimal = decimal_from_i128_with_scale(value, 0).unwrap();
+            assert_eq!(
+                decimal_to_be_bytes_min(&decimal),
+                i128_to_be_bytes_min(value)
+            );
+        }
+    }
 }

iceberg-rust-spec/src/spec/values.rs

@@ -45,16 +45,13 @@ use uuid::Uuid;
 use crate::error::Error;
 
 use super::{
-    decimal::{
-        decimal_from_i128_with_scale, decimal_mantissa, decimal_scale, i128_to_be_bytes_min,
-        Decimal,
-    },
+    decimal::{decimal_from_i128_with_scale, decimal_scale, decimal_to_be_bytes_min, Decimal},
     partition::{PartitionField, Transform},
     types::{PrimitiveType, StructType, Type},
 };
 
 #[cfg(test)]
-use super::decimal::decimal_from_str_exact;
+use super::decimal::{decimal_from_str_exact, decimal_mantissa};
 
 pub static YEARS_BEFORE_UNIX_EPOCH: i32 = 1970;
 
@@ -139,7 +136,7 @@ impl From<Value> for ByteBuf {
             Value::UUID(val) => ByteBuf::from(val.as_u128().to_be_bytes()),
             Value::Fixed(_, val) => ByteBuf::from(val),
             Value::Binary(val) => ByteBuf::from(val),
-            Value::Decimal(val) => ByteBuf::from(i128_to_be_bytes_min(decimal_mantissa(&val))),
+            Value::Decimal(val) => ByteBuf::from(decimal_to_be_bytes_min(&val)),
             _ => todo!(),
         }
     }
@@ -529,7 +526,7 @@ impl Value {
                     } else {
                         return Err(Error::Type("decimal".to_string(), "bytes".to_string()));
                     };
-                    Ok(Value::Decimal(decimal_from_i128_with_scale(val, *scale)))
+                    Ok(Value::Decimal(decimal_from_i128_with_scale(val, *scale)?))
                 }
                 PrimitiveType::TimestampNs
                 | PrimitiveType::TimestamptzNs
@@ -1447,7 +1444,7 @@ mod tests {
     fn decimal_native_little_endian_hint_round_trips() {
         let decimal = decimal_from_str_exact("104899.50").unwrap();
         let value = Value::Decimal(decimal);
-        let bytes = i64::try_from(decimal_mantissa(&decimal))
+        let bytes = i64::try_from(decimal_mantissa(&decimal).unwrap())
             .unwrap()
             .to_le_bytes();
 
@@ -1819,7 +1816,7 @@ mod tests {
     fn test_identity_cast_returns_same_value_for_every_supported_primitive_variant() {
         // Same-type Value::cast is a no-op. Decimal datatype() hardcodes precision=38, so
         // the identity cast must target precision=38 too.
-        let dec_38_2 = decimal_from_i128_with_scale(1234, 2);
+        let dec_38_2 = decimal_from_i128_with_scale(1234, 2).unwrap();
         let cases = vec![
             Value::Boolean(true),
             Value::Int(123),
@@ -1961,7 +1958,7 @@ mod tests {
 
     #[test]
     fn test_decimal_value_rejects_every_non_decimal_target_type() {
-        let value = Value::Decimal(decimal_from_i128_with_scale(3411, 2));
+        let value = Value::Decimal(decimal_from_i128_with_scale(3411, 2).unwrap());
         // Decimal datatype() hardcodes precision=38 so identity uses precision=38; any other
         // decimal precision/scale variant is therefore "not the same type" but still allowed.
         let targets = all_other_primitive_types(&[PrimitiveType::Decimal {