treewide: Replace rust_decimal with bigdecimal

We wrap the type with our own enum capable of representing +/-Infinity
and NaN. This allows us to snapshot and replicate tables with any
`DECIMAL`/`NUMERIC` precision, removing the limit of 27 (previously 28)
digits.

`bigdecimal` is certainly slower for very large numbers, and it may be
worthwhile to add a `Decimal::SmallNumber` variant which can fit into
`rust_decimal`'s i128 representation, but I don't know how much faster
it is for smaller numbers.

Fixes: REA-1933, REA-5720
Release-Note-Core: Add support for arbitrarily large `DECIMAL` and
  `NUMERIC` values. We can now replicate and use the full range allowed
  by MySQL and Postgres for these columns. This also fixes an issue with
  replicating NaN values, which will now work as expected.
BuildDistroReleasePkg: decimalalpha

Change-Id: I9ed0219e1f78d5e1aa8466a41ff6d42239fb9467
Reviewed-on: https://gerrit.readyset.name/c/readyset/+/9519
Reviewed-by: Johnathan Davis <jcd@readyset.io>
Tested-by: Buildkite CI

Author: mvzink

Cargo.lock

@@ -40,6 +40,7 @@ members = [
     "readyset-common",
     "readyset-data",
     "readyset-dataflow",
+    "readyset-decimal",
     "readyset-e2e-tests",
     "readyset-errors",
     "readyset-logictest",
@@ -99,6 +100,7 @@ async-stream = "0.3.5"
 async-trait = "0.1"
 backtrace = "0.3.65"
 base64 = { version = "0.22.1", default-features = false, features = ["std"] }
+bigdecimal = "0.4.8"
 bimap = "0.6.3"
 bincode = "1.3.3"
 bit-vec = "0.6"
@@ -174,6 +176,8 @@ nom_locate = "4.2.0"
 notify = { version = "8.0.0", default-features = false, features = [
     "macos_kqueue",
 ] }
+num-integer = "0.1.46"
+num-traits = "0.2.19"
 num_cpus = "1.16.0"
 once_cell = "1.20.2"
 opentelemetry = "0.21.0"
@@ -204,7 +208,6 @@ rocksdb = { version = "0.22.0", default-features = false, features = [
     "lz4",
     "jemalloc",
 ] }
-rust_decimal = "1.26"
 rusty-fork = "0.3.0"
 rustyline = "14.0"
 rustyline-derive = "0.10.0"

Cargo.toml

@@ -13,12 +13,12 @@ eui48 = { workspace = true }
 growable-bloom-filter = { workspace = true }
 rand = { workspace = true }
 rand_regex = { workspace = true }
-rust_decimal = { workspace = true }
 uuid = { workspace = true }
 rand_distr = { workspace = true }
 
 readyset-sql = { path = "../readyset-sql" }
-readyset-data = { path = "../readyset-data/" }
+readyset-data = { path = "../readyset-data" }
+readyset-decimal = { path = "../readyset-decimal" }
 
 [dev-dependencies]
 test-strategy = { workspace = true }

data-generator/Cargo.toml

@@ -13,8 +13,8 @@ use rand::seq::SliceRandom;
 use rand::{thread_rng, Rng, RngCore};
 use rand_distr::Zipf;
 use readyset_data::{encoding::Encoding, DfType, DfValue, Dialect};
+use readyset_decimal::Decimal;
 use readyset_sql::ast::SqlType;
-use rust_decimal::Decimal;
 
 mod distribution_annotation;
 
@@ -756,11 +756,13 @@ pub fn unique_value_of_type(typ: &SqlType, idx: u32) -> DfValue {
         SqlType::MediumIntUnsigned(_) => (idx).into(),
         SqlType::Float | SqlType::Double => (1.5 + idx as f64).try_into().unwrap(),
         SqlType::Real => (1.5 + idx as f32).try_into().unwrap(),
-        SqlType::Decimal(prec, scale) => Decimal::new(clamp_digits(*prec as _), *scale as _).into(),
+        SqlType::Decimal(prec, scale) => {
+            Decimal::new(clamp_digits(*prec as _) as _, *scale as _).into()
+        }
         SqlType::Numeric(prec_scale) => match prec_scale {
-            Some((prec, None)) => Decimal::new(clamp_digits(*prec as _), 1),
-            Some((prec, Some(scale))) => Decimal::new(clamp_digits(*prec as _), *scale as _),
-            None => Decimal::new((15 + idx) as i64, 2),
+            Some((prec, None)) => Decimal::new(clamp_digits(*prec as _) as _, 1),
+            Some((prec, Some(scale))) => Decimal::new(clamp_digits(*prec as _) as _, *scale as _),
+            None => Decimal::new((15 + idx) as _, 2),
         }
         .into(),
         SqlType::DateTime(_) | SqlType::Timestamp => (NaiveDate::from_ymd_opt(2020, 1, 1)

data-generator/src/lib.rs

@@ -11,9 +11,9 @@ chrono = { workspace = true, features = ["serde"] }
 chrono-tz = { workspace = true, features = ["serde"] }
 itertools = { workspace = true }
 lazy_static = { workspace = true }
+num-traits = { workspace = true }
 proptest = { workspace = true }
 regex = { workspace = true }
-rust_decimal = { workspace = true }
 serde = { workspace = true, features = ["derive"] }
 serde_json = { workspace = true }
 test-strategy = { workspace = true }
@@ -25,6 +25,7 @@ readyset-sql = { path = "../readyset-sql" }
 readyset-util = { path = "../readyset-util" }
 mysql-time = { path = "../mysql-time" }
 readyset-data = { path = "../readyset-data" }
+readyset-decimal = { path = "../readyset-decimal" }
 readyset-errors = { path = "../readyset-errors" }
 partial-map = { path = "../partial-map" }
 

dataflow-expression/Cargo.toml

@@ -1,7 +1,7 @@
 use std::borrow::Borrow;
 use std::cmp::Ordering;
 use std::fmt::{self, Write};
-use std::ops::{Add, Div, Mul, Sub};
+use std::ops::{Add, Sub};
 use std::str::FromStr;
 
 use chrono::{
@@ -11,13 +11,13 @@ use chrono::{
 use chrono_tz::Tz;
 use itertools::Either;
 use mysql_time::MySqlTime;
+use num_traits::cast::FromPrimitive;
 use readyset_data::dialect::SqlEngine;
 use readyset_data::{Array, Collation, DfType, DfValue, TimestampTz};
+use readyset_decimal::{Decimal, RoundingMode};
 use readyset_errors::{internal, invalid_query_err, unsupported, ReadySetError, ReadySetResult};
 use readyset_sql::ast::TimestampField;
 use readyset_util::math::integer_rnd;
-use rust_decimal::prelude::{FromPrimitive, ToPrimitive};
-use rust_decimal::Decimal;
 use serde_json::Value as JsonValue;
 use test_strategy::Arbitrary;
 use vec1::Vec1;
@@ -891,14 +891,12 @@ impl BuiltinFunction {
                     DfValue::UnsignedInt(inner) => inner as i32,
                     DfValue::Float(f) => f.round() as i32,
                     DfValue::Double(f) => f.round() as i32,
-                    DfValue::Numeric(d) => {
-                        // TODO(fran): I don't know if this is the right thing to do.
-                        d.round().to_i32().ok_or_else(|| {
-                            ReadySetError::BadRequest(format!(
-                                "NUMERIC value {d} exceeds 32-bit integer size"
-                            ))
-                        })?
-                    }
+                    // XXX(mvzink): Yes, MySQL does round the second argument (up) if it's a decimal
+                    // before using that to round the first argument. However, Postgres throws an
+                    // error, so we are not matching that.
+                    DfValue::Numeric(d) => d
+                        .round_dp_with_strategy(0, RoundingMode::HalfUp)
+                        .try_into()?,
                     _ => 0,
                 };
 
@@ -938,22 +936,10 @@ impl BuiltinFunction {
                         Ok(DfValue::Int(rounded as _))
                     }
                     DfValue::Numeric(d) => {
-                        let rounded_dec = if rnd_prec >= 0 {
-                            d.round_dp_with_strategy(
-                                rnd_prec as _,
-                                rust_decimal::RoundingStrategy::MidpointAwayFromZero,
-                            )
-                        } else {
-                            let factor = Decimal::from_f64(10.0f64.powf(-rnd_prec as _)).unwrap();
-
-                            d.div(factor)
-                                .round_dp_with_strategy(
-                                    0,
-                                    rust_decimal::RoundingStrategy::MidpointAwayFromZero,
-                                )
-                                .mul(factor)
-                        };
-
+                        let rounded_dec = d.round_dp_with_strategy(
+                            rnd_prec as _,
+                            readyset_decimal::RoundingMode::HalfUp,
+                        );
                         Ok(DfValue::Numeric(rounded_dec.into()))
                     }
                     dt => {
@@ -1966,29 +1952,37 @@ mod tests {
         let expr = parse_and_lower("round(c0, c1)", MySQL);
         assert_eq!(
             expr.eval::<DfValue>(&[
-                DfValue::from(Decimal::from_f64(52.123).unwrap()),
+                DfValue::from(Decimal::try_from(52.123).unwrap()),
                 DfValue::from(1)
             ])
             .unwrap(),
-            DfValue::from(Decimal::from_f64(52.1)),
+            DfValue::from(Decimal::from_str("52.1").unwrap()),
         );
 
         assert_eq!(
             expr.eval::<DfValue>(&[
-                DfValue::from(Decimal::from_f64(-52.666).unwrap()),
+                DfValue::from(Decimal::try_from(-52.666).unwrap()),
                 DfValue::from(2)
             ])
             .unwrap(),
-            DfValue::from(Decimal::from_f64(-52.67)),
+            DfValue::from(Decimal::from_str("-52.67").unwrap()),
         );
 
         assert_eq!(
             expr.eval::<DfValue>(&[
-                DfValue::from(Decimal::from_f64(-52.666).unwrap()),
+                DfValue::from(Decimal::try_from(-52.666).unwrap()),
                 DfValue::from(-1)
             ])
             .unwrap(),
-            DfValue::from(Decimal::from_f64(-50.)),
+            DfValue::from(Decimal::try_from(-50.).unwrap()),
+        );
+        assert_eq!(
+            expr.eval::<DfValue>(&[
+                DfValue::from(Decimal::try_from(-5266.666).unwrap()),
+                DfValue::from(-2)
+            ])
+            .unwrap(),
+            DfValue::from(Decimal::try_from(-5300.).unwrap()),
         );
     }
 

dataflow-expression/src/eval/builtins.rs

@@ -42,16 +42,13 @@ readyset-util = { path = "../readyset-util" }
 replication-offset = { path = "../replication-offset" }
 
 [dev-dependencies]
-proptest-stateful = { path = "../proptest-stateful" }
 async-trait = { workspace = true }
-pretty_assertions = { workspace = true }
-lazy_static = { workspace = true }
 criterion = { workspace = true, features = ["async_tokio"] }
-rust_decimal = { workspace = true, features = [
-    "db-tokio-postgres",
-    "serde-str",
-] }
+lazy_static = { workspace = true }
+pretty_assertions = { workspace = true }
 test-utils = { path = "../test-utils" }
+proptest-stateful = { path = "../proptest-stateful" }
+readyset-decimal = { path = "../readyset-decimal" }
 
 [[bench]]
 name = "persistent_state"

dataflow-state/Cargo.toml

@@ -2498,11 +2498,16 @@ mod tests {
     use std::path::PathBuf;
 
     use pretty_assertions::assert_eq;
+
     use readyset_data::Bound::*;
+
     use readyset_data::Collation;
+
     use replication_offset::mysql::MySqlPosition;
+
     use replication_offset::postgres::PostgresPosition;
-    use rust_decimal::Decimal;
+
+    use readyset_decimal::Decimal;
 
     use super::*;
 
@@ -2709,7 +2714,7 @@ mod tests {
         let mut state = setup_persistent("lookup_numeric_with_different_precision", None);
         state.add_index(Index::btree_map(vec![0]), None);
 
-        let records = vec![vec![DfValue::from(Decimal::from_str_exact("4.0").unwrap())]];
+        let records = vec![vec![DfValue::from(Decimal::from_str("4.0").unwrap())]];
 
         state
             .process_records(&mut records.clone().into(), None, None)
@@ -2718,13 +2723,13 @@ mod tests {
         let res = state
             .lookup(
                 &[0],
-                &PointKey::Single(DfValue::from(Decimal::from_str_exact("4").unwrap())),
+                &PointKey::Single(DfValue::from(Decimal::from_str("4").unwrap())),
             )
             .unwrap();
 
         assert_eq!(res, records.into());
         let val = Decimal::try_from(&res.into_iter().next().unwrap()[0]).unwrap();
-        assert_eq!(val.scale(), 1);
+        assert_eq!(val.scale(), Some(1));
     }
 
     #[test]

dataflow-state/src/persistent_state/mod.rs

@@ -132,24 +132,26 @@ deny = [
     #{ name = "ansi_term", version = "=0.11.0", wrappers = [] },
 ]
 # Certain crates/versions that will be skipped when doing duplicate detection.
-skip = []
+skip = [
+    { name = "bit-vec", version = "0.6.3" },
+    { name = "uuid", version = "0.8.2" },
+]
 # Similarly to `skip` allows you to skip certain crates during duplicate
 # detection. Unlike skip, it also includes the entire tree of transitive
 # dependencies starting at the specified crate, up to a certain depth, which is
 # by default infinite
 skip-tree = [
-    { name = "rusty-fork", version = "0.3.0" },
-    { name = "fail", version = "0.5.0" },
-    { name = "rust_decimal", version = "1.17.0" },
     { name = "consulrs", version = "0.1.0" },
+    { name = "fail", version = "0.5.0" },
     { name = "hashbrown", version = "0.12.3" },
-    { name = "unicode-width", version = "0.1.14" },
-    { name = "web-time", version = "0.2.4" },
+    { name = "hermit-abi", version = "0.3.9" },
     { name = "itertools", version = "0.10.5" },
     { name = "itertools", version = "0.12.1" },
-    { name = "winnow", version = "0.6.26" },
     { name = "regex-automata", version = "0.1.10" },
-    { name = "hermit-abi", version = "0.3.9" },
+    { name = "rusty-fork", version = "0.3.0" },
+    { name = "unicode-width", version = "0.1.14" },
+    { name = "web-time", version = "0.2.4" },
+    { name = "winnow", version = "0.6.26" },
 ]
 
 # This section is considered when running `cargo deny check sources`.
@@ -166,9 +168,7 @@ unknown-git = "warn"
 # if not specified. If it is specified but empty, no registries are allowed.
 allow-registry = ["https://github.com/rust-lang/crates.io-index"]
 # List of URLs for allowed Git repositories
-allow-git = [
-    "git+https://github.com/apache/datafusion-sqlparser-rs",
-]
+allow-git = ["git+https://github.com/apache/datafusion-sqlparser-rs"]
 
 [sources.allow-org]
 # 1 or more github.com organizations to allow git sources for

deny.toml

@@ -26,7 +26,6 @@ postgres-types = { workspace = true, features = [
     "with-cidr-0_2",
 ] }
 rand = { workspace = true }
-rust_decimal = { workspace = true }
 serde_json = { workspace = true, features = ["arbitrary_precision"] }
 sha2 = { workspace = true }
 smallvec = { workspace = true }
@@ -48,6 +47,7 @@ uuid = { workspace = true }
 database-utils = { path = "../database-utils" }
 readyset-adapter-types = { path = "../readyset-adapter-types" }
 readyset-data = { path = "../readyset-data" }
+readyset-decimal = { path = "../readyset-decimal" }
 readyset-sql = { path = "../readyset-sql" }
 readyset-util = { path = "../readyset-util" }