Support Iceberg decimals with 38-digit precision

Cargo.toml

@@ -33,6 +33,7 @@ datafusion-sql = "55"
 derive-getters = "0.5.0"
 derive_builder = "0.20"
 futures = "0.3.31"
+fastnum = { version = "0.7", default-features = false, features = ["std", "serde"] }
 getrandom = { version = "0.4.2", features = ["std"] }
 itertools = "0.14.0"
 lazy_static = "1.5.0"
@@ -42,7 +43,6 @@ murmur3 = { version = "0.5.2" }
 parquet = { version = "59", features = ["async", "object_store", "variant_experimental"] }
 pin-project-lite = "0.2"
 regex = "1.11.1"
-rust_decimal = "1.42.0"
 serde = "^1.0"
 serde_derive = "^1.0"
 serde_json = "^1.0"

datafusion_iceberg/Cargo.toml

@@ -25,7 +25,6 @@ object_store = { workspace = true }
 parquet-variant-compute = "59"
 pin-project-lite = "0.2.17"
 regex = { workspace = true }
-rust_decimal = { workspace = true }
 serde_json = { workspace = true }
 thiserror = { workspace = true }
 tokio = { version = "1.50", features = ["rt-multi-thread", "sync"] }

datafusion_iceberg/src/pruning_statistics.rs

