Repository: scylladb/scylla-rust-driver
Pull request #141 (2021-01-19 10:53:59)

Implement all CQL collection types. (Issue #97)
Issue: https://github.com/scylladb/scylla-rust-driver/issues/97

Implemented simple CQL Collection types (Map, Set, Tuples, List).
Implemented example of usage.

Converted HashMap to BTreeMap in existing UDT type so it could implement default Hash trait.

Co-authored-by: @Kejmer <kajetanhus@gmail.com>
Co-authored-by: @asledz <anita.gaja.sledz@gmail.com>

Kejmer: @psarna @piodul 
new clear history PR

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/select-paging.rs b/examples/select-paging.rs
index 12904a1d..497e2537 100644
--- a/examples/select-paging.rs
+++ b/examples/select-paging.rs
@@ -32,8 +32,7 @@ async fn main() -> Result<()> {
 
     // Iterate through select result with paging
     let mut rows_stream = session
-        .query_iter("SELECT a, b, c FROM ks.t", &[])
-        .await?
+        .query_iter("SELECT a, b, c FROM ks.t", &[])?
         .into_typed::<(i32, i32, String)>();
 
     while let Some(next_row_res) = rows_stream.next().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..9b261f23 100644
--- a/scylla/Cargo.toml
+++ b/scylla/Cargo.toml
@@ -20,7 +20,6 @@ tokio = { version = "0.3.0", features = ["net", "time", "io-util", "sync", "rt",
 snap = "1.0"
 uuid = "0.8.1"
 rand = "0.7.3"
-serde_json = "1.0.60"
 thiserror = "1.0"
 
 [dev-dependencies]
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..36f28c7c 100644
--- a/scylla/src/frame/response/cql_to_rust.rs
+++ b/scylla/src/frame/response/cql_to_rust.rs
@@ -1,8 +1,10 @@
 use super::result::{CQLValue, Row};
+use std::collections::HashMap;
+use std::hash::Hash;
 use std::net::IpAddr;
 use thiserror::Error;
 
-#[derive(Error, Debug, Clone, PartialEq, Eq)]
+#[derive(Error, Debug, PartialEq, Eq)]
 pub enum FromRowError {
     #[error("Bad CQL value")]
     BadCQLVal(#[from] FromCQLValError),
@@ -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
@@ -17,7 +25,7 @@ pub trait FromCQLVal<T>: Sized {
     fn from_cql(cql_val: T) -> Result<Self, FromCQLValError>;
 }
 
-#[derive(Error, Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, PartialEq, Eq, Error)]
 pub enum FromCQLValError {
     #[error("Bad CQL type")]
     BadCQLType,
@@ -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/error.rs b/scylla/src/frame/response/error.rs
index a5f0ea02..fba03277 100644
--- a/scylla/src/frame/response/error.rs
+++ b/scylla/src/frame/response/error.rs
@@ -1,6 +1,6 @@
 use crate::frame::frame_errors::ParseError;
 use crate::frame::types;
-use crate::transport::errors::{DBError, QueryError};
+use crate::transport::transport_errors::{DBError, TransportError};
 
 #[derive(Debug)]
 pub struct Error {
@@ -17,8 +17,8 @@ impl Error {
     }
 }
 
-impl Into<QueryError> for Error {
-    fn into(self) -> QueryError {
-        QueryError::DBError(DBError::ErrorMsg(self.code, self.reason))
+impl Into<TransportError> for Error {
+    fn into(self) -> TransportError {
+        TransportError::DBError(DBError::ErrorMsg(self.code, self.reason))
     }
 }
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..6d408109 100644
--- a/scylla/src/routing.rs
+++ b/scylla/src/routing.rs
@@ -1,9 +1,11 @@
 use bytes::{Buf, Bytes};
 use rand::Rng;
+use std::collections::BTreeMap;
 use std::collections::HashMap;
 use std::convert::TryFrom;
 use std::net::SocketAddr;
 use std::num::Wrapping;
+use std::ops::Bound::{Included, Unbounded};
 use thiserror::Error;
 
 #[derive(PartialEq, Eq, PartialOrd, Ord, Hash, Clone, Copy)]
@@ -18,13 +20,19 @@ pub struct Token {
     pub value: i64,
 }
 
+#[derive(Clone)]
+pub struct Ring {
+    // invariant: nonempty
+    pub owners: BTreeMap<Token, Node>,
+}
+
 pub type Shard = u32;
 
 #[derive(PartialEq, Eq, Clone, Debug)]
 pub struct ShardInfo {
-    pub shard: u16,
-    pub nr_shards: u16,
-    pub msb_ignore: u8,
+    shard: u16,
+    nr_shards: u16,
+    msb_ignore: u8,
 }
 
 impl std::str::FromStr for Token {
@@ -40,6 +48,15 @@ pub fn murmur3_token(pk: Bytes) -> Token {
     }
 }
 
+impl Ring {
+    pub fn owner(&self, t: Token) -> Node {
+        if let Some((_, &n)) = self.owners.range((Included(t), Unbounded)).next() {
+            return n;
+        }
+        *self.owners.iter().next().unwrap().1 // safe by invariant
+    }
+}
+
 impl ShardInfo {
     pub fn new(shard: u16, nr_shards: u16, msb_ignore: u8) -> Self {
         assert!(nr_shards > 0);
@@ -62,17 +79,14 @@ impl ShardInfo {
         (source_port % self.nr_shards) as Shard
     }
 
-    /// 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);
+    /// Randomly choose a source port `p` such that `shard_of(t) == shard_of_source_port(p)`.
+    pub fn draw_source_port_for_token(&self, t: Token) -> u16 {
+        let shard = self.shard_of(t) as u16;
+        assert!(shard < self.nr_shards);
         rand::thread_rng().gen_range(49152 + self.nr_shards - 1, 65535 - self.nr_shards + 1)
             / self.nr_shards
             * self.nr_shards
-            + shard as u16
-    }
-
-    pub fn get_nr_shards(&self) -> u16 {
-        self.nr_shards
+            + shard
     }
 }
 
diff --git a/scylla/src/transport/cluster.rs b/scylla/src/transport/cluster.rs
deleted file mode 100644
index 0b939782..00000000
--- a/scylla/src/transport/cluster.rs
+++ /dev/null
@@ -1,232 +0,0 @@
-/// Cluster manages up to date information and connections to database nodes
-use crate::routing::Token;
-use crate::transport::connection::Connection;
-use crate::transport::errors::QueryError;
-use crate::transport::node::{Node, NodeConnections};
-use crate::transport::topology::{Keyspace, TopologyReader};
-use crate::transport::Compression;
-
-use futures::{future::RemoteHandle, FutureExt};
-use std::collections::{BTreeMap, HashMap};
-use std::net::SocketAddr;
-use std::sync::{Arc, RwLock};
-
-/// Cluster manages up to date information and connections to database nodes.
-/// All data can be accessed by cloning Arc<ClusterData> in the `data` field
-pub struct Cluster {
-    data: Arc<RwLock<Arc<ClusterData>>>,
-
-    refresh_channel: tokio::sync::mpsc::Sender<RefreshRequest>,
-    _worker_handle: RemoteHandle<()>,
-}
-
-pub struct ClusterData {
-    pub known_peers: HashMap<SocketAddr, Arc<Node>>, // Invariant: nonempty after Cluster::new()
-    pub ring: BTreeMap<Token, Arc<Node>>,            // Invariant: nonempty after Cluster::new()
-    pub keyspaces: HashMap<String, Keyspace>,
-}
-
-// Works in the background to keep the cluster updated
-struct ClusterWorker {
-    // Cluster data to keep updated:
-    cluster_data: Arc<RwLock<Arc<ClusterData>>>,
-
-    // Cluster connections
-    topology_reader: TopologyReader,
-    compression: Option<Compression>,
-
-    // To listen for refresh requests
-    refresh_channel: tokio::sync::mpsc::Receiver<RefreshRequest>,
-}
-
-#[derive(Debug)]
-struct RefreshRequest {
-    response_chan: tokio::sync::oneshot::Sender<Result<(), QueryError>>,
-}
-
-impl Cluster {
-    pub async fn new(
-        initial_peers: &[SocketAddr],
-        compression: Option<Compression>,
-    ) -> Result<Cluster, QueryError> {
-        let cluster_data = Arc::new(RwLock::new(Arc::new(ClusterData {
-            known_peers: HashMap::new(),
-            ring: BTreeMap::new(),
-            keyspaces: HashMap::new(),
-        })));
-
-        let (refresh_sender, refresh_receiver) = tokio::sync::mpsc::channel(32);
-
-        let worker = ClusterWorker {
-            cluster_data: cluster_data.clone(),
-
-            topology_reader: TopologyReader::new(initial_peers, compression),
-            compression,
-
-            refresh_channel: refresh_receiver,
-        };
-
-        let (fut, worker_handle) = worker.work().remote_handle();
-        tokio::spawn(fut);
-
-        let result = Cluster {
-            data: cluster_data,
-            refresh_channel: refresh_sender,
-            _worker_handle: worker_handle,
-        };
-
-        result.refresh_topology().await?;
-
-        Ok(result)
-    }
-
-    pub fn get_data(&self) -> Arc<ClusterData> {
-        self.data.read().unwrap().clone()
-    }
-
-    pub async fn refresh_topology(&self) -> Result<(), QueryError> {
-        let (response_sender, response_receiver) = tokio::sync::oneshot::channel();
-
-        self.refresh_channel
-            .send(RefreshRequest {
-                response_chan: response_sender,
-            })
-            .await
-            .expect("Bug in Cluster::refresh_topology sending");
-        // Other end of this channel is in ClusterWorker, can't be dropped while we have &self to Cluster with _worker_handle
-
-        response_receiver
-            .await
-            .expect("Bug in Cluster::refresh_topology receiving")
-        // ClusterWorker always responds
-    }
-
-    /// Returns nonempty list of working connections to all shards
-    pub async fn get_working_connections(&self) -> Result<Vec<Arc<Connection>>, QueryError> {
-        let cluster_data: Arc<ClusterData> = self.get_data();
-        let peers = &cluster_data.known_peers;
-
-        let mut result: Vec<Arc<Connection>> = Vec::with_capacity(peers.len());
-
-        let mut last_error: Option<QueryError> = None;
-
-        // Takes result of ConnectionKeeper::get_connection() and pushes it onto result list or sets last_error
-        let mut push_to_result = |get_conn_res: Result<Arc<Connection>, QueryError>| {
-            match get_conn_res {
-                Ok(conn) => result.push(conn),
-                Err(e) => last_error = Some(e),
-            };
-        };
-
-        for node in peers.values() {
-            match &*node.connections.read().await {
-                NodeConnections::Single(conn_keeper) => {
-                    push_to_result(conn_keeper.get_connection().await)
-                }
-                NodeConnections::Sharded { shard_conns, .. } => {
-                    for conn_keeper in shard_conns {
-                        push_to_result(conn_keeper.get_connection().await);
-                    }
-                }
-            }
-        }
-
-        if result.is_empty() {
-            return Err(last_error.unwrap()); // By invariant peers is nonempty
-        }
-
-        Ok(result)
-    }
-}
-
-impl ClusterWorker {
-    pub async fn work(mut self) {
-        use tokio::time::{Duration, Instant};
-
-        let refresh_duration = Duration::from_secs(60); // Refresh topology every 60 seconds
-        let mut last_refresh_time = Instant::now();
-
-        loop {
-            let mut cur_request: Option<RefreshRequest> = None;
-
-            // Wait until it's time for the next refresh
-            let sleep_until: Instant = last_refresh_time
-                .checked_add(refresh_duration)
-                .unwrap_or_else(Instant::now);
-
-            let sleep_future = tokio::time::sleep_until(sleep_until);
-            tokio::pin!(sleep_future);
-
-            tokio::select! {
-                _ = sleep_future => {},
-                recv_res = self.refresh_channel.recv() => {
-                    match recv_res {
-                        Some(request) => cur_request = Some(request),
-                        None => return, // If refresh_channel was closed then cluster was dropped, we can stop working
-                    }
-                }
-            }
-
-            // Perform the refresh
-            last_refresh_time = Instant::now();
-            let refresh_res = self.perform_refresh().await;
-
-            // Send refresh result if there was a request
-            if let Some(request) = cur_request {
-                // We can ignore sending error - if no one waits for the response we can drop it
-                let _ = request.response_chan.send(refresh_res);
-            }
-        }
-    }
-
-    async fn perform_refresh(&mut self) -> Result<(), QueryError> {
-        // Read latest TopologyInfo
-        let topo_info = self.topology_reader.read_topology_info().await?;
-
-        // Create new updated known_peers and ring
-        let mut new_known_peers: HashMap<SocketAddr, Arc<Node>> =
-            HashMap::with_capacity(topo_info.peers.len());
-        let mut new_ring: BTreeMap<Token, Arc<Node>> = BTreeMap::new();
-
-        let cluster_data: Arc<ClusterData> = self.cluster_data.read().unwrap().clone();
-
-        for peer in topo_info.peers {
-            // Take existing Arc<Node> if possible, otherwise create new one
-            // Changing rack/datacenter but not ip address seems improbable
-            // so we can just create new node and connections then
-            let node: Arc<Node> = match cluster_data.known_peers.get(&peer.address) {
-                Some(node) if node.datacenter == peer.datacenter && node.rack == peer.rack => {
-                    node.clone()
-                }
-                _ => Arc::new(Node::new(
-                    peer.address,
-                    self.compression,
-                    peer.datacenter,
-                    peer.rack,
-                )),
-            };
-
-            new_known_peers.insert(peer.address, node.clone());
-
-            for token in peer.tokens {
-                new_ring.insert(token, node.clone());
-            }
-        }
-
-        let mut new_cluster_data: Arc<ClusterData> = Arc::new(ClusterData {
-            known_peers: new_known_peers,
-            ring: new_ring,
-            keyspaces: topo_info.keyspaces,
-        });
-
-        // Update current cluster_data
-        // Use std::mem::swap to avoid dropping large structures with active write lock
-        let mut cluster_data_lock = self.cluster_data.write().unwrap();
-
-        std::mem::swap(&mut *cluster_data_lock, &mut new_cluster_data);
-
-        drop(cluster_data_lock);
-
-        Ok(())
-    }
-}
diff --git a/scylla/src/transport/connection.rs b/scylla/src/transport/connection.rs
index acfa059a..57d2816f 100644
--- a/scylla/src/transport/connection.rs
+++ b/scylla/src/transport/connection.rs
@@ -6,12 +6,10 @@ use tokio::sync::{mpsc, oneshot};
 use std::cmp::Ordering;
 use std::collections::HashMap;
 use std::convert::TryFrom;
-use std::io::ErrorKind;
 use std::net::{IpAddr, Ipv4Addr, SocketAddr};
-use std::sync::Arc;
 use std::sync::Mutex as StdMutex;
 
-use super::errors::{BadQuery, QueryError};
+use super::transport_errors::{InternalDriverError, PrepareError, TransportError};
 
 use crate::batch::{Batch, BatchStatement};
 use crate::frame::{
@@ -55,7 +53,7 @@ impl Connection {
         addr: SocketAddr,
         source_port: Option<u16>,
         compression: Option<Compression>,
-    ) -> Result<Self, std::io::Error> {
+    ) -> Result<Self, TransportError> {
         let stream = match source_port {
             Some(p) => connect_with_source_port(addr, p).await?,
             None => TcpStream::connect(addr).await?,
@@ -78,16 +76,19 @@ impl Connection {
         })
     }
 
-    pub async fn startup(&self, options: HashMap<String, String>) -> Result<Response, QueryError> {
+    pub async fn startup(
+        &self,
+        options: HashMap<String, String>,
+    ) -> Result<Response, TransportError> {
         self.send_request(&request::Startup { options }, false)
             .await
     }
 
-    pub async fn get_options(&self) -> Result<Response, QueryError> {
+    pub async fn get_options(&self) -> Result<Response, TransportError> {
         self.send_request(&request::Options {}, false).await
     }
 
-    pub async fn prepare(&self, query: &str) -> Result<PreparedStatement, QueryError> {
+    pub async fn prepare(&self, query: &str) -> Result<PreparedStatement, TransportError> {
         let result = self.send_request(&request::Prepare { query }, true).await?;
         match result {
             Response::Error(err) => Err(err.into()),
@@ -96,8 +97,8 @@ impl Connection {
                 p.prepared_metadata,
                 query.to_owned(),
             )),
-            _ => Err(QueryError::ProtocolError(
-                "PREPARE: Unexpected server response",
+            _ => Err(TransportError::PrepareError(
+                PrepareError::InternalDriverError(InternalDriverError::UnexpectedResponse),
             )),
         }
     }
@@ -106,14 +107,14 @@ impl Connection {
         &self,
         query: impl Into<Query>,
         values: impl ValueList,
-    ) -> Result<Option<Vec<result::Row>>, QueryError> {
+    ) -> Result<Option<Vec<result::Row>>, TransportError> {
         let result = self.query(&query.into(), values, None).await?;
         match result {
             Response::Error(err) => Err(err.into()),
             Response::Result(result::Result::Rows(rs)) => Ok(Some(rs.rows)),
             Response::Result(_) => Ok(None),
-            _ => Err(QueryError::ProtocolError(
-                "QUERY: Unexpected server response",
+            _ => Err(TransportError::PrepareError(
+                PrepareError::InternalDriverError(InternalDriverError::UnexpectedResponse),
             )),
         }
     }
@@ -123,7 +124,7 @@ impl Connection {
         query: &Query,
         values: impl ValueList,
         paging_state: Option<Bytes>,
-    ) -> Result<Response, QueryError> {
+    ) -> Result<Response, TransportError> {
         let serialized_values = values.serialized()?;
 
         let query_frame = query::Query {
@@ -144,7 +145,7 @@ impl Connection {
         prepared_statement: &PreparedStatement,
         values: impl ValueList,
         paging_state: Option<Bytes>,
-    ) -> Result<Response, QueryError> {
+    ) -> Result<Response, TransportError> {
         let serialized_values = values.serialized()?;
 
         let execute_frame = execute::Execute {
@@ -164,13 +165,13 @@ impl Connection {
         &self,
         batch: &Batch,
         values: impl BatchValues,
-    ) -> Result<Response, QueryError> {
+    ) -> Result<Response, TransportError> {
         let statements_count = batch.get_statements().len();
         if statements_count != values.len() {
-            return Err(QueryError::BadQuery(BadQuery::ValueLenMismatch(
+            return Err(TransportError::ValueLenMismatch(
                 values.len(),
                 statements_count,
-            )));
+            ));
         }
 
         let statements_iter = batch.get_statements().iter().map(|s| match s {
@@ -198,7 +199,7 @@ impl Connection {
         &self,
         request: &R,
         compress: bool,
-    ) -> Result<Response, QueryError> {
+    ) -> Result<Response, TransportError> {
         let body = request.to_bytes()?;
         let compression = if compress { self.compression } else { None };
         let body_with_ext = RequestBodyWithExtensions { body };
@@ -216,19 +217,9 @@ impl Connection {
                 response_handler: sender,
             })
             .await
-            .map_err(|_| {
-                QueryError::IOError(Arc::new(std::io::Error::new(
-                    ErrorKind::Other,
-                    "Connection broken",
-                )))
-            })?;
-
-        let task_response = receiver.await.map_err(|_| {
-            QueryError::IOError(Arc::new(std::io::Error::new(
-                ErrorKind::Other,
-                "Connection broken",
-            )))
-        })?;
+            .map_err(|_| TransportError::DroppedRequest)?;
+
+        let task_response = receiver.await?;
         let body_with_ext = frame::parse_response_body_extensions(
             task_response.params.flags,
             self.compression,
@@ -273,7 +264,7 @@ impl Connection {
     async fn reader(
         mut read_half: tcp::ReadHalf<'_>,
         handler_map: &StdMutex<ResponseHandlerMap>,
-    ) -> Result<(), QueryError> {
+    ) -> Result<(), TransportError> {
         loop {
             let (params, opcode, body) = frame::read_response_frame(&mut read_half).await?;
 
@@ -310,9 +301,7 @@ impl Connection {
             } else {
                 // Unsolicited frame. This should not happen and indicates
                 // a bug either in the driver, or in the database
-                return Err(QueryError::ProtocolError(
-                    "Received reponse with unexpected StreamId",
-                ));
+                panic!(format!("Unsolicited frame on stream {}", params.stream));
             }
         }
     }
@@ -321,7 +310,7 @@ impl Connection {
         mut write_half: tcp::WriteHalf<'_>,
         handler_map: &StdMutex<ResponseHandlerMap>,
         mut task_receiver: mpsc::Receiver<Task>,
-    ) -> Result<(), QueryError> {
+    ) -> Result<(), TransportError> {
         // When the Connection object is dropped, the sender half
         // of the channel will be dropped, this task will return an error
         // and the whole worker will be stopped
@@ -355,7 +344,7 @@ impl Connection {
             .await?;
         }
 
-        Err(std::io::Error::new(ErrorKind::Other, "connection broken").into())
+        Err(TransportError::TaskQueueClosed)
     }
 
     pub fn get_shard_info(&self) -> &Option<ShardInfo> {
@@ -389,7 +378,7 @@ pub async fn open_connection(
     addr: SocketAddr,
     source_port: Option<u16>,
     compression: Option<Compression>,
-) -> Result<Connection, QueryError> {
+) -> Result<Connection, TransportError> {
     open_named_connection(
         addr,
         source_port,
@@ -404,7 +393,7 @@ pub async fn open_named_connection(
     source_port: Option<u16>,
     compression: Option<Compression>,
     driver_name: Option<String>,
-) -> Result<Connection, QueryError> {
+) -> Result<Connection, TransportError> {
     // TODO: shouldn't all this logic be in Connection::new?
     let mut connection = Connection::new(addr, source_port, compression).await?;
 
@@ -452,8 +441,8 @@ pub async fn open_named_connection(
         Response::Ready => {}
         Response::Authenticate => unimplemented!("Authentication is not yet implemented"),
         _ => {
-            return Err(QueryError::ProtocolError(
-                "Unexpected response to STARTUP message",
+            return Err(TransportError::InternalDriverError(
+                InternalDriverError::UnexpectedResponse,
             ))
         }
     }
@@ -461,10 +450,12 @@ pub async fn open_named_connection(
     Ok(connection)
 }
 
+// TODO: if we use the same source port twice in a row, we'll get an "address already in use" error.
+// We should try with multiple source ports in a loop.
 async fn connect_with_source_port(
     addr: SocketAddr,
     source_port: u16,
-) -> Result<TcpStream, std::io::Error> {
+) -> Result<TcpStream, TransportError> {
     // TODO: handle ipv6?
     let socket = TcpSocket::new_v4()?;
     let source_addr = IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0));
diff --git a/scylla/src/transport/connection_keeper.rs b/scylla/src/transport/connection_keeper.rs
deleted file mode 100644
index 449e62b0..00000000
--- a/scylla/src/transport/connection_keeper.rs
+++ /dev/null
@@ -1,148 +0,0 @@
-/// ConnectionKeeper keeps a Connection to some address and works to keep it open
-use crate::routing::ShardInfo;
-use crate::transport::errors::QueryError;
-use crate::transport::Compression;
-use crate::transport::{connection, connection::Connection};
-
-use futures::{future::RemoteHandle, FutureExt};
-use std::net::SocketAddr;
-use std::sync::Arc;
-
-/// ConnectionKeeper keeps a Connection to some address and works to keep it open
-pub struct ConnectionKeeper {
-    conn_state_receiver: tokio::sync::watch::Receiver<ConnectionState>,
-    _worker_handle: RemoteHandle<()>,
-}
-
-#[derive(Clone)]
-pub enum ConnectionState {
-    Initializing, // First connect attempt ongoing
-    Connected(Arc<Connection>),
-    Broken(QueryError),
-}
-
-/// Works in the background to keep the connection open
-struct ConnectionKeeperWorker {
-    address: SocketAddr,
-    compression: Option<Compression>,
-    shard_info: Option<ShardInfo>,
-
-    shard_info_sender: Option<ShardInfoSender>,
-    conn_state_sender: tokio::sync::watch::Sender<ConnectionState>,
-}
-
-pub type ShardInfoSender = Arc<std::sync::Mutex<tokio::sync::watch::Sender<Option<ShardInfo>>>>;
-
-impl ConnectionKeeper {
-    /// Creates new ConnectionKeeper that starts a connection in the background
-    /// # Arguments
-    ///
-    /// * `address` - IP address to connect to
-    /// * `compression` - preferred compression method to use
-    /// * `shard_info` - ShardInfo to use, will connect to shard number `shard_info.shard`
-    /// * `shard_info_sender` - channel to send new ShardInfo after each connection creation
-    pub fn new(
-        address: SocketAddr,
-        compression: Option<Compression>,
-        shard_info: Option<ShardInfo>,
-        shard_info_sender: Option<ShardInfoSender>,
-    ) -> Self {
-        let (conn_state_sender, conn_state_receiver) =
-            tokio::sync::watch::channel(ConnectionState::Initializing);
-
-        let worker = ConnectionKeeperWorker {
-            address,
-            compression,
-            shard_info,
-            shard_info_sender,
-            conn_state_sender,
-        };
-
-        let (fut, worker_handle) = worker.work().remote_handle();
-        tokio::spawn(fut);
-
-        ConnectionKeeper {
-            conn_state_receiver,
-            _worker_handle: worker_handle,
-        }
-    }
-
-    /// Get current connection state, returns immediately
-    pub fn connection_state(&self) -> ConnectionState {
-        self.conn_state_receiver.borrow().clone()
-    }
-
-    pub async fn wait_until_initialized(&self) {
-        match &*self.conn_state_receiver.borrow() {
-            ConnectionState::Initializing => {}
-            _ => return,
-        };
-
-        let mut my_receiver = self.conn_state_receiver.clone();
-
-        my_receiver
-            .changed()
-            .await
-            .expect("Bug in ConnectionKeeper::wait_until_initialized");
-        // Worker can't stop while we have &self to struct with worker_handle
-
-        // Now state must be != Initializing
-        debug_assert!(!matches!(
-            &*self.conn_state_receiver.borrow(),
-            ConnectionState::Initializing
-        ));
-    }
-
-    /// Wait for the connection to initialize and get it if succesfylly connected
-    pub async fn get_connection(&self) -> Result<Arc<Connection>, QueryError> {
-        self.wait_until_initialized().await;
-
-        match self.connection_state() {
-            ConnectionState::Connected(conn) => Ok(conn),
-            ConnectionState::Broken(e) => Err(e),
-            _ => unreachable!(),
-        }
-    }
-}
-
-impl ConnectionKeeperWorker {
-    pub async fn work(self) {
-        let cur_connection = self.open_new_connection().await;
-
-        match &cur_connection {
-            Ok(conn) => {
-                let _ = self
-                    .conn_state_sender
-                    .send(ConnectionState::Connected(conn.clone()));
-
-                let new_shard_info: Option<ShardInfo> = conn.get_shard_info().clone();
-
-                if let Some(sender) = &self.shard_info_sender {
-                    // Ignore sending error
-                    // If no one wants to get shard_info that's OK
-                    // If lock is poisoned do nothing
-                    if let Ok(sender_locked) = sender.lock() {
-                        let _ = sender_locked.send(new_shard_info);
-                    }
-                }
-            }
-            Err(e) => {
-                let _ = self
-                    .conn_state_sender
-                    .send(ConnectionState::Broken(e.clone()));
-            } // TODO: Wait for connection to fail, then create new, loop it
-        };
-    }
-
-    async fn open_new_connection(&self) -> Result<Arc<Connection>, QueryError> {
-        let mut source_port: Option<u16> = None;
-        if let Some(info) = &self.shard_info {
-            source_port = Some(info.draw_source_port_for_shard(info.shard.into()));
-        }
-
-        let new_conn =
-            connection::open_connection(self.address, source_port, self.compression).await?;
-
-        Ok(Arc::new(new_conn))
-    }
-}
diff --git a/scylla/src/transport/errors.rs b/scylla/src/transport/errors.rs
deleted file mode 100644
index 0fd08300..00000000
--- a/scylla/src/transport/errors.rs
+++ /dev/null
@@ -1,117 +0,0 @@
-use crate::frame::frame_errors::{FrameError, ParseError};
-use crate::frame::value::SerializeValuesError;
-use std::sync::Arc;
-use thiserror::Error;
-
-/// Error that occured during query execution
-#[derive(Error, Debug, Clone)]
-pub enum QueryError {
-    /// Database sent a response containing some error
-    #[error(transparent)]
-    DBError(#[from] DBError),
-
-    /// Caller passed an invalid query
-    #[error(transparent)]
-    BadQuery(#[from] BadQuery),
-
-    /// Input/Output error has occured, connection broken etc.
-    #[error("IO Error: {0}")]
-    IOError(Arc<std::io::Error>),
-
-    /// Unexpected or invalid message received
-    #[error("Protocol Error: {0}")]
-    ProtocolError(&'static str),
-}
-
-/// An error sent from database in response to a query
-#[derive(Error, Debug, Clone)]
-#[error("Database response contains an error")]
-pub enum DBError {
-    // TODO develop this while implementing retry policies
-    // to react differently to each error.
-    // Make it fit what we encounter.
-    #[error("Response is an error message: code: {0} message: {1}")]
-    ErrorMsg(i32, String),
-}
-
-/// Error caused by caller creating an invalid query
-#[derive(Error, Debug, Clone)]
-#[error("Invalid query passed to Session")]
-pub enum BadQuery {
-    /// Failed to serialize values passed to a query - values too big
-    #[error("Serializing values failed: {0} ")]
-    SerializeValuesError(#[from] SerializeValuesError),
-
-    /// Number of values provided doesn't match number of statements in a batch
-    #[error("Length of provided values ({0}) must be equal to number of batch statements ({1})")]
-    ValueLenMismatch(usize, usize),
-
-    /// Serialized values are too long to compute parition key
-    #[error("Serialized values are too long to compute parition key! Length: {0}, Max allowed length: {1}")]
-    ValuesTooLongForKey(usize, usize),
-}
-
-/// Error that occured during session creation
-#[derive(Error, Debug, Clone)]
-pub enum NewSessionError {
-    /// Failed to resolve hostname passed in Session creation
-    #[error("Couldn't resolve address: {0}")]
-    FailedToResolveAddress(String),
-
-    /// Database sent a response containing some error
-    #[error(transparent)]
-    DBError(#[from] DBError),
-
-    /// Caller passed an invalid query
-    #[error(transparent)]
-    BadQuery(#[from] BadQuery),
-
-    /// Input/Output error has occured, connection broken etc.
-    #[error("IO Error: {0}")]
-    IOError(Arc<std::io::Error>),
-
-    /// Unexpected or invalid message received
-    #[error("Protocol Error: {0}")]
-    ProtocolError(&'static str),
-}
-
-impl From<std::io::Error> for QueryError {
-    fn from(io_error: std::io::Error) -> QueryError {
-        QueryError::IOError(Arc::new(io_error))
-    }
-}
-
-impl From<SerializeValuesError> for QueryError {
-    fn from(serialized_err: SerializeValuesError) -> QueryError {
-        QueryError::BadQuery(BadQuery::SerializeValuesError(serialized_err))
-    }
-}
-
-impl From<ParseError> for QueryError {
-    fn from(_parse_error: ParseError) -> QueryError {
-        QueryError::ProtocolError("Error parsing message")
-    }
-}
-
-impl From<FrameError> for QueryError {
-    fn from(_frame_error: FrameError) -> QueryError {
-        QueryError::ProtocolError("Error parsing message frame")
-    }
-}
-
-impl From<std::io::Error> for NewSessionError {
-    fn from(io_error: std::io::Error) -> NewSessionError {
-        NewSessionError::IOError(Arc::new(io_error))
-    }
-}
-
-impl From<QueryError> for NewSessionError {
-    fn from(query_error: QueryError) -> NewSessionError {
-        match query_error {
-            QueryError::DBError(e) => NewSessionError::DBError(e),
-            QueryError::BadQuery(e) => NewSessionError::BadQuery(e),
-            QueryError::IOError(e) => NewSessionError::IOError(e),
-            QueryError::ProtocolError(m) => NewSessionError::ProtocolError(m),
-        }
-    }
-}
diff --git a/scylla/src/transport/iterator.rs b/scylla/src/transport/iterator.rs
index affa3ce0..1e70c8a5 100644
--- a/scylla/src/transport/iterator.rs
+++ b/scylla/src/transport/iterator.rs
@@ -8,11 +8,10 @@ use std::time::Instant;
 use bytes::Bytes;
 use futures::Stream;
 use std::result::Result as StdResult;
-use thiserror::Error;
 use tokio::sync::mpsc;
 
-use super::errors::QueryError;
-use crate::cql_to_rust::{FromRow, FromRowError};
+use super::transport_errors::{InternalDriverError, TransportError};
+use crate::cql_to_rust::FromRow;
 
 use crate::frame::{
     response::{
@@ -28,11 +27,11 @@ use crate::transport::metrics::Metrics;
 pub struct RowIterator {
     current_row_idx: usize,
     current_page: Rows,
-    page_receiver: mpsc::Receiver<StdResult<Rows, QueryError>>,
+    page_receiver: mpsc::Receiver<StdResult<Rows, TransportError>>,
 }
 
 impl Stream for RowIterator {
-    type Item = StdResult<Row, QueryError>;
+    type Item = StdResult<Row, TransportError>;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         let mut s = self.as_mut();
@@ -146,18 +145,18 @@ impl RowIterator {
     }
 }
 struct WorkerHelper {
-    sender: mpsc::Sender<StdResult<Rows, QueryError>>,
+    sender: mpsc::Sender<StdResult<Rows, TransportError>>,
     metrics: Arc<Metrics>,
 }
 
 impl WorkerHelper {
-    fn new(sender: mpsc::Sender<StdResult<Rows, QueryError>>, metrics: Arc<Metrics>) -> Self {
+    fn new(sender: mpsc::Sender<StdResult<Rows, TransportError>>, metrics: Arc<Metrics>) -> Self {
         Self { sender, metrics }
     }
 
     async fn handle_response(
         &mut self,
-        response: StdResult<Response, QueryError>,
+        response: StdResult<Response, TransportError>,
     ) -> Option<Bytes> {
         match response {
             Ok(Response::Result(Result::Rows(rows))) => {
@@ -178,8 +177,8 @@ impl WorkerHelper {
                 self.metrics.inc_failed_paged_queries();
                 let _ = self
                     .sender
-                    .send(Err(QueryError::ProtocolError(
-                        "Unexpected response to next page query",
+                    .send(Err(TransportError::InternalDriverError(
+                        InternalDriverError::UnexpectedResponse,
                     )))
                     .await;
                 None
@@ -198,33 +197,21 @@ pub struct TypedRowIterator<RowT> {
     phantom_data: std::marker::PhantomData<RowT>,
 }
 
-/// Couldn't get next typed row from the iterator
-#[derive(Error, Debug, Clone)]
-pub enum NextRowError {
-    /// Query to fetch next page has failed
-    #[error(transparent)]
-    QueryError(#[from] QueryError),
-
-    /// Parsing values in row as given types failed
-    #[error(transparent)]
-    FromRowError(#[from] FromRowError),
-}
-
 impl<RowT: FromRow> Stream for TypedRowIterator<RowT> {
-    type Item = StdResult<RowT, NextRowError>;
+    type Item = StdResult<RowT, TransportError>;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         let mut s = self.as_mut();
 
-        let next_elem: Option<StdResult<Row, QueryError>> =
+        let next_elem: Option<StdResult<Row, TransportError>> =
             match Pin::new(&mut s.row_iterator).poll_next(cx) {
                 Poll::Ready(next_elem) => next_elem,
                 Poll::Pending => return Poll::Pending,
             };
 
         let next_ready: Option<Self::Item> = match next_elem {
-            Some(Ok(next_row)) => Some(RowT::from_row(next_row).map_err(|e| e.into())),
-            Some(Err(e)) => Some(Err(e.into())),
+            Some(Ok(next_row)) => Some(RowT::from_row(next_row).map_err(TransportError::from)),
+            Some(Err(e)) => Some(Err(e)),
             None => None,
         };
 
diff --git a/scylla/src/transport/mod.rs b/scylla/src/transport/mod.rs
index c4cedf81..981815f8 100644
--- a/scylla/src/transport/mod.rs
+++ b/scylla/src/transport/mod.rs
@@ -1,16 +1,9 @@
-mod cluster;
 pub mod connection;
-mod connection_keeper;
-mod node;
-pub mod session;
-mod topology;
-
-pub mod errors;
 pub mod iterator;
 mod metrics;
-
-#[cfg(test)]
-mod session_test;
+pub mod session;
+mod topology;
+pub mod transport_errors;
 
 /// The wire protocol compression algorithm.
 #[derive(Copy, Clone)]
@@ -29,3 +22,6 @@ impl ToString for Compression {
         }
     }
 }
+
+#[cfg(test)]
+mod session_test;
diff --git a/scylla/src/transport/node.rs b/scylla/src/transport/node.rs
deleted file mode 100644
index 0647ef63..00000000
--- a/scylla/src/transport/node.rs
+++ /dev/null
@@ -1,172 +0,0 @@
-/// Node represents a cluster node along with it's data and connections
-use crate::routing::{Shard, ShardInfo, Token};
-use crate::transport::connection::Connection;
-use crate::transport::connection_keeper::{ConnectionKeeper, ShardInfoSender};
-use crate::transport::errors::QueryError;
-use crate::transport::Compression;
-
-use futures::{future::RemoteHandle, FutureExt};
-use std::net::SocketAddr;
-use std::sync::Arc;
-use tokio::sync::RwLock as AsyncRwLock;
-
-/// Node represents a cluster node along with it's data and connections
-pub struct Node {
-    pub address: SocketAddr,
-    pub datacenter: Option<String>,
-    pub rack: Option<String>,
-
-    pub connections: Arc<AsyncRwLock<NodeConnections>>,
-
-    _worker_handle: RemoteHandle<()>,
-}
-
-pub enum NodeConnections {
-    /// Non shard-aware ex. a Cassandra node connection
-    Single(ConnectionKeeper),
-    /// Shard aware Scylla node connections
-    Sharded {
-        shard_info: ShardInfo,
-        /// shard_conns always contains shard_info.nr_shards ConnectionKeepers
-        shard_conns: Vec<ConnectionKeeper>,
-    },
-}
-
-// Works in the background to detect ShardInfo changes and keep node connections updated
-struct NodeWorker {
-    node_conns: Arc<AsyncRwLock<NodeConnections>>,
-    node_addr: SocketAddr,
-    compression: Option<Compression>,
-
-    shard_info_sender: ShardInfoSender,
-    shard_info_receiver: tokio::sync::watch::Receiver<Option<ShardInfo>>,
-}
-
-impl Node {
-    /// Creates new node which starts connecting in the background
-    /// # Arguments
-    ///
-    /// `address` - address to connect to
-    /// `compression` - preferred compression to use
-    /// `datacenter` - optional datacenter name
-    /// `rack` - optional rack name
-    pub fn new(
-        address: SocketAddr,
-        compression: Option<Compression>,
-        datacenter: Option<String>,
-        rack: Option<String>,
-    ) -> Self {
-        let (shard_info_sender, shard_info_receiver) = tokio::sync::watch::channel(None);
-
-        let shard_info_sender = Arc::new(std::sync::Mutex::new(shard_info_sender));
-
-        let connections = Arc::new(AsyncRwLock::new(NodeConnections::Single(
-            ConnectionKeeper::new(address, compression, None, Some(shard_info_sender.clone())),
-        )));
-
-        let worker = NodeWorker {
-            node_conns: connections.clone(),
-            node_addr: address,
-            compression,
-            shard_info_sender,
-            shard_info_receiver,
-        };
-
-        let (fut, worker_handle) = worker.work().remote_handle();
-        tokio::spawn(fut);
-
-        Node {
-            address,
-            datacenter,
-            rack,
-            connections,
-            _worker_handle: worker_handle,
-        }
-    }
-
-    /// Get connection which should be used to connect using given token
-    pub async fn connection_for_token(&self, token: Token) -> Result<Arc<Connection>, QueryError> {
-        let connections = &*self.connections.read().await;
-
-        match connections {
-            NodeConnections::Single(conn_keeper) => conn_keeper.get_connection().await,
-            NodeConnections::Sharded {
-                shard_info,
-                shard_conns,
-            } => {
-                let shard: Shard = shard_info.shard_of(token);
-                // It's guaranteed by NodeWorker that shard_conns.len() == shard_info.nr_shards
-                shard_conns[shard as usize].get_connection().await
-            }
-        }
-    }
-}
-
-impl NodeWorker {
-    pub async fn work(mut self) {
-        let mut cur_shard_info: Option<ShardInfo> = self.shard_info_receiver.borrow().clone();
-
-        loop {
-            // Wait for current shard_info to change
-            // We own one sending end of this channel so it can't return None
-            self.shard_info_receiver
-                .changed()
-                .await
-                .expect("Bug in NodeWorker::work");
-            let new_shard_info: Option<ShardInfo> = self.shard_info_receiver.borrow().clone();
-
-            // See if the node has resharded
-            match (&cur_shard_info, &new_shard_info) {
-                (Some(cur), Some(new)) => {
-                    if cur.nr_shards == new.nr_shards && cur.msb_ignore == new.msb_ignore {
-                        // Nothing chaged, go back to waiting for a change
-                        continue;
-                    }
-                }
-                (None, None) => continue, // Nothing chaged, go back to waiting for a change
-                _ => {}
-            }
-
-            cur_shard_info = new_shard_info;
-
-            // We received updated node ShardInfo
-            // Create new node connections. It will happen rarely so we can probably afford it
-            // TODO: Maybe save some connections instead of recreating?
-            let mut new_connections: NodeConnections = match &cur_shard_info {
-                None => NodeConnections::Single(ConnectionKeeper::new(
-                    self.node_addr,
-                    self.compression,
-                    None,
-                    Some(self.shard_info_sender.clone()),
-                )),
-                Some(shard_info) => {
-                    let mut connections: Vec<ConnectionKeeper> =
-                        Vec::with_capacity(shard_info.nr_shards as usize);
-
-                    for shard in 0..shard_info.nr_shards {
-                        let mut cur_conn_shard_info = shard_info.clone();
-                        cur_conn_shard_info.shard = shard;
-                        let cur_conn = ConnectionKeeper::new(
-                            self.node_addr,
-                            self.compression,
-                            Some(cur_conn_shard_info),
-                            Some(self.shard_info_sender.clone()),
-                        );
-                        connections.push(cur_conn);
-                    }
-
-                    NodeConnections::Sharded {
-                        shard_info: shard_info.clone(),
-                        shard_conns: connections,
-                    }
-                }
-            };
-
-            // Update node.connections
-            // Use std::mem::swap to minimalize time spent holding write lock
-            let mut node_conns_lock = self.node_conns.write().await;
-            std::mem::swap(&mut *node_conns_lock, &mut new_connections);
-            drop(node_conns_lock);
-        }
-    }
-}
diff --git a/scylla/src/transport/session.rs b/scylla/src/transport/session.rs
index e9d9ced5..8c112ab9 100644
--- a/scylla/src/transport/session.rs
+++ b/scylla/src/transport/session.rs
@@ -1,13 +1,16 @@
-use core::ops::Bound::{Included, Unbounded};
-use futures::future::join_all;
-use rand::Rng;
+use futures::{
+    future::{try_join_all, RemoteHandle},
+    FutureExt,
+};
+use std::collections::hash_map::Entry;
+use std::collections::HashMap;
 use std::fmt::Display;
 use std::net::SocketAddr;
-use std::sync::Arc;
+use std::sync::{Arc, RwLock};
 use std::time::Instant;
 use tokio::net::{lookup_host, ToSocketAddrs};
 
-use super::errors::{BadQuery, NewSessionError, QueryError};
+use super::transport_errors::{InternalDriverError, TransportError};
 use crate::batch::Batch;
 use crate::cql_to_rust::FromRow;
 use crate::frame::response::cql_to_rust::FromRowError;
@@ -16,20 +19,44 @@ use crate::frame::response::Response;
 use crate::frame::value::{BatchValues, SerializedValues, ValueList};
 use crate::prepared_statement::{PartitionKeyError, PreparedStatement};
 use crate::query::Query;
-use crate::routing::{murmur3_token, Token};
-use crate::transport::cluster::{Cluster, ClusterData};
-use crate::transport::connection::Connection;
+use crate::routing::{murmur3_token, Node, Shard, ShardInfo, Token};
+use crate::transport::connection::{open_connection, Connection};
 use crate::transport::iterator::RowIterator;
 use crate::transport::metrics::{Metrics, MetricsView};
-use crate::transport::node::Node;
+use crate::transport::topology::{Topology, TopologyReader};
 use crate::transport::Compression;
 
+type Connections = Vec<(Node, Vec<Arc<Connection>>)>;
+
+const POOL_LOCK_POISONED: &str =
+    "Connection pool lock is poisoned, this session is no longer usable. \
+                                  Drop it and create a new one.";
+
 pub struct Session {
-    cluster: Cluster,
+    // invariant: nonempty
+    // PLEASE DO NOT EXPOSE IT TO THE USER OR PEOPLE WILL DIE
+    pool: Arc<RwLock<HashMap<Node, NodePool>>>,
+
+    // compression options passed to the session when it was created
+    compression: Option<Compression>,
+
+    topology: Topology,
+    _topology_reader_handle: RemoteHandle<()>,
 
     metrics: Arc<Metrics>,
 }
 
+// Pool of connections for a single node.
+// Also stores the sharding info for this node if we have one.
+// If we're connecting using a non-shard-aware port, or connecting to Cassandra,
+// the pool will always have a single connection under shard 0
+// and the sharding info will be empty.
+struct NodePool {
+    // invariant: nonempty
+    connections: HashMap<Shard, Arc<Connection>>,
+    shard_info: Option<ShardInfo>,
+}
+
 // Trait used to implement Vec<result::Row>::into_typed<RowT>(self)
 // This is the only way to add custom method to Vec
 pub trait IntoTypedRows {
@@ -65,6 +92,7 @@ impl<RowT: FromRow> Iterator for TypedRowIter<RowT> {
 /// Represents a CQL session, which can be used to communicate
 /// with the database
 impl Session {
+    // FIXME: Use ToSocketAddrs instead of SocketAddr in public interfaces
     // because it's more convenient
     /// Estabilishes a CQL session with the database
     /// # Arguments
@@ -74,14 +102,30 @@ impl Session {
     pub async fn connect(
         addr: impl ToSocketAddrs + Display,
         compression: Option<Compression>,
-    ) -> Result<Self, NewSessionError> {
+    ) -> Result<Self, TransportError> {
         let addr = resolve(addr).await?;
+        let connection = open_connection(addr, None, compression).await?;
+        let node = Node { addr };
+
+        let (topology_reader, topology) = TopologyReader::new(node).await?;
+        let (fut, _topology_reader_handle) = topology_reader.run().remote_handle();
+        tokio::task::spawn(fut);
 
-        let cluster = Cluster::new(&[addr], compression).await?;
+        let pool = Arc::new(RwLock::new(
+            vec![(node, NodePool::new(Arc::new(connection)))]
+                .into_iter()
+                .collect(),
+        ));
 
         let metrics = Arc::new(Metrics::new());
 
-        Ok(Session { cluster, metrics })
+        Ok(Session {
+            pool,
+            compression,
+            topology,
+            _topology_reader_handle,
+            metrics,
+        })
     }
 
     // TODO: Should return an iterator over results
@@ -98,7 +142,7 @@ impl Session {
         &self,
         query: impl Into<Query>,
         values: impl ValueList,
-    ) -> Result<Option<Vec<result::Row>>, QueryError> {
+    ) -> Result<Option<Vec<result::Row>>, TransportError> {
         let now = Instant::now();
         self.metrics.inc_total_nonpaged_queries();
         let result = self.query_no_metrics(query, values).await;
@@ -113,22 +157,21 @@ impl Session {
         &self,
         query: impl Into<Query>,
         values: impl ValueList,
-    ) -> Result<Option<Vec<result::Row>>, QueryError> {
-        self.any_connection()
-            .await?
+    ) -> Result<Option<Vec<result::Row>>, TransportError> {
+        self.any_connection()?
             .query_single_page(query, values)
             .await
     }
 
-    pub async fn query_iter(
+    pub fn query_iter(
         &self,
         query: impl Into<Query>,
         values: impl ValueList,
-    ) -> Result<RowIterator, QueryError> {
+    ) -> Result<RowIterator, TransportError> {
         let serialized_values = values.serialized()?;
 
         Ok(RowIterator::new_for_query(
-            self.any_connection().await?,
+            self.any_connection()?,
             query.into(),
             serialized_values.into_owned(),
             self.metrics.clone(),
@@ -139,42 +182,25 @@ impl Session {
     /// which can later be used to perform more efficient queries
     /// # Arguments
     ///#
-    pub async fn prepare(&self, query: &str) -> Result<PreparedStatement, QueryError> {
-        let connections = self.cluster.get_working_connections().await?;
+    pub async fn prepare(&self, query: &str) -> Result<PreparedStatement, TransportError> {
+        let connections = self.get_connections()?;
+        let all_connections = connections.iter().flat_map(|(_, v)| v);
 
         // Prepare statements on all connections concurrently
-        let handles = connections.iter().map(|c| c.prepare(query));
-        let mut results = join_all(handles).await;
+        let handles = all_connections.map(|c| c.prepare(query));
+        let mut results = try_join_all(handles).await?;
 
-        // If at least one prepare was succesfull prepare returns Ok
-
-        // Find first result that is Ok, or Err if all failed
-        let mut first_ok: Option<Result<PreparedStatement, QueryError>> = None;
-
-        while let Some(res) = results.pop() {
-            let is_ok: bool = res.is_ok();
-
-            first_ok = Some(res);
-
-            if is_ok {
-                break;
-            }
+        if !Self::validate_prepared_ids_equality(&results) {
+            return Err(TransportError::PreparedStatementsIDsNotMatch);
         }
 
-        let prepared: PreparedStatement = first_ok.unwrap()?;
-
-        // Validate prepared ids equality
-        for res in results {
-            if let Ok(statement) = res {
-                if prepared.get_id() != statement.get_id() {
-                    return Err(QueryError::ProtocolError(
-                        "Prepared statement Ids differ, all should be equal",
-                    ));
-                }
-            }
-        }
+        results.pop().ok_or(TransportError::NoConnectionsAvailable)
+    }
 
-        Ok(prepared)
+    fn validate_prepared_ids_equality(statements: &[PreparedStatement]) -> bool {
+        statements
+            .windows(2)
+            .all(|p| p[0].get_id() == p[1].get_id())
     }
 
     /// Executes a previously prepared statement
@@ -186,7 +212,7 @@ impl Session {
         &self,
         prepared: &PreparedStatement,
         values: impl ValueList,
-    ) -> Result<Option<Vec<result::Row>>, QueryError> {
+    ) -> Result<Option<Vec<result::Row>>, TransportError> {
         let now = Instant::now();
         self.metrics.inc_total_nonpaged_queries();
         let result = self.execute_no_metrics(prepared, values).await;
@@ -201,7 +227,7 @@ impl Session {
         &self,
         prepared: &PreparedStatement,
         values: impl ValueList,
-    ) -> Result<Option<Vec<result::Row>>, QueryError> {
+    ) -> Result<Option<Vec<result::Row>>, TransportError> {
         // FIXME: Prepared statement ids are local to a node, so we must make sure
         // that prepare() sends to all nodes and keeps all ids.
         let serialized_values = values.serialized()?;
@@ -220,9 +246,7 @@ impl Session {
                         // Reprepared statement should keep its id - it's the md5 sum
                         // of statement contents
                         if reprepared.get_id() != prepared.get_id() {
-                            return Err(QueryError::ProtocolError(
-                                "Prepared statement Id changed, md5 sum should stay the same",
-                            ));
+                            return Err(TransportError::RepreparedStatmentIDChanged);
                         }
 
                         let result = connection
@@ -232,8 +256,8 @@ impl Session {
                             Response::Error(err) => Err(err.into()),
                             Response::Result(result::Result::Rows(rs)) => Ok(Some(rs.rows)),
                             Response::Result(_) => Ok(None),
-                            _ => Err(QueryError::ProtocolError(
-                                "EXECUTE: Unexpected server response",
+                            _ => Err(TransportError::InternalDriverError(
+                                InternalDriverError::UnexpectedResponse,
                             )),
                         }
                     }
@@ -242,21 +266,21 @@ impl Session {
             }
             Response::Result(result::Result::Rows(rs)) => Ok(Some(rs.rows)),
             Response::Result(_) => Ok(None),
-            _ => Err(QueryError::ProtocolError(
-                "EXECUTE: Unexpected server response",
+            _ => Err(TransportError::InternalDriverError(
+                InternalDriverError::UnexpectedResponse,
             )),
         }
     }
 
-    pub async fn execute_iter(
+    pub fn execute_iter(
         &self,
         prepared: impl Into<PreparedStatement>,
         values: impl ValueList,
-    ) -> Result<RowIterator, QueryError> {
+    ) -> Result<RowIterator, TransportError> {
         let serialized_values = values.serialized()?;
 
         Ok(RowIterator::new_for_prepared_statement(
-            self.any_connection().await?,
+            self.any_connection()?,
             prepared.into(),
             serialized_values.into_owned(),
             self.metrics.clone(),
@@ -268,47 +292,158 @@ impl Session {
     ///
     /// * `batch` - batch to be performed
     /// * `values` - values bound to the query
-    pub async fn batch(&self, batch: &Batch, values: impl BatchValues) -> Result<(), QueryError> {
+    pub async fn batch(
+        &self,
+        batch: &Batch,
+        values: impl BatchValues,
+    ) -> Result<(), TransportError> {
         // FIXME: Prepared statement ids are local to a node
         // this method does not handle this
-        let response = self.any_connection().await?.batch(&batch, values).await?;
+        let response = self.any_connection()?.batch(&batch, values).await?;
         match response {
             Response::Error(err) => Err(err.into()),
             Response::Result(_) => Ok(()),
-            _ => Err(QueryError::ProtocolError(
-                "BATCH: Unexpected server response",
+            _ => Err(TransportError::InternalDriverError(
+                InternalDriverError::UnexpectedResponse,
             )),
         }
     }
 
-    pub async fn refresh_topology(&self) -> Result<(), QueryError> {
-        self.cluster.refresh_topology().await
+    /// Returns all connections that the session has currently opened.
+    pub fn get_connections(&self) -> Result<Connections, TransportError> {
+        Ok(self
+            .pool
+            .read()
+            .map_err(|_| {
+                TransportError::InternalDriverError(InternalDriverError::PoolLockPoisoned(
+                    POOL_LOCK_POISONED.to_string(),
+                ))
+            })?
+            .iter()
+            .map(|(&node, node_pool)| (node, node_pool.get_connections()))
+            .collect())
     }
 
-    async fn pick_connection(&self, t: Token) -> Result<Arc<Connection>, QueryError> {
+    pub async fn refresh_topology(&self) -> Result<(), TransportError> {
+        self.topology.refresh().await
+    }
+
+    async fn pick_connection(&self, t: Token) -> Result<Arc<Connection>, TransportError> {
         // TODO: we try only the owner of the range (vnode) that the token lies in
         // we should calculate the *set* of replicas for this token, using the replication strategy
         // of the table being queried.
-        let cluster_data: Arc<ClusterData> = self.cluster.get_data();
+        let owner = self.topology.read_ring()?.owner(t);
+
+        let mut found_pool = false;
+        let mut shard_info = None;
+        if let Some(node_pool) = self
+            .pool
+            .read()
+            .map_err(|_| {
+                TransportError::InternalDriverError(InternalDriverError::PoolLockPoisoned(
+                    POOL_LOCK_POISONED.to_string(),
+                ))
+            })?
+            .get(&owner)
+        {
+            // We have a pool for this node. Check if we have a connection for the token.
+            if let Some(c) = node_pool.connection_for_token(t) {
+                return Ok(c);
+            }
 
-        let first_node: Arc<Node> = cluster_data.ring.values().next().unwrap().clone();
+            found_pool = true;
+            shard_info = node_pool.get_shard_info().cloned();
+        };
 
-        let owner: Arc<Node> = cluster_data
-            .ring
-            .range((Included(t), Unbounded))
-            .next()
-            .map(|(_token, node)| node.clone())
-            .unwrap_or(first_node);
+        // We can't put this block in `else` for the previous if statement, because the block
+        // prolongs the lifetime of the read guard from `self.pool.read()`.
+        if !found_pool {
+            // We don't even have a pool for this node; in particular, we don't know its
+            // sharding info. We will just connect to a random shard this time so we don't do
+            // too much work on the query path (and we might still get lucky).
+            // The next request using this node's pool will pick the right connection, creating
+            // it if necessary.
+            let new_conn = open_connection(owner.addr, None, self.compression).await?;
+            let new_node_pool = NodePool::new(Arc::new(new_conn));
+
+            return Ok(self
+                .pool
+                .write()
+                .map_err(|_| {
+                    TransportError::InternalDriverError(InternalDriverError::PoolLockPoisoned(
+                        POOL_LOCK_POISONED.to_string(),
+                    ))
+                })?
+                .entry(owner)
+                // If someone created a node pool for this node faster while we were creating
+                // our own and inserted it into the session pool, drop our new pool and use the
+                // existing one.
+                // From this existing pool we will take any connection if there is no
+                // connection for this token yet.
+                .or_insert(new_node_pool)
+                .connection_for_token_or_any(t)?);
+        };
 
-        owner.connection_for_token(t).await
-    }
+        // We had a pool for this node, but there was no connection for this token in that pool.
+        // Let's create one.
+        // TODO: don't do this on the query path, but concurrently in some other fiber?
+        // Take any connection if there is no good connection for this node yet?
+        // TODO: opening this connection may actually fail because the port is already in
+        // use. We should try binding to a couple of ports before we fail. For that we need
+        // open_connection to return a well-typed error so we can check if it was an AddrInUse
+        // error. Or refactor this whole thing.
+        let new_conn_source_port = shard_info
+            .as_ref()
+            .map(|info| info.draw_source_port_for_token(t));
+        let new_conn =
+            Arc::new(open_connection(owner.addr, new_conn_source_port, self.compression).await?);
+
+        Ok(
+            match self
+                .pool
+                .write()
+                .map_err(|_| {
+                    TransportError::InternalDriverError(InternalDriverError::PoolLockPoisoned(
+                        POOL_LOCK_POISONED.to_string(),
+                    ))
+                })?
+                .entry(owner)
+            {
+                Entry::Vacant(entry) => {
+                    entry.insert(NodePool::new(new_conn.clone()));
+                    new_conn
+                }
+                Entry::Occupied(mut entry) => {
+                    let node_pool = entry.get_mut();
+                    if shard_info.as_ref() != node_pool.get_shard_info() {
+                        // The node has resharded while we were creating the connection.
+                        // Take any connection.
+                        return node_pool.connection_for_token_or_any(t);
+                    }
 
-    async fn any_connection(&self) -> Result<Arc<Connection>, QueryError> {
-        let random_token: Token = Token {
-            value: rand::thread_rng().gen(),
-        };
+                    let shard = node_pool.shard_for_token(t);
+                    node_pool.or_insert(shard, new_conn)
+                }
+            },
+        )
+    }
 
-        self.pick_connection(random_token).await
+    fn any_connection(&self) -> Result<Arc<Connection>, TransportError> {
+        Ok(self
+            .pool
+            .read()
+            // TODO: better error message?
+            .map_err(|_| {
+                TransportError::InternalDriverError(InternalDriverError::PoolLockPoisoned(
+                    POOL_LOCK_POISONED.to_string(),
+                ))
+            })?
+            .values()
+            .next()
+            .ok_or(TransportError::InternalDriverError(
+                InternalDriverError::FatalConnectionError,
+            ))?
+            .any_connection()?)
     }
 
     pub fn get_metrics(&self) -> MetricsView {
@@ -322,35 +457,79 @@ impl Session {
     }
 }
 
+impl NodePool {
+    pub fn new(connection: Arc<Connection>) -> Self {
+        let (shard, shard_info) =
+            match (connection.get_shard_info(), connection.get_is_shard_aware()) {
+                (Some(info), true) => (
+                    info.shard_of_source_port(connection.get_source_port()),
+                    Some(info.to_owned()),
+                ),
+                _ => (0, None),
+            };
+
+        Self {
+            connections: vec![(shard, connection)].into_iter().collect(),
+            shard_info,
+        }
+    }
+
+    /// Returns all connections that the session has currently opened.
+    pub fn get_connections(&self) -> Vec<Arc<Connection>> {
+        self.connections
+            .values()
+            .map(|conn| conn.to_owned())
+            .collect()
+    }
+
+    pub fn get_shard_info(&self) -> Option<&ShardInfo> {
+        self.shard_info.as_ref()
+    }
+
+    pub fn shard_for_token(&self, t: Token) -> Shard {
+        self.shard_info.as_ref().map_or(0, |info| info.shard_of(t))
+    }
+
+    pub fn connection_for_token(&self, t: Token) -> Option<Arc<Connection>> {
+        let shard = self.shard_for_token(t);
+        self.connections.get(&shard).map(|c| c.to_owned())
+    }
+
+    pub fn connection_for_token_or_any(&self, t: Token) -> Result<Arc<Connection>, TransportError> {
+        self.connection_for_token(t)
+            .map_or_else(|| self.any_connection(), Ok)
+    }
+
+    pub fn any_connection(&self) -> Result<Arc<Connection>, TransportError> {
+        Ok(self
+            .connections
+            .values()
+            .next()
+            .ok_or(TransportError::InternalDriverError(
+                InternalDriverError::FatalConnectionError,
+            ))?
+            .to_owned())
+    }
+
+    pub fn or_insert(&mut self, shard: Shard, conn: Arc<Connection>) -> Arc<Connection> {
+        self.connections.entry(shard).or_insert(conn).to_owned()
+    }
+}
+
 fn calculate_token(
     stmt: &PreparedStatement,
     values: &SerializedValues,
-) -> Result<Token, QueryError> {
+) -> Result<Token, PartitionKeyError> {
     // TODO: take the partitioner of the table that is being queried and calculate the token using
     // that partitioner. The below logic gives correct token only for murmur3partitioner
-    let partition_key = match stmt.compute_partition_key(values) {
-        Ok(key) => key,
-        Err(PartitionKeyError::NoPkIndexValue(_, _)) => {
-            return Err(QueryError::ProtocolError(
-                "No pk indexes - can't calculate token",
-            ))
-        }
-        Err(PartitionKeyError::ValueTooLong(values_len)) => {
-            return Err(QueryError::BadQuery(BadQuery::ValuesTooLongForKey(
-                values_len,
-                u16::max_value().into(),
-            )))
-        }
-    };
-
-    Ok(murmur3_token(partition_key))
+    Ok(murmur3_token(stmt.compute_partition_key(values)?))
 }
 
 // Resolve the given `ToSocketAddrs` using a DNS lookup if necessary.
 // The resolution may return multiple IPs and the function returns one of them.
 // It prefers to return IPv4s first, and only if there are none, IPv6s.
-async fn resolve(addr: impl ToSocketAddrs + Display) -> Result<SocketAddr, NewSessionError> {
-    let failed_err = NewSessionError::FailedToResolveAddress(addr.to_string());
+async fn resolve(addr: impl ToSocketAddrs + Display) -> Result<SocketAddr, TransportError> {
+    let failed_err = TransportError::FailedToResolveAddress(addr.to_string());
     let mut ret = None;
     for a in lookup_host(addr).await? {
         match a {
diff --git a/scylla/src/transport/topology.rs b/scylla/src/transport/topology.rs
index f9291cde..8a30012d 100644
--- a/scylla/src/transport/topology.rs
+++ b/scylla/src/transport/topology.rs
@@ -1,316 +1,364 @@
-use crate::routing::Token;
-use crate::transport::connection::Connection;
-use crate::transport::connection_keeper::ConnectionKeeper;
-use crate::transport::errors::QueryError;
-use crate::transport::session::IntoTypedRows;
-use crate::transport::Compression;
-
-use std::collections::HashMap;
-use std::net::{IpAddr, SocketAddr};
+use super::transport_errors::{InternalDriverError, TokenError, TopologyError, TransportError};
+
+use std::collections::hash_map::Entry;
+use std::collections::{BTreeMap, HashMap};
+use std::net::SocketAddr;
 use std::str::FromStr;
+use std::sync::{Arc, RwLock, RwLockReadGuard};
 
-/// Allows to read current topology info from the cluster
-pub struct TopologyReader {
-    control_connections: HashMap<SocketAddr, ConnectionKeeper>,
-    connect_port: u16,
-    compression: Option<Compression>,
-}
+use tokio::sync::{mpsc, oneshot};
+use tokio::time;
 
-/// Describes all topology information retrieved from the cluster
-pub struct TopologyInfo {
-    pub peers: Vec<Peer>,
-    pub keyspaces: HashMap<String, Keyspace>,
-}
+use crate::frame::response::result;
+use crate::routing::*;
+use crate::transport::connection::{open_connection, open_named_connection, Connection};
+
+const UPDATER_CRASHED: &str = "the topology updater thread has crashed. Need to restart it.";
 
-pub struct Peer {
-    pub address: SocketAddr,
-    pub tokens: Vec<Token>,
-    pub datacenter: Option<String>,
-    pub rack: Option<String>,
+// Represents a view of the cluster's topology, including the shape of the token ring.
+// Assiociated with a TopologyReader which may send update messages.
+pub struct Topology {
+    // TODO accessing the ring through Arc<RwLock<...>> on each read is not the best for efficiency
+    // the ring should probably be cached, and the cache refreshed periodically using the Arc
+    // calling refresh() would refresh it synchronously
+    ring: Arc<RwLock<Ring>>,
+
+    refresh_tx: mpsc::Sender<RefreshRequest>,
 }
 
-#[derive(Clone, Debug, PartialEq, Eq)]
-pub struct Keyspace {
-    pub strategy: Strategy,
+struct RefreshRequest {
+    ack_tx: oneshot::Sender<Result<(), TransportError>>,
 }
 
-#[derive(Clone, Debug, PartialEq, Eq)]
-pub enum Strategy {
-    SimpleStrategy {
-        replication_factor: usize,
-    },
-    NetworkTopologyStrategy {
-        // Replication factors of datacenters with given names
-        datacenter_repfactors: HashMap<String, usize>,
-    },
-    LocalStrategy, // replication_factor == 1
-    Other {
-        name: String,
-        data: HashMap<String, String>,
-    },
+// Responsible for periodically updating a Topology.
+pub struct TopologyReader {
+    // invariant: nonempty
+    pool: HashMap<Node, Connection>,
+
+    // invariant:
+    //  1. for every key `n` in `tokens`,
+    //     for every `t` in tokens[`n`],
+    //     `ring.owners` has a `(t, n)` entry
+    //  2. for every `(t, n)` entry in `ring.owners`,
+    //     `tokens` has an entry with key `n`,
+    //     and `tokens[n]` contains `t`.
+    //
+    // `tokens` also serves as a container for cluster members that we know (using the keys of the map).
+    // We assume that every ``normal'' member of the cluster (so not a joining or leaving one) has
+    // tokens in the ring. TODO: this might change in the future; in this case we'll need a
+    // separate set.
+    tokens: HashMap<Node, Vec<Token>>,
+    ring: Arc<RwLock<Ring>>,
+
+    refresh_rx: mpsc::Receiver<RefreshRequest>,
+
+    // Global port for the cluster. Assumes every node listens on this
+    port: u16,
 }
 
-impl TopologyReader {
-    /// Creates new TopologyReader, which connects to known_peers in the background
-    pub fn new(known_peers: &[SocketAddr], compression: Option<Compression>) -> Self {
-        let connect_port: u16 = known_peers
-            .first()
-            .expect("Tried to initialize TopologyReader with empty known_peers list!")
-            .port();
-
-        let mut control_connections: HashMap<SocketAddr, ConnectionKeeper> =
-            HashMap::with_capacity(known_peers.len());
-
-        for address in known_peers {
-            control_connections.insert(
-                *address,
-                ConnectionKeeper::new(*address, compression, None, None),
-            );
-        }
+impl Topology {
+    pub fn read_ring(&self) -> Result<RwLockReadGuard<'_, Ring>, TransportError> {
+        self.ring.read().map_err(|_| {
+            TransportError::InternalDriverError(InternalDriverError::UpdaterError(
+                UPDATER_CRASHED.to_string(),
+            ))
+        })
+    }
 
-        TopologyReader {
-            control_connections,
-            connect_port,
-            compression,
-        }
+    // Performs a synchronous refresh of the topology information.
+    // After the returned future is awaited, stored topology is not older
+    // than the cluster's topology at the moment `refresh` was called.
+    pub async fn refresh(&self) -> Result<(), TransportError> {
+        let (ack_tx, ack_rx) = oneshot::channel();
+        self.refresh_tx
+            .send(RefreshRequest { ack_tx })
+            .await
+            .map_err(|_| {
+                TransportError::InternalDriverError(InternalDriverError::UpdaterError(
+                    UPDATER_CRASHED.to_string(),
+                ))
+            })?;
+        ack_rx.await.map_err(|_| {
+            TransportError::InternalDriverError(InternalDriverError::UpdaterError(
+                UPDATER_CRASHED.to_string(),
+            ))
+        })?
     }
+}
 
-    /// Fetches current topology info from the cluster
-    pub async fn read_topology_info(&mut self) -> Result<TopologyInfo, QueryError> {
-        let result = self.fetch_topology_info().await;
+impl TopologyReader {
+    pub async fn new(n: Node) -> Result<(Self, Topology), TransportError> {
+        // TODO: use compression? maybe not necessarily, since the communicated objects are
+        // small and the communication doesn't happen often?
+        let conn = open_named_connection(
+            n.addr,
+            None,
+            None,
+            Some("scylla-rust-driver:control-connection".to_string()),
+        )
+        .await?;
+
+        // TODO: When querying system.local of `n` or system.peers of other nodes, we might find
+        // that the address of `n` is different than `n.addr` (`n` might have multiple addresses).
+        // We need to deal with that. Since the data structures are filled using data obtained from
+        // `system.peers` and `system.local`, we probably need to query `system.local` for the
+        // address that `n` itself prefers.
+
+        let toks = query_local_tokens(&conn).await?;
+        if toks.is_empty() {
+            return Err(TransportError::InternalDriverError(
+                InternalDriverError::LocalTokensEmptyOnNodes,
+            ));
+        }
 
-        if let Ok(topology_info) = &result {
-            self.update_control_connections(topology_info);
+        let mut owners = BTreeMap::new();
+        for &t in &toks {
+            owners.insert(t, n);
         }
 
-        return result;
+        let ring = Arc::new(RwLock::new(Ring { owners }));
+        let tokens = vec![(n, toks)].into_iter().collect();
+        let pool = vec![(n, conn)].into_iter().collect();
+        let (refresh_tx, refresh_rx) = mpsc::channel(32);
+
+        Ok((
+            TopologyReader {
+                pool,
+                tokens,
+                ring: ring.clone(),
+                refresh_rx,
+                port: n.addr.port(),
+            },
+            Topology { ring, refresh_tx },
+        ))
     }
 
-    async fn fetch_topology_info(&self) -> Result<TopologyInfo, QueryError> {
-        // TODO: Timeouts? New attempts in parallel?
-
-        let mut last_error: QueryError = QueryError::ProtocolError(
-            "Invariant broken - TopologyReader control_connections not empty",
-        );
-
-        for conn in self.control_connections.values() {
-            match query_topology_info(conn, self.connect_port).await {
-                Ok(info) => return Ok(info),
-                Err(e) => last_error = e,
-            };
+    pub async fn run(mut self) {
+        use time::Duration;
+
+        let mut sleep_dur = Duration::from_secs(0);
+        loop {
+            tokio::select! {
+                _ = time::sleep(sleep_dur) => {
+                    sleep_dur = match self.refresh().await {
+                        Ok(()) => Duration::from_secs(10),
+                        Err(e) => {
+                            // TODO: use appropriate logging mechanism
+                            eprintln!("TopologyReader: error on periodic topology refresh: {}. Will retry", e);
+                            Duration::from_secs(2)
+                        }
+                    };
+                },
+                Some(request) = self.refresh_rx.recv() => {
+                    let _ = request.ack_tx.send(self.refresh().await);
+                    // If there was a send error, the other end must have closed the channel, i.e.
+                    // it wasn't interested in the result anymore. We don't need to do anything.
+                    sleep_dur = Duration::from_secs(10)
+                }
+            }
         }
-
-        Err(last_error)
     }
 
-    fn update_control_connections(&mut self, topology_info: &TopologyInfo) {
-        let mut new_control_connections: HashMap<SocketAddr, ConnectionKeeper> =
-            HashMap::with_capacity(topology_info.peers.len());
+    async fn refresh(&mut self) -> Result<(), TransportError> {
+        use time::{timeout, Duration};
+
+        let mut broken_connections = vec![];
+        let mut refreshed = false;
+        for (&node, conn) in self.pool.iter() {
+            // TODO: handle broken connections
+            // TODO: after short timeout, attempt to refresh on another connection *in parallel*
+            // while continuing waiting on the previous connection
+            let (peer_addrs, _) =
+                match timeout(Duration::from_secs(5), query_peers(conn, false, self.port)).await {
+                    Err(_) => {
+                        // A timeout doesn't necessarily mean that the connection is useless.
+                        // It may be a temporary network partition between us and the server.
+                        continue;
+                    }
+                    Ok(Ok(ret)) => ret,
+                    Ok(Err(e)) => {
+                        eprintln!("Error when querying system.peers of {}: {}", node.addr, e);
+                        // TODO distinguish between different error types and react accordingly
+                        // for now, we just drop the connection on every error.
+                        broken_connections.push(node);
+                        continue;
+                    }
+                };
 
-        for peer in &topology_info.peers {
-            let cur_connection = self
-                .control_connections
-                .remove(&peer.address)
-                .unwrap_or_else(|| {
-                    ConnectionKeeper::new(peer.address, self.compression, None, None)
-                });
+            // We assume that tokens of existing peers don't change.
+            // TODO: in the future this assumption might become false
+            let need_new_tokens = peer_addrs
+                .iter()
+                .any(|addr| !self.tokens.contains_key(&Node { addr: *addr }));
+
+            if need_new_tokens {
+                let (peer_addrs, peer_toks) =
+                    match timeout(Duration::from_secs(5), query_peers(conn, true, self.port)).await
+                    {
+                        Err(_) => continue,
+                        Ok(Ok(ret)) => ret,
+                        Ok(Err(e)) => {
+                            eprintln!("Error when querying system.peers of {}: {}", node.addr, e);
+                            // TODO get rid of copy-pasta
+                            broken_connections.push(node);
+                            continue;
+                        }
+                    };
+                assert_eq!(peer_addrs.len(), peer_toks.len());
+                for (addr, toks) in peer_addrs.into_iter().zip(peer_toks.into_iter()) {
+                    let n = Node { addr };
+                    let new_ring = {
+                        let mut ring = self
+                            .ring
+                            .read()
+                            .unwrap_or_else(|_| panic!("poisoned ring lock")) // TODO: handle this better?
+                            .clone();
+                        for &t in &toks {
+                            ring.owners.insert(t, n);
+                        }
+                        ring
+                    };
+
+                    let mut ring_ref = self
+                        .ring
+                        .write()
+                        .unwrap_or_else(|_| panic!("poisoned ring lock")); // TODO same as above?
+                    *ring_ref = new_ring;
+
+                    self.tokens.insert(n, toks);
+                }
+            }
 
-            new_control_connections.insert(peer.address, cur_connection);
-        }
+            // TODO: Remove nodes that are not in peers anymore
+            // (from pool, tokens, and ring)
 
-        self.control_connections = new_control_connections;
-    }
-}
+            refreshed = true;
+        }
 
-async fn query_topology_info(
-    conn_keeper: &ConnectionKeeper,
-    connect_port: u16,
-) -> Result<TopologyInfo, QueryError> {
-    let conn: &Connection = &*conn_keeper.get_connection().await?;
+        for node in broken_connections {
+            self.pool.remove(&node);
+        }
 
-    let peers_query = query_peers(conn, connect_port);
-    let keyspaces_query = query_keyspaces(conn);
+        // Try opening new connections
+        let mut any_new = false;
+        for &node in self.tokens.keys() {
+            if let Entry::Vacant(entry) = self.pool.entry(node) {
+                // TODO: use compression?
+                let conn = match timeout(
+                    Duration::from_secs(5),
+                    open_connection(node.addr, None, None),
+                )
+                .await
+                {
+                    Err(_) => {
+                        // TODO: logging
+                        eprintln!(
+                            "TopologyReader: timeout when trying to open connection to {}",
+                            node.addr
+                        );
+                        continue;
+                    }
+                    Ok(Err(e)) => {
+                        // TODO: logging
+                        eprintln!(
+                            "TopologyReader: failed to open connection to {}: {}",
+                            node.addr, e
+                        );
+                        continue;
+                    }
+                    Ok(Ok(conn)) => conn,
+                };
+                eprintln!("TopologyReader: opened new connection to {}", node.addr);
+                entry.insert(conn);
+                any_new = true;
+            }
+        }
 
-    let (peers, keyspaces) = tokio::try_join!(peers_query, keyspaces_query)?;
+        if refreshed {
+            return Ok(());
+        }
 
-    // There must be at least one peer
-    if peers.is_empty() {
-        return Err(QueryError::ProtocolError(
-            "Bad TopologyInfo: peers list is empty",
-        ));
-    }
+        if any_new {
+            return Err(TransportError::InternalDriverError(
+                InternalDriverError::RefreshingFailedOnEveryConnection(
+                    "new connections have been opened".to_string(),
+                ),
+            ));
+        }
 
-    // At least one peer has to have some tokens
-    if peers.iter().all(|peer| peer.tokens.is_empty()) {
-        return Err(QueryError::ProtocolError(
-            "Bad TopoologyInfo: All peers have empty token list",
-        ));
+        Err(TransportError::InternalDriverError(
+            InternalDriverError::RefreshingFailedOnEveryConnection(
+                "no new connections could be opened".to_string(),
+            ),
+        ))
     }
-
-    Ok(TopologyInfo { peers, keyspaces })
 }
 
-async fn query_peers(conn: &Connection, connect_port: u16) -> Result<Vec<Peer>, QueryError> {
-    // There shouldn't be more peers than a single page capacity
-    let peers_query = conn.query_single_page(
-        "select peer, data_center, rack, tokens from system.peers",
-        &[],
-    );
-    let local_query = conn.query_single_page(
-        "select listen_address, data_center, rack, tokens from system.local",
-        &[],
-    );
-
-    let (peers_res, local_res) = tokio::try_join!(peers_query, local_query)?;
-
-    let peers_rows = peers_res.ok_or(QueryError::ProtocolError(
-        "system.peers query response was not Rows",
-    ))?;
-
-    let local_rows = local_res.ok_or(QueryError::ProtocolError(
-        "system.local query response was not Rows",
-    ))?;
-
-    let mut result: Vec<Peer> = Vec::with_capacity(peers_rows.len() + 1);
-
-    let typed_peers_rows =
-        peers_rows.into_typed::<(IpAddr, Option<String>, Option<String>, Option<Vec<String>>)>();
-    let typed_local_rows =
-        local_rows.into_typed::<(IpAddr, Option<String>, Option<String>, Option<Vec<String>>)>();
-
-    for row in typed_peers_rows.chain(typed_local_rows) {
-        let (ip_address, datacenter, rack, tokens) = row.map_err(|_| {
-            QueryError::ProtocolError("system.peers or system.local has invalid column type")
-        })?;
-
-        let tokens_str: Vec<String> = tokens.unwrap_or_default();
-
-        let address = SocketAddr::new(ip_address, connect_port);
-
-        // Parse string representation of tokens as integer values
-        let tokens: Vec<Token> = tokens_str
-            .iter()
-            .map(|s| Token::from_str(&s))
-            .collect::<Result<Vec<Token>, _>>()
-            .map_err(|_| QueryError::ProtocolError("Couldn't parse tokens as integer values"))?;
-
-        result.push(Peer {
-            address,
-            datacenter,
-            rack,
-            tokens,
-        });
-    }
-
-    Ok(result)
+async fn query_local_tokens(c: &Connection) -> Result<Vec<Token>, TransportError> {
+    unwrap_tokens(
+        c.query_single_page("SELECT tokens FROM system.local", &[])
+            .await?
+            .ok_or(TopologyError::LocalExpectedRowResults)?
+            .into_iter()
+            .next()
+            .ok_or(TopologyError::LocalQueryResultEmpty)?
+            .columns
+            .into_iter()
+            .next()
+            .ok_or(TopologyError::LocalTokensNoColumnsReturned)?,
+    )
 }
 
-async fn query_keyspaces(conn: &Connection) -> Result<HashMap<String, Keyspace>, QueryError> {
-    let rows = conn
+async fn query_peers(
+    c: &Connection,
+    get_tokens: bool,
+    port: u16,
+) -> Result<(Vec<SocketAddr>, Vec<Vec<Token>>), TransportError> {
+    // TODO: do we want `peer`, `preferred_ip`, or `rpc_address`? Which one is `external`?
+    let rows = c
         .query_single_page(
-            "select keyspace_name, toJson(replication) from system_schema.keyspaces",
+            format!(
+                "SELECT {} FROM system.peers",
+                if get_tokens { "peer, tokens" } else { "peer" }
+            ),
             &[],
         )
         .await?
-        .ok_or(QueryError::ProtocolError(
-            "system_schema.keyspaces query response was not Rows",
-        ))?;
-
-    let mut result = HashMap::with_capacity(rows.len());
-
-    for row in rows.into_typed::<(String, String)>() {
-        let (keyspace_name, keyspace_json_text) = row.map_err(|_| {
-            QueryError::ProtocolError("system_schema.keyspaces has invalid column type")
-        })?;
-
-        let strategy_map: HashMap<String, String> = json_to_string_map(&keyspace_json_text)?;
-
-        let strategy: Strategy = strategy_from_string_map(strategy_map)?;
+        .ok_or(TopologyError::PeersRowError)?;
 
-        result.insert(keyspace_name, Keyspace { strategy });
-    }
-
-    Ok(result)
-}
-
-fn json_to_string_map(json_text: &str) -> Result<HashMap<String, String>, QueryError> {
-    use serde_json::Value;
-
-    let json: Value = serde_json::from_str(json_text)
-        .map_err(|_| QueryError::ProtocolError("Couldn't parse keyspaces as json"))?;
-
-    let object_map = match json {
-        Value::Object(map) => map,
-        _ => {
-            return Err(QueryError::ProtocolError(
-                "keyspaces map json is not a json object",
-            ))
-        }
+    let mut addrs = Vec::with_capacity(rows.len());
+    let mut tokens = if get_tokens {
+        vec![]
+    } else {
+        Vec::with_capacity(rows.len())
     };
-
-    let mut result = HashMap::with_capacity(object_map.len());
-
-    for (key, val) in object_map.into_iter() {
-        match val {
-            Value::String(string) => result.insert(key, string),
-            _ => {
-                return Err(QueryError::ProtocolError(
-                    "json keyspaces map does not contain strings",
-                ))
-            }
-        };
+    for row in rows {
+        let mut col_iter = row.columns.into_iter();
+
+        let peer = col_iter
+            .next()
+            .ok_or(TopologyError::PeersColumnError)?
+            .ok_or(TopologyError::PeersValueTypeError)?
+            .as_inet()
+            .ok_or(TopologyError::PeersValueTypeError)?;
+        addrs.push(SocketAddr::new(peer, port));
+
+        if get_tokens {
+            let toks = unwrap_tokens(col_iter.next().ok_or(TopologyError::PeersColumnError)?)?;
+            tokens.push(toks);
+        }
     }
-
-    Ok(result)
+    Ok((addrs, tokens))
 }
 
-fn strategy_from_string_map(
-    mut strategy_map: HashMap<String, String>,
-) -> Result<Strategy, QueryError> {
-    let strategy_name: String = strategy_map
-        .remove("class")
-        .ok_or(QueryError::ProtocolError(
-            "strategy map should have a 'class' field",
-        ))?;
-
-    let strategy: Strategy = match strategy_name.as_str() {
-        "org.apache.cassandra.locator.SimpleStrategy" => {
-            let rep_factor_str: String =
-                strategy_map
-                    .remove("replication_factor")
-                    .ok_or(QueryError::ProtocolError(
-                        "SimpleStrategy in strategy map does not have a replication factor",
-                    ))?;
-
-            let replication_factor: usize = usize::from_str(&rep_factor_str).map_err(|_| {
-                QueryError::ProtocolError("Could not parse replication factor as an integer")
-            })?;
+fn unwrap_tokens(tokens_val_opt: Option<result::CQLValue>) -> Result<Vec<Token>, TransportError> {
+    let tokens_val = tokens_val_opt.ok_or(TokenError::NullTokens)?;
+    let tokens = tokens_val.as_set().ok_or(TokenError::SetValue)?;
 
-            Strategy::SimpleStrategy { replication_factor }
-        }
-        "org.apache.cassandra.locator.NetworkTopologyStrategy" => {
-            let mut datacenter_repfactors: HashMap<String, usize> =
-                HashMap::with_capacity(strategy_map.len());
-
-            for (datacenter, rep_factor_str) in strategy_map.drain() {
-                let rep_factor: usize = match usize::from_str(&rep_factor_str) {
-                    Ok(number) => number,
-                    Err(_) => continue, // There might be other things in the map, we care only about rep_factors
-                };
-
-                datacenter_repfactors.insert(datacenter, rep_factor);
-            }
-
-            Strategy::NetworkTopologyStrategy {
-                datacenter_repfactors,
-            }
-        }
-        "org.apache.cassandra.locator.LocalStrategy" => Strategy::LocalStrategy,
-        _ => Strategy::Other {
-            name: strategy_name,
-            data: strategy_map,
-        },
-    };
-
-    Ok(strategy)
+    let mut ret = Vec::with_capacity(tokens.len());
+    for tok_val in tokens {
+        let tok = tok_val.as_text().ok_or(TokenError::NonTextValue)?;
+        ret.push(Token::from_str(tok).map_err(|_| TokenError::ParseFailure(tok.to_string()))?);
+    }
+    Ok(ret)
 }
diff --git a/scylla/src/transport/transport_errors.rs b/scylla/src/transport/transport_errors.rs
new file mode 100644
index 00000000..b540d6fd
--- /dev/null
+++ b/scylla/src/transport/transport_errors.rs
@@ -0,0 +1,110 @@
+use crate::frame::frame_errors;
+use crate::frame::frame_errors::ParseError;
+use crate::frame::response::cql_to_rust::FromRowError;
+use crate::frame::value::SerializeValuesError;
+use crate::prepared_statement::PartitionKeyError;
+use thiserror::Error;
+
+#[derive(Error, Debug)]
+pub enum TokenError {
+    #[error("tokens value null")]
+    NullTokens,
+    #[error("tokens value type is not a set")]
+    SetValue,
+    #[error("tokens set contained a non-text value")]
+    NonTextValue,
+    #[error("failed to parse token {0}")]
+    ParseFailure(String),
+}
+
+#[derive(Error, Debug)]
+pub enum TopologyError {
+    #[error("system.peers query returned no columns")]
+    PeersColumnError,
+    #[error("system.peers `peer` query returned wrong value type or null")]
+    PeersValueTypeError,
+    #[error("Expected rows result when querying system.peers")]
+    PeersRowError,
+
+    #[error("Expected rows result when querying system.local")]
+    LocalExpectedRowResults,
+    #[error("system.local query result empty")]
+    LocalQueryResultEmpty,
+    #[error("system.local `tokens` query returned no columns")]
+    LocalTokensNoColumnsReturned,
+}
+
+#[derive(Error, Debug)]
+pub enum InternalDriverError {
+    #[error("Unexpected response!")]
+    UnexpectedResponse,
+    #[error("refreshing failed on every connection, {0}")]
+    RefreshingFailedOnEveryConnection(String),
+    #[error("system.local tokens empty on the node")]
+    LocalTokensEmptyOnNodes,
+    #[error("Pool Lock Poisoned: {0}")]
+    PoolLockPoisoned(String),
+    #[error("Updater Crashed: {0}")]
+    UpdaterError(String),
+    #[error("fatal error, broken invariant: no connections available")]
+    FatalConnectionError,
+}
+
+#[derive(Error, Debug)]
+pub enum PrepareError {
+    #[error("Internal Driver Error")]
+    InternalDriverError(#[from] InternalDriverError),
+}
+
+#[derive(Error, Debug)]
+pub enum DBError {
+    // TODO develop this while implementing retry policies
+    // to react differently to each error.
+    // Make it fit what we encounter.
+    #[error("Response is an error message: code: {0} message: {1}")]
+    ErrorMsg(i32, String),
+}
+
+#[derive(Error, Debug)]
+pub enum TransportError {
+    #[error("Task queue closed")]
+    TaskQueueClosed,
+    #[error("Request dropped!")]
+    DroppedRequest,
+    #[error("failed to resolve {0}")]
+    FailedToResolveAddress(String),
+    #[error("Reprepared statement unexpectedly changed its id")]
+    RepreparedStatmentIDChanged,
+    #[error("IDs of statements prepared using different connections do not match")]
+    PreparedStatementsIDsNotMatch,
+    #[error("Length of provided values ({0}) must be equal to number of batch statements ({1})")]
+    ValueLenMismatch(usize, usize),
+    #[error("No connections available")]
+    NoConnectionsAvailable,
+
+    #[error("Internal Driver Error")]
+    InternalDriverError(#[from] InternalDriverError),
+
+    #[error(transparent)]
+    DBError(#[from] DBError),
+    #[error(transparent)]
+    PrepareError(#[from] PrepareError),
+    #[error(transparent)]
+    ParseError(#[from] ParseError),
+    #[error(transparent)]
+    ToplogyError(#[from] TopologyError),
+    #[error(transparent)]
+    TokenError(#[from] TokenError),
+    #[error(transparent)]
+    StdIOError(#[from] std::io::Error),
+    #[error(transparent)]
+    FrameError(#[from] frame_errors::FrameError),
+    #[error(transparent)]
+    TokioSyncOneshotError(#[from] tokio::sync::oneshot::error::RecvError),
+    #[error(transparent)]
+    FromRowError(#[from] FromRowError),
+    #[error(transparent)]
+    SerializeValuesError(#[from] SerializeValuesError),
+    #[error(transparent)]
+    PartitionKeyError(#[from] PartitionKeyError),
+}


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);
+        }
+    }
+}