Repository: scylladb/scylla-rust-driver
Pull request #146 (2021-01-28 15:56:52)

Remove needless refresh_topology()
`Session::connect()` now does `refresh_topology()` on startup so we call it twice in examples, which is not needed.


Patch:
diff --git a/examples/Cargo.toml b/examples/Cargo.toml
index d90a379f..756fe217 100644
--- a/examples/Cargo.toml
+++ b/examples/Cargo.toml
@@ -36,4 +36,4 @@ path = "select-paging.rs"
 
 [[example]]
 name = "user-defined-type"
-path = "user-defined-type.rs"
\ No newline at end of file
+path = "user-defined-type.rs"
diff --git a/examples/basic.rs b/examples/basic.rs
index 78957769..99f24e7b 100644
--- a/examples/basic.rs
+++ b/examples/basic.rs
@@ -11,7 +11,6 @@ async fn main() -> Result<()> {
     println!("Connecting to {} ...", uri);
 
     let session = Session::connect(uri, None).await?;
-    session.refresh_topology().await?;
 
     session.query("CREATE KEYSPACE IF NOT EXISTS ks WITH REPLICATION = {'class' : 'SimpleStrategy', 'replication_factor' : 1}", &[]).await?;
 
diff --git a/examples/compare-tokens.rs b/examples/compare-tokens.rs
index ed277c24..d4a6da4f 100644
--- a/examples/compare-tokens.rs
+++ b/examples/compare-tokens.rs
@@ -11,7 +11,6 @@ async fn main() -> Result<()> {
     println!("Connecting to {} ...", uri);
 
     let session = Session::connect(uri, None).await?;
-    session.refresh_topology().await?;
 
     session.query("CREATE KEYSPACE IF NOT EXISTS ks WITH REPLICATION = {'class' : 'SimpleStrategy', 'replication_factor' : 1}", &[]).await?;
 
diff --git a/examples/select-paging.rs b/examples/select-paging.rs
index 12904a1d..4007e719 100644
--- a/examples/select-paging.rs
+++ b/examples/select-paging.rs
@@ -10,7 +10,6 @@ async fn main() -> Result<()> {
     println!("Connecting to {} ...", uri);
 
     let session = Session::connect(uri, None).await?;
-    session.refresh_topology().await?;
 
     session.query("CREATE KEYSPACE IF NOT EXISTS ks WITH REPLICATION = {'class' : 'SimpleStrategy', 'replication_factor' : 1}", &[]).await?;
 
diff --git a/examples/user-defined-type.rs b/examples/user-defined-type.rs
index d05078d9..f4e22a36 100644
--- a/examples/user-defined-type.rs
+++ b/examples/user-defined-type.rs
@@ -11,7 +11,6 @@ async fn main() -> Result<()> {
     println!("Connecting to {} ...", uri);
 
     let session = Session::connect(uri, None).await?;
-    session.refresh_topology().await?;
 
     session.query("CREATE KEYSPACE IF NOT EXISTS ks WITH REPLICATION = {'class' : 'SimpleStrategy', 'replication_factor' : 1}", &[]).await?;
 
diff --git a/scylla-macros/src/from_user_type.rs b/scylla-macros/src/from_user_type.rs
index aee4a751..06e67f0a 100644
--- a/scylla-macros/src/from_user_type.rs
+++ b/scylla-macros/src/from_user_type.rs
@@ -25,12 +25,12 @@ pub fn from_user_type_derive(tokens_input: TokenStream) -> TokenStream {
         impl FromCQLVal<scylla::frame::response::result::CQLValue> for #struct_name {
             fn from_cql(cql_val: scylla::frame::response::result::CQLValue)
             -> Result<Self, scylla::cql_to_rust::FromCQLValError> {
-                use std::collections::HashMap;
+                use std::collections::BTreeMap;
                 use scylla::cql_to_rust::{FromCQLVal, FromCQLValError};
                 use scylla::frame::response::result::CQLValue;
 
                 // Interpret CQLValue as CQlValue::UserDefinedType
-                let mut fields: HashMap<String, Option<CQLValue>> = match cql_val {
+                let mut fields: BTreeMap<String, Option<CQLValue>> = match cql_val {
                     CQLValue::UserDefinedType{fields, ..} => fields,
                     _ => return Err(FromCQLValError::BadCQLType),
                 };
diff --git a/scylla/Cargo.toml b/scylla/Cargo.toml
index 1ff2e156..aed37d2e 100644
--- a/scylla/Cargo.toml
+++ b/scylla/Cargo.toml
@@ -11,15 +11,15 @@ categories = ["database"]
 [dependencies]
 scylla-macros = { version = "0.1.0", path = "../scylla-macros"}
 byteorder = "1.3.4"
-bytes = "0.6.0"
+bytes = "1.0.1"
 futures = "0.3.6"
 histogram = "0.6.9"
 num_enum = "0.5"
 compress = "0.2.1"
-tokio = { version = "0.3.0", features = ["net", "time", "io-util", "sync", "rt", "macros"] }
+tokio = { version = "1.1.0", features = ["net", "time", "io-util", "sync", "rt", "macros"] }
 snap = "1.0"
 uuid = "0.8.1"
-rand = "0.7.3"
+rand = "0.8.3"
 serde_json = "1.0.60"
 thiserror = "1.0"
 
diff --git a/scylla/src/frame/frame_errors.rs b/scylla/src/frame/frame_errors.rs
index 377f5854..81692f0a 100644
--- a/scylla/src/frame/frame_errors.rs
+++ b/scylla/src/frame/frame_errors.rs
@@ -1,4 +1,5 @@
 use super::response;
+use crate::cql_to_rust::CQLTypeError;
 use crate::frame::value::SerializeValuesError;
 use thiserror::Error;
 
@@ -36,4 +37,6 @@ pub enum ParseError {
     TypeNotImplemented(i16),
     #[error(transparent)]
     SerializeValuesError(#[from] SerializeValuesError),
+    #[error(transparent)]
+    CQLTypeError(#[from] CQLTypeError),
 }
diff --git a/scylla/src/frame/mod.rs b/scylla/src/frame/mod.rs
index 2ab64b78..213d8f96 100644
--- a/scylla/src/frame/mod.rs
+++ b/scylla/src/frame/mod.rs
@@ -7,6 +7,9 @@ pub mod value;
 #[cfg(test)]
 mod value_tests;
 
+#[cfg(test)]
+mod cql_collections_test;
+
 use crate::frame::frame_errors::FrameError;
 use crate::transport::Compression;
 use bytes::{Buf, BufMut, Bytes};
diff --git a/scylla/src/frame/response/cql_to_rust.rs b/scylla/src/frame/response/cql_to_rust.rs
index 2ab7c722..10ba0721 100644
--- a/scylla/src/frame/response/cql_to_rust.rs
+++ b/scylla/src/frame/response/cql_to_rust.rs
@@ -1,4 +1,6 @@
 use super::result::{CQLValue, Row};
+use std::collections::HashMap;
+use std::hash::Hash;
 use std::net::IpAddr;
 use thiserror::Error;
 
@@ -10,6 +12,12 @@ pub enum FromRowError {
     RowTooShort,
 }
 
+#[derive(Error, Debug, PartialEq, Eq)]
+pub enum CQLTypeError {
+    #[error("Invalid number of set elements: {0}")]
+    InvalidNumberOfElements(i32),
+}
+
 /// 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
@@ -76,6 +84,19 @@ impl<T: FromCQLVal<CQLValue>> FromCQLVal<CQLValue> for Vec<T> {
     }
 }
 
+impl<T1: FromCQLVal<CQLValue> + Eq + Hash, T2: FromCQLVal<CQLValue>> FromCQLVal<CQLValue>
+    for HashMap<T1, T2>
+{
+    fn from_cql(cql_val: CQLValue) -> Result<Self, FromCQLValError> {
+        let vec = cql_val.into_pair_vec().ok_or(FromCQLValError::BadCQLType)?;
+        let mut res = HashMap::with_capacity(vec.len());
+        for (key, value) in vec {
+            res.insert(T1::from_cql(key)?, T2::from_cql(value)?);
+        }
+        Ok(res)
+    }
+}
+
 // This macro implements FromRow for tuple of types that have FromCQLVal
 macro_rules! impl_tuple_from_row {
     ( $($Ti:tt),+ ) => {
@@ -117,6 +138,47 @@ impl_tuple_from_row!(T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14
 impl_tuple_from_row!(T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15);
 impl_tuple_from_row!(T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16);
 
+macro_rules! impl_tuple_from_cql {
+    ( $($Ti:tt),+ ) => {
+        impl<$($Ti),+> FromCQLVal<CQLValue> for ($($Ti,)+)
+        where
+            $($Ti: FromCQLVal<CQLValue>),+
+        {
+            fn from_cql(cql_val: CQLValue) -> Result<Self, FromCQLValError> {
+                let tuple_fields = match cql_val {
+                    CQLValue::Tuple(fields) => fields,
+                    _ => return Err(FromCQLValError::BadCQLType)
+                };
+
+                let mut tuple_fields_iter = tuple_fields.into_iter();
+
+                Ok((
+                    $(
+                        $Ti::from_cql(tuple_fields_iter.next().ok_or(FromCQLValError::BadCQLType) ?) ?
+                    ,)+
+                ))
+            }
+        }
+    }
+}
+
+impl_tuple_from_cql!(T1);
+impl_tuple_from_cql!(T1, T2);
+impl_tuple_from_cql!(T1, T2, T3);
+impl_tuple_from_cql!(T1, T2, T3, T4);
+impl_tuple_from_cql!(T1, T2, T3, T4, T5);
+impl_tuple_from_cql!(T1, T2, T3, T4, T5, T6);
+impl_tuple_from_cql!(T1, T2, T3, T4, T5, T6, T7);
+impl_tuple_from_cql!(T1, T2, T3, T4, T5, T6, T7, T8);
+impl_tuple_from_cql!(T1, T2, T3, T4, T5, T6, T7, T8, T9);
+impl_tuple_from_cql!(T1, T2, T3, T4, T5, T6, T7, T8, T9, T10);
+impl_tuple_from_cql!(T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11);
+impl_tuple_from_cql!(T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12);
+impl_tuple_from_cql!(T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13);
+impl_tuple_from_cql!(T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14);
+impl_tuple_from_cql!(T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15);
+impl_tuple_from_cql!(T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15, T16);
+
 #[cfg(test)]
 mod tests {
     use super::{CQLValue, FromCQLVal, FromCQLValError, FromRow, FromRowError, Row};
diff --git a/scylla/src/frame/response/result.rs b/scylla/src/frame/response/result.rs
index c6856528..49539a72 100644
--- a/scylla/src/frame/response/result.rs
+++ b/scylla/src/frame/response/result.rs
@@ -1,15 +1,14 @@
-use super::cql_to_rust::FromRowError;
-use crate::cql_to_rust::FromRow;
-use crate::frame::frame_errors::ParseError;
-use crate::frame::types;
+use crate::cql_to_rust::{FromRow, FromRowError};
+use crate::frame::{frame_errors::ParseError, types};
 use byteorder::{BigEndian, ReadBytesExt};
 use bytes::{Buf, Bytes};
-use std::collections::HashMap;
-use std::convert::TryFrom;
-use std::convert::TryInto;
-use std::net::IpAddr;
-use std::result::Result as StdResult;
-use std::str;
+use std::{
+    collections::BTreeMap,
+    convert::{TryFrom, TryInto},
+    net::IpAddr,
+    result::Result as StdResult,
+    str,
+};
 
 #[derive(Debug)]
 pub struct SetKeyspace {
@@ -41,28 +40,33 @@ enum ColumnType {
     BigInt,
     Text,
     Inet,
+    List(Box<ColumnType>),
+    Map(Box<ColumnType>, Box<ColumnType>),
     Set(Box<ColumnType>),
     UserDefinedType {
         type_name: String,
         keyspace: String,
         field_types: Vec<(String, ColumnType)>,
     },
-    // TODO
+    Tuple(Vec<ColumnType>),
 }
 
-#[derive(Debug)]
+#[derive(Debug, PartialEq, Eq, Hash)]
 pub enum CQLValue {
     Ascii(String),
     Int(i32),
     BigInt(i64),
     Text(String),
     Inet(IpAddr),
+    List(Vec<CQLValue>),
+    Map(Vec<(CQLValue, CQLValue)>),
     Set(Vec<CQLValue>),
     UserDefinedType {
         keyspace: String,
         type_name: String,
-        fields: HashMap<String, Option<CQLValue>>,
-    }, // TODO
+        fields: BTreeMap<String, Option<CQLValue>>,
+    },
+    Tuple(Vec<CQLValue>),
 }
 
 impl CQLValue {
@@ -109,6 +113,13 @@ impl CQLValue {
         }
     }
 
+    pub fn as_list(&self) -> Option<&Vec<CQLValue>> {
+        match self {
+            Self::List(s) => Some(&s),
+            _ => None,
+        }
+    }
+
     pub fn as_set(&self) -> Option<&Vec<CQLValue>> {
         match self {
             Self::Set(s) => Some(&s),
@@ -118,11 +129,19 @@ impl CQLValue {
 
     pub fn into_vec(self) -> Option<Vec<CQLValue>> {
         match self {
+            Self::List(s) => Some(s),
             Self::Set(s) => Some(s),
             _ => None,
         }
     }
 
+    pub fn into_pair_vec(self) -> Option<Vec<(CQLValue, CQLValue)>> {
+        match self {
+            Self::Map(s) => Some(s),
+            _ => None,
+        }
+    }
+
     // TODO
 }
 
@@ -193,6 +212,8 @@ fn deser_type(buf: &mut &[u8]) -> StdResult<ColumnType, ParseError> {
         0x0009 => Int,
         0x000D => Text,
         0x0010 => Inet,
+        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)?)),
         0x0030 => {
             let keyspace_name: String = types::read_string(buf)?.to_string();
@@ -214,6 +235,14 @@ fn deser_type(buf: &mut &[u8]) -> StdResult<ColumnType, ParseError> {
                 field_types,
             }
         }
+        0x0031 => {
+            let len: usize = types::read_short(buf)?.try_into()?;
+            let mut types = Vec::with_capacity(len as usize);
+            for _ in 0..len {
+                types.push(deser_type(buf)?);
+            }
+            Tuple(types)
+        }
         id => {
             // TODO implement other types
             return Err(ParseError::TypeNotImplemented(id));
@@ -355,14 +384,34 @@ fn deser_cql_value(typ: &ColumnType, buf: &mut &[u8]) -> StdResult<CQLValue, Par
                 )));
             }
         }),
-        Set(typ) => {
+        List(type_name) => {
+            let len: usize = types::read_int(buf)?.try_into()?;
+            let mut res = Vec::with_capacity(len);
+            for _ in 0..len {
+                let mut b = types::read_bytes(buf)?;
+                res.push(deser_cql_value(type_name, &mut b)?);
+            }
+            CQLValue::List(res)
+        }
+        Map(key_type, value_type) => {
+            let len: usize = types::read_int(buf)?.try_into()?;
+            let mut res = Vec::with_capacity(len);
+            for _ in 0..len {
+                let mut b = types::read_bytes(buf)?;
+                let key = deser_cql_value(key_type, &mut b)?;
+                b = types::read_bytes(buf)?;
+                let val = deser_cql_value(value_type, &mut b)?;
+                res.push((key, val));
+            }
+            CQLValue::Map(res)
+        }
+        Set(type_name) => {
             let len: usize = types::read_int(buf)?.try_into()?;
-
             let mut res = Vec::with_capacity(len);
             for _ in 0..len {
                 // TODO: is `null` allowed as set element? Should we use read_bytes_opt?
                 let mut b = types::read_bytes(buf)?;
-                res.push(deser_cql_value(typ, &mut b)?);
+                res.push(deser_cql_value(type_name, &mut b)?);
             }
             CQLValue::Set(res)
         }
@@ -371,8 +420,7 @@ fn deser_cql_value(typ: &ColumnType, buf: &mut &[u8]) -> StdResult<CQLValue, Par
             keyspace,
             field_types,
         } => {
-            let mut fields: HashMap<String, Option<CQLValue>> =
-                HashMap::with_capacity(field_types.len());
+            let mut fields: BTreeMap<String, Option<CQLValue>> = BTreeMap::new();
 
             for (field_name, field_type) in field_types {
                 let mut field_value: Option<CQLValue> = None;
@@ -389,6 +437,14 @@ fn deser_cql_value(typ: &ColumnType, buf: &mut &[u8]) -> StdResult<CQLValue, Par
                 fields,
             }
         }
+        Tuple(type_names) => {
+            let mut res = Vec::with_capacity(type_names.len());
+            for type_name in type_names {
+                let mut b = types::read_bytes(buf)?;
+                res.push(deser_cql_value(type_name, &mut b)?);
+            }
+            CQLValue::Tuple(res)
+        }
     })
 }
 
@@ -460,3 +516,59 @@ pub fn deserialize(buf: &mut &[u8]) -> StdResult<Result, ParseError> {
         }
     })
 }
+
+#[cfg(test)]
+mod tests {
+    use crate as scylla;
+    use scylla::frame::response::result::CQLValue;
+
+    #[test]
+    fn test_list_from_cql() {
+        let mut my_vec: Vec<CQLValue> = Vec::new();
+
+        my_vec.push(CQLValue::Int(20));
+        my_vec.push(CQLValue::Int(2));
+        my_vec.push(CQLValue::Int(13));
+
+        let cql: CQLValue = CQLValue::List(my_vec);
+        let decoded = cql.into_vec().unwrap();
+
+        assert_eq!(decoded[0], CQLValue::Int(20));
+        assert_eq!(decoded[1], CQLValue::Int(2));
+        assert_eq!(decoded[2], CQLValue::Int(13));
+    }
+
+    #[test]
+    fn test_set_from_cql() {
+        let mut my_vec: Vec<CQLValue> = Vec::new();
+
+        my_vec.push(CQLValue::Int(20));
+        my_vec.push(CQLValue::Int(2));
+        my_vec.push(CQLValue::Int(13));
+
+        let cql: CQLValue = CQLValue::Set(my_vec);
+        let decoded = cql.as_set().unwrap();
+
+        assert_eq!(decoded[0], CQLValue::Int(20));
+        assert_eq!(decoded[1], CQLValue::Int(2));
+        assert_eq!(decoded[2], CQLValue::Int(13));
+    }
+
+    #[test]
+    fn test_map_from_cql() {
+        let mut my_vec: Vec<(CQLValue, CQLValue)> = Vec::new();
+
+        my_vec.push((CQLValue::Int(20), CQLValue::Int(21)));
+        my_vec.push((CQLValue::Int(2), CQLValue::Int(3)));
+
+        let cql: CQLValue = CQLValue::Map(my_vec);
+
+        let decoded = cql.into_pair_vec().unwrap();
+
+        assert_eq!(CQLValue::Int(20), decoded[0].0);
+        assert_eq!(CQLValue::Int(21), decoded[0].1);
+
+        assert_eq!(CQLValue::Int(2), decoded[1].0);
+        assert_eq!(CQLValue::Int(3), decoded[1].1);
+    }
+}
diff --git a/scylla/src/frame/value.rs b/scylla/src/frame/value.rs
index 9f467ca4..a40e54ad 100644
--- a/scylla/src/frame/value.rs
+++ b/scylla/src/frame/value.rs
@@ -1,5 +1,6 @@
 use bytes::BufMut;
 use std::borrow::Cow;
+use std::collections::HashMap;
 use std::convert::TryInto;
 use thiserror::Error;
 
@@ -224,6 +225,90 @@ impl<T: Value> Value for &T {
     }
 }
 