@@ -36,6 +36,7 @@ use iceberg_rust::{
     arrow::transform::transform_arrow,
     error::Error,
     spec::{
+        decimal::{decimal_mantissa, decimal_scale, Decimal},
         manifest::ManifestEntry,
         manifest_list::ManifestListEntry,
         partition::{BoundPartitionField, Transform},
@@ -44,7 +45,6 @@ use iceberg_rust::{
     },
     table::ManifestPath,
 };
-use rust_decimal::Decimal;
 
 pub(crate) struct PruneManifests<'table, 'manifests> {
     partition_fields: &'table [BoundPartitionField<'table>],
@@ -291,7 +291,7 @@ fn any_iter_to_array(
                 ScalarValue::Decimal128(
                     opt.and_then(|value| {
                         let d = *value.downcast::<Decimal>().ok()?;
-                        (d.scale() == scale as u32).then(|| d.mantissa())
+                        (decimal_scale(&d) == scale as u32).then(|| decimal_mantissa(&d))
                     }),
                     precision,
                     scale,
@@ -510,7 +510,7 @@ mod tests {
     };
     use datafusion::arrow::datatypes::Field;
     use datafusion::common::config::ConfigOptions;
-    use rust_decimal::Decimal;
+    use iceberg_rust::spec::decimal::decimal_from_i128_with_scale;
     use std::sync::Arc;
 
     /// Helper: invoke `DateTransform` directly with a transform name and scalar value.
@@ -751,7 +751,7 @@ mod tests {
     #[test]
     fn any_iter_to_array_decimal128() {
         let iter = vec![
-            Some(Value::Decimal(Decimal::new(12345, 2)).into_any()),
+            Some(Value::Decimal(decimal_from_i128_with_scale(12345, 2)).into_any()),
             None,
         ]
         .into_iter();
@@ -766,7 +766,9 @@ mod tests {
     #[test]
     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::new(12345, 2)).into_any()));
+        let iter = std::iter::once(Some(
+            Value::Decimal(decimal_from_i128_with_scale(12345, 2)).into_any(),
+        ));
         let array = any_iter_to_array(iter, &DataType::Decimal128(10, 4)).unwrap();
         let dec = array.as_any().downcast_ref::<Decimal128Array>().unwrap();
         assert!(dec.is_null(0));

datafusion_iceberg/src/statistics.rs

@@ -6,6 +6,7 @@ use datafusion::{
 use iceberg_rust::error::Error;
 use iceberg_rust::file_format::parquet::estimate_distinct_count;
 use iceberg_rust::spec::{
+    decimal::{decimal_mantissa, decimal_scale},
     manifest::{ManifestEntry, Status},
     schema::Schema,
     types::{PrimitiveType, Type},
@@ -147,11 +148,11 @@ fn convert_value_to_scalar_value(value: Value, field_type: &Type) -> Result<Scal
                 _ => {
                     // Fallback: use the decimal's own scale and assume max precision
                     // This matches the behavior in Value::datatype()
-                    (38, decimal.scale() as i8)
+                    (38, decimal_scale(&decimal) as i8)
                 }
             };
             Ok(ScalarValue::Decimal128(
-                Some(decimal.mantissa()),
+                Some(decimal_mantissa(&decimal)),
                 precision,
                 scale,
             ))
@@ -208,3 +209,26 @@ fn new_distinct_count(acc: &ColumnStatistics, x: &ColumnStatistics) -> Precision
         _ => acc.distinct_count.add(&x.distinct_count),
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use iceberg_rust::spec::decimal::decimal_from_i128_with_scale;
+
+    #[test]
+    fn converts_precision_38_decimal_bound_to_datafusion() {
+        let mantissa = 99_999_999_999_999_999_999_999_999_999_999_999_999_i128;
+        let field_type = Type::Primitive(PrimitiveType::Decimal {
+            precision: 38,
+            scale: 0,
+        });
+
+        let scalar = convert_value_to_scalar_value(
+            Value::Decimal(decimal_from_i128_with_scale(mantissa, 0)),
+            &field_type,
+        )
+        .unwrap();
+
+        assert_eq!(scalar, ScalarValue::Decimal128(Some(mantissa), 38, 0));
+    }
+}

datafusion_iceberg/tests/roundtrip_types.rs

@@ -20,6 +20,7 @@ use iceberg_rust::catalog::Catalog;
 use iceberg_rust::error::Error;
 use iceberg_rust::file_format::parquet::parquet_to_datafile;
 use iceberg_rust::object_store::ObjectStoreBuilder;
+use iceberg_rust::spec::decimal::decimal_from_i128_with_scale;
 use iceberg_rust::spec::manifest::DataFile;
 use iceberg_rust::spec::namespace::Namespace;
 use iceberg_rust::spec::partition::{BoundPartitionField, PartitionField, Transform};
@@ -30,7 +31,6 @@ use iceberg_rust::table::Table;
 use iceberg_sql_catalog::SqlCatalog;
 use parquet::arrow::ArrowWriter;
 use parquet::file::reader::{FileReader, SerializedFileReader};
-use rust_decimal::Decimal;
 use uuid::Uuid;
 
 /// Build an in-memory catalog with a single `public.t(id INT, amount DECIMAL(18,2))`
@@ -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));
     assert_eq!(partition_value, amount);
 
     let uuid_val = Value::UUID(Uuid::parse_str(uuid_str).unwrap());

iceberg-rust-spec/Cargo.toml

@@ -14,11 +14,11 @@ arrow-schema = { workspace = true }
 chrono = { workspace = true }
 derive-getters = { workspace = true }
 derive_builder = { workspace = true }
+fastnum = { workspace = true }
 getrandom = { workspace = true }
 itertools = { workspace = true }
 murmur3 = { workspace = true }
 ordered-float = { version = "5.3.0", features = ["serde"] }
-rust_decimal = { workspace = true }
 serde = { workspace = true }
 serde_bytes = "0.11.15"
 serde_derive = { workspace = true }

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

@@ -0,0 +1,117 @@
+//! Decimal helpers for Iceberg's maximum 38-digit precision.
+
+use fastnum::{decimal::Context, D128};
+
+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
+        )
+    } else {
+        let decimal_point = digits.len() - scale;
+        format!(
+            "{}{}.{}",
+            if is_negative { "-" } else { "" },
+            &digits[..decimal_point],
+            &digits[decimal_point..]
+        )
+    };
+
+    D128::from_str(&value, Context::default())
+        .expect("a decimal assembled from an i128 and scale is valid")
+}
+
+/// Parses an exact decimal value.
+pub fn decimal_from_str_exact(value: &str) -> Result<Decimal, Error> {
+    D128::from_str(value, Context::default())
+        .map_err(|_| Error::Conversion(value.to_string(), "decimal".to_string()))
+}
+
+/// Returns the signed unscaled value.
+#[must_use]
+pub fn decimal_mantissa(decimal: &Decimal) -> i128 {
+    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");
+    if decimal.is_sign_negative() {
+        -magnitude
+    } else {
+        magnitude
+    }
+}
+
+/// Returns the number of digits after the decimal point.
+#[must_use]
+pub fn decimal_scale(decimal: &Decimal) -> u32 {
+    decimal.fractional_digits_count().max(0) as u32
+}
+
+/// Encodes an i128 using the minimum-length big-endian two's-complement form.
+#[must_use]
+pub fn i128_to_be_bytes_min(value: i128) -> Vec<u8> {
+    let bytes = value.to_be_bytes();
+    let is_negative = value < 0;
+    let padding = if is_negative { 0xff } else { 0x00 };
+    let mut start = 0;
+
+    while start < bytes.len() - 1 && bytes[start] == padding {
+        let next_is_negative = bytes[start + 1] & 0x80 != 0;
+        if next_is_negative != is_negative {
+            break;
+        }
+        start += 1;
+    }
+
+    bytes[start..].to_vec()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn supports_iceberg_precision_38() {
+        for value in [
+            "99999999999999999999999999999999999999",
+            "-99999999999999999999999999999999999999",
+        ] {
+            let decimal = decimal_from_str_exact(value).unwrap();
+            assert_eq!(decimal.to_string(), value);
+        }
+    }
+
+    #[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);
+    }
+
+    #[test]
+    fn minimal_big_endian_encoding_preserves_sign() {
+        assert_eq!(i128_to_be_bytes_min(127), vec![0x7f]);
+        assert_eq!(i128_to_be_bytes_min(128), vec![0x00, 0x80]);
+        assert_eq!(i128_to_be_bytes_min(-128), vec![0x80]);
+        assert_eq!(i128_to_be_bytes_min(-129), vec![0xff, 0x7f]);
+    }
+}

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

@@ -13,6 +13,7 @@
 //! Each submodule implements a specific part of the specification, providing
 //! serialization/deserialization and validation logic.
 
+pub mod decimal;
 pub mod expressions;
 pub mod identifier;
 pub mod manifest;