fix(ser): fix i128/u128 SQL bind serialization

Integer types larger than 64 bits cannot be written as literals without casting,
or they end up `Float64`. See ClickHouse/ClickHouse#38480

This commit fixes #208

Author: marcin-ptaszynski

src/sql/ser.rs

@@ -95,12 +95,10 @@ impl<'a, W: Write> Serializer for SqlSerializer<'a, W> {
         serialize_i16(i16),
         serialize_i32(i32),
         serialize_i64(i64),
-        serialize_i128(i128),
         serialize_u8(u8),
         serialize_u16(u16),
         serialize_u32(u32),
         serialize_u64(u64),
-        serialize_u128(u128),
         serialize_f32(f32),
         serialize_f64(f64),
         serialize_bool(bool),
@@ -112,6 +110,18 @@ impl<'a, W: Write> Serializer for SqlSerializer<'a, W> {
         self.serialize_str(value.encode_utf8(&mut tmp))
     }
 
+    #[inline]
+    fn serialize_i128(self, value: i128) -> Result {
+        write!(self.writer, "{}::Int128", value)?;
+        Ok(())
+    }
+
+    #[inline]
+    fn serialize_u128(self, value: u128) -> Result {
+        write!(self.writer, "{}::UInt128", value)?;
+        Ok(())
+    }
+
     #[inline]
     fn serialize_str(self, value: &str) -> Result {
         escape::string(value, self.writer)?;
@@ -452,7 +462,8 @@ mod tests {
     fn it_writes_numeric_primitives() {
         assert_eq!(check(42), "42");
         assert_eq!(check(42.5), "42.5");
-        assert_eq!(check(42u128), "42");
+        assert_eq!(check(42u128), "42::UInt128");
+        assert_eq!(check(42i128), "42::Int128");
     }
 
     #[test]

tests/it/int128.rs

@@ -0,0 +1,122 @@
+use rand::random;
+use serde::{Deserialize, Serialize};
+
+use clickhouse::Row;
+
+#[tokio::test]
+async fn u128() {
+    let client = prepare_database!();
+
+    #[derive(Debug, PartialEq, Eq, Serialize, Deserialize, Row)]
+    struct MyRow {
+        id: u128,
+        value: String,
+    }
+
+    client
+        .query(
+            "
+            CREATE TABLE test(
+                id UInt128,
+                value String,
+            ) ENGINE = MergeTree ORDER BY id
+        ",
+        )
+        .execute()
+        .await
+        .unwrap();
+
+    let (id0, id1, id2) = (random(), random(), random());
+    println!("ids: {id0}, {id1}, {id2}");
+
+    let original_rows = vec![
+        MyRow {
+            id: id0,
+            value: "test_0".to_string(),
+        },
+        MyRow {
+            id: id1,
+            value: "test_1".to_string(),
+        },
+        MyRow {
+            id: id2,
+            value: "test_2".to_string(),
+        },
+    ];
+
+    let mut insert = client.insert("test").unwrap();
+    for row in &original_rows {
+        insert.write(row).await.unwrap();
+    }
+    insert.end().await.unwrap();
+
+    let rows = client
+        .query("SELECT ?fields FROM test WHERE id IN ? ORDER BY value")
+        .bind(vec![id0, id2])
+        .fetch_all::<MyRow>()
+        .await
+        .unwrap();
+
+    assert_eq!(rows.len(), 2);
+    assert_eq!(rows[0], original_rows[0]);
+    assert_eq!(rows[1], original_rows[2]);
+}
+
+#[tokio::test]
+async fn i128() {
+    let client = prepare_database!();
+
+    #[derive(Debug, PartialEq, Eq, Serialize, Deserialize, Row)]
+    struct MyRow {
+        id: i128,
+        value: String,
+    }
+
+    client
+        .query(
+            "
+            CREATE TABLE test(
+                id Int128,
+                value String,
+            ) ENGINE = MergeTree ORDER BY id
+        ",
+        )
+        .execute()
+        .await
+        .unwrap();
+
+    let (id0, id1, id2) = (random(), random(), random());
+    println!("ids: {id0}, {id1}, {id2}");
+
+    let original_rows = vec![
+        MyRow {
+            id: id0,
+            value: "test_0".to_string(),
+        },
+        MyRow {
+            id: id1,
+            value: "test_1".to_string(),
+        },
+        MyRow {
+            id: id2,
+            value: "test_2".to_string(),
+        },
+    ];
+
+    let mut insert = client.insert("test").unwrap();
+    for row in &original_rows {
+        insert.write(row).await.unwrap();
+    }
+    insert.end().await.unwrap();
+
+    let rows = client
+        .query("SELECT ?fields FROM test WHERE id IN ? ORDER BY value")
+        .bind(vec![id0, id2])
+        .fetch_all::<MyRow>()
+        .await
+        .unwrap();
+
+    assert_eq!(rows.len(), 2);
+    assert_eq!(rows[0], original_rows[0]);
+    assert_eq!(rows[1], original_rows[2]);
+}

tests/it/main.rs

@@ -61,6 +61,7 @@ mod cursor_stats;
 mod fetch_bytes;
 mod insert;
 mod inserter;
+mod int128;
 mod ip;
 mod mock;
 mod nested;