+impl<K: Value, V: Value> Value for HashMap<K, V> {
+    fn serialize(&self, buf: &mut Vec<u8>) -> Result<(), ValueTooBig> {
+        let bytes_num_pos: usize = buf.len();
+        buf.put_i32(0);
+
+        buf.put_i32(self.len().try_into().map_err(|_| ValueTooBig)?);
+        for (key, value) in self {
+            <K as Value>::serialize(key, buf)?;
+            <V as Value>::serialize(value, 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<T: Value> Value for Vec<T> {
+    fn serialize(&self, buf: &mut Vec<u8>) -> Result<(), ValueTooBig> {
+        let bytes_num_pos: usize = buf.len();
+        buf.put_i32(0);
+
+        buf.put_i32(self.len().try_into().map_err(|_| ValueTooBig)?);
+        for value in self {
+            <T as Value>::serialize(value, 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(())
+    }
+}
+
+macro_rules! impl_value_for_tuple {
+    ( $($Ti:ident),* ; $($FieldI:tt),* ) => {
+    impl<$($Ti),+> Value for ($($Ti,)+)
+        where
+            $($Ti: Value),+
+        {
+            fn serialize(&self, buf: &mut Vec<u8>) -> Result<(), ValueTooBig> {
+                let bytes_num_pos: usize = buf.len();
+                buf.put_i32(0);
+                $(
+                    <$Ti as Value>::serialize(&self.$FieldI, 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_value_for_tuple!(T0; 0);
+impl_value_for_tuple!(T0, T1; 0, 1);
+impl_value_for_tuple!(T0, T1, T2; 0, 1, 2);
+impl_value_for_tuple!(T0, T1, T2, T3; 0, 1, 2, 3);
+impl_value_for_tuple!(T0, T1, T2, T3, T4; 0, 1, 2, 3, 4);
+impl_value_for_tuple!(T0, T1, T2, T3, T4, T5; 0, 1, 2, 3, 4, 5);
+impl_value_for_tuple!(T0, T1, T2, T3, T4, T5, T6; 0, 1, 2, 3, 4, 5, 6);
+impl_value_for_tuple!(T0, T1, T2, T3, T4, T5, T6, T7; 0, 1, 2, 3, 4, 5, 6, 7);
+impl_value_for_tuple!(T0, T1, T2, T3, T4, T5, T6, T7, T8; 0, 1, 2, 3, 4, 5, 6, 7, 8);
+impl_value_for_tuple!(T0, T1, T2, T3, T4, T5, T6, T7, T8, T9; 
+                           0, 1, 2, 3, 4, 5, 6, 7, 8, 9);
+impl_value_for_tuple!(T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10; 
+                           0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
+impl_value_for_tuple!(T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11;
+                           0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11);
+impl_value_for_tuple!(T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12;
+                           0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12);
+impl_value_for_tuple!(T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13;
+                           0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13);
+impl_value_for_tuple!(T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14;
+                           0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14);
+impl_value_for_tuple!(T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15;
+                           0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15);
+
 //
 //  ValueList impls
 //
diff --git a/scylla/src/routing.rs b/scylla/src/routing.rs
index bc472ae4..50c7965f 100644
--- a/scylla/src/routing.rs
+++ b/scylla/src/routing.rs
@@ -65,7 +65,7 @@ impl ShardInfo {
     /// Randomly choose a source port `p` such that `shard == shard_of_source_port(p)`.
     pub fn draw_source_port_for_shard(&self, shard: Shard) -> u16 {
         assert!(shard < self.nr_shards as u32);
-        rand::thread_rng().gen_range(49152 + self.nr_shards - 1, 65535 - self.nr_shards + 1)
+        rand::thread_rng().gen_range((49152 + self.nr_shards - 1)..(65535 - self.nr_shards + 1))
             / self.nr_shards
             * self.nr_shards
             + shard as u16
diff --git a/scylla/src/transport/iterator.rs b/scylla/src/transport/iterator.rs
index affa3ce0..8814b720 100644
--- a/scylla/src/transport/iterator.rs
+++ b/scylla/src/transport/iterator.rs
@@ -38,7 +38,7 @@ impl Stream for RowIterator {
         let mut s = self.as_mut();
 
         if s.is_current_page_exhausted() {
-            match Pin::new(&mut s.page_receiver).poll_next(cx) {
+            match Pin::new(&mut s.page_receiver).poll_recv(cx) {
                 Poll::Ready(Some(Ok(rows))) => {
                     s.current_page = rows;
                     s.current_row_idx = 0;


Test patch:
diff --git a/scylla/src/frame/cql_collections_test.rs b/scylla/src/frame/cql_collections_test.rs
new file mode 100644
index 00000000..fa2ebb5b
--- /dev/null
+++ b/scylla/src/frame/cql_collections_test.rs
@@ -0,0 +1,192 @@
+// use anyhow::Result;
+use crate::transport::session::{IntoTypedRows, Session};
+use std::collections::HashMap;
+use std::env;
+
+// TODO: Requires a running local Scylla instance
+#[tokio::test]
+#[ignore]
+async fn test_cql_collections() {
+    // Create connection
+    let uri = env::var("SCYLLA_URI").unwrap_or_else(|_| "127.0.0.1:9042".to_string());
+
+    println!("Connecting to {} ...", uri);
+
+    let session = Session::connect(uri, None).await.unwrap();
+    session.refresh_topology().await.unwrap();
+
+    session.query("CREATE KEYSPACE IF NOT EXISTS ks WITH REPLICATION = {'class' : 'SimpleStrategy', 'replication_factor' : 1}", &[]).await.unwrap();
+
+    // HashMap example
+
+    // Create the table
+
+    session
+        .query(
+            "CREATE TABLE IF NOT EXISTS ks.teams (id int, team map<int, text>, primary key (id))",
+            &[],
+        )
+        .await
+        .unwrap();
+
+    let team1: HashMap<i32, String> = vec![(1, "Ginny".to_owned()), (2, "Cho".to_owned())]
+        .into_iter()
+        .collect();
+
+    let team2: HashMap<i32, String> = vec![(1, "John".to_owned()), (2, "Jessica".to_owned())]
+        .into_iter()
+        .collect();
+
+    session
+        .query("INSERT INTO ks.teams (id, team) VALUES (1, ?)", (team1,))
+        .await
+        .unwrap();
+
+    session
+        .query("INSERT INTO ks.teams (id, team) VALUES (2, ?)", (team2,))
+        .await
+        .unwrap();
+
+    // Recieve data in singular map:
+
+    if let Some(rows) = session
+        .query("SELECT team FROM ks.teams", &[])
+        .await
+        .unwrap()
+    {
+        for row in rows.into_typed::<(HashMap<i32, String>,)>() {
+            let teams: HashMap<i32, String> = row.unwrap().0;
+            for (key, value) in &teams {
+                println!("{}: {}", key, value);
+            }
+        }
+    }
+
+    // Set example
+
+    session
+        .query(
+            "CREATE TABLE IF NOT EXISTS ks.images (name text PRIMARY KEY, owner text, tags set<text>)",
+            &[],
+        )
+        .await.unwrap();
+
+    let tags1: Vec<String> = vec!["pet".to_owned(), "cute".to_owned()];
+    let tags2: Vec<String> = vec!["kitten".to_owned(), "cat".to_owned(), "lol".to_owned()];
+
+    session
+        .query(
+            "INSERT INTO ks.images (name, owner, tags) VALUES ('cat.jpg', 'jsmith', ?)",
+            (tags1,),
+        )
+        .await
+        .unwrap();
+
+    if let Some(rows) = session
+        .query("SELECT tags FROM ks.images", &[])
+        .await
+        .unwrap()
+    {
+        for row in rows.into_typed::<(Vec<String>,)>() {
+            let tag: Vec<String> = row.unwrap().0;
+            println!("{:?}", tag);
+        }
+    }
+
+    session
+        .query(
+            "UPDATE ks.images SET tags = ? WHERE name = 'cat.jpg';",
+            (tags2,),
+        )
+        .await
+        .unwrap();
+    if let Some(rows) = session
+        .query("SELECT tags FROM ks.images", &[])
+        .await
+        .unwrap()
+    {
+        for row in rows.into_typed::<(Vec<String>,)>() {
+            let tag: Vec<String> = row.unwrap().0;
+            println!("{:?}", tag);
+        }
+    }
+
+    // List example
+
+    session
+        .query(
+            "CREATE TABLE IF NOT EXISTS ks.plays ( id text PRIMARY KEY, game text, players int, scores list<int>)",
+            &[],
+        )
+        .await.unwrap();
+
+    let scores1: Vec<i32> = vec![17, 4, 2];
+    session
+        .query(
+            "INSERT INTO ks.plays (id, game, players, scores) VALUES ('123-afde', 'quake', 3, ?)",
+            (&scores1,),
+        )
+        .await
+        .unwrap();
+
+    if let Some(rows) = session
+        .query("SELECT scores FROM ks.plays", &[])
+        .await
+        .unwrap()
+    {
+        for row in rows.into_typed::<(Vec<i32>,)>() {
+            let scores: Vec<i32> = row.unwrap().0;
+            println!("{:?}", scores);
+        }
+    }
+
+    let scores2: Vec<i32> = vec![3, 9, 4];
+
+    session
+        .query(
+            "UPDATE ks.plays SET scores = ? WHERE id = '123-afde'",
+            (&scores2,),
+        )
+        .await
+        .unwrap();
+
+    if let Some(rows) = session
+        .query("SELECT scores FROM ks.plays", &[])
+        .await
+        .unwrap()
+    {
+        for row in rows.into_typed::<(Vec<i32>,)>() {
+            let scores: Vec<i32> = row.unwrap().0;
+            println!("{:?}", scores);
+        }
+    }
+
+    // Tuple example
+
+    session
+        .query(
+            "CREATE TABLE IF NOT EXISTS ks.durations (event text, duration tuple<int, text>, primary key (event))",
+            &[],
+        )
+        .await.unwrap();
+
+    let duration = (3, "hours");
+    session
+        .query(
+            "INSERT INTO ks.durations (event, duration) VALUES ('ev1', ?)",
+            (duration,),
+        )
+        .await
+        .unwrap();
+
+    if let Some(rows) = session
+        .query("SELECT duration FROM ks.durations", &[])
+        .await
+        .unwrap()
+    {
+        for row in rows.into_typed::<((i32, String),)>() {
+            let duration = row.unwrap().0;
+            println!("{:?}", duration);
+        }
+    }
+}