Repository: AlexPikalov/cdrs (language: rust)

Issue:
Epoch timestamps are being modified by the driver before being stored in Cassandra DB

For example:

suppose that there is a table in Cassandra called tableA and it has two columns: ID column and timestamp column.

id is INT
timestamp is DECIMAL

When executing those lines of code:

    let update_struct_cql: String = format!("update tableA SET  timestamp = ? where id = ?;");

    ddb.query_with_values(update_struct_cql,query_values!(current_time, id)).expect("[Err]: Could not update table");  
  
When I print out timestamp BEFORE it is being used in above query, it will print something normal, like 1545870134, however, when I check column value in Cassandra DB, I see something like:

6.1805539111220E-825570344


The question is: What happened to the timestamp ?



Discussion / hints:
Hi @MaenMax ,

First of all thank you for your question. If your timestamp is decimal the driver converts it into decimal data type defined in Cassandra Native Protocol spec https://github.com/apache/cassandra/blob/trunk/doc/native_protocol_v4.spec#L863

> The decimal format represents an arbitrary-precision number.  It contains an
  [int] "scale" component followed by a varint encoding (see section 6.17)
  of the unscaled value.  The encoded value represents "unscaledE-scale".
  In other words, "unscaled * 10 ^ (-1 * scale)".

The string format is more or less in agreement with the spec.

PS Feel free to create an issue if you have any problems with actual values and how it's been converted.
Thanks Alex for your reply.

I want to save it as it is in DB. Do you know how to save it so that it appear the same in Cassandra without the conversion? I don't want to change the data type of Cassandra column, I would like to keep it as decimal, but how to save something like "1545870134" into a decimal column type without being converted as above ? 
I can try to modify an algorithm of encoding so that for integer numbers (numbers without decimal point) the value will be without `E-XXX`, but just `1545870134`.

I'll try to prepare the PR for this change as soon as I can. Will keep you updated.
Thanks!

Fix patch:
diff --git a/.travis.yml b/.travis.yml
index 827ee82..2fa1e23 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -25,6 +25,7 @@ install:
 script:
   # - cargo fmt -- --write-mode=diff
   - cargo build --verbose
+  - cargo test --verbose
   - cargo test --verbose -- --ignored
 
 deploy:
diff --git a/src/types/data_serialization_types.rs b/src/types/data_serialization_types.rs
index b054163..2518df5 100644
--- a/src/types/data_serialization_types.rs
+++ b/src/types/data_serialization_types.rs
@@ -1,13 +1,12 @@
-use std::ops::Mul;
 use std::io;
 use std::net;
 use std::string::FromUtf8Error;
 
-use uuid;
-use super::*;
 use super::blob::Blob;
+use super::*;
 use error;
 use frame::FromCursor;
+use uuid;
 
 // https://github.com/apache/cassandra/blob/trunk/doc/native_protocol_v4.spec#L813
 
