Repository: scylladb/scylla-rust-driver
Pull request #776 (2023-07-30 16:32:10)

configurable topology refresh duration
This pull requests a configurable way to topology refresh duration. Two files session.rs and cluster.rs have been modified to set the session configuration and consume the duration configuration respectively.

Note: I am not sure if we need any test cases in this case as the duration is still provided from where its called. 

- [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 have provided docstrings for the public items that I want to introduce.
- [x] I have adjusted the documentation in `./docs/source/`.
- [x] I added appropriate `Fixes:` annotations to PR description.


Fixes: #232 



Patch:
diff --git a/docs/source/connecting/connecting.md b/docs/source/connecting/connecting.md
index a8892f62..1522cb25 100644
--- a/docs/source/connecting/connecting.md
+++ b/docs/source/connecting/connecting.md
@@ -20,6 +20,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
         .known_node("127.0.0.72:4321")
         .known_node("localhost:8000")
         .connection_timeout(Duration::from_secs(3))
+        .cluster_metadata_refresh_interval(Duration::from_secs(10))
         .known_node_addr(SocketAddr::new(
             IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)),
             9000,
@@ -34,6 +35,9 @@ async fn main() -> Result<(), Box<dyn Error>> {
 After successfully connecting to some specified node the driver will fetch topology information about
 other nodes in this cluster and connect to them as well.
 
+The driver refreshes the cluster metadata periodically, which contains information about cluster topology as well as the cluster schema. By default, the driver refreshes the cluster metadata every 60 seconds. 
+However, you can set the `cluster_metadata_refresh_interval` to a non-negative value to periodically refresh the cluster metadata. This is useful when you do not have unexpected amount of traffic or when you have an extra traffic causing topology to change frequently.
+
 Scylla Serverless is an elastic and dynamic deployment model. When creating a `Session` you need to
 specify the secure connection bundle as follows:
 
diff --git a/docs/source/data-types/udt.md b/docs/source/data-types/udt.md
index 85c401fa..ceb8ae8c 100644
--- a/docs/source/data-types/udt.md
+++ b/docs/source/data-types/udt.md
@@ -1,13 +1,36 @@
 # User defined types
-Scylla allows users to define their own data types with named fields.\
-The driver supports this by allowing to create a custom struct with derived functionality.
+Scylla allows users to define their own data types with named fields (See [the official documentation](https://opensource.docs.scylladb.com/stable/cql/types.html#user-defined-types))\
+To use user defined types in the driver, you can create a corresponding struct in Rust, and use it to read and write UDT values.
+
 
 For example let's say `my_type` was created using this query:
 ```sql
 CREATE TYPE ks.my_type (int_val int, text_val text)
 ```
 
-To use this type in the driver create a matching struct and derive `IntoUserType` and `FromUserType`:
+To use this type in the driver, create a matching struct and derive `IntoUserType` and `FromUserType`:
+
+```rust
+# extern crate scylla;
+# async fn check_only_compiles() {
+use scylla::macros::{FromUserType, IntoUserType};
+
+// Define a custom struct that matches the User Defined Type created earlier.
+// Fields must be in the same order as they are in the database.
+// Wrapping a field in Option will gracefully handle null field values.
+#[derive(Debug, IntoUserType, FromUserType)]
+struct MyType {
+    int_val: i32,
+    text_val: Option<String>,
+}
+# }
+```
+
+> ***Important***\
+> Fields in the Rust struct must be defined in the same order as they are in the database.
+> When sending and receiving values, the driver will (de)serialize fields one after another, without looking at field names.
+
+Now it can be sent and received just like any other CQL value:
 ```rust
 # extern crate scylla;
 # use scylla::Session;
@@ -17,16 +40,12 @@ use scylla::IntoTypedRows;
 use scylla::macros::{FromUserType, IntoUserType};
 use scylla::cql_to_rust::FromCqlVal;
 
-// Define custom struct that matches User Defined Type created earlier
-// wrapping field in Option will gracefully handle null field values
 #[derive(Debug, IntoUserType, FromUserType)]
 struct MyType {
     int_val: i32,
     text_val: Option<String>,
 }
 
-// Now it can be sent and received like any other value
-
 // Insert my_type into the table
 let to_insert = MyType {
     int_val: 17,
diff --git a/docs/source/queries/schema-agreement.md b/docs/source/queries/schema-agreement.md
index b9a519b4..2df38743 100644
--- a/docs/source/queries/schema-agreement.md
+++ b/docs/source/queries/schema-agreement.md
@@ -1,25 +1,33 @@
 # Schema agreement
 
-Sometimes after performing queries some nodes have not been updated so we need a mechanism that checks if every node have agreed schema version.
-There are four methods in `Session` that assist us. 
-Every method raise `QueryError` if something goes wrong, but they should never raise any errors, unless there is a DB or connection malfunction.
+Sometimes after performing queries some nodes have not been updated, so we need a mechanism that checks if every node have agreed on schema version.
 
-### Checking schema version
-`Session::fetch_schema_version` returns an `Uuid` of local node's schema version. 
+### Automated awaiting schema agreement
+
+The driver automatically awaits schema agreement after a schema-altering query is executed.
+Waiting for schema agreement more than necessary is never a bug, but might slow down applications which do a lot of schema changes (e.g. a migration).
+For instance, in case where somebody wishes to create a keyspace and then a lot of tables in it, it makes sense only to wait after creating a keyspace
+and after creating all the tables rather than after every query. Therefore, the said behaviour can be disabled:
 
 ```rust
 # extern crate scylla;
-# use scylla::Session;
+# use scylla::SessionBuilder;
 # use std::error::Error;
-# async fn check_only_compiles(session: &Session) -> Result<(), Box<dyn Error>> {
-println!("Local schema version is: {}", session.fetch_schema_version().await?);
+# async fn check_only_compiles() -> Result<(), Box<dyn Error>> {
+let session = SessionBuilder::new()
+    .known_node("127.0.0.1:9042")
+    .auto_await_schema_agreement(false)
+    .build()
+    .await?;
 # Ok(())
 # }
 ```
 
-### Awaiting schema agreement
+### Manually awaiting schema agreement
 
-`Session::await_schema_agreement` returns a `Future` that can be `await`ed on as long as schema is not in an agreement.
+`Session::await_schema_agreement` returns a `Future` that can be `await`ed as long as schema is not in an agreement.
+However, it won't wait forever; `SessionConfig` defines a timeout that limits the time of waiting. If the timeout elapses,
+the return value is `Err(QueryError::RequestTimeout)`, otherwise it is `Ok(schema_version)`.
 
 ```rust
 # extern crate scylla;
@@ -31,28 +39,10 @@ session.await_schema_agreement().await?;
 # }
 ```
 
-### Awaiting with timeout
-We can also set timeout in milliseconds with `Session::await_timed_schema_agreement`.
-It takes one argument, an `std::time::Duration` value that tells how long our driver should await for schema agreement. If the timeout is met the return value is `false` otherwise it is `true`.
-
-```rust
-# extern crate scylla;
-# use scylla::Session;
-# use std::error::Error;
-# use std::time::Duration;
-# async fn check_only_compiles(session: &Session) -> Result<(), Box<dyn Error>> {
-if session.await_timed_schema_agreement(Duration::from_secs(5)).await? { // wait for 5 seconds
-    println!("SCHEMA AGREED");
-} else {
-    println!("SCHEMA IS NOT IN AGREEMENT - TIMED OUT");
-}
-# Ok(())
-# }
-```
-
-### Checking for schema interval
-If schema is not agreed driver sleeps for a duration before checking it again. Default value is 200 milliseconds but it can be changed with `SessionBuilder::schema_agreement_interval`.
+### Interval of checking for schema agreement
 
+If the schema is not agreed upon, the driver sleeps for a duration before checking it again. The default value is 200 milliseconds,
+but it can be changed with `SessionBuilder::schema_agreement_interval`.
 
 ```rust
 # extern crate scylla;
@@ -70,15 +60,15 @@ SessionBuilder::new()
 ```
 
 ### Checking if schema is in agreement now
-If you want to check if schema is in agreement now without retrying after failure you can use `Session::check_schema_agreement` function.
 
+If you want to check if schema is in agreement now, without retrying after failure, you can use `Session::check_schema_agreement` function.
 
 ```rust
 # extern crate scylla;
 # use scylla::Session;
 # use std::error::Error;
 # async fn check_only_compiles(session: &Session) -> Result<(), Box<dyn Error>> {
-if session.check_schema_agreement().await? { 
+if session.check_schema_agreement().await?.is_some() {
     println!("SCHEMA AGREED");
 } else {
     println!("SCHEMA IS NOT IN AGREEMENT");
@@ -86,5 +76,3 @@ if session.check_schema_agreement().await? {
 # Ok(())
 # }
 ```
-
-
diff --git a/examples/compare-tokens.rs b/examples/compare-tokens.rs
index 793aae2e..5c56aa5f 100644
--- a/examples/compare-tokens.rs
+++ b/examples/compare-tokens.rs
@@ -1,7 +1,6 @@
 use anyhow::Result;
 use scylla::frame::value::ValueList;
 use scylla::routing::Token;
-use scylla::transport::partitioner::{Murmur3Partitioner, Partitioner};
 use scylla::transport::NodeAddr;
 use scylla::{Session, SessionBuilder};
 use std::env;
@@ -31,7 +30,7 @@ async fn main() -> Result<()> {
             .await?;
 
         let serialized_pk = (pk,).serialized()?.into_owned();
-        let t = Murmur3Partitioner::hash(&prepared.compute_partition_key(&serialized_pk)?).value;
+        let t = prepared.calculate_token(&serialized_pk)?.unwrap().value;
 
         println!(
             "Token endpoints for query: {:?}",
diff --git a/examples/schema_agreement.rs b/examples/schema_agreement.rs
index b1825ddf..08e5a593 100644
--- a/examples/schema_agreement.rs
+++ b/examples/schema_agreement.rs
@@ -1,4 +1,5 @@
-use anyhow::Result;
+use anyhow::{bail, Result};
+use scylla::transport::errors::QueryError;
 use scylla::transport::session::{IntoTypedRows, Session};
 use scylla::SessionBuilder;
 use std::env;
@@ -17,21 +18,17 @@ async fn main() -> Result<()> {
         .build()
         .await?;
 
-    let schema_version = session.fetch_schema_version().await?;
+    let schema_version = session.await_schema_agreement().await?;
+
     println!("Schema version: {}", schema_version);
 
-    session.await_schema_agreement().await?; // without timeout example
     session.query("CREATE KEYSPACE IF NOT EXISTS ks WITH REPLICATION = {'class' : 'NetworkTopologyStrategy', 'replication_factor' : 1}", &[]).await?;
 
-    if session
-        .await_timed_schema_agreement(Duration::from_secs(5))
-        .await?
-    {
-        // with timeout example
-        println!("Timed schema is in agreement");
-    } else {
-        println!("Timed schema is NOT in agreement");
-    }
+    match session.await_schema_agreement().await {
+        Ok(_schema_version) => println!("Schema is in agreement in time"),
+        Err(QueryError::RequestTimeout(_)) => println!("Schema is NOT in agreement in time"),
+        Err(err) => bail!(err),
+    };
     session
         .query(
             "CREATE TABLE IF NOT EXISTS ks.t (a int, b int, c text, primary key (a, b))",
@@ -71,7 +68,7 @@ async fn main() -> Result<()> {
     }
     println!("Ok.");
 
-    let schema_version = session.fetch_schema_version().await?;
+    let schema_version = session.await_schema_agreement().await?;
     println!("Schema version: {}", schema_version);
 
     Ok(())
diff --git a/scylla-cql/src/errors.rs b/scylla-cql/src/errors.rs
index 6456f126..40587cfe 100644
--- a/scylla-cql/src/errors.rs
+++ b/scylla-cql/src/errors.rs
@@ -2,8 +2,8 @@
 
 use crate::frame::frame_errors::{FrameError, ParseError};
 use crate::frame::protocol_features::ProtocolFeatures;
-use crate::frame::types::LegacyConsistency;
 use crate::frame::value::SerializeValuesError;
+use crate::Consistency;
 use bytes::Bytes;
 use std::io::ErrorKind;
 use std::sync::Arc;
@@ -108,7 +108,7 @@ pub enum DbError {
     )]
     Unavailable {
         /// Consistency level of the query
-        consistency: LegacyConsistency,
+        consistency: Consistency,
         /// Number of nodes required to be alive to satisfy required consistency level
         required: i32,
         /// Found number of active nodes
@@ -132,7 +132,7 @@ pub enum DbError {
             (consistency: {consistency}, received: {received}, required: {required}, data_present: {data_present})")]
     ReadTimeout {
         /// Consistency level of the query
-        consistency: LegacyConsistency,
+        consistency: Consistency,
         /// Number of nodes that responded to the read request
         received: i32,
         /// Number of nodes required to respond to satisfy required consistency level
@@ -146,7 +146,7 @@ pub enum DbError {
             (consistency: {consistency}, received: {received}, required: {required}, write_type: {write_type})")]
     WriteTimeout {
         /// Consistency level of the query
-        consistency: LegacyConsistency,
+        consistency: Consistency,
         /// Number of nodes that responded to the write request
         received: i32,
         /// Number of nodes required to respond to satisfy required consistency level
@@ -163,7 +163,7 @@ pub enum DbError {
     )]
     ReadFailure {
         /// Consistency level of the query
-        consistency: LegacyConsistency,
+        consistency: Consistency,
         /// Number of nodes that responded to the read request
         received: i32,
         /// Number of nodes required to respond to satisfy required consistency level
@@ -182,7 +182,7 @@ pub enum DbError {
     )]
     WriteFailure {
         /// Consistency level of the query
-        consistency: LegacyConsistency,
+        consistency: Consistency,
         /// Number of nodes that responded to the read request
         received: i32,
         /// Number of nodes required to respond to satisfy required consistency level
@@ -550,7 +550,7 @@ impl WriteType {
 #[cfg(test)]
 mod tests {
     use super::{DbError, QueryError, WriteType};
-    use crate::frame::types::{Consistency, LegacyConsistency};
+    use crate::frame::types::Consistency;
 
     #[test]
     fn write_type_from_str() {
@@ -581,7 +581,7 @@ mod tests {
     fn dberror_full_info() {
         // Test that DbError::Unavailable is displayed correctly
         let db_error = DbError::Unavailable {
-            consistency: LegacyConsistency::Regular(Consistency::Three),
+            consistency: Consistency::Three,
             required: 3,
             alive: 2,
         };
diff --git a/scylla-cql/src/frame/request/batch.rs b/scylla-cql/src/frame/request/batch.rs
index 91326a51..3c0bad39 100644
--- a/scylla-cql/src/frame/request/batch.rs
+++ b/scylla-cql/src/frame/request/batch.rs
@@ -4,7 +4,7 @@ use std::{borrow::Cow, convert::TryInto};
 use crate::frame::{
     frame_errors::ParseError,
     request::{RequestOpcode, SerializableRequest},
-    types,
+    types::{self, SerialConsistency},
     value::{BatchValues, BatchValuesIterator, SerializedValues},
 };
 
@@ -202,13 +202,7 @@ impl<'b> DeserializableRequest for Batch<'b, BatchStatement<'b>, Vec<SerializedV
             })
             .collect::<Result<Vec<_>, ParseError>>()?;
 
-        let consistency = match types::read_consistency(buf)? {
-            types::LegacyConsistency::Regular(reg) => Ok(reg),
-            types::LegacyConsistency::Serial(ser) => Err(ParseError::BadIncomingData(format!(
-                "Expected regular Consistency, got SerialConsistency {}",
-                ser
-            ))),
-        }?;
+        let consistency = types::read_consistency(buf)?;
 
         let flags = buf.get_u8();
         let unknown_flags = flags & (!ALL_FLAGS);
@@ -224,15 +218,15 @@ impl<'b> DeserializableRequest for Batch<'b, BatchStatement<'b>, Vec<SerializedV
         let serial_consistency = serial_consistency_flag
             .then(|| types::read_consistency(buf))
             .transpose()?
-            .map(|legacy_consistency| match legacy_consistency {
-                types::LegacyConsistency::Regular(reg) => {
-                    Err(ParseError::BadIncomingData(format!(
+            .map(
+                |consistency| match SerialConsistency::try_from(consistency) {
+                    Ok(serial_consistency) => Ok(serial_consistency),
+                    Err(_) => Err(ParseError::BadIncomingData(format!(
                         "Expected SerialConsistency, got regular Consistency {}",
-                        reg
-                    )))
-                }
-                types::LegacyConsistency::Serial(ser) => Ok(ser),
-            })
+                        consistency
+                    ))),
+                },
+            )
             .transpose()?;
 
         let timestamp = default_timestamp_flag
diff --git a/scylla-cql/src/frame/request/mod.rs b/scylla-cql/src/frame/request/mod.rs
index 8202ccde..cd41d6bc 100644
--- a/scylla-cql/src/frame/request/mod.rs
+++ b/scylla-cql/src/frame/request/mod.rs
@@ -112,7 +112,7 @@ mod tests {
                 query::{Query, QueryParameters},
                 DeserializableRequest, SerializableRequest,
             },
-            types::{self, LegacyConsistency, SerialConsistency},
+            types::{self, SerialConsistency},
             value::SerializedValues,
         },
         Consistency,
@@ -236,10 +236,7 @@ mod tests {
 
             // Now buf_ptr points at consistency.
             let consistency = types::read_consistency(&mut buf_ptr).unwrap();
-            assert_eq!(
-                consistency,
-                LegacyConsistency::Regular(Consistency::default())
-            );
+            assert_eq!(consistency, Consistency::default());
 
             // Now buf_ptr points at flags, but it is immutable. Get mutable reference into the buffer.
             let flags_idx = buf.len() - buf_ptr.len();
diff --git a/scylla-cql/src/frame/request/query.rs b/scylla-cql/src/frame/request/query.rs
index 1b5a52ed..ff0b0cc8 100644
--- a/scylla-cql/src/frame/request/query.rs
+++ b/scylla-cql/src/frame/request/query.rs
@@ -1,6 +1,6 @@
 use std::borrow::Cow;
 
-use crate::frame::frame_errors::ParseError;
+use crate::frame::{frame_errors::ParseError, types::SerialConsistency};
 use bytes::{Buf, BufMut, Bytes};
 
 use crate::{
@@ -134,13 +134,7 @@ impl QueryParameters<'_> {
 
 impl<'q> QueryParameters<'q> {
     pub fn deserialize(buf: &mut &[u8]) -> Result<Self, ParseError> {
-        let consistency = match types::read_consistency(buf)? {
-            types::LegacyConsistency::Regular(reg) => Ok(reg),
-            types::LegacyConsistency::Serial(ser) => Err(ParseError::BadIncomingData(format!(
-                "Expected regular Consistency, got SerialConsistency {}",
-                ser
-            ))),
-        }?;
+        let consistency = types::read_consistency(buf)?;
 
         let flags = buf.get_u8();
         let unknown_flags = flags & (!ALL_FLAGS);
@@ -169,19 +163,19 @@ impl<'q> QueryParameters<'q> {
         } else {
             None
         };
-        let serial_consistency = if serial_consistency_flag {
-            match types::read_consistency(buf)? {
-                types::LegacyConsistency::Regular(reg) => {
-                    return Err(ParseError::BadIncomingData(format!(
+        let serial_consistency = serial_consistency_flag
+            .then(|| types::read_consistency(buf))
+            .transpose()?
+            .map(
+                |consistency| match SerialConsistency::try_from(consistency) {
+                    Ok(serial_consistency) => Ok(serial_consistency),
+                    Err(_) => Err(ParseError::BadIncomingData(format!(
                         "Expected SerialConsistency, got regular Consistency {}",
-                        reg
-                    )))
-                }
-                types::LegacyConsistency::Serial(ser) => Some(ser),
-            }
-        } else {
-            None
-        };
+                        consistency
+                    ))),
+                },
+            )
+            .transpose()?;
         let timestamp = if default_timestamp_flag {
             Some(types::read_long(buf)?)
         } else {
diff --git a/scylla-cql/src/frame/response/error.rs b/scylla-cql/src/frame/response/error.rs
index 8f438243..e99d5686 100644
--- a/scylla-cql/src/frame/response/error.rs
+++ b/scylla-cql/src/frame/response/error.rs
@@ -92,7 +92,6 @@ mod tests {
     use super::Error;
     use crate::errors::{DbError, OperationType, WriteType};
     use crate::frame::protocol_features::ProtocolFeatures;
-    use crate::frame::types::LegacyConsistency;
     use crate::Consistency;
     use bytes::Bytes;
     use std::convert::TryInto;
@@ -151,7 +150,7 @@ mod tests {
         assert_eq!(
             error.error,
             DbError::Unavailable {
-                consistency: LegacyConsistency::Regular(Consistency::One),
+                consistency: Consistency::One,
                 required: 2,
                 alive: 3,
             }
@@ -178,7 +177,7 @@ mod tests {
         assert_eq!(
             error.error,
             DbError::WriteTimeout {
-                consistency: LegacyConsistency::Regular(Consistency::Quorum),
+                consistency: Consistency::Quorum,
                 received: -5, // Allow negative values when they don't make sense, it's better than crashing with ProtocolError
                 required: 100,
                 write_type: WriteType::Simple,
@@ -202,7 +201,7 @@ mod tests {
         assert_eq!(
             error.error,
             DbError::ReadTimeout {
-                consistency: LegacyConsistency::Regular(Consistency::Two),
+                consistency: Consistency::Two,
                 received: 8,
                 required: 32,
                 data_present: false,
@@ -227,7 +226,7 @@ mod tests {
         assert_eq!(
             error.error,
             DbError::ReadFailure {
-                consistency: LegacyConsistency::Regular(Consistency::Three),
+                consistency: Consistency::Three,
                 received: 4,
                 required: 5,
                 numfailures: 6,
@@ -299,7 +298,7 @@ mod tests {
         assert_eq!(
             error.error,
             DbError::WriteFailure {
-                consistency: LegacyConsistency::Regular(Consistency::Any),
+                consistency: Consistency::Any,
                 received: 2,
                 required: 4,
                 numfailures: 8,
diff --git a/scylla-cql/src/frame/response/result.rs b/scylla-cql/src/frame/response/result.rs
index 80e4aa9b..288baf91 100644
--- a/scylla-cql/src/frame/response/result.rs
+++ b/scylla-cql/src/frame/response/result.rs
@@ -34,7 +34,7 @@ pub struct SchemaChange {
     pub event: SchemaChangeEvent,
 }
 
-#[derive(Clone, Debug)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub struct TableSpec {
     pub ks_name: String,
     pub table_name: String,
@@ -329,7 +329,7 @@ impl CqlValue {
     // TODO
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub struct ColumnSpec {
     pub table_spec: TableSpec,
     pub name: String,
diff --git a/scylla-cql/src/frame/types.rs b/scylla-cql/src/frame/types.rs
index 858db35f..71b0f782 100644
--- a/scylla-cql/src/frame/types.rs
+++ b/scylla-cql/src/frame/types.rs
@@ -10,6 +10,7 @@ use std::convert::TryInto;
 use std::net::IpAddr;
 use std::net::SocketAddr;
 use std::str;
+use thiserror::Error;
 use uuid::Uuid;
 
 #[derive(Debug, Copy, Clone, Default, PartialEq, Eq, PartialOrd, Ord, TryFromPrimitive)]
@@ -27,6 +28,11 @@ pub enum Consistency {
     LocalQuorum = 0x0006,
     EachQuorum = 0x0007,
     LocalOne = 0x000A,
+
+    // Apparently, Consistency can be set to Serial or LocalSerial in SELECT statements
+    // to make them use Paxos.
+    Serial = 0x0008,
+    LocalSerial = 0x0009,
 }
 
 #[derive(Debug, Copy, Clone, PartialEq, Eq, PartialOrd, Ord, TryFromPrimitive)]
@@ -38,13 +44,34 @@ pub enum SerialConsistency {
     LocalSerial = 0x0009,
 }
 
-// LegacyConsistency exists, because Scylla may return a SerialConsistency value
-// as Consistency when returning certain error types - the distinction between
-// Consistency and SerialConsistency is not really a thing in CQL.
-#[derive(Debug, Copy, Clone, PartialEq, Eq, PartialOrd, Ord)]
-pub enum LegacyConsistency {
-    Regular(Consistency),
-    Serial(SerialConsistency),
+impl Consistency {
+    pub fn is_serial(&self) -> bool {
+        matches!(self, Consistency::Serial | Consistency::LocalSerial)
+    }
+}
+
+#[derive(Debug, Error)]
+#[error("Expected Consistency Serial or LocalSerial, got: {0}")]
+pub struct NonSerialConsistencyError(Consistency);
+
+impl TryFrom<Consistency> for SerialConsistency {
+    type Error = NonSerialConsistencyError;
+
+    fn try_from(c: Consistency) -> Result<Self, Self::Error> {
+        match c {
+            Consistency::Any
+            | Consistency::One
+            | Consistency::Two
+            | Consistency::Three
+            | Consistency::Quorum
+            | Consistency::All
+            | Consistency::LocalQuorum
+            | Consistency::EachQuorum
+            | Consistency::LocalOne => Err(NonSerialConsistencyError(c)),
+            Consistency::Serial => Ok(SerialConsistency::Serial),
+            Consistency::LocalSerial => Ok(SerialConsistency::LocalSerial),
+        }
+    }
 }
 
 impl std::fmt::Display for Consistency {
@@ -59,15 +86,6 @@ impl std::fmt::Display for SerialConsistency {
     }
 }
 
-impl std::fmt::Display for LegacyConsistency {
-    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
-        match self {
-            Self::Regular(c) => c.fmt(f),
-            Self::Serial(c) => c.fmt(f),
-        }
-    }
-}
-
 impl From<std::num::TryFromIntError> for ParseError {
     fn from(_err: std::num::TryFromIntError) -> Self {
         ParseError::BadIncomingData("Integer conversion out of range".to_string())
@@ -439,18 +457,10 @@ fn type_uuid() {
     assert_eq!(u, u2);
 }
 
-pub fn read_consistency(buf: &mut &[u8]) -> Result<LegacyConsistency, ParseError> {
+pub fn read_consistency(buf: &mut &[u8]) -> Result<Consistency, ParseError> {
     let raw = read_short(buf)?;
-    let parsed = match Consistency::try_from(raw) {
-        Ok(c) => LegacyConsistency::Regular(c),
-        Err(_) => {
-            let parsed_serial = SerialConsistency::try_from(raw).map_err(|_| {
-                ParseError::BadIncomingData(format!("unknown consistency: {}", raw))
-            })?;
-            LegacyConsistency::Serial(parsed_serial)
-        }
-    };
-    Ok(parsed)
+    Consistency::try_from(raw)
+        .map_err(|_| ParseError::BadIncomingData(format!("unknown consistency: {}", raw)))
 }
 
 pub fn write_consistency(c: Consistency, buf: &mut impl BufMut) {
@@ -467,7 +477,7 @@ fn type_consistency() {
     let mut buf = Vec::new();
     write_consistency(c, &mut buf);
     let c2 = read_consistency(&mut &*buf).unwrap();
-    assert_eq!(LegacyConsistency::Regular(c), c2);
+    assert_eq!(c, c2);
 
     let c: i16 = 0x1234;
     buf.clear();
diff --git a/scylla-cql/src/frame/value.rs b/scylla-cql/src/frame/value.rs
index 9cc482c2..9e927be3 100644
--- a/scylla-cql/src/frame/value.rs
+++ b/scylla-cql/src/frame/value.rs
@@ -504,6 +504,12 @@ impl Value for &str {
 }
 
 impl Value for Vec<u8> {
+    fn serialize(&self, buf: &mut Vec<u8>) -> Result<(), ValueTooBig> {
+        <&[u8] as Value>::serialize(&self.as_slice(), buf)
+    }
+}
+
+impl Value for &[u8] {
     fn serialize(&self, buf: &mut Vec<u8>) -> Result<(), ValueTooBig> {
         let val_len: i32 = self.len().try_into().map_err(|_| ValueTooBig)?;
         buf.put_i32(val_len);
diff --git a/scylla-proxy/src/actions.rs b/scylla-proxy/src/actions.rs
index 4e617bfb..ef0945ba 100644
--- a/scylla-proxy/src/actions.rs
+++ b/scylla-proxy/src/actions.rs
@@ -9,7 +9,6 @@ use crate::frame::{
 };
 use scylla_cql::{
     errors::{DbError, WriteType},
-    frame::types::LegacyConsistency,
     Consistency,
 };
 
@@ -427,7 +426,7 @@ impl ExampleDbErrors {
     }
     pub fn unavailable() -> DbError {
         DbError::Unavailable {
-            consistency: LegacyConsistency::Regular(Consistency::One),
+            consistency: Consistency::One,
             required: 2,
             alive: 1,
         }
@@ -443,7 +442,7 @@ impl ExampleDbErrors {
     }
     pub fn read_timeout() -> DbError {
         DbError::ReadTimeout {
-            consistency: LegacyConsistency::Regular(Consistency::One),
+            consistency: Consistency::One,
             received: 2,
             required: 3,
             data_present: true,
@@ -451,7 +450,7 @@ impl ExampleDbErrors {
     }
     pub fn write_timeout() -> DbError {
         DbError::WriteTimeout {
-            consistency: LegacyConsistency::Regular(Consistency::One),
+            consistency: Consistency::One,
             received: 2,
             required: 3,
             write_type: WriteType::UnloggedBatch,
@@ -459,7 +458,7 @@ impl ExampleDbErrors {
     }
     pub fn read_failure() -> DbError {
         DbError::ReadFailure {
-            consistency: LegacyConsistency::Regular(Consistency::One),
+            consistency: Consistency::One,
             received: 2,
             required: 3,
             data_present: true,
@@ -468,7 +467,7 @@ impl ExampleDbErrors {
     }
     pub fn write_failure() -> DbError {
         DbError::WriteFailure {
-            consistency: LegacyConsistency::Regular(Consistency::One),
+            consistency: Consistency::One,
             received: 2,
             required: 3,
             write_type: WriteType::UnloggedBatch,
diff --git a/scylla-proxy/src/frame.rs b/scylla-proxy/src/frame.rs
index cf61a6b0..5aab0340 100644
--- a/scylla-proxy/src/frame.rs
+++ b/scylla-proxy/src/frame.rs
@@ -6,8 +6,6 @@ use scylla_cql::frame::protocol_features::ProtocolFeatures;
 use scylla_cql::frame::request::Request;
 pub use scylla_cql::frame::request::RequestOpcode;
 pub use scylla_cql::frame::response::ResponseOpcode;
-use scylla_cql::frame::types::LegacyConsistency;
-use scylla_cql::Consistency;
 use scylla_cql::{errors::DbError, frame::types};
 use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
 
@@ -147,19 +145,13 @@ impl ResponseFrame {
 }
 
 fn serialize_error_specific_fields(buf: &mut BytesMut, error: DbError) -> Result<(), ParseError> {
-    fn unwrap_cl(c: LegacyConsistency) -> Consistency {
-        match c {
-            types::LegacyConsistency::Regular(c) => c,
-            types::LegacyConsistency::Serial(_) => unreachable!(),
-        }
-    }
     match error {
         DbError::Unavailable {
             consistency,
             required,
             alive,
         } => {
-            types::write_consistency(unwrap_cl(consistency), buf);
+            types::write_consistency(consistency, buf);
             types::write_int(required, buf);
             types::write_int(alive, buf);
         }
@@ -169,7 +161,7 @@ fn serialize_error_specific_fields(buf: &mut BytesMut, error: DbError) -> Result
             required,
             write_type,
         } => {
-            types::write_consistency(unwrap_cl(consistency), buf);
+            types::write_consistency(consistency, buf);
             types::write_int(received, buf);
             types::write_int(required, buf);
             types::write_string(write_type.as_str(), buf)?;
@@ -180,7 +172,7 @@ fn serialize_error_specific_fields(buf: &mut BytesMut, error: DbError) -> Result
             required,
             data_present,
         } => {
-            types::write_consistency(unwrap_cl(consistency), buf);
+            types::write_consistency(consistency, buf);
             types::write_int(received, buf);
             types::write_int(required, buf);
             buf.put_u8(u8::from(data_present));
@@ -192,7 +184,7 @@ fn serialize_error_specific_fields(buf: &mut BytesMut, error: DbError) -> Result
             numfailures,
             data_present,
         } => {
-            types::write_consistency(unwrap_cl(consistency), buf);
+            types::write_consistency(consistency, buf);
             types::write_int(received, buf);
             types::write_int(required, buf);
             types::write_int(numfailures, buf);
@@ -205,7 +197,7 @@ fn serialize_error_specific_fields(buf: &mut BytesMut, error: DbError) -> Result
             numfailures,
             write_type,
         } => {
-            types::write_consistency(unwrap_cl(consistency), buf);
+            types::write_consistency(consistency, buf);
             types::write_int(received, buf);
             types::write_int(required, buf);
             types::write_int(numfailures, buf);
diff --git a/scylla/benches/benchmark.rs b/scylla/benches/benchmark.rs
index 08c13a3f..c6058923 100644
--- a/scylla/benches/benchmark.rs
+++ b/scylla/benches/benchmark.rs
@@ -1,7 +1,11 @@
 use criterion::{criterion_group, criterion_main, Criterion};
 
 use bytes::BytesMut;
-use scylla::frame::types;
+use scylla::{
+    frame::types,
+    frame::value::ValueList,
+    transport::partitioner::{calculate_token_for_partition_key, Murmur3Partitioner},
+};
 
 fn types_benchmark(c: &mut Criterion) {
     let mut buf = BytesMut::with_capacity(64);
@@ -35,5 +39,57 @@ fn types_benchmark(c: &mut Criterion) {
     });
 }
 
-criterion_group!(benches, types_benchmark);
+fn calculate_token_bench(c: &mut Criterion) {
+    let simple_pk = ("I'm prepared!!!",);
+    let serialized_simple_pk = simple_pk.serialized().unwrap().into_owned();
+    let simple_pk_long_column = (
+        17_i32,
+        16_i32,
+        String::from_iter(std::iter::repeat('.').take(2000)),
+    );
+    let serialized_simple_pk_long_column = simple_pk_long_column.serialized().unwrap().into_owned();
+
+    let complex_pk = (17_i32, 16_i32, "I'm prepared!!!");
+    let serialized_complex_pk = complex_pk.serialized().unwrap().into_owned();
+    let complex_pk_long_column = (
+        17_i32,
+        16_i32,
+        String::from_iter(std::iter::repeat('.').take(2000)),
+    );
+    let serialized_values_long_column = complex_pk_long_column.serialized().unwrap().into_owned();
+
+    c.bench_function("calculate_token_from_partition_key simple pk", |b| {
+        b.iter(|| calculate_token_for_partition_key(&serialized_simple_pk, &Murmur3Partitioner))
+    });
+
+    c.bench_function(
+        "calculate_token_from_partition_key simple pk long column",
+        |b| {
+            b.iter(|| {
+                calculate_token_for_partition_key(
+                    &serialized_simple_pk_long_column,
+                    &Murmur3Partitioner,
+                )
+            })
+        },
+    );
+
+    c.bench_function("calculate_token_from_partition_key complex pk", |b| {
+        b.iter(|| calculate_token_for_partition_key(&serialized_complex_pk, &Murmur3Partitioner))
+    });
+
+    c.bench_function(
+        "calculate_token_from_partition_key complex pk long column",
+        |b| {
+            b.iter(|| {
+                calculate_token_for_partition_key(
+                    &serialized_values_long_column,
+                    &Murmur3Partitioner,
+                )
+            })
+        },
+    );
+}
+
+criterion_group!(benches, types_benchmark, calculate_token_bench);
 criterion_main!(benches);
diff --git a/scylla/src/history.rs b/scylla/src/history.rs
index 4767690e..6b1bd1a9 100644
--- a/scylla/src/history.rs
+++ b/scylla/src/history.rs
@@ -468,7 +468,6 @@ mod tests {
     use futures::StreamExt;
     use scylla_cql::{
         errors::{DbError, QueryError},
-        frame::types::LegacyConsistency,
         Consistency,
     };
 
@@ -576,7 +575,7 @@ mod tests {
     fn unavailable_error() -> QueryError {
         QueryError::DbError(
             DbError::Unavailable {
-                consistency: LegacyConsistency::Regular(Consistency::Quorum),
+                consistency: Consistency::Quorum,
                 required: 2,
                 alive: 1,
             },
diff --git a/scylla/src/routing.rs b/scylla/src/routing.rs
index 5df667fe..d6fff764 100644
--- a/scylla/src/routing.rs
+++ b/scylla/src/routing.rs
@@ -13,10 +13,10 @@ pub type Shard = u32;
 pub type ShardCount = NonZeroU16;
 
 #[derive(PartialEq, Eq, Clone, Debug)]
-pub struct ShardInfo {
-    pub shard: u16,
-    pub nr_shards: ShardCount,
-    pub msb_ignore: u8,
+pub(crate) struct ShardInfo {
+    pub(crate) shard: u16,
+    pub(crate) nr_shards: ShardCount,
+    pub(crate) msb_ignore: u8,
 }
 
 #[derive(PartialEq, Eq, Clone, Debug)]
@@ -33,7 +33,7 @@ impl std::str::FromStr for Token {
 }
 
 impl ShardInfo {
-    pub fn new(shard: u16, nr_shards: ShardCount, msb_ignore: u8) -> Self {
+    pub(crate) fn new(shard: u16, nr_shards: ShardCount, msb_ignore: u8) -> Self {
         ShardInfo {
             shard,
             nr_shards,
@@ -41,7 +41,7 @@ impl ShardInfo {
         }
     }
 
-    pub fn get_sharder(&self) -> Sharder {
+    pub(crate) fn get_sharder(&self) -> Sharder {
         Sharder::new(self.nr_shards, self.msb_ignore)
     }
 }
diff --git a/scylla/src/statement/prepared_statement.rs b/scylla/src/statement/prepared_statement.rs
index 1bb41135..b57d5d4b 100644
--- a/scylla/src/statement/prepared_statement.rs
+++ b/scylla/src/statement/prepared_statement.rs
@@ -1,5 +1,5 @@
-use byteorder::{BigEndian, ReadBytesExt};
-use bytes::{BufMut, Bytes, BytesMut};
+use bytes::{Bytes, BytesMut};
+use scylla_cql::errors::{BadQuery, QueryError};
 use smallvec::{smallvec, SmallVec};
 use std::convert::TryInto;
 use std::sync::Arc;
@@ -7,8 +7,7 @@ use std::time::Duration;
 use thiserror::Error;
 use uuid::Uuid;
 
-use scylla_cql::frame::frame_errors::ParseError;
-use scylla_cql::frame::response::result::{deser_cql_value, CqlValue};
+use scylla_cql::frame::response::result::ColumnSpec;
 
 use super::StatementConfig;
 use crate::frame::response::result::PreparedMetadata;
@@ -16,8 +15,9 @@ use crate::frame::types::{Consistency, SerialConsistency};
 use crate::frame::value::SerializedValues;
 use crate::history::HistoryListener;
 use crate::retry_policy::RetryPolicy;
+use crate::routing::Token;
 use crate::transport::execution_profile::ExecutionProfileHandle;
-use crate::transport::partitioner::PartitionerName;
+use crate::transport::partitioner::{Partitioner, PartitionerHasher, PartitionerName};
 
 /// Represents a statement prepared on the server.
 #[derive(Debug)]
@@ -124,68 +124,76 @@ impl PreparedStatement {
 
     /// Computes the partition key of the target table from given values —
     /// it assumes that all partition key columns are passed in values.
-    /// Partition keys have a specific serialization rules.
+    /// Partition keys have specific serialization rules.
     /// Ref: <https://github.com/scylladb/scylla/blob/40adf38915b6d8f5314c621a94d694d172360833/compound_compat.hh#L33-L47>
+    ///
+    /// Note: this is no longer required to compute a token. This is because partitioners
+    /// are now stateful and stream-based, so they do not require materialised partition key.
+    /// Therefore, for computing a token, there are more efficient methods, such as
+    /// [Self::calculate_token()].
     pub fn compute_partition_key(
         &self,
         bound_values: &SerializedValues,
     ) -> Result<Bytes, PartitionKeyError> {
+        let partition_key = self.extract_partition_key(bound_values)?;
         let mut buf = BytesMut::new();
+        let mut writer = |chunk: &[u8]| buf.extend_from_slice(chunk);
 
-        // Single-value partition key case
-        if self.get_prepared_metadata().pk_indexes.len() == 1 {
-            let pk_index = self.get_prepared_metadata().pk_indexes[0].index;
-            if let Some(v) = bound_values
-                .iter()
-                .nth(pk_index as usize)
-                .ok_or_else(|| PartitionKeyError::NoPkIndexValue(pk_index, bound_values.len()))?
-            {
-                buf.extend_from_slice(v);
-            }
-            return Ok(buf.into());
-        }
+        partition_key.write_encoded_partition_key(&mut writer)?;
 
-        // Composite partition key case
+        Ok(buf.freeze())
+    }
 
-        // Iterate on values using sorted pk_indexes (see deser_prepared_metadata),
-        // and use PartitionKeyIndex.sequence to insert the value in pk_values with the correct order.
-        // At the same time, compute the size of the buffer to reserve it before writing in it.
-        let mut pk_values: SmallVec<[_; 8]> =
-            smallvec![None; self.get_prepared_metadata().pk_indexes.len()];
-        let mut values_iter = bound_values.iter();
-        let mut buf_size = 0;
-        // pk_indexes contains the indexes of the partition key value, so the current offset of the
-        // iterator must be kept, in order to compute the next position of the pk in the iterator.
-        // At each iteration values_iter.nth(0) will roughly correspond to values[values_iter_offset],
-        // so values[pk_index.index] will be retrieved with values_iter.nth(pk_index.index - values_iter_offset)
-        let mut values_iter_offset = 0;
-        for pk_index in self.get_prepared_metadata().pk_indexes.iter().copied() {
-            // Find value matching current pk_index
-            let next_val = values_iter
-                .nth((pk_index.index - values_iter_offset) as usize)
-                .ok_or_else(|| {
-                    PartitionKeyError::NoPkIndexValue(pk_index.index, bound_values.len())
-                })?;
-            // Add it in sequence order to pk_values
-            if let Some(v) = next_val {
-                pk_values[pk_index.sequence as usize] = Some(v);
-                buf_size += std::mem::size_of::<u16>() + v.len() + std::mem::size_of::<u8>();
-            }
-            values_iter_offset = pk_index.index + 1;
-        }
-        // Add values' bytes
-        buf.reserve(buf_size);
-        for v in pk_values.iter().flatten() {
-            let v_len_u16: u16 = v
-                .len()
-                .try_into()
-                .map_err(|_| PartitionKeyError::ValueTooLong(v.len()))?;
-
-            buf.put_u16(v_len_u16);
-            buf.extend_from_slice(v);
-            buf.put_u8(0);
+    /// Determines which values consistute the partition key and puts them in order.
+    ///
+    /// This is a preparation step necessary for calculating token based on a prepared statement.
+    pub(crate) fn extract_partition_key<'ps>(
+        &'ps self,
+        bound_values: &'ps SerializedValues,
+    ) -> Result<PartitionKey, PartitionKeyExtractionError> {
+        PartitionKey::new(self.get_prepared_metadata(), bound_values)
+    }
+
+    pub(crate) fn extract_partition_key_and_calculate_token<'ps>(
+        &'ps self,
+        partitioner_name: &'ps PartitionerName,
+        serialized_values: &'ps SerializedValues,
+    ) -> Result<Option<(PartitionKey<'ps>, Token)>, QueryError> {
+        if !self.is_token_aware() {
+            return Ok(None);
         }
-        Ok(buf.into())
+
+        let partition_key =
+            self.extract_partition_key(serialized_values)
+                .map_err(|err| match err {
+                    PartitionKeyExtractionError::NoPkIndexValue(_, _) => {
+                        QueryError::ProtocolError("No pk indexes - can't calculate token")
+                    }
+                })?;
+        let token = partition_key
+            .calculate_token(partitioner_name)
+            .map_err(|err| match err {
+                TokenCalculationError::ValueTooLong(values_len) => {
+                    QueryError::BadQuery(BadQuery::ValuesTooLongForKey(values_len, u16::MAX.into()))
+                }
+            })?;
+
+        Ok(Some((partition_key, token)))
+    }
+
+    /// Calculates the token for given prepared statement and serialized values.
+    ///
+    /// Returns the token that would be computed for executing the provided
+    /// prepared statement with the provided values.
+    // As this function creates a `PartitionKey`, it is intended rather for external usage (by users).
+    // For internal purposes, `PartitionKey::calculate_token()` is preferred, as `PartitionKey`
+    // is either way used internally, among others for display in traces.
+    pub fn calculate_token(
+        &self,
+        serialized_values: &SerializedValues,
+    ) -> Result<Option<Token>, QueryError> {
+        self.extract_partition_key_and_calculate_token(&self.partitioner_name, serialized_values)
+            .map(|opt| opt.map(|(_pk, token)| token))
     }
 
     /// Returns the name of the keyspace this statement is operating on.
@@ -329,91 +337,132 @@ impl PreparedStatement {
 }
 
 #[derive(Clone, Debug, Error, PartialEq, Eq, PartialOrd, Ord)]
-pub enum PartitionKeyError {
+pub enum PartitionKeyExtractionError {
     #[error("No value with given pk_index! pk_index: {0}, values.len(): {1}")]
     NoPkIndexValue(u16, i16),
+}
+
+#[derive(Clone, Debug, Error, PartialEq, Eq, PartialOrd, Ord)]
+pub enum TokenCalculationError {
     #[error("Value bytes too long to create partition key, max 65 535 allowed! value.len(): {0}")]
     ValueTooLong(usize),
 }
 
-// The PartitionKeyDecoder reverses the process of PreparedStatement::compute_partition_key:
-// it returns the consecutive values of partition key column that were encoded
-// by the function into the Bytes object, additionally decoding them to CqlValue.
-//
-// The format follows the description here:
-// <https://github.com/scylladb/scylla/blob/40adf38915b6d8f5314c621a94d694d172360833/compound_compat.hh#L33-L47>
-//
-// TODO: Currently, if there is a null value specified for a partition column,
-// it will be skipped when creating the serialized partition key. We should
-// not create such partition keys in the future, i.e. fail the request or
-// route it to a random replica instead and let the DB reject it. In the
-// meantime, this struct will return some garbage data while it tries to
-// decode the key, but nothing bad like a panic should happen otherwise.
-// The struct is currently only used for printing the partition key, so that's
-// completely fine.
-#[derive(Clone, Copy)]
-pub(crate) struct PartitionKeyDecoder<'pk> {
-    prepared_metadata: &'pk PreparedMetadata,
-    partition_key: &'pk [u8],
-    value_index: usize,
+#[derive(Clone, Debug, Error, PartialEq, Eq, PartialOrd, Ord)]
+pub enum PartitionKeyError {
+    #[error(transparent)]
+    PartitionKeyExtraction(PartitionKeyExtractionError),
+    #[error(transparent)]
+    TokenCalculation(TokenCalculationError),
+}
+
+impl From<PartitionKeyExtractionError> for PartitionKeyError {
+    fn from(err: PartitionKeyExtractionError) -> Self {
+        Self::PartitionKeyExtraction(err)
+    }
 }
 
-impl<'pk> PartitionKeyDecoder<'pk> {
-    pub(crate) fn new(prepared_metadata: &'pk PreparedMetadata, partition_key: &'pk [u8]) -> Self {
-        Self {
-            prepared_metadata,
-            partition_key,
-            value_index: 0,
-        }
+impl From<TokenCalculationError> for PartitionKeyError {
+    fn from(err: TokenCalculationError) -> Self {
+        Self::TokenCalculation(err)
     }
 }
 
-impl<'pk> Iterator for PartitionKeyDecoder<'pk> {
-    type Item = Result<CqlValue, ParseError>;
+pub(crate) type PartitionKeyValue<'ps> = (&'ps [u8], &'ps ColumnSpec);
 
-    fn next(&mut self) -> Option<Self::Item> {
-        if self.value_index >= self.prepared_metadata.pk_indexes.len() {
-            return None;
-        }
+pub(crate) struct PartitionKey<'ps> {
+    pk_values: SmallVec<[Option<PartitionKeyValue<'ps>>; PartitionKey::SMALLVEC_ON_STACK_SIZE]>,
+}
 
-        let col_idx = self.prepared_metadata.pk_indexes[self.value_index].index as usize;
-        let spec = &self.prepared_metadata.col_specs[col_idx];
+impl<'ps> PartitionKey<'ps> {
+    const SMALLVEC_ON_STACK_SIZE: usize = 8;
 
-        let cell = if self.prepared_metadata.pk_indexes.len() == 1 {
-            Ok(self.partition_key)
-        } else {
-            self.parse_cell()
-        };
-        self.value_index += 1;
-        Some(cell.and_then(|mut cell| deser_cql_value(&spec.typ, &mut cell)))
+    fn new(
+        prepared_metadata: &'ps PreparedMetadata,
+        bound_values: &'ps SerializedValues,
+    ) -> Result<Self, PartitionKeyExtractionError> {
+        // Iterate on values using sorted pk_indexes (see deser_prepared_metadata),
+        // and use PartitionKeyIndex.sequence to insert the value in pk_values with the correct order.
+        let mut pk_values: SmallVec<[_; PartitionKey::SMALLVEC_ON_STACK_SIZE]> =
+            smallvec![None; prepared_metadata.pk_indexes.len()];
+        let mut values_iter = bound_values.iter();
+        // pk_indexes contains the indexes of the partition key value, so the current offset of the
+        // iterator must be kept, in order to compute the next position of the pk in the iterator.
+        // At each iteration values_iter.nth(0) will roughly correspond to values[values_iter_offset],
+        // so values[pk_index.index] will be retrieved with values_iter.nth(pk_index.index - values_iter_offset)
+        let mut values_iter_offset = 0;
+        for pk_index in prepared_metadata.pk_indexes.iter().copied() {
+            // Find value matching current pk_index
+            let next_val = values_iter
+                .nth((pk_index.index - values_iter_offset) as usize)
+                .ok_or_else(|| {
+                    PartitionKeyExtractionError::NoPkIndexValue(pk_index.index, bound_values.len())
+                })?;
+            // Add it in sequence order to pk_values
+            if let Some(v) = next_val {
+                let spec = &prepared_metadata.col_specs[pk_index.index as usize];
+                pk_values[pk_index.sequence as usize] = Some((v, spec));
+            }
+            values_iter_offset = pk_index.index + 1;
+        }
+        Ok(Self { pk_values })
     }
-}
 
-impl<'pk> PartitionKeyDecoder<'pk> {
-    fn parse_cell(&mut self) -> Result<&'pk [u8], ParseError> {
-        let buf = &mut self.partition_key;
-        let len = buf.read_u16::<BigEndian>()? as usize;
-        if buf.len() < len {
-            return Err(ParseError::IoError(std::io::Error::new(
-                std::io::ErrorKind::UnexpectedEof,
-                "value too short",
-            )));
+    pub(crate) fn iter(&self) -> impl Iterator<Item = PartitionKeyValue<'ps>> + Clone + '_ {
+        self.pk_values.iter().flatten().copied()
+    }
+
+    fn write_encoded_partition_key(
+        &self,
+        writer: &mut impl FnMut(&[u8]),
+    ) -> Result<(), TokenCalculationError> {
+        let mut pk_val_iter = self.iter().map(|(val, _spec)| val);
+        if let Some(first_value) = pk_val_iter.next() {
+            if let Some(second_value) = pk_val_iter.next() {
+                // Composite partition key case
+                for value in std::iter::once(first_value)
+                    .chain(std::iter::once(second_value))
+                    .chain(pk_val_iter)
+                {
+                    let v_len_u16: u16 = value
+                        .len()
+                        .try_into()
+                        .map_err(|_| TokenCalculationError::ValueTooLong(value.len()))?;
+                    writer(&v_len_u16.to_be_bytes());
+                    writer(value);
+                    writer(&[0u8]);
+                }
+            } else {
+                // Single-value partition key case
+                writer(first_value);
+            }
         }
-        let col = &buf[..len];
-        *buf = &buf[len..];
-        buf.read_u8()?;
-        Ok(col)
+        Ok(())
+    }
+
+    pub(crate) fn calculate_token(
+        &self,
+        partitioner_name: &PartitionerName,
+    ) -> Result<Token, TokenCalculationError> {
+        let mut partitioner_hasher = partitioner_name.build_hasher();
+        let mut writer = |chunk: &[u8]| partitioner_hasher.write(chunk);
+
+        self.write_encoded_partition_key(&mut writer)?;
+
+        Ok(partitioner_hasher.finish())
     }
 }
 
 #[cfg(test)]
 mod tests {
-    use bytes::{BufMut, Bytes, BytesMut};
-    use scylla_cql::frame::response::result::{
-        ColumnSpec, ColumnType, CqlValue, PartitionKeyIndex, PreparedMetadata, TableSpec,
+    use scylla_cql::frame::{
+        response::result::{
+            ColumnSpec, ColumnType, PartitionKeyIndex, PreparedMetadata, TableSpec,
+        },
+        value::SerializedValues,
     };
 
-    use super::PartitionKeyDecoder;
+    use crate::prepared_statement::PartitionKey;
 
     fn make_meta(
         cols: impl IntoIterator<Item = ColumnType>,
@@ -432,14 +481,15 @@ mod tests {
                 typ,
             })
             .collect();
-        let pk_indexes = idx
+        let mut pk_indexes = idx
             .into_iter()
             .enumerate()
             .map(|(sequence, index)| PartitionKeyIndex {
                 index: index as u16,
                 sequence: sequence as u16,
             })
-            .collect();
+            .collect::<Vec<_>>();
+        pk_indexes.sort_unstable_by_key(|pki| pki.index);
         PreparedMetadata {
             flags: 0,
             col_count: col_specs.len(),
@@ -448,53 +498,8 @@ mod tests {
         }
     }
 
-    fn make_key<'pk>(cols: impl IntoIterator<Item = &'pk [u8]>) -> Bytes {
-        let cols: Vec<_> = cols.into_iter().collect();
-        // TODO: Use compute_partition_key or one of the variants
-        // after they are moved to a more sensible place
-        // instead of constructing the PK manually
-        let mut b = BytesMut::new();
-        if cols.len() == 1 {
-            b.extend_from_slice(cols[0]);
-        } else {
-            for c in cols {
-                b.put_i16(c.len() as i16);
-                b.extend_from_slice(c);
-                b.put_u8(0);
-            }
-        }
-        b.freeze()
-    }
-
-    #[test]
-    fn test_pk_decoder_single_column() {
-        let meta = make_meta([ColumnType::Int], [0]);
-        let pk = make_key([0i32.to_be_bytes().as_ref()]);
-        let cells = PartitionKeyDecoder::new(&meta, &pk)
-            .collect::<Result<Vec<_>, _>>()
-            .unwrap();
-        assert_eq!(cells, vec![CqlValue::Int(0)]);
-    }
-
-    #[test]
-    fn test_pk_decoder_multiple_columns() {
-        let meta = make_meta(std::iter::repeat(ColumnType::Int).take(3), [0, 1, 2]);
-        let pk = make_key([
-            12i32.to_be_bytes().as_ref(),
-            34i32.to_be_bytes().as_ref(),
-            56i32.to_be_bytes().as_ref(),
-        ]);
-        let cells = PartitionKeyDecoder::new(&meta, &pk)
-            .collect::<Result<Vec<_>, _>>()
-            .unwrap();
-        assert_eq!(
-            cells,
-            vec![CqlValue::Int(12), CqlValue::Int(34), CqlValue::Int(56)],
-        );
-    }
-
     #[test]
-    fn test_pk_decoder_multiple_columns_shuffled() {
+    fn test_partition_key_multiple_columns_shuffled() {
         let meta = make_meta(
             [
                 ColumnType::TinyInt,
@@ -505,21 +510,22 @@ mod tests {
             ],
             [4, 0, 3],
         );
-        let pk = make_key([
-            &[1, 2, 3, 4, 5],
-            67i8.to_be_bytes().as_ref(),
-            89i64.to_be_bytes().as_ref(),
-        ]);
-        let cells = PartitionKeyDecoder::new(&meta, &pk)
-            .collect::<Result<Vec<_>, _>>()
-            .unwrap();
+        let mut values = SerializedValues::new();
+        values.add_value(&67i8).unwrap();
+        values.add_value(&42i16).unwrap();
+        values.add_value(&23i32).unwrap();
+        values.add_value(&89i64).unwrap();
+        values.add_value(&[1u8, 2, 3, 4, 5]).unwrap();
+
+        let pk = PartitionKey::new(&meta, &values).unwrap();
+        let pk_cols = Vec::from_iter(pk.iter());
         assert_eq!(
-            cells,
+            pk_cols,
             vec![
-                CqlValue::Blob(vec![1, 2, 3, 4, 5]),
-                CqlValue::TinyInt(67),
-                CqlValue::BigInt(89),
-            ],
+                ([1u8, 2, 3, 4, 5].as_slice(), &meta.col_specs[4]),
+                (67i8.to_be_bytes().as_ref(), &meta.col_specs[0]),
+                (89i64.to_be_bytes().as_ref(), &meta.col_specs[3]),
+            ]
         );
     }
 }
diff --git a/scylla/src/transport/cluster.rs b/scylla/src/transport/cluster.rs
index 0e6e9047..4e436d29 100644
--- a/scylla/src/transport/cluster.rs
+++ b/scylla/src/transport/cluster.rs
@@ -1,6 +1,7 @@
 /// Cluster manages up to date information and connections to database nodes
 use crate::frame::response::event::{Event, StatusChangeEvent};
 use crate::frame::value::ValueList;
+use crate::prepared_statement::TokenCalculationError;
 use crate::routing::Token;
 use crate::transport::host_filter::HostFilter;
 use crate::transport::{
@@ -13,7 +14,6 @@ use crate::transport::{
 };
 
 use arc_swap::ArcSwap;
-use bytes::{BufMut, Bytes, BytesMut};
 use futures::future::join_all;
 use futures::{future::RemoteHandle, FutureExt};
 use itertools::Itertools;
@@ -36,6 +36,7 @@ pub struct ContactPoint {
 }
 
 use super::locator::ReplicaLocator;
+use super::partitioner::calculate_token_for_partition_key;
 use super::topology::Strategy;
 
 /// Cluster manages up to date information and connections to database nodes.
@@ -123,6 +124,10 @@ struct ClusterWorker {
     // The host filter determines towards which nodes we should open
     // connections
     host_filter: Option<Arc<dyn HostFilter>>,
+
+    // This value determines how frequently the cluster
+    // worker will refresh the cluster metadata
+    cluster_metadata_refresh_interval: Duration,
 }
 
 #[derive(Debug)]
@@ -143,6 +148,7 @@ impl Cluster {
         keyspaces_to_fetch: Vec<String>,
         fetch_schema_metadata: bool,
         host_filter: Option<Arc<dyn HostFilter>>,
+        cluster_metadata_refresh_interval: Duration,
     ) -> Result<Cluster, QueryError> {
         let (refresh_sender, refresh_receiver) = tokio::sync::mpsc::channel(32);
         let (use_keyspace_sender, use_keyspace_receiver) = tokio::sync::mpsc::channel(32);
@@ -184,6 +190,7 @@ impl Cluster {
             used_keyspace: None,
 
             host_filter,
+            cluster_metadata_refresh_interval,
         };
 
         let (fut, worker_handle) = worker.work().remote_handle();
@@ -237,29 +244,6 @@ impl Cluster {
 
         response_receiver.await.unwrap() // ClusterWorker always responds
     }
-
-    /// Returns nonempty list of working connections to all shards
-    pub(crate) 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;
-
-        for node in peers.values() {
-            match node.get_working_connections() {
-                Ok(conns) => result.extend(conns),
-                Err(e) => last_error = Some(e),
-            }
-        }
-
-        if result.is_empty() {
-            return Err(last_error.unwrap()); // By invariant peers is nonempty
-        }
-
-        Ok(result)
-    }
 }
 
 impl ClusterData {
@@ -269,7 +253,7 @@ impl ClusterData {
             datacenter.rack_count = datacenter
                 .nodes
                 .iter()
-                .filter_map(|node| node.rack.clone())
+                .filter_map(|node| node.rack.as_ref())
                 .unique()
                 .count();
         }
@@ -301,19 +285,27 @@ impl ClusterData {
             // 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 known_peers.get(&peer.host_id) {
+            let peer_host_id = peer.host_id;
+            let peer_address = peer.address;
+            let peer_tokens;
+
+            let node: Arc<Node> = match known_peers.get(&peer_host_id) {
                 Some(node) if node.datacenter == peer.datacenter && node.rack == peer.rack => {
-                    if node.address == peer.address {
+                    let (peer_endpoint, tokens) = peer.into_peer_endpoint_and_tokens();
+                    peer_tokens = tokens;
+                    if node.address == peer_address {
                         node.clone()
                     } else {
                         // If IP changes, the Node struct is recreated, but the underlying pool is preserved and notified about the IP change.
-                        Arc::new(Node::inherit_with_ip_changed(node, peer.to_peer_endpoint()))
+                        Arc::new(Node::inherit_with_ip_changed(node, peer_endpoint))
                     }
                 }
                 _ => {
                     let is_enabled = host_filter.map_or(true, |f| f.accept(&peer));
+                    let (peer_endpoint, tokens) = peer.into_peer_endpoint_and_tokens();
+                    peer_tokens = tokens;
                     Arc::new(Node::new(
-                        peer.to_peer_endpoint(),
+                        peer_endpoint,
                         pool_config.clone(),
                         used_keyspace.clone(),
                         is_enabled,
@@ -321,7 +313,7 @@ impl ClusterData {
                 }
             };
 
-            new_known_peers.insert(peer.host_id, node.clone());
+            new_known_peers.insert(peer_host_id, node.clone());
 
             if let Some(dc) = &node.datacenter {
                 match datacenters.get_mut(dc) {
@@ -336,7 +328,7 @@ impl ClusterData {
                 }
             }
 
-            for token in peer.tokens {
+            for token in peer_tokens {
                 ring.push((token, node.clone()));
             }
 
@@ -345,21 +337,18 @@ impl ClusterData {
 
         Self::update_rack_count(&mut datacenters);
 
-        let keyspace_strategies: Vec<Strategy> = metadata
-            .keyspaces
-            .values()
-            .map(|ks| ks.strategy.clone())
-            .collect();
-
-        let locator = tokio::task::spawn_blocking(move || {
-            ReplicaLocator::new(ring.into_iter(), keyspace_strategies.iter())
+        let keyspaces = metadata.keyspaces;
+        let (locator, keyspaces) = tokio::task::spawn_blocking(move || {
+            let keyspace_strategies = keyspaces.values().map(|ks| &ks.strategy);
+            let locator = ReplicaLocator::new(ring.into_iter(), keyspace_strategies);
+            (locator, keyspaces)
         })
         .await
         .unwrap();
 
         ClusterData {
             known_peers: new_known_peers,
-            keyspaces: metadata.keyspaces,
+            keyspaces,
             locator,
         }
     }
@@ -409,31 +398,13 @@ impl ClusterData {
             .and_then(|t| t.partitioner.as_deref())
             .and_then(PartitionerName::from_str)
             .unwrap_or_default();
-        let serialized_values = partition_key.serialized()?;
-        // Null values are skipped in computation; null values in partition key are unsound,
-        // but it is consistent with computation of prepared statements token.
-        let serialized_pk = match serialized_values.len() {
-            0 => None,
-            1 => serialized_values
-                .iter()
-                .next()
-                .unwrap()
-                .map(Bytes::copy_from_slice),
-            _ => {
-                let mut buf = BytesMut::new();
-                for value in serialized_values.iter().flatten() {
-                    let value_size = value
-                        .len()
-                        .try_into()
-                        .map_err(|_| BadQuery::ValuesTooLongForKey(value.len(), u16::MAX.into()))?;
-                    buf.put_u16(value_size);
-                    buf.extend_from_slice(value);
-                    buf.put_u8(0);
+
+        calculate_token_for_partition_key(&partition_key.serialized().unwrap(), &partitioner)
+            .map_err(|err| match err {
+                TokenCalculationError::ValueTooLong(values_len) => {
+                    BadQuery::ValuesTooLongForKey(values_len, u16::MAX.into())
                 }
-                Some(buf.into())
-            }
-        };
-        Ok(partitioner.hash(&serialized_pk.unwrap_or_default()))
+            })
     }
 
     /// Access to replicas owning a given token
@@ -476,13 +447,38 @@ impl ClusterData {
     pub fn replica_locator(&self) -> &ReplicaLocator {
         &self.locator
     }
+
+    /// Returns nonempty iterator of working connections to all shards.
+    pub(crate) fn iter_working_connections(
+        &self,
+    ) -> Result<impl Iterator<Item = Arc<Connection>> + '_, QueryError> {
+        // The returned iterator is nonempty by nonemptiness invariant of `self.known_peers`.
+        assert!(!self.known_peers.is_empty());
+        let mut peers_iter = self.known_peers.values();
+
+        // First we try to find the first working pool of connections.
+        // If none is found, return error.
+        let first_working_pool = peers_iter
+            .by_ref()
+            .map(|node| node.get_working_connections())
+            .find_or_first(Result::is_ok)
+            .expect("impossible: known_peers was asserted to be nonempty")?;
+
+        let remaining_pools_iter = peers_iter
+            .map(|node| node.get_working_connections())
+            .flatten_ok()
+            .flatten();
+
+        Ok(first_working_pool.into_iter().chain(remaining_pools_iter))
+        // By an invariant `self.known_peers` is nonempty, so the returned iterator
+        // is nonempty, too.
+    }
 }
 
 impl ClusterWorker {
     pub(crate) async fn work(mut self) {
         use tokio::time::Instant;
 
-        let refresh_duration = Duration::from_secs(60); // Refresh topology every 60 seconds
         let mut last_refresh_time = Instant::now();
 
         loop {
@@ -490,7 +486,7 @@ impl ClusterWorker {
 
             // Wait until it's time for the next refresh
             let sleep_until: Instant = last_refresh_time
-                .checked_add(refresh_duration)
+                .checked_add(self.cluster_metadata_refresh_interval)
                 .unwrap_or_else(Instant::now);
 
             let sleep_future = tokio::time::sleep_until(sleep_until);
@@ -591,13 +587,10 @@ impl ClusterWorker {
         cluster_data: Arc<ClusterData>,
         keyspace_name: &VerifiedKeyspaceName,
     ) -> Result<(), QueryError> {
-        let mut use_keyspace_futures = Vec::new();
-
-        for node in cluster_data.known_peers.values() {
-            let fut = node.use_keyspace(keyspace_name.clone());
-            use_keyspace_futures.push(fut);
-        }
-
+        let use_keyspace_futures = cluster_data
+            .known_peers
+            .values()
+            .map(|node| node.use_keyspace(keyspace_name.clone()));
         let use_keyspace_results: Vec<Result<(), QueryError>> =
             join_all(use_keyspace_futures).await;
 
diff --git a/scylla/src/transport/downgrading_consistency_retry_policy.rs b/scylla/src/transport/downgrading_consistency_retry_policy.rs
index 879e6b10..f55c25bc 100644
--- a/scylla/src/transport/downgrading_consistency_retry_policy.rs
+++ b/scylla/src/transport/downgrading_consistency_retry_policy.rs
@@ -1,6 +1,5 @@
 use scylla_cql::{
     errors::{DbError, QueryError, WriteType},
-    frame::types::LegacyConsistency,
     Consistency,
 };
 use tracing::debug;
@@ -56,7 +55,7 @@ impl Default for DowngradingConsistencyRetrySession {
 impl RetrySession for DowngradingConsistencyRetrySession {
     fn decide_should_retry(&mut self, query_info: QueryInfo) -> RetryDecision {
         let cl = match query_info.consistency {
-            LegacyConsistency::Serial(_) => {
+            Consistency::Serial | Consistency::LocalSerial => {
                 return match query_info.error {
                     QueryError::DbError(DbError::Unavailable { .. }, _) => {
                         // JAVA-764: if the requested consistency level is serial, it means that the operation failed at
@@ -67,7 +66,7 @@ impl RetrySession for DowngradingConsistencyRetrySession {
                     _ => RetryDecision::DontRetry,
                 };
             }
-            LegacyConsistency::Regular(cl) => cl,
+            cl => cl,
         };
 
         fn max_likely_to_work_cl(known_ok: i32, previous_cl: Consistency) -> RetryDecision {
@@ -209,7 +208,7 @@ mod tests {
         QueryInfo {
             error,
             is_idempotent,
-            consistency: LegacyConsistency::Regular(cl),
+            consistency: cl,
         }
     }
 
@@ -246,14 +245,14 @@ mod tests {
             DbError::Unauthorized,
             DbError::ConfigError,
             DbError::ReadFailure {
-                consistency: LegacyConsistency::Regular(Consistency::Two),
+                consistency: Consistency::Two,
                 received: 1,
                 required: 2,
                 numfailures: 1,
                 data_present: false,
             },
             DbError::WriteFailure {
-                consistency: LegacyConsistency::Regular(Consistency::Two),
+                consistency: Consistency::Two,
                 received: 1,
                 required: 2,
                 numfailures: 1,
@@ -361,7 +360,7 @@ mod tests {
         let alive = 1;
         let error = QueryError::DbError(
             DbError::Unavailable {
-                consistency: LegacyConsistency::Regular(Consistency::Two),
+                consistency: Consistency::Two,
                 required: 2,
                 alive,
             },
@@ -399,7 +398,7 @@ mod tests {
         // Enough responses and data_present == false - coordinator received only checksums
         let enough_responses_no_data = QueryError::DbError(
             DbError::ReadTimeout {
-                consistency: LegacyConsistency::Regular(Consistency::Two),
+                consistency: Consistency::Two,
                 received: 2,
                 required: 2,
                 data_present: false,
@@ -450,7 +449,7 @@ mod tests {
         // waiting for read-repair acknowledgement.
         let enough_responses_with_data = QueryError::DbError(
             DbError::ReadTimeout {
-                consistency: LegacyConsistency::Regular(Consistency::Two),
+                consistency: Consistency::Two,
                 received: 2,
                 required: 2,
                 data_present: true,
@@ -486,7 +485,7 @@ mod tests {
         let received = 1;
         let not_enough_responses_with_data = QueryError::DbError(
             DbError::ReadTimeout {
-                consistency: LegacyConsistency::Regular(Consistency::Two),
+                consistency: Consistency::Two,
                 received,
                 required: 2,
                 data_present: true,
@@ -547,7 +546,7 @@ mod tests {
             // WriteType == BatchLog
             let write_type_batchlog = QueryError::DbError(
                 DbError::WriteTimeout {
-                    consistency: LegacyConsistency::Regular(Consistency::Two),
+                    consistency: Consistency::Two,
                     received,
                     required,
                     write_type: WriteType::BatchLog,
@@ -590,7 +589,7 @@ mod tests {
             // WriteType == UnloggedBatch
             let write_type_unlogged_batch = QueryError::DbError(
                 DbError::WriteTimeout {
-                    consistency: LegacyConsistency::Regular(Consistency::Two),
+                    consistency: Consistency::Two,
                     received,
                     required,
                     write_type: WriteType::UnloggedBatch,
@@ -633,7 +632,7 @@ mod tests {
             // WriteType == other
             let write_type_other = QueryError::DbError(
                 DbError::WriteTimeout {
-                    consistency: LegacyConsistency::Regular(Consistency::Two),
+                    consistency: Consistency::Two,
                     received,
                     required,
                     write_type: WriteType::Simple,
diff --git a/scylla/src/transport/iterator.rs b/scylla/src/transport/iterator.rs
index 0cc95569..a9f5a7ec 100644
--- a/scylla/src/transport/iterator.rs
+++ b/scylla/src/transport/iterator.rs
@@ -22,7 +22,6 @@ use super::execution_profile::ExecutionProfileInner;
 use super::session::RequestSpan;
 use crate::cql_to_rust::{FromRow, FromRowError};
 
-use crate::frame::types::LegacyConsistency;
 use crate::frame::{
     response::{
         result,
@@ -31,7 +30,6 @@ use crate::frame::{
     value::SerializedValues,
 };
 use crate::history::{self, HistoryListener};
-use crate::routing::Token;
 use crate::statement::Consistency;
 use crate::statement::{prepared_statement::PreparedStatement, query::Query};
 use crate::transport::cluster::ClusterData;
@@ -40,6 +38,7 @@ use crate::transport::load_balancing::{self, RoutingInfo};
 use crate::transport::metrics::Metrics;
 use crate::transport::retry_policy::{QueryInfo, RetryDecision, RetrySession};
 use crate::transport::{Node, NodeRef};
+use crate::utils::unzip_option;
 use tracing::{trace, trace_span, warn, Instrument};
 use uuid::Uuid;
 
@@ -76,8 +75,6 @@ struct ReceivedPage {
 pub(crate) struct PreparedIteratorConfig {
     pub(crate) prepared: PreparedStatement,
     pub(crate) values: SerializedValues,
-    pub(crate) partition_key: Option<Bytes>,
-    pub(crate) token: Option<Token>,
     pub(crate) execution_profile: Arc<ExecutionProfileInner>,
     pub(crate) cluster_data: Arc<ClusterData>,
     pub(crate) metrics: Arc<Metrics>,
@@ -241,19 +238,29 @@ impl RowIterator {
 
         let parent_span = tracing::Span::current();
         let worker_task = async move {
+            let prepared_ref = &config.prepared;
+            let values_ref = &config.values;
+
+            let (partition_key, token) = match prepared_ref
+                .extract_partition_key_and_calculate_token(
+                    prepared_ref.get_partitioner_name(),
+                    values_ref,
+                ) {
+                Ok(res) => unzip_option(res),
+                Err(err) => {
+                    let (proof, _res) = ProvingSender::from(sender).send(Err(err)).await;
+                    return proof;
+                }
+            };
+
             let statement_info = RoutingInfo {
                 consistency,
                 serial_consistency,
-                token: config.token,
+                token,
                 keyspace: config.prepared.get_keyspace_name(),
                 is_confirmed_lwt: config.prepared.is_confirmed_lwt(),
             };
 
-            let prepared_ref = &config.prepared;
-            let values_ref = &config.values;
-            let partition_key = config.partition_key;
-            let token = config.token;
-
             let choose_connection = |node: Arc<Node>| async move {
                 match token {
                     Some(token) => node.connection_for_token(token).await,
@@ -294,8 +301,7 @@ impl RowIterator {
 
             let span_creator = move || {
                 let span = RequestSpan::new_prepared(
-                    prepared_ref.get_prepared_metadata(),
-                    partition_key.as_ref(),
+                    partition_key.as_ref().map(|pk| pk.iter()),
                     token,
                     serialized_values_size,
                 );
@@ -565,7 +571,7 @@ where
                 let query_info = QueryInfo {
                     error: &last_error,
                     is_idempotent: self.query_is_idempotent,
-                    consistency: LegacyConsistency::Regular(self.query_consistency),
+                    consistency: self.query_consistency,
                 };
 
                 let retry_decision = self.retry_session.decide_should_retry(query_info);
diff --git a/scylla/src/transport/locator/mod.rs b/scylla/src/transport/locator/mod.rs
index dda46626..db55b9fe 100644
--- a/scylla/src/transport/locator/mod.rs
+++ b/scylla/src/transport/locator/mod.rs
@@ -38,7 +38,7 @@ impl ReplicaLocator {
     /// Creates a new `ReplicaLocator` in which the specified replication strategies
     /// (`precompute_replica_sets_for`) will have its token ranges precomputed. This function can
     /// potentially be CPU-intensive (if a ring & replication factors in given strategies are big).
-    pub fn new<'a>(
+    pub(crate) fn new<'a>(
         ring_iter: impl Iterator<Item = (Token, Arc<Node>)>,
         precompute_replica_sets_for: impl Iterator<Item = &'a Strategy>,
     ) -> Self {
diff --git a/scylla/src/transport/partitioner.rs b/scylla/src/transport/partitioner.rs
index 887acb2f..9c8a5423 100644
--- a/scylla/src/transport/partitioner.rs
+++ b/scylla/src/transport/partitioner.rs
@@ -1,7 +1,9 @@
 use bytes::Buf;
 use std::num::Wrapping;
 
-use crate::routing::Token;
+use crate::{
+    frame::value::SerializedValues, prepared_statement::TokenCalculationError, routing::Token,
+};
 
 #[allow(clippy::upper_case_acronyms)]
 #[derive(Clone, PartialEq, Debug, Default)]
@@ -21,87 +23,229 @@ impl PartitionerName {
             None
         }
     }
+}
+
+impl Partitioner for PartitionerName {
+    type Hasher = PartitionerHasherAny;
+
+    fn build_hasher(&self) -> Self::Hasher {
+        match self {
+            PartitionerName::Murmur3 => {
+                PartitionerHasherAny::Murmur3(Murmur3Partitioner.build_hasher())
+            }
+            PartitionerName::CDC => PartitionerHasherAny::CDC(CDCPartitioner.build_hasher()),
+        }
+    }
+}
+
+#[allow(clippy::upper_case_acronyms)]
+pub(crate) enum PartitionerHasherAny {
+    Murmur3(Murmur3PartitionerHasher),
+    CDC(CDCPartitionerHasher),
+}
+
+impl PartitionerHasher for PartitionerHasherAny {
+    fn write(&mut self, pk_part: &[u8]) {
+        match self {
+            PartitionerHasherAny::Murmur3(h) => h.write(pk_part),
+            PartitionerHasherAny::CDC(h) => h.write(pk_part),
+        }
+    }
 
-    pub(crate) fn hash(&self, pk: &[u8]) -> Token {
+    fn finish(&self) -> Token {
         match self {
-            PartitionerName::Murmur3 => Murmur3Partitioner::hash(pk),
-            PartitionerName::CDC => CDCPartitioner::hash(pk),
+            PartitionerHasherAny::Murmur3(h) => h.finish(),
+            PartitionerHasherAny::CDC(h) => h.finish(),
         }
     }
 }
 
+/// A trait for creating instances of `PartitionHasher`, which ultimately compute the token.
+///
+/// The Partitioners' design is based on std::hash design: `Partitioner`
+/// corresponds to `HasherBuilder`, and `PartitionerHasher` to `Hasher`.
+/// See their documentation for more details.
 pub trait Partitioner {
-    fn hash(pk: &[u8]) -> Token;
+    type Hasher: PartitionerHasher;
+
+    fn build_hasher(&self) -> Self::Hasher;
+
+    fn hash_one(&self, data: &[u8]) -> Token {
+        let mut hasher = self.build_hasher();
+        hasher.write(data);
+        hasher.finish()
+    }
+}
+
+/// A trait for hashing a stream of serialized CQL values.
+///
+/// Instances of this trait are created by a `Partitioner` and are stateful.
+/// At any point, one can call `finish()` and a `Token` will be computed
+/// based on values that has been fed so far.
+pub trait PartitionerHasher {
+    fn write(&mut self, pk_part: &[u8]);
+    fn finish(&self) -> Token;
 }
 
 pub struct Murmur3Partitioner;
-pub struct CDCPartitioner;
 
-impl Murmur3Partitioner {
-    // An implementation of MurmurHash3 ported from Scylla. Please note that this
-    // is not a "correct" implementation of MurmurHash3 - it replicates the same
-    // bugs made in the original Cassandra implementation in order to be compatible.
-    fn hash3_x64_128(mut data: &[u8]) -> i128 {
-        let length = data.len();
+impl Partitioner for Murmur3Partitioner {
+    type Hasher = Murmur3PartitionerHasher;
+
+    fn build_hasher(&self) -> Self::Hasher {
+        Self::Hasher {
+            total_len: 0,
+            buf: Default::default(),
+            h1: Wrapping(0),
+            h2: Wrapping(0),
+        }
+    }
+}
+
+pub struct Murmur3PartitionerHasher {
+    total_len: usize,
+    buf: [u8; Self::BUF_CAPACITY],
+    h1: Wrapping<i64>,
+    h2: Wrapping<i64>,
+}
 
-        let c1: Wrapping<i64> = Wrapping(0x87c3_7b91_1142_53d5_u64 as i64);
-        let c2: Wrapping<i64> = Wrapping(0x4cf5_ad43_2745_937f_u64 as i64);
+impl Murmur3PartitionerHasher {
+    const BUF_CAPACITY: usize = 16;
 
-        let mut h1 = Wrapping(0_i64);
-        let mut h2 = Wrapping(0_i64);
+    const C1: Wrapping<i64> = Wrapping(0x87c3_7b91_1142_53d5_u64 as i64);
+    const C2: Wrapping<i64> = Wrapping(0x4cf5_ad43_2745_937f_u64 as i64);
 
-        while data.len() >= 16 {
-            let mut k1 = Wrapping(data.get_i64_le());
-            let mut k2 = Wrapping(data.get_i64_le());
+    fn hash_16_bytes(&mut self, mut k1: Wrapping<i64>, mut k2: Wrapping<i64>) {
+        k1 *= Self::C1;
+        k1 = Self::rotl64(k1, 31);
+        k1 *= Self::C2;
+        self.h1 ^= k1;
 
-            k1 *= c1;
-            k1 = Self::rotl64(k1, 31);
-            k1 *= c2;
-            h1 ^= k1;
+        self.h1 = Self::rotl64(self.h1, 27);
+        self.h1 += self.h2;
+        self.h1 = self.h1 * Wrapping(5) + Wrapping(0x52dce729);
 
-            h1 = Self::rotl64(h1, 27);
-            h1 += h2;
-            h1 = h1 * Wrapping(5) + Wrapping(0x52dce729);
+        k2 *= Self::C2;
+        k2 = Self::rotl64(k2, 33);
+        k2 *= Self::C1;
+        self.h2 ^= k2;
 
-            k2 *= c2;
-            k2 = Self::rotl64(k2, 33);
-            k2 *= c1;
-            h2 ^= k2;
+        self.h2 = Self::rotl64(self.h2, 31);
+        self.h2 += self.h1;
+        self.h2 = self.h2 * Wrapping(5) + Wrapping(0x38495ab5);
+    }
+
+    fn fetch_16_bytes_from_buf(buf: &mut &[u8]) -> (Wrapping<i64>, Wrapping<i64>) {
+        let k1 = Wrapping(buf.get_i64_le());
+        let k2 = Wrapping(buf.get_i64_le());
+        (k1, k2)
+    }
+
+    #[inline]
+    fn rotl64(v: Wrapping<i64>, n: u32) -> Wrapping<i64> {
+        Wrapping((v.0 << n) | (v.0 as u64 >> (64 - n)) as i64)
+    }
 
-            h2 = Self::rotl64(h2, 31);
-            h2 += h1;
-            h2 = h2 * Wrapping(5) + Wrapping(0x38495ab5);
+    #[inline]
+    fn fmix(mut k: Wrapping<i64>) -> Wrapping<i64> {
+        k ^= Wrapping((k.0 as u64 >> 33) as i64);
+        k *= Wrapping(0xff51afd7ed558ccd_u64 as i64);
+        k ^= Wrapping((k.0 as u64 >> 33) as i64);
+        k *= Wrapping(0xc4ceb9fe1a85ec53_u64 as i64);
+        k ^= Wrapping((k.0 as u64 >> 33) as i64);
+
+        k
+    }
+}
+
+// The implemented Murmur3 algorithm is roughly as follows:
+// 1. while there are at least 16 bytes given:
+//      consume 16 bytes by parsing them into i64s, then
+//      include them in h1, h2, k1, k2;
+// 2. do some magic with remaining n < 16 bytes,
+//      include them in h1, h2, k1, k2;
+// 3. compute the token based on h1, h2, k1, k2.
+//
+// Therefore, the buffer of capacity 16 is used. As soon as it gets full,
+// point 1. is executed. Points 2. and 3. are exclusively done in `finish()`,
+// so they don't mutate the state.
+impl PartitionerHasher for Murmur3PartitionerHasher {
+    fn write(&mut self, mut pk_part: &[u8]) {
+        let mut buf_len = self.total_len % Self::BUF_CAPACITY;
+        self.total_len += pk_part.len();
+
+        // If the buffer is nonempty and can be filled completely, so that we can fetch two i64s from it,
+        // fill it and hash its contents, then make it empty.
+        if buf_len > 0 && Self::BUF_CAPACITY - buf_len <= pk_part.len() {
+            // First phase: populate buffer until full, then consume two i64s.
+            let to_write = Ord::min(Self::BUF_CAPACITY - buf_len, pk_part.len());
+            self.buf[buf_len..buf_len + to_write].copy_from_slice(&pk_part[..to_write]);
+            pk_part.advance(to_write);
+            buf_len += to_write;
+
+            debug_assert_eq!(buf_len, Self::BUF_CAPACITY);
+            // consume 16 bytes from internal buf
+            let mut buf_ptr = &self.buf[..];
+            let (k1, k2) = Self::fetch_16_bytes_from_buf(&mut buf_ptr);
+            debug_assert!(buf_ptr.is_empty());
+            self.hash_16_bytes(k1, k2);
+            buf_len = 0;
         }
 
+        // If there were enough data, now we have an empty buffer. Further data, if enough, can be hence
+        // hashed directly from the external buffer.
+        if buf_len == 0 {
+            // Second phase: fast path for big values.
+            while pk_part.len() >= Self::BUF_CAPACITY {
+                let (k1, k2) = Self::fetch_16_bytes_from_buf(&mut pk_part);
+                self.hash_16_bytes(k1, k2);
+            }
+        }
+
+        // Third phase: move remaining bytes to the buffer.
+        debug_assert!(pk_part.len() < Self::BUF_CAPACITY - buf_len);
+        let to_write = pk_part.len();
+        self.buf[buf_len..buf_len + to_write].copy_from_slice(&pk_part[..to_write]);
+        pk_part.advance(to_write);
+        buf_len += to_write;
+        debug_assert!(pk_part.is_empty());
+
+        debug_assert!(buf_len < Self::BUF_CAPACITY);
+    }
+
+    fn finish(&self) -> Token {
+        let mut h1 = self.h1;
+        let mut h2 = self.h2;
+
         let mut k1 = Wrapping(0_i64);
         let mut k2 = Wrapping(0_i64);
 
-        debug_assert!(data.len() < 16);
+        let buf_len = self.total_len % Self::BUF_CAPACITY;
 
-        if data.len() > 8 {
-            for i in (8..data.len()).rev() {
-                k2 ^= Wrapping(data[i] as i8 as i64) << ((i - 8) * 8);
+        if buf_len > 8 {
+            for i in (8..buf_len).rev() {
+                k2 ^= Wrapping(self.buf[i] as i8 as i64) << ((i - 8) * 8);
             }
 
-            k2 *= c2;
+            k2 *= Self::C2;
             k2 = Self::rotl64(k2, 33);
-            k2 *= c1;
+            k2 *= Self::C1;
             h2 ^= k2;
         }
 
-        if !data.is_empty() {
-            for i in (0..std::cmp::min(8, data.len())).rev() {
-                k1 ^= Wrapping(data[i] as i8 as i64) << (i * 8);
+        if buf_len > 0 {
+            for i in (0..std::cmp::min(8, buf_len)).rev() {
+                k1 ^= Wrapping(self.buf[i] as i8 as i64) << (i * 8);
             }
 
-            k1 *= c1;
+            k1 *= Self::C1;
             k1 = Self::rotl64(k1, 31);
-            k1 *= c2;
+            k1 *= Self::C2;
             h1 ^= k1;
         }
 
-        h1 ^= Wrapping(length as i64);
-        h2 ^= Wrapping(length as i64);
+        h1 ^= Wrapping(self.total_len as i64);
+        h2 ^= Wrapping(self.total_len as i64);
 
         h1 += h2;
         h2 += h1;
@@ -112,47 +256,122 @@ impl Murmur3Partitioner {
         h1 += h2;
         h2 += h1;
 
-        ((h2.0 as i128) << 64) | h1.0 as i128
+        Token {
+            value: (((h2.0 as i128) << 64) | h1.0 as i128) as i64,
+        }
     }
+}
 
-    #[inline]
-    fn rotl64(v: Wrapping<i64>, n: u32) -> Wrapping<i64> {
-        Wrapping((v.0 << n) | (v.0 as u64 >> (64 - n)) as i64)
-    }
+enum CDCPartitionerHasherState {
+    Feeding {
+        len: usize,
+        buf: [u8; CDCPartitionerHasher::BUF_CAPACITY],
+    },
+    Computed(Token),
+}
 
-    #[inline]
-    fn fmix(mut k: Wrapping<i64>) -> Wrapping<i64> {
-        k ^= Wrapping((k.0 as u64 >> 33) as i64);
-        k *= Wrapping(0xff51afd7ed558ccd_u64 as i64);
-        k ^= Wrapping((k.0 as u64 >> 33) as i64);
-        k *= Wrapping(0xc4ceb9fe1a85ec53_u64 as i64);
-        k ^= Wrapping((k.0 as u64 >> 33) as i64);
+pub struct CDCPartitioner;
 
-        k
+pub struct CDCPartitionerHasher {
+    state: CDCPartitionerHasherState,
+}
+
+impl Partitioner for CDCPartitioner {
+    type Hasher = CDCPartitionerHasher;
+
+    fn build_hasher(&self) -> Self::Hasher {
+        Self::Hasher {
+            state: CDCPartitionerHasherState::Feeding {
+                len: 0,
+                buf: Default::default(),
+            },
+        }
     }
 }
 
-impl Partitioner for Murmur3Partitioner {
-    fn hash(pk: &[u8]) -> Token {
-        Token {
-            value: Self::hash3_x64_128(pk) as i64,
+impl CDCPartitionerHasher {
+    const BUF_CAPACITY: usize = 8;
+}
+
+impl PartitionerHasher for CDCPartitionerHasher {
+    fn write(&mut self, pk_part: &[u8]) {
+        match &mut self.state {
+            CDCPartitionerHasherState::Feeding { len, buf } => {
+                // We feed the buffer until it's full.
+                let copied_len = Ord::min(pk_part.len(), Self::BUF_CAPACITY - *len);
+                buf[*len..*len + copied_len].copy_from_slice(&pk_part[..copied_len]);
+                *len += copied_len;
+
+                // If the buffer is full, we can compute and fix the token.
+                if *len == Self::BUF_CAPACITY {
+                    let token = Token {
+                        value: (&mut &buf[..]).get_i64(),
+                    };
+                    self.state = CDCPartitionerHasherState::Computed(token);
+                }
+            }
+            CDCPartitionerHasherState::Computed(_) => (),
+        }
+    }
+
+    fn finish(&self) -> Token {
+        match self.state {
+            CDCPartitionerHasherState::Feeding { .. } => Token { value: i64::MIN },
+            CDCPartitionerHasherState::Computed(token) => token,
         }
     }
 }
 
-impl Partitioner for CDCPartitioner {
-    fn hash(mut pk: &[u8]) -> Token {
-        let value = if pk.len() < 8 { i64::MIN } else { pk.get_i64() };
-        Token { value }
+/// Calculates the token for given partitioner and serialized partition key.
+///
+/// The ordinary way to calculate token is based on a PreparedStatement
+/// and values for that statement. However, if a user knows:
+/// - the order of the columns in the partition key,
+/// - the values of the columns of the partition key,
+/// - the partitioner of the table that the statement operates on,
+///
+/// then having a `PreparedStatement` is not necessary and the token can
+/// be calculated based on that information.
+///
+/// NOTE: the provided values must completely constitute partition key
+/// and be in the order defined in CREATE TABLE statement.
+pub fn calculate_token_for_partition_key<P: Partitioner>(
+    serialized_partition_key_values: &SerializedValues,
+    partitioner: &P,
+) -> Result<Token, TokenCalculationError> {
+    let mut partitioner_hasher = partitioner.build_hasher();
+
+    if serialized_partition_key_values.len() == 1 {
+        let val = serialized_partition_key_values.iter().next().unwrap();
+        if let Some(val) = val {
+            partitioner_hasher.write(val);
+        }
+    } else {
+        for val in serialized_partition_key_values.iter().flatten() {
+            let val_len_u16: u16 = val
+                .len()
+                .try_into()
+                .map_err(|_| TokenCalculationError::ValueTooLong(val.len()))?;
+            partitioner_hasher.write(&val_len_u16.to_be_bytes());
+            partitioner_hasher.write(val);
+            partitioner_hasher.write(&[0u8]);
+        }
     }
+
+    Ok(partitioner_hasher.finish())
 }
 
 #[cfg(test)]
 mod tests {
+    use rand::Rng;
+    use rand_pcg::Pcg32;
+
+    use crate::transport::partitioner::PartitionerHasher;
+
     use super::{CDCPartitioner, Murmur3Partitioner, Partitioner};
 
     fn assert_correct_murmur3_hash(pk: &'static str, expected_hash: i64) {
-        let hash = Murmur3Partitioner::hash(pk.as_bytes()).value;
+        let hash = Murmur3Partitioner.hash_one(pk.as_bytes()).value;
         assert_eq!(hash, expected_hash);
     }
 
@@ -169,10 +388,68 @@ mod tests {
     }
 
     fn assert_correct_cdc_hash(pk: &'static str, expected_hash: i64) {
-        let hash = CDCPartitioner::hash(pk.as_bytes()).value;
+        let hash = CDCPartitioner.hash_one(pk.as_bytes()).value;
         assert_eq!(hash, expected_hash);
     }
 
+    #[test]
+    fn partitioners_output_same_result_no_matter_how_input_is_partitioned() {
+        let inputs: &[&[u8]] = &[
+            b"",
+            b"0",
+            "Ala ma kota, a kota ma Ala.".as_bytes(),
+            "Zażółć gęślą jaźń. Wsiadł rycerz Szaławiła na bułanego konia. Litwo, ojczyzno moja, ...".as_bytes(),
+        ];
+
+        let seed = 0x2137;
+        let mut randgen = Pcg32::new(seed, 0);
+
+        // Splits the given data 2^n times and feeds partitioner with the chunks got.
+        fn split_and_feed(
+            randgen: &mut impl Rng,
+            partitioner: &mut impl PartitionerHasher,
+            data: &[u8],
+            n: usize,
+        ) {
+            if n == 0 {
+                partitioner.write(data);
+            } else {
+                let pivot = if !data.is_empty() {
+                    randgen.gen_range(0..data.len())
+                } else {
+                    0
+                };
+                let (data1, data2) = data.split_at(pivot);
+                for data in [data1, data2] {
+                    split_and_feed(randgen, partitioner, data, n - 1);
+                }
+            }
+        }
+
+        fn check_for_partitioner<P: Partitioner>(
+            partitioner: P,
+            randgen: &mut impl Rng,
+            input: &[u8],
+        ) {
+            let result_single_batch = partitioner.hash_one(input);
+
+            let results_chunks = (0..1000).map(|_| {
+                let mut partitioner_hasher = partitioner.build_hasher();
+                split_and_feed(randgen, &mut partitioner_hasher, input, 2);
+                partitioner_hasher.finish()
+            });
+
+            for result_chunk in results_chunks {
+                assert_eq!(result_single_batch, result_chunk)
+            }
+        }
+
+        for input in inputs {
+            check_for_partitioner(Murmur3Partitioner, &mut randgen, input);
+            check_for_partitioner(CDCPartitioner, &mut randgen, input);
+        }
+    }
+
     #[test]
     fn test_cdc_partitioner() {
         for s in [
diff --git a/scylla/src/transport/retry_policy.rs b/scylla/src/transport/retry_policy.rs
index 1fb0279e..ee0c50aa 100644
--- a/scylla/src/transport/retry_policy.rs
+++ b/scylla/src/transport/retry_policy.rs
@@ -2,7 +2,7 @@
 //! To decide when to retry a query the `Session` can use any object which implements
 //! the `RetryPolicy` trait
 
-use crate::frame::types::{Consistency, LegacyConsistency};
+use crate::frame::types::Consistency;
 use crate::transport::errors::{DbError, QueryError, WriteType};
 
 /// Information about a failed query
@@ -14,7 +14,7 @@ pub struct QueryInfo<'a> {
     /// If set to `false` it is unknown whether it is idempotent
     pub is_idempotent: bool,
     /// Consistency with which the query failed
-    pub consistency: LegacyConsistency,
+    pub consistency: Consistency,
 }
 
 #[derive(Clone, Debug, PartialEq, Eq)]
@@ -136,7 +136,7 @@ impl Default for DefaultRetrySession {
 
 impl RetrySession for DefaultRetrySession {
     fn decide_should_retry(&mut self, query_info: QueryInfo) -> RetryDecision {
-        if let LegacyConsistency::Serial(_) = query_info.consistency {
+        if query_info.consistency.is_serial() {
             return RetryDecision::DontRetry;
         };
         match query_info.error {
@@ -219,7 +219,6 @@ impl RetrySession for DefaultRetrySession {
 #[cfg(test)]
 mod tests {
     use super::{DefaultRetryPolicy, QueryInfo, RetryDecision, RetryPolicy};
-    use crate::frame::types::LegacyConsistency;
     use crate::statement::Consistency;
     use crate::transport::errors::{BadQuery, DbError, QueryError, WriteType};
     use bytes::Bytes;
@@ -230,7 +229,7 @@ mod tests {
         QueryInfo {
             error,
             is_idempotent,
-            consistency: LegacyConsistency::Regular(Consistency::One),
+            consistency: Consistency::One,
         }
     }
 
@@ -267,14 +266,14 @@ mod tests {
             DbError::Unauthorized,
             DbError::ConfigError,
             DbError::ReadFailure {
-                consistency: LegacyConsistency::Regular(Consistency::Two),
+                consistency: Consistency::Two,
                 received: 2,
                 required: 1,
                 numfailures: 1,
                 data_present: false,
             },
             DbError::WriteFailure {
-                consistency: LegacyConsistency::Regular(Consistency::Two),
+                consistency: Consistency::Two,
                 received: 1,
                 required: 2,
                 numfailures: 1,
@@ -351,7 +350,7 @@ mod tests {
     fn default_unavailable() {
         let error = QueryError::DbError(
             DbError::Unavailable {
-                consistency: LegacyConsistency::Regular(Consistency::Two),
+                consistency: Consistency::Two,
                 required: 2,
                 alive: 1,
             },
@@ -385,7 +384,7 @@ mod tests {
         // Enough responses and data_present == false - coordinator received only checksums
         let enough_responses_no_data = QueryError::DbError(
             DbError::ReadTimeout {
-                consistency: LegacyConsistency::Regular(Consistency::Two),
+                consistency: Consistency::Two,
                 received: 2,
                 required: 2,
                 data_present: false,
@@ -419,7 +418,7 @@ mod tests {
         // waiting for read-repair acknowledgement.
         let enough_responses_with_data = QueryError::DbError(
             DbError::ReadTimeout {
-                consistency: LegacyConsistency::Regular(Consistency::Two),
+                consistency: Consistency::Two,
                 received: 2,
                 required: 2,
                 data_present: true,
@@ -444,7 +443,7 @@ mod tests {
         // Not enough responses, data_present == true
         let not_enough_responses_with_data = QueryError::DbError(
             DbError::ReadTimeout {
-                consistency: LegacyConsistency::Regular(Consistency::Two),
+                consistency: Consistency::Two,
                 received: 1,
                 required: 2,
                 data_present: true,
@@ -473,7 +472,7 @@ mod tests {
         // WriteType == BatchLog
         let good_write_type = QueryError::DbError(
             DbError::WriteTimeout {
-                consistency: LegacyConsistency::Regular(Consistency::Two),
+                consistency: Consistency::Two,
                 received: 1,
                 required: 2,
                 write_type: WriteType::BatchLog,
@@ -502,7 +501,7 @@ mod tests {
         // WriteType != BatchLog
         let bad_write_type = QueryError::DbError(
             DbError::WriteTimeout {
-                consistency: LegacyConsistency::Regular(Consistency::Two),
+                consistency: Consistency::Two,
                 received: 4,
                 required: 2,
                 write_type: WriteType::Simple,
diff --git a/scylla/src/transport/session.rs b/scylla/src/transport/session.rs
index 9c85d23a..28fa0593 100644
--- a/scylla/src/transport/session.rs
+++ b/scylla/src/transport/session.rs
@@ -4,21 +4,18 @@
 #[cfg(feature = "cloud")]
 use crate::cloud::CloudConfig;
 
-use crate::frame::types::LegacyConsistency;
 use crate::history;
 use crate::history::HistoryListener;
-use crate::prepared_statement::PartitionKeyDecoder;
 use crate::utils::pretty::{CommaSeparatedDisplayer, CqlValueDisplayer};
+use crate::utils::unzip_option;
 use arc_swap::ArcSwapOption;
 use async_trait::async_trait;
-use bytes::BufMut;
 use bytes::Bytes;
-use bytes::BytesMut;
 use futures::future::join_all;
 use futures::future::try_join_all;
-use itertools::Either;
+use itertools::{Either, Itertools};
 pub use scylla_cql::errors::TranslationError;
-use scylla_cql::frame::response::result::{PreparedMetadata, Rows};
+use scylla_cql::frame::response::result::{deser_cql_value, ColumnSpec, Rows};
 use scylla_cql::frame::response::NonErrorResponse;
 use std::borrow::Borrow;
 use std::collections::HashMap;
@@ -35,7 +32,7 @@ use std::time::Duration;
 use tokio::net::lookup_host;
 use tokio::time::timeout;
 use tracing::warn;
-use tracing::{debug, error, trace, trace_span, Instrument};
+use tracing::{debug, trace, trace_span, Instrument};
 use uuid::Uuid;
 
 use super::cluster::ContactPoint;
@@ -43,9 +40,8 @@ use super::connection::NonErrorQueryResponse;
 use super::connection::QueryResponse;
 #[cfg(feature = "ssl")]
 use super::connection::SslConfig;
-use super::errors::{BadQuery, NewSessionError, QueryError};
+use super::errors::{NewSessionError, QueryError};
 use super::execution_profile::{ExecutionProfile, ExecutionProfileHandle, ExecutionProfileInner};
-use super::partitioner::Partitioner;
 use super::partitioner::PartitionerName;
 use super::topology::UntranslatedPeer;
 use super::NodeRef;
@@ -53,12 +49,12 @@ use crate::cql_to_rust::FromRow;
 use crate::frame::response::cql_to_rust::FromRowError;
 use crate::frame::response::result;
 use crate::frame::value::{
-    BatchValues, BatchValuesFirstSerialized, BatchValuesIterator, SerializedValues, ValueList,
+    BatchValues, BatchValuesFirstSerialized, BatchValuesIterator, ValueList,
 };
-use crate::prepared_statement::{PartitionKeyError, PreparedStatement};
+use crate::prepared_statement::PreparedStatement;
 use crate::query::Query;
 use crate::routing::Token;
-use crate::statement::{Consistency, SerialConsistency};
+use crate::statement::Consistency;
 use crate::tracing::{TracingEvent, TracingInfo};
 use crate::transport::cluster::{Cluster, ClusterData, ClusterNeatDebug};
 use crate::transport::connection::{Connection, ConnectionConfig, VerifiedKeyspaceName};
@@ -83,6 +79,22 @@ use crate::authentication::AuthenticatorProvider;
 #[cfg(feature = "ssl")]
 use openssl::ssl::SslContext;
 
+/// Translates IP addresses received from ScyllaDB nodes into locally reachable addresses.
+///
+/// The driver auto-detects new ScyllaDB nodes added to the cluster through server side pushed
+/// notifications and through checking the system tables. For each node, the address the driver
+/// receives corresponds to the address set as `rpc_address` in the node yaml file. In most
+/// cases, this is the correct address to use by the driver and that is what is used by default.
+/// However, sometimes the addresses received through this mechanism will either not be reachable
+/// directly by the driver or should not be the preferred address to use to reach the node (for
+/// instance, the `rpc_address` set on ScyllaDB nodes might be a private IP, but some clients
+/// may have to use a public IP, or pass by a router, e.g. through NAT, to reach that node).
+/// This interface allows to deal with such cases, by allowing to translate an address as sent
+/// by a ScyllaDB node to another address to be used by the driver for connection.
+///
+/// Please note that the "known nodes" addresses provided while creating the [`Session`]
+/// instance are not translated, only IP address retrieved from or sent by Cassandra nodes
+/// to the driver are.
 #[async_trait]
 pub trait AddressTranslator: Send + Sync {
     async fn translate_address(
@@ -130,7 +142,8 @@ pub struct Session {
     default_execution_profile_handle: ExecutionProfileHandle,
     schema_agreement_interval: Duration,
     metrics: Arc<Metrics>,
-    auto_await_schema_agreement_timeout: Option<Duration>,
+    schema_agreement_timeout: Duration,
+    schema_agreement_automatic_waiting: bool,
     refresh_metadata_on_auto_schema_agreement: bool,
     keyspace_name: ArcSwapOption<String>,
     tracing_info_fetch_attempts: NonZeroU32,
@@ -152,7 +165,7 @@ impl std::fmt::Debug for Session {
             .field("metrics", &self.metrics)
             .field(
                 "auto_await_schema_agreement_timeout",
-                &self.auto_await_schema_agreement_timeout,
+                &self.schema_agreement_timeout,
             )
             .finish()
     }
@@ -186,7 +199,6 @@ pub struct SessionConfig {
 
     pub authenticator: Option<Arc<dyn AuthenticatorProvider>>,
 
-    pub schema_agreement_interval: Duration,
     pub connect_timeout: Duration,
 
     /// Size of the per-node connection pool, i.e. how many connections the driver should keep to each node.
@@ -211,10 +223,26 @@ pub struct SessionConfig {
     /// If `None`, connections are never closed due to lack of response to a keepalive message.
     pub keepalive_timeout: Option<Duration>,
 
-    /// Controls the timeout for the automatic wait for schema agreement after sending a schema-altering statement.
-    /// If `None`, the automatic schema agreement is disabled.
-    pub auto_await_schema_agreement_timeout: Option<Duration>,
+    /// How often the driver should ask if schema is in agreement.
+    pub schema_agreement_interval: Duration,
+
+    /// Controls the timeout for waiting for schema agreement.
+    /// This works both for manual awaiting schema agreement and for
+    /// automatic waiting after a schema-altering statement is sent.
+    pub schema_agreement_timeout: Duration,
 
+    /// Controls whether schema agreement is automatically awaited
+    /// after sending a schema-altering statement.
+    pub schema_agreement_automatic_waiting: bool,
+
+    /// If true, full schema metadata is fetched after successfully reaching a schema agreement.
+    /// It is true by default but can be disabled if successive schema-altering statements should be performed.
+    pub refresh_metadata_on_auto_schema_agreement: bool,
+
+    /// The address translator is used to translate addresses received from ScyllaDB nodes
+    /// (either with cluster metadata or with an event) to addresses that can be used to
+    /// actually connect to those nodes. This may be needed e.g. when there is NAT
+    /// between the nodes and the driver.
     pub address_translator: Option<Arc<dyn AddressTranslator>>,
 
     /// The host filter decides whether any connections should be opened
@@ -222,10 +250,6 @@ pub struct SessionConfig {
     /// re-establishing the control connection.
     pub host_filter: Option<Arc<dyn HostFilter>>,
 
-    /// If true, full schema metadata is fetched after successfully reaching a schema agreement.
-    /// It is true by default but can be disabled if successive schema-altering statements should be performed.
-    pub refresh_metadata_on_auto_schema_agreement: bool,
-
     /// If the driver is to connect to ScyllaCloud, there is a config for it.
     #[cfg(feature = "cloud")]
     pub cloud_config: Option<Arc<CloudConfig>>,
@@ -257,6 +281,12 @@ pub struct SessionConfig {
     /// Consistency level of fetching [`TracingInfo`]
     /// in [`Session::get_tracing_info`].
     pub tracing_info_fetch_consistency: Consistency,
+
+    /// Interval between refreshing cluster metadata. This
+    /// can be configured according to the traffic pattern
+    /// for e.g: if they do not want unexpected traffic
+    /// or they expect the topology to change frequently.
+    pub cluster_metadata_refresh_interval: Duration,
 }
 
 /// Describes database server known on Session startup.
@@ -309,7 +339,8 @@ impl SessionConfig {
             fetch_schema_metadata: true,
             keepalive_interval: Some(Duration::from_secs(30)),
             keepalive_timeout: Some(Duration::from_secs(30)),
-            auto_await_schema_agreement_timeout: Some(std::time::Duration::from_secs(60)),
+            schema_agreement_timeout: Duration::from_secs(60),
+            schema_agreement_automatic_waiting: true,
             address_translator: None,
             host_filter: None,
             refresh_metadata_on_auto_schema_agreement: true,
@@ -319,6 +350,7 @@ impl SessionConfig {
             tracing_info_fetch_attempts: NonZeroU32::new(5).unwrap(),
             tracing_info_fetch_interval: Duration::from_millis(3),
             tracing_info_fetch_consistency: Consistency::One,
+            cluster_metadata_refresh_interval: Duration::from_secs(60),
         }
     }
 
@@ -423,7 +455,7 @@ impl<RowT: FromRow> Iterator for TypedRowIter<RowT> {
     }
 }
 
-pub enum RunQueryResult<ResT> {
+pub(crate) enum RunQueryResult<ResT> {
     IgnoredWriteError,
     Completed(ResT),
 }
@@ -555,6 +587,7 @@ impl Session {
             config.keyspaces_to_fetch,
             config.fetch_schema_metadata,
             config.host_filter,
+            config.cluster_metadata_refresh_interval,
         )
         .await?;
 
@@ -565,7 +598,8 @@ impl Session {
             default_execution_profile_handle,
             schema_agreement_interval: config.schema_agreement_interval,
             metrics: Arc::new(Metrics::new()),
-            auto_await_schema_agreement_timeout: config.auto_await_schema_agreement_timeout,
+            schema_agreement_timeout: config.schema_agreement_timeout,
+            schema_agreement_automatic_waiting: config.schema_agreement_automatic_waiting,
             refresh_metadata_on_auto_schema_agreement: config
                 .refresh_metadata_on_auto_schema_agreement,
             keyspace_name: ArcSwapOption::default(), // will be set by use_keyspace
@@ -715,8 +749,7 @@ impl Session {
         };
 
         self.handle_set_keyspace_response(&response).await?;
-        self.handle_auto_await_schema_agreement(&query.contents, &response)
-            .await?;
+        self.handle_auto_await_schema_agreement(&response).await?;
 
         let result = response.into_query_result()?;
         span.record_result_fields(&result);
@@ -741,20 +774,11 @@ impl Session {
 
     async fn handle_auto_await_schema_agreement(
         &self,
-        contents: &str,
         response: &NonErrorQueryResponse,
     ) -> Result<(), QueryError> {
-        if let Some(timeout) = self.auto_await_schema_agreement_timeout {
-            if response.as_schema_change().is_some()
-                && !self.await_timed_schema_agreement(timeout).await?
-            {
-                // TODO: The TimeoutError should allow to provide more context.
-                // For now, print an error to the logs
-                error!(
-                    "Failed to reach schema agreement after a schema-altering statement: {}",
-                    contents,
-                );
-                return Err(QueryError::TimeoutError);
+        if self.schema_agreement_automatic_waiting {
+            if response.as_schema_change().is_some() {
+                self.await_schema_agreement().await?;
             }
 
             if self.refresh_metadata_on_auto_schema_agreement
@@ -860,32 +884,26 @@ impl Session {
     /// ```
     pub async fn prepare(&self, query: impl Into<Query>) -> Result<PreparedStatement, QueryError> {
         let query = query.into();
+        let query_ref = &query;
 
-        let connections = self.cluster.get_working_connections().await?;
+        let cluster_data = self.get_cluster_data();
+        let connections_iter = cluster_data.iter_working_connections()?;
 
         // Prepare statements on all connections concurrently
-        let handles = connections.iter().map(|c| c.prepare(&query));
-        let mut results = join_all(handles).await;
+        let handles = connections_iter.map(|c| async move { c.prepare(query_ref).await });
+        let mut results = join_all(handles).await.into_iter();
 
-        // If at least one prepare was successful prepare returns Ok
+        // If at least one prepare was successful, `prepare()` returns Ok.
+        // Find the first result that is Ok, or Err if all failed.
 
-        // 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;
-            }
-        }
-
-        let mut prepared: PreparedStatement = first_ok.unwrap()?;
+        // Safety: there is at least one node in the cluster, and `Cluster::iter_working_connections()`
+        // returns either an error or an iterator with at least one connection, so there will be at least one result.
+        let first_ok: Result<PreparedStatement, QueryError> =
+            results.by_ref().find_or_first(Result::is_ok).unwrap();
+        let mut prepared: PreparedStatement = first_ok?;
 
         // Validate prepared ids equality
-        for statement in results.into_iter().flatten() {
+        for statement in results.flatten() {
             if prepared.get_id() != statement.get_id() {
                 return Err(QueryError::ProtocolError(
                     "Prepared statement Ids differ, all should be equal",
@@ -982,10 +1000,11 @@ impl Session {
         let values_ref = &serialized_values;
         let paging_state_ref = &paging_state;
 
-        let partition_key = self.calculate_partition_key(prepared, &serialized_values)?;
-        let token = partition_key
-            .as_ref()
-            .map(|pk| prepared.get_partitioner_name().hash(pk));
+        let (partition_key, token) =
+            unzip_option(prepared.extract_partition_key_and_calculate_token(
+                prepared.get_partitioner_name(),
+                &serialized_values,
+            )?);
 
         let execution_profile = prepared
             .get_execution_profile_handle()
@@ -1007,8 +1026,7 @@ impl Session {
         };
 
         let span = RequestSpan::new_prepared(
-            prepared.get_prepared_metadata(),
-            partition_key.as_ref(),
+            partition_key.as_ref().map(|pk| pk.iter()),
             token,
             serialized_values.size(),
         );
@@ -1069,8 +1087,7 @@ impl Session {
         };
 
         self.handle_set_keyspace_response(&response).await?;
-        self.handle_auto_await_schema_agreement(prepared.get_statement(), &response)
-            .await?;
+        self.handle_auto_await_schema_agreement(&response).await?;
 
         let result = response.into_query_result()?;
         span.record_result_fields(&result);
@@ -1125,10 +1142,6 @@ impl Session {
     ) -> Result<RowIterator, QueryError> {
         let prepared = prepared.into();
         let serialized_values = values.serialized()?;
-        let partition_key = self.calculate_partition_key(&prepared, &serialized_values)?;
-        let token = partition_key
-            .as_ref()
-            .map(|pk| prepared.get_partitioner_name().hash(pk));
 
         let execution_profile = prepared
             .get_execution_profile_handle()
@@ -1138,8 +1151,6 @@ impl Session {
         RowIterator::new_for_prepared_statement(PreparedIteratorConfig {
             prepared,
             values: serialized_values.into_owned(),
-            partition_key,
-            token,
             execution_profile,
             cluster_data: self.cluster.get_data(),
             metrics: self.metrics.clone(),
@@ -1219,7 +1230,7 @@ impl Session {
                 RoutingInfo {
                     consistency,
                     serial_consistency,
-                    token: self.calculate_token(ps, first_serialized_value)?,
+                    token: ps.calculate_token(first_serialized_value)?,
                     keyspace: ps.get_keyspace_name(),
                     is_confirmed_lwt: false,
                 }
@@ -1782,11 +1793,9 @@ impl Session {
                 let query_info = QueryInfo {
                     error: the_error,
                     is_idempotent: context.is_idempotent,
-                    consistency: LegacyConsistency::Regular(
-                        context
-                            .consistency_set_on_statement
-                            .unwrap_or(execution_profile.consistency),
-                    ),
+                    consistency: context
+                        .consistency_set_on_statement
+                        .unwrap_or(execution_profile.consistency),
                 };
 
                 let retry_decision = context.retry_session.decide_should_retry(query_info);
@@ -1818,146 +1827,36 @@ impl Session {
         last_error.map(Result::Err)
     }
 
-    pub async fn await_schema_agreement(&self) -> Result<(), QueryError> {
-        while !self.check_schema_agreement().await? {
-            tokio::time::sleep(self.schema_agreement_interval).await
+    async fn await_schema_agreement_indefinitely(&self) -> Result<Uuid, QueryError> {
+        loop {
+            tokio::time::sleep(self.schema_agreement_interval).await;
+            if let Some(agreed_version) = self.check_schema_agreement().await? {
+                return Ok(agreed_version);
+            }
         }
-        Ok(())
     }
 
-    pub async fn await_timed_schema_agreement(
-        &self,
-        timeout_duration: Duration,
-    ) -> Result<bool, QueryError> {
-        timeout(timeout_duration, self.await_schema_agreement())
-            .await
-            .map_or(Ok(false), |res| res.and(Ok(true)))
-    }
-
-    async fn schema_agreement_auxiliary<ResT, QueryFut>(
-        &self,
-        do_query: impl Fn(Arc<Connection>, Consistency, &ExecutionProfileInner) -> QueryFut,
-    ) -> Result<ResT, QueryError>
-    where
-        QueryFut: Future<Output = Result<ResT, QueryError>>,
-        ResT: AllowedRunQueryResTType,
-    {
-        let info = RoutingInfo::default();
-        let config = StatementConfig {
-            is_idempotent: true,
-            serial_consistency: Some(Some(SerialConsistency::LocalSerial)),
-            ..Default::default()
-        };
-
-        let span = RequestSpan::new_none();
-
-        match self
-            .run_query(
-                info,
-                &config,
-                self.get_default_execution_profile_handle().access(),
-                |node: Arc<Node>| async move { node.random_connection().await },
-                do_query,
-                &span,
-            )
-            .await?
-        {
-            RunQueryResult::IgnoredWriteError => Err(QueryError::ProtocolError(
-                "Retry policy has made the driver ignore schema's agreement query.",
-            )),
-            RunQueryResult::Completed(result) => Ok(result),
-        }
+    pub async fn await_schema_agreement(&self) -> Result<Uuid, QueryError> {
+        timeout(
+            self.schema_agreement_timeout,
+            self.await_schema_agreement_indefinitely(),
+        )
+        .await
+        .unwrap_or(Err(QueryError::RequestTimeout(
+            "schema agreement not reached in time".to_owned(),
+        )))
     }
 
-    pub async fn check_schema_agreement(&self) -> Result<bool, QueryError> {
-        let connections = self.cluster.get_working_connections().await?;
+    pub async fn check_schema_agreement(&self) -> Result<Option<Uuid>, QueryError> {
+        let cluster_data = self.get_cluster_data();
+        let connections_iter = cluster_data.iter_working_connections()?;
 
-        let handles = connections.iter().map(|c| c.fetch_schema_version());
+        let handles = connections_iter.map(|c| async move { c.fetch_schema_version().await });
         let versions = try_join_all(handles).await?;
 
         let local_version: Uuid = versions[0];
         let in_agreement = versions.into_iter().all(|v| v == local_version);
-        Ok(in_agreement)
-    }
-
-    pub async fn fetch_schema_version(&self) -> Result<Uuid, QueryError> {
-        // We ignore custom Consistency that a retry policy could decide to put here, using the default instead.
-        self.schema_agreement_auxiliary(
-            |connection: Arc<Connection>, _: Consistency, _: &ExecutionProfileInner| async move {
-                connection.fetch_schema_version().await
-            },
-        )
-        .await
-    }
-
-    fn calculate_partition_key(
-        &self,
-        prepared: &PreparedStatement,
-        serialized_values: &SerializedValues,
-    ) -> Result<Option<Bytes>, QueryError> {
-        if !prepared.is_token_aware() {
-            return Ok(None);
-        }
-        let partition_key = calculate_partition_key(prepared, serialized_values)?;
-        Ok(Some(partition_key))
-    }
-
-    /// Calculates the token for given prepared statement and serialized values.
-    ///
-    /// Returns the token that would be computed for executing the provided
-    /// prepared statement with the provided values.
-    pub fn calculate_token(
-        &self,
-        prepared: &PreparedStatement,
-        serialized_values: &SerializedValues,
-    ) -> Result<Option<Token>, QueryError> {
-        match self.calculate_partition_key(prepared, serialized_values) {
-            Ok(Some(partition_key)) => {
-                let partitioner_name = prepared.get_partitioner_name();
-                Ok(Some(partitioner_name.hash(&partition_key)))
-            }
-            Ok(None) => Ok(None),
-            Err(err) => Err(err),
-        }
-    }
-
-    /// Calculates the token for given partitioner and serialized partition key.
-    ///
-    /// The ordinary way to calculate token is based on a PreparedStatement
-    /// and values for that statement. However, if a user knows:
-    /// - the order of the columns in the partition key,
-    /// - the values of the columns of the partition key,
-    /// - the partitioner of the table that the statement operates on,
-    ///
-    /// then having a `PreparedStatement` is not necessary and the token can
-    /// be calculated based on that information.
-    ///
-    /// NOTE: the provided values must completely constitute partition key
-    /// and be in the order defined in CREATE TABLE statement.
-    pub fn calculate_token_for_partition_key<P: Partitioner>(
-        serialized_partition_key_values: &SerializedValues,
-        _partitioner: &P,
-    ) -> Result<Token, PartitionKeyError> {
-        let mut buf: BytesMut = BytesMut::new();
-
-        if serialized_partition_key_values.len() == 1 {
-            let val = serialized_partition_key_values.iter().next().unwrap();
-            if let Some(val) = val {
-                buf.extend_from_slice(val);
-            }
-        } else {
-            for val in serialized_partition_key_values.iter().flatten() {
-                let val_len_u16: u16 = val
-                    .len()
-                    .try_into()
-                    .map_err(|_| PartitionKeyError::ValueTooLong(val.len()))?;
-                buf.put_u16(val_len_u16);
-                buf.extend_from_slice(val);
-                buf.put_u8(0);
-            }
-        }
-
-        Ok(P::hash(&buf.freeze()))
+        Ok(in_agreement.then_some(local_version))
     }
 
     /// Retrieves the handle to execution profile that is used by this session
@@ -1967,21 +1866,6 @@ impl Session {
     }
 }
 
-fn calculate_partition_key(
-    stmt: &PreparedStatement,
-    values: &SerializedValues,
-) -> Result<Bytes, QueryError> {
-    match stmt.compute_partition_key(values) {
-        Ok(key) => Ok(key),
-        Err(PartitionKeyError::NoPkIndexValue(_, _)) => Err(QueryError::ProtocolError(
-            "No pk indexes - can't calculate token",
-        )),
-        Err(PartitionKeyError::ValueTooLong(values_len)) => Err(QueryError::BadQuery(
-            BadQuery::ValuesTooLongForKey(values_len, u16::MAX.into()),
-        )),
-    }
-}
-
 // Resolve the given hostname 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.
@@ -2018,7 +1902,7 @@ pub(crate) async fn resolve_hostname(hostname: &str) -> Result<SocketAddr, io::E
 // When using run_query make sure that the ResT type is NOT able
 // to contain any errors.
 // See https://github.com/scylladb/scylla-rust-driver/issues/501
-pub trait AllowedRunQueryResTType {}
+pub(crate) trait AllowedRunQueryResTType {}
 
 impl AllowedRunQueryResTType for Uuid {}
 impl AllowedRunQueryResTType for QueryResult {}
@@ -2111,9 +1995,8 @@ impl RequestSpan {
         }
     }
 
-    pub(crate) fn new_prepared(
-        prepared_metadata: &PreparedMetadata,
-        partition_key: Option<&Bytes>,
+    pub(crate) fn new_prepared<'ps>(
+        partition_key: Option<impl Iterator<Item = (&'ps [u8], &'ps ColumnSpec)> + Clone>,
         token: Option<Token>,
         request_size: usize,
     ) -> Self {
@@ -2136,10 +2019,9 @@ impl RequestSpan {
         if let Some(partition_key) = partition_key {
             span.record(
                 "partition_key",
-                tracing::field::display(format_args!(
-                    "{}",
-                    partition_key_displayer(prepared_metadata, partition_key),
-                )),
+                tracing::field::display(
+                    format_args!("{}", partition_key_displayer(partition_key),),
+                ),
             );
         }
         if let Some(token) = token {
@@ -2173,13 +2055,6 @@ impl RequestSpan {
         }
     }
 
-    pub(crate) fn new_none() -> Self {
-        Self {
-            span: tracing::Span::none(),
-            speculative_executions: 0.into(),
-        }
-    }
-
     pub(crate) fn record_shard_id(&self, conn: &Connection) {
         if let Some(info) = conn.get_shard_info() {
             self.span.record("shard", info.shard);
@@ -2238,12 +2113,16 @@ impl Drop for RequestSpan {
     }
 }
 
-fn partition_key_displayer<'pk>(
-    prepared_metadata: &'pk PreparedMetadata,
-    partition_key: &'pk [u8],
-) -> impl Display + 'pk {
+fn partition_key_displayer<'ps, 'res>(
+    mut pk_values_iter: impl Iterator<Item = (&'ps [u8], &'ps ColumnSpec)> + 'res + Clone,
+) -> impl Display + 'res {
     CommaSeparatedDisplayer(
-        PartitionKeyDecoder::new(prepared_metadata, partition_key).map(|c| match c {
+        std::iter::from_fn(move || {
+            pk_values_iter
+                .next()
+                .map(|(mut cell, spec)| deser_cql_value(&spec.typ, &mut cell))
+        })
+        .map(|c| match c {
             Ok(c) => Either::Left(CqlValueDisplayer(c)),
             Err(_) => Either::Right("<decoding error>"),
         }),
diff --git a/scylla/src/transport/session_builder.rs b/scylla/src/transport/session_builder.rs
index 8804e10e..7460fdb1 100644
--- a/scylla/src/transport/session_builder.rs
+++ b/scylla/src/transport/session_builder.rs
@@ -717,8 +717,8 @@ impl<K: SessionBuilderKind> GenericSessionBuilder<K> {
         self
     }
 
-    /// Enables automatic wait for schema agreement and sets the timeout for it.
-    /// By default, it is enabled and the timeout is 60 seconds.
+    /// Sets the timeout for waiting for schema agreement.
+    /// By default, the timeout is 60 seconds.
     ///
     /// # Example
     /// ```
@@ -726,19 +726,19 @@ impl<K: SessionBuilderKind> GenericSessionBuilder<K> {
     /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
     /// let session: Session = SessionBuilder::new()
     ///     .known_node("127.0.0.1:9042")
-    ///     .auto_schema_agreement_timeout(std::time::Duration::from_secs(120))
+    ///     .schema_agreement_timeout(std::time::Duration::from_secs(120))
     ///     .build()
     ///     .await?;
     /// # Ok(())
     /// # }
     /// ```
-    pub fn auto_schema_agreement_timeout(mut self, timeout: Duration) -> Self {
-        self.config.auto_await_schema_agreement_timeout = Some(timeout);
+    pub fn schema_agreement_timeout(mut self, timeout: Duration) -> Self {
+        self.config.schema_agreement_timeout = timeout;
         self
     }
 
-    /// Disables automatic wait for schema agreement.
-    /// By default, it is enabled and the timeout is 60 seconds.
+    /// Controls automatic waiting for schema agreement after a schema-altering
+    /// statement is sent. By default, it is enabled.
     ///
     /// # Example
     /// ```
@@ -746,14 +746,14 @@ impl<K: SessionBuilderKind> GenericSessionBuilder<K> {
     /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
     /// let session: Session = SessionBuilder::new()
     ///     .known_node("127.0.0.1:9042")
-    ///     .no_auto_schema_agreement()
+    ///     .auto_await_schema_agreement(false)
     ///     .build()
     ///     .await?;
     /// # Ok(())
     /// # }
     /// ```
-    pub fn no_auto_schema_agreement(mut self) -> Self {
-        self.config.auto_await_schema_agreement_timeout = None;
+    pub fn auto_await_schema_agreement(mut self, enabled: bool) -> Self {
+        self.config.schema_agreement_automatic_waiting = enabled;
         self
     }
 
@@ -910,6 +910,30 @@ impl<K: SessionBuilderKind> GenericSessionBuilder<K> {
         self.config.enable_write_coalescing = enable;
         self
     }
+
+    /// Set the interval at which the driver refreshes the cluster metadata which contains information
+    /// about the cluster topology as well as the cluster schema.
+    ///
+    /// The default is 60 seconds.
+    ///
+    /// In the given example, we have set the duration value to 20 seconds, which
+    /// means that the metadata is refreshed every 20 seconds.
+    /// # 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")
+    ///         .cluster_metadata_refresh_interval(std::time::Duration::from_secs(20))
+    ///         .build()
+    ///         .await?;
+    /// #   Ok(())
+    /// # }
+    /// ```
+    pub fn cluster_metadata_refresh_interval(mut self, interval: Duration) -> Self {
+        self.config.cluster_metadata_refresh_interval = interval;
+        self
+    }
 }
 
 /// Creates a [`SessionBuilder`] with default configuration, same as [`SessionBuilder::new`]
@@ -1124,6 +1148,15 @@ mod tests {
         );
     }
 
+    #[test]
+    fn cluster_metadata_refresh_interval() {
+        let builder = SessionBuilder::new();
+        assert_eq!(
+            builder.config.cluster_metadata_refresh_interval,
+            std::time::Duration::from_secs(60)
+        );
+    }
+
     #[test]
     fn all_features() {
         let mut builder = SessionBuilder::new();
@@ -1140,6 +1173,7 @@ mod tests {
         builder = builder.tcp_nodelay(true);
         builder = builder.use_keyspace("ks_name", true);
         builder = builder.fetch_schema_metadata(false);
+        builder = builder.cluster_metadata_refresh_interval(Duration::from_secs(1));
 
         assert_eq!(
             builder.config.known_nodes,
@@ -1155,6 +1189,10 @@ mod tests {
 
         assert_eq!(builder.config.compression, Some(Compression::Snappy));
         assert!(builder.config.tcp_nodelay);
+        assert_eq!(
+            builder.config.cluster_metadata_refresh_interval,
+            Duration::from_secs(1)
+        );
 
         assert_eq!(builder.config.used_keyspace, Some("ks_name".to_string()));
 
diff --git a/scylla/src/transport/speculative_execution.rs b/scylla/src/transport/speculative_execution.rs
index 17ed8064..516ba8cd 100644
--- a/scylla/src/transport/speculative_execution.rs
+++ b/scylla/src/transport/speculative_execution.rs
@@ -91,7 +91,7 @@ fn can_be_ignored<ResT>(result: &Result<ResT, QueryError>) -> bool {
 
 const EMPTY_PLAN_ERROR: QueryError = QueryError::ProtocolError("Empty query plan - driver bug!");
 
-pub async fn execute<QueryFut, ResT>(
+pub(crate) async fn execute<QueryFut, ResT>(
     policy: &dyn SpeculativeExecutionPolicy,
     context: &Context,
     query_runner_generator: impl Fn(bool) -> QueryFut,
diff --git a/scylla/src/transport/topology.rs b/scylla/src/transport/topology.rs
index 3379420c..ffad6283 100644
--- a/scylla/src/transport/topology.rs
+++ b/scylla/src/transport/topology.rs
@@ -49,9 +49,9 @@ pub(crate) struct MetadataReader {
 }
 
 /// Describes all metadata retrieved from the cluster
-pub struct Metadata {
-    pub peers: Vec<Peer>,
-    pub keyspaces: HashMap<String, Keyspace>,
+pub(crate) struct Metadata {
+    pub(crate) peers: Vec<Peer>,
+    pub(crate) keyspaces: HashMap<String, Keyspace>,
 }
 
 #[non_exhaustive] // <- so that we can add more fields in a backwards-compatible way
@@ -113,6 +113,18 @@ impl Peer {
             rack: self.rack.clone(),
         }
     }
+
+    pub(crate) fn into_peer_endpoint_and_tokens(self) -> (PeerEndpoint, Vec<Token>) {
+        (
+            PeerEndpoint {
+                host_id: self.host_id,
+                address: self.address,
+                datacenter: self.datacenter,
+                rack: self.rack,
+            },
+            self.tokens,
+        )
+    }
 }
 
 /// Data used to issue connections to a node that is possibly subject to address translation.
@@ -349,7 +361,7 @@ impl Metadata {
     ///
     /// It can be used as a replacement for real metadata when initial
     /// metadata read fails.
-    pub fn new_dummy(initial_peers: &[UntranslatedEndpoint]) -> Self {
+    pub(crate) fn new_dummy(initial_peers: &[UntranslatedEndpoint]) -> Self {
         let peers = initial_peers
             .iter()
             .enumerate()
diff --git a/scylla/src/utils/mod.rs b/scylla/src/utils/mod.rs
index c83adbf7..9399e332 100644
--- a/scylla/src/utils/mod.rs
+++ b/scylla/src/utils/mod.rs
@@ -2,3 +2,11 @@ pub(crate) mod parse;
 
 pub(crate) mod pretty;
 pub mod test_utils;
+
+// FIXME: replace this with `Option::unzip()` once MSRV is bumped to at least 1.66
+pub(crate) fn unzip_option<T, U>(opt: Option<(T, U)>) -> (Option<T>, Option<U>) {
+    match opt {
+        Some((a, b)) => (Some(a), Some(b)),
+        None => (None, None),
+    }
+}


Test patch:
diff --git a/scylla-cql/src/frame/value_tests.rs b/scylla-cql/src/frame/value_tests.rs
index 95641464..f7e48036 100644
--- a/scylla-cql/src/frame/value_tests.rs
+++ b/scylla-cql/src/frame/value_tests.rs
@@ -35,6 +35,12 @@ fn u8_array_serialization() {
     assert_eq!(serialized(val), vec![0, 0, 0, 4, 1, 1, 1, 1]);
 }
 
+#[test]
+fn u8_slice_serialization() {
+    let val = vec![1u8, 1, 1, 1];
+    assert_eq!(serialized(val.as_slice()), vec![0, 0, 0, 4, 1, 1, 1, 1]);
+}
+
 #[test]
 fn naive_date_serialization() {
     // 1970-01-31 is 2^31
diff --git a/scylla/src/transport/session_test.rs b/scylla/src/transport/session_test.rs
index 35e2f7bf..6bc9f0e4 100644
--- a/scylla/src/transport/session_test.rs
+++ b/scylla/src/transport/session_test.rs
@@ -10,7 +10,9 @@ use crate::statement::Consistency;
 use crate::tracing::TracingInfo;
 use crate::transport::cluster::Datacenter;
 use crate::transport::errors::{BadKeyspaceName, BadQuery, DbError, QueryError};
-use crate::transport::partitioner::{Murmur3Partitioner, Partitioner, PartitionerName};
+use crate::transport::partitioner::{
+    calculate_token_for_partition_key, Murmur3Partitioner, Partitioner, PartitionerName,
+};
 use crate::transport::topology::Strategy::NetworkTopologyStrategy;
 use crate::transport::topology::{
     CollectionType, ColumnKind, CqlType, NativeType, UserDefinedType,
@@ -229,7 +231,7 @@ async fn test_prepared_statement() {
                 .as_bigint()
                 .unwrap(),
         };
-        let prepared_token = Murmur3Partitioner::hash(
+        let prepared_token = Murmur3Partitioner.hash_one(
             &prepared_statement
                 .compute_partition_key(&serialized_values)
                 .unwrap(),
@@ -255,7 +257,7 @@ async fn test_prepared_statement() {
                 .as_bigint()
                 .unwrap(),
         };
-        let prepared_token = Murmur3Partitioner::hash(
+        let prepared_token = Murmur3Partitioner.hash_one(
             &prepared_complex_pk_statement
                 .compute_partition_key(&serialized_values)
                 .unwrap(),
@@ -527,7 +529,7 @@ async fn test_token_calculation() {
                 .as_bigint()
                 .unwrap(),
         };
-        let prepared_token = Murmur3Partitioner::hash(
+        let prepared_token = Murmur3Partitioner.hash_one(
             &prepared_statement
                 .compute_partition_key(&serialized_values)
                 .unwrap(),
@@ -1103,26 +1105,16 @@ async fn assert_in_tracing_table(session: &Session, tracing_uuid: Uuid) {
     panic!("No rows for tracing with this session id!");
 }
 
-#[tokio::test]
-async fn test_fetch_schema_version() {
-    let session = create_new_session_builder().build().await.unwrap();
-    session.fetch_schema_version().await.unwrap();
-}
-
 #[tokio::test]
 async fn test_await_schema_agreement() {
     let session = create_new_session_builder().build().await.unwrap();
-    session.await_schema_agreement().await.unwrap();
+    let _schema_version = session.await_schema_agreement().await.unwrap();
 }
 
 #[tokio::test]
 async fn test_await_timed_schema_agreement() {
-    use std::time::Duration;
     let session = create_new_session_builder().build().await.unwrap();
-    session
-        .await_timed_schema_agreement(Duration::from_millis(50))
-        .await
-        .unwrap();
+    session.await_schema_agreement().await.unwrap();
 }
 
 #[tokio::test]
@@ -2797,15 +2789,13 @@ async fn test_manual_primary_key_computation() {
             .await
             .unwrap();
 
-        let token_by_prepared = session
-            .calculate_token(prepared, &serialized_values_in_query_order)
+        let token_by_prepared = prepared
+            .calculate_token(&serialized_values_in_query_order)
             .unwrap()
             .unwrap();
-        let token_by_hand = Session::calculate_token_for_partition_key(
-            &serialized_values_in_pk_order,
-            &Murmur3Partitioner,
-        )
-        .unwrap();
+        let token_by_hand =
+            calculate_token_for_partition_key(&serialized_values_in_pk_order, &Murmur3Partitioner)
+                .unwrap();
         println!(
             "by_prepared: {}, by_hand: {}",
             token_by_prepared.value, token_by_hand.value
diff --git a/scylla/tests/integration/consistency.rs b/scylla/tests/integration/consistency.rs
index 41c32c73..9eaabf03 100644
--- a/scylla/tests/integration/consistency.rs
+++ b/scylla/tests/integration/consistency.rs
@@ -452,3 +452,24 @@ async fn consistency_is_correctly_set_in_routing_info() {
     )
     .await;
 }
+
+// Performs a read using Paxos, by setting Consistency to Serial.
+// This not really checks that functionality works properly, but stays here
+// to ensure that it is even expressible to issue such query.
+// Before, Consistency did not contain serial variants, so it used to be impossible.
+#[tokio::test]
+#[ntest::timeout(60000)]
+#[cfg(not(scylla_cloud_tests))]
+async fn consistency_allows_for_paxos_selects() {
+    let uri = std::env::var("SCYLLA_URI").unwrap_or_else(|_| "127.0.0.1:9042".to_string());
+
+    let session = SessionBuilder::new()
+        .known_node(uri.as_str())
+        .build()
+        .await
+        .unwrap();
+
+    let mut query = Query::from("SELECT host_id FROM system.peers WHERE peer = '127.0.0.1'");
+    query.set_consistency(Consistency::Serial);
+    session.query(query, ()).await.unwrap();
+}
diff --git a/scylla/tests/integration/execution_profiles.rs b/scylla/tests/integration/execution_profiles.rs
index 93334106..119487a6 100644
--- a/scylla/tests/integration/execution_profiles.rs
+++ b/scylla/tests/integration/execution_profiles.rs
@@ -9,7 +9,6 @@ use scylla::query::Query;
 use scylla::statement::SerialConsistency;
 use scylla::transport::NodeRef;
 use scylla::{
-    frame::types::LegacyConsistency,
     load_balancing::{LoadBalancingPolicy, RoutingInfo},
     retry_policy::{RetryPolicy, RetrySession},
     speculative_execution::SpeculativeExecutionPolicy,
@@ -34,14 +33,14 @@ enum Report {
 #[derive(Debug, Clone)]
 struct BoundToPredefinedNodePolicy<const NODE: u8> {
     profile_reporter: mpsc::UnboundedSender<(Report, u8)>,
-    consistency_reporter: mpsc::UnboundedSender<LegacyConsistency>,
+    consistency_reporter: mpsc::UnboundedSender<Consistency>,
 }
 
 impl<const NODE: u8> BoundToPredefinedNodePolicy<NODE> {
     fn report_node(&self, report: Report) {
         self.profile_reporter.send((report, NODE)).unwrap();
     }
-    fn report_consistency(&self, c: LegacyConsistency) {
+    fn report_consistency(&self, c: Consistency) {
         self.consistency_reporter.send(c).unwrap();
     }
 }
@@ -260,55 +259,55 @@ async fn test_execution_profiles() {
         // Run non-LWT on default per-session execution profile
         session.query(query.clone(), &[]).await.unwrap_err();
         let report_consistency = consistency_rx.recv().await.unwrap();
-        assert_matches!(report_consistency, LegacyConsistency::Regular(Consistency::One));
+        assert_matches!(report_consistency, Consistency::One);
         consistency_rx.try_recv().unwrap_err();
 
         session.execute(&prepared, &[]).await.unwrap_err();
         let report_consistency = consistency_rx.recv().await.unwrap();
-        assert_matches!(report_consistency, LegacyConsistency::Regular(Consistency::One));
+        assert_matches!(report_consistency, Consistency::One);
         consistency_rx.try_recv().unwrap_err();
 
         session.batch(&batch, ((),)).await.unwrap_err();
         let report_consistency = consistency_rx.recv().await.unwrap();
-        assert_matches!(report_consistency, LegacyConsistency::Regular(Consistency::One));
+        assert_matches!(report_consistency, Consistency::One);
         consistency_rx.try_recv().unwrap_err();
 
         // Run on statement-specific execution profile
         query.set_execution_profile_handle(Some(profile2.clone().into_handle()));
         session.query(query.clone(), &[]).await.unwrap_err();
         let report_consistency = consistency_rx.recv().await.unwrap();
-        assert_matches!(report_consistency, LegacyConsistency::Regular(Consistency::Two));
+        assert_matches!(report_consistency, Consistency::Two);
         consistency_rx.try_recv().unwrap_err();
 
         prepared.set_execution_profile_handle(Some(profile2.clone().into_handle()));
         session.execute(&prepared, &[]).await.unwrap_err();
         let report_consistency = consistency_rx.recv().await.unwrap();
-        assert_matches!(report_consistency, LegacyConsistency::Regular(Consistency::Two));
+        assert_matches!(report_consistency, Consistency::Two);
         consistency_rx.try_recv().unwrap_err();
 
         batch.set_execution_profile_handle(Some(profile2.clone().into_handle()));
         session.batch(&batch, ((),)).await.unwrap_err();
         let report_consistency = consistency_rx.recv().await.unwrap();
-        assert_matches!(report_consistency, LegacyConsistency::Regular(Consistency::Two));
+        assert_matches!(report_consistency, Consistency::Two);
         consistency_rx.try_recv().unwrap_err();
 
         // Run with statement-set specific options
         query.set_consistency(Consistency::Three);
         session.query(query.clone(), &[]).await.unwrap_err();
         let report_consistency = consistency_rx.recv().await.unwrap();
-        assert_matches!(report_consistency, LegacyConsistency::Regular(Consistency::Three));
+        assert_matches!(report_consistency, Consistency::Three);
         consistency_rx.try_recv().unwrap_err();
 
         prepared.set_consistency(Consistency::Three);
         session.execute(&prepared, &[]).await.unwrap_err();
         let report_consistency = consistency_rx.recv().await.unwrap();
-        assert_matches!(report_consistency, LegacyConsistency::Regular(Consistency::Three));
+        assert_matches!(report_consistency, Consistency::Three);
         consistency_rx.try_recv().unwrap_err();
 
         batch.set_consistency(Consistency::Three);
         session.batch(&batch, ((),)).await.unwrap_err();
         let report_consistency = consistency_rx.recv().await.unwrap();
-        assert_matches!(report_consistency, LegacyConsistency::Regular(Consistency::Three));
+        assert_matches!(report_consistency, Consistency::Three);
         consistency_rx.try_recv().unwrap_err();
 
         for i in 0..=2 {