From 446feb6534626d08ca29a9a3c1c170fd4b47ac73 Mon Sep 17 00:00:00 2001 From: osipovartem Date: Wed, 2 Sep 2026 15:08:24 +0300 Subject: [PATCH] Support Iceberg decimals with 38-digit precision --- Cargo.lock | 34 +++++- Cargo.toml | 2 +- datafusion_iceberg/Cargo.toml | 1 - datafusion_iceberg/src/pruning_statistics.rs | 12 +- datafusion_iceberg/src/statistics.rs | 28 ++++- datafusion_iceberg/tests/roundtrip_types.rs | 4 +- iceberg-rust-spec/Cargo.toml | 2 +- iceberg-rust-spec/src/spec/decimal.rs | 117 +++++++++++++++++++ iceberg-rust-spec/src/spec/mod.rs | 1 + iceberg-rust-spec/src/spec/values.rs | 54 ++++++--- 10 files changed, 223 insertions(+), 32 deletions(-) create mode 100644 iceberg-rust-spec/src/spec/decimal.rs diff --git a/Cargo.lock b/Cargo.lock index 8cd46bb6..5b969d3d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1373,6 +1373,16 @@ dependencies = [ "piper", ] +[[package]] +name = "bnum" +version = "0.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f781dba93de3a5ef6dc5b17c9958b208f6f3f021623b360fb605ea51ce443f10" +dependencies = [ + "serde", + "serde-big-array", +] + [[package]] name = "bollard" version = "0.19.4" @@ -2716,7 +2726,6 @@ dependencies = [ "regex", "reqwest", "rstest 0.26.1", - "rust_decimal", "serde_json", "sqlx", "tempfile", @@ -2980,6 +2989,18 @@ version = "0.1.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7360491ce676a36bf9bb3c56c1aa791658183a54d2744120f27285738d90465a" +[[package]] +name = "fastnum" +version = "0.7.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "020d1b59a944bc239d79903fbac2eda2365138b44890c27979562f6592059dcd" +dependencies = [ + "bnum", + "num-integer", + "num-traits", + "serde", +] + [[package]] name = "fastrand" version = "1.9.0" @@ -3804,12 +3825,12 @@ dependencies = [ "chrono", "derive-getters", "derive_builder", + "fastnum", "getrandom 0.4.2", "itertools 0.14.0", "murmur3", "ordered-float 5.3.0", "rstest 0.23.0", - "rust_decimal", "serde", "serde_bytes", "serde_derive", @@ -5810,6 +5831,15 @@ dependencies = [ "serde_derive", ] +[[package]] +name = "serde-big-array" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "11fc7cc2c76d73e0f27ee52abbd64eec84d46f370c88371120433196934e4b7f" +dependencies = [ + "serde", +] + [[package]] name = "serde_bytes" version = "0.11.19" diff --git a/Cargo.toml b/Cargo.toml index 3418d3bf..43a989fb 100644 --- a/Cargo.toml +++ b/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" diff --git a/datafusion_iceberg/Cargo.toml b/datafusion_iceberg/Cargo.toml index 7ec79a36..3ca514ef 100644 --- a/datafusion_iceberg/Cargo.toml +++ b/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"] } diff --git a/datafusion_iceberg/src/pruning_statistics.rs b/datafusion_iceberg/src/pruning_statistics.rs index 8a3c5eaf..4c9b5bba 100644 --- a/datafusion_iceberg/src/pruning_statistics.rs +++ b/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::().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::().unwrap(); assert!(dec.is_null(0)); diff --git a/datafusion_iceberg/src/statistics.rs b/datafusion_iceberg/src/statistics.rs index 41b5d5e9..36e6fc11 100644 --- a/datafusion_iceberg/src/statistics.rs +++ b/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 { // 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)); + } +} diff --git a/datafusion_iceberg/tests/roundtrip_types.rs b/datafusion_iceberg/tests/roundtrip_types.rs index ea9b8e35..60406e6f 100644 --- a/datafusion_iceberg/tests/roundtrip_types.rs +++ b/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()); diff --git a/iceberg-rust-spec/Cargo.toml b/iceberg-rust-spec/Cargo.toml index 502a686f..cb512c87 100644 --- a/iceberg-rust-spec/Cargo.toml +++ b/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 } diff --git a/iceberg-rust-spec/src/spec/decimal.rs b/iceberg-rust-spec/src/spec/decimal.rs new file mode 100644 index 00000000..747b4d90 --- /dev/null +++ b/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 { + 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 { + 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]); + } +} diff --git a/iceberg-rust-spec/src/spec/mod.rs b/iceberg-rust-spec/src/spec/mod.rs index e2c3312a..89ff5bf3 100644 --- a/iceberg-rust-spec/src/spec/mod.rs +++ b/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; diff --git a/iceberg-rust-spec/src/spec/values.rs b/iceberg-rust-spec/src/spec/values.rs index 0e2a55bc..66d97aa3 100644 --- a/iceberg-rust-spec/src/spec/values.rs +++ b/iceberg-rust-spec/src/spec/values.rs @@ -33,7 +33,6 @@ use datetime::{ }; use itertools::Itertools; use ordered_float::OrderedFloat; -use rust_decimal::Decimal; use serde::{ de::{MapAccess, Visitor}, ser::SerializeStruct, @@ -46,10 +45,17 @@ 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, + }, partition::{PartitionField, Transform}, types::{PrimitiveType, StructType, Type}, }; +#[cfg(test)] +use super::decimal::decimal_from_str_exact; + pub static YEARS_BEFORE_UNIX_EPOCH: i32 = 1970; /// How the bytes passed to [`Value::try_from_bytes`] are physically encoded, @@ -133,12 +139,7 @@ impl From 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) => { - // rust_decimal mantissa is 96 bits - // so we can remove the first 32 bits of the i128 representation - let bytes = val.mantissa().to_be_bytes()[4..].to_vec(); - ByteBuf::from(bytes) - } + Value::Decimal(val) => ByteBuf::from(i128_to_be_bytes_min(decimal_mantissa(&val))), _ => todo!(), } } @@ -528,7 +529,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 @@ -719,7 +720,7 @@ impl Value { Value::UUID(_) => Type::Primitive(PrimitiveType::Uuid), Value::Decimal(dec) => Type::Primitive(PrimitiveType::Decimal { precision: 38, - scale: dec.scale(), + scale: decimal_scale(dec), }), _ => unimplemented!(), } @@ -1413,15 +1414,12 @@ mod tests { #[test] fn avro_bytes_decimal() { - let value = Value::Decimal(Decimal::from_str_exact("104899.50").unwrap()); + let value = Value::Decimal(decimal_from_str_exact("104899.50").unwrap()); // Test serialization let byte_buf: ByteBuf = value.clone().into(); let bytes: Vec = byte_buf.into_vec(); - assert_eq!( - bytes, - vec![0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 160u8, 16u8, 94u8] - ); + assert_eq!(bytes, vec![0u8, 160u8, 16u8, 94u8]); // Test deserialization check_avro_bytes_serde( @@ -1447,9 +1445,11 @@ mod tests { #[test] fn decimal_native_little_endian_hint_round_trips() { - let decimal = Decimal::from_str_exact("104899.50").unwrap(); + let decimal = decimal_from_str_exact("104899.50").unwrap(); let value = Value::Decimal(decimal); - let bytes = (decimal.mantissa() as i64).to_le_bytes(); + let bytes = i64::try_from(decimal_mantissa(&decimal)) + .unwrap() + .to_le_bytes(); let decoded = Value::try_from_bytes_with_hint( &bytes, @@ -1463,6 +1463,24 @@ mod tests { assert_eq!(decoded, value); } + #[test] + fn decimal_precision_38_spec_bytes_round_trip() { + let decimal_type = Type::Primitive(PrimitiveType::Decimal { + precision: 38, + scale: 0, + }); + + for text in [ + "99999999999999999999999999999999999999", + "-99999999999999999999999999999999999999", + ] { + let value = Value::Decimal(decimal_from_str_exact(text).unwrap()); + let bytes: ByteBuf = value.clone().into(); + let decoded = Value::try_from_bytes(&bytes, &decimal_type).unwrap(); + assert_eq!(decoded, value); + } + } + #[test] fn uuid_byte_array_hint_round_trips() { let uuid = Uuid::parse_str("f79c3e09-677c-4bbd-a479-3f349cb785e7").unwrap(); @@ -1801,7 +1819,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); let cases = vec![ Value::Boolean(true), Value::Int(123), @@ -1943,7 +1961,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)); // 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 {