@@ -41,8 +40,10 @@ pub fn decode_blob(bytes: &Vec<u8>) -> Result<Blob, io::Error> {
 pub fn decode_boolean(bytes: &[u8]) -> Result<bool, io::Error> {
     let false_byte: u8 = 0;
     if bytes.is_empty() {
-        Err(io::Error::new(io::ErrorKind::UnexpectedEof,
-                           "no bytes were found"))
+        Err(io::Error::new(
+            io::ErrorKind::UnexpectedEof,
+            "no bytes were found",
+        ))
     } else {
         Ok(bytes[0] != false_byte)
     }
@@ -61,26 +62,16 @@ pub fn decode_date(bytes: &[u8]) -> Result<i32, io::Error> {
     try_from_bytes(bytes).map(|i| i as i32)
 }
 
-// TODO: make sure this method meets the specification.
 // Decodes Cassandra `decimal` data (bytes) into Rust's `Result<f32, io::Error>`
 pub fn decode_decimal(bytes: &[u8]) -> Result<f32, io::Error> {
-    let ref separator = b'E';
-    let lr: Vec<Vec<u8>> = bytes.split(|ch| ch == separator)
-                                .map(|p| p.to_vec())
-                                .collect();
-    let unscaled = try_i_from_bytes(lr[0].as_slice());
-    if unscaled.is_err() {
-        return Err(unscaled.unwrap_err());
-    }
-    let scaled = try_i_from_bytes(lr[1].as_slice());
-    if scaled.is_err() {
-        return Err(scaled.unwrap_err());
-    }
-
-    let unscaled_unwrapped = try!(unscaled) as f32;
-    let scaled_unwrapped = try!(scaled) as i32;
-    let dec: f32 = 10.0;
-    Ok(unscaled_unwrapped.mul(dec.powi(scaled_unwrapped)))
+    let lr = bytes.split_at(INT_LEN);
+
+    let scale = try_i_from_bytes(lr.0)? as u32;
+    let unscaled = try_i_from_bytes(lr.1)? as f32;
+
+    let mult = 10i32.pow(scale) as f32;
+
+    Ok(unscaled / mult)
 }
 
 // Decodes Cassandra `double` data (bytes) into Rust's `Result<f32, io::Error>`
@@ -97,10 +88,9 @@ pub fn decode_float(bytes: &[u8]) -> Result<f32, io::Error> {
 pub fn decode_inet(bytes: &[u8]) -> Result<net::IpAddr, io::Error> {
     match bytes.len() {
         // v4
-        4 => Ok(net::IpAddr::V4(net::Ipv4Addr::new(bytes[0],
-                                                   bytes[1],
-                                                   bytes[2],
-                                                   bytes[3]))),
+        4 => Ok(net::IpAddr::V4(net::Ipv4Addr::new(
+            bytes[0], bytes[1], bytes[2], bytes[3],
+        ))),
         // v6
         16 => {
             let a = from_u16_bytes(&bytes[0..2]);
@@ -115,9 +105,10 @@ pub fn decode_inet(bytes: &[u8]) -> Result<net::IpAddr, io::Error> {
         }
         _ => {
             // let message = format!("Unparseable  Ip address {:?}", bytes);
-            Err(io::Error::new(io::ErrorKind::Other,
-                               format!("Unparseable  Ip address {:?}",
-                                       bytes)))
+            Err(io::Error::new(
+                io::ErrorKind::Other,
+                format!("Unparseable  Ip address {:?}", bytes),
+            ))
         }
     }
 }
@@ -133,13 +124,12 @@ pub fn decode_timestamp(bytes: &[u8]) -> Result<i64, io::Error> {
 // Decodes Cassandra `list` data (bytes) into Rust's `Result<Vec<CBytes>, io::Error>`
 pub fn decode_list(bytes: &[u8]) -> Result<Vec<CBytes>, io::Error> {
     let mut cursor: io::Cursor<&[u8]> = io::Cursor::new(bytes);
-    let l =
-        CInt::from_cursor(&mut cursor).map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
+    let l = CInt::from_cursor(&mut cursor)
+        .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
     let mut list = Vec::with_capacity(l as usize);
     for _ in 0..l {
-        let b = CBytes::from_cursor(&mut cursor).map_err(|err| {
-            io::Error::new(io::ErrorKind::InvalidData, err)
-        })?;
+        let b = CBytes::from_cursor(&mut cursor)
+            .map_err(|err| io::Error::new(io::ErrorKind::InvalidData, err))?;
         list.push(b);
     }
     Ok(list)
@@ -153,17 +143,14 @@ pub fn decode_set(bytes: &[u8]) -> Result<Vec<CBytes>, io::Error> {
 // Decodes Cassandra `map` data (bytes) into Rust's `Result<Vec<(CBytes, CBytes)>, io::Error>`
 pub fn decode_map(bytes: &[u8]) -> Result<Vec<(CBytes, CBytes)>, io::Error> {
     let mut cursor: io::Cursor<&[u8]> = io::Cursor::new(bytes);
-    let l = CInt::from_cursor(&mut cursor).map_err(|err| {
-        io::Error::new(io::ErrorKind::InvalidData, err)
-    })?;
+    let l = CInt::from_cursor(&mut cursor)
+        .map_err(|err| io::Error::new(io::ErrorKind::InvalidData, err))?;
     let mut map = Vec::with_capacity(l as usize);
     for _ in 0..l {
-        let n = CBytes::from_cursor(&mut cursor).map_err(|err| {
-            io::Error::new(io::ErrorKind::InvalidData, err)
-        })?;
-        let v = CBytes::from_cursor(&mut cursor).map_err(|err| {
-            io::Error::new(io::ErrorKind::InvalidData, err)
-        })?;
+        let n = CBytes::from_cursor(&mut cursor)
+            .map_err(|err| io::Error::new(io::ErrorKind::InvalidData, err))?;
+        let v = CBytes::from_cursor(&mut cursor)
+            .map_err(|err| io::Error::new(io::ErrorKind::InvalidData, err))?;
         map.push((n, v));
     }
     Ok(map)
@@ -205,21 +192,17 @@ pub fn decode_udt(bytes: &[u8], l: usize) -> Result<Vec<CBytes>, io::Error> {
     let mut cursor: io::Cursor<&[u8]> = io::Cursor::new(bytes);
     let mut udt = Vec::with_capacity(l);
     for _ in 0..l {
-        let v =
-            CBytes::from_cursor(&mut cursor).or_else(|err| match err {
-                                                         error::Error::Io(io_err) => {
-                                                             if io_err.kind()
-                                                                == io::ErrorKind::UnexpectedEof {
-                                                                 Ok(CBytes::new_empty())
-                                                             } else {
-                                                                 Err(io_err.into())
-                                                             }
-                                                         }
-                                                         _ => Err(err),
-                                                     })
-                                            .map_err(|err| {
-                io::Error::new(io::ErrorKind::InvalidData, err)
-            })?;
+        let v = CBytes::from_cursor(&mut cursor)
+            .or_else(|err| match err {
+                error::Error::Io(io_err) => {
+                    if io_err.kind() == io::ErrorKind::UnexpectedEof {
+                        Ok(CBytes::new_empty())
+                    } else {
+                        Err(io_err.into())
+                    }
+                }
+                _ => Err(err),
+            }).map_err(|err| io::Error::new(io::ErrorKind::InvalidData, err))?;
         udt.push(v);
     }
     Ok(udt)
@@ -231,9 +214,8 @@ pub fn decode_tuple(bytes: &[u8], l: usize) -> Result<Vec<CBytes>, io::Error> {
     let mut cursor: io::Cursor<&[u8]> = io::Cursor::new(bytes);
     let mut udt = Vec::with_capacity(l);
     for _ in 0..l {
-        let v = CBytes::from_cursor(&mut cursor).map_err(|err| {
-            io::Error::new(io::ErrorKind::InvalidData, err)
-        })?;
+        let v = CBytes::from_cursor(&mut cursor)
+            .map_err(|err| io::Error::new(io::ErrorKind::InvalidData, err))?;
         udt.push(v);
     }
     Ok(udt)
@@ -241,10 +223,10 @@ pub fn decode_tuple(bytes: &[u8], l: usize) -> Result<Vec<CBytes>, io::Error> {
 
 #[cfg(test)]
 mod tests {
-    use std::net::IpAddr;
-    use super::*;
-    use super::super::super::frame::frame_result::*;
     use super::super::super::error::*;
+    use super::super::super::frame::frame_result::*;
+    use super::*;
+    use std::net::IpAddr;
 
     #[test]
     fn decode_custom_test() {
@@ -268,8 +250,10 @@ mod tests {
 
     #[test]
     fn decode_blob_test() {
-        assert_eq!(decode_blob(&vec![0, 0, 0, 3]).unwrap().into_vec(),
-                   vec![0, 0, 0, 3]);
+        assert_eq!(
+            decode_blob(&vec![0, 0, 0, 3]).unwrap().into_vec(),
+            vec![0, 0, 0, 3]
+        );
     }
 
     #[test]
@@ -353,6 +337,19 @@ mod tests {
         assert_eq!(decode_tinyint(&[10]).unwrap(), 10);
     }
 
+    #[test]
+    fn decode_decimal_test() {
+        assert_eq!(decode_decimal(&[0, 0, 0, 0, 10u8]).unwrap(), 10f32);
+
+        assert_eq!(129.0, decode_decimal(&[0, 0, 0, 0, 0x00, 0x81]).unwrap());
+
+        assert_eq!(-129.0, decode_decimal(&[0, 0, 0, 0, 0xFF, 0x7F]).unwrap());
+
+        assert_eq!(12.9, decode_decimal(&[0, 0, 0, 1, 0x00, 0x81]).unwrap());
+
+        assert_eq!(-12.9, decode_decimal(&[0, 0, 0, 1, 0xFF, 0x7F]).unwrap());
+    }
+
     #[test]
     fn decode_text_test() {
         assert_eq!(decode_text(b"abcba").unwrap(), "abcba");
@@ -375,7 +372,14 @@ mod tests {
 
     #[test]
     fn decode_varint_test() {
-        assert_eq!(decode_varint(&[0, 0, 0, 0, 0, 0, 0, 10]).unwrap(), 10);
+        assert_eq!(decode_varint(&[0x00]).unwrap(), 0);
+        assert_eq!(decode_varint(&[0x01]).unwrap(), 1);
+        assert_eq!(decode_varint(&[0x7F]).unwrap(), 127);
+        assert_eq!(decode_varint(&[0x00, 0x80]).unwrap(), 128);
+        assert_eq!(decode_varint(&[0x00, 0x81]).unwrap(), 129);
+        assert_eq!(decode_varint(&[0xFF]).unwrap(), -1);
+        assert_eq!(decode_varint(&[0x80]).unwrap(), -128);
+        assert_eq!(decode_varint(&[0xFF, 0x7F]).unwrap(), -129);
     }
 
     #[test]
@@ -389,55 +393,84 @@ mod tests {
     fn as_rust_blob_test() {
         let d_type = DataType { id: ColType::Blob };
         let data = CBytes::new(vec![1, 2, 3]);
-        assert_eq!(as_rust_type!(d_type, data, Blob).unwrap()
-                                                    .unwrap()
-                                                    .into_vec(),
-                   vec![1, 2, 3]);
+        assert_eq!(
+            as_rust_type!(d_type, data, Blob)
+                .unwrap()
+                .unwrap()
+                .into_vec(),
+            vec![1, 2, 3]
+        );
         let wrong_type = DataType { id: ColType::Map };
         assert!(as_rust_type!(wrong_type, data, Blob).is_err());
     }
 
     #[test]
     fn as_rust_string_test() {
-        let type_custom = DataType { id: ColType::Custom, };
+        let type_custom = DataType {
+            id: ColType::Custom,
+        };
         let type_ascii = DataType { id: ColType::Ascii };
-        let type_varchar = DataType { id: ColType::Varchar, };
+        let type_varchar = DataType {
+            id: ColType::Varchar,
+        };
         let data = CBytes::new(b"abc".to_vec());
-        assert_eq!(as_rust_type!(type_custom, data, String).unwrap().unwrap(),
-                   "abc");
-        assert_eq!(as_rust_type!(type_ascii, data, String).unwrap().unwrap(),
-                   "abc");
-        assert_eq!(as_rust_type!(type_varchar, data, String).unwrap().unwrap(),
-                   "abc");
+        assert_eq!(
+            as_rust_type!(type_custom, data, String).unwrap().unwrap(),
+            "abc"
+        );
+        assert_eq!(
+            as_rust_type!(type_ascii, data, String).unwrap().unwrap(),
+            "abc"
+        );
+        assert_eq!(
+            as_rust_type!(type_varchar, data, String).unwrap().unwrap(),
+            "abc"
+        );
         let wrong_type = DataType { id: ColType::Map };
         assert!(as_rust_type!(wrong_type, data, String).is_err());
     }
 
     #[test]
     fn as_rust_bool_test() {
-        let type_boolean = DataType { id: ColType::Boolean, };
+        let type_boolean = DataType {
+            id: ColType::Boolean,
+        };
         let data_true = CBytes::new(vec![1]);
         let data_false = CBytes::new(vec![0]);
-        assert_eq!(as_rust_type!(type_boolean, data_true, bool).unwrap()
-                                                               .unwrap(),
-                   true);
-        assert_eq!(as_rust_type!(type_boolean, data_false, bool).unwrap()
-                                                                .unwrap(),
-                   false);
+        assert_eq!(
+            as_rust_type!(type_boolean, data_true, bool)
+                .unwrap()
+                .unwrap(),
+            true
+        );
+        assert_eq!(
+            as_rust_type!(type_boolean, data_false, bool)
+                .unwrap()
+                .unwrap(),
+            false
+        );
         let wrong_type = DataType { id: ColType::Map };
         assert!(as_rust_type!(wrong_type, data_false, bool).is_err());
     }
 
     #[test]
     fn as_rust_i64_test() {
-        let type_bigint = DataType { id: ColType::Bigint, };
-        let type_timestamp = DataType { id: ColType::Timestamp, };
+        let type_bigint = DataType {
+            id: ColType::Bigint,
+        };
+        let type_timestamp = DataType {
+            id: ColType::Timestamp,
+        };
         let type_time = DataType { id: ColType::Time };
-        let type_varint = DataType { id: ColType::Varint, };
+        let type_varint = DataType {
+            id: ColType::Varint,
+        };
         let data = CBytes::new(vec![0, 0, 0, 0, 0, 0, 0, 100]);
         assert_eq!(as_rust_type!(type_bigint, data, i64).unwrap().unwrap(), 100);
-        assert_eq!(as_rust_type!(type_timestamp, data, i64).unwrap().unwrap(),
-                   100);
+        assert_eq!(
+            as_rust_type!(type_timestamp, data, i64).unwrap().unwrap(),
+            100
+        );
         assert_eq!(as_rust_type!(type_time, data, i64).unwrap().unwrap(), 100);
         assert_eq!(as_rust_type!(type_varint, data, i64).unwrap().unwrap(), 100);
         let wrong_type = DataType { id: ColType::Map };
@@ -457,17 +490,23 @@ mod tests {
 
     #[test]
     fn as_rust_i16_test() {
-        let type_smallint = DataType { id: ColType::Smallint, };
+        let type_smallint = DataType {
+            id: ColType::Smallint,
+        };
         let data = CBytes::new(vec![0, 100]);
-        assert_eq!(as_rust_type!(type_smallint, data, i16).unwrap().unwrap(),
-                   100);
+        assert_eq!(
+            as_rust_type!(type_smallint, data, i16).unwrap().unwrap(),
+            100
+        );
         let wrong_type = DataType { id: ColType::Map };
         assert!(as_rust_type!(wrong_type, data, i16).is_err());
     }
 
     #[test]
     fn as_rust_i8_test() {
-        let type_tinyint = DataType { id: ColType::Tinyint, };
+        let type_tinyint = DataType {
+            id: ColType::Tinyint,
+        };
         let data = CBytes::new(vec![100]);
         assert_eq!(as_rust_type!(type_tinyint, data, i8).unwrap().unwrap(), 100);
         let wrong_type = DataType { id: ColType::Map };
@@ -476,7 +515,9 @@ mod tests {
 
     #[test]
     fn as_rust_f64_test() {
-        let type_double = DataType { id: ColType::Double, };
+        let type_double = DataType {
+            id: ColType::Double,
+        };
         let data = CBytes::new(to_float_big(0.1 as f64));
         assert_eq!(as_rust_type!(type_double, data, f64).unwrap().unwrap(), 0.1);
         let wrong_type = DataType { id: ColType::Map };
diff --git a/src/types/decimal.rs b/src/types/decimal.rs
new file mode 100644
index 0000000..4bba245
--- /dev/null
+++ b/src/types/decimal.rs
@@ -0,0 +1,74 @@
+use super::{to_int, to_varint};
+use frame::traits::IntoBytes;
+
+/// Cassandra Decimal type
+#[derive(Debug, Clone, PartialEq)]
+pub struct Decimal(f32);
+
+impl Decimal {
+  /// Method that returns plain `f32` value.
+  pub fn as_plain(&self) -> f32 {
+    self.0
+  }
+}
+
+impl IntoBytes for Decimal {
+  fn into_cbytes(&self) -> Vec<u8> {
+    let mut scaled: f32 = self.0;
+    let mut scale: i32 = 0;
+
+    while scaled.trunc() != scaled {
+      scaled *= 10f32;
+      scale += 1;
+    }
+
+    let mut bytes: Vec<u8> = vec![];
+    bytes.extend(to_int(scale));
+    bytes.extend(to_varint(scaled as i32));
+
+    bytes
+  }
+}
+
+macro_rules! impl_from_for_decimal {
+  ($t:ty) => {
+    impl From<$t> for Decimal {
+      fn from(i: $t) -> Self {
+        Decimal(i.into())
+      }
+    }
+  };
+}
+
+impl_from_for_decimal!(i8);
+impl_from_for_decimal!(i16);
+impl_from_for_decimal!(u8);
+impl_from_for_decimal!(u16);
+impl_from_for_decimal!(f32);
+
+#[cfg(test)]
+mod test {
+  use super::*;
+  #[test]
+  fn into_cbytes_test() {
+    let int_positive: f32 = 129.0;
+    assert_eq!(
+      Decimal::from(int_positive).into_cbytes(),
+      vec![0, 0, 0, 0, 0x00, 0x81]
+    );
+
+    let int_negative: f32 = -129.0;
+    assert_eq!(
+      Decimal::from(int_negative).into_cbytes(),
+      vec![0, 0, 0, 0, 0xFF, 0x7F]
+    );
+
+    let float_positive: f32 = 12.9;
+    let expected: Vec<u8> = vec![0, 0, 0, 1, 0x00, 0x81];
+    assert_eq!(Decimal::from(float_positive).into_cbytes(), expected);
+
+    let float_negative: f32 = -12.9;
+    let expected: Vec<u8> = vec![0, 0, 0, 1, 0xFF, 0x7F];
+    assert_eq!(Decimal::from(float_negative).into_cbytes(), expected);
+  }
+}
diff --git a/src/types/mod.rs b/src/types/mod.rs
index 9bc66a4..e508269 100644
--- a/src/types/mod.rs
+++ b/src/types/mod.rs
@@ -16,6 +16,7 @@ pub const UUID_LEN: usize = 16;
 #[macro_use]
 pub mod blob;
 pub mod data_serialization_types;
+pub mod decimal;
 pub mod from_cdrs;
 pub mod list;
 pub mod map;
@@ -264,6 +265,40 @@ pub fn to_bigint(int: i64) -> Vec<u8> {
     bytes
 }
 
+/// Converts integer into Cassandra's [varint].
+pub fn to_varint(int: i32) -> Vec<u8> {
+    if int == 0 {
+        return vec![0];
+    }
+
+    let mut int_bytes = to_int(int);
+    match int.signum() {
+        1 => {
+            int_bytes = int_bytes.into_iter().skip_while(|b| *b == 0x00).collect();
+            if int_bytes
+                .get(0)
+                .map(|b| b.leading_zeros() == 0)
+                .unwrap_or(true)
+            {
+                int_bytes.insert(0, 0x00);
+            }
+        }
+        -1 => {
+            int_bytes = int_bytes.into_iter().skip_while(|b| *b == 0xFF).collect();
+            if int_bytes
+                .get(0)
+                .map(|b| b.leading_zeros() > 0)
+                .unwrap_or(true)
+            {
+                int_bytes.insert(0, 0xFF);
+            }
+        }
+        _ => unreachable!(),
+    }
+
+    int_bytes
+}
+
 /// Converts number i16 into Cassandra's `short`.
 ///
 /// # Panics
@@ -849,4 +884,16 @@ mod tests {
         assert_eq!(val, 12i64);
     }
 
+    #[test]
+    fn test_to_varint() {
+        assert_eq!(to_varint(0), vec![0x00]);
+        assert_eq!(to_varint(1), vec![0x01]);
+        assert_eq!(to_varint(127), vec![0x7F]);
+        assert_eq!(to_varint(128), vec![0x00, 0x80]);
+        assert_eq!(to_varint(129), vec![0x00, 0x81]);
+        assert_eq!(to_varint(-1), vec![0xFF]);
+        assert_eq!(to_varint(-128), vec![0x80]);
+        assert_eq!(to_varint(-129), vec![0xFF, 0x7F]);
+    }
+
 }
diff --git a/src/types/value.rs b/src/types/value.rs
index 698ed89..4706ee1 100644
--- a/src/types/value.rs
+++ b/src/types/value.rs
@@ -1,16 +1,17 @@
+use std::cmp::Eq;
 use std::collections::HashMap;
 use std::convert::Into;
+use std::fmt::Debug;
 use std::hash::Hash;
-use std::cmp::Eq;
 use std::net::IpAddr;
-use std::fmt::Debug;
 
-use uuid::Uuid;
-use time::Timespec;
 use frame::IntoBytes;
+use time::Timespec;
+use uuid::Uuid;
 
-use super::*;
 use super::blob::Blob;
+use super::decimal::Decimal;
+use super::*;
 
 /// Types of Cassandra value: normal value (bits), null value and not-set value
 #[derive(Debug, Clone)]
@@ -40,24 +41,31 @@ pub struct Value {
 impl Value {
     /// The factory method which creates a normal type value basing on provided bytes.
     pub fn new_normal<B>(v: B) -> Value
-        where B: Into<Bytes>
+    where
+        B: Into<Bytes>,
     {
         let bytes = v.into().0;
         let l = bytes.len() as i32;
-        Value { body: bytes,
-                value_type: ValueType::Normal(l), }
+        Value {
+            body: bytes,
+            value_type: ValueType::Normal(l),
+        }
     }
 
     /// The factory method which creates null Cassandra value.
     pub fn new_null() -> Value {
-        Value { body: vec![],
-                value_type: ValueType::Null, }
+        Value {
+            body: vec![],
+            value_type: ValueType::Null,
+        }
     }
 
     /// The factory method which creates non-set Cassandra value.
     pub fn new_not_set() -> Value {
-        Value { body: vec![],
-                value_type: ValueType::NotSet, }
+        Value {
+            body: vec![],
+            value_type: ValueType::NotSet,
+        }
     }
 }
 
@@ -204,6 +212,12 @@ impl Into<Bytes> for Blob {
     }
 }
 
+impl Into<Bytes> for Decimal {
+    fn into(self) -> Bytes {
+        Bytes(self.into_cbytes())
+    }
+}
+
 impl<T: Into<Bytes> + Clone + Debug> From<Vec<T>> for Bytes {
     fn from(vec: Vec<T>) -> Bytes {
         let mut bytes: Vec<u8> = vec![];
@@ -218,8 +232,9 @@ impl<T: Into<Bytes> + Clone + Debug> From<Vec<T>> for Bytes {
 }
 
 impl<K, V> From<HashMap<K, V>> for Bytes
-    where K: Into<Bytes> + Clone + Debug + Hash + Eq,
-          V: Into<Bytes> + Clone + Debug
+where
+    K: Into<Bytes> + Clone + Debug + Hash + Eq,
+    V: Into<Bytes> + Clone + Debug,
 {
     fn from(map: HashMap<K, V>) -> Bytes {
         let mut bytes: Vec<u8> = vec![];


Test patch:
diff --git a/tests/native_types.rs b/tests/native_types.rs
index decee9a..cb232b8 100644
--- a/tests/native_types.rs
+++ b/tests/native_types.rs
@@ -10,6 +10,7 @@ use common::*;
 
 use cdrs::query::QueryExecutor;
 use cdrs::types::blob::Blob;
+use cdrs::types::decimal::Decimal;
 use cdrs::types::map::Map;
 use cdrs::types::value::Bytes;
 use cdrs::types::{AsRust, ByName, IntoRustByName};
@@ -181,14 +182,16 @@ fn integer_v4() {
 #[ignore]
 fn float() {
     let cql = "CREATE TABLE IF NOT EXISTS cdrs_test.test_float \
-               (my_float float PRIMARY KEY, my_double double)";
+               (my_float float PRIMARY KEY, my_double double, my_decimal decimal)";
     let session = setup(cql).expect("setup");
 
     let my_float: f32 = 123.456;
     let my_double: f64 = 987.654;
-    let values = query_values!(my_float, my_double);
+    let my_decimal: f32 = 120.01;
+    let values = query_values!(my_float, my_double, Decimal::from(my_decimal));
 
-    let query = "INSERT INTO cdrs_test.test_float (my_float, my_double) VALUES (?, ?)";
+    let query =
+        "INSERT INTO cdrs_test.test_float (my_float, my_double, my_decimal) VALUES (?, ?, ?)";
     session
         .query_with_values(query, values)
         .expect("insert floats error");
@@ -206,8 +209,10 @@ fn float() {
     for row in rows {
         let my_float_row: f32 = row.get_r_by_name("my_float").expect("my_float");
         let my_double_row: f64 = row.get_r_by_name("my_double").expect("my_double");
+        let my_decimal_row: f32 = row.get_r_by_name("my_decimal").expect("my_decimal");
         assert_eq!(my_float_row, my_float);
         assert_eq!(my_double_row, my_double);
+        assert_eq!(my_decimal_row, my_decimal);
     }
 }
 
@@ -224,7 +229,8 @@ fn blob() {
         ("b".to_owned(), b"bbbbb".to_vec().into()),
         ("c".to_owned(), b"ccccc".to_vec().into()),
         ("d".to_owned(), b"ddddd".to_vec().into()),
-    ].into_iter()
+    ]
+        .into_iter()
         .map(|x| x.clone())
         .collect();