Repository: scylladb/scylla-rust-driver
Pull request #349 (2021-11-23 16:58:03)

Expose schema metadata
This pull request introduces an insight into the schema of the cluster. From now on it is possible to access keyspace elements such as tables, their columns, types, kinds (regular, static, clustering, partition_key), ordering of partition key and clustering columns, UDTs, partitioner which are stored in `Metadata` structure (former `TopologyInfo`). The client has a way to disable this feature using a proper flag in `SessionConfig`.

This pull request introduces new external dependency namely strum.

## Pre-review checklist

- [x] I have split my patch into logically separate commits.
- [x] All commit messages clearly explain what they change and why.
- [x] I added relevant tests for new features and bug fixes.
- [x] All commits compile, pass static checks and pass test.
- [x] PR description sums up the changes and reasons why they should be introduced.
- [x] I added appropriate `Fixes:` annotations to PR description.

Fixes: #346 

piodul: @Ponewor One more thing - some users have an enormous amount of keyspaces and tables in their cluster but they might be not interested in accessing information about schema. We should provide an opt-out switch in the session configuration to turn off fetching schema information (fetching topology information should be always on).
Ponewor: > @Ponewor One more thing - some users have an enormous amount of keyspaces and tables in their cluster but they might be not interested in accessing information about schema. We should provide an opt-out switch in the session configuration to turn off fetching schema information (fetching topology information should be always on).

I have tried to do this the simplest way and here is what I have achieved:
The struct
```rust
pub struct Keyspace {
    pub strategy: Strategy,
    pub tables: HashMap<String, Table>,
    pub user_defined_types: HashMap<String, Vec<(String, CqlType)>>,
}
```
has turned into

```rust
pub struct Keyspace {
    pub strategy: Strategy,
    pub tables: Option<HashMap<String, Table>>,
    pub user_defined_types: Option<HashMap<String, Vec<(String, CqlType)>>>,
}
```
as `strategy` is required and `tables` and `user_defined_types are optional`. I do not like it. It makes using the API cumbersome as it may be observed in tests. In the future `Keyspace` will gain additional fields and all of them would be `Option`s if we continue this way. Maybe it would be better to turn it into
```rust
pub struct Keyspace {
    pub strategy: Strategy,
    pub schema: Option<Schema>,
}

pub struct Schema {
    pub tables: HashMap<String, Table>,
    pub user_defined_types: HashMap<String, Vec<(String, CqlType)>>,
}
```
? What do you think? Maybe I am being overly dramatic over few `unwrap`s? If I am not, is schema the right name for this additional struct? I wonder if schema is a more general term containing all of the keyspaces.
piodul: > ? What do you think? Maybe I am being overly dramatic over few `unwrap`s? If I am not, is schema the right name for this additional struct? I wonder if schema is a more general term containing all of the keyspaces.

@Ponewor I agree that the `Option`s make it cumbersome. I think we can get rid of them and just leave the hashmaps empty if fetching schema information is disabled - if the user disabled it then they won't be interested in looking at those fields. A docstring comment for those fields mentioning this behavior would be nice to have.

Patch:
diff --git a/.github/workflows/cassandra.yml b/.github/workflows/cassandra.yml
index 54e7a75c..13999f68 100644
--- a/.github/workflows/cassandra.yml
+++ b/.github/workflows/cassandra.yml
@@ -18,6 +18,8 @@ jobs:
         image: cassandra
         ports:
           - 9042:9042
+    env:
+      CDC: 'disabled'
     steps:
     - uses: actions/checkout@v2
       # A separate step for building to separate measuring time of compilation and testing
diff --git a/docs/source/data-types/collections.md b/docs/source/data-types/collections.md
index 43301d31..1557ca68 100644
--- a/docs/source/data-types/collections.md
+++ b/docs/source/data-types/collections.md
@@ -11,9 +11,9 @@
 use scylla::IntoTypedRows;
 
 // Insert a list of ints into the table
-let my_list: Vec<i32> = vec![1, 2, 3, 4, 5];
+let to_insert: Vec<i32> = vec![1, 2, 3, 4, 5];
 session
-    .query("INSERT INTO keyspace.table (a) VALUES(?)", (&my_list,))
+    .query("INSERT INTO keyspace.table (a) VALUES(?)", (&to_insert,))
     .await?;
 
 // Read a list of ints from the table
@@ -27,7 +27,7 @@ if let Some(rows) = session.query("SELECT a FROM keyspace.table", &[]).await?.ro
 ```
 
 ## Set
-`Set` is represented as `Vec<T>`, `HashSet<T>` or `BTreeSet<T>`:
+`Set` is represented as `Vec<T>`
 
 ```rust
 # extern crate scylla;
@@ -37,9 +37,9 @@ if let Some(rows) = session.query("SELECT a FROM keyspace.table", &[]).await?.ro
 use scylla::IntoTypedRows;
 
 // Insert a set of ints into the table
-let my_set: Vec<i32> = vec![1, 2, 3, 4, 5];
+let to_insert: Vec<i32> = vec![1, 2, 3, 4, 5];
 session
-    .query("INSERT INTO keyspace.table (a) VALUES(?)", (&my_set,))
+    .query("INSERT INTO keyspace.table (a) VALUES(?)", (&to_insert,))
     .await?;
 
 // Read a set of ints from the table
@@ -52,56 +52,8 @@ if let Some(rows) = session.query("SELECT a FROM keyspace.table", &[]).await?.ro
 # }
 ```
 
-```rust
-# extern crate scylla;
-# use scylla::Session;
-# use std::error::Error;
-# async fn check_only_compiles(session: &Session) -> Result<(), Box<dyn Error>> {
-use scylla::IntoTypedRows;
-use std::collections::HashSet;
-
-// Insert a set of ints into the table
-let my_set: HashSet<i32> = vec![1, 2, 3, 4, 5].into_iter().collect();
-session
-    .query("INSERT INTO keyspace.table (a) VALUES(?)", (&my_set,))
-    .await?;
-
-// Read a set of ints from the table
-if let Some(rows) = session.query("SELECT a FROM keyspace.table", &[]).await?.rows {
-    for row in rows.into_typed::<(HashSet<i32>,)>() {
-        let (set_value,): (HashSet<i32>,) = row?;
-    }
-}
-# Ok(())
-# }
-```
-
-```rust
-# extern crate scylla;
-# use scylla::Session;
-# use std::error::Error;
-# async fn check_only_compiles(session: &Session) -> Result<(), Box<dyn Error>> {
-use scylla::IntoTypedRows;
-use std::collections::BTreeSet;
-
-// Insert a set of ints into the table
-let my_set: BTreeSet<i32> = vec![1, 2, 3, 4, 5].into_iter().collect();
-session
-    .query("INSERT INTO keyspace.table (a) VALUES(?)", (&my_set,))
-    .await?;
-
-// Read a set of ints from the table
-if let Some(rows) = session.query("SELECT a FROM keyspace.table", &[]).await?.rows {
-    for row in rows.into_typed::<(BTreeSet<i32>,)>() {
-        let (set_value,): (BTreeSet<i32>,) = row?;
-    }
-}
-# Ok(())
-# }
-```
-
 ## Map
-`Map` is represented as `HashMap<K, V>` or `BTreeMap<K, V>`
+`Map` is represented as `std::collections::HashMap<K, V>`
 
 ```rust
 # extern crate scylla;
@@ -112,11 +64,11 @@ use scylla::IntoTypedRows;
 use std::collections::HashMap;
 
 // Insert a map of text and int into the table
-let mut my_map: HashMap<String, i32> = HashMap::new();
-my_map.insert("abcd".to_string(), 16);
+let mut to_insert: HashMap<String, i32> = HashMap::new();
+to_insert.insert("abcd".to_string(), 16);
 
 session
-    .query("INSERT INTO keyspace.table (a) VALUES(?)", (&my_map,))
+    .query("INSERT INTO keyspace.table (a) VALUES(?)", (&to_insert,))
     .await?;
 
 // Read a map from the table
@@ -127,30 +79,4 @@ if let Some(rows) = session.query("SELECT a FROM keyspace.table", &[]).await?.ro
 }
 # Ok(())
 # }
-```
-
-```rust
-# extern crate scylla;
-# use scylla::Session;
-# use std::error::Error;
-# async fn check_only_compiles(session: &Session) -> Result<(), Box<dyn Error>> {
-use scylla::IntoTypedRows;
-use std::collections::BTreeMap;
-
-// Insert a map of text and int into the table
-let mut my_map: BTreeMap<String, i32> = BTreeMap::new();
-my_map.insert("abcd".to_string(), 16);
-
-session
-    .query("INSERT INTO keyspace.table (a) VALUES(?)", (&my_map,))
-    .await?;
-
-// Read a map from the table
-if let Some(rows) = session.query("SELECT a FROM keyspace.table", &[]).await?.rows {
-    for row in rows.into_typed::<(BTreeMap<String, i32>,)>() {
-        let (map_value,): (BTreeMap<String, i32>,) = row?;
-    }
-}
-# Ok(())
-# }
 ```
\ No newline at end of file
diff --git a/scylla/Cargo.toml b/scylla/Cargo.toml
index 378d2daa..ef2b7e6b 100644
--- a/scylla/Cargo.toml
+++ b/scylla/Cargo.toml
@@ -35,6 +35,8 @@ openssl = { version = "0.10.32", optional = true }
 tokio-openssl = { version = "0.6.1", optional = true }
 arc-swap = "1.3.0"
 dashmap = "4.0.2"
+strum = "0.23"
+strum_macros = "0.23"
 
 [dev-dependencies]
 criterion = "0.3"
diff --git a/scylla/src/frame/response/cql_to_rust.rs b/scylla/src/frame/response/cql_to_rust.rs
index f8a3efaa..4ce3f2ee 100644
--- a/scylla/src/frame/response/cql_to_rust.rs
+++ b/scylla/src/frame/response/cql_to_rust.rs
@@ -3,7 +3,8 @@ use crate::frame::value::Counter;
 use bigdecimal::BigDecimal;
 use chrono::{Duration, NaiveDate};
 use num_bigint::BigInt;
-use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
+use std::collections::HashMap;
+use std::collections::HashSet;
 use std::hash::Hash;
 use std::net::IpAddr;
 use thiserror::Error;
@@ -43,13 +44,6 @@ pub trait FromRow: Sized {
     fn from_row(row: Row) -> Result<Self, FromRowError>;
 }
 
-// CqlValue can be converted to CqlValue
-impl FromCqlVal<CqlValue> for CqlValue {
-    fn from_cql(cql_val: CqlValue) -> Result<CqlValue, FromCqlValError> {
-        Ok(cql_val)
-    }
-}
-
 // Implement from_cql<Option<CqlValue>> for every type that has from_cql<CqlValue>
 // This tries to unwrap the option or fails with an error
 impl<T: FromCqlVal<CqlValue>> FromCqlVal<Option<CqlValue>> for T {
@@ -133,30 +127,6 @@ impl<T: FromCqlVal<CqlValue> + Eq + Hash> FromCqlVal<CqlValue> for HashSet<T> {
     }
 }
 
-impl<T: FromCqlVal<CqlValue> + Ord> FromCqlVal<CqlValue> for BTreeSet<T> {
-    fn from_cql(cql_val: CqlValue) -> Result<Self, FromCqlValError> {
-        cql_val
-            .into_vec()
-            .ok_or(FromCqlValError::BadCqlType)?
-            .into_iter()
-            .map(T::from_cql)
-            .collect::<Result<BTreeSet<T>, FromCqlValError>>()
-    }
-}
-
-impl<K: FromCqlVal<CqlValue> + Ord, V: FromCqlVal<CqlValue>> FromCqlVal<CqlValue>
-    for BTreeMap<K, V>
-{
-    fn from_cql(cql_val: CqlValue) -> Result<Self, FromCqlValError> {
-        let vec = cql_val.into_pair_vec().ok_or(FromCqlValError::BadCqlType)?;
-        let mut res = BTreeMap::new();
-        for (key, value) in vec {
-            res.insert(K::from_cql(key)?, V::from_cql(value)?);
-        }
-        Ok(res)
-    }
-}
-
 macro_rules! replace_expr {
     ($_t:tt $sub:expr) => {
         $sub
diff --git a/scylla/src/frame/response/result.rs b/scylla/src/frame/response/result.rs
index 0f69b57f..beac805c 100644
--- a/scylla/src/frame/response/result.rs
+++ b/scylla/src/frame/response/result.rs
@@ -326,7 +326,7 @@ impl Row {
     }
 }
 
-#[derive(Debug)]
+#[derive(Debug, Default)]
 pub struct Rows {
     pub metadata: ResultMetadata,
     rows_count: usize,
diff --git a/scylla/src/frame/value.rs b/scylla/src/frame/value.rs
index e25ff216..787ada24 100644
--- a/scylla/src/frame/value.rs
+++ b/scylla/src/frame/value.rs
@@ -4,7 +4,7 @@ use chrono::prelude::*;
 use chrono::Duration;
 use num_bigint::BigInt;
 use std::borrow::Cow;
-use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
+use std::collections::HashMap;
 use std::convert::TryInto;
 use std::net::IpAddr;
 use thiserror::Error;
@@ -447,53 +447,27 @@ fn serialize_map<K: Value, V: Value>(
     Ok(())
 }
 
-fn serialize_list_or_set<'a, V: 'a + Value>(
-    elements_iter: impl Iterator<Item = &'a V>,
-    element_count: usize,
-    buf: &mut Vec<u8>,
-) -> Result<(), ValueTooBig> {
-    let bytes_num_pos: usize = buf.len();
-    buf.put_i32(0);
-
-    buf.put_i32(element_count.try_into().map_err(|_| ValueTooBig)?);
-    for value in elements_iter {
-        <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<V: Value> Value for HashSet<V> {
-    fn serialize(&self, buf: &mut Vec<u8>) -> Result<(), ValueTooBig> {
-        serialize_list_or_set(self.iter(), self.len(), buf)
-    }
-}
-
 impl<K: Value, V: Value> Value for HashMap<K, V> {
     fn serialize(&self, buf: &mut Vec<u8>) -> Result<(), ValueTooBig> {
         serialize_map(self.iter(), self.len(), buf)
     }
 }
 
-impl<V: Value> Value for BTreeSet<V> {
+impl<T: Value> Value for Vec<T> {
     fn serialize(&self, buf: &mut Vec<u8>) -> Result<(), ValueTooBig> {
-        serialize_list_or_set(self.iter(), self.len(), buf)
-    }
-}
+        let bytes_num_pos: usize = buf.len();
+        buf.put_i32(0);
 
-impl<K: Value, V: Value> Value for BTreeMap<K, V> {
-    fn serialize(&self, buf: &mut Vec<u8>) -> Result<(), ValueTooBig> {
-        serialize_map(self.iter(), self.len(), buf)
-    }
-}
+        buf.put_i32(self.len().try_into().map_err(|_| ValueTooBig)?);
+        for value in self {
+            <T as Value>::serialize(value, buf)?;
+        }
 
-impl<T: Value> Value for Vec<T> {
-    fn serialize(&self, buf: &mut Vec<u8>) -> Result<(), ValueTooBig> {
-        serialize_list_or_set(self.iter(), self.len(), 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(())
     }
 }
 
diff --git a/scylla/src/transport/cluster.rs b/scylla/src/transport/cluster.rs
index 985e59cf..1333327e 100644
--- a/scylla/src/transport/cluster.rs
+++ b/scylla/src/transport/cluster.rs
@@ -5,7 +5,7 @@ use crate::transport::connection::{Connection, VerifiedKeyspaceName};
 use crate::transport::connection_pool::PoolConfig;
 use crate::transport::errors::QueryError;
 use crate::transport::node::Node;
-use crate::transport::topology::{Keyspace, TopologyInfo, TopologyReader};
+use crate::transport::topology::{Keyspace, Metadata, MetadataReader};
 
 use arc_swap::ArcSwap;
 use futures::future::join_all;
@@ -37,11 +37,11 @@ pub struct Datacenter {
 
 #[derive(Clone)]
 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>,
-    pub all_nodes: Vec<Arc<Node>>,
-    pub datacenters: HashMap<String, Datacenter>,
+    pub(crate) known_peers: HashMap<SocketAddr, Arc<Node>>, // Invariant: nonempty after Cluster::new()
+    pub(crate) ring: BTreeMap<Token, Arc<Node>>, // Invariant: nonempty after Cluster::new()
+    pub(crate) keyspaces: HashMap<String, Keyspace>,
+    pub(crate) all_nodes: Vec<Arc<Node>>,
+    pub(crate) datacenters: HashMap<String, Datacenter>,
 }
 
 // Works in the background to keep the cluster updated
@@ -50,7 +50,7 @@ struct ClusterWorker {
     cluster_data: Arc<ArcSwap<ClusterData>>,
 
     // Cluster connections
-    topology_reader: TopologyReader,
+    metadata_reader: MetadataReader,
     pool_config: PoolConfig,
 
     // To listen for refresh requests
@@ -81,6 +81,7 @@ impl Cluster {
     pub async fn new(
         initial_peers: &[SocketAddr],
         pool_config: PoolConfig,
+        fetch_schema_metadata: bool,
     ) -> Result<Cluster, QueryError> {
         let cluster_data = Arc::new(ArcSwap::from(Arc::new(ClusterData {
             known_peers: HashMap::new(),
@@ -97,10 +98,11 @@ impl Cluster {
         let worker = ClusterWorker {
             cluster_data: cluster_data.clone(),
 
-            topology_reader: TopologyReader::new(
+            metadata_reader: MetadataReader::new(
                 initial_peers,
                 pool_config.connection_config.clone(),
                 server_events_sender,
+                fetch_schema_metadata,
             ),
             pool_config,
 
@@ -121,7 +123,7 @@ impl Cluster {
             _worker_handle: worker_handle,
         };
 
-        result.refresh_topology().await?;
+        result.refresh_metadata().await?;
 
         Ok(result)
     }
@@ -130,7 +132,7 @@ impl Cluster {
         self.data.load_full()
     }
 
-    pub async fn refresh_topology(&self) -> Result<(), QueryError> {
+    pub async fn refresh_metadata(&self) -> Result<(), QueryError> {
         let (response_sender, response_receiver) = tokio::sync::oneshot::channel();
 
         self.refresh_channel
@@ -138,12 +140,12 @@ impl Cluster {
                 response_chan: response_sender,
             })
             .await
-            .expect("Bug in Cluster::refresh_topology sending");
+            .expect("Bug in Cluster::refresh_metadata 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")
+            .expect("Bug in Cluster::refresh_metadata receiving")
         // ClusterWorker always responds
     }
 
@@ -192,7 +194,7 @@ impl Cluster {
 impl ClusterData {
     /// Returns an iterator to the sequence of ends of vnodes, starting at the vnode in which t
     /// lies and going clockwise. Returned sequence has the same length as ring.
-    pub fn ring_range<'a>(&'a self, t: &Token) -> impl Iterator<Item = Arc<Node>> + 'a {
+    pub(crate) fn ring_range<'a>(&'a self, t: &Token) -> impl Iterator<Item = Arc<Node>> + 'a {
         let before_wrap = self.ring.range(t..).map(|(_token, node)| node.clone());
         let after_wrap = self.ring.values().cloned();
 
@@ -211,28 +213,28 @@ impl ClusterData {
         }
     }
 
-    pub async fn wait_until_all_pools_are_initialized(&self) {
+    pub(crate) async fn wait_until_all_pools_are_initialized(&self) {
         for node in self.all_nodes.iter() {
             node.wait_until_pool_initialized().await;
         }
     }
 
-    /// Creates new ClusterData using information about topology held in `info`.
+    /// Creates new ClusterData using information about topology held in `metadata`.
     /// Uses provided `known_peers` hashmap to recycle nodes if possible.
-    pub fn new(
-        info: TopologyInfo,
+    pub(crate) fn new(
+        metadata: Metadata,
         pool_config: &PoolConfig,
         known_peers: &HashMap<SocketAddr, Arc<Node>>,
         used_keyspace: &Option<VerifiedKeyspaceName>,
     ) -> Self {
         // Create new updated known_peers and ring
         let mut new_known_peers: HashMap<SocketAddr, Arc<Node>> =
-            HashMap::with_capacity(info.peers.len());
+            HashMap::with_capacity(metadata.peers.len());
         let mut ring: BTreeMap<Token, Arc<Node>> = BTreeMap::new();
         let mut datacenters: HashMap<String, Datacenter> = HashMap::new();
-        let mut all_nodes: Vec<Arc<Node>> = Vec::with_capacity(info.peers.len());
+        let mut all_nodes: Vec<Arc<Node>> = Vec::with_capacity(metadata.peers.len());
 
-        for peer in info.peers {
+        for peer in metadata.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
@@ -276,11 +278,18 @@ impl ClusterData {
         ClusterData {
             known_peers: new_known_peers,
             ring,
-            keyspaces: info.keyspaces,
+            keyspaces: metadata.keyspaces,
             all_nodes,
             datacenters,
         }
     }
+
+    /// Access keyspaces details collected by the driver
+    /// Driver collects various schema details like tables, partitioners, columns, types.
+    /// They can be read using this method
+    pub fn get_keyspace_info(&self) -> &HashMap<String, Keyspace> {
+        &self.keyspaces
+    }
 }
 
 impl ClusterWorker {
@@ -427,12 +436,12 @@ impl ClusterWorker {
     }
 
     async fn perform_refresh(&mut self) -> Result<(), QueryError> {
-        // Read latest TopologyInfo
-        let topo_info = self.topology_reader.read_topology_info().await?;
+        // Read latest Metadata
+        let metadata = self.metadata_reader.read_metadata().await?;
         let cluster_data: Arc<ClusterData> = self.cluster_data.load_full();
 
         let new_cluster_data = Arc::new(ClusterData::new(
-            topo_info,
+            metadata,
             &self.pool_config,
             &cluster_data.known_peers,
             &self.used_keyspace,
diff --git a/scylla/src/transport/iterator.rs b/scylla/src/transport/iterator.rs
index 93d641f0..37d6bcae 100644
--- a/scylla/src/transport/iterator.rs
+++ b/scylla/src/transport/iterator.rs
@@ -18,7 +18,7 @@ use crate::cql_to_rust::{FromRow, FromRowError};
 use crate::frame::{
     response::{
         result,
-        result::{ColumnSpec, Row, Rows},
+        result::{Row, Rows},
         Response,
     },
     value::SerializedValues,
@@ -106,7 +106,7 @@ impl RowIterator {
         }
     }
 
-    pub(crate) async fn new_for_query(
+    pub(crate) fn new_for_query(
         query: Query,
         values: SerializedValues,
         default_consistency: Consistency,
@@ -114,8 +114,8 @@ impl RowIterator {
         load_balancer: Arc<dyn LoadBalancingPolicy>,
         cluster_data: Arc<ClusterData>,
         metrics: Arc<Metrics>,
-    ) -> Result<RowIterator, QueryError> {
-        let (sender, mut receiver) = mpsc::channel(1);
+    ) -> RowIterator {
+        let (sender, receiver) = mpsc::channel(1);
         let consistency = query.config.determine_consistency(default_consistency);
 
         let worker_task = async move {
@@ -146,24 +146,16 @@ impl RowIterator {
 
         tokio::task::spawn(worker_task);
 
-        let pages_received = receiver.recv().await.unwrap()?;
-
-        Ok(RowIterator {
+        RowIterator {
             current_row_idx: 0,
-            current_page: pages_received.rows,
+            current_page: Default::default(),
             page_receiver: receiver,
-            tracing_ids: if let Some(tracing_id) = pages_received.tracing_id {
-                vec![tracing_id]
-            } else {
-                Vec::new()
-            },
-        })
+            tracing_ids: Vec::new(),
+        }
     }
 
-    pub(crate) async fn new_for_prepared_statement(
-        config: PreparedIteratorConfig,
-    ) -> Result<RowIterator, QueryError> {
-        let (sender, mut receiver) = mpsc::channel(1);
+    pub(crate) fn new_for_prepared_statement(config: PreparedIteratorConfig) -> RowIterator {
+        let (sender, receiver) = mpsc::channel(1);
         let consistency = config
             .prepared
             .config
@@ -206,18 +198,12 @@ impl RowIterator {
 
         tokio::task::spawn(worker_task);
 
-        let pages_received = receiver.recv().await.unwrap()?;
-
-        Ok(RowIterator {
+        RowIterator {
             current_row_idx: 0,
-            current_page: pages_received.rows,
+            current_page: Default::default(),
             page_receiver: receiver,
-            tracing_ids: if let Some(tracing_id) = pages_received.tracing_id {
-                vec![tracing_id]
-            } else {
-                Vec::new()
-            },
-        })
+            tracing_ids: Vec::new(),
+        }
     }
 
     /// If tracing was enabled returns tracing ids of all finished page queries
@@ -225,11 +211,6 @@ impl RowIterator {
         &self.tracing_ids
     }
 
-    /// Returns specification of row columns
-    pub fn get_column_specs(&self) -> &[ColumnSpec] {
-        &self.current_page.metadata.col_specs
-    }
-
     fn is_current_page_exhausted(&self) -> bool {
         self.current_row_idx >= self.current_page.rows.len()
     }
@@ -383,11 +364,6 @@ impl<RowT> TypedRowIterator<RowT> {
     pub fn get_tracing_ids(&self) -> &[Uuid] {
         self.row_iterator.get_tracing_ids()
     }
-
-    /// Returns specification of row columns
-    pub fn get_column_specs(&self) -> &[ColumnSpec] {
-        self.row_iterator.get_column_specs()
-    }
 }
 
 /// Couldn't get next typed row from the iterator
diff --git a/scylla/src/transport/load_balancing/mod.rs b/scylla/src/transport/load_balancing/mod.rs
index b107ef55..c558fe16 100644
--- a/scylla/src/transport/load_balancing/mod.rs
+++ b/scylla/src/transport/load_balancing/mod.rs
@@ -84,8 +84,8 @@ impl Default for Statement<'_> {
 mod tests {
     use super::*;
 
+    use crate::transport::topology::Metadata;
     use crate::transport::topology::Peer;
-    use crate::transport::topology::TopologyInfo;
     use std::collections::HashMap;
     use std::net::SocketAddr;
 
@@ -134,7 +134,7 @@ mod tests {
             })
             .collect::<Vec<_>>();
 
-        let info = TopologyInfo {
+        let info = Metadata {
             peers,
             keyspaces: HashMap::new(),
         };
diff --git a/scylla/src/transport/load_balancing/token_aware.rs b/scylla/src/transport/load_balancing/token_aware.rs
index 73ef736c..17270424 100644
--- a/scylla/src/transport/load_balancing/token_aware.rs
+++ b/scylla/src/transport/load_balancing/token_aware.rs
@@ -153,9 +153,9 @@ mod tests {
 
     use crate::transport::load_balancing::tests;
     use crate::transport::topology::Keyspace;
+    use crate::transport::topology::Metadata;
     use crate::transport::topology::Peer;
     use crate::transport::topology::Strategy;
-    use crate::transport::topology::TopologyInfo;
     use std::collections::HashMap;
 
     // ConnectionKeeper (which lives in Node) requires context of Tokio runtime
@@ -287,6 +287,8 @@ mod tests {
                     strategy: Strategy::SimpleStrategy {
                         replication_factor: 2,
                     },
+                    tables: HashMap::new(),
+                    user_defined_types: HashMap::new(),
                 },
             ),
             (
@@ -295,6 +297,8 @@ mod tests {
                     strategy: Strategy::SimpleStrategy {
                         replication_factor: 3,
                     },
+                    tables: HashMap::new(),
+                    user_defined_types: HashMap::new(),
                 },
             ),
         ]
@@ -302,7 +306,7 @@ mod tests {
         .cloned()
         .collect();
 
-        let info = TopologyInfo {
+        let info = Metadata {
             peers: Vec::from(peers),
             keyspaces,
         };
@@ -384,13 +388,15 @@ mod tests {
                         .cloned()
                         .collect::<HashMap<_, _>>(),
                 },
+                tables: HashMap::new(),
+                user_defined_types: HashMap::new(),
             },
         )]
         .iter()
         .cloned()
         .collect();
 
-        let info = TopologyInfo {
+        let info = Metadata {
             peers: Vec::from(peers),
             keyspaces,
         };
diff --git a/scylla/src/transport/mod.rs b/scylla/src/transport/mod.rs
index 450714f7..62cec123 100644
--- a/scylla/src/transport/mod.rs
+++ b/scylla/src/transport/mod.rs
@@ -9,6 +9,7 @@ pub mod session;
 pub mod session_builder;
 pub mod speculative_execution;
 mod topology;
+pub use cluster::ClusterData;
 
 pub mod errors;
 pub mod iterator;
diff --git a/scylla/src/transport/session.rs b/scylla/src/transport/session.rs
index ad467897..671e4b61 100644
--- a/scylla/src/transport/session.rs
+++ b/scylla/src/transport/session.rs
@@ -15,6 +15,7 @@ use uuid::Uuid;
 
 use super::connection::QueryResponse;
 use super::errors::{BadQuery, NewSessionError, QueryError};
+use crate::cql_to_rust::FromRow;
 use crate::frame::response::cql_to_rust::FromRowError;
 use crate::frame::response::result;
 use crate::frame::value::{BatchValues, SerializedValues, ValueList};
@@ -23,24 +24,27 @@ use crate::query::Query;
 use crate::routing::{murmur3_token, Token};
 use crate::statement::{Consistency, SerialConsistency};
 use crate::tracing::{GetTracingConfig, TracingEvent, TracingInfo};
+use crate::transport::cluster::{Cluster, ClusterData};
+use crate::transport::connection::{
+    BatchResult, Connection, ConnectionConfig, QueryResult, VerifiedKeyspaceName,
+};
 use crate::transport::connection_pool::PoolConfig;
-use crate::transport::{
-    cluster::Cluster,
-    connection::{BatchResult, Connection, ConnectionConfig, QueryResult, VerifiedKeyspaceName},
-    iterator::RowIterator,
-    load_balancing::{LoadBalancingPolicy, RoundRobinPolicy, Statement, TokenAwarePolicy},
-    metrics::Metrics,
-    node::Node,
-    retry_policy::{DefaultRetryPolicy, QueryInfo, RetryDecision, RetryPolicy, RetrySession},
-    speculative_execution::SpeculativeExecutionPolicy,
-    Compression,
+use crate::transport::iterator::{PreparedIteratorConfig, RowIterator};
+use crate::transport::load_balancing::{
+    LoadBalancingPolicy, RoundRobinPolicy, Statement, TokenAwarePolicy,
+};
+use crate::transport::metrics::Metrics;
+use crate::transport::node::Node;
+use crate::transport::retry_policy::{
+    DefaultRetryPolicy, QueryInfo, RetryDecision, RetryPolicy, RetrySession,
 };
+use crate::transport::speculative_execution;
+use crate::transport::speculative_execution::SpeculativeExecutionPolicy;
+use crate::transport::Compression;
 use crate::{batch::Batch, statement::StatementConfig};
-use crate::{cql_to_rust::FromRow, transport::speculative_execution};
 
 pub use crate::transport::connection_pool::PoolSize;
 
-use crate::transport::iterator::PreparedIteratorConfig;
 #[cfg(feature = "ssl")]
 use openssl::ssl::SslContext;
 
@@ -98,6 +102,9 @@ pub struct SessionConfig {
     pub disallow_shard_aware_port: bool,
 
     pub default_consistency: Consistency,
+
+    /// If true, full schema is fetched with every metadata refresh.
+    pub fetch_schema_metadata: bool,
     /*
     These configuration options will be added in the future:
 
@@ -143,6 +150,7 @@ impl SessionConfig {
             connection_pool_size: Default::default(),
             disallow_shard_aware_port: false,
             default_consistency: Consistency::LocalQuorum,
+            fetch_schema_metadata: true,
         }
     }
 
@@ -317,7 +325,12 @@ impl Session {
 
         node_addresses.extend(resolved);
 
-        let cluster = Cluster::new(&node_addresses, config.get_pool_config()).await?;
+        let cluster = Cluster::new(
+            &node_addresses,
+            config.get_pool_config(),
+            config.fetch_schema_metadata,
+        )
+        .await?;
 
         let session = Session {
             cluster,
@@ -493,7 +506,7 @@ impl Session {
             None => self.retry_policy.new_session(),
         };
 
-        RowIterator::new_for_query(
+        Ok(RowIterator::new_for_query(
             query,
             serialized_values.into_owned(),
             self.default_consistency,
@@ -501,8 +514,7 @@ impl Session {
             self.load_balancer.clone(),
             self.cluster.get_data(),
             self.metrics.clone(),
-        )
-        .await
+        ))
     }
 
     /// Prepares a statement on the server side and returns a prepared statement,
@@ -724,17 +736,18 @@ impl Session {
             None => self.retry_policy.new_session(),
         };
 
-        RowIterator::new_for_prepared_statement(PreparedIteratorConfig {
-            prepared,
-            values: serialized_values.into_owned(),
-            default_consistency: self.default_consistency,
-            token,
-            retry_session,
-            load_balancer: self.load_balancer.clone(),
-            cluster_data: self.cluster.get_data(),
-            metrics: self.metrics.clone(),
-        })
-        .await
+        Ok(RowIterator::new_for_prepared_statement(
+            PreparedIteratorConfig {
+                prepared,
+                values: serialized_values.into_owned(),
+                default_consistency: self.default_consistency,
+                token,
+                retry_session,
+                load_balancer: self.load_balancer.clone(),
+                cluster_data: self.cluster.get_data(),
+                metrics: self.metrics.clone(),
+            },
+        ))
     }
 
     /// Perform a batch query\
@@ -844,13 +857,13 @@ impl Session {
         Ok(())
     }
 
-    /// Manually trigger a topology refresh\
-    /// The driver will fetch current nodes in the cluster and update its topology information
+    /// Manually trigger a metadata refresh\
+    /// The driver will fetch current nodes in the cluster and update its metadata
     ///
     /// Normally this is not needed,
-    /// the driver should automatically detect all topology changes in the cluster
-    pub async fn refresh_topology(&self) -> Result<(), QueryError> {
-        self.cluster.refresh_topology().await
+    /// the driver should automatically detect all metadata changes in the cluster
+    pub async fn refresh_metadata(&self) -> Result<(), QueryError> {
+        self.cluster.refresh_metadata().await
     }
 
     /// Access metrics collected by the driver\
@@ -860,6 +873,13 @@ impl Session {
         self.metrics.clone()
     }
 
+    /// Access cluster data collected by the driver\
+    /// Driver collects various information about network topology or schema.
+    /// They can be read using this method
+    pub fn get_cluster_data(&self) -> Arc<ClusterData> {
+        self.cluster.get_data()
+    }
+
     /// Get [`TracingInfo`] of a traced query performed earlier
     ///
     /// See [the book](https://cvybhu.github.io/scyllabook/tracing/tracing.html)
diff --git a/scylla/src/transport/session_builder.rs b/scylla/src/transport/session_builder.rs
index dd251606..2b852326 100644
--- a/scylla/src/transport/session_builder.rs
+++ b/scylla/src/transport/session_builder.rs
@@ -482,6 +482,26 @@ impl SessionBuilder {
         self.config.disallow_shard_aware_port = disallow;
         self
     }
+
+    /// Set the fetch schema metadata flag.
+    /// The default is true.
+    ///
+    /// # Example
+    /// ```
+    /// # use scylla::{Session, SessionBuilder};
+    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
+    /// let session: Session = SessionBuilder::new()
+    ///     .known_node("127.0.0.1:9042")
+    ///     .fetch_schema_metadata(true)
+    ///     .build()
+    ///     .await?;
+    /// # Ok(())
+    /// # }
+    /// ```
+    pub fn fetch_schema_metadata(mut self, fetch: bool) -> Self {
+        self.config.fetch_schema_metadata = fetch;
+        self
+    }
 }
 
 /// Creates a [`SessionBuilder`] with default configuration, same as [`SessionBuilder::new`]
@@ -636,6 +656,18 @@ mod tests {
         );
     }
 
+    #[test]
+    fn fetch_schema_metadata() {
+        let mut builder = SessionBuilder::new();
+        assert!(builder.config.fetch_schema_metadata);
+
+        builder = builder.fetch_schema_metadata(false);
+        assert!(!builder.config.fetch_schema_metadata);
+
+        builder = builder.fetch_schema_metadata(true);
+        assert!(builder.config.fetch_schema_metadata);
+    }
+
     #[test]
     fn all_features() {
         let mut builder = SessionBuilder::new();
@@ -652,6 +684,7 @@ mod tests {
         builder = builder.tcp_nodelay(true);
         builder = builder.load_balancing(Arc::new(RoundRobinPolicy::new()));
         builder = builder.use_keyspace("ks_name", true);
+        builder = builder.fetch_schema_metadata(false);
 
         assert_eq!(
             builder.config.known_nodes,
@@ -675,5 +708,6 @@ mod tests {
         assert_eq!(builder.config.used_keyspace, Some("ks_name".to_string()));
 
         assert!(builder.config.keyspace_case_sensitive);
+        assert!(!builder.config.fetch_schema_metadata);
     }
 }
diff --git a/scylla/src/transport/topology.rs b/scylla/src/transport/topology.rs
index f144daf3..53755aa3 100644
--- a/scylla/src/transport/topology.rs
+++ b/scylla/src/transport/topology.rs
@@ -1,32 +1,37 @@
 use crate::frame::response::event::Event;
 use crate::routing::Token;
+use crate::statement::query::Query;
 use crate::transport::connection::{Connection, ConnectionConfig};
 use crate::transport::connection_pool::{NodeConnectionPool, PoolConfig, PoolSize};
-use crate::transport::errors::QueryError;
+use crate::transport::errors::{DbError, QueryError};
 use crate::transport::session::IntoTypedRows;
 
-use crate::statement::query::Query;
 use rand::seq::SliceRandom;
 use rand::{thread_rng, Rng};
+use std::borrow::BorrowMut;
 use std::collections::HashMap;
+use std::fmt;
+use std::fmt::Formatter;
 use std::net::{IpAddr, SocketAddr};
 use std::num::NonZeroUsize;
 use std::str::FromStr;
+use strum_macros::EnumString;
 use tokio::sync::mpsc;
 use tracing::{debug, error, trace, warn};
 
-/// Allows to read current topology info from the cluster
-pub struct TopologyReader {
+/// Allows to read current metadata from the cluster
+pub struct MetadataReader {
     connection_config: ConnectionConfig,
     control_connection_address: SocketAddr,
     control_connection: NodeConnectionPool,
 
-    // when control connection fails, TopologyReader tries to connect to one of known_peers
+    // when control connection fails, MetadataReader tries to connect to one of known_peers
     known_peers: Vec<SocketAddr>,
+    fetch_schema: bool,
 }
 
-/// Describes all topology information retrieved from the cluster
-pub struct TopologyInfo {
+/// Describes all metadata retrieved from the cluster
+pub struct Metadata {
     pub peers: Vec<Peer>,
     pub keyspaces: HashMap<String, Keyspace>,
 }
@@ -41,6 +46,73 @@ pub struct Peer {
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub struct Keyspace {
     pub strategy: Strategy,
+    /// Empty HashMap may as well mean that the client disabled schema fetching in SessionConfig
+    pub tables: HashMap<String, Table>,
+    /// Empty HashMap may as well mean that the client disabled schema fetching in SessionConfig
+    pub user_defined_types: HashMap<String, Vec<(String, CqlType)>>,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct Table {
+    pub columns: HashMap<String, Column>,
+    pub partition_key: Vec<String>,
+    pub clustering_key: Vec<String>,
+    pub partitioner: Option<String>,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct Column {
+    pub type_: CqlType,
+    pub kind: ColumnKind,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub enum CqlType {
+    Native(NativeType),
+    Collection { frozen: bool, type_: CollectionType },
+    Tuple(Vec<CqlType>),
+    UserDefinedType { frozen: bool, name: String },
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, EnumString)]
+#[strum(serialize_all = "lowercase")]
+pub enum NativeType {
+    Ascii,
+    Boolean,
+    Blob,
+    Counter,
+    Date,
+    Decimal,
+    Double,
+    Duration,
+    Float,
+    Int,
+    BigInt,
+    Text,
+    Timestamp,
+    Inet,
+    SmallInt,
+    TinyInt,
+    Time,
+    Timeuuid,
+    Uuid,
+    Varint,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub enum CollectionType {
+    List(Box<CqlType>),
+    Map(Box<CqlType>, Box<CqlType>),
+    Set(Box<CqlType>),
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, EnumString)]
+#[strum(serialize_all = "snake_case")]
+pub enum ColumnKind {
+    Regular,
+    Static,
+    Clustering,
+    PartitionKey,
 }
 
 #[derive(Clone, Debug, PartialEq, Eq)]
@@ -60,16 +132,33 @@ pub enum Strategy {
     },
 }
 
-impl TopologyReader {
-    /// Creates new TopologyReader, which connects to known_peers in the background
+#[derive(Clone, Debug)]
+struct InvalidCqlType(String);
+
+impl fmt::Display for InvalidCqlType {
+    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
+        write!(f, "{:?}", self)
+    }
+}
+
+impl From<InvalidCqlType> for QueryError {
+    fn from(e: InvalidCqlType) -> Self {
+        // FIXME: The correct error type is QueryError:ProtocolError but at the moment it accepts only &'static str
+        QueryError::InvalidMessage(format!("type {} is not implemented", e.0))
+    }
+}
+
+impl MetadataReader {
+    /// Creates new MetadataReader, which connects to known_peers in the background
     pub fn new(
         known_peers: &[SocketAddr],
         mut connection_config: ConnectionConfig,
         server_event_sender: mpsc::Sender<Event>,
+        fetch_schema: bool,
     ) -> Self {
         let control_connection_address = *known_peers
             .choose(&mut thread_rng())
-            .expect("Tried to initialize TopologyReader with empty known_peers list!");
+            .expect("Tried to initialize MetadataReader with empty known_peers list!");
 
         // setting event_sender field in connection config will cause control connection to
         // - send REGISTER message to receive server events
@@ -81,20 +170,21 @@ impl TopologyReader {
             connection_config.clone(),
         );
 
-        TopologyReader {
+        MetadataReader {
             control_connection_address,
             control_connection,
             connection_config,
             known_peers: known_peers.into(),
+            fetch_schema,
         }
     }
 
-    /// Fetches current topology info from the cluster
-    pub async fn read_topology_info(&mut self) -> Result<TopologyInfo, QueryError> {
-        let mut result = self.fetch_topology_info().await;
-        if let Ok(topology_info) = result {
-            self.update_known_peers(&topology_info);
-            return Ok(topology_info);
+    /// Fetches current metadata from the cluster
+    pub async fn read_metadata(&mut self) -> Result<Metadata, QueryError> {
+        let mut result = self.fetch_metadata().await;
+        if let Ok(metadata) = result {
+            self.update_known_peers(&metadata);
+            return Ok(metadata);
         }
 
         // shuffle known_peers to iterate through them in random order later
@@ -106,8 +196,8 @@ impl TopologyReader {
             .iter()
             .filter(|&peer| peer != &address_of_failed_control_connection);
 
-        // if fetching topology info on current control connection failed,
-        // try to fetch topology info from other known peer
+        // if fetching metadata on current control connection failed,
+        // try to fetch metadata from other known peer
         for peer in filtered_known_peers {
             let err = match result {
                 Ok(_) => break,
@@ -117,7 +207,7 @@ impl TopologyReader {
             warn!(
                 control_connection_address = self.control_connection_address.to_string().as_str(),
                 error = err.to_string().as_str(),
-                "Falied to fetch topology info using current control connection"
+                "Failed to fetch metadata using current control connection"
             );
 
             self.control_connection_address = *peer;
@@ -126,39 +216,36 @@ impl TopologyReader {
                 self.connection_config.clone(),
             );
 
-            result = self.fetch_topology_info().await;
+            result = self.fetch_metadata().await;
         }
 
         match &result {
-            Ok(topology_info) => {
-                self.update_known_peers(topology_info);
-                debug!("Fetched new topology info");
+            Ok(metadata) => {
+                self.update_known_peers(metadata);
+                debug!("Fetched new metadata");
             }
             Err(error) => error!(
                 error = error.to_string().as_str(),
-                "Could not fetch topology info"
+                "Could not fetch metadata"
             ),
         }
 
         result
     }
 
-    async fn fetch_topology_info(&self) -> Result<TopologyInfo, QueryError> {
+    async fn fetch_metadata(&self) -> Result<Metadata, QueryError> {
         // TODO: Timeouts?
 
-        query_topology_info(
+        query_metadata(
             &self.control_connection,
             self.control_connection_address.port(),
+            self.fetch_schema,
         )
         .await
     }
 
-    fn update_known_peers(&mut self, topology_info: &TopologyInfo) {
-        self.known_peers = topology_info
-            .peers
-            .iter()
-            .map(|peer| peer.address)
-            .collect();
+    fn update_known_peers(&mut self, metadata: &Metadata) {
+        self.known_peers = metadata.peers.iter().map(|peer| peer.address).collect();
     }
 
     fn make_control_connection_pool(
@@ -180,33 +267,34 @@ impl TopologyReader {
     }
 }
 
-async fn query_topology_info(
+async fn query_metadata(
     pool: &NodeConnectionPool,
     connect_port: u16,
-) -> Result<TopologyInfo, QueryError> {
+    fetch_schema: bool,
+) -> Result<Metadata, QueryError> {
     pool.wait_until_initialized().await;
     let conn: &Connection = &*pool.random_connection()?;
 
     let peers_query = query_peers(conn, connect_port);
-    let keyspaces_query = query_keyspaces(conn);
+    let keyspaces_query = query_keyspaces(conn, fetch_schema);
 
     let (peers, keyspaces) = tokio::try_join!(peers_query, keyspaces_query)?;
 
     // There must be at least one peer
     if peers.is_empty() {
         return Err(QueryError::ProtocolError(
-            "Bad TopologyInfo: peers list is empty",
+            "Bad Metadata: peers list is empty",
         ));
     }
 
     // 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",
+            "Bad Metadata: All peers have empty token list",
         ));
     }
 
-    Ok(TopologyInfo { peers, keyspaces })
+    Ok(Metadata { peers, keyspaces })
 }
 
 async fn query_peers(conn: &Connection, connect_port: u16) -> Result<Vec<Peer>, QueryError> {
@@ -279,7 +367,10 @@ async fn query_peers(conn: &Connection, connect_port: u16) -> Result<Vec<Peer>,
     Ok(result)
 }
 
-async fn query_keyspaces(conn: &Connection) -> Result<HashMap<String, Keyspace>, QueryError> {
+async fn query_keyspaces(
+    conn: &Connection,
+    fetch_schema: bool,
+) -> Result<HashMap<String, Keyspace>, QueryError> {
     let mut keyspaces_query =
         Query::new("select keyspace_name, replication from system_schema.keyspaces");
     keyspaces_query.set_page_size(1024);
@@ -293,6 +384,14 @@ async fn query_keyspaces(conn: &Connection) -> Result<HashMap<String, Keyspace>,
             ))?;
 
     let mut result = HashMap::with_capacity(rows.len());
+    let (mut all_tables, mut all_user_defined_types) = if fetch_schema {
+        (
+            query_tables(conn).await?,
+            query_user_defined_types(conn).await?,
+        )
+    } else {
+        (HashMap::new(), HashMap::new())
+    };
 
     for row in rows.into_typed::<(String, HashMap<String, String>)>() {
         let (keyspace_name, strategy_map) = row.map_err(|_| {
@@ -300,13 +399,285 @@ async fn query_keyspaces(conn: &Connection) -> Result<HashMap<String, Keyspace>,
         })?;
 
         let strategy: Strategy = strategy_from_string_map(strategy_map)?;
+        let tables = all_tables.remove(&keyspace_name).unwrap_or_default();
+        let user_defined_types = all_user_defined_types
+            .remove(&keyspace_name)
+            .unwrap_or_default();
+
+        result.insert(
+            keyspace_name,
+            Keyspace {
+                strategy,
+                tables,
+                user_defined_types,
+            },
+        );
+    }
+
+    Ok(result)
+}
+
+async fn query_user_defined_types(
+    conn: &Connection,
+) -> Result<HashMap<String, HashMap<String, Vec<(String, CqlType)>>>, QueryError> {
+    let mut user_defined_types_query = Query::new(
+        "select keyspace_name, type_name, field_names, field_types from system_schema.types",
+    );
+    user_defined_types_query.set_page_size(1024);
+
+    let rows = conn
+        .query_all(&user_defined_types_query, &[])
+        .await?
+        .rows
+        .ok_or(QueryError::ProtocolError(
+            "system_schema.types query response was not Rows",
+        ))?;
+
+    let mut result = HashMap::with_capacity(rows.len());
+
+    for row in rows.into_typed::<(String, String, Vec<String>, Vec<String>)>() {
+        let (keyspace_name, type_name, field_names, field_types) = row.map_err(|_| {
+            QueryError::ProtocolError("system_schema.types has invalid column type")
+        })?;
+
+        let mut fields = Vec::with_capacity(field_names.len());
+
+        for (field_name, field_type) in field_names.into_iter().zip(field_types.iter()) {
+            fields.push((field_name, map_string_to_cql_type(field_type)?));
+        }
 
-        result.insert(keyspace_name, Keyspace { strategy });
+        result
+            .entry(keyspace_name)
+            .or_insert_with(HashMap::new)
+            .insert(type_name, fields);
     }
 
     Ok(result)
 }
 
+async fn query_tables(
+    conn: &Connection,
+) -> Result<HashMap<String, HashMap<String, Table>>, QueryError> {
+    let mut tables_query = Query::new("select keyspace_name, table_name from system_schema.tables");
+    tables_query.set_page_size(1024);
+
+    let rows = conn
+        .query_all(&tables_query, &[])
+        .await?
+        .rows
+        .ok_or(QueryError::ProtocolError(
+            "system_schema.tables query response was not Rows",
+        ))?;
+
+    let mut result = HashMap::with_capacity(rows.len());
+    let mut tables = query_tables_schema(conn).await?;
+
+    for row in rows.into_typed::<(String, String)>() {
+        let (keyspace_name, table_name) = row.map_err(|_| {
+            QueryError::ProtocolError("system_schema.tables has invalid column type")
+        })?;
+
+        let keyspace_and_table_name = (keyspace_name, table_name);
+
+        let table = tables.remove(&keyspace_and_table_name).unwrap_or(Table {
+            columns: HashMap::new(),
+            partition_key: vec![],
+            clustering_key: vec![],
+            partitioner: None,
+        });
+
+        result
+            .entry(keyspace_and_table_name.0)
+            .or_insert_with(HashMap::new)
+            .insert(keyspace_and_table_name.1, table);
+    }
+
+    Ok(result)
+}
+
+async fn query_tables_schema(
+    conn: &Connection,
+) -> Result<HashMap<(String, String), Table>, QueryError> {
+    // Upon migration from thrift to CQL, Cassandra internally creates a surrogate column "value" of
+    // type EmptyType for dense tables. This resolves into this CQL type name.
+    // This column shouldn't be exposed to the user but is currently exposed in system tables.
+    const THRIFT_EMPTY_TYPE: &str = "empty";
+
+    let mut columns_query = Query::new(
+        "select keyspace_name, table_name, column_name, kind, position, type from system_schema.columns",
+    );
+    columns_query.set_page_size(1024);
+
+    let rows = conn
+        .query_all(&columns_query, &[])
+        .await?
+        .rows
+        .ok_or(QueryError::ProtocolError(
+            "system_schema.columns query response was not Rows",
+        ))?;
+
+    let mut tables_schema = HashMap::with_capacity(rows.len());
+
+    for row in rows.into_typed::<(String, String, String, String, i32, String)>() {
+        let (keyspace_name, table_name, column_name, kind, position, type_) =
+            row.map_err(|_| {
+                QueryError::ProtocolError("system_schema.columns has invalid column type")
+            })?;
+
+        if type_ == THRIFT_EMPTY_TYPE {
+            continue;
+        }
+
+        let entry = tables_schema.entry((keyspace_name, table_name)).or_insert((
+            HashMap::new(), // columns
+            HashMap::new(), // partition key
+            HashMap::new(), // clustering key
+        ));
+
+        let cql_type = map_string_to_cql_type(&type_)?;
+
+        let kind = ColumnKind::from_str(&kind)
+            // FIXME: The correct error type is QueryError:ProtocolError but at the moment it accepts only &'static str
+            .map_err(|_| QueryError::InvalidMessage(format!("invalid column kind {}", kind)))?;
+
+        if kind == ColumnKind::PartitionKey || kind == ColumnKind::Clustering {
+            let key_map = if kind == ColumnKind::PartitionKey {
+                entry.1.borrow_mut()
+            } else {
+                entry.2.borrow_mut()
+            };
+            key_map.insert(position, column_name.clone());
+        }
+
+        entry.0.insert(
+            column_name,
+            Column {
+                type_: cql_type,
+                kind,
+            },
+        );
+    }
+
+    let mut all_partitioners = query_table_partitioners(conn).await?;
+    let mut result = HashMap::new();
+
+    for ((keyspace_name, table_name), (columns, partition_key_columns, clustering_key_columns)) in
+        tables_schema
+    {
+        let mut partition_key = vec!["".to_string(); partition_key_columns.len()];
+        for (position, column_name) in partition_key_columns {
+            partition_key[position as usize] = column_name;
+        }
+
+        let mut clustering_key = vec!["".to_string(); clustering_key_columns.len()];
+        for (position, column_name) in clustering_key_columns {
+            clustering_key[position as usize] = column_name;
+        }
+
+        let keyspace_and_table_name = (keyspace_name, table_name);
+
+        let partitioner = all_partitioners
+            .remove(&keyspace_and_table_name)
+            .unwrap_or_default();
+
+        result.insert(
+            keyspace_and_table_name,
+            Table {
+                columns,
+                partition_key,
+                clustering_key,
+                partitioner,
+            },
+        );
+    }
+
+    Ok(result)
+}
+
+fn map_string_to_cql_type(type_: &str) -> Result<CqlType, InvalidCqlType> {
+    use CollectionType::*;
+
+    let frozen = type_.starts_with("frozen<");
+
+    let type_ = if frozen {
+        let langle_position = type_.find('<').unwrap();
+        &type_[langle_position + 1..type_.len() - 1]
+    } else {
+        type_
+    };
+
+    if type_.ends_with('>') && type_.contains('<') {
+        let langle_position = type_.find('<').unwrap();
+        let inner_type = &type_[langle_position + 1..type_.len() - 1];
+        let outer_type = &type_[..langle_position];
+        let type_ = match outer_type {
+            "map" | "tuple" => {
+                let types = inner_type.split(", ").collect::<Vec<_>>();
+                if outer_type == "map" {
+                    if types.len() != 2 {
+                        return Err(InvalidCqlType(type_.to_string()));
+                    }
+                    Map(
+                        Box::new(map_string_to_cql_type(types[0])?),
+                        Box::new(map_string_to_cql_type(types[1])?),
+                    )
+                } else {
+                    return Ok(CqlType::Tuple(
+                        types
+                            .iter()
+                            .map(|type_| map_string_to_cql_type(type_))
+                            .collect::<Result<Vec<_>, _>>()?,
+                    ));
+                }
+            }
+            "set" => Set(Box::new(map_string_to_cql_type(inner_type)?)),
+            "list" => List(Box::new(map_string_to_cql_type(inner_type)?)),
+            _ => {
+                return Err(InvalidCqlType(type_.to_string()));
+            }
+        };
+        return Ok(CqlType::Collection { frozen, type_ });
+    }
+
+    Ok(match NativeType::from_str(type_) {
+        Ok(type_) => CqlType::Native(type_),
+        _ => CqlType::UserDefinedType {
+            frozen,
+            name: type_.to_string(),
+        },
+    })
+}
+
+async fn query_table_partitioners(
+    conn: &Connection,
+) -> Result<HashMap<(String, String), Option<String>>, QueryError> {
+    let mut partitioner_query = Query::new(
+        "select keyspace_name, table_name, partitioner from system_schema.scylla_tables",
+    );
+    partitioner_query.set_page_size(1024);
+
+    let rows = match conn.query_all(&partitioner_query, &[]).await {
+        // FIXME: This match catches all database errors with this error code despite the fact
+        // that we are only interested in the ones resulting from non-existent table
+        // system_schema.scylla_tables.
+        // For more information please refer to https://github.com/scylladb/scylla-rust-driver/pull/349#discussion_r762050262
+        Err(QueryError::DbError(DbError::Invalid, _)) => return Ok(HashMap::new()),
+        query_result => query_result?.rows.ok_or(QueryError::ProtocolError(
+            "system_schema.scylla_tables query response was not Rows",
+        ))?,
+    };
+
+    let mut result = HashMap::with_capacity(rows.len());
+
+    for row in rows.into_typed::<(String, String, Option<String>)>() {
+        let (keyspace_name, table_name, partitioner) = row.map_err(|_| {
+            QueryError::ProtocolError("system_schema.tables has invalid column type")
+        })?;
+        result.insert((keyspace_name, table_name), partitioner);
+    }
+    Ok(result)
+}
+
 fn strategy_from_string_map(
     mut strategy_map: HashMap<String, String>,
 ) -> Result<Strategy, QueryError> {


Test patch:
diff --git a/scylla/src/frame/cql_collections_test.rs b/scylla/src/frame/cql_collections_test.rs
index 42ba857c..0968bda9 100644
--- a/scylla/src/frame/cql_collections_test.rs
+++ b/scylla/src/frame/cql_collections_test.rs
@@ -1,227 +1,211 @@
-use crate::cql_to_rust::FromCqlVal;
-use crate::frame::value::Value;
-use crate::{frame::response::result::CqlValue, IntoTypedRows, Session, SessionBuilder};
-use std::{
-    collections::{BTreeMap, BTreeSet, HashMap, HashSet},
-    env,
-};
-
-async fn connect() -> Session {
+use crate::{frame::response::result::CqlValue, IntoTypedRows, SessionBuilder};
+use std::{collections::HashMap, env};
+
+#[tokio::test]
+async fn test_cql_collections() {
+    // Create connection
     let uri = env::var("SCYLLA_URI").unwrap_or_else(|_| "127.0.0.1:9042".to_string());
     let session = SessionBuilder::new().known_node(uri).build().await.unwrap();
+
     session.query("CREATE KEYSPACE IF NOT EXISTS ks WITH REPLICATION = {'class' : 'SimpleStrategy', 'replication_factor' : 1}", &[]).await.unwrap();
 
-    session
-}
+    // HashMap test
 
-async fn create_table(session: &Session, table_name: &str, value_type: &str) {
     session
         .query(
-            format!(
-                "CREATE TABLE IF NOT EXISTS ks.{} (p int PRIMARY KEY, val {})",
-                table_name, value_type
-            ),
-            (),
+            "CREATE TABLE IF NOT EXISTS ks.teams (id int, team map<int, text>, primary key (id))",
+            &[],
         )
         .await
         .unwrap();
-}
 
-async fn insert_and_select<InsertT, SelectT>(
-    session: &Session,
-    table_name: &str,
-    to_insert: &InsertT,
-    expected: &SelectT,
-) where
-    InsertT: Value,
-    SelectT: FromCqlVal<Option<CqlValue>> + PartialEq + std::fmt::Debug,
-{
+    let team1: HashMap<i32, String> = vec![(1, "Ginny".to_owned()), (2, "Cho".to_owned())]
+        .into_iter()
+        .collect();
+    // Test CqlValue serialization too
+    let team2: CqlValue = CqlValue::Map(vec![
+        (CqlValue::Int(1), CqlValue::Text("John".into())),
+        (CqlValue::Int(2), CqlValue::Text("Jessica".into())),
+    ]);
+
+    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();
+
+    let mut received_kv_pairs: Vec<Vec<(i32, String)>> = session
+        .query("SELECT team FROM ks.teams", &[])
+        .await
+        .unwrap()
+        .rows
+        .unwrap()
+        .into_typed::<(HashMap<i32, String>,)>()
+        .map(Result::unwrap)
+        .map(|(hashmap,)| {
+            let mut v: Vec<_> = hashmap.into_iter().collect();
+            v.sort_unstable();
+            v
+        })
+        .collect();
+    received_kv_pairs.sort_unstable();
+
+    assert_eq!(
+        received_kv_pairs,
+        vec![
+            vec![(1, "Ginny".into()), (2, "Cho".into())],
+            vec![(1, "John".into()), (2, "Jessica".into())]
+        ]
+    );
+
+    // Set test
+
+    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: CqlValue = CqlValue::Set(vec![
+        CqlValue::Text("kitten".into()),
+        CqlValue::Text("cat".into()),
+        CqlValue::Text("lol".into()),
+    ]);
+
     session
         .query(
-            format!("INSERT INTO ks.{} (p, val) VALUES (0, ?)", table_name),
-            (&to_insert,),
+            "INSERT INTO ks.images (name, owner, tags) VALUES ('cat.jpg', 'jsmith', ?)",
+            (tags1,),
+        )
+        .await
+        .unwrap();
+    session
+        .query(
+            "UPDATE ks.images SET tags = ? WHERE name = 'cat.jpg';",
+            (tags2,),
         )
         .await
         .unwrap();
 
-    let selected_value: SelectT = session
-        .query(format!("SELECT val FROM ks.{} WHERE p = 0", table_name), ())
+    let mut received_elements: Vec<String> = session
+        .query("SELECT tags FROM ks.images", &[])
         .await
         .unwrap()
         .rows
         .unwrap()
-        .into_typed::<(SelectT,)>()
-        .next()
+        .into_typed::<(Vec<String>,)>()
+        .map(Result::unwrap)
+        .map(|(v,)| v.into_iter())
+        .flatten()
+        .collect();
+    received_elements.sort_unstable();
+
+    assert_eq!(
+        received_elements,
+        vec!["cat".to_string(), "kitten".to_string(), "lol".to_string()]
+    );
+
+    // List test
+
+    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];
+    let scores2: CqlValue =
+        CqlValue::List(vec![CqlValue::Int(3), CqlValue::Int(9), CqlValue::Int(4)]);
+
+    session
+        .query(
+            "INSERT INTO ks.plays (id, game, players, scores) VALUES ('123-afde', 'quake', 3, ?)",
+            (&scores1,),
+        )
+        .await
+        .unwrap();
+    session
+        .query(
+            "INSERT INTO ks.plays (id, game, players, scores) VALUES ('1234-qwer', 'cs', 10, ?)",
+            (&scores2,),
+        )
+        .await
+        .unwrap();
+
+    let mut received_elements: Vec<Vec<i32>> = session
+        .query("SELECT scores FROM ks.plays", &[])
+        .await
         .unwrap()
+        .rows
         .unwrap()
-        .0;
+        .into_typed::<(Vec<i32>,)>()
+        .map(Result::unwrap)
+        .map(|(v,)| v.into_iter().collect())
+        .collect();
+    received_elements.sort_unstable();
 
-    assert_eq!(&selected_value, expected);
-}
+    assert_eq!(received_elements, vec![vec![3, 9, 4], vec![17, 4, 2]]);
 
-#[tokio::test]
-async fn test_cql_list() {
-    let session: Session = connect().await;
-
-    let table_name: &str = "test_cql_list_tab";
-    create_table(&session, table_name, "list<int>").await;
-
-    // Vec
-    let list1: Vec<i32> = vec![-1, 0, 1, 1, 2];
-    let list2: Vec<i32> = vec![100, 212, 2323];
-    let list_empty: Vec<i32> = vec![];
-    let list_empty_selected: Option<Vec<i32>> = None;
-    insert_and_select(&session, table_name, &list1, &list1).await;
-    insert_and_select(&session, table_name, &list2, &list2).await;
-    insert_and_select(&session, table_name, &list_empty, &list_empty_selected).await;
-
-    // CqlValue
-    let list_cql_value: CqlValue =
-        CqlValue::List(vec![CqlValue::Int(-1), CqlValue::Int(1), CqlValue::Int(0)]);
-    let list_cql_value_empty: CqlValue = CqlValue::List(Vec::new());
-    let list_cql_value_empty_selected: Option<CqlValue> = None;
-
-    insert_and_select(&session, table_name, &list_cql_value, &list_cql_value).await;
-    insert_and_select(
-        &session,
-        table_name,
-        &list_cql_value_empty,
-        &list_cql_value_empty_selected,
-    )
-    .await;
-}
+    // Tuple test
 
-#[tokio::test]
-async fn test_cql_set() {
-    let session: Session = connect().await;
-
-    let table_name: &str = "test_cql_set_tab";
-    create_table(&session, table_name, "set<int>").await;
-
-    // Vec
-    let set_vec_sorted: Vec<i32> = vec![-1, 0, 1, 2];
-    let set_vec_unordered: Vec<i32> = vec![1, -1, 1, 2, 0];
-    let set_vec_empty: Vec<i32> = vec![];
-    let set_vec_empty_selected: Option<Vec<i32>> = None;
-    insert_and_select(&session, table_name, &set_vec_sorted, &set_vec_sorted).await;
-    insert_and_select(&session, table_name, &set_vec_unordered, &set_vec_sorted).await;
-    insert_and_select(
-        &session,
-        table_name,
-        &set_vec_empty,
-        &set_vec_empty_selected,
-    )
-    .await;
-
-    // HashSet
-    let set_hashset: HashSet<i32> = vec![-1, 1, -2].into_iter().collect();
-    let set_hashset_empty: HashSet<i32> = HashSet::new();
-    let set_hashset_empty_selected: Option<HashSet<i32>> = None;
-    insert_and_select(&session, table_name, &set_hashset, &set_hashset).await;
-    insert_and_select(
-        &session,
-        table_name,
-        &set_hashset_empty,
-        &set_hashset_empty_selected,
-    )
-    .await;
-
-    // BTreeSet
-    let set_btreeset: BTreeSet<i32> = vec![0, -2, -1].into_iter().collect();
-    let set_btreeset_empty: BTreeSet<i32> = BTreeSet::new();
-    let set_btreeset_empty_selected: Option<BTreeSet<i32>> = None;
-    insert_and_select(&session, table_name, &set_btreeset, &set_btreeset).await;
-    insert_and_select(
-        &session,
-        table_name,
-        &set_btreeset_empty,
-        &set_btreeset_empty_selected,
-    )
-    .await;
-
-    // CqlValue
-    let set_cql_value: CqlValue =
-        CqlValue::Set(vec![CqlValue::Int(-1), CqlValue::Int(1), CqlValue::Int(2)]);
-    let set_cql_value_empty: CqlValue = CqlValue::Set(Vec::new());
-    let set_cql_value_empty_selected: Option<CqlValue> = None;
-    insert_and_select(&session, table_name, &set_cql_value, &set_cql_value).await;
-    insert_and_select(
-        &session,
-        table_name,
-        &set_cql_value_empty,
-        &set_cql_value_empty_selected,
-    )
-    .await;
-}
+    session
+        .query(
+            "CREATE TABLE IF NOT EXISTS ks.durations (event text, duration tuple<int, text>, primary key (event))",
+            &[],
+        )
+        .await.unwrap();
 
-#[tokio::test]
-async fn test_cql_map() {
-    let session: Session = connect().await;
-
-    let table_name: &str = "test_cql_map_tab";
-    create_table(&session, table_name, "map<int, int>").await;
-
-    // HashMap
-    let map_hashmap: HashMap<i32, i32> = vec![(-1, 0), (0, 1), (2, 1)].into_iter().collect();
-    let map_hashmap_empty: HashMap<i32, i32> = HashMap::new();
-    let map_hashmap_empty_selected: Option<HashMap<i32, i32>> = None;
-    insert_and_select(&session, table_name, &map_hashmap, &map_hashmap).await;
-    insert_and_select(
-        &session,
-        table_name,
-        &map_hashmap_empty,
-        &map_hashmap_empty_selected,
-    )
-    .await;
-
-    // BTreeMap
-    let map_btreemap: BTreeMap<i32, i32> = vec![(10, 20), (30, 10), (4, 5)].into_iter().collect();
-    let map_btreemap_empty: BTreeMap<i32, i32> = BTreeMap::new();
-    let map_btreemap_empty_selected: Option<BTreeMap<i32, i32>> = None;
-    insert_and_select(&session, table_name, &map_btreemap, &map_btreemap).await;
-    insert_and_select(
-        &session,
-        table_name,
-        &map_btreemap_empty,
-        &map_btreemap_empty_selected,
-    )
-    .await;
-
-    // CqlValue
-    let map_cql_value: CqlValue = CqlValue::Map(vec![
-        (CqlValue::Int(2), CqlValue::Int(4)),
-        (CqlValue::Int(8), CqlValue::Int(16)),
+    let duration1 = (3, "hours");
+    let duration2 = CqlValue::Tuple(vec![
+        Some(CqlValue::Int(2)),
+        Some(CqlValue::Text("minutes".into())),
     ]);
-    let map_cql_value_empty: CqlValue = CqlValue::Map(Vec::new());
-    let map_cql_value_empty_selected: Option<CqlValue> = None;
-    insert_and_select(&session, table_name, &map_cql_value, &map_cql_value).await;
-    insert_and_select(
-        &session,
-        table_name,
-        &map_cql_value_empty,
-        &map_cql_value_empty_selected,
-    )
-    .await;
-}
+    session
+        .query(
+            "INSERT INTO ks.durations (event, duration) VALUES ('ev1', ?)",
+            (duration1,),
+        )
+        .await
+        .unwrap();
+    session
+        .query(
+            "INSERT INTO ks.durations (event, duration) VALUES ('ev2', ?)",
+            (duration2,),
+        )
+        .await
+        .unwrap();
+    session
+        .query(
+            "INSERT INTO ks.durations (event, duration) VALUES ('ev3', (4, null))",
+            &[],
+        )
+        .await
+        .unwrap();
 
-#[tokio::test]
-async fn test_cql_tuple() {
-    let session: Session = connect().await;
-
-    let table_name: &str = "test_cql_tuple_tab";
-    create_table(&session, table_name, "tuple<int, int, text>").await;
-
-    // Rust tuple ()
-    let tuple1: (i32, i32, String) = (1, 2, "three".to_string());
-    let tuple2: (Option<i32>, Option<i32>, String) = (Some(4), None, "sixteen".to_string());
-    insert_and_select(&session, table_name, &tuple1, &tuple1).await;
-    insert_and_select(&session, table_name, &tuple2, &tuple2).await;
-
-    // CqlValue
-    let tuple_cql_value: CqlValue = CqlValue::Tuple(vec![
-        None,
-        Some(CqlValue::Int(1024)),
-        Some(CqlValue::Text("cql_value_text".to_string())),
-    ]);
-    insert_and_select(&session, table_name, &tuple_cql_value, &tuple_cql_value).await;
+    let mut received_elements: Vec<(i32, Option<String>)> = session
+        .query("SELECT duration FROM ks.durations", &[])
+        .await
+        .unwrap()
+        .rows
+        .unwrap()
+        .into_typed::<((i32, Option<String>),)>()
+        .map(Result::unwrap)
+        .map(|(x,)| x)
+        .collect();
+    received_elements.sort_unstable();
+
+    assert_eq!(
+        received_elements,
+        vec![
+            (2, Some("minutes".into())),
+            (3, Some("hours".into())),
+            (4, None)
+        ]
+    );
 }
diff --git a/scylla/src/transport/session_test.rs b/scylla/src/transport/session_test.rs
index 4cdcb3e6..1c3a5276 100644
--- a/scylla/src/transport/session_test.rs
+++ b/scylla/src/transport/session_test.rs
@@ -8,9 +8,12 @@ use crate::statement::Consistency;
 use crate::tracing::TracingInfo;
 use crate::transport::connection::{BatchResult, QueryResult};
 use crate::transport::errors::{BadKeyspaceName, BadQuery, DbError, QueryError};
+use crate::transport::topology::Strategy::SimpleStrategy;
+use crate::transport::topology::{CollectionType, ColumnKind, CqlType, NativeType};
 use crate::{IntoTypedRows, Session, SessionBuilder};
 use bytes::Bytes;
 use futures::StreamExt;
+use itertools::Itertools;
 use uuid::Uuid;
 
 #[tokio::test]
@@ -75,16 +78,6 @@ async fn test_unprepared_statement() {
             (7, 11, &String::from(""))
         ]
     );
-    let query_result = session
-        .query_iter("SELECT a, b, c FROM ks.t", &[])
-        .await
-        .unwrap();
-    let specs = query_result.get_column_specs();
-    assert_eq!(specs.len(), 3);
-    for (spec, name) in specs.iter().zip(["a", "b", "c"]) {
-        assert_eq!(spec.name, name); // Check column name.
-        assert_eq!(spec.table_spec.ks_name, "ks");
-    }
     let mut results_from_manual_paging: Vec<Row> = vec![];
     let query = Query::new("SELECT a, b, c FROM ks.t").with_page_size(1);
     let mut paging_state: Option<Bytes> = None;
@@ -131,16 +124,6 @@ async fn test_prepared_statement() {
         .unwrap();
     // Wait for schema agreement
     std::thread::sleep(std::time::Duration::from_millis(300));
-
-    let prepared_statement = session.prepare("SELECT a, b, c FROM ks.t2").await.unwrap();
-    let query_result = session.execute_iter(prepared_statement, &[]).await.unwrap();
-    let specs = query_result.get_column_specs();
-    assert_eq!(specs.len(), 3);
-    for (spec, name) in specs.iter().zip(["a", "b", "c"]) {
-        assert_eq!(spec.name, name); // Check column name.
-        assert_eq!(spec.table_spec.ks_name, "ks");
-    }
-
     let prepared_statement = session
         .prepare("INSERT INTO ks.t2 (a, b, c) VALUES (?, ?, ?)")
         .await
@@ -1093,3 +1076,491 @@ async fn test_prepared_config() {
     assert_eq!(prepared_statement.get_is_idempotent(), true);
     assert_eq!(prepared_statement.get_page_size(), Some(42));
 }
+
+#[tokio::test]
+async fn test_schema_types_in_metadata() {
+    let uri = std::env::var("SCYLLA_URI").unwrap_or_else(|_| "127.0.0.1:9042".to_string());
+    let session = SessionBuilder::new().known_node(uri).build().await.unwrap();
+
+    session
+        .query("DROP KEYSPACE IF EXISTS test_metadata_ks", &[])
+        .await
+        .unwrap();
+
+    session
+        .query("CREATE KEYSPACE test_metadata_ks WITH REPLICATION = {'class' : 'SimpleStrategy', 'replication_factor' : 1}", &[])
+        .await
+        .unwrap();
+
+    session.query("USE test_metadata_ks", &[]).await.unwrap();
+
+    session
+        .query(
+            "CREATE TYPE IF NOT EXISTS type_a (
+                    a map<frozen<list<int>>, text>,
+                    b frozen<map<frozen<list<int>>, frozen<set<text>>>>
+                   )",
+            &[],
+        )
+        .await
+        .unwrap();
+
+    session
+        .query("CREATE TYPE IF NOT EXISTS type_b (a int, b text)", &[])
+        .await
+        .unwrap();
+
+    session
+        .query(
+            "CREATE TYPE IF NOT EXISTS type_c (a map<frozen<set<text>>, frozen<type_b>>)",
+            &[],
+        )
+        .await
+        .unwrap();
+
+    session
+        .query(
+            "CREATE TABLE IF NOT EXISTS table_a (
+                    a frozen<type_a> PRIMARY KEY,
+                    b type_b,
+                    c frozen<type_c>,
+                    d map<text, frozen<list<int>>>,
+                    e tuple<int, text>
+                  )",
+            &[],
+        )
+        .await
+        .unwrap();
+
+    session
+        .query(
+            "CREATE TABLE IF NOT EXISTS table_b (
+                        a text PRIMARY KEY,
+                        b frozen<map<int, int>>
+                     )",
+            &[],
+        )
+        .await
+        .unwrap();
+
+    session.await_schema_agreement().await.unwrap();
+    session.refresh_metadata().await.unwrap();
+
+    let cluster_data = session.get_cluster_data();
+    let tables = &cluster_data.get_keyspace_info()["test_metadata_ks"].tables;
+
+    assert_eq!(
+        tables.keys().sorted().collect::<Vec<_>>(),
+        vec!["table_a", "table_b"]
+    );
+
+    let table_a_columns = &tables["table_a"].columns;
+
+    assert_eq!(
+        table_a_columns.keys().sorted().collect::<Vec<_>>(),
+        vec!["a", "b", "c", "d", "e"]
+    );
+
+    let a = &table_a_columns["a"];
+
+    assert_eq!(
+        a.type_,
+        CqlType::UserDefinedType {
+            name: "type_a".to_string(),
+            frozen: true
+        }
+    );
+
+    let b = &table_a_columns["b"];
+
+    assert_eq!(
+        b.type_,
+        CqlType::UserDefinedType {
+            name: "type_b".to_string(),
+            frozen: false,
+        }
+    );
+
+    let c = &table_a_columns["c"];
+
+    assert_eq!(
+        c.type_,
+        CqlType::UserDefinedType {
+            name: "type_c".to_string(),
+            frozen: true
+        }
+    );
+
+    let d = &table_a_columns["d"];
+
+    assert_eq!(
+        d.type_,
+        CqlType::Collection {
+            type_: CollectionType::Map(
+                Box::new(CqlType::Native(NativeType::Text)),
+                Box::new(CqlType::Collection {
+                    type_: CollectionType::List(Box::new(CqlType::Native(NativeType::Int))),
+                    frozen: true
+                })
+            ),
+            frozen: false
+        }
+    );
+
+    let e = &table_a_columns["e"];
+
+    assert_eq!(
+        e.type_,
+        CqlType::Tuple(vec![
+            CqlType::Native(NativeType::Int),
+            CqlType::Native(NativeType::Text)
+        ])
+    );
+
+    let table_b_columns = &tables["table_b"].columns;
+
+    let a = &table_b_columns["a"];
+
+    assert_eq!(a.type_, CqlType::Native(NativeType::Text));
+
+    let b = &table_b_columns["b"];
+
+    assert_eq!(
+        b.type_,
+        CqlType::Collection {
+            type_: CollectionType::Map(
+                Box::new(CqlType::Native(NativeType::Int),),
+                Box::new(CqlType::Native(NativeType::Int),)
+            ),
+            frozen: true
+        }
+    );
+}
+
+#[tokio::test]
+async fn test_user_defined_types_in_metadata() {
+    let uri = std::env::var("SCYLLA_URI").unwrap_or_else(|_| "127.0.0.1:9042".to_string());
+    let session = SessionBuilder::new().known_node(uri).build().await.unwrap();
+
+    session
+        .query("DROP KEYSPACE IF EXISTS test_metadata_ks", &[])
+        .await
+        .unwrap();
+
+    session
+        .query("CREATE KEYSPACE test_metadata_ks WITH REPLICATION = {'class' : 'SimpleStrategy', 'replication_factor' : 1}", &[])
+        .await
+        .unwrap();
+
+    session.query("USE test_metadata_ks", &[]).await.unwrap();
+
+    session
+        .query(
+            "CREATE TYPE IF NOT EXISTS type_a (
+                    a map<frozen<list<int>>, text>,
+                    b frozen<map<frozen<list<int>>, frozen<set<text>>>>
+                   )",
+            &[],
+        )
+        .await
+        .unwrap();
+
+    session
+        .query("CREATE TYPE IF NOT EXISTS type_b (a int, b text)", &[])
+        .await
+        .unwrap();
+
+    session
+        .query(
+            "CREATE TYPE IF NOT EXISTS type_c (a map<frozen<set<text>>, frozen<type_b>>)",
+            &[],
+        )
+        .await
+        .unwrap();
+
+    session.await_schema_agreement().await.unwrap();
+    session.refresh_metadata().await.unwrap();
+
+    let cluster_data = session.get_cluster_data();
+    let user_defined_types =
+        &cluster_data.get_keyspace_info()["test_metadata_ks"].user_defined_types;
+
+    assert_eq!(
+        user_defined_types.keys().sorted().collect::<Vec<_>>(),
+        vec!["type_a", "type_b", "type_c"]
+    );
+
+    let type_a = &user_defined_types["type_a"];
+
+    assert_eq!(
+        type_a,
+        &vec![
+            (
+                "a".to_string(),
+                CqlType::Collection {
+                    frozen: false,
+                    type_: CollectionType::Map(
+                        Box::new(CqlType::Collection {
+                            frozen: true,
+                            type_: CollectionType::List(Box::new(CqlType::Native(NativeType::Int)))
+                        }),
+                        Box::new(CqlType::Native(NativeType::Text))
+                    )
+                }
+            ),
+            (
+                "b".to_string(),
+                CqlType::Collection {
+                    frozen: true,
+                    type_: CollectionType::Map(
+                        Box::new(CqlType::Collection {
+                            frozen: true,
+                            type_: CollectionType::List(Box::new(CqlType::Native(NativeType::Int)))
+                        }),
+                        Box::new(CqlType::Collection {
+                            frozen: true,
+                            type_: CollectionType::Set(Box::new(CqlType::Native(NativeType::Text)))
+                        })
+                    )
+                }
+            )
+        ]
+    );
+
+    let type_b = &user_defined_types["type_b"];
+
+    assert_eq!(
+        type_b,
+        &vec![
+            ("a".to_string(), CqlType::Native(NativeType::Int)),
+            ("b".to_string(), CqlType::Native(NativeType::Text))
+        ]
+    );
+
+    let type_c = &user_defined_types["type_c"];
+
+    assert_eq!(
+        type_c,
+        &vec![(
+            "a".to_string(),
+            CqlType::Collection {
+                frozen: false,
+                type_: CollectionType::Map(
+                    Box::new(CqlType::Collection {
+                        frozen: true,
+                        type_: CollectionType::Set(Box::new(CqlType::Native(NativeType::Text)))
+                    }),
+                    Box::new(CqlType::UserDefinedType {
+                        frozen: true,
+                        name: "type_b".to_string()
+                    })
+                )
+            }
+        )]
+    );
+}
+
+#[tokio::test]
+async fn test_column_kinds_in_metadata() {
+    let uri = std::env::var("SCYLLA_URI").unwrap_or_else(|_| "127.0.0.1:9042".to_string());
+    let session = SessionBuilder::new().known_node(uri).build().await.unwrap();
+
+    session
+        .query("DROP KEYSPACE IF EXISTS test_metadata_ks", &[])
+        .await
+        .unwrap();
+
+    session
+        .query("CREATE KEYSPACE test_metadata_ks WITH REPLICATION = {'class' : 'SimpleStrategy', 'replication_factor' : 1}", &[])
+        .await
+        .unwrap();
+
+    session.query("USE test_metadata_ks", &[]).await.unwrap();
+
+    session
+        .query(
+            "CREATE TABLE IF NOT EXISTS t (
+                    a int,
+                    b int,
+                    c int,
+                    d int STATIC,
+                    e int,
+                    f int,
+                    PRIMARY KEY ((c, e), b, a)
+                  )",
+            &[],
+        )
+        .await
+        .unwrap();
+
+    session.await_schema_agreement().await.unwrap();
+    session.refresh_metadata().await.unwrap();
+
+    let cluster_data = session.get_cluster_data();
+    let columns = &cluster_data.get_keyspace_info()["test_metadata_ks"].tables["t"].columns;
+
+    assert_eq!(columns["a"].kind, ColumnKind::Clustering);
+    assert_eq!(columns["b"].kind, ColumnKind::Clustering);
+    assert_eq!(columns["c"].kind, ColumnKind::PartitionKey);
+    assert_eq!(columns["d"].kind, ColumnKind::Static);
+    assert_eq!(columns["e"].kind, ColumnKind::PartitionKey);
+    assert_eq!(columns["f"].kind, ColumnKind::Regular);
+}
+
+#[tokio::test]
+async fn test_primary_key_ordering_in_metadata() {
+    let uri = std::env::var("SCYLLA_URI").unwrap_or_else(|_| "127.0.0.1:9042".to_string());
+    let session = SessionBuilder::new().known_node(uri).build().await.unwrap();
+
+    session
+        .query("DROP KEYSPACE IF EXISTS test_metadata_ks", &[])
+        .await
+        .unwrap();
+
+    session
+        .query("CREATE KEYSPACE test_metadata_ks WITH REPLICATION = {'class' : 'SimpleStrategy', 'replication_factor' : 1}", &[])
+        .await
+        .unwrap();
+
+    session.query("USE test_metadata_ks", &[]).await.unwrap();
+
+    session
+        .query(
+            "CREATE TABLE IF NOT EXISTS t (
+                    a int,
+                    b int,
+                    c int,
+                    d int STATIC,
+                    e int,
+                    f int,
+                    g int,
+                    h int,
+                    i int STATIC,
+                    PRIMARY KEY ((c, e), b, a)
+                  )",
+            &[],
+        )
+        .await
+        .unwrap();
+
+    session.await_schema_agreement().await.unwrap();
+    session.refresh_metadata().await.unwrap();
+
+    let cluster_data = session.get_cluster_data();
+    let table = &cluster_data.get_keyspace_info()["test_metadata_ks"].tables["t"];
+
+    assert_eq!(table.partition_key, vec!["c", "e"]);
+    assert_eq!(table.clustering_key, vec!["b", "a"]);
+}
+
+#[tokio::test]
+async fn test_table_partitioner_in_metadata() {
+    if option_env!("CDC") == Some("disabled") {
+        return;
+    }
+
+    let uri = std::env::var("SCYLLA_URI").unwrap_or_else(|_| "127.0.0.1:9042".to_string());
+    let session = SessionBuilder::new().known_node(uri).build().await.unwrap();
+
+    session
+        .query("DROP KEYSPACE IF EXISTS test_metadata_ks", &[])
+        .await
+        .unwrap();
+
+    session
+        .query("CREATE KEYSPACE test_metadata_ks WITH REPLICATION = {'class' : 'SimpleStrategy', 'replication_factor' : 1}", &[])
+        .await
+        .unwrap();
+
+    session.query("USE test_metadata_ks", &[]).await.unwrap();
+
+    session
+        .query(
+            "CREATE TABLE t (pk int, ck int, v int, PRIMARY KEY (pk, ck, v))WITH cdc = {'enabled':true}",
+            &[],
+        )
+        .await
+        .unwrap();
+
+    session.await_schema_agreement().await.unwrap();
+    session.refresh_metadata().await.unwrap();
+
+    let cluster_data = session.get_cluster_data();
+    let tables = &cluster_data.get_keyspace_info()["test_metadata_ks"].tables;
+    let table = &tables["t"];
+    let cdc_table = &tables["t_scylla_cdc_log"];
+
+    assert_eq!(table.partitioner, None);
+    assert_eq!(
+        cdc_table.partitioner.as_ref().unwrap(),
+        "com.scylladb.dht.CDCPartitioner"
+    );
+}
+
+#[tokio::test]
+async fn test_turning_off_schema_fetching() {
+    let uri = std::env::var("SCYLLA_URI").unwrap_or_else(|_| "127.0.0.1:9042".to_string());
+    let session = SessionBuilder::new()
+        .fetch_schema_metadata(false)
+        .known_node(uri)
+        .build()
+        .await
+        .unwrap();
+
+    session
+        .query("CREATE KEYSPACE IF NOT EXISTS test_metadata_ks WITH REPLICATION = {'class' : 'SimpleStrategy', 'replication_factor' : 1}", &[])
+        .await
+        .unwrap();
+
+    session.query("USE test_metadata_ks", &[]).await.unwrap();
+
+    session
+        .query(
+            "CREATE TYPE IF NOT EXISTS type_a (
+                    a map<frozen<list<int>>, text>,
+                    b frozen<map<frozen<list<int>>, frozen<set<text>>>>
+                   )",
+            &[],
+        )
+        .await
+        .unwrap();
+
+    session
+        .query("CREATE TYPE IF NOT EXISTS type_b (a int, b text)", &[])
+        .await
+        .unwrap();
+
+    session
+        .query(
+            "CREATE TYPE IF NOT EXISTS type_c (a map<frozen<set<text>>, frozen<type_b>>)",
+            &[],
+        )
+        .await
+        .unwrap();
+
+    session
+        .query(
+            "CREATE TABLE IF NOT EXISTS table_a (
+                    a frozen<type_a> PRIMARY KEY,
+                    b type_b,
+                    c frozen<type_c>,
+                    d map<text, frozen<list<int>>>,
+                    e tuple<int, text>
+                  )",
+            &[],
+        )
+        .await
+        .unwrap();
+
+    let cluster_data = &session.get_cluster_data();
+    let keyspace = &cluster_data.get_keyspace_info()["test_metadata_ks"];
+
+    assert_eq!(
+        keyspace.strategy,
+        SimpleStrategy {
+            replication_factor: 1
+        }
+    );
+    assert_eq!(keyspace.tables.len(), 0);
+    assert_eq!(keyspace.user_defined_types.len(), 0);
+}