use spg_storage::Value;
use crate::EngineError;
fn apply_negative_scale(scaled: i128, src_scale: u16, k: u16) -> Option<i128> {
let denom = pow10_i128_checked(src_scale.checked_add(k)?)?;
let half = denom / 2;
let biased = if scaled >= 0 {
scaled.checked_add(half)?
} else {
scaled.checked_sub(half)?
};
(biased / denom).checked_mul(pow10_i128_checked(k)?)
}
const fn split_declared_scale(scale: i16) -> (bool, u16) {
if scale < 0 {
(true, scale.unsigned_abs())
} else {
#[allow(clippy::cast_sign_loss)]
(false, scale as u16)
}
}
pub(crate) fn numeric_from_integer(
n: i128,
precision: u16,
scale: i16,
col_name: &str,
) -> Result<Value<'static>, EngineError> {
let (neg, k) = split_declared_scale(scale);
if neg {
let rounded = apply_negative_scale(n, 0, k).ok_or_else(|| {
EngineError::Unsupported(alloc::format!(
"integer overflow scaling value for column `{col_name}` to scale {scale}"
))
})?;
check_precision(rounded, precision, scale, col_name)?;
return Ok(Value::Numeric {
scaled: rounded,
scale: 0,
kind: spg_storage::NumericKind::Finite,
});
}
let scale = k;
let factor = pow10_i128(scale);
let scaled = n.checked_mul(factor).ok_or_else(|| {
EngineError::Unsupported(alloc::format!(
"integer overflow scaling value for column `{col_name}` to scale {scale}"
))
})?;
#[allow(clippy::cast_possible_wrap)]
let signed_scale = scale as i16;
check_precision(scaled, precision, signed_scale, col_name)?;
Ok(Value::Numeric {
scaled,
scale,
kind: spg_storage::NumericKind::Finite,
})
}
#[allow(clippy::cast_precision_loss, clippy::cast_possible_truncation)]
pub(crate) fn numeric_from_float(
x: f64,
precision: u16,
scale: i16,
col_name: &str,
) -> Result<Value<'static>, EngineError> {
let (neg_scale, k) = split_declared_scale(scale);
if neg_scale {
if !x.is_finite() {
return Err(EngineError::Unsupported(alloc::format!(
"cannot store non-finite float in NUMERIC column `{col_name}`"
)));
}
let mut f = 1.0_f64;
for _ in 0..k {
f *= 10.0;
}
let q = (x / f).round();
#[allow(clippy::cast_possible_truncation)]
let rounded = (q as i128).checked_mul(pow10_i128(k)).ok_or_else(|| {
EngineError::Unsupported(alloc::format!(
"value overflows NUMERIC column `{col_name}`"
))
})?;
check_precision(rounded, precision, scale, col_name)?;
return Ok(Value::Numeric {
scaled: rounded,
scale: 0,
kind: spg_storage::NumericKind::Finite,
});
}
let scale = k;
if !x.is_finite() {
return Err(EngineError::Unsupported(alloc::format!(
"cannot store non-finite float in NUMERIC column `{col_name}`"
)));
}
let mut factor = 1.0_f64;
for _ in 0..scale {
factor *= 10.0;
}
let shifted = x * factor;
let biased = if shifted >= 0.0 {
shifted + 0.5
} else {
shifted - 0.5
};
if !(-1e38..=1e38).contains(&biased) {
return Err(EngineError::Unsupported(alloc::format!(
"value {x} overflows NUMERIC range for column `{col_name}`"
)));
}
let scaled = biased as i128;
#[allow(clippy::cast_possible_wrap)]
let signed_scale = scale as i16;
check_precision(scaled, precision, signed_scale, col_name)?;
Ok(Value::Numeric {
scaled,
scale,
kind: spg_storage::NumericKind::Finite,
})
}
pub(crate) fn parse_numeric_special(s: &str) -> Option<spg_storage::NumericKind> {
use spg_storage::NumericKind;
let t = s.trim();
let lower = t.to_ascii_lowercase();
match lower.as_str() {
"nan" => Some(NumericKind::NaN),
"infinity" | "inf" | "+infinity" | "+inf" => Some(NumericKind::PosInf),
"-infinity" | "-inf" => Some(NumericKind::NegInf),
_ => None,
}
}
pub(crate) fn special_math(
name: &str,
args: &[Value<'_>],
) -> Option<Result<Value<'static>, crate::eval::EvalError>> {
use spg_storage::NumericKind as K;
let kind_of = |v: &Value<'_>| match v {
Value::Numeric { kind, .. } if *kind != K::Finite => Some(*kind),
_ => None,
};
if !args.iter().any(|a| kind_of(a).is_some()) {
return None;
}
let special = |k: K| {
Ok(Value::Numeric {
scaled: 0,
scale: 0,
kind: k,
})
};
let finite = |n: i128| {
Ok(Value::Numeric {
scaled: n,
scale: 0,
kind: K::Finite,
})
};
let neg_err = |what: &str| {
Err(crate::eval::EvalError::TypeMismatch {
detail: alloc::format!("cannot take {what} of a negative number"),
})
};
let a0 = kind_of(&args[0]);
let a1 = args.get(1).and_then(kind_of);
Some(match (name, a0) {
("abs", Some(K::NegInf)) => special(K::PosInf),
("abs", Some(k)) => special(k),
("trunc" | "truncate" | "round" | "ceil" | "ceiling" | "floor" | "trim_scale", Some(k)) => {
special(k)
}
("sign", Some(K::PosInf)) => finite(1),
("sign", Some(K::NegInf)) => finite(-1),
("sign", Some(K::NaN)) => special(K::NaN),
("scale" | "min_scale", Some(_)) => Ok(Value::Null),
("sqrt", Some(K::NegInf)) => neg_err("square root"),
("sqrt", Some(k)) => special(k),
("ln", Some(K::NegInf)) => neg_err("logarithm"),
("ln", Some(k)) => special(k),
("exp", Some(K::NegInf)) => finite(0),
("exp", Some(k)) => special(k),
("log", Some(K::NegInf)) if args.len() == 1 => neg_err("logarithm"),
("log", Some(k)) if args.len() == 1 => special(k),
("log", Some(K::PosInf)) => finite(0),
("log", _) if a1 == Some(K::PosInf) => special(K::PosInf),
("log", _) if a1 == Some(K::NaN) || a0 == Some(K::NaN) => special(K::NaN),
("div", Some(K::NaN)) => special(K::NaN),
("div", Some(k)) if a1.is_none() => special(k),
("div", _) if a1 == Some(K::NaN) => special(K::NaN),
("div", _) => finite(0),
("mod", _) => special(K::NaN),
("width_bucket", _) => Err(crate::eval::EvalError::TypeMismatch {
detail: alloc::string::String::from(
"operand, lower bound, and upper bound cannot be NaN",
),
}),
_ => return None,
})
}
pub(crate) fn strip_digit_underscores(s: &str) -> Option<alloc::borrow::Cow<'_, str>> {
if !s.contains('_') {
return Some(alloc::borrow::Cow::Borrowed(s));
}
let b = s.as_bytes();
for (i, &c) in b.iter().enumerate() {
if c == b'_'
&& !(i > 0 && i + 1 < b.len() && b[i - 1].is_ascii_digit() && b[i + 1].is_ascii_digit())
{
return None;
}
}
Some(alloc::borrow::Cow::Owned(s.replace('_', "")))
}
pub(crate) fn parse_numeric_text(s: &str) -> Option<(i128, u16)> {
let s = s.trim();
if s.is_empty() {
return None;
}
let s: &str = &strip_digit_underscores(s)?;
if let Some(idx) = s.find(['e', 'E']) {
let exp: i32 = s[idx + 1..].parse().ok()?;
let (mantissa, base_scale) = parse_plain_numeric(&s[..idx])?;
let eff = i32::from(base_scale) - exp;
return if eff >= 0 {
Some((mantissa, u16::try_from(eff).ok()?))
} else {
let shift = u16::try_from(-eff).ok()?;
if shift > 38 {
return None;
}
Some((mantissa.checked_mul(pow10_i128(shift))?, 0))
};
}
parse_plain_numeric(s)
}
fn parse_plain_numeric(s: &str) -> Option<(i128, u16)> {
if s.is_empty() {
return None;
}
let (negative, rest) = match s.as_bytes()[0] {
b'-' => (true, &s[1..]),
b'+' => (false, &s[1..]),
_ => (false, s),
};
if rest.is_empty() {
return None;
}
let (int_part, frac_part) = match rest.find('.') {
Some(idx) => (&rest[..idx], &rest[idx + 1..]),
None => (rest, ""),
};
if int_part.is_empty() && frac_part.is_empty() {
return None;
}
if int_part.bytes().any(|b| !b.is_ascii_digit()) {
return None;
}
if frac_part.bytes().any(|b| !b.is_ascii_digit()) {
return None;
}
let scale_u32 = u32::try_from(frac_part.len()).ok()?;
if scale_u32 > u32::from(u16::MAX) {
return None;
}
#[allow(clippy::cast_possible_truncation)]
let scale = scale_u32 as u16;
let mut digits = alloc::string::String::with_capacity(int_part.len() + frac_part.len() + 1);
if negative {
digits.push('-');
}
digits.push_str(int_part);
digits.push_str(frac_part);
let digits = if digits == "-" {
return None;
} else if digits.is_empty() {
"0"
} else {
digits.as_str()
};
let mantissa: i128 = digits.parse().ok()?;
Some((mantissa, scale))
}
pub(crate) fn numeric_rescale(
scaled: i128,
src_scale: u16,
precision: u16,
dst_scale: i16,
col_name: &str,
) -> Result<Value<'static>, EngineError> {
let (neg_scale, k) = split_declared_scale(dst_scale);
if neg_scale {
let rounded = apply_negative_scale(scaled, src_scale, k).ok_or_else(|| {
EngineError::Unsupported(alloc::format!(
"overflow rescaling NUMERIC for column `{col_name}`"
))
})?;
check_precision(rounded, precision, dst_scale, col_name)?;
return Ok(Value::Numeric {
scaled: rounded,
scale: 0,
kind: spg_storage::NumericKind::Finite,
});
}
let dst_scale = k;
if dst_scale >= src_scale && pow10_i128_checked(dst_scale - src_scale).is_none() {
let widened =
spg_storage::bignum::BigNumeric::from_i128(scaled, src_scale).round_to(dst_scale);
return Ok(crate::eval::binop::bignum_to_value(widened));
}
let new_scaled = if dst_scale >= src_scale {
let bump = pow10_i128(dst_scale - src_scale);
match scaled.checked_mul(bump) {
Some(v) => v,
None => {
let widened = spg_storage::bignum::BigNumeric::from_i128(scaled, src_scale)
.round_to(dst_scale);
return Ok(crate::eval::binop::bignum_to_value(widened));
}
}
} else {
let drop = pow10_i128(src_scale - dst_scale);
let half = drop / 2;
if scaled >= 0 {
(scaled + half) / drop
} else {
(scaled - half) / drop
}
};
#[allow(clippy::cast_possible_wrap)]
let signed_dst = dst_scale as i16;
check_precision(new_scaled, precision, signed_dst, col_name)?;
Ok(Value::Numeric {
scaled: new_scaled,
scale: dst_scale,
kind: spg_storage::NumericKind::Finite,
})
}
pub(crate) const fn numeric_truncate_to_integer(scaled: i128, scale: u16) -> i128 {
if scale == 0 {
return scaled;
}
let factor = pow10_i128_const(scale);
scaled / factor
}
pub(crate) const fn numeric_round_to_integer(scaled: i128, scale: u16) -> i128 {
if scale == 0 {
return scaled;
}
let factor = pow10_i128_const(scale);
let neg = scaled < 0;
let abs = scaled.unsigned_abs() as i128;
let q = abs / factor;
let r = abs % factor;
let mag = if 2 * r >= factor { q + 1 } else { q };
if neg { -mag } else { mag }
}
pub(crate) fn check_precision_text(
v: &Value<'static>,
precision: u16,
scale: i16,
_col_name: &str,
) -> Result<(), EngineError> {
if precision == 0 {
return Ok(());
}
let text = match v {
Value::Numeric { scaled, scale, .. } => crate::eval::format_numeric(*scaled, *scale),
Value::NumericBig(b) => b.to_decimal_str(),
_ => return Ok(()),
};
let body = text.trim_start_matches('-');
let int_part = body
.split('.')
.next()
.unwrap_or(body)
.trim_start_matches('0');
let allowed = usize::try_from(i32::from(precision) - i32::from(scale)).unwrap_or(0);
if int_part.len() > allowed {
return Err(numeric_field_overflow(precision, scale));
}
Ok(())
}
fn numeric_field_overflow(precision: u16, scale: i16) -> EngineError {
EngineError::Unsupported(alloc::format!(
"numeric field overflow DETAIL: A field with precision {precision}, scale {scale} \
must round to an absolute value less than 10^{}.",
i32::from(precision) - i32::from(scale)
))
}
fn check_precision(
scaled: i128,
precision: u16,
scale: i16,
col_name: &str,
) -> Result<(), EngineError> {
if precision == 0 {
return Ok(());
}
if precision > 38 || scale < 0 {
return check_precision_text(
&Value::Numeric {
scaled,
scale: 0,
kind: spg_storage::NumericKind::Finite,
},
precision,
scale,
col_name,
);
}
#[allow(clippy::cast_sign_loss)]
let limit = pow10_i128(precision);
if scaled.unsigned_abs() >= limit.unsigned_abs() {
return Err(numeric_field_overflow(precision, scale));
}
Ok(())
}
pub(crate) fn numeric_add_checked(
a: i128,
a_scale: u16,
b: i128,
b_scale: u16,
) -> Option<(i128, u16)> {
if a_scale == b_scale {
a.checked_add(b).map(|s| (s, a_scale))
} else if a_scale > b_scale {
let f = 10i128.checked_pow(u32::from(a_scale - b_scale))?;
a.checked_add(b.checked_mul(f)?).map(|s| (s, a_scale))
} else {
let f = 10i128.checked_pow(u32::from(b_scale - a_scale))?;
a.checked_mul(f)?.checked_add(b).map(|s| (s, b_scale))
}
}
pub(crate) fn numeric_add(a: i128, a_scale: u16, b: i128, b_scale: u16) -> (i128, u16) {
if a_scale == b_scale {
(a.saturating_add(b), a_scale)
} else if a_scale > b_scale {
let f = pow10_sat(a_scale - b_scale);
(a.saturating_add(b.saturating_mul(f)), a_scale)
} else {
let f = pow10_sat(b_scale - a_scale);
(a.saturating_mul(f).saturating_add(b), b_scale)
}
}
pub(crate) fn numeric_avg(sum_scaled: i128, sum_scale: u16, count: i128) -> (i128, u16) {
let rscale = division_display_scale(sum_scaled, sum_scale, count, 0);
let (num, den) = if i32::from(rscale) >= i32::from(sum_scale) {
let k = rscale - sum_scale;
(sum_scaled.saturating_mul(pow10_sat(k)), count)
} else {
let k = sum_scale - rscale;
(sum_scaled, count.saturating_mul(pow10_sat(k)))
};
(div_round_half_away(num, den), rscale)
}
pub(crate) fn numeric_div(a: i128, sa: u16, b: i128, sb: u16) -> Option<(i128, u16)> {
let rscale = division_display_scale(a, sa, b, sb);
let e = i32::from(sb) + i32::from(rscale) - i32::from(sa);
let (num, den) = if e >= 0 {
(a.checked_mul(pow10_checked(e as u32)?)?, b)
} else {
(a, b.checked_mul(pow10_checked((-e) as u32)?)?)
};
Some((div_round_half_away(num, den), rscale))
}
fn pow10_checked(p: u32) -> Option<i128> {
let mut acc: i128 = 1;
for _ in 0..p {
acc = acc.checked_mul(10)?;
}
Some(acc)
}
fn division_display_scale(
dividend: i128,
dividend_scale: u16,
divisor: i128,
divisor_scale: u16,
) -> u16 {
let dwf = base10000_weight_firstdigit(dividend, dividend_scale);
let vwf = base10000_weight_firstdigit(divisor, divisor_scale);
division_display_scale_from_wf(dwf, dividend_scale, vwf)
}
pub(crate) fn division_display_scale_big(
dividend: &spg_storage::bignum::BigNumeric,
divisor: &spg_storage::bignum::BigNumeric,
) -> u16 {
let (dm, ds) = mantissa_and_scale_big(dividend);
let (vm, vs) = mantissa_and_scale_big(divisor);
let dwf = weight_firstdigit_core(&dm, ds);
let vwf = weight_firstdigit_core(&vm, vs);
division_display_scale_from_wf(dwf, ds, vwf)
}
pub(crate) fn sqrt_display_scale_big(arg: &spg_storage::bignum::BigNumeric) -> u16 {
let (m, s) = mantissa_and_scale_big(arg);
let (weight, _lead) = weight_firstdigit_core(&m, s);
let sweight = (weight + 1) * 2 - 1;
let scale = (16 - sweight).max(i32::from(s)).max(0).min(1000);
scale as u16
}
fn division_display_scale_from_wf(
(dividend_group, dividend_lead): (i32, i32),
dividend_scale: u16,
(divisor_group, divisor_lead): (i32, i32),
) -> u16 {
let mut quotient_group = dividend_group - divisor_group;
if dividend_lead <= divisor_lead {
quotient_group -= 1;
}
let scale = (16 - quotient_group * 4)
.max(i32::from(dividend_scale))
.max(0)
.min(1000);
scale as u16
}
fn mantissa_and_scale_big(b: &spg_storage::bignum::BigNumeric) -> (alloc::string::String, u16) {
use alloc::string::{String, ToString};
let s = b.to_decimal_str();
let s = s.strip_prefix('-').unwrap_or(&s);
let digits: String = s.chars().filter(|c| *c != '.').collect();
let trimmed = digits.trim_start_matches('0');
let m = if trimmed.is_empty() {
"0".to_string()
} else {
trimmed.to_string()
};
(m, b.scale())
}
fn base10000_weight_firstdigit(scaled: i128, scale: u16) -> (i32, i32) {
let a = scaled.unsigned_abs();
if a == 0 {
return (0, 0);
}
weight_firstdigit_core(&alloc::string::ToString::to_string(&a), scale)
}
fn weight_firstdigit_core(s: &str, scale: u16) -> (i32, i32) {
if s == "0" {
return (0, 0);
}
let ndigits = s.len() as i32;
let scale = i32::from(scale);
let int_digits = ndigits - scale;
if int_digits > 0 {
let weight = (int_digits - 1) / 4;
let top_len = (int_digits - weight * 4) as usize;
let firstdigit: i32 = s[..top_len].parse().unwrap_or(0);
(weight, firstdigit)
} else {
let lead_zeros = (-int_digits) as usize;
let g = (lead_zeros as i32) / 4;
let weight = -(g + 1);
let mut frac = alloc::string::String::with_capacity(lead_zeros + s.len());
for _ in 0..lead_zeros {
frac.push('0');
}
frac.push_str(s);
let start = (4 * g) as usize;
let mut group: alloc::string::String = frac[start..].chars().take(4).collect();
while group.len() < 4 {
group.push('0');
}
let firstdigit: i32 = group.parse().unwrap_or(0);
(weight, firstdigit)
}
}
fn div_round_half_away(num: i128, den: i128) -> i128 {
let q = num / den;
let r = num % den;
if r.unsigned_abs() * 2 >= den.unsigned_abs() {
if num >= 0 { q + 1 } else { q - 1 }
} else {
q
}
}
fn pow10_sat(p: u16) -> i128 {
let mut acc: i128 = 1;
for _ in 0..p {
match acc.checked_mul(10) {
Some(v) => acc = v,
None => return i128::MAX,
}
}
acc
}
const fn pow10_i128_const(p: u16) -> i128 {
let mut acc: i128 = 1;
let mut i = 0;
while i < p {
match acc.checked_mul(10) {
Some(v) => acc = v,
None => return i128::MAX,
}
i += 1;
}
acc
}
fn pow10_i128(p: u16) -> i128 {
pow10_i128_const(p)
}
const fn pow10_i128_checked(p: u16) -> Option<i128> {
let mut acc: i128 = 1;
let mut i = 0;
while i < p {
match acc.checked_mul(10) {
Some(v) => acc = v,
None => return None,
}
i += 1;
}
Some(acc)
}