macro_rules! dispatch_arith {
(
$left:expr, $right:expr;
fixed: $fh:ident, arb: $ah:ident ($ctx:expr, $target:expr, $fragment:expr);
$($extra:tt)*
) => {
dispatch_arith!(@rows
($left, $right) $fh $ah ($ctx, $target, $fragment)
[Float4 Float8 Int1 Int2 Int4 Int8 Int16 Uint1 Uint2 Uint4 Uint8 Uint16]
{$($extra)*}
{}
)
};
(@rows
($left:expr, $right:expr) $fh:ident $ah:ident ($ctx:expr, $target:expr, $fragment:expr)
[$L:ident $($rest:ident)*]
{$($extra:tt)*}
{$($acc:tt)*}
) => {
dispatch_arith!(@rows
($left, $right) $fh $ah ($ctx, $target, $fragment)
[$($rest)*]
{$($extra)*}
{
$($acc)*
(ViewData::$L(l), ViewData::Float4(r)) => $fh($ctx, dispatch_arith!(@values $L l), r.values(), $target, $fragment),
(ViewData::$L(l), ViewData::Float8(r)) => $fh($ctx, dispatch_arith!(@values $L l), r.values(), $target, $fragment),
(ViewData::$L(l), ViewData::Int1(r)) => $fh($ctx, dispatch_arith!(@values $L l), r.values(), $target, $fragment),
(ViewData::$L(l), ViewData::Int2(r)) => $fh($ctx, dispatch_arith!(@values $L l), r.values(), $target, $fragment),
(ViewData::$L(l), ViewData::Int4(r)) => $fh($ctx, dispatch_arith!(@values $L l), r.values(), $target, $fragment),
(ViewData::$L(l), ViewData::Int8(r)) => $fh($ctx, dispatch_arith!(@values $L l), r.values(), $target, $fragment),
(ViewData::$L(l), ViewData::Int16(r)) => $fh($ctx, dispatch_arith!(@values $L l), &wides::<i128>(r), $target, $fragment),
(ViewData::$L(l), ViewData::Uint1(r)) => $fh($ctx, dispatch_arith!(@values $L l), r.values(), $target, $fragment),
(ViewData::$L(l), ViewData::Uint2(r)) => $fh($ctx, dispatch_arith!(@values $L l), r.values(), $target, $fragment),
(ViewData::$L(l), ViewData::Uint4(r)) => $fh($ctx, dispatch_arith!(@values $L l), r.values(), $target, $fragment),
(ViewData::$L(l), ViewData::Uint8(r)) => $fh($ctx, dispatch_arith!(@values $L l), r.values(), $target, $fragment),
(ViewData::$L(l), ViewData::Uint16(r)) => $fh($ctx, dispatch_arith!(@values $L l), &wides::<u128>(r), $target, $fragment),
(ViewData::$L(l), ViewData::Decimal(r)) => $ah($ctx, dispatch_arith!(@values $L l), &decimals(r), $target, $fragment),
}
)
};
(@rows
($left:expr, $right:expr) $fh:ident $ah:ident ($ctx:expr, $target:expr, $fragment:expr)
[]
{$($extra:tt)*}
{$($acc:tt)*}
) => {
match ($left, $right) {
$($acc)*
(ViewData::Decimal(l), ViewData::Float4(r)) => $ah($ctx, &decimals(l), r.values(), $target, $fragment),
(ViewData::Decimal(l), ViewData::Float8(r)) => $ah($ctx, &decimals(l), r.values(), $target, $fragment),
(ViewData::Decimal(l), ViewData::Int1(r)) => $ah($ctx, &decimals(l), r.values(), $target, $fragment),
(ViewData::Decimal(l), ViewData::Int2(r)) => $ah($ctx, &decimals(l), r.values(), $target, $fragment),
(ViewData::Decimal(l), ViewData::Int4(r)) => $ah($ctx, &decimals(l), r.values(), $target, $fragment),
(ViewData::Decimal(l), ViewData::Int8(r)) => $ah($ctx, &decimals(l), r.values(), $target, $fragment),
(ViewData::Decimal(l), ViewData::Int16(r)) => $ah($ctx, &decimals(l), &wides::<i128>(r), $target, $fragment),
(ViewData::Decimal(l), ViewData::Uint1(r)) => $ah($ctx, &decimals(l), r.values(), $target, $fragment),
(ViewData::Decimal(l), ViewData::Uint2(r)) => $ah($ctx, &decimals(l), r.values(), $target, $fragment),
(ViewData::Decimal(l), ViewData::Uint4(r)) => $ah($ctx, &decimals(l), r.values(), $target, $fragment),
(ViewData::Decimal(l), ViewData::Uint8(r)) => $ah($ctx, &decimals(l), r.values(), $target, $fragment),
(ViewData::Decimal(l), ViewData::Uint16(r)) => $ah($ctx, &decimals(l), &wides::<u128>(r), $target, $fragment),
(ViewData::Decimal(l), ViewData::Decimal(r)) => $ah($ctx, &decimals(l), &decimals(r), $target, $fragment),
$($extra)*
}
};
(@values Int16 $array:ident) => {
&wides::<i128>($array)
};
(@values Uint16 $array:ident) => {
&wides::<u128>($array)
};
(@values $L:ident $array:ident) => {
$array.values()
};
}
pub mod add;
pub mod context;
pub mod div;
pub mod mul;
pub mod option;
pub mod rem;
pub mod scalar;
pub mod sub;
use reifydb_value::{
error::{Error, RuntimeErrorKind, TypeError},
fragment::Fragment,
value::{
constraint::{precision::Precision, scale::Scale},
decimal::unscaled::MAX_DIGITS,
value_type::ValueType,
},
};
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
pub enum ArithOp {
Add,
Sub,
Mul,
Div,
Rem,
}
fn arith_digits(ty: &ValueType) -> Option<(u8, u8)> {
if ty.is_floating_point() {
return family_digits(&ValueType::DECIMAL);
}
family_digits(ty)
}
pub fn arith_target(op: ArithOp, left: ValueType, right: ValueType) -> ValueType {
if !(is_family(&left) || is_family(&right)) {
return ValueType::promote(left, right);
}
let (Some((left_digits, left_scale)), Some((right_digits, right_scale))) =
(arith_digits(&left), arith_digits(&right))
else {
return ValueType::promote(left, right);
};
let (digits, scale) = match op {
ArithOp::Add | ArithOp::Sub => (left_digits.max(right_digits) + 1, left_scale.max(right_scale)),
ArithOp::Mul => {
let (digits, scale) = (left_digits + right_digits, left_scale + right_scale);
if digits.saturating_add(scale) > MAX_DIGITS {
(digits, scale.min(6).max(MAX_DIGITS.saturating_sub(digits)))
} else {
(digits, scale)
}
}
ArithOp::Div => (left_digits + right_scale, left_scale.max(right_scale).max(6)),
ArithOp::Rem => (left_digits.min(right_digits), left_scale.max(right_scale)),
};
let scale = scale.min(MAX_DIGITS);
let precision = Precision::new(digits.saturating_add(scale).clamp(1, MAX_DIGITS));
ValueType::decimal(precision, Scale::new(scale))
}
fn integer_digits(ty: &ValueType) -> Option<u8> {
match ty {
ValueType::Int1 | ValueType::Uint1 => Some(3),
ValueType::Int2 | ValueType::Uint2 => Some(5),
ValueType::Int4 | ValueType::Uint4 => Some(10),
ValueType::Int8 => Some(19),
ValueType::Uint8 => Some(20),
ValueType::Int16 | ValueType::Uint16 => Some(39),
_ => None,
}
}
pub fn family_digits(ty: &ValueType) -> Option<(u8, u8)> {
match ty {
ValueType::Decimal {
precision,
scale,
} => Some((precision.value() - scale.value(), scale.value())),
_ => integer_digits(ty).map(|digits| (digits, 0)),
}
}
pub fn is_family(ty: &ValueType) -> bool {
matches!(ty, ValueType::Decimal { .. })
}
pub(crate) fn row_of(len: usize, row: usize) -> usize {
if len == 1 {
0
} else {
row
}
}
pub(crate) fn output_len(left: usize, right: usize) -> usize {
if left == 1 {
right
} else {
left
}
}
pub fn length_mismatch(left: usize, right: usize, fragment: &Fragment) -> Error {
TypeError::Runtime {
kind: RuntimeErrorKind::ColumnLengthMismatch {
left,
right,
fragment: fragment.clone(),
},
message: format!("cannot combine a column of {left} rows with a column of {right} rows"),
}
.into()
}