This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new d33082c6 Align decimal schema evolution with Java (#1118)
d33082c6 is described below
commit d33082c6bc806cfedc5089574e4c28284fa217ed
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Oct 13 14:46:00 2026 +0800
Align decimal schema evolution with Java (#1118)
---
crates/paimon/src/arrow/evolution_decimal.rs | 340 +++++++++++++++++++
crates/paimon/src/arrow/mod.rs | 1 +
crates/paimon/src/arrow/nested_evolution.rs | 5 +
crates/paimon/src/common/mod.rs | 1 +
crates/paimon/src/common/number.rs | 375 +++++++++++++++++++++
crates/paimon/src/spec/partition_utils.rs | 85 +----
.../paimon/tests/decimal_schema_evolution_test.rs | 299 ++++++++++++++++
7 files changed, 1022 insertions(+), 84 deletions(-)
diff --git a/crates/paimon/src/arrow/evolution_decimal.rs
b/crates/paimon/src/arrow/evolution_decimal.rs
new file mode 100644
index 00000000..548bf37a
--- /dev/null
+++ b/crates/paimon/src/arrow/evolution_decimal.rs
@@ -0,0 +1,340 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Java decimal conversions for existing Python-supported schema changes.
+
+use std::sync::Arc;
+
+use arrow_array::{
+ cast::AsArray,
+ types::{Float32Type, Float64Type},
+ ArrayRef, Decimal128Array,
+};
+
+use crate::common::number::java_double_decimal_text;
+use crate::spec::DataType;
+
+/// Match NumericPrimitiveToDecimalCastRule and StringToDecimalCastRule.
+/// Java first builds a BigDecimal from decimal text, rounds HALF_UP, then
+/// checks precision. Multiplying the binary float by 10^scale loses digits.
+pub(super) fn cast_decimal(
+ source: &ArrayRef,
+ source_type: &DataType,
+ target_type: &DataType,
+) -> crate::Result<Option<ArrayRef>> {
+ let DataType::Decimal(target) = target_type else {
+ return Ok(None);
+ };
+ let precision = target.precision();
+ let scale = target.scale();
+ let values = match source_type {
+ DataType::Float(_) => source
+ .as_primitive::<Float32Type>()
+ .iter()
+ // Java Number.doubleValue expands FLOAT before Double.toString.
+ .map(|v| {
+ v.map(|v| float_decimal(f64::from(v), precision, scale))
+ .transpose()
+ })
+ .collect::<crate::Result<Vec<_>>>()?,
+ DataType::Double(_) => source
+ .as_primitive::<Float64Type>()
+ .iter()
+ .map(|v| v.map(|v| float_decimal(v, precision, scale)).transpose())
+ .collect::<crate::Result<Vec<_>>>()?,
+ DataType::Char(_) | DataType::VarChar(_) => source
+ .as_string::<i32>()
+ .iter()
+ .map(|v| v.map(|v| decimal_text(v, precision, scale)).transpose())
+ .collect::<crate::Result<Vec<_>>>()?,
+ _ => return Ok(None),
+ };
+ let values = values.into_iter().map(Option::flatten);
+ let result = Decimal128Array::from_iter(values)
+ .with_precision_and_scale(precision as u8, scale as i8)
+ .map_err(|e| invalid(e.to_string()))?;
+ Ok(Some(Arc::new(result)))
+}
+
+fn float_decimal(value: f64, precision: u32, scale: u32) ->
crate::Result<Option<i128>> {
+ if !value.is_finite() {
+ return Err(invalid("Non-finite decimal schema-evolution value"));
+ }
+ decimal_text(&java_double_decimal_text(value), precision, scale)
+}
+
+/// Parse only the digits needed by the target, so a long mantissa or exponent
+/// never requires a large power of ten. Invalid text is an error; a valid
+/// number exceeding target precision returns NULL, as Decimal.fromBigDecimal.
+fn decimal_text(text: &str, precision: u32, scale: u32) ->
crate::Result<Option<i128>> {
+ let invalid_text = || invalid(format!("Invalid decimal schema-evolution
value: {text}"));
+ let (mantissa, exponent) = text
+ .split_once(['e', 'E'])
+ .map_or(Ok((text, 0)), |(m, e)| {
+ e.parse::<i64>().map(|e| (m, e)).map_err(|_| invalid_text())
+ })?;
+ let negative = mantissa.starts_with('-');
+ let mantissa = mantissa.strip_prefix(['-', '+']).unwrap_or(mantissa);
+ let (whole, fraction) = mantissa.split_once('.').unwrap_or((mantissa, ""));
+ if whole.is_empty() && fraction.is_empty()
+ || !whole
+ .bytes()
+ .chain(fraction.bytes())
+ .all(|b| b.is_ascii_digit())
+ {
+ return Err(invalid_text());
+ }
+ // BigDecimal checks the adjusted input scale, rather than the exponent
+ // alone: 0e2147483648 is valid, but 0e-2147483648 is not.
+ let input_scale = (fraction.len() as i64)
+ .checked_sub(exponent)
+ .and_then(|scale| i32::try_from(scale).ok())
+ .ok_or_else(invalid_text)?;
+ let digits = format!("{whole}{fraction}");
+ let digits = digits.trim_start_matches('0');
+ if digits.is_empty() {
+ return Ok(Some(0));
+ }
+ let shift = i64::from(scale) - i64::from(input_scale);
+ let power = shift.abs();
+ // Match setScale's checkScale and BigInteger.TEN.pow supported-range
+ // check. Java evaluates this before target precision, except for zero.
+ let max_magnitude_length = i64::from(i32::MAX) / 32 + 1;
+ if power > i64::from(i32::MAX) || 4 * power / 32 > max_magnitude_length {
+ return Err(invalid(
+ "Decimal scale conversion exceeds Java's supported range",
+ ));
+ }
+ let kept = digits.len() as i64 + shift.min(0);
+ if kept + shift.max(0) > i64::from(precision) {
+ return Ok(None);
+ }
+ let (mut value, round) = if kept > 0 {
+ let kept = kept as usize;
+ (
+ digits[..kept].parse::<i128>().map_err(|_| invalid_text())?,
+ digits.as_bytes().get(kept).is_some_and(|b| *b >= b'5'),
+ )
+ } else {
+ (0, kept == 0 && digits.as_bytes()[0] >= b'5')
+ };
+ if shift > 0 {
+ value *= 10_i128.pow(shift as u32);
+ }
+ value += i128::from(round);
+ if value >= 10_i128.pow(precision) {
+ return Ok(None);
+ }
+ Ok(Some(if negative { -value } else { value }))
+}
+
+fn invalid(message: impl Into<String>) -> crate::Error {
+ crate::Error::DataInvalid {
+ message: message.into(),
+ source: None,
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use std::sync::Arc;
+
+ use arrow_array::{ArrayRef, Decimal128Array, Float32Array, Float64Array,
StringArray};
+
+ use crate::arrow::nested_evolution::evolve_column;
+ use crate::spec::{DataType, DecimalType, DoubleType, FloatType,
VarCharType};
+
+ fn decimals(source: &ArrayRef, from: DataType, precision: u32, scale: u32)
-> Decimal128Array {
+ evolve_column(
+ source,
+ &from,
+ &DataType::Decimal(DecimalType::new(precision, scale).unwrap()),
+ )
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Decimal128Array>()
+ .unwrap()
+ .clone()
+ }
+
+ #[test]
+ fn doubles_use_java_decimal_text_before_half_up_rounding() {
+ let source: ArrayRef = Arc::new(Float64Array::from(vec![
+ None,
+ Some(1.005),
+ Some(-1.005),
+ Some(2.675),
+ Some(-2.675),
+ Some(0.145),
+ Some(999.995),
+ Some(1e30),
+ None,
+ Some(0.0),
+ ]));
+ let actual = decimals(
+ &source.slice(1, 8),
+ DataType::Double(DoubleType::new()),
+ 5,
+ 2,
+ );
+ assert_eq!(
+ actual,
+ Decimal128Array::from(vec![
+ Some(101),
+ Some(-101),
+ Some(268),
+ Some(-268),
+ Some(15),
+ None,
+ None,
+ None,
+ ])
+ .with_precision_and_scale(5, 2)
+ .unwrap()
+ );
+ }
+
+ #[test]
+ fn
decimal_digits_match_java_8_11_and_17_rather_than_modern_shortest_text() {
+ let value = f64::from_bits(0xc381_e88f_38c0_c8fd);
+ let source: ArrayRef = Arc::new(Float64Array::from(vec![Some(value),
Some(-value), None]));
+ assert_eq!(
+ decimals(&source, DataType::Double(DoubleType::new()), 38, 20),
+ Decimal128Array::from(vec![
+ Some(-16_130_484_589_462_313_600_000_000_000_000_000_000),
+ Some(16_130_484_589_462_313_600_000_000_000_000_000_000),
+ None,
+ ])
+ .with_precision_and_scale(38, 20)
+ .unwrap()
+ );
+ let source: ArrayRef = Arc::new(Float64Array::from(vec![1e23]));
+ assert_eq!(
+ decimals(&source, DataType::Double(DoubleType::new()), 38, 6),
+
Decimal128Array::from(vec![99_999_999_999_999_990_000_000_000_000_i128])
+ .with_precision_and_scale(38, 6)
+ .unwrap()
+ );
+ }
+
+ #[test]
+ fn float_is_promoted_to_double_before_decimal_text_conversion() {
+ let source: ArrayRef =
+ Arc::new(Float32Array::from(vec![Some(1.005), Some(1.2345679),
None]));
+ let actual = decimals(&source, DataType::Float(FloatType::new()), 38,
20);
+ assert_eq!(
+ actual,
+ Decimal128Array::from(vec![
+ Some(100_499_999_523_162_840_000),
+ Some(123_456_788_063_049_320_000),
+ None,
+ ])
+ .with_precision_and_scale(38, 20)
+ .unwrap()
+ );
+ }
+
+ #[test]
+ fn doubles_keep_decimal_significant_digits_at_high_scale() {
+ let source: ArrayRef = Arc::new(Float64Array::from(vec![
+ Some(1.005),
+ Some(1.2345678901234567),
+ Some(f64::MIN_POSITIVE),
+ None,
+ ]));
+ let actual = decimals(&source, DataType::Double(DoubleType::new()),
38, 20);
+ assert_eq!(
+ actual,
+ Decimal128Array::from(vec![
+ Some(100_500_000_000_000_000_000),
+ Some(123_456_789_012_345_670_000),
+ Some(0),
+ None,
+ ])
+ .with_precision_and_scale(38, 20)
+ .unwrap()
+ );
+ }
+
+ #[test]
+ fn decimal_text_rounds_and_nulls_overflow_after_rounding() {
+ let source: ArrayRef = Arc::new(StringArray::from(vec![
+ "1.005", "-1.005", "+.005", "2.675", "1e-2", "999.995", "1e100",
+ ]));
+ assert_eq!(
+ decimals(&source, DataType::VarChar(VarCharType::string_type()),
5, 2),
+ Decimal128Array::from(vec![
+ Some(101),
+ Some(-101),
+ Some(1),
+ Some(268),
+ Some(1),
+ None,
+ None
+ ])
+ .with_precision_and_scale(5, 2)
+ .unwrap()
+ );
+ }
+
+ #[test]
+ fn invalid_decimal_text_is_an_error_instead_of_silent_null() {
+ for text in [
+ "",
+ " 1.2 ",
+ "1_2",
+ "1.2.3",
+ "1e",
+ "Infinity",
+ "0e-2147483648",
+ "0.1e-2147483647",
+ "1e2147483647",
+ "1e-2147483647",
+ "1e9999999999999999999",
+ ] {
+ let source: ArrayRef = Arc::new(StringArray::from(vec![None,
Some(text)]));
+ assert!(
+ evolve_column(
+ &source,
+ &DataType::VarChar(VarCharType::string_type()),
+ &DataType::Decimal(DecimalType::new(5, 2).unwrap())
+ )
+ .is_err(),
+ "{text:?}"
+ );
+ }
+ }
+
+ #[test]
+ fn
decimal_text_handles_long_mantissas_and_zero_exponents_without_large_powers() {
+ let long = format!("1.005{}", "0".repeat(200));
+ let tiny = format!("0.{}1", "0".repeat(200));
+ let source: ArrayRef = Arc::new(StringArray::from(vec![
+ long.as_str(),
+ tiny.as_str(),
+ "0e2147483648",
+ "0.0e2147483648",
+ "0e2147483647",
+ ]));
+ assert_eq!(
+ decimals(&source, DataType::VarChar(VarCharType::string_type()),
5, 2),
+ Decimal128Array::from(vec![101, 0, 0, 0, 0])
+ .with_precision_and_scale(5, 2)
+ .unwrap()
+ );
+ }
+}
diff --git a/crates/paimon/src/arrow/mod.rs b/crates/paimon/src/arrow/mod.rs
index fbd9ac27..34140b97 100644
--- a/crates/paimon/src/arrow/mod.rs
+++ b/crates/paimon/src/arrow/mod.rs
@@ -16,6 +16,7 @@
// under the License.
mod evolution_cast;
+mod evolution_decimal;
mod evolution_string;
mod evolution_temporal;
pub(crate) mod filtering;
diff --git a/crates/paimon/src/arrow/nested_evolution.rs
b/crates/paimon/src/arrow/nested_evolution.rs
index 8e2d3e75..20e5d33e 100644
--- a/crates/paimon/src/arrow/nested_evolution.rs
+++ b/crates/paimon/src/arrow/nested_evolution.rs
@@ -204,6 +204,11 @@ pub(crate) fn evolve_column(
_ => {}
}
+ if let Some(result) = super::evolution_decimal::cast_decimal(source,
source_type, target_type)?
+ {
+ return Ok(result);
+ }
+
if let Some(result) =
super::evolution_string::temporal_to_string(source, source_type,
target_type)?
{
diff --git a/crates/paimon/src/common/mod.rs b/crates/paimon/src/common/mod.rs
index 5376e76d..543fb6fa 100644
--- a/crates/paimon/src/common/mod.rs
+++ b/crates/paimon/src/common/mod.rs
@@ -18,6 +18,7 @@
//! Common utilities for Paimon.
pub(crate) mod date_time;
+pub(crate) mod number;
pub mod options;
pub use options::{CatalogOptions, Options};
diff --git a/crates/paimon/src/common/number.rs
b/crates/paimon/src/common/number.rs
new file mode 100644
index 00000000..cc30c512
--- /dev/null
+++ b/crates/paimon/src/common/number.rs
@@ -0,0 +1,375 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Java numeric text shared by partition paths and value conversions.
+
+use std::cmp::Ordering;
+
+use num_bigint::BigUint;
+use num_traits::ToPrimitive;
+
+/// Numeric digits selected by Java 8/11/17 Double.toString, for
BigDecimal.valueOf.
+/// Its integral shortcut retains digits that the JDK 19+ shortest converter
+/// discards. The remaining cases generate digits from the IEEE-754 interval;
+/// fixed-width and arbitrary-precision arithmetic use different upper
endpoints.
+/// The result is coefficient/exponent text, rather than a display format.
+pub(crate) fn java_double_decimal_text(value: f64) -> String {
+ debug_assert!(value.is_finite());
+ if value == 0.0 {
+ return "0".into();
+ }
+ let bits = value.to_bits();
+ let sign = if bits >> 63 != 0 { "-" } else { "" };
+ let mut exponent = ((bits >> 52) & 2047) as i32 - 1023;
+ let mut mantissa = bits & ((1_u64 << 52) - 1);
+ let significant = if exponent == -1023 {
+ let significant = 64 - mantissa.leading_zeros() as i32;
+ let normalize = 53 - significant;
+ mantissa <<= normalize;
+ exponent = -1022 - normalize;
+ significant
+ } else {
+ mantissa |= 1_u64 << 52;
+ 53
+ };
+ let trailing = mantissa.trailing_zeros() as i32;
+ let width = 53 - trailing;
+ let tiny = (width - exponent - 1).max(0);
+ if (-21..=62).contains(&exponent) && tiny == 0 {
+ let integer = if exponent >= 52 {
+ mantissa << (exponent - 52)
+ } else {
+ mantissa >> (52 - exponent)
+ };
+ let discarded = if exponent > significant {
+ (1_u64 << (exponent - significant - 1)).ilog10()
+ } else {
+ 0
+ };
+ let factor = 10_u64.pow(discarded);
+ return format!("{sign}{}e{discarded}", (integer + factor / 2) /
factor);
+ }
+
+ // Approximate log10 from a normalized binary significand. A leading zero
+ // corrects an overestimate while keeping the original rounding interval.
+ let normalized = mantissa as f64 / (1_u64 << 52) as f64;
+ // Retain the JVM's truncated log10(2) estimate for arithmetic-path
selection.
+ #[allow(clippy::approx_constant)]
+ let mut decimal_exponent =
+ ((normalized - 1.5) * 0.289529654 + 0.176091259 + f64::from(exponent)
* 0.301029995663981)
+ .floor() as i32;
+ let mut interval = DecimalInterval::new(
+ mantissa,
+ exponent,
+ significant,
+ width,
+ tiny,
+ decimal_exponent,
+ );
+ let mut coefficient = 0_u128;
+ let mut count = 0;
+ loop {
+ let (digit, mut low, mut high, direction) = interval.next_digit(count
!= 0);
+ if digit == 0 && count == 0 && !high {
+ decimal_exponent -= 1;
+ } else {
+ coefficient = coefficient * 10 + u128::from(digit);
+ count += 1;
+ }
+ if count <= 1 && !(-3..8).contains(&decimal_exponent) {
+ low = false;
+ high = false;
+ }
+ if low || high {
+ if high
+ && (!low
+ || direction == Ordering::Greater
+ || direction == Ordering::Equal &&
!coefficient.is_multiple_of(2))
+ {
+ coefficient += 1;
+ }
+ return format!("{sign}{coefficient}e{}", decimal_exponent - count
+ 1);
+ }
+ }
+}
+
+enum DecimalInterval {
+ Small {
+ numerator: i128,
+ denominator: i128,
+ margin: i128,
+ word: u32,
+ },
+ Big {
+ numerator: BigUint,
+ denominator: BigUint,
+ margin: BigUint,
+ },
+}
+
+impl DecimalInterval {
+ fn new(
+ mantissa: u64,
+ exponent: i32,
+ significant: i32,
+ width: i32,
+ tiny: i32,
+ decimal_exponent: i32,
+ ) -> Self {
+ // Factor the scaled value and half-ULP margin into powers of 2 and 5,
+ // removing common binary factors before choosing an arithmetic width.
+ let negative_power = (-decimal_exponent).max(0);
+ let positive_power = decimal_exponent.max(0);
+ let mut numerator_power = negative_power + tiny + exponent - width + 1;
+ let mut denominator_power = positive_power + tiny;
+ let mut margin_power = negative_power + tiny + exponent - significant;
+ let common = numerator_power.min(denominator_power);
+ numerator_power -= common;
+ denominator_power -= common;
+ margin_power -= common;
+ if width == 1 {
+ margin_power -= 1;
+ }
+ if margin_power < 0 {
+ numerator_power -= margin_power;
+ denominator_power -= margin_power;
+ margin_power = 0;
+ }
+ let five = |power: i32| BigUint::from(5_u8).pow(power as u32);
+ let numerator = (BigUint::from(mantissa >> (53 - width)) *
five(negative_power))
+ << numerator_power as usize;
+ let denominator = five(positive_power) << denominator_power as usize;
+ let margin = five(negative_power) << margin_power as usize;
+ let five_bits = |power: i32| {
+ if (1..27).contains(&power) {
+ five(power).bits() as i32
+ } else {
+ power * 3
+ }
+ };
+ let bits = (width + numerator_power + five_bits(negative_power))
+ .max(denominator_power + 1 + five_bits(positive_power + 1));
+ if bits < 64 {
+ Self::Small {
+ numerator: numerator.to_i128().expect("fixed-width numerator"),
+ denominator: denominator.to_i128().expect("fixed-width
denominator"),
+ margin: margin.to_i128().expect("fixed-width margin"),
+ word: if bits < 32 { 32 } else { 64 },
+ }
+ } else {
+ Self::Big {
+ numerator,
+ denominator,
+ margin,
+ }
+ }
+ }
+
+ fn next_digit(&mut self, after_first: bool) -> (u8, bool, bool, Ordering) {
+ match self {
+ Self::Small {
+ numerator,
+ denominator,
+ margin,
+ word,
+ } => {
+ let wrap = |value: i128| {
+ if *word == 32 {
+ value as i32 as i128
+ } else {
+ value as i64 as i128
+ }
+ };
+ let digit = (*numerator / *denominator) as u8;
+ *numerator = (*numerator % *denominator) * 10;
+ *margin = wrap(*margin * 10);
+ let limit = *denominator * 10;
+ let (low, high) = if after_first && *margin <= 0 {
+ (true, true)
+ } else {
+ (*numerator < *margin, wrap(*numerator + *margin) > limit)
+ };
+ (digit, low, high, wrap(2 * *numerator - limit).cmp(&0))
+ }
+ Self::Big {
+ numerator,
+ denominator,
+ margin,
+ } => {
+ let digit = (&*numerator / &*denominator)
+ .to_u8()
+ .expect("decimal digit");
+ *numerator %= &*denominator;
+ *numerator *= 10_u8;
+ *margin *= 10_u8;
+ let limit = &*denominator * 10_u8;
+ let low = *numerator < *margin;
+ let high = &*numerator + &*margin >= limit;
+ (digit, low, high, (&*numerator << 1_usize).cmp(&limit))
+ }
+ }
+ }
+}
+
+/// Java's shortest-decimal string contract (JDK 19+), including its notation
+/// thresholds and closest choice among one- and two-digit decimals. Rust's
+/// shortest Display chooses different digits at ties; Ryū uses ties-to-even.
+pub(crate) fn format_java_float<T: ryu::Float + std::fmt::LowerExp + Copy +
Into<f64>>(
+ value: T,
+) -> String {
+ let number = value.into();
+ if number.is_nan() {
+ return "NaN".into();
+ }
+ if number.is_infinite() {
+ return if number.is_sign_negative() {
+ "-Infinity"
+ } else {
+ "Infinity"
+ }
+ .into();
+ }
+ if number == 0.0 {
+ return if number.is_sign_negative() {
+ "-0.0"
+ } else {
+ "0.0"
+ }
+ .into();
+ }
+
+ let mut buffer = ryu::Buffer::new();
+ let (mut negative, mut digits, mut exponent) =
decimal_parts(buffer.format_finite(value));
+ if digits.len() == 1 {
+ // Padding the shortest digit with zero would miss the closer two-digit
+ // form of small subnormals, e.g. Double.MIN_VALUE -> 4.9E-324.
+ (negative, digits, exponent) = decimal_parts(&format!("{value:.1e}"));
+ }
+ let mut result = if negative {
+ "-".to_string()
+ } else {
+ String::new()
+ };
+ if (-3..7).contains(&exponent) {
+ let point = exponent + 1;
+ if point <= 0 {
+ result.push_str("0.");
+ result.push_str(&"0".repeat((-point) as usize));
+ result.push_str(&digits);
+ } else if point as usize >= digits.len() {
+ result.push_str(&digits);
+ result.push_str(&"0".repeat(point as usize - digits.len()));
+ result.push_str(".0");
+ } else {
+ result.push_str(&digits[..point as usize]);
+ result.push('.');
+ result.push_str(&digits[point as usize..]);
+ }
+ } else {
+ result.push_str(&digits[..1]);
+ result.push('.');
+ result.push_str(if digits.len() > 1 { &digits[1..] } else { "0" });
+ result.push('E');
+ result.push_str(&exponent.to_string());
+ }
+ result
+}
+
+fn decimal_parts(text: &str) -> (bool, String, i32) {
+ let (mantissa, exponent) = text
+ .split_once('e')
+ .map_or((text, 0), |(mantissa, exponent)| {
+ (mantissa, exponent.parse::<i32>().expect("decimal exponent"))
+ });
+ let negative = mantissa.starts_with('-');
+ let mantissa = mantissa.trim_start_matches('-');
+ let point = mantissa.find('.').unwrap_or(mantissa.len()) as i32;
+ let digits = mantissa.replace('.', "");
+ let first = digits
+ .bytes()
+ .position(|byte| byte != b'0')
+ .expect("nonzero decimal");
+ let exponent = exponent + point - first as i32 - 1;
+ let mut digits = digits[first..].to_string();
+ while digits.len() > 1 && digits.ends_with('0') {
+ digits.pop();
+ }
+ (negative, digits, exponent)
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn double_decimal_digits_match_java_8_11_and_17_golden_outputs() {
+ // Outputs obtained from the actual JVMs, including integral rounding,
+ // subnormal normalization, and both sides of binary exponent
boundaries.
+ for (bits, expected) in [
+ (0xc381_e88f_38c0_c8fd, "-1.61304845894623136E17"),
+ (0x44b5_2d02_c7e1_4af6, "9.999999999999999E22"),
+ (0x0000_0000_0000_0001, "4.9E-324"),
+ (0x0010_0000_0000_0000, "2.2250738585072014E-308"),
+ (0x7fef_ffff_ffff_ffff, "1.7976931348623157E308"),
+ (0x0000_0000_0000_0002, "1.0E-323"),
+ (0x000f_ffff_ffff_ffff, "2.225073858507201E-308"),
+ (0x0010_0000_0000_0001, "2.225073858507202E-308"),
+ (0x016f_ffff_ffff_ffff, "9.332636185032188E-302"),
+ (0x0170_0000_0000_0000, "9.332636185032189E-302"),
+ (0x0170_0000_0000_0001, "9.33263618503219E-302"),
+ (0x39af_ffff_ffff_ffff, "7.888609052210117E-31"),
+ (0x39b0_0000_0000_0000, "7.888609052210118E-31"),
+ (0x39b0_0000_0000_0001, "7.88860905221012E-31"),
+ (0x3bef_ffff_ffff_ffff, "5.4210108624275216E-20"),
+ (0x3bf0_0000_0000_0000, "5.421010862427522E-20"),
+ (0x3bf0_0000_0000_0001, "5.421010862427523E-20"),
+ (0x3c9f_ffff_ffff_ffff, "1.1102230246251564E-16"),
+ (0x3ca0_0000_0000_0000, "1.1102230246251565E-16"),
+ (0x3ca0_0000_0000_0001, "1.1102230246251568E-16"),
+ (0x3e9f_ffff_ffff_ffff, "4.7683715820312495E-7"),
+ (0x3ea0_0000_0000_0000, "4.76837158203125E-7"),
+ (0x3ea0_0000_0000_0001, "4.768371582031251E-7"),
+ (0x3fbf_ffff_ffff_ffff, "0.12499999999999999"),
+ (0x3fc0_0000_0000_0000, "0.125"),
+ (0x3fc0_0000_0000_0001, "0.12500000000000003"),
+ (0x3fef_ffff_ffff_ffff, "0.9999999999999999"),
+ (0x3ff0_0000_0000_0000, "1.0"),
+ (0x3ff0_0000_0000_0001, "1.0000000000000002"),
+ (0x3fff_ffff_ffff_ffff, "1.9999999999999998"),
+ (0x4000_0000_0000_0000, "2.0"),
+ (0x4000_0000_0000_0001, "2.0000000000000004"),
+ (0x41ef_ffff_ffff_ffff, "4.2949672959999995E9"),
+ (0x41f0_0000_0000_0000, "4.294967296E9"),
+ (0x41f0_0000_0000_0001, "4.294967296000001E9"),
+ (0x432f_ffff_ffff_ffff, "4.5035996273704955E15"),
+ (0x4330_0000_0000_0000, "4.503599627370496E15"),
+ (0x4330_0000_0000_0001, "4.503599627370497E15"),
+ (0x433f_ffff_ffff_ffff, "9.007199254740991E15"),
+ (0x4340_0000_0000_0000, "9.007199254740992E15"),
+ (0x4340_0000_0000_0001, "9.007199254740994E15"),
+ (0x434f_ffff_ffff_ffff, "1.8014398509481982E16"),
+ (0x4350_0000_0000_0000, "1.8014398509481984E16"),
+ ] {
+ let actual = java_double_decimal_text(f64::from_bits(bits));
+ assert_eq!(
+ decimal_parts(&actual),
+ decimal_parts(&expected.replace('E', "e")),
+ "{bits:x}"
+ );
+ }
+ }
+}
diff --git a/crates/paimon/src/spec/partition_utils.rs
b/crates/paimon/src/spec/partition_utils.rs
index 3785e619..e6fe1dda 100644
--- a/crates/paimon/src/spec/partition_utils.rs
+++ b/crates/paimon/src/spec/partition_utils.rs
@@ -24,6 +24,7 @@
//! - Java `PartitionPathUtils` for path escaping and assembly
use crate::common::date_time::{format_time, format_timestamp};
+use crate::common::number::format_java_float;
use crate::error::*;
use crate::spec::types::DataType;
use crate::spec::BinaryRow;
@@ -484,90 +485,6 @@ fn format_partition_value(
Ok(value)
}
-/// Java's shortest-decimal string contract (JDK 19+), including its notation
-/// thresholds and closest choice among one- and two-digit decimals. Rust's
-/// shortest Display chooses different digits at ties; Ryū uses ties-to-even.
-fn format_java_float<T: ryu::Float + std::fmt::LowerExp + Copy +
Into<f64>>(value: T) -> String {
- let number = value.into();
- if number.is_nan() {
- return "NaN".into();
- }
- if number.is_infinite() {
- return if number.is_sign_negative() {
- "-Infinity"
- } else {
- "Infinity"
- }
- .into();
- }
- if number == 0.0 {
- return if number.is_sign_negative() {
- "-0.0"
- } else {
- "0.0"
- }
- .into();
- }
-
- let mut buffer = ryu::Buffer::new();
- let (mut negative, mut digits, mut exponent) =
decimal_parts(buffer.format_finite(value));
- if digits.len() == 1 {
- // Padding the shortest digit with zero would miss the closer two-digit
- // form of small subnormals, e.g. Double.MIN_VALUE -> 4.9E-324.
- (negative, digits, exponent) = decimal_parts(&format!("{value:.1e}"));
- }
- let mut result = if negative {
- "-".to_string()
- } else {
- String::new()
- };
- if (-3..7).contains(&exponent) {
- let point = exponent + 1;
- if point <= 0 {
- result.push_str("0.");
- result.push_str(&"0".repeat((-point) as usize));
- result.push_str(&digits);
- } else if point as usize >= digits.len() {
- result.push_str(&digits);
- result.push_str(&"0".repeat(point as usize - digits.len()));
- result.push_str(".0");
- } else {
- result.push_str(&digits[..point as usize]);
- result.push('.');
- result.push_str(&digits[point as usize..]);
- }
- } else {
- result.push_str(&digits[..1]);
- result.push('.');
- result.push_str(if digits.len() > 1 { &digits[1..] } else { "0" });
- result.push('E');
- result.push_str(&exponent.to_string());
- }
- result
-}
-
-fn decimal_parts(text: &str) -> (bool, String, i32) {
- let (mantissa, exponent) = text
- .split_once('e')
- .map_or((text, 0), |(mantissa, exponent)| {
- (mantissa, exponent.parse::<i32>().expect("decimal exponent"))
- });
- let negative = mantissa.starts_with('-');
- let mantissa = mantissa.trim_start_matches('-');
- let point = mantissa.find('.').unwrap_or(mantissa.len()) as i32;
- let digits = mantissa.replace('.', "");
- let first = digits
- .bytes()
- .position(|byte| byte != b'0')
- .expect("nonzero decimal");
- let exponent = exponent + point - first as i32 - 1;
- let mut digits = digits[first..].to_string();
- while digits.len() > 1 && digits.ends_with('0') {
- digits.pop();
- }
- (negative, digits, exponent)
-}
-
/// Decode as Java's UTF-8 decoder does for `BinaryString.toString()`.
/// Rust's lossy decoder replaces each byte of a UTF-8 encoded surrogate,
/// while Java replaces the complete malformed surrogate sequence once.
diff --git a/crates/paimon/tests/decimal_schema_evolution_test.rs
b/crates/paimon/tests/decimal_schema_evolution_test.rs
new file mode 100644
index 00000000..0d1de6a4
--- /dev/null
+++ b/crates/paimon/tests/decimal_schema_evolution_test.rs
@@ -0,0 +1,299 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+mod common;
+
+use std::collections::BTreeMap;
+use std::sync::Arc;
+
+use arrow_array::{
+ Array, ArrayRef, Decimal128Array, Float32Array, Float64Array, Int32Array,
RecordBatch,
+ StringArray,
+};
+use futures::TryStreamExt;
+use paimon::spec::{
+ DataType, Datum, DecimalType, DoubleType, FloatType, IntType, Predicate,
PredicateBuilder,
+ Schema, SchemaChange, TableSchema, VarCharType,
+};
+use paimon::table::Table;
+
+use common::incremental_helpers::{memory_table, persist_table_schema,
setup_dirs, write_batch};
+
+async fn read(table: &Table, predicate: Option<Predicate>, only_id: bool) ->
Vec<RecordBatch> {
+ let mut builder = table.new_read_builder();
+ if let Some(predicate) = predicate {
+ builder.with_filter(predicate);
+ }
+ if only_id {
+ builder.with_projection(&["id"]).unwrap();
+ }
+ let plan = builder.new_scan().plan().await.unwrap();
+ builder
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap()
+}
+
+async fn ids(table: &Table, predicate: Predicate) -> Vec<i32> {
+ let mut ids = read(table, Some(predicate), true)
+ .await
+ .into_iter()
+ .flat_map(|batch| {
+ batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .values()
+ .to_vec()
+ })
+ .collect::<Vec<_>>();
+ ids.sort_unstable();
+ ids
+}
+
+fn decimal(precision: u32, scale: u32) -> DataType {
+ DataType::Decimal(DecimalType::new(precision, scale).unwrap())
+}
+
+#[tokio::test]
+async fn decimal_casts_read_mixed_files_and_filter_in_all_table_modes() {
+ for mode in ["append", "pk", "de"] {
+ let mut schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("d", DataType::Double(DoubleType::new()))
+ .column("f", DataType::Float(FloatType::new()))
+ .column("s", DataType::VarChar(VarCharType::string_type()))
+ .column("wide", DataType::Float(FloatType::new()))
+ .option("read.batch-size", "1");
+ if mode == "pk" {
+ schema = schema.primary_key(["id"]).option("bucket", "1");
+ } else if mode == "de" {
+ schema = schema
+ .option("data-evolution.enabled", "true")
+ .option("row-tracking.enabled", "true");
+ }
+ let (io, old) = memory_table(
+ &format!("memory:/decimal_evolution_{mode}"),
+ TableSchema::new(0, &schema.build().unwrap()),
+ );
+ setup_dirs(&io, old.location()).await;
+ persist_table_schema(&io, old.location(), old.schema()).await;
+ let batch = RecordBatch::try_from_iter([
+ (
+ "id",
+ Arc::new(Int32Array::from(vec![0, 1, 2, 3])) as ArrayRef,
+ ),
+ (
+ "d",
+ Arc::new(Float64Array::from(vec![
+ Some(1.005),
+ Some(-1.005),
+ Some(999.995),
+ None,
+ ])) as ArrayRef,
+ ),
+ (
+ "f",
+ Arc::new(Float32Array::from(vec![
+ Some(1.005),
+ Some(-1.005),
+ Some(999.995),
+ None,
+ ])) as ArrayRef,
+ ),
+ (
+ "s",
+ Arc::new(StringArray::from(vec![
+ Some("1e-2"),
+ Some("+.005"),
+ Some("1e100"),
+ None,
+ ])) as ArrayRef,
+ ),
+ (
+ "wide",
+ Arc::new(Float32Array::from(vec![
+ Some(1.2345679),
+ Some(-1.2345679),
+ Some(1e30),
+ None,
+ ])) as ArrayRef,
+ ),
+ ])
+ .unwrap();
+ write_batch(&old, &batch).await;
+ // Single-value stats must not prune a match after decimal rounding.
+ let single = RecordBatch::try_new(
+ batch.schema(),
+ vec![
+ Arc::new(Int32Array::from(vec![4])) as ArrayRef,
+ batch.column(1).slice(0, 1),
+ batch.column(2).slice(0, 1),
+ batch.column(3).slice(0, 1),
+ batch.column(4).slice(0, 1),
+ ],
+ )
+ .unwrap();
+ write_batch(&old, &single).await;
+ let schema = old
+ .schema()
+ .apply_changes(vec![
+ SchemaChange::update_column_type("d".into(), decimal(5, 2)),
+ SchemaChange::update_column_type("f".into(), decimal(5, 2)),
+ SchemaChange::update_column_type("s".into(), decimal(5, 2)),
+ SchemaChange::update_column_type("wide".into(), decimal(38,
20)),
+ ])
+ .unwrap();
+ persist_table_schema(&io, old.location(), &schema).await;
+ let current = old.copy_with_resolved_schema(schema, "main").unwrap();
+ let new = RecordBatch::try_from_iter([
+ ("id", Arc::new(Int32Array::from(vec![5])) as ArrayRef),
+ (
+ "d",
+ Arc::new(
+ Decimal128Array::from(vec![101])
+ .with_precision_and_scale(5, 2)
+ .unwrap(),
+ ) as ArrayRef,
+ ),
+ (
+ "f",
+ Arc::new(
+ Decimal128Array::from(vec![100])
+ .with_precision_and_scale(5, 2)
+ .unwrap(),
+ ) as ArrayRef,
+ ),
+ (
+ "s",
+ Arc::new(
+ Decimal128Array::from(vec![1])
+ .with_precision_and_scale(5, 2)
+ .unwrap(),
+ ) as ArrayRef,
+ ),
+ (
+ "wide",
+ Arc::new(
+ Decimal128Array::from(vec![123_456_788_063_049_320_000])
+ .with_precision_and_scale(38, 20)
+ .unwrap(),
+ ) as ArrayRef,
+ ),
+ ])
+ .unwrap();
+ write_batch(¤t, &new).await;
+ let mut values = BTreeMap::new();
+ for batch in read(¤t, None, false).await {
+ let keys = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ let decimals = batch.columns()[1..]
+ .iter()
+ .map(|c| c.as_any().downcast_ref::<Decimal128Array>().unwrap())
+ .collect::<Vec<_>>();
+ for i in 0..batch.num_rows() {
+ values.insert(
+ keys.value(i),
+ decimals
+ .iter()
+ .map(|c| (!c.is_null(i)).then(|| c.value(i)))
+ .collect::<Vec<_>>(),
+ );
+ }
+ }
+ let first = vec![
+ Some(101),
+ Some(100),
+ Some(1),
+ Some(123_456_788_063_049_320_000),
+ ];
+ assert_eq!(
+ values,
+ BTreeMap::from([
+ (0, first.clone()),
+ (
+ 1,
+ vec![
+ Some(-101),
+ Some(-100),
+ Some(1),
+ Some(-123_456_788_063_049_320_000)
+ ]
+ ),
+ (2, vec![None, Some(99_999), None, None]),
+ (3, vec![None; 4]),
+ (4, first.clone()),
+ (5, first),
+ ]),
+ "{mode}"
+ );
+ let pb = PredicateBuilder::new(current.schema().fields());
+ for (column, unscaled, precision, scale) in [
+ ("d", 101, 5, 2),
+ ("f", 100, 5, 2),
+ ("wide", 123_456_788_063_049_320_000, 38, 20),
+ ] {
+ assert_eq!(
+ ids(
+ ¤t,
+ pb.equal(
+ column,
+ Datum::Decimal {
+ unscaled,
+ precision,
+ scale
+ }
+ )
+ .unwrap()
+ )
+ .await,
+ vec![0, 4, 5],
+ "{mode} {column}"
+ );
+ }
+ assert_eq!(
+ ids(
+ ¤t,
+ pb.equal(
+ "s",
+ Datum::Decimal {
+ unscaled: 1,
+ precision: 5,
+ scale: 2
+ }
+ )
+ .unwrap()
+ )
+ .await,
+ vec![0, 1, 4, 5],
+ "{mode} exponent"
+ );
+ assert_eq!(
+ ids(¤t, pb.is_null("d").unwrap()).await,
+ vec![2, 3],
+ "{mode} overflow"
+ );
+ }
+}
