feat: support decimal

questdb-rs-ffi/src/lib.rs

@@ -224,6 +224,9 @@ pub enum line_sender_error_code {
 
     /// Line sender protocol version error.
     line_sender_error_protocol_version_error,
+
+    /// The supplied decimal is invalid.
+    line_sender_error_invalid_decimal,
 }
 
 impl From<ErrorCode> for line_sender_error_code {
@@ -252,6 +255,9 @@ impl From<ErrorCode> for line_sender_error_code {
             ErrorCode::ProtocolVersionError => {
                 line_sender_error_code::line_sender_error_protocol_version_error
             }
+            ErrorCode::InvalidDecimal => {
+                line_sender_error_code::line_sender_error_invalid_decimal
+            }
         }
     }
 }

questdb-rs/Cargo.toml

@@ -28,18 +28,26 @@ itoa = "1.0"
 aws-lc-rs = { version = "1.13", optional = true }
 ring = { version = "0.17.14", optional = true }
 rustls-pki-types = "1.0.1"
-rustls = { version = "0.23.25", default-features = false, features = ["logging", "std", "tls12"] }
+rustls = { version = "0.23.25", default-features = false, features = [
+    "logging",
+    "std",
+    "tls12",
+] }
 rustls-native-certs = { version = "0.8.1", optional = true }
 webpki-roots = { version = "1.0.1", default-features = false, optional = true }
 chrono = { version = "0.4.40", optional = true }
 
 # We need to limit the `ureq` version to 3.0.x since we use
 # the `ureq::unversioned` module which does not respect semantic versioning.
-ureq = { version = "3.0.10, <3.1.0", default-features = false, features = ["_tls"], optional = true }
+ureq = { version = "3.0.10, <3.1.0", default-features = false, features = [
+    "_tls",
+], optional = true }
 serde_json = { version = "1", optional = true }
 questdb-confstr = "0.1.1"
 rand = { version = "0.9.0", optional = true }
 ndarray = { version = "0.16", optional = true }
+rust_decimal = { version = "1.38.0", optional = true }
+bigdecimal = { version = "0.4.8", optional = true }
 
 [target.'cfg(windows)'.dependencies]
 winapi = { version = "0.3.9", features = ["ws2def"] }
@@ -68,7 +76,13 @@ sync-sender = ["sync-sender-tcp", "sync-sender-http"]
 sync-sender-tcp = ["_sync-sender", "_sender-tcp", "dep:socket2"]
 
 ## Sync ILP/HTTP
-sync-sender-http = ["_sync-sender", "_sender-http", "dep:ureq", "dep:serde_json", "dep:rand"]
+sync-sender-http = [
+    "_sync-sender",
+    "_sender-http",
+    "dep:ureq",
+    "dep:serde_json",
+    "dep:rand",
+]
 
 ## Allow use OS-provided root TLS certificates
 tls-native-certs = ["dep:rustls-native-certs"]
@@ -91,6 +105,12 @@ json_tests = []
 ## Enable methods to create timestamp objects from chrono::DateTime objects.
 chrono_timestamp = ["chrono"]
 
+## Enable serialization of rust_decimal::Decimal in ILP
+rust_decimal = ["dep:rust_decimal"]
+
+## Enable serialization of bigdecimal::BigDecimal in ILP
+bigdecimal = ["dep:bigdecimal"]
+
 # Hidden derived features, used in code to enable-disable code sections. Don't use directly.
 _sender-tcp = []
 _sender-http = []
@@ -109,7 +129,9 @@ almost-all-features = [
     "insecure-skip-verify",
     "json_tests",
     "chrono_timestamp",
-    "ndarray"
+    "ndarray",
+    "rust_decimal",
+    "bigdecimal",
 ]
 
 [[example]]
@@ -126,8 +148,8 @@ required-features = ["chrono_timestamp"]
 
 [[example]]
 name = "http"
-required-features = ["sync-sender-http", "ndarray"]
+required-features = ["sync-sender-http", "ndarray", "rust_decimal"]
 
 [[example]]
 name = "protocol_version"
-required-features = ["sync-sender-http", "ndarray"]
+required-features = ["sync-sender-http", "ndarray", "bigdecimal"]

questdb-rs/examples/basic.rs

@@ -8,15 +8,15 @@ use questdb::{
 fn main() -> Result<()> {
     let host: String = std::env::args().nth(1).unwrap_or("localhost".to_string());
     let port: &str = &std::env::args().nth(2).unwrap_or("9009".to_string());
-    let mut sender = Sender::from_conf(format!("tcp::addr={host}:{port};"))?;
+    let mut sender = Sender::from_conf(format!("tcp::addr={host}:{port};protocol_version=2;"))?;
     let mut buffer = sender.new_buffer();
     let designated_timestamp =
         TimestampNanos::from_datetime(Utc.with_ymd_and_hms(1997, 7, 4, 4, 56, 55).unwrap())?;
     buffer
         .table("trades")?
         .symbol("symbol", "ETH-USD")?
         .symbol("side", "sell")?
-        .column_f64("price", 2615.54)?
+        .column_decimal("price", "2615.54")?
         .column_f64("amount", 0.00044)?
         // QuestDB server version 9.0.0 or later is required for array support.
         .column_arr("location", &arr1(&[100.0, 100.1, 100.2]).view())?

questdb-rs/examples/http.rs

@@ -1,17 +1,23 @@
+use std::str::FromStr;
+
 use ndarray::arr1;
 use questdb::{
     ingress::{Sender, TimestampNanos},
     Result,
 };
+use rust_decimal::Decimal;
 
 fn main() -> Result<()> {
-    let mut sender = Sender::from_conf("https::addr=localhost:9000;username=foo;password=bar;")?;
+    let mut sender = Sender::from_conf(
+        "https::addr=localhost:9000;username=foo;password=bar;protocol_version=2;",
+    )?;
     let mut buffer = sender.new_buffer();
+    let price = Decimal::from_str("2615.54").unwrap();
     buffer
         .table("trades")?
         .symbol("symbol", "ETH-USD")?
         .symbol("side", "sell")?
-        .column_f64("price", 2615.54)?
+        .column_decimal("price", &price)?
         .column_f64("amount", 0.00044)?
         // QuestDB server version 9.0.0 or later is required for array support.
         .column_arr("location", &arr1(&[100.0, 100.1, 100.2]).view())?

questdb-rs/examples/protocol_version.rs

@@ -1,33 +1,38 @@
+use std::str::FromStr;
+
+use bigdecimal::BigDecimal;
 use ndarray::arr1;
 use questdb::{
     ingress::{Sender, TimestampNanos},
     Result,
 };
 
 fn main() -> Result<()> {
+    let price = BigDecimal::from_str("2615.54").unwrap();
+
     let mut sender = Sender::from_conf(
-        "https::addr=localhost:9000;username=foo;password=bar;protocol_version=1;",
+        "http::addr=localhost:9000;username=foo;password=bar;protocol_version=1;",
     )?;
     let mut buffer = sender.new_buffer();
     buffer
         .table("trades_ilp_v1")?
         .symbol("symbol", "ETH-USD")?
         .symbol("side", "sell")?
-        .column_f64("price", 2615.54)?
+        .column_decimal("price", &price)?
         .column_f64("amount", 0.00044)?
         .at(TimestampNanos::now())?;
     sender.flush(&mut buffer)?;
 
     // QuestDB server version 9.0.0 or later is required for `protocol_version=2` support.
     let mut sender2 = Sender::from_conf(
-        "https::addr=localhost:9000;username=foo;password=bar;protocol_version=2;",
+        "http::addr=localhost:9000;username=foo;password=bar;protocol_version=2;",
     )?;
-    let mut buffer2 = sender.new_buffer();
+    let mut buffer2 = sender2.new_buffer();
     buffer2
         .table("trades_ilp_v2")?
         .symbol("symbol", "ETH-USD")?
         .symbol("side", "sell")?
-        .column_f64("price", 2615.54)?
+        .column_decimal("price", &price)?
         .column_f64("amount", 0.00044)?
         .column_arr("location", &arr1(&[100.0, 100.1, 100.2]).view())?
         .at(TimestampNanos::now())?;

questdb-rs/src/error.rs

@@ -78,6 +78,9 @@ pub enum ErrorCode {
 
     /// Validate protocol version error.
     ProtocolVersionError,
+
+    /// The supplied decimal is invalid.
+    InvalidDecimal,
 }
 
 /// An error that occurred when using QuestDB client library.

questdb-rs/src/ingress/buffer.rs

@@ -21,6 +21,7 @@
  *  limitations under the License.
  *
  ******************************************************************************/
+use crate::ingress::decimal::DecimalSerializer;
 use crate::ingress::ndarr::{check_and_get_array_bytes_size, ArrayElementSealed};
 use crate::ingress::{
     ndarr, ArrayElement, DebugBytes, NdArrayView, ProtocolVersion, Timestamp, TimestampNanos,
@@ -71,7 +72,7 @@ where
     quoting_fn(output);
 }
 
-fn must_escape_unquoted(c: u8) -> bool {
+pub(crate) fn must_escape_unquoted(c: u8) -> bool {
     matches!(c, b' ' | b',' | b'=' | b'\n' | b'\r' | b'\\')
 }
 
@@ -974,6 +975,92 @@ impl Buffer {
         Ok(self)
     }
 
+    /// Record a decimal value for the given column.
+    ///
+    /// ```no_run
+    /// # use questdb::Result;
+    /// # use questdb::ingress::{Buffer, SenderBuilder};
+    /// # fn main() -> Result<()> {
+    /// # let mut sender = SenderBuilder::from_conf("https::addr=localhost:9000;")?.build()?;
+    /// # let mut buffer = sender.new_buffer();
+    /// # buffer.table("x")?;
+    /// buffer.column_decimal("col_name", "123.45")?;
+    /// # Ok(())
+    /// # }
+    /// ```
+    ///
+    /// or
+    ///
+    /// ```no_run
+    /// # use questdb::Result;
+    /// # use questdb::ingress::{Buffer, SenderBuilder};
+    /// use questdb::ingress::ColumnName;
+    ///
+    /// # fn main() -> Result<()> {
+    /// # let mut sender = SenderBuilder::from_conf("https::addr=localhost:9000;")?.build()?;
+    /// # let mut buffer = sender.new_buffer();
+    /// # buffer.table("x")?;
+    /// let col_name = ColumnName::new("col_name")?;
+    /// buffer.column_decimal(col_name, "123.45")?;
+    /// # Ok(())
+    /// # }
+    /// ```
+    ///
+    /// With `rust_decimal` feature enabled:
+    ///
+    /// ```no_run
+    /// # #[cfg(feature = "rust_decimal")]
+    /// # {
+    /// # use questdb::Result;
+    /// # use questdb::ingress::{Buffer, SenderBuilder};
+    /// use rust_decimal::Decimal;
+    /// use std::str::FromStr;
+    ///
+    /// # fn main() -> Result<()> {
+    /// # let mut sender = SenderBuilder::from_conf("https::addr=localhost:9000;")?.build()?;
+    /// # let mut buffer = sender.new_buffer();
+    /// # buffer.table("x")?;
+    /// let value = Decimal::from_str("123.45")?;
+    /// buffer.column_decimal("col_name", &value)?;
+    /// # Ok(())
+    /// # }
+    /// # }
+    /// ```
+    ///
+    /// With `bigdecimal` feature enabled:
+    ///
+    /// ```no_run
+    /// # #[cfg(feature = "bigdecimal")]
+    /// # {
+    /// # use questdb::Result;
+    /// # use questdb::ingress::{Buffer, SenderBuilder};
+    /// use bigdecimal::BigDecimal;
+    /// use std::str::FromStr;
+    ///
+    /// # fn main() -> Result<()> {
+    /// # let mut sender = SenderBuilder::from_conf("https::addr=localhost:9000;")?.build()?;
+    /// # let mut buffer = sender.new_buffer();
+    /// # buffer.table("x")?;
+    /// let value = BigDecimal::from_str("0.123456789012345678901234567890")?;
+    /// buffer.column_decimal("col_name", &value)?;
+    /// # Ok(())
+    /// # }
+    /// # }
+    /// ```
+    pub fn column_decimal<'a, N, S>(&mut self, name: N, value: S) -> crate::Result<&mut Self>
+    where
+        N: TryInto<ColumnName<'a>>,
+        S: DecimalSerializer,
+        Error: From<N::Error>,
+    {
+        self.write_column_key(name)?;
+        value.serialize(
+            &mut self.output,
+            self.protocol_version.supports_binary_encoding(),
+        )?;
+        Ok(self)
+    }
+
     /// Record a multidimensional array value for the given column.
     ///
     /// Supports arrays with up to [`MAX_ARRAY_DIMS`] dimensions. The array elements must