Repository: scylladb/scylla-rust-driver
Pull request #363 (2021-12-05 04:02:51)

Add support for Duration type
Fixes: #320 

This works great with Scylla but for some reason it fails with Cassandra (at least with 4.0.1). The column type code that Cassandra is sending for Duration is 0 instead of 0x15 and it is followed with type's name: "org.apache.cassandra.db.marshal.DurationType". How should we deal with it?

Apart from that (and a few missing tests or docstrings), is there anything serious missing here? 


## 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.


psarna: @Ponewor I have no idea why Cassandra decided to encode duration as a custom type instead of a native type, but it is legal in CQL v4. Here's an excerpt from https://github.com/apache/cassandra/blob/trunk/doc/native_protocol_v4.spec:
```
The option for <type> is either a native
          type (see below), in which case the option has no value, or a
          'custom' type, in which case the value is a [string] representing
          the fully qualified class name of the type represented. Valid option
          ids are:
            0x0000    Custom: the value is a [string], see above.
            0x0001    Ascii
            0x0002    Bigint
            0x0003    Blob
            (...)
```

So this is actually a mild compatibility bug - not a serious one, because our driver should be free to not recognize specific custom types, but perhaps we should have a map of handlers for a few known ones - e.g. `org.apache.cassandra.db.marshal.DurationType`.

I created https://github.com/scylladb/scylla-rust-driver/issues/364 to track the problem. For now, either make the test optional or add a Cassandra-specific bypass - namely, let the test pass either when it returns a correct duration, or when it fails because it received a custom type. If you decide to go with the latter, please also add a comment about it in the test case.
piodul: @Ponewor The Duration type was introduced in v5 of the CQL protocol. There was a need for this type to be supported in Scylla - however, Scylla only supports v4 so the support for it was backported to the older version of the protocol.

If you look at the protocol specifications as defined in the Cassandra repository, you can see that in v4 there is no concept of Duration ([see here](https://github.com/apache/cassandra/blob/185893256f10c14207bffe49ae733fb1a970aec5/doc/native_protocol_v4.spec#L636-L637)), it is only introduced in v5 ([here](https://github.com/apache/cassandra/blob/185893256f10c14207bffe49ae733fb1a970aec5/doc/native_protocol_v5.spec#L841-L843)).

In order to be compatible with the protocol v4, I guess that Cassandra decides to encode Durations as a "Custom" type. I'm not aware of how the encoding looks, though. I agree with @psarna on how to handle this for Cassandra.
psarna: > @Ponewor The Duration type was introduced in v5 of the CQL protocol. There was a need for this type to be supported in Scylla - however, Scylla only supports v4 so the support for it was backported to the older version of the protocol.

Ah, makes sense. Btw, we could consider allowing the user to declare the version explicitly to allow "version spoofing". That would allow users to set v5 here, e.g. if they know that they'll only use the subset of v5 that's actually implemented. It would act as a trivial workaround for issues like this.

> 
> If you look at the protocol specifications as defined in the Cassandra repository, you can see that in v4 there is no concept of Duration ([see here](https://github.com/apache/cassandra/blob/185893256f10c14207bffe49ae733fb1a970aec5/doc/native_protocol_v4.spec#L636-L637)), it is only introduced in v5 ([here](https://github.com/apache/cassandra/blob/185893256f10c14207bffe49ae733fb1a970aec5/doc/native_protocol_v5.spec#L841-L843)).
> 
> In order to be compatible with the protocol v4, I guess that Cassandra decides to encode Durations as a "Custom" type. I'm not aware of how the encoding looks, though. I agree with @psarna on how to handle this for Cassandra.


psarna: any updates on this one?

Patch:
diff --git a/scylla/src/frame/response/result.rs b/scylla/src/frame/response/result.rs
index 18e8984b..f9e7538c 100644
--- a/scylla/src/frame/response/result.rs
+++ b/scylla/src/frame/response/result.rs
@@ -1,12 +1,13 @@
 use crate::cql_to_rust::{FromRow, FromRowError};
 use crate::frame::response::event::SchemaChangeEvent;
-use crate::frame::value::Counter;
+use crate::frame::types::vint_decode;
+use crate::frame::value::{Counter, CqlDuration};
 use crate::frame::{frame_errors::ParseError, types};
 use bigdecimal::BigDecimal;
 use byteorder::{BigEndian, ReadBytesExt};
 use bytes::{Buf, Bytes};
+use chrono;
 use chrono::prelude::*;
-use chrono::Duration;
 use num_bigint::BigInt;
 use std::{
     convert::{TryFrom, TryInto},
@@ -49,6 +50,7 @@ pub enum ColumnType {
     Date,
     Decimal,
     Double,
+    Duration,
     Float,
     Int,
     BigInt,
@@ -83,13 +85,14 @@ pub enum CqlValue {
     /// Can be converted to chrono::NaiveDate (-262145-1-1 to 262143-12-31) using as_date
     Date(u32),
     Double(f64),
+    Duration(CqlDuration),
     Empty,
     Float(f32),
     Int(i32),
     BigInt(i64),
     Text(String),
     /// Milliseconds since unix epoch
-    Timestamp(Duration),
+    Timestamp(chrono::Duration),
     Inet(IpAddr),
     List(Vec<CqlValue>),
     Map(Vec<(CqlValue, CqlValue)>),
@@ -105,7 +108,7 @@ pub enum CqlValue {
     SmallInt(i16),
     TinyInt(i8),
     /// Nanoseconds since midnight
-    Time(Duration),
+    Time(chrono::Duration),
     Timeuuid(Uuid),
     Tuple(Vec<Option<CqlValue>>),
     Uuid(Uuid),
@@ -129,12 +132,13 @@ impl CqlValue {
 
         // date_days is u32 then converted to i64
         // then we substract 2^31 - this can't panic
-        let days_since_epoch = Duration::days(date_days.into()) - Duration::days(1 << 31);
+        let days_since_epoch =
+            chrono::Duration::days(date_days.into()) - chrono::Duration::days(1 << 31);
 
         NaiveDate::from_ymd(1970, 1, 1).checked_add_signed(days_since_epoch)
     }
 
-    pub fn as_duration(&self) -> Option<Duration> {
+    pub fn as_duration(&self) -> Option<chrono::Duration> {
         match self {
             Self::Timestamp(i) => Some(*i),
             Self::Time(i) => Some(*i),
@@ -142,6 +146,13 @@ impl CqlValue {
         }
     }
 
+    pub fn as_cql_duration(&self) -> Option<CqlDuration> {
+        match self {
+            Self::Duration(i) => Some(*i),
+            _ => None,
+        }
+    }
+
     pub fn as_counter(&self) -> Option<Counter> {
         match self {
             Self::Counter(i) => Some(*i),
@@ -380,7 +391,10 @@ fn deser_type(buf: &mut &[u8]) -> StdResult<ColumnType, ParseError> {
     Ok(match id {
         0x0000 => {
             let type_str: String = types::read_string(buf)?.to_string();
-            Custom(type_str)
+            match type_str.as_str() {
+                "org.apache.cassandra.db.marshal.DurationType" => Duration,
+                _ => Custom(type_str),
+            }
         }
         0x0001 => Ascii,
         0x0002 => BigInt,
@@ -401,6 +415,7 @@ fn deser_type(buf: &mut &[u8]) -> StdResult<ColumnType, ParseError> {
         0x0012 => Time,
         0x0013 => SmallInt,
         0x0014 => TinyInt,
+        0x0015 => Duration,
         0x0020 => List(Box::new(deser_type(buf)?)),
         0x0021 => Map(Box::new(deser_type(buf)?), Box::new(deser_type(buf)?)),
         0x0022 => Set(Box::new(deser_type(buf)?)),
@@ -654,7 +669,7 @@ fn deser_cql_value(typ: &ColumnType, buf: &mut &[u8]) -> StdResult<CqlValue, Par
             }
             let millis = buf.read_i64::<BigEndian>()?;
 
-            CqlValue::Timestamp(Duration::milliseconds(millis))
+            CqlValue::Timestamp(chrono::Duration::milliseconds(millis))
         }
         Time => {
             if buf.len() != 8 {
@@ -672,7 +687,7 @@ fn deser_cql_value(typ: &ColumnType, buf: &mut &[u8]) -> StdResult<CqlValue, Par
                 }));
             }
 
-            CqlValue::Time(Duration::nanoseconds(nanoseconds))
+            CqlValue::Time(chrono::Duration::nanoseconds(nanoseconds))
         }
         Timeuuid => {
             if buf.len() != 16 {
@@ -684,6 +699,17 @@ fn deser_cql_value(typ: &ColumnType, buf: &mut &[u8]) -> StdResult<CqlValue, Par
             let uuid = uuid::Uuid::from_slice(buf).expect("Deserializing Uuid failed.");
             CqlValue::Timeuuid(uuid)
         }
+        Duration => {
+            let months = i32::try_from(vint_decode(buf)?)?;
+            let days = i32::try_from(vint_decode(buf)?)?;
+            let nanoseconds = vint_decode(buf)?;
+
+            CqlValue::Duration(CqlDuration {
+                months,
+                days,
+                nanoseconds,
+            })
+        }
         Inet => CqlValue::Inet(match buf.len() {
             4 => {
                 let ret = IpAddr::from(<[u8; 4]>::try_from(&buf[0..4])?);
@@ -864,7 +890,7 @@ pub fn deserialize(buf: &mut &[u8]) -> StdResult<Result, ParseError> {
 #[cfg(test)]
 mod tests {
     use crate as scylla;
-    use crate::frame::value::Counter;
+    use crate::frame::value::{Counter, CqlDuration};
     use bigdecimal::BigDecimal;
     use chrono::Duration;
     use chrono::NaiveDate;
@@ -1307,6 +1333,21 @@ mod tests {
         assert_eq!(v, vec![0, 0, 0, 0]);
     }
 
+    #[test]
+    fn test_duration_deserialize() {
+        let bytes = [0xc, 0x12, 0xe2, 0x8c, 0x39, 0xd2];
+        let cql_value: CqlValue =
+            super::deser_cql_value(&ColumnType::Duration, &mut &bytes[..]).unwrap();
+        assert_eq!(
+            cql_value,
+            CqlValue::Duration(CqlDuration {
+                months: 6,
+                days: 9,
+                nanoseconds: 21372137
+            })
+        );
+    }
+
     #[test]
     fn test_deserialize_empty_payload() {
         for (test_type, res_cql) in [
diff --git a/scylla/src/frame/types.rs b/scylla/src/frame/types.rs
index ccdb9549..8e8b36e3 100644
--- a/scylla/src/frame/types.rs
+++ b/scylla/src/frame/types.rs
@@ -525,3 +525,327 @@ fn type_inet() {
     let read_iv6 = read_inet(&mut &*buf).unwrap();
     assert_eq!(iv6, read_iv6);
 }
+
+fn zig_zag_encode(v: i64) -> u64 {
+    ((v >> 63) ^ (v << 1)) as u64
+}
+
+fn zig_zag_decode(v: u64) -> i64 {
+    ((v >> 1) as i64) ^ -((v & 1) as i64)
+}
+
+fn unsigned_vint_encode(v: u64, buf: &mut Vec<u8>) {
+    let mut v = v;
+    let mut number_of_bytes = (639 - 9 * v.leading_zeros()) >> 6;
+    if number_of_bytes <= 1 {
+        return buf.put_u8(v as u8);
+    }
+
+    if number_of_bytes != 9 {
+        let extra_bytes = number_of_bytes - 1;
+        let length_bits = !(0xff >> extra_bytes);
+        v |= (length_bits as u64) << (8 * extra_bytes);
+    } else {
+        buf.put_u8(0xff);
+        number_of_bytes -= 1;
+    }
+    buf.put_uint(v, number_of_bytes as usize)
+}
+
+fn unsigned_vint_decode(buf: &mut &[u8]) -> Result<u64, ParseError> {
+    let first_byte = buf.read_u8()?;
+    let extra_bytes = first_byte.leading_ones() as usize;
+
+    let mut v = if extra_bytes != 8 {
+        let first_byte_bits = first_byte & (0xffu8 >> extra_bytes);
+        (first_byte_bits as u64) << (8 * extra_bytes)
+    } else {
+        0
+    };
+
+    if extra_bytes != 0 {
+        v += buf.read_uint::<BigEndian>(extra_bytes)?;
+    }
+
+    Ok(v)
+}
+
+pub(crate) fn vint_encode(v: i64, buf: &mut Vec<u8>) {
+    unsigned_vint_encode(zig_zag_encode(v), buf)
+}
+
+pub(crate) fn vint_decode(buf: &mut &[u8]) -> Result<i64, ParseError> {
+    unsigned_vint_decode(buf).map(zig_zag_decode)
+}
+
+#[test]
+fn zig_zag_encode_test() {
+    assert_eq!(zig_zag_encode(0), 0);
+    assert_eq!(zig_zag_encode(-1), 1);
+    assert_eq!(zig_zag_encode(1), 2);
+    assert_eq!(zig_zag_encode(-2), 3);
+    assert_eq!(zig_zag_encode(2), 4);
+    assert_eq!(zig_zag_encode(-3), 5);
+    assert_eq!(zig_zag_encode(3), 6);
+}
+
+#[test]
+fn zig_zag_decode_test() {
+    assert_eq!(zig_zag_decode(0), 0);
+    assert_eq!(zig_zag_decode(1), -1);
+    assert_eq!(zig_zag_decode(2), 1);
+    assert_eq!(zig_zag_decode(3), -2);
+    assert_eq!(zig_zag_decode(4), 2);
+    assert_eq!(zig_zag_decode(5), -3);
+    assert_eq!(zig_zag_decode(6), 3);
+}
+
+#[test]
+fn unsigned_vint_encode_and_decode_test() {
+    let unsigned_vint_encoding = vec![
+        (0, vec![0]),
+        (1, vec![1]),
+        (2, vec![2]),
+        ((1 << 2) - 1, vec![3]),
+        (1 << 2, vec![4]),
+        ((1 << 2) + 1, vec![5]),
+        ((1 << 3) - 1, vec![7]),
+        (1 << 3, vec![8]),
+        ((1 << 3) + 1, vec![9]),
+        ((1 << 4) - 1, vec![15]),
+        (1 << 4, vec![16]),
+        ((1 << 4) + 1, vec![17]),
+        ((1 << 5) - 1, vec![31]),
+        (1 << 5, vec![32]),
+        ((1 << 5) + 1, vec![33]),
+        ((1 << 6) - 1, vec![63]),
+        (1 << 6, vec![64]),
+        ((1 << 6) + 1, vec![65]),
+        ((1 << 7) - 1, vec![127]),
+        (1 << 7, vec![128, 128]),
+        ((1 << 7) + 1, vec![128, 129]),
+        ((1 << 8) - 1, vec![128, 255]),
+        (1 << 8, vec![129, 0]),
+        ((1 << 8) + 1, vec![129, 1]),
+        ((1 << 9) - 1, vec![129, 255]),
+        (1 << 9, vec![130, 0]),
+        ((1 << 9) + 1, vec![130, 1]),
+        ((1 << 10) - 1, vec![131, 255]),
+        (1 << 10, vec![132, 0]),
+        ((1 << 10) + 1, vec![132, 1]),
+        ((1 << 11) - 1, vec![135, 255]),
+        (1 << 11, vec![136, 0]),
+        ((1 << 11) + 1, vec![136, 1]),
+        ((1 << 12) - 1, vec![143, 255]),
+        (1 << 12, vec![144, 0]),
+        ((1 << 12) + 1, vec![144, 1]),
+        ((1 << 13) - 1, vec![159, 255]),
+        (1 << 13, vec![160, 0]),
+        ((1 << 13) + 1, vec![160, 1]),
+        ((1 << 14) - 1, vec![191, 255]),
+        (1 << 14, vec![192, 64, 0]),
+        ((1 << 14) + 1, vec![192, 64, 1]),
+        ((1 << 15) - 1, vec![192, 127, 255]),
+        (1 << 15, vec![192, 128, 0]),
+        ((1 << 15) + 1, vec![192, 128, 1]),
+        ((1 << 16) - 1, vec![192, 255, 255]),
+        (1 << 16, vec![193, 0, 0]),
+        ((1 << 16) + 1, vec![193, 0, 1]),
+        ((1 << 17) - 1, vec![193, 255, 255]),
+        (1 << 17, vec![194, 0, 0]),
+        ((1 << 17) + 1, vec![194, 0, 1]),
+        ((1 << 18) - 1, vec![195, 255, 255]),
+        (1 << 18, vec![196, 0, 0]),
+        ((1 << 18) + 1, vec![196, 0, 1]),
+        ((1 << 19) - 1, vec![199, 255, 255]),
+        (1 << 19, vec![200, 0, 0]),
+        ((1 << 19) + 1, vec![200, 0, 1]),
+        ((1 << 20) - 1, vec![207, 255, 255]),
+        (1 << 20, vec![208, 0, 0]),
+        ((1 << 20) + 1, vec![208, 0, 1]),
+        ((1 << 21) - 1, vec![223, 255, 255]),
+        (1 << 21, vec![224, 32, 0, 0]),
+        ((1 << 21) + 1, vec![224, 32, 0, 1]),
+        ((1 << 22) - 1, vec![224, 63, 255, 255]),
+        (1 << 22, vec![224, 64, 0, 0]),
+        ((1 << 22) + 1, vec![224, 64, 0, 1]),
+        ((1 << 23) - 1, vec![224, 127, 255, 255]),
+        (1 << 23, vec![224, 128, 0, 0]),
+        ((1 << 23) + 1, vec![224, 128, 0, 1]),
+        ((1 << 24) - 1, vec![224, 255, 255, 255]),
+        (1 << 24, vec![225, 0, 0, 0]),
+        ((1 << 24) + 1, vec![225, 0, 0, 1]),
+        ((1 << 25) - 1, vec![225, 255, 255, 255]),
+        (1 << 25, vec![226, 0, 0, 0]),
+        ((1 << 25) + 1, vec![226, 0, 0, 1]),
+        ((1 << 26) - 1, vec![227, 255, 255, 255]),
+        (1 << 26, vec![228, 0, 0, 0]),
+        ((1 << 26) + 1, vec![228, 0, 0, 1]),
+        ((1 << 27) - 1, vec![231, 255, 255, 255]),
+        (1 << 27, vec![232, 0, 0, 0]),
+        ((1 << 27) + 1, vec![232, 0, 0, 1]),
+        ((1 << 28) - 1, vec![239, 255, 255, 255]),
+        (1 << 28, vec![240, 16, 0, 0, 0]),
+        ((1 << 28) + 1, vec![240, 16, 0, 0, 1]),
+        ((1 << 29) - 1, vec![240, 31, 255, 255, 255]),
+        (1 << 29, vec![240, 32, 0, 0, 0]),
+        ((1 << 29) + 1, vec![240, 32, 0, 0, 1]),
+        ((1 << 30) - 1, vec![240, 63, 255, 255, 255]),
+        (1 << 30, vec![240, 64, 0, 0, 0]),
+        ((1 << 30) + 1, vec![240, 64, 0, 0, 1]),
+        ((1 << 31) - 1, vec![240, 127, 255, 255, 255]),
+        (1 << 31, vec![240, 128, 0, 0, 0]),
+        ((1 << 31) + 1, vec![240, 128, 0, 0, 1]),
+        ((1 << 32) - 1, vec![240, 255, 255, 255, 255]),
+        (1 << 32, vec![241, 0, 0, 0, 0]),
+        ((1 << 32) + 1, vec![241, 0, 0, 0, 1]),
+        ((1 << 33) - 1, vec![241, 255, 255, 255, 255]),
+        (1 << 33, vec![242, 0, 0, 0, 0]),
+        ((1 << 33) + 1, vec![242, 0, 0, 0, 1]),
+        ((1 << 34) - 1, vec![243, 255, 255, 255, 255]),
+        (1 << 34, vec![244, 0, 0, 0, 0]),
+        ((1 << 34) + 1, vec![244, 0, 0, 0, 1]),
+        ((1 << 35) - 1, vec![247, 255, 255, 255, 255]),
+        (1 << 35, vec![248, 8, 0, 0, 0, 0]),
+        ((1 << 35) + 1, vec![248, 8, 0, 0, 0, 1]),
+        ((1 << 36) - 1, vec![248, 15, 255, 255, 255, 255]),
+        (1 << 36, vec![248, 16, 0, 0, 0, 0]),
+        ((1 << 36) + 1, vec![248, 16, 0, 0, 0, 1]),
+        ((1 << 37) - 1, vec![248, 31, 255, 255, 255, 255]),
+        (1 << 37, vec![248, 32, 0, 0, 0, 0]),
+        ((1 << 37) + 1, vec![248, 32, 0, 0, 0, 1]),
+        ((1 << 38) - 1, vec![248, 63, 255, 255, 255, 255]),
+        (1 << 38, vec![248, 64, 0, 0, 0, 0]),
+        ((1 << 38) + 1, vec![248, 64, 0, 0, 0, 1]),
+        ((1 << 39) - 1, vec![248, 127, 255, 255, 255, 255]),
+        (1 << 39, vec![248, 128, 0, 0, 0, 0]),
+        ((1 << 39) + 1, vec![248, 128, 0, 0, 0, 1]),
+        ((1 << 40) - 1, vec![248, 255, 255, 255, 255, 255]),
+        (1 << 40, vec![249, 0, 0, 0, 0, 0]),
+        ((1 << 40) + 1, vec![249, 0, 0, 0, 0, 1]),
+        ((1 << 41) - 1, vec![249, 255, 255, 255, 255, 255]),
+        (1 << 41, vec![250, 0, 0, 0, 0, 0]),
+        ((1 << 41) + 1, vec![250, 0, 0, 0, 0, 1]),
+        ((1 << 42) - 1, vec![251, 255, 255, 255, 255, 255]),
+        (1 << 42, vec![252, 4, 0, 0, 0, 0, 0]),
+        ((1 << 42) + 1, vec![252, 4, 0, 0, 0, 0, 1]),
+        ((1 << 43) - 1, vec![252, 7, 255, 255, 255, 255, 255]),
+        (1 << 43, vec![252, 8, 0, 0, 0, 0, 0]),
+        ((1 << 43) + 1, vec![252, 8, 0, 0, 0, 0, 1]),
+        ((1 << 44) - 1, vec![252, 15, 255, 255, 255, 255, 255]),
+        (1 << 44, vec![252, 16, 0, 0, 0, 0, 0]),
+        ((1 << 44) + 1, vec![252, 16, 0, 0, 0, 0, 1]),
+        ((1 << 45) - 1, vec![252, 31, 255, 255, 255, 255, 255]),
+        (1 << 45, vec![252, 32, 0, 0, 0, 0, 0]),
+        ((1 << 45) + 1, vec![252, 32, 0, 0, 0, 0, 1]),
+        ((1 << 46) - 1, vec![252, 63, 255, 255, 255, 255, 255]),
+        (1 << 46, vec![252, 64, 0, 0, 0, 0, 0]),
+        ((1 << 46) + 1, vec![252, 64, 0, 0, 0, 0, 1]),
+        ((1 << 47) - 1, vec![252, 127, 255, 255, 255, 255, 255]),
+        (1 << 47, vec![252, 128, 0, 0, 0, 0, 0]),
+        ((1 << 47) + 1, vec![252, 128, 0, 0, 0, 0, 1]),
+        ((1 << 48) - 1, vec![252, 255, 255, 255, 255, 255, 255]),
+        (1 << 48, vec![253, 0, 0, 0, 0, 0, 0]),
+        ((1 << 48) + 1, vec![253, 0, 0, 0, 0, 0, 1]),
+        ((1 << 49) - 1, vec![253, 255, 255, 255, 255, 255, 255]),
+        (1 << 49, vec![254, 2, 0, 0, 0, 0, 0, 0]),
+        ((1 << 49) + 1, vec![254, 2, 0, 0, 0, 0, 0, 1]),
+        ((1 << 50) - 1, vec![254, 3, 255, 255, 255, 255, 255, 255]),
+        (1 << 50, vec![254, 4, 0, 0, 0, 0, 0, 0]),
+        ((1 << 50) + 1, vec![254, 4, 0, 0, 0, 0, 0, 1]),
+        ((1 << 51) - 1, vec![254, 7, 255, 255, 255, 255, 255, 255]),
+        (1 << 51, vec![254, 8, 0, 0, 0, 0, 0, 0]),
+        ((1 << 51) + 1, vec![254, 8, 0, 0, 0, 0, 0, 1]),
+        ((1 << 52) - 1, vec![254, 15, 255, 255, 255, 255, 255, 255]),
+        (1 << 52, vec![254, 16, 0, 0, 0, 0, 0, 0]),
+        ((1 << 52) + 1, vec![254, 16, 0, 0, 0, 0, 0, 1]),
+        ((1 << 53) - 1, vec![254, 31, 255, 255, 255, 255, 255, 255]),
+        (1 << 53, vec![254, 32, 0, 0, 0, 0, 0, 0]),
+        ((1 << 53) + 1, vec![254, 32, 0, 0, 0, 0, 0, 1]),
+        ((1 << 54) - 1, vec![254, 63, 255, 255, 255, 255, 255, 255]),
+        (1 << 54, vec![254, 64, 0, 0, 0, 0, 0, 0]),
+        ((1 << 54) + 1, vec![254, 64, 0, 0, 0, 0, 0, 1]),
+        ((1 << 55) - 1, vec![254, 127, 255, 255, 255, 255, 255, 255]),
+        (1 << 55, vec![254, 128, 0, 0, 0, 0, 0, 0]),
+        ((1 << 55) + 1, vec![254, 128, 0, 0, 0, 0, 0, 1]),
+        ((1 << 56) - 1, vec![254, 255, 255, 255, 255, 255, 255, 255]),
+        (1 << 56, vec![255, 1, 0, 0, 0, 0, 0, 0, 0]),
+        ((1 << 56) + 1, vec![255, 1, 0, 0, 0, 0, 0, 0, 1]),
+        (
+            (1 << 57) - 1,
+            vec![255, 1, 255, 255, 255, 255, 255, 255, 255],
+        ),
+        (1 << 57, vec![255, 2, 0, 0, 0, 0, 0, 0, 0]),
+        ((1 << 57) + 1, vec![255, 2, 0, 0, 0, 0, 0, 0, 1]),
+        (
+            (1 << 58) - 1,
+            vec![255, 3, 255, 255, 255, 255, 255, 255, 255],
+        ),
+        (1 << 58, vec![255, 4, 0, 0, 0, 0, 0, 0, 0]),
+        ((1 << 58) + 1, vec![255, 4, 0, 0, 0, 0, 0, 0, 1]),
+        (
+            (1 << 59) - 1,
+            vec![255, 7, 255, 255, 255, 255, 255, 255, 255],
+        ),
+        (1 << 59, vec![255, 8, 0, 0, 0, 0, 0, 0, 0]),
+        ((1 << 59) + 1, vec![255, 8, 0, 0, 0, 0, 0, 0, 1]),
+        (
+            (1 << 60) - 1,
+            vec![255, 15, 255, 255, 255, 255, 255, 255, 255],
+        ),
+        (1 << 60, vec![255, 16, 0, 0, 0, 0, 0, 0, 0]),
+        ((1 << 60) + 1, vec![255, 16, 0, 0, 0, 0, 0, 0, 1]),
+        (
+            (1 << 61) - 1,
+            vec![255, 31, 255, 255, 255, 255, 255, 255, 255],
+        ),
+        (1 << 61, vec![255, 32, 0, 0, 0, 0, 0, 0, 0]),
+        ((1 << 61) + 1, vec![255, 32, 0, 0, 0, 0, 0, 0, 1]),
+        (
+            (1 << 62) - 1,
+            vec![255, 63, 255, 255, 255, 255, 255, 255, 255],
+        ),
+        (1 << 62, vec![255, 64, 0, 0, 0, 0, 0, 0, 0]),
+        ((1 << 62) + 1, vec![255, 64, 0, 0, 0, 0, 0, 0, 1]),
+        (
+            (1 << 63) - 1,
+            vec![255, 127, 255, 255, 255, 255, 255, 255, 255],
+        ),
+        (1 << 63, vec![255, 128, 0, 0, 0, 0, 0, 0, 0]),
+        ((1 << 63) + 1, vec![255, 128, 0, 0, 0, 0, 0, 0, 1]),
+        (u64::MAX, vec![255, 255, 255, 255, 255, 255, 255, 255, 255]),
+    ];
+
+    let mut buf = Vec::new();
+
+    for (v, result) in unsigned_vint_encoding.into_iter() {
+        unsigned_vint_encode(v, &mut buf);
+        assert_eq!(buf, result);
+        let decoded_v = unsigned_vint_decode(&mut buf.as_slice()).unwrap();
+        assert_eq!(v, decoded_v);
+        buf.clear();
+    }
+}
+
+#[test]
+fn vint_encode_and_decode_test() {
+    let mut buf: Vec<u8> = Vec::with_capacity(128);
+
+    let mut check = |n: i64| {
+        vint_encode(n, &mut buf);
+        assert_eq!(vint_decode(&mut buf.as_slice()).unwrap(), n);
+        buf.clear();
+    };
+
+    for i in 0..63 {
+        check((1 << i) - 1);
+        check(1 - (1 << i));
+        check(1 << i);
+        check(-(1 << i));
+        check((1 << i) + 1);
+        check(-1 - (1 << i));
+    }
+    check(i64::MAX);
+    check(-i64::MAX);
+    check(i64::MIN)
+}
diff --git a/scylla/src/frame/value.rs b/scylla/src/frame/value.rs
index 32f8c418..af127f0f 100644
--- a/scylla/src/frame/value.rs
+++ b/scylla/src/frame/value.rs
@@ -12,6 +12,7 @@ use thiserror::Error;
 use uuid::Uuid;
 
 use super::response::result::CqlValue;
+use super::types::vint_encode;
 
 /// Every value being sent in a query must implement this trait
 /// serialize() should write the Value as [bytes] to the provided buffer
@@ -60,6 +61,14 @@ pub struct SerializedValues {
     contains_names: bool,
 }
 
+/// Represents a CQL Duration value
+#[derive(Clone, Debug, Copy, PartialEq)]
+pub struct CqlDuration {
+    pub months: i32,
+    pub days: i32,
+    pub nanoseconds: i64,
+}
+
 #[derive(Debug, Error, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
 pub enum SerializeValuesError {
     #[error("Too many values to add, max 32 767 values can be sent in a request")]
@@ -437,6 +446,23 @@ impl Value for Counter {
     }
 }
 
+impl Value for CqlDuration {
+    fn serialize(&self, buf: &mut Vec<u8>) -> Result<(), ValueTooBig> {
+        let bytes_num_pos: usize = buf.len();
+        buf.put_i32(0);
+
+        vint_encode(self.months as i64, buf);
+        vint_encode(self.days as i64, buf);
+        vint_encode(self.nanoseconds, buf);
+
+        let written_bytes: usize = buf.len() - bytes_num_pos - 4;
+        let written_bytes_i32: i32 = written_bytes.try_into().map_err(|_| ValueTooBig)?;
+        buf[bytes_num_pos..(bytes_num_pos + 4)].copy_from_slice(&written_bytes_i32.to_be_bytes());
+
+        Ok(())
+    }
+}
+
 impl<V: Value> Value for MaybeUnset<V> {
     fn serialize(&self, buf: &mut Vec<u8>) -> Result<(), ValueTooBig> {
         match self {
@@ -566,6 +592,7 @@ impl Value for CqlValue {
             }
 
             CqlValue::Date(d) => Date(*d).serialize(buf),
+            CqlValue::Duration(d) => d.serialize(buf),
             CqlValue::Timestamp(t) => Timestamp(*t).serialize(buf),
             CqlValue::Time(t) => Time(*t).serialize(buf),
 


Test patch:
diff --git a/scylla/src/frame/value_tests.rs b/scylla/src/frame/value_tests.rs
index a2121de2..40bedf37 100644
--- a/scylla/src/frame/value_tests.rs
+++ b/scylla/src/frame/value_tests.rs
@@ -5,6 +5,7 @@ use super::{
         Unset, Value, ValueList, ValueTooBig,
     },
 };
+use crate::frame::value::CqlDuration;
 use crate::{Session, SessionBuilder};
 use bytes::BufMut;
 use chrono::{Duration, NaiveDate};
@@ -723,3 +724,83 @@ async fn test_cqlvalue_udt() {
 
     assert_eq!(received_udt_cql_value, &udt_cql_value);
 }
+
+#[tokio::test]
+async fn test_cqlvalue_duration() {
+    let uri = env::var("SCYLLA_URI").unwrap_or_else(|_| "127.0.0.1:9042".to_string());
+    let session: Session = SessionBuilder::new().known_node(uri).build().await.unwrap();
+
+    let ks = crate::transport::session_test::unique_name();
+    session
+        .query(
+            format!(
+                "CREATE KEYSPACE IF NOT EXISTS {} WITH REPLICATION = \
+                {{'class' : 'SimpleStrategy', 'replication_factor' : 1}}",
+                ks
+            ),
+            &[],
+        )
+        .await
+        .unwrap();
+    session.use_keyspace(&ks, false).await.unwrap();
+
+    let duration_cql_value = CqlValue::Duration(CqlDuration {
+        months: 6,
+        days: 9,
+        nanoseconds: 21372137,
+    });
+
+    let fixture_queries = vec![
+        ("CREATE TABLE IF NOT EXISTS cqlvalue_duration_test (pk int, ck int, v duration, primary key (pk, ck))", vec![],),
+        ("INSERT INTO cqlvalue_duration_test (pk, ck, v) VALUES (0, 0, ?)", vec![&duration_cql_value,],),
+        ("INSERT INTO cqlvalue_duration_test (pk, ck, v) VALUES (0, 1, 89h4m48s)", vec![],),
+        ("INSERT INTO cqlvalue_duration_test (pk, ck, v) VALUES (0, 2, PT89H8M53S)", vec![],),
+        ("INSERT INTO cqlvalue_duration_test (pk, ck, v) VALUES (0, 3, P0000-00-00T89:09:09)", vec![],),
+    ];
+
+    for query in fixture_queries {
+        session.query(query.0, query.1).await.unwrap();
+    }
+
+    let rows = session
+        .query(
+            "SELECT v FROM cqlvalue_duration_test WHERE pk = ?",
+            (CqlValue::Int(0),),
+        )
+        .await
+        .unwrap()
+        .rows
+        .unwrap();
+
+    assert_eq!(rows.len(), 4);
+    assert_eq!(rows[0].columns.len(), 1);
+
+    assert_eq!(rows[0].columns[0].as_ref().unwrap(), &duration_cql_value);
+
+    assert_eq!(
+        rows[1].columns[0].as_ref().unwrap(),
+        &CqlValue::Duration(CqlDuration {
+            months: 0,
+            days: 0,
+            nanoseconds: 320_688_000_000_000,
+        })
+    );
+
+    assert_eq!(
+        rows[2].columns[0].as_ref().unwrap(),
+        &CqlValue::Duration(CqlDuration {
+            months: 0,
+            days: 0,
+            nanoseconds: 320_933_000_000_000,
+        })
+    );
+
+    assert_eq!(
+        rows[3].columns[0].as_ref().unwrap(),
+        &CqlValue::Duration(CqlDuration {
+            months: 0,
+            days: 0,
+            nanoseconds: 320_949_000_000_000,
+        })
+    );
+}