fix(spec): validate decimal mantissas before conversion

crates/iceberg/src/spec/values/datum.rs

@@ -29,7 +29,7 @@ use serde::{Deserialize, Serialize};
 use serde_bytes::ByteBuf;
 
 use super::decimal_utils::{
-    Decimal, decimal_from_i128_with_scale, decimal_from_str_exact, decimal_mantissa,
+    Decimal, decimal_from_i128_with_scale, decimal_from_str_exact, decimal_mantissa_at_scale,
     decimal_precision, decimal_scale, i128_from_be_bytes, i128_to_be_bytes_min,
 };
 use super::literal::Literal;
@@ -1029,18 +1029,7 @@ impl Datum {
     /// assert_eq!(&format!("{t}"), "1.23");
     /// ```
     pub fn decimal(value: Decimal) -> Result<Self> {
-        ensure_data_valid!(value.is_finite(), "Decimal value must be finite: {}", value);
-        let scale = decimal_scale(&value);
-
-        let r#type = Type::decimal(MAX_DECIMAL_PRECISION, scale)?;
-        if let Type::Primitive(p) = r#type {
-            Ok(Self {
-                r#type: p,
-                literal: PrimitiveLiteral::Int128(decimal_mantissa(&value)),
-            })
-        } else {
-            unreachable!("Decimal type must be primitive.")
-        }
+        Self::decimal_with_precision(value, MAX_DECIMAL_PRECISION)
     }
 
     /// Try to create a decimal literal from [`Decimal`] with precision.
@@ -1049,11 +1038,12 @@ impl Datum {
     /// which is useful when you need to control the storage requirements.
     /// Use [`Datum::decimal`] if you want to use the maximum precision (38).
     pub fn decimal_with_precision(value: Decimal, precision: u32) -> Result<Self> {
-        ensure_data_valid!(value.is_finite(), "Decimal value must be finite: {}", value);
         let scale = decimal_scale(&value);
-        let mantissa = decimal_mantissa(&value);
-
-        Self::decimal_from_mantissa(mantissa, precision, scale)
+        let Type::Primitive(data_type) = Type::decimal(precision, scale)? else {
+            unreachable!("Decimal type must be primitive.");
+        };
+        let mantissa = decimal_mantissa_at_scale(value, scale, precision)?;
+        Ok(Self::new(data_type, PrimitiveLiteral::Int128(mantissa)))
     }
 
     fn i64_to_i32<T: Into<i64> + PartialOrd<i64>>(val: T) -> Datum {

crates/iceberg/src/spec/values/decimal_utils.rs

@@ -24,6 +24,7 @@ use fastnum::D128;
 use fastnum::decimal::Context;
 
 use crate::error::invalid_data;
+use crate::spec::MAX_DECIMAL_PRECISION;
 use crate::{Result, ensure_data_valid};
 
 /// Re-export D128 as the Decimal type for use throughout the crate.
@@ -103,28 +104,49 @@ pub fn decimal_from_str_exact(s: &str) -> Result<Decimal> {
     Ok(decimal)
 }
 
-/// Get the mantissa (unscaled coefficient) as i128.
-///
-/// This is equivalent to rust_decimal's `decimal.mantissa()`.
-///
-/// The mantissa is signed: negative decimals return negative mantissa.
-pub fn decimal_mantissa(d: &Decimal) -> i128 {
-    // digits() returns unsigned coefficient as UInt<N>
-    // For Iceberg decimals (max 38 digits), this always fits in u128/i128
-    let digits = d.digits();
-
-    // Convert UInt<2> to u128 - this always succeeds for Iceberg-compliant decimals
-    // since 38 digits requires ~127 bits and u128 has 128 bits
-    let unsigned: u128 = digits
+/// Returns the signed coefficient at the value's non-negative scale.
+pub(crate) fn decimal_mantissa(value: Decimal) -> Result<i128> {
+    decimal_mantissa_at_scale(value, decimal_scale(&value), MAX_DECIMAL_PRECISION)
+}
+
+/// Returns the signed coefficient at the requested scale and precision.
+pub(super) fn decimal_mantissa_at_scale(
+    mut value: Decimal,
+    scale: u32,
+    precision: u32,
+) -> Result<i128> {
+    ensure_data_valid!(value.is_finite(), "Decimal value must be finite: {}", value);
+
+    let shift = i64::from(scale) - i64::from(value.fractional_digits_count());
+    if shift < 0 {
+        value = value.rescale(scale as i16);
+    }
+    if value.is_zero() {
+        return Ok(0);
+    }
+
+    let appended_digits = shift.max(0);
+    // Check before multiplying so the expanded coefficient fits in i128.
+    ensure_data_valid!(
+        value.digits_count() as i64 + appended_digits
+            <= i64::from(precision.min(MAX_DECIMAL_PRECISION)),
+        "Decimal value is too large for precision {} at scale {}",
+        precision,
+        scale
+    );
+    let coefficient = value
+        .digits()
         .to_u128()
-        .expect("Iceberg decimals (max 38 digits) always fit in u128");
+        .ok()
+        .and_then(|digits| i128::try_from(digits).ok())
+        .ok_or_else(|| invalid_data!("Decimal coefficient does not fit in i128"))?;
+    let mantissa = coefficient * 10i128.pow(appended_digits as u32);
 
-    let signed = unsigned as i128;
-    if d.is_sign_negative() {
-        -signed
+    Ok(if value.is_sign_negative() {
+        -mantissa
     } else {
-        signed
-    }
+        mantissa
+    })
 }
 
 /// Get the decimal digit precision of a mantissa.
@@ -145,13 +167,6 @@ pub fn decimal_scale(d: &Decimal) -> u32 {
     if frac < 0 { 0 } else { frac as u32 }
 }
 
-/// Rescale a decimal to the given scale, returning the rescaled value.
-///
-/// This is equivalent to rust_decimal's `decimal.rescale(scale)`.
-pub fn decimal_rescale(d: Decimal, scale: u32) -> Decimal {
-    d.rescale(scale as i16)
-}
-
 /// Convert big-endian signed bytes to i128.
 ///
 /// This handles variable-length byte arrays (up to 16 bytes) with sign extension.
@@ -270,10 +285,10 @@ mod tests {
     #[test]
     fn test_decimal_mantissa() {
         let d = decimal_from_i128_with_scale(12345, 2);
-        assert_eq!(decimal_mantissa(&d), 12345);
+        assert_eq!(decimal_mantissa(d).unwrap(), 12345);
 
         let d = decimal_from_i128_with_scale(-12345, 2);
-        assert_eq!(decimal_mantissa(&d), -12345);
+        assert_eq!(decimal_mantissa(d).unwrap(), -12345);
     }
 
     #[test]
@@ -315,9 +330,7 @@ mod tests {
     #[test]
     fn test_decimal_rescale() {
         let d = decimal_from_str_exact("123.45").unwrap();
-        let rescaled = decimal_rescale(d, 4);
-        assert_eq!(decimal_scale(&rescaled), 4);
-        assert_eq!(decimal_mantissa(&rescaled), 1234500);
+        assert_eq!(decimal_mantissa_at_scale(d, 4, 7).unwrap(), 1234500);
     }
 
     #[test]

crates/iceberg/src/spec/values/literal.rs

@@ -27,7 +27,8 @@ use uuid::Uuid;
 
 use super::Map;
 use super::decimal_utils::{
-    decimal_from_str_exact, decimal_mantissa, decimal_rescale, try_decimal_from_i128_with_scale,
+    decimal_from_str_exact, decimal_mantissa, decimal_mantissa_at_scale,
+    try_decimal_from_i128_with_scale,
 };
 use super::primitive::PrimitiveLiteral;
 use super::struct_value::Struct;
@@ -391,7 +392,7 @@ impl Literal {
     /// ```
     pub fn decimal_from_str<S: AsRef<str>>(s: S) -> Result<Self> {
         let decimal = decimal_from_str_exact(s.as_ref())?;
-        Ok(Self::decimal(decimal_mantissa(&decimal)))
+        Ok(Self::decimal(decimal_mantissa(decimal)?))
     }
 
     /// Attempts to convert the Literal to a PrimitiveLiteral
@@ -515,18 +516,10 @@ impl Literal {
                     ErrorKind::FeatureUnsupported,
                     "WKT deserialization for geospatial types is not yet supported",
                 )),
-                (
-                    PrimitiveType::Decimal {
-                        precision: _,
-                        scale,
-                    },
-                    JsonValue::String(s),
-                ) => {
+                (PrimitiveType::Decimal { precision, scale }, JsonValue::String(s)) => {
                     let decimal = decimal_from_str_exact(&s)?;
-                    let rescaled = decimal_rescale(decimal, *scale);
-                    Ok(Some(Literal::Primitive(PrimitiveLiteral::Int128(
-                        decimal_mantissa(&rescaled),
-                    ))))
+                    let mantissa = decimal_mantissa_at_scale(decimal, *scale, *precision)?;
+                    Ok(Some(Literal::decimal(mantissa)))
                 }
                 (PrimitiveType::Unknown, value) if !value.is_null() => Err(invalid_data!(
                     "Unknown type only supports null default values"

crates/iceberg/src/spec/values/tests.rs

@@ -1995,6 +1995,92 @@ fn test_decimal_constructors_accept_finite_values() {
     }
 }
 
+#[test]
+fn test_decimal_positive_exponents() {
+    for (input, mantissa, scale) in [
+        ("1e3", 1000, 0),
+        ("-1.25e3", -1250, 0),
+        ("1.2300e2", 12300, 2),
+        ("1e-3", 1, 3),
+        ("-0e1000", 0, 0),
+        ("9e37", 9 * 10i128.pow(37), 0),
+    ] {
+        let expected = Datum::new(
+            PrimitiveType::Decimal {
+                precision: 38,
+                scale,
+            },
+            PrimitiveLiteral::Int128(mantissa),
+        );
+        let value = Decimal::from_str(input, fastnum::decimal::Context::default()).unwrap();
+        assert_eq!(Datum::decimal_from_str(input).unwrap(), expected, "{input}");
+        assert_eq!(Datum::decimal(value).unwrap(), expected, "{input}");
+        assert_eq!(
+            Literal::decimal_from_str(input).unwrap(),
+            Literal::decimal(mantissa),
+            "{input}"
+        );
+    }
+
+    let value = Decimal::from_str("1e3", fastnum::decimal::Context::default()).unwrap();
+    assert_eq!(
+        Datum::decimal_with_precision(value, 4).unwrap(),
+        Datum::new(
+            PrimitiveType::Decimal {
+                precision: 4,
+                scale: 0,
+            },
+            PrimitiveLiteral::Int128(1000),
+        )
+    );
+    assert_eq!(
+        Datum::decimal_with_precision(value, 3).unwrap_err().kind(),
+        ErrorKind::DataInvalid
+    );
+}
+
+#[test]
+fn test_decimal_rejects_oversized_mantissa() {
+    for input in [
+        "1e38",
+        "-1e1000",
+        "100000000000000000000000000000000000000",
+        "170141183460469231731687303715884105729",
+        "-170141183460469231731687303715884105728",
+    ] {
+        let value = Decimal::from_str(input, fastnum::decimal::Context::default()).unwrap();
+        for kind in [
+            Datum::decimal_from_str(input).unwrap_err().kind(),
+            Datum::decimal(value).unwrap_err().kind(),
+            Datum::decimal_with_precision(value, 38).unwrap_err().kind(),
+            Literal::decimal_from_str(input).unwrap_err().kind(),
+        ] {
+            assert_eq!(kind, ErrorKind::DataInvalid, "{input}");
+        }
+    }
+}
+
+#[test]
+fn test_decimal_from_json_validates_precision_at_target_scale() {
+    let decimal_type = Type::decimal(6, 2).unwrap();
+    for (input, mantissa) in [("1e3", 100000), ("1.234", 123), ("-0e1000", 0)] {
+        assert_eq!(
+            Literal::try_from_json(JsonValue::String(input.to_string()), &decimal_type).unwrap(),
+            Some(Literal::decimal(mantissa)),
+            "{input}"
+        );
+    }
+    for input in ["1e4", "9999.999", "1e1000"] {
+        assert_eq!(
+            Literal::try_from_json(JsonValue::String(input.to_string()), &decimal_type)
+                .unwrap_err()
+                .kind(),
+            ErrorKind::DataInvalid,
+            "{input}"
+        );
+    }
+}
+
 #[test]
 fn test_datum_to_decimal_narrows_precision_when_scale_matches() {
     let target_type = Primitive(PrimitiveType::Decimal {

crates/iceberg/src/writer/file_writer/parquet_writer.rs

@@ -1849,8 +1849,8 @@ mod tests {
             .await?;
         let col0 = Arc::new(
             Decimal128Array::from(vec![
-                Some(decimal_mantissa(&decimal_max)),
-                Some(decimal_mantissa(&decimal_min)),
+                Some(decimal_mantissa(decimal_max).unwrap()),
+                Some(decimal_mantissa(decimal_min).unwrap()),
             ])
             .with_data_type(DataType::Decimal128(38, 0)),
         ) as ArrayRef;