Repository: scylladb/scylla-rust-driver
Pull request #185 (2021-03-03 21:40:02)

CQL missing types
## Description
Adding CQL missing types: varint, counter, decimal. 
Varchar was already added (named Text: `0x000D => Text,`)

### Fixes: #181

## Pre-review checklist

<!--
    Make sure you took care of the issues on the list.
    Put 'x' into those boxes which apply.
    You can also create the PR now and click on all relevant checkboxes.
    See CONTRIBUTING.md for more details.
-->

- [x] I have split my patch into logically separate commits.
- [x] All commit messages clearly explain what they change and why.
- [x] I added relevant tests for new features and bug fixes.
- [x] All commits compile, pass static checks and pass test.
- [x] PR description sums up the changes and reasons why they should be introduced.
- [x] I added appropriate `Fixes:` annotations to PR description.



Patch:
diff --git a/scylla/Cargo.toml b/scylla/Cargo.toml
index 840affe8..50caec97 100644
--- a/scylla/Cargo.toml
+++ b/scylla/Cargo.toml
@@ -23,6 +23,8 @@ rand = "0.8.3"
 serde_json = "1.0.60"
 thiserror = "1.0"
 itertools = "0.10.0"
+bigdecimal = "0.2.0"
+num-bigint = "0.3"
 
 [dev-dependencies]
 criterion = "0.3"
diff --git a/scylla/src/frame/response/cql_to_rust.rs b/scylla/src/frame/response/cql_to_rust.rs
index 715bb568..a4165814 100644
--- a/scylla/src/frame/response/cql_to_rust.rs
+++ b/scylla/src/frame/response/cql_to_rust.rs
@@ -1,4 +1,7 @@
 use super::result::{CQLValue, Row};
+use crate::frame::value::Counter;
+use bigdecimal::BigDecimal;
+use num_bigint::BigInt;
 use std::collections::HashMap;
 use std::hash::Hash;
 use std::net::IpAddr;
@@ -19,7 +22,7 @@ pub enum CQLTypeError {
     InvalidNumberOfElements(i32),
 }
 
-/// This trait defines a way to convert CQLValue or Option<CQLValue> into some rust type  
+/// This trait defines a way to convert CQLValue or Option<CQLValue> into some rust type
 // We can't use From trait because impl From<Option<CQLValue>> for String {...}
 // is forbidden since neither From nor String are defined in this crate
 pub trait FromCQLVal<T>: Sized {
@@ -70,7 +73,9 @@ macro_rules! impl_from_cql_val {
 
 impl_from_cql_val!(i32, as_int); // i32::from_cql<CQLValue>
 impl_from_cql_val!(i64, as_bigint); // i64::from_cql<CQLValue>
+impl_from_cql_val!(Counter, as_counter); // Counter::from_cql<CQLValue>
 impl_from_cql_val!(i16, as_smallint); // i16::from_cql<CQLValue>
+impl_from_cql_val!(BigInt, into_varint); // BigInt::from_cql<CQLValue>
 impl_from_cql_val!(i8, as_tinyint); // i8::from_cql<CQLValue>
 impl_from_cql_val!(u32, as_date); // u32::from_cql<CQLValue>
 impl_from_cql_val!(f32, as_float); // f32::from_cql<CQLValue>
@@ -79,6 +84,7 @@ impl_from_cql_val!(bool, as_boolean); // bool::from_cql<CQLValue>
 impl_from_cql_val!(String, into_string); // String::from_cql<CQLValue>
 impl_from_cql_val!(IpAddr, as_inet); // IpAddr::from_cql<CQLValue>
 impl_from_cql_val!(Uuid, as_uuid); // Uuid::from_cql<CQLValue>
+impl_from_cql_val!(BigDecimal, into_decimal); // BigDecimal::from_cql<CQLValue>
 
 // Vec<T>::from_cql<CQLValue>
 impl<T: FromCQLVal<CQLValue>> FromCQLVal<CQLValue> for Vec<T> {
@@ -191,19 +197,54 @@ impl_tuple_from_cql!(T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14
 mod tests {
     use super::{CQLValue, FromCQLVal, FromCQLValError, FromRow, FromRowError, Row};
     use crate as scylla;
+    use crate::frame::value::Counter;
     use crate::macros::FromRow;
+    use bigdecimal::BigDecimal;
+    use num_bigint::{BigInt, ToBigInt};
     use std::net::{IpAddr, Ipv4Addr};
+    use std::str::FromStr;
+    use uuid::Uuid;
+
+    #[test]
+    fn uuid_from_cql() {
+        let my_uuid = Uuid::parse_str("936DA01F9ABD4d9d80C702AF85C822A8").unwrap();
+        assert_eq!(Ok(my_uuid), Uuid::from_cql(CQLValue::Uuid(my_uuid)));
+    }
 
     #[test]
     fn i32_from_cql() {
         assert_eq!(Ok(1234), i32::from_cql(CQLValue::Int(1234)));
     }
 
+    #[test]
+    fn bool_from_cql() {
+        assert_eq!(Ok(true), bool::from_cql(CQLValue::Boolean(true)));
+        assert_eq!(Ok(false), bool::from_cql(CQLValue::Boolean(false)));
+    }
+
+    #[test]
+    fn floatingpoints_from_cql() {
+        let float: f32 = 2.13;
+        let double: f64 = 4.26;
+        assert_eq!(Ok(float), f32::from_cql(CQLValue::Float(float)));
+        assert_eq!(Ok(double), f64::from_cql(CQLValue::Double(double)));
+    }
+
     #[test]
     fn i64_from_cql() {
         assert_eq!(Ok(1234), i64::from_cql(CQLValue::BigInt(1234)));
     }
 
+    #[test]
+    fn i8_from_cql() {
+        assert_eq!(Ok(6), i8::from_cql(CQLValue::TinyInt(6)));
+    }
+
+    #[test]
+    fn i16_from_cql() {
+        assert_eq!(Ok(16), i16::from_cql(CQLValue::SmallInt(16)));
+    }
+
     #[test]
     fn string_from_cql() {
         assert_eq!(
@@ -222,6 +263,30 @@ mod tests {
         assert_eq!(Ok(ip_addr), IpAddr::from_cql(CQLValue::Inet(ip_addr)));
     }
 
+    #[test]
+    fn varint_from_cql() {
+        let big_int = 0.to_bigint().unwrap();
+        assert_eq!(
+            Ok(big_int),
+            BigInt::from_cql(CQLValue::Varint(0.to_bigint().unwrap()))
+        );
+    }
+
+    #[test]
+    fn decimal_from_cql() {
+        let decimal = BigDecimal::from_str("123.4").unwrap();
+        assert_eq!(
+            Ok(decimal.clone()),
+            BigDecimal::from_cql(CQLValue::Decimal(decimal))
+        );
+    }
+
+    #[test]
+    fn counter_from_cql() {
+        let counter = Counter(1);
+        assert_eq!(Ok(counter), Counter::from_cql(CQLValue::Counter(counter)));
+    }
+
     #[test]
     fn vec_from_cql() {
         let cql_val = CQLValue::Set(vec![CQLValue::Int(1), CQLValue::Int(2), CQLValue::Int(3)]);
diff --git a/scylla/src/frame/response/result.rs b/scylla/src/frame/response/result.rs
index 756361a6..ca9535df 100644
--- a/scylla/src/frame/response/result.rs
+++ b/scylla/src/frame/response/result.rs
@@ -1,7 +1,10 @@
 use crate::cql_to_rust::{FromRow, FromRowError};
+use crate::frame::value::Counter;
 use crate::frame::{frame_errors::ParseError, types};
+use bigdecimal::BigDecimal;
 use byteorder::{BigEndian, ReadBytesExt};
 use bytes::{Buf, Bytes};
+use num_bigint::BigInt;
 use std::{
     collections::BTreeMap,
     convert::{TryFrom, TryInto},
@@ -39,7 +42,9 @@ enum ColumnType {
     Ascii,
     Boolean,
     Blob,
+    Counter,
     Date,
+    Decimal,
     Double,
     Float,
     Int,
@@ -61,6 +66,7 @@ enum ColumnType {
     Timeuuid,
     Tuple(Vec<ColumnType>),
     Uuid,
+    Varint,
 }
 
 #[derive(Debug, PartialEq)]
@@ -68,7 +74,9 @@ pub enum CQLValue {
     Ascii(String),
     Boolean(bool),
     Blob(Vec<u8>),
+    Counter(Counter),
     Date(u32),
+    Decimal(BigDecimal),
     Double(f64),
     Float(f32),
     Int(i32),
@@ -90,6 +98,7 @@ pub enum CQLValue {
     Timeuuid(Uuid),
     Tuple(Vec<CQLValue>),
     Uuid(Uuid),
+    Varint(BigInt),
 }
 
 impl CQLValue {
@@ -114,6 +123,13 @@ impl CQLValue {
         }
     }
 
+    pub fn as_counter(&self) -> Option<Counter> {
+        match self {
+            Self::Counter(i) => Some(*i),
+            _ => None,
+        }
+    }
+
     pub fn as_date(&self) -> Option<u32> {
         match self {
             Self::Date(i) => Some(*i),
@@ -242,6 +258,19 @@ impl CQLValue {
         }
     }
 
+    pub fn into_varint(self) -> Option<BigInt> {
+        match self {
+            Self::Varint(i) => Some(i),
+            _ => None,
+        }
+    }
+
+    pub fn into_decimal(self) -> Option<BigDecimal> {
+        match self {
+            Self::Decimal(i) => Some(i),
+            _ => None,
+        }
+    }
     // TODO
 }
 
@@ -311,12 +340,15 @@ fn deser_type(buf: &mut &[u8]) -> StdResult<ColumnType, ParseError> {
         0x0002 => BigInt,
         0x0003 => Blob,
         0x0004 => Boolean,
+        0x0005 => Counter,
+        0x0006 => Decimal,
         0x0007 => Double,
         0x0008 => Float,
         0x0009 => Int,
         0x000B => Timestamp,
         0x000C => Uuid,
         0x000D => Text,
+        0x000E => Varint,
         0x000F => Timeuuid,
         0x0010 => Inet,
         0x0011 => Date,
@@ -477,6 +509,22 @@ fn deser_cql_value(typ: &ColumnType, buf: &mut &[u8]) -> StdResult<CQLValue, Par
             }
             CQLValue::Date(buf.read_u32::<BigEndian>()?)
         }
+        Counter => {
+            if buf.len() != 8 {
+                return Err(ParseError::BadData(format!(
+                    "Buffer length should be 8 not {}",
+                    buf.len()
+                )));
+            }
+            CQLValue::Counter(crate::frame::value::Counter(buf.read_i64::<BigEndian>()?))
+        }
+        Decimal => {
+            let scale = types::read_int(buf)? as i64;
+            let int_value = num_bigint::BigInt::from_signed_bytes_be(buf);
+            let big_decimal: BigDecimal = BigDecimal::from((int_value, scale));
+
+            CQLValue::Decimal(big_decimal)
+        }
         Double => {
             if buf.len() != 8 {
                 return Err(ParseError::BadData(format!(
@@ -589,6 +637,7 @@ fn deser_cql_value(typ: &ColumnType, buf: &mut &[u8]) -> StdResult<CQLValue, Par
             let uuid = uuid::Uuid::from_slice(buf).expect("Deserializing Uuid failed.");
             CQLValue::Uuid(uuid)
         }
+        Varint => CQLValue::Varint(num_bigint::BigInt::from_signed_bytes_be(buf)),
         List(type_name) => {
             let len: usize = types::read_int(buf)?.try_into()?;
             let mut res = Vec::with_capacity(len);
@@ -728,7 +777,217 @@ pub fn deserialize(buf: &mut &[u8]) -> StdResult<Result, ParseError> {
 #[cfg(test)]
 mod tests {
     use crate as scylla;
-    use scylla::frame::response::result::CQLValue;
+    use crate::frame::value::Counter;
+    use bigdecimal::BigDecimal;
+    use num_bigint::BigInt;
+    use num_bigint::ToBigInt;
+    use scylla::frame::response::result::{CQLValue, ColumnType};
+    use std::str::FromStr;
+    use uuid::Uuid;
+
+    #[test]
+    fn test_deserialize_text_types() {
+        let mut buf = Vec::<u8>::new();
+        buf.push(0x41);
+        let int_slice = &mut &buf[..];
+        let ascii_serialized = super::deser_cql_value(&ColumnType::Ascii, int_slice).unwrap();
+        let text_serialized = super::deser_cql_value(&ColumnType::Text, int_slice).unwrap();
+        assert_eq!(ascii_serialized, CQLValue::Ascii("A".to_string()));
+        assert_eq!(text_serialized, CQLValue::Text("A".to_string()));
+    }
+
+    #[test]
+    fn test_deserialize_uuid_inet_types() {
+        let my_uuid = Uuid::parse_str("00000000000000000000000000000001").unwrap();
+
+        let uuid_buf: Vec<u8> = vec![0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1];
+        let uuid_slice = &mut &uuid_buf[..];
+        let uuid_serialize = super::deser_cql_value(&ColumnType::Uuid, uuid_slice).unwrap();
+        assert_eq!(uuid_serialize, CQLValue::Uuid(my_uuid));
+
+        let time_uuid_serialize =
+            super::deser_cql_value(&ColumnType::Timeuuid, uuid_slice).unwrap();
+        assert_eq!(time_uuid_serialize, CQLValue::Timeuuid(my_uuid));
+
+        let my_ip = "::1".parse().unwrap();
+        let ip_buf: Vec<u8> = vec![0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1];
+        let ip_slice = &mut &ip_buf[..];
+        let ip_serialize = super::deser_cql_value(&ColumnType::Inet, ip_slice).unwrap();
+        assert_eq!(ip_serialize, CQLValue::Inet(my_ip));
+
+        let max_ip = "ffff:ffff:ffff:ffff:ffff:ffff:ffff:ffff".parse().unwrap();
+        let max_ip_buf: Vec<u8> = vec![
+            255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255,
+        ];
+        let max_ip_slice = &mut &max_ip_buf[..];
+        let max_ip_serialize = super::deser_cql_value(&ColumnType::Inet, max_ip_slice).unwrap();
+        assert_eq!(max_ip_serialize, CQLValue::Inet(max_ip));
+    }
+
+    #[test]
+    fn test_floating_points() {
+        let float: f32 = 0.5;
+        let double: f64 = 2.0;
+
+        let float_buf: Vec<u8> = vec![63, 0, 0, 0];
+        let float_slice = &mut &float_buf[..];
+        let float_serialize = super::deser_cql_value(&ColumnType::Float, float_slice).unwrap();
+        assert_eq!(float_serialize, CQLValue::Float(float));
+
+        let double_buf: Vec<u8> = vec![64, 0, 0, 0, 0, 0, 0, 0];
+        let double_slice = &mut &double_buf[..];
+        let double_serialize = super::deser_cql_value(&ColumnType::Double, double_slice).unwrap();
+        assert_eq!(double_serialize, CQLValue::Double(double));
+    }
+
+    #[test]
+    fn test_varint() {
+        struct Test<'a> {
+            value: BigInt,
+            encoding: &'a [u8],
+        }
+
+        /*
+            Table taken from CQL Binary Protocol v4 spec
+
+            Value | Encoding
+            ------|---------
+                0 |     0x00
+                1 |     0x01
+              127 |     0x7F
+              128 |   0x0080
+              129 |   0x0081
+               -1 |     0xFF
+             -128 |     0x80
+             -129 |   0xFF7F
+        */
+        let tests = [
+            Test {
+                value: 0.to_bigint().unwrap(),
+                encoding: &[0x00],
+            },
+            Test {
+                value: 1.to_bigint().unwrap(),
+                encoding: &[0x01],
+            },
+            Test {
+                value: 127.to_bigint().unwrap(),
+                encoding: &[0x7F],
+            },
+            Test {
+                value: 128.to_bigint().unwrap(),
+                encoding: &[0x00, 0x80],
+            },
+            Test {
+                value: 129.to_bigint().unwrap(),
+                encoding: &[0x00, 0x81],
+            },
+            Test {
+                value: (-1).to_bigint().unwrap(),
+                encoding: &[0xFF],
+            },
+            Test {
+                value: (-128).to_bigint().unwrap(),
+                encoding: &[0x80],
+            },
+            Test {
+                value: (-129).to_bigint().unwrap(),
+                encoding: &[0xFF, 0x7F],
+            },
+        ];
+
+        for t in tests.iter() {
+            let value = super::deser_cql_value(&ColumnType::Varint, &mut &t.encoding[..]).unwrap();
+            assert_eq!(CQLValue::Varint(t.value.clone()), value);
+        }
+    }
+
+    #[test]
+    fn test_decimal() {
+        struct Test<'a> {
+            value: BigDecimal,
+            encoding: &'a [u8],
+        }
+
+        let tests = [
+            Test {
+                value: BigDecimal::from_str("-1.28").unwrap(),
+                encoding: &[0x0, 0x0, 0x0, 0x2, 0x80],
+            },
+            Test {
+                value: BigDecimal::from_str("1.29").unwrap(),
+                encoding: &[0x0, 0x0, 0x0, 0x2, 0x0, 0x81],
+            },
+            Test {
+                value: BigDecimal::from_str("0").unwrap(),
+                encoding: &[0x0, 0x0, 0x0, 0x0, 0x0],
+            },
+            Test {
+                value: BigDecimal::from_str("123").unwrap(),
+                encoding: &[0x0, 0x0, 0x0, 0x0, 0x7b],
+            },
+        ];
+
+        for t in tests.iter() {
+            let value = super::deser_cql_value(&ColumnType::Decimal, &mut &t.encoding[..]).unwrap();
+            assert_eq!(CQLValue::Decimal(t.value.clone()), value);
+        }
+    }
+
+    #[test]
+    fn test_deserialize_counter() {
+        let counter: Vec<u8> = vec![0, 0, 0, 0, 0, 0, 1, 0];
+        let counter_slice = &mut &counter[..];
+        let counter_serialize =
+            super::deser_cql_value(&ColumnType::Counter, counter_slice).unwrap();
+        assert_eq!(counter_serialize, CQLValue::Counter(Counter(256)));
+    }
+
+    #[test]
+    fn test_deserialize_blob() {
+        let blob: Vec<u8> = vec![0, 1, 2, 3];
+        let blob_slice = &mut &blob[..];
+        let blob_serialize = super::deser_cql_value(&ColumnType::Blob, blob_slice).unwrap();
+        assert_eq!(blob_serialize, CQLValue::Blob(blob));
+    }
+
+    #[test]
+    fn test_deserialize_bool() {
+        let bool_buf: Vec<u8> = vec![0x00];
+        let bool_slice = &mut &bool_buf[..];
+        let bool_serialize = super::deser_cql_value(&ColumnType::Boolean, bool_slice).unwrap();
+        assert_eq!(bool_serialize, CQLValue::Boolean(false));
+
+        let bool_buf: Vec<u8> = vec![0x01];
+        let bool_slice = &mut &bool_buf[..];
+        let bool_serialize = super::deser_cql_value(&ColumnType::Boolean, bool_slice).unwrap();
+        assert_eq!(bool_serialize, CQLValue::Boolean(true));
+    }
+
+    #[test]
+    fn test_deserialize_int_types() {
+        let int_buf: Vec<u8> = vec![0, 0, 0, 4];
+        let int_slice = &mut &int_buf[..];
+        let int_serialized = super::deser_cql_value(&ColumnType::Int, int_slice).unwrap();
+        assert_eq!(int_serialized, CQLValue::Int(4));
+
+        let smallint_buf: Vec<u8> = vec![0, 4];
+        let smallint_slice = &mut &smallint_buf[..];
+        let smallint_serialized =
+            super::deser_cql_value(&ColumnType::SmallInt, smallint_slice).unwrap();
+        assert_eq!(smallint_serialized, CQLValue::SmallInt(4));
+
+        let tinyint_buf: Vec<u8> = vec![4];
+        let tinyint_slice = &mut &tinyint_buf[..];
+        let tinyint_serialized =
+            super::deser_cql_value(&ColumnType::TinyInt, tinyint_slice).unwrap();
+        assert_eq!(tinyint_serialized, CQLValue::TinyInt(4));
+
+        let bigint_buf: Vec<u8> = vec![0, 0, 0, 0, 0, 0, 0, 4];
+        let bigint_slice = &mut &bigint_buf[..];
+        let bigint_serialized = super::deser_cql_value(&ColumnType::BigInt, bigint_slice).unwrap();
+        assert_eq!(bigint_serialized, CQLValue::BigInt(4));
+    }
 
     #[test]
     fn test_list_from_cql() {
diff --git a/scylla/src/frame/value.rs b/scylla/src/frame/value.rs
index cba91ccc..0205e207 100644
--- a/scylla/src/frame/value.rs
+++ b/scylla/src/frame/value.rs
@@ -1,4 +1,6 @@
+use bigdecimal::BigDecimal;
 use bytes::BufMut;
+use num_bigint::BigInt;
 use std::borrow::Cow;
 use std::collections::HashMap;
 use std::convert::TryInto;
@@ -18,6 +20,10 @@ pub struct ValueTooBig;
 /// Represents an unset value
 pub struct Unset;
 
+/// Represents an counter value
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
+pub struct Counter(pub i64);
+
 /// Enum providing a way to represent a value that might be unset
 pub enum MaybeUnset<V: Value> {
     Unset,
@@ -197,18 +203,17 @@ impl Value for i64 {
     }
 }
 
-impl Value for u32 {
+impl Value for BigDecimal {
     fn serialize(&self, buf: &mut Vec<u8>) -> Result<(), ValueTooBig> {
-        buf.put_i32(4);
-        buf.put_u32(*self);
-        Ok(())
-    }
-}
+        let (value, scale) = self.as_bigint_and_exponent();
+
+        let serialized = value.to_signed_bytes_be();
+        let serialized_len: i32 = serialized.len().try_into().map_err(|_| ValueTooBig)?;
+
+        buf.put_i32(serialized_len + 4);
+        buf.put_i32(scale.try_into().map_err(|_| ValueTooBig)?);
+        buf.extend_from_slice(&serialized);
 
-impl Value for u64 {
-    fn serialize(&self, buf: &mut Vec<u8>) -> Result<(), ValueTooBig> {
-        buf.put_i32(8);
-        buf.put_u64(*self);
         Ok(())
     }
 }
@@ -252,6 +257,18 @@ impl Value for Uuid {
     }
 }
 
+impl Value for BigInt {
+    fn serialize(&self, buf: &mut Vec<u8>) -> Result<(), ValueTooBig> {
+        let serialized = self.to_signed_bytes_be();
+        let serialized_len: i32 = serialized.len().try_into().map_err(|_| ValueTooBig)?;
+
+        buf.put_i32(serialized_len);
+        buf.extend_from_slice(&serialized);
+
+        Ok(())
+    }
+}
+
 impl Value for &str {
     fn serialize(&self, buf: &mut Vec<u8>) -> Result<(), ValueTooBig> {
         let str_bytes: &[u8] = self.as_bytes();
@@ -302,6 +319,12 @@ impl Value for Unset {
     }
 }
 
+impl Value for Counter {
+    fn serialize(&self, buf: &mut Vec<u8>) -> Result<(), ValueTooBig> {
+        self.0.serialize(buf)
+    }
+}
+
 impl<V: Value> Value for MaybeUnset<V> {
     fn serialize(&self, buf: &mut Vec<u8>) -> Result<(), ValueTooBig> {
         match self {


Test patch:
diff --git a/scylla/src/frame/cql_types_test.rs b/scylla/src/frame/cql_types_test.rs
index c510559e..603b116d 100644
--- a/scylla/src/frame/cql_types_test.rs
+++ b/scylla/src/frame/cql_types_test.rs
@@ -1,99 +1,186 @@
+use crate::cql_to_rust::FromCQLVal;
+use crate::frame::response::result::CQLValue;
+use crate::frame::value::Counter;
+use crate::frame::value::Value;
 use crate::transport::session::IntoTypedRows;
 use crate::transport::session::Session;
 use crate::SessionBuilder;
+use bigdecimal::BigDecimal;
+use num_bigint::BigInt;
+use std::cmp::PartialEq;
 use std::env;
+use std::fmt::Debug;
+use std::str::FromStr;
 
-// TODO: Requires a running local Scylla instance
-#[tokio::test]
-#[ignore]
-async fn test_cql_types() {
-    // Create connection
+// Used to prepare a table for test
+// Creates keyspace ks
+// Drops and creates table ks.{table_name} (id int PRIMARY KEY, val {type_name})
+async fn init_test(table_name: &str, type_name: &str) -> Session {
     let uri = env::var("SCYLLA_URI").unwrap_or_else(|_| "127.0.0.1:9042".to_string());
 
     println!("Connecting to {} ...", uri);
-
     let session: Session = SessionBuilder::new().known_node(uri).build().await.unwrap();
 
-    session.query("CREATE KEYSPACE IF NOT EXISTS ks WITH REPLICATION = {'class' : 'SimpleStrategy', 'replication_factor' : 1}", &[]).await.unwrap();
-
-    // Date type test
     session
         .query(
-            "CREATE TABLE IF NOT EXISTS ks.days (day date, id int PRIMARY KEY)",
+            "CREATE KEYSPACE IF NOT EXISTS ks WITH REPLICATION = \
+            {'class' : 'SimpleStrategy', 'replication_factor' : 1}",
             &[],
         )
         .await
         .unwrap();
 
-    let date_ts: u32 = 10;
     session
-        .query("INSERT INTO ks.days (day, id) VALUES (?, 1)", (date_ts,))
+        .query(format!("DROP TABLE IF EXISTS ks.{}", table_name), &[])
         .await
         .unwrap();
 
-    if let Some(rows) = session.query("SELECT day FROM ks.days", &[]).await.unwrap() {
-        for row in rows.into_typed::<(u32,)>() {
-            let day: u32 = row.unwrap().0;
-            println!("day: {}", day);
-        }
-    }
-
-    // Bool test
     session
         .query(
-            "CREATE TABLE IF NOT EXISTS ks.truefalse (boolvalue boolean, id int PRIMARY KEY)",
+            format!(
+                "CREATE TABLE IF NOT EXISTS ks.{} (id int PRIMARY KEY, val {})",
+                table_name, type_name
+            ),
             &[],
         )
         .await
         .unwrap();
 
-    let val: bool = true;
     session
-        .query(
-            "INSERT INTO ks.truefalse (boolvalue, id) VALUES (?, 1)",
-            (val,),
-        )
-        .await
-        .unwrap();
+}
 
-    if let Some(rows) = session
-        .query("SELECT boolvalue FROM ks.truefalse", &[])
-        .await
-        .unwrap()
-    {
-        for row in rows.into_typed::<(bool,)>() {
-            let bool_val: bool = row.unwrap().0;
-            println!("bool value: {}", bool_val);
-        }
+// This function tests serialization and deserialization mechanisms by sending insert and select
+// queries to running Scylla instance.
+// To do so, it:
+// Prepares a table for tests (by creating keyspace ks and table ks.{table_name} using init_test)
+// Runs a test that, for every element of `tests`:
+// - inserts 2 values (one encoded as string and one as bound values) into table ks.{type_name}
+// - selects this 2 values and compares them with expected value
+// Expected values and bound values are computed using T::from_str
+async fn run_tests<T>(tests: &[&str], type_name: &str)
+where
+    T: Value + FromCQLVal<CQLValue> + FromStr + Debug + Clone + PartialEq,
+{
+    let session: Session = init_test(type_name, type_name).await;
+
+    for test in tests.iter() {
+        let insert_string_encoded_value = format!(
+            "INSERT INTO ks.{} (id, val) VALUES (0, {})",
+            type_name, test
+        );
+        session
+            .query(insert_string_encoded_value, &[])
+            .await
+            .unwrap();
+
+        let insert_bound_value = format!("INSERT INTO ks.{} (id, val) VALUES (1, ?)", type_name);
+        let value_to_bound = T::from_str(test).ok().unwrap();
+        session
+            .query(insert_bound_value, (value_to_bound,))
+            .await
+            .unwrap();
+
+        let select_values = format!("SELECT val from ks.{}", type_name);
+        let read_values: Vec<T> = session
+            .query(select_values, &[])
+            .await
+            .unwrap()
+            .unwrap()
+            .into_typed::<(T,)>()
+            .map(Result::unwrap)
+            .map(|row| row.0)
+            .collect::<Vec<_>>();
+
+        let expected_value = T::from_str(test).ok().unwrap();
+        assert_eq!(read_values, vec![expected_value.clone(), expected_value]);
     }
+}
 
-    // Float test
+#[tokio::test]
+async fn test_varint() {
+    let tests = [
+        "0",
+        "1",
+        "127",
+        "128",
+        "129",
+        "-1",
+        "-128",
+        "-129",
+        "123456789012345678901234567890",
+        "-123456789012345678901234567890",
+    ];
 
-    session
-        .query(
-            "CREATE TABLE IF NOT EXISTS ks.floatingpoint (floatval float, id int PRIMARY KEY)",
-            &[],
-        )
-        .await
-        .unwrap();
+    run_tests::<BigInt>(&tests, "varint").await;
+}
 
-    let val: f32 = 1.42;
-    session
-        .query(
-            "INSERT INTO ks.floatingpoint (floatval, id) VALUES (?, 1)",
-            (val,),
-        )
-        .await
-        .unwrap();
+#[tokio::test]
+async fn test_decimal() {
+    let tests = [
+        "4.2",
+        "0",
+        "1.999999999999999999999999999999999999999",
+        "997",
+        "123456789012345678901234567890.1234567890",
+        "-123456789012345678901234567890.1234567890",
+    ];
 
-    if let Some(rows) = session
-        .query("SELECT floatval FROM ks.floatingpoint", &[])
-        .await
-        .unwrap()
-    {
-        for row in rows.into_typed::<(f32,)>() {
-            let bool_val: f32 = row.unwrap().0;
-            println!("float value: {}", bool_val);
-        }
+    run_tests::<BigDecimal>(&tests, "decimal").await;
+}
+
+#[tokio::test]
+async fn test_bool() {
+    let tests = ["true", "false"];
+
+    run_tests::<bool>(&tests, "boolean").await;
+}
+
+#[tokio::test]
+async fn test_float() {
+    let max = f32::MAX.to_string();
+    let min = f32::MIN.to_string();
+    let tests = [
+        "3.14",
+        "997",
+        "0.1",
+        "128",
+        "-128",
+        max.as_str(),
+        min.as_str(),
+    ];
+
+    run_tests::<f32>(&tests, "float").await;
+}
+
+#[tokio::test]
+async fn test_counter() {
+    let big_increment = i64::MAX.to_string();
+    let tests = ["1", "997", big_increment.as_str()];
+
+    // Can't use run_tests, because counters are special and can't be inserted
+    let type_name = "counter";
+    let session: Session = init_test(type_name, type_name).await;
+
+    for (i, test) in tests.iter().enumerate() {
+        let update_bound_value = format!("UPDATE ks.{} SET val = val + ? WHERE id = ?", type_name);
+        let value_to_bound = Counter(i64::from_str(test).unwrap());
+        session
+            .query(update_bound_value, (value_to_bound, i as i32))
+            .await
+            .unwrap();
+
+        let select_values = format!("SELECT val FROM ks.{} WHERE id = ?", type_name);
+        let read_values: Vec<Counter> = session
+            .query(select_values, (i as i32,))
+            .await
+            .unwrap()
+            .unwrap()
+            .into_typed::<(Counter,)>()
+            .map(Result::unwrap)
+            .map(|row| row.0)
+            .collect::<Vec<_>>();
+
+        let expected_value = Counter(i64::from_str(test).unwrap());
+        assert_eq!(read_values, vec![expected_value]);
     }
 }