Repository: scylladb/scylla-rust-driver
Pull request #743 (2023-06-12 16:54:29)

Fix RoutingInfo consistency
A bug has been found that put wrong `Consistency` and `SerialConsistency` into `RoutingInfo`. In order to:
- be sure that it's fully fixed,
- prevent regressions,
- check consistencies in other contexts,

the following is done:
- proxy acquires deserialization capabilities for request frames; it understands `Query`, `Execute` and `Batch` frames,
- a test is added to check that request frames contains consistencies that match the specified ones,
- a test is added to check that `RoutingInfo` for `LoadBalancingPolicy` contains the specified consistencies,
- as the latter test confirms the bug in `Session`, the affected methods are fixed,
- minor refactor is done in `Session` to make fiddling with consistency clearer,
- bonus: utilities for single-node dry-mode cluster are added, based on the bindings' test implementation. These will most probably prove useful for writing future tests.

## Pre-review checklist

- [x] I have split my patch into logically separate commits.
- [x] All commit messages clearly explain what they change and why.
- [x] I added relevant tests for new features and bug fixes.
- [x] All commits compile, pass static checks and pass test.
- [x] PR description sums up the changes and reasons why they should be introduced.
- [x] I have provided docstrings for the public items that I want to introduce.
- ~~[ ] I have adjusted the documentation in `./docs/source/`.~~
- ~~[ ] I added appropriate `Fixes:` annotations to PR description.~~



Patch:
diff --git a/examples/compare-tokens.rs b/examples/compare-tokens.rs
index 793aae2e..30c631a1 100644
--- a/examples/compare-tokens.rs
+++ b/examples/compare-tokens.rs
@@ -1,9 +1,8 @@
 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 scylla::{load_balancing, Session, SessionBuilder};
 use std::env;
 
 #[tokio::main]
@@ -33,11 +32,16 @@ async fn main() -> Result<()> {
         let serialized_pk = (pk,).serialized()?.into_owned();
         let t = Murmur3Partitioner::hash(&prepared.compute_partition_key(&serialized_pk)?).value;
 
+        let statement_info = load_balancing::RoutingInfo {
+            token: Some(scylla::routing::Token { value: t }),
+            keyspace: Some("ks"),
+            is_confirmed_lwt: false,
+            ..Default::default()
+        };
         println!(
-            "Token endpoints for query: {:?}",
+            "Estimated replicas for query: {:?}",
             session
-                .get_cluster_data()
-                .get_token_endpoints("ks", Token { value: t })
+                .estimate_replicas_for_query(&statement_info)
                 .iter()
                 .map(|n| n.address)
                 .collect::<Vec<NodeAddr>>()
diff --git a/examples/tracing.rs b/examples/tracing.rs
index bff0bca9..2e1d7df2 100644
--- a/examples/tracing.rs
+++ b/examples/tracing.rs
@@ -7,7 +7,7 @@ use scylla::batch::Batch;
 use scylla::statement::{
     prepared_statement::PreparedStatement, query::Query, Consistency, SerialConsistency,
 };
-use scylla::tracing::TracingInfo;
+use scylla::tracing::{GetTracingConfig, TracingInfo};
 use scylla::transport::iterator::RowIterator;
 use scylla::QueryResult;
 use scylla::{Session, SessionBuilder};
@@ -21,10 +21,7 @@ async fn main() -> Result<()> {
     let uri = env::var("SCYLLA_URI").unwrap_or_else(|_| "127.0.0.1:9042".to_string());
 
     println!("Connecting to {} ...", uri);
-    let session: Session = SessionBuilder::new()
-        .known_node(uri.as_str())
-        .build()
-        .await?;
+    let session: Session = SessionBuilder::new().known_node(uri).build().await?;
 
     session.query("CREATE KEYSPACE IF NOT EXISTS ks WITH REPLICATION = {'class' : 'NetworkTopologyStrategy', 'replication_factor' : 1}", &[]).await?;
 
@@ -109,18 +106,19 @@ async fn main() -> Result<()> {
     println!("Batch tracing id: {:?}\n", batch_result.tracing_id);
 
     // CUSTOM
-    // Session configuration allows specifying custom settings for querying tracing info.
+    // GetTracingConfig allows to specify a custom settings for querying tracing info
     // Tracing info might not immediately be available on queried node
     // so the driver performs a few attempts with sleeps in between.
-    let session: Session = SessionBuilder::new()
-        .known_node(uri)
-        .tracing_info_fetch_attempts(NonZeroU32::new(8).unwrap())
-        .tracing_info_fetch_interval(Duration::from_millis(100))
-        .tracing_info_fetch_consistency(Consistency::One)
-        .build()
-        .await?;
 
-    let _custom_info: TracingInfo = session.get_tracing_info(&query_tracing_id).await?;
+    let custom_config: GetTracingConfig = GetTracingConfig {
+        attempts: NonZeroU32::new(8).unwrap(),
+        interval: Duration::from_millis(100),
+        consistency: Consistency::One,
+    };
+
+    let _custom_info: TracingInfo = session
+        .get_tracing_info_custom(&query_tracing_id, &custom_config)
+        .await?;
 
     Ok(())
 }
diff --git a/scylla-cql/Cargo.toml b/scylla-cql/Cargo.toml
index a5eda905..36e8a76e 100644
--- a/scylla-cql/Cargo.toml
+++ b/scylla-cql/Cargo.toml
@@ -13,7 +13,7 @@ license = "MIT OR Apache-2.0"
 scylla-macros = { version = "0.2.0", path = "../scylla-macros" }
 byteorder = "1.3.4"
 bytes = "1.0.1"
-num_enum = "0.6"
+num_enum = "0.5"
 tokio = { version = "1.12", features = ["io-util", "time"] }
 secrecy = { version = "0.7.0", optional = true }
 snap = "1.0"
diff --git a/scylla-cql/benches/benchmark.rs b/scylla-cql/benches/benchmark.rs
index b8e9cbcc..0aa6c891 100644
--- a/scylla-cql/benches/benchmark.rs
+++ b/scylla-cql/benches/benchmark.rs
@@ -1,17 +1,19 @@
+use std::borrow::Cow;
+
 use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion};
 
-use scylla_cql::frame::request::Request;
+use scylla_cql::frame::request::SerializableRequest;
 use scylla_cql::frame::value::SerializedValues;
 use scylla_cql::frame::value::ValueList;
 use scylla_cql::frame::{request::query, Compression, SerializedRequest};
 
 fn make_query<'a>(contents: &'a str, values: &'a SerializedValues) -> query::Query<'a> {
     query::Query {
-        contents,
+        contents: Cow::Borrowed(contents),
         parameters: query::QueryParameters {
             consistency: scylla_cql::Consistency::LocalQuorum,
             serial_consistency: None,
-            values,
+            values: Cow::Borrowed(values),
             page_size: None,
             paging_state: None,
             timestamp: None,
@@ -25,7 +27,7 @@ fn serialized_request_make_bench(c: &mut Criterion) {
         ("INSERT foo INTO ks.table_name (?)", &(1234,).serialized().unwrap()),
         ("INSERT foo, bar, baz INTO ks.table_name (?, ?, ?)", &(1234, "a value", "i am storing a string").serialized().unwrap()),
         (
-            "INSERT foo, bar, baz, boop, blah INTO longer_keyspace.a_big_table_name (?, ?, ?, ?, 1000)", 
+            "INSERT foo, bar, baz, boop, blah INTO longer_keyspace.a_big_table_name (?, ?, ?, ?, 1000)",
             &(1234, "a value", "i am storing a string", "dc0c8cd7-d954-47c1-8722-a857941c43fb").serialized().unwrap()
         ),
     ];
diff --git a/scylla-cql/src/frame/mod.rs b/scylla-cql/src/frame/mod.rs
index e2f313cf..52e6dad2 100644
--- a/scylla-cql/src/frame/mod.rs
+++ b/scylla-cql/src/frame/mod.rs
@@ -16,7 +16,7 @@ use uuid::Uuid;
 
 use std::convert::TryFrom;
 
-use request::Request;
+use request::SerializableRequest;
 use response::ResponseOpcode;
 
 const HEADER_SIZE: usize = 9;
@@ -60,7 +60,7 @@ pub struct SerializedRequest {
 }
 
 impl SerializedRequest {
-    pub fn make<R: Request>(
+    pub fn make<R: SerializableRequest>(
         req: &R,
         compression: Option<Compression>,
         tracing: bool,
diff --git a/scylla-cql/src/frame/request/auth_response.rs b/scylla-cql/src/frame/request/auth_response.rs
index 61932774..aa515b8a 100644
--- a/scylla-cql/src/frame/request/auth_response.rs
+++ b/scylla-cql/src/frame/request/auth_response.rs
@@ -1,7 +1,7 @@
 use crate::frame::frame_errors::ParseError;
 use bytes::BufMut;
 
-use crate::frame::request::{Request, RequestOpcode};
+use crate::frame::request::{RequestOpcode, SerializableRequest};
 use crate::frame::types::write_bytes_opt;
 
 // Implements Authenticate Response
@@ -9,7 +9,7 @@ pub struct AuthResponse {
     pub response: Option<Vec<u8>>,
 }
 
-impl Request for AuthResponse {
+impl SerializableRequest for AuthResponse {
     const OPCODE: RequestOpcode = RequestOpcode::AuthResponse;
 
     fn serialize(&self, buf: &mut impl BufMut) -> Result<(), ParseError> {
diff --git a/scylla-cql/src/frame/request/batch.rs b/scylla-cql/src/frame/request/batch.rs
index d7876c21..91326a51 100644
--- a/scylla-cql/src/frame/request/batch.rs
+++ b/scylla-cql/src/frame/request/batch.rs
@@ -1,24 +1,28 @@
-use crate::frame::{frame_errors::ParseError, value::BatchValuesIterator};
-use bytes::{BufMut, Bytes};
-use std::convert::TryInto;
+use bytes::{Buf, BufMut};
+use std::{borrow::Cow, convert::TryInto};
 
 use crate::frame::{
-    request::{Request, RequestOpcode},
+    frame_errors::ParseError,
+    request::{RequestOpcode, SerializableRequest},
     types,
-    value::BatchValues,
+    value::{BatchValues, BatchValuesIterator, SerializedValues},
 };
 
+use super::DeserializableRequest;
+
 // Batch flags
 const FLAG_WITH_SERIAL_CONSISTENCY: u8 = 0x10;
 const FLAG_WITH_DEFAULT_TIMESTAMP: u8 = 0x20;
+const ALL_FLAGS: u8 = FLAG_WITH_SERIAL_CONSISTENCY | FLAG_WITH_DEFAULT_TIMESTAMP;
 
-pub struct Batch<'a, StatementsIter, Values>
+#[cfg_attr(test, derive(Debug, PartialEq, Eq))]
+pub struct Batch<'b, Statement, Values>
 where
-    StatementsIter: Iterator<Item = BatchStatement<'a>> + Clone,
+    BatchStatement<'b>: From<&'b Statement>,
+    Statement: Clone,
     Values: BatchValues,
 {
-    pub statements: StatementsIter,
-    pub statements_count: usize,
+    pub statements: Cow<'b, [Statement]>,
     pub batch_type: BatchType,
     pub consistency: types::Consistency,
     pub serial_consistency: Option<types::SerialConsistency>,
@@ -28,21 +32,46 @@ where
 
 /// The type of a batch.
 #[derive(Clone, Copy)]
+#[cfg_attr(test, derive(Debug, PartialEq, Eq))]
 pub enum BatchType {
     Logged = 0,
     Unlogged = 1,
     Counter = 2,
 }
 
-#[derive(Debug, Clone, Copy, Eq, PartialEq, PartialOrd, Ord)]
+pub struct BatchTypeParseError {
+    value: u8,
+}
+
+impl From<BatchTypeParseError> for ParseError {
+    fn from(err: BatchTypeParseError) -> Self {
+        Self::BadIncomingData(format!("Bad BatchType value: {}", err.value))
+    }
+}
+
+impl TryFrom<u8> for BatchType {
+    type Error = BatchTypeParseError;
+
+    fn try_from(value: u8) -> Result<Self, Self::Error> {
+        match value {
+            0 => Ok(Self::Logged),
+            1 => Ok(Self::Unlogged),
+            2 => Ok(Self::Counter),
+            _ => Err(BatchTypeParseError { value }),
+        }
+    }
+}
+
+#[derive(Debug, Clone, Eq, PartialEq, PartialOrd, Ord)]
 pub enum BatchStatement<'a> {
-    Query { text: &'a str },
-    Prepared { id: &'a Bytes },
+    Query { text: Cow<'a, str> },
+    Prepared { id: Cow<'a, [u8]> },
 }
 
-impl<'a, StatementsIter, Values> Request for Batch<'a, StatementsIter, Values>
+impl<Statement, Values> SerializableRequest for Batch<'_, Statement, Values>
 where
-    StatementsIter: Iterator<Item = BatchStatement<'a>> + Clone,
+    for<'s> BatchStatement<'s>: From<&'s Statement>,
+    Statement: Clone,
     Values: BatchValues,
 {
     const OPCODE: RequestOpcode = RequestOpcode::Batch;
@@ -52,7 +81,7 @@ where
         buf.put_u8(self.batch_type as u8);
 
         // Serializing queries
-        types::write_short(self.statements_count.try_into()?, buf);
+        types::write_short(self.statements.len().try_into()?, buf);
 
         let counts_mismatch_err = |n_values: usize, n_statements: usize| {
             ParseError::BadDataToSerialize(format!(
@@ -62,26 +91,27 @@ where
         };
         let mut n_serialized_statements = 0usize;
         let mut value_lists = self.values.batch_values_iter();
-        for (idx, statement) in self.statements.clone().enumerate() {
-            statement.serialize(buf)?;
+        for (idx, statement) in self.statements.iter().enumerate() {
+            BatchStatement::from(statement).serialize(buf)?;
             value_lists
                 .write_next_to_request(buf)
-                .ok_or_else(|| counts_mismatch_err(idx, self.statements.clone().count()))??;
+                .ok_or_else(|| counts_mismatch_err(idx, self.statements.len()))??;
             n_serialized_statements += 1;
         }
+        // At this point, we have all statements serialized. If any values are still left, we have a mismatch.
         if value_lists.skip_next().is_some() {
             return Err(counts_mismatch_err(
-                std::iter::from_fn(|| value_lists.skip_next()).count() + 1,
+                n_serialized_statements + 1 /*skipped above*/ + value_lists.count(),
                 n_serialized_statements,
             ));
         }
-        if n_serialized_statements != self.statements_count {
+        if n_serialized_statements != self.statements.len() {
             // We want to check this to avoid propagating an invalid construction of self.statements_count as a
             // hard-to-debug silent fail
             return Err(ParseError::BadDataToSerialize(format!(
                 "Invalid Batch constructed: not as many statements serialized as announced \
                     (batch.statement_count: {announced_statement_count}, {n_serialized_statements}",
-                announced_statement_count = self.statements_count
+                announced_statement_count = self.statements.len()
             )));
         }
 
@@ -110,19 +140,115 @@ where
     }
 }
 
+impl BatchStatement<'_> {
+    fn deserialize(buf: &mut &[u8]) -> Result<Self, ParseError> {
+        let kind = buf.get_u8();
+        match kind {
+            0 => {
+                let text = Cow::Owned(types::read_long_string(buf)?.to_owned());
+                Ok(BatchStatement::Query { text })
+            }
+            1 => {
+                let id = types::read_short_bytes(buf)?.to_vec().into();
+                Ok(BatchStatement::Prepared { id })
+            }
+            _ => Err(ParseError::BadIncomingData(format!(
+                "Unexpected batch statement kind: {}",
+                kind
+            ))),
+        }
+    }
+}
+
 impl BatchStatement<'_> {
     fn serialize(&self, buf: &mut impl BufMut) -> Result<(), ParseError> {
         match self {
-            BatchStatement::Query { text } => {
+            Self::Query { text } => {
                 buf.put_u8(0);
                 types::write_long_string(text, buf)?;
             }
-            BatchStatement::Prepared { id } => {
+            Self::Prepared { id } => {
                 buf.put_u8(1);
-                types::write_short_bytes(&id[..], buf)?;
+                types::write_short_bytes(id, buf)?;
             }
         }
 
         Ok(())
     }
 }
+
+impl<'s, 'b> From<&'s BatchStatement<'b>> for BatchStatement<'s> {
+    fn from(value: &'s BatchStatement) -> Self {
+        match value {
+            BatchStatement::Query { text } => BatchStatement::Query { text: text.clone() },
+            BatchStatement::Prepared { id } => BatchStatement::Prepared { id: id.clone() },
+        }
+    }
+}
+
+impl<'b> DeserializableRequest for Batch<'b, BatchStatement<'b>, Vec<SerializedValues>> {
+    fn deserialize(buf: &mut &[u8]) -> Result<Self, ParseError> {
+        let batch_type = buf.get_u8().try_into()?;
+
+        let statements_count: usize = types::read_short(buf)?.try_into()?;
+        let statements_with_values = (0..statements_count)
+            .map(|_| {
+                let batch_statement = BatchStatement::deserialize(buf)?;
+
+                // As stated in CQL protocol v4 specification, values names in Batch are broken and should be never used.
+                let values = SerializedValues::new_from_frame(buf, false)?;
+
+                Ok((batch_statement, values))
+            })
+            .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 flags = buf.get_u8();
+        let unknown_flags = flags & (!ALL_FLAGS);
+        if unknown_flags != 0 {
+            return Err(ParseError::BadIncomingData(format!(
+                "Specified flags are not recognised: {:02x}",
+                unknown_flags
+            )));
+        }
+        let serial_consistency_flag = (flags & FLAG_WITH_SERIAL_CONSISTENCY) != 0;
+        let default_timestamp_flag = (flags & FLAG_WITH_DEFAULT_TIMESTAMP) != 0;
+
+        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!(
+                        "Expected SerialConsistency, got regular Consistency {}",
+                        reg
+                    )))
+                }
+                types::LegacyConsistency::Serial(ser) => Ok(ser),
+            })
+            .transpose()?;
+
+        let timestamp = default_timestamp_flag
+            .then(|| types::read_long(buf))
+            .transpose()?;
+
+        let (statements, values): (Vec<BatchStatement>, Vec<SerializedValues>) =
+            statements_with_values.into_iter().unzip();
+
+        Ok(Self {
+            batch_type,
+            consistency,
+            serial_consistency,
+            timestamp,
+            statements: Cow::Owned(statements),
+            values,
+        })
+    }
+}
diff --git a/scylla-cql/src/frame/request/execute.rs b/scylla-cql/src/frame/request/execute.rs
index 5f8a35b9..59cdf1ef 100644
--- a/scylla-cql/src/frame/request/execute.rs
+++ b/scylla-cql/src/frame/request/execute.rs
@@ -2,16 +2,19 @@ use crate::frame::frame_errors::ParseError;
 use bytes::{BufMut, Bytes};
 
 use crate::{
-    frame::request::{query, Request, RequestOpcode},
+    frame::request::{query, RequestOpcode, SerializableRequest},
     frame::types,
 };
 
+use super::{query::QueryParameters, DeserializableRequest};
+
+#[cfg_attr(test, derive(Debug, PartialEq, Eq))]
 pub struct Execute<'a> {
     pub id: Bytes,
     pub parameters: query::QueryParameters<'a>,
 }
 
-impl Request for Execute<'_> {
+impl SerializableRequest for Execute<'_> {
     const OPCODE: RequestOpcode = RequestOpcode::Execute;
 
     fn serialize(&self, buf: &mut impl BufMut) -> Result<(), ParseError> {
@@ -23,3 +26,12 @@ impl Request for Execute<'_> {
         Ok(())
     }
 }
+
+impl<'e> DeserializableRequest for Execute<'e> {
+    fn deserialize(buf: &mut &[u8]) -> Result<Self, ParseError> {
+        let id = types::read_short_bytes(buf)?.to_vec().into();
+        let parameters = QueryParameters::deserialize(buf)?;
+
+        Ok(Self { id, parameters })
+    }
+}
diff --git a/scylla-cql/src/frame/request/mod.rs b/scylla-cql/src/frame/request/mod.rs
index f09dd3c8..8202ccde 100644
--- a/scylla-cql/src/frame/request/mod.rs
+++ b/scylla-cql/src/frame/request/mod.rs
@@ -7,17 +7,23 @@ pub mod query;
 pub mod register;
 pub mod startup;
 
-use crate::frame::frame_errors::ParseError;
+use crate::{frame::frame_errors::ParseError, Consistency};
 use bytes::{BufMut, Bytes};
 use num_enum::TryFromPrimitive;
 
 pub use auth_response::AuthResponse;
 pub use batch::Batch;
+pub use execute::Execute;
 pub use options::Options;
 pub use prepare::Prepare;
 pub use query::Query;
 pub use startup::Startup;
 
+use self::batch::BatchStatement;
+
+use super::types::SerialConsistency;
+use super::value::SerializedValues;
+
 #[derive(Debug, Copy, Clone, PartialEq, Eq, PartialOrd, Ord, TryFromPrimitive)]
 #[repr(u8)]
 pub enum RequestOpcode {
@@ -31,7 +37,7 @@ pub enum RequestOpcode {
     AuthResponse = 0x0F,
 }
 
-pub trait Request {
+pub trait SerializableRequest {
     const OPCODE: RequestOpcode;
 
     fn serialize(&self, buf: &mut impl BufMut) -> Result<(), ParseError>;
@@ -42,3 +48,242 @@ pub trait Request {
         Ok(v.into())
     }
 }
+
+/// Not intended for driver's direct usage (as driver has no interest in deserialising CQL requests),
+/// but very useful for testing (e.g. asserting that the sent requests have proper parameters set).
+pub trait DeserializableRequest: SerializableRequest + Sized {
+    fn deserialize(buf: &mut &[u8]) -> Result<Self, ParseError>;
+}
+
+#[non_exhaustive] // TODO: add remaining request types
+pub enum Request<'r> {
+    Query(Query<'r>),
+    Execute(Execute<'r>),
+    Batch(Batch<'r, BatchStatement<'r>, Vec<SerializedValues>>),
+}
+
+impl<'r> Request<'r> {
+    pub fn deserialize(buf: &mut &[u8], opcode: RequestOpcode) -> Result<Self, ParseError> {
+        match opcode {
+            RequestOpcode::Query => Query::deserialize(buf).map(Self::Query),
+            RequestOpcode::Execute => Execute::deserialize(buf).map(Self::Execute),
+            RequestOpcode::Batch => Batch::deserialize(buf).map(Self::Batch),
+            _ => unimplemented!(
+                "Deserialization of opcode {:?} is not yet supported",
+                opcode
+            ),
+        }
+    }
+
+    /// Retrieves consistency from request frame, if present.
+    pub fn get_consistency(&self) -> Option<Consistency> {
+        match self {
+            Request::Query(q) => Some(q.parameters.consistency),
+            Request::Execute(e) => Some(e.parameters.consistency),
+            Request::Batch(b) => Some(b.consistency),
+            #[allow(unreachable_patterns)] // until other opcodes are supported
+            _ => None,
+        }
+    }
+
+    /// Retrieves serial consistency from request frame.
+    pub fn get_serial_consistency(&self) -> Option<Option<SerialConsistency>> {
+        match self {
+            Request::Query(q) => Some(q.parameters.serial_consistency),
+            Request::Execute(e) => Some(e.parameters.serial_consistency),
+            Request::Batch(b) => Some(b.serial_consistency),
+            #[allow(unreachable_patterns)] // until other opcodes are supported
+            _ => None,
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::{borrow::Cow, ops::Deref};
+
+    use bytes::Bytes;
+
+    use crate::{
+        frame::{
+            request::{
+                batch::{Batch, BatchStatement, BatchType},
+                execute::Execute,
+                query::{Query, QueryParameters},
+                DeserializableRequest, SerializableRequest,
+            },
+            types::{self, LegacyConsistency, SerialConsistency},
+            value::SerializedValues,
+        },
+        Consistency,
+    };
+
+    #[test]
+    fn request_ser_de_identity() {
+        // Query
+        let contents = Cow::Borrowed("SELECT host_id from system.peers");
+        let parameters = QueryParameters {
+            consistency: Consistency::All,
+            serial_consistency: Some(SerialConsistency::Serial),
+            timestamp: None,
+            page_size: Some(323),
+            paging_state: Some(vec![2, 1, 3, 7].into()),
+            values: {
+                let mut vals = SerializedValues::new();
+                vals.add_value(&2137).unwrap();
+                Cow::Owned(vals)
+            },
+        };
+        let query = Query {
+            contents,
+            parameters,
+        };
+
+        {
+            let mut buf = Vec::new();
+            query.serialize(&mut buf).unwrap();
+
+            let query_deserialized = Query::deserialize(&mut &buf[..]).unwrap();
+            assert_eq!(&query_deserialized, &query);
+        }
+
+        // Execute
+        let id: Bytes = vec![2, 4, 5, 2, 6, 7, 3, 1].into();
+        let parameters = QueryParameters {
+            consistency: Consistency::Any,
+            serial_consistency: None,
+            timestamp: Some(3423434),
+            page_size: None,
+            paging_state: None,
+            values: {
+                let mut vals = SerializedValues::new();
+                vals.add_named_value("the_answer", &42).unwrap();
+                vals.add_named_value("really?", &2137).unwrap();
+                Cow::Owned(vals)
+            },
+        };
+        let execute = Execute { id, parameters };
+        {
+            let mut buf = Vec::new();
+            execute.serialize(&mut buf).unwrap();
+
+            let execute_deserialized = Execute::deserialize(&mut &buf[..]).unwrap();
+            assert_eq!(&execute_deserialized, &execute);
+        }
+
+        // Batch
+        let statements = vec![
+            BatchStatement::Query {
+                text: query.contents,
+            },
+            BatchStatement::Prepared {
+                id: Cow::Borrowed(&execute.id),
+            },
+        ];
+        let batch = Batch {
+            statements: Cow::Owned(statements),
+            batch_type: BatchType::Logged,
+            consistency: Consistency::EachQuorum,
+            serial_consistency: Some(SerialConsistency::LocalSerial),
+            timestamp: Some(32432),
+
+            // Not execute's values, because named values are not supported in batches.
+            values: vec![
+                query.parameters.values.deref().clone(),
+                query.parameters.values.deref().clone(),
+            ],
+        };
+        {
+            let mut buf = Vec::new();
+            batch.serialize(&mut buf).unwrap();
+
+            let batch_deserialized = Batch::deserialize(&mut &buf[..]).unwrap();
+            assert_eq!(&batch_deserialized, &batch);
+        }
+    }
+
+    #[test]
+    fn deser_rejects_unknown_flags() {
+        // Query
+        let contents = Cow::Borrowed("SELECT host_id from system.peers");
+        let parameters = QueryParameters {
+            consistency: Default::default(),
+            serial_consistency: Some(SerialConsistency::LocalSerial),
+            timestamp: None,
+            page_size: None,
+            paging_state: None,
+            values: Cow::Owned(SerializedValues::new()),
+        };
+        let query = Query {
+            contents: contents.clone(),
+            parameters,
+        };
+
+        {
+            let mut buf = Vec::new();
+            query.serialize(&mut buf).unwrap();
+
+            // Sanity check: query deserializes to the equivalent.
+            let query_deserialized = Query::deserialize(&mut &buf[..]).unwrap();
+            assert_eq!(&query_deserialized.contents, &query.contents);
+            assert_eq!(&query_deserialized.parameters, &query.parameters);
+
+            // Now modify flags by adding an unknown one.
+            // Find flags in buffer:
+            let mut buf_ptr = buf.as_slice();
+            let serialised_contents = types::read_long_string(&mut buf_ptr).unwrap();
+            assert_eq!(serialised_contents, contents);
+
+            // Now buf_ptr points at consistency.
+            let consistency = types::read_consistency(&mut buf_ptr).unwrap();
+            assert_eq!(
+                consistency,
+                LegacyConsistency::Regular(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();
+            let flags_mut = &mut buf[flags_idx];
+
+            // This assumes that the following flag is unknown, which is true at the time of writing this test.
+            *flags_mut |= 0x80;
+
+            // Unknown flag should lead to frame rejection, as unknown flags can be new protocol extensions
+            // leading to different semantics.
+            let _parse_error = Query::deserialize(&mut &buf[..]).unwrap_err();
+        }
+
+        // Batch
+        let statements = vec![BatchStatement::Query {
+            text: query.contents,
+        }];
+        let batch = Batch {
+            statements: Cow::Owned(statements),
+            batch_type: BatchType::Logged,
+            consistency: Consistency::EachQuorum,
+            serial_consistency: None,
+            timestamp: None,
+
+            values: vec![query.parameters.values.deref().clone()],
+        };
+        {
+            let mut buf = Vec::new();
+            batch.serialize(&mut buf).unwrap();
+
+            // Sanity check: batch deserializes to the equivalent.
+            let batch_deserialized = Batch::deserialize(&mut &buf[..]).unwrap();
+            assert_eq!(batch, batch_deserialized);
+
+            // Now modify flags by adding an unknown one.
+            // There are no timestamp nor serial consistency, so flags are the last byte in the buf.
+            let buf_len = buf.len();
+            let flags_mut = &mut buf[buf_len - 1];
+            // This assumes that the following flag is unknown, which is true at the time of writing this test.
+            *flags_mut |= 0x80;
+
+            // Unknown flag should lead to frame rejection, as unknown flags can be new protocol extensions
+            // leading to different semantics.
+            let _parse_error = Batch::deserialize(&mut &buf[..]).unwrap_err();
+        }
+    }
+}
diff --git a/scylla-cql/src/frame/request/options.rs b/scylla-cql/src/frame/request/options.rs
index 5a0561ed..a1a5e8d5 100644
--- a/scylla-cql/src/frame/request/options.rs
+++ b/scylla-cql/src/frame/request/options.rs
@@ -1,11 +1,11 @@
 use crate::frame::frame_errors::ParseError;
 use bytes::BufMut;
 
-use crate::frame::request::{Request, RequestOpcode};
+use crate::frame::request::{RequestOpcode, SerializableRequest};
 
 pub struct Options;
 
-impl Request for Options {
+impl SerializableRequest for Options {
     const OPCODE: RequestOpcode = RequestOpcode::Options;
 
     fn serialize(&self, _buf: &mut impl BufMut) -> Result<(), ParseError> {
diff --git a/scylla-cql/src/frame/request/prepare.rs b/scylla-cql/src/frame/request/prepare.rs
index 7141f439..d4273891 100644
--- a/scylla-cql/src/frame/request/prepare.rs
+++ b/scylla-cql/src/frame/request/prepare.rs
@@ -2,7 +2,7 @@ use crate::frame::frame_errors::ParseError;
 use bytes::BufMut;
 
 use crate::{
-    frame::request::{Request, RequestOpcode},
+    frame::request::{RequestOpcode, SerializableRequest},
     frame::types,
 };
 
@@ -10,7 +10,7 @@ pub struct Prepare<'a> {
     pub query: &'a str,
 }
 
-impl<'a> Request for Prepare<'a> {
+impl<'a> SerializableRequest for Prepare<'a> {
     const OPCODE: RequestOpcode = RequestOpcode::Prepare;
 
     fn serialize(&self, buf: &mut impl BufMut) -> Result<(), ParseError> {
diff --git a/scylla-cql/src/frame/request/query.rs b/scylla-cql/src/frame/request/query.rs
index 70286721..1b5a52ed 100644
--- a/scylla-cql/src/frame/request/query.rs
+++ b/scylla-cql/src/frame/request/query.rs
@@ -1,44 +1,67 @@
+use std::borrow::Cow;
+
 use crate::frame::frame_errors::ParseError;
-use bytes::{BufMut, Bytes};
+use bytes::{Buf, BufMut, Bytes};
 
 use crate::{
-    frame::request::{Request, RequestOpcode},
+    frame::request::{RequestOpcode, SerializableRequest},
     frame::types,
     frame::value::SerializedValues,
 };
 
+use super::DeserializableRequest;
+
 // Query flags
-// Unused flags are commented out so that they don't trigger warnings
 const FLAG_VALUES: u8 = 0x01;
-// const FLAG_SKIP_METADATA: u8 = 0x02;
+const FLAG_SKIP_METADATA: u8 = 0x02;
 const FLAG_PAGE_SIZE: u8 = 0x04;
 const FLAG_WITH_PAGING_STATE: u8 = 0x08;
 const FLAG_WITH_SERIAL_CONSISTENCY: u8 = 0x10;
 const FLAG_WITH_DEFAULT_TIMESTAMP: u8 = 0x20;
 const FLAG_WITH_NAMES_FOR_VALUES: u8 = 0x40;
-
-pub struct Query<'a> {
-    pub contents: &'a str,
-    pub parameters: QueryParameters<'a>,
+const ALL_FLAGS: u8 = FLAG_VALUES
+    | FLAG_SKIP_METADATA
+    | FLAG_PAGE_SIZE
+    | FLAG_WITH_PAGING_STATE
+    | FLAG_WITH_SERIAL_CONSISTENCY
+    | FLAG_WITH_DEFAULT_TIMESTAMP
+    | FLAG_WITH_NAMES_FOR_VALUES;
+
+#[cfg_attr(test, derive(Debug, PartialEq, Eq))]
+pub struct Query<'q> {
+    pub contents: Cow<'q, str>,
+    pub parameters: QueryParameters<'q>,
 }
 
-impl Request for Query<'_> {
+impl SerializableRequest for Query<'_> {
     const OPCODE: RequestOpcode = RequestOpcode::Query;
 
     fn serialize(&self, buf: &mut impl BufMut) -> Result<(), ParseError> {
-        types::write_long_string(self.contents, buf)?;
+        types::write_long_string(&self.contents, buf)?;
         self.parameters.serialize(buf)?;
         Ok(())
     }
 }
 
+impl<'q> DeserializableRequest for Query<'q> {
+    fn deserialize(buf: &mut &[u8]) -> Result<Self, ParseError> {
+        let contents = Cow::Owned(types::read_long_string(buf)?.to_owned());
+        let parameters = QueryParameters::deserialize(buf)?;
+
+        Ok(Self {
+            contents,
+            parameters,
+        })
+    }
+}
+#[cfg_attr(test, derive(Debug, PartialEq, Eq))]
 pub struct QueryParameters<'a> {
     pub consistency: types::Consistency,
     pub serial_consistency: Option<types::SerialConsistency>,
     pub timestamp: Option<i64>,
     pub page_size: Option<i32>,
     pub paging_state: Option<Bytes>,
-    pub values: &'a SerializedValues,
+    pub values: Cow<'a, SerializedValues>,
 }
 
 impl Default for QueryParameters<'_> {
@@ -49,7 +72,7 @@ impl Default for QueryParameters<'_> {
             timestamp: None,
             page_size: None,
             paging_state: None,
-            values: SerializedValues::EMPTY,
+            values: Cow::Borrowed(SerializedValues::EMPTY),
         }
     }
 }
@@ -108,3 +131,70 @@ impl QueryParameters<'_> {
         Ok(())
     }
 }
+
+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 flags = buf.get_u8();
+        let unknown_flags = flags & (!ALL_FLAGS);
+        if unknown_flags != 0 {
+            return Err(ParseError::BadIncomingData(format!(
+                "Specified flags are not recognised: {:02x}",
+                unknown_flags
+            )));
+        }
+        let values_flag = (flags & FLAG_VALUES) != 0;
+        let page_size_flag = (flags & FLAG_PAGE_SIZE) != 0;
+        let paging_state_flag = (flags & FLAG_WITH_PAGING_STATE) != 0;
+        let serial_consistency_flag = (flags & FLAG_WITH_SERIAL_CONSISTENCY) != 0;
+        let default_timestamp_flag = (flags & FLAG_WITH_DEFAULT_TIMESTAMP) != 0;
+        let values_have_names_flag = (flags & FLAG_WITH_NAMES_FOR_VALUES) != 0;
+
+        let values = Cow::Owned(if values_flag {
+            SerializedValues::new_from_frame(buf, values_have_names_flag)?
+        } else {
+            SerializedValues::new()
+        });
+
+        let page_size = page_size_flag.then(|| types::read_int(buf)).transpose()?;
+        let paging_state = if paging_state_flag {
+            Some(Bytes::copy_from_slice(types::read_bytes(buf)?))
+        } else {
+            None
+        };
+        let serial_consistency = if serial_consistency_flag {
+            match types::read_consistency(buf)? {
+                types::LegacyConsistency::Regular(reg) => {
+                    return Err(ParseError::BadIncomingData(format!(
+                        "Expected SerialConsistency, got regular Consistency {}",
+                        reg
+                    )))
+                }
+                types::LegacyConsistency::Serial(ser) => Some(ser),
+            }
+        } else {
+            None
+        };
+        let timestamp = if default_timestamp_flag {
+            Some(types::read_long(buf)?)
+        } else {
+            None
+        };
+
+        Ok(Self {
+            consistency,
+            serial_consistency,
+            timestamp,
+            page_size,
+            paging_state,
+            values,
+        })
+    }
+}
diff --git a/scylla-cql/src/frame/request/register.rs b/scylla-cql/src/frame/request/register.rs
index 2c008b05..d8f88ea3 100644
--- a/scylla-cql/src/frame/request/register.rs
+++ b/scylla-cql/src/frame/request/register.rs
@@ -2,7 +2,7 @@ use bytes::BufMut;
 
 use crate::frame::{
     frame_errors::ParseError,
-    request::{Request, RequestOpcode},
+    request::{RequestOpcode, SerializableRequest},
     server_event_type::EventType,
     types,
 };
@@ -11,7 +11,7 @@ pub struct Register {
     pub event_types_to_register_for: Vec<EventType>,
 }
 
-impl Request for Register {
+impl SerializableRequest for Register {
     const OPCODE: RequestOpcode = RequestOpcode::Register;
 
     fn serialize(&self, buf: &mut impl BufMut) -> Result<(), ParseError> {
diff --git a/scylla-cql/src/frame/request/startup.rs b/scylla-cql/src/frame/request/startup.rs
index 5e75d85f..044a9883 100644
--- a/scylla-cql/src/frame/request/startup.rs
+++ b/scylla-cql/src/frame/request/startup.rs
@@ -4,7 +4,7 @@ use bytes::BufMut;
 use std::collections::HashMap;
 
 use crate::{
-    frame::request::{Request, RequestOpcode},
+    frame::request::{RequestOpcode, SerializableRequest},
     frame::types,
 };
 
@@ -12,7 +12,7 @@ pub struct Startup {
     pub options: HashMap<String, String>,
 }
 
-impl Request for Startup {
+impl SerializableRequest for Startup {
     const OPCODE: RequestOpcode = RequestOpcode::Startup;
 
     fn serialize(&self, buf: &mut impl BufMut) -> Result<(), ParseError> {
diff --git a/scylla-cql/src/frame/types.rs b/scylla-cql/src/frame/types.rs
index 858db35f..c7daff52 100644
--- a/scylla-cql/src/frame/types.rs
+++ b/scylla-cql/src/frame/types.rs
@@ -12,7 +12,7 @@ use std::net::SocketAddr;
 use std::str;
 use uuid::Uuid;
 
-#[derive(Debug, Copy, Clone, Default, PartialEq, Eq, PartialOrd, Ord, TryFromPrimitive)]
+#[derive(Debug, Copy, Clone, PartialEq, Eq, PartialOrd, Ord, TryFromPrimitive)]
 #[cfg_attr(feature = "serde", derive(serde::Deserialize))]
 #[cfg_attr(feature = "serde", serde(rename_all = "SCREAMING_SNAKE_CASE"))]
 #[repr(i16)]
@@ -23,7 +23,6 @@ pub enum Consistency {
     Three = 0x0003,
     Quorum = 0x0004,
     All = 0x0005,
-    #[default]
     LocalQuorum = 0x0006,
     EachQuorum = 0x0007,
     LocalOne = 0x000A,
@@ -47,6 +46,22 @@ pub enum LegacyConsistency {
     Serial(SerialConsistency),
 }
 
+// Although we could `impl Default for Consistency` with an automatic derive,
+// this would require adding a #[default] annotation on the default variant,
+// but - unfortunately - that annotation is also recognized by `TryFromPrimitive`
+// derive macro. If there is a #[default] variant then `TryFromPrimitive`
+// falls back to it - if not, then it returns an error. This breaks one of the
+// tests.
+//
+// It will be possible to fix this properly after the following issue is closed:
+// https://github.com/illicitonion/num_enum/issues/75
+#[allow(clippy::derivable_impls)]
+impl Default for Consistency {
+    fn default() -> Self {
+        Consistency::LocalQuorum
+    }
+}
+
 impl std::fmt::Display for Consistency {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         write!(f, "{:?}", self)
diff --git a/scylla-cql/src/frame/value.rs b/scylla-cql/src/frame/value.rs
index 8791a563..9cc482c2 100644
--- a/scylla-cql/src/frame/value.rs
+++ b/scylla-cql/src/frame/value.rs
@@ -1,3 +1,4 @@
+use crate::frame::frame_errors::ParseError;
 use crate::frame::types;
 use bigdecimal::BigDecimal;
 use bytes::BufMut;
@@ -197,6 +198,39 @@ impl SerializedValues {
     pub fn size(&self) -> usize {
         self.serialized_values.len()
     }
+
+    /// Creates value list from the request frame
+    pub fn new_from_frame(buf: &mut &[u8], contains_names: bool) -> Result<Self, ParseError> {
+        let values_num = types::read_short(buf)?;
+        let values_beg = *buf;
+        for _ in 0..values_num {
+            if contains_names {
+                let _name = types::read_string(buf)?;
+            }
+            let _serialized = types::read_bytes_opt(buf)?;
+        }
+
+        let values_len_in_buf = values_beg.len() - buf.len();
+        let values_in_frame = &values_beg[0..values_len_in_buf];
+        Ok(SerializedValues {
+            serialized_values: values_in_frame.to_vec(),
+            values_num,
+            contains_names,
+        })
+    }
+
+    pub fn iter_name_value_pairs(&self) -> impl Iterator<Item = (Option<&str>, &[u8])> {
+        let mut buf = &self.serialized_values[..];
+        (0..self.values_num).map(move |_| {
+            // `unwrap()`s here are safe, as we assume type-safety: if `SerializedValues` exits,
+            // we have a guarantee that the layout of the serialized values is valid.
+            let name = self
+                .contains_names
+                .then(|| types::read_string(&mut buf).unwrap());
+            let serialized = types::read_bytes(&mut buf).unwrap();
+            (name, serialized)
+        })
+    }
 }
 
 #[derive(Clone, Copy)]
@@ -249,6 +283,16 @@ pub trait BatchValuesIterator<'a> {
         buf: &mut impl BufMut,
     ) -> Option<Result<(), SerializeValuesError>>;
     fn skip_next(&mut self) -> Option<()>;
+    fn count(mut self) -> usize
+    where
+        Self: Sized,
+    {
+        let mut count = 0;
+        while self.skip_next().is_some() {
+            count += 1;
+        }
+        count
+    }
 }
 
 /// Implements `BatchValuesIterator` from an `Iterator` over references to things that implement `ValueList`
diff --git a/scylla-proxy/src/frame.rs b/scylla-proxy/src/frame.rs
index 0a4bbe89..cf61a6b0 100644
--- a/scylla-proxy/src/frame.rs
+++ b/scylla-proxy/src/frame.rs
@@ -3,6 +3,7 @@ use std::collections::HashMap;
 use bytes::{Buf, BufMut, Bytes, BytesMut};
 use scylla_cql::frame::frame_errors::{FrameError, ParseError};
 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;
@@ -69,6 +70,10 @@ impl RequestFrame {
         )
         .await
     }
+
+    pub fn deserialize(&self) -> Result<Request, ParseError> {
+        Request::deserialize(&mut &self.body[..], self.opcode)
+    }
 }
 #[derive(Clone, Debug)]
 pub struct ResponseFrame {
diff --git a/scylla/Cargo.toml b/scylla/Cargo.toml
index 426e5d9a..b34fca0f 100644
--- a/scylla/Cargo.toml
+++ b/scylla/Cargo.toml
@@ -26,7 +26,7 @@ byteorder = "1.3.4"
 bytes = "1.0.1"
 futures = "0.3.6"
 histogram = "0.6.9"
-num_enum = "0.6"
+num_enum = "0.5"
 tokio = { version = "1.27", features = ["net", "time", "io-util", "sync", "rt", "macros"] }
 snap = "1.0"
 uuid = { version = "1.0", features = ["v4"] }
diff --git a/scylla/src/statement/batch.rs b/scylla/src/statement/batch.rs
index 90286eca..6b440f1f 100644
--- a/scylla/src/statement/batch.rs
+++ b/scylla/src/statement/batch.rs
@@ -1,3 +1,4 @@
+use std::borrow::Cow;
 use std::sync::Arc;
 
 use crate::history::HistoryListener;
@@ -181,3 +182,22 @@ impl From<PreparedStatement> for BatchStatement {
         BatchStatement::PreparedStatement(p)
     }
 }
+
+impl<'a: 'b, 'b> From<&'a BatchStatement>
+    for scylla_cql::frame::request::batch::BatchStatement<'b>
+{
+    fn from(val: &'a BatchStatement) -> Self {
+        match val {
+            BatchStatement::Query(query) => {
+                scylla_cql::frame::request::batch::BatchStatement::Query {
+                    text: Cow::Borrowed(&query.contents),
+                }
+            }
+            BatchStatement::PreparedStatement(prepared) => {
+                scylla_cql::frame::request::batch::BatchStatement::Prepared {
+                    id: Cow::Borrowed(prepared.get_id()),
+                }
+            }
+        }
+    }
+}
diff --git a/scylla/src/tracing.rs b/scylla/src/tracing.rs
index 3d40e364..dbdff599 100644
--- a/scylla/src/tracing.rs
+++ b/scylla/src/tracing.rs
@@ -1,6 +1,9 @@
+use crate::statement::Consistency;
 use itertools::Itertools;
 use std::collections::HashMap;
 use std::net::IpAddr;
+use std::num::NonZeroU32;
+use std::time::Duration;
 use uuid::Uuid;
 
 use crate::cql_to_rust::{FromRow, FromRowError};
@@ -32,6 +35,20 @@ pub struct TracingEvent {
     pub thread: Option<String>,
 }
 
+/// Used to configure a custom retry strategy when querying tracing info
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct GetTracingConfig {
+    /// Number of attempts to be made before giving up.
+    /// Default value: 5
+    pub attempts: NonZeroU32,
+    /// Interval to wait between each attempt.
+    /// Default value: 3 milliseconds
+    pub interval: Duration,
+    /// Consistency to use in queries that read TracingInfo.
+    /// Default value: One
+    pub consistency: Consistency,
+}
+
 impl TracingInfo {
     /// Returns a list of unique nodes involved in the query
     pub fn nodes(&self) -> Vec<IpAddr> {
@@ -43,6 +60,16 @@ impl TracingInfo {
     }
 }
 
+impl Default for GetTracingConfig {
+    fn default() -> GetTracingConfig {
+        GetTracingConfig {
+            attempts: NonZeroU32::new(5).unwrap(),
+            interval: Duration::from_millis(3),
+            consistency: Consistency::One,
+        }
+    }
+}
+
 // A query used to query TracingInfo from system_traces.sessions
 pub(crate) const TRACES_SESSION_QUERY_STR: &str =
     "SELECT client, command, coordinator, duration, parameters, request, started_at \
diff --git a/scylla/src/transport/connection.rs b/scylla/src/transport/connection.rs
index 7017104d..41ab3ce5 100644
--- a/scylla/src/transport/connection.rs
+++ b/scylla/src/transport/connection.rs
@@ -13,6 +13,7 @@ use tracing::instrument::WithSubscriber;
 use tracing::{debug, error, trace, warn};
 use uuid::Uuid;
 
+use std::borrow::Cow;
 #[cfg(feature = "ssl")]
 use std::pin::Pin;
 use std::sync::atomic::AtomicU64;
@@ -48,7 +49,7 @@ use crate::batch::{Batch, BatchStatement};
 use crate::frame::protocol_features::ProtocolFeatures;
 use crate::frame::{
     self,
-    request::{self, batch, execute, query, register, Request},
+    request::{self, batch, execute, query, register, SerializableRequest},
     response::{event::Event, result, NonErrorResponse, Response, ResponseOpcode},
     server_event_type::EventType,
     value::{BatchValues, ValueList},
@@ -105,7 +106,7 @@ impl RouterHandle {
 
     async fn send_request(
         &self,
-        request: &impl Request,
+        request: &impl SerializableRequest,
         compression: Option<Compression>,
         tracing: bool,
     ) -> Result<TaskResponse, QueryError> {
@@ -655,11 +656,11 @@ impl Connection {
         let serialized_values = values.serialized()?;
 
         let query_frame = query::Query {
-            contents: &query.contents,
+            contents: Cow::Borrowed(&query.contents),
             parameters: query::QueryParameters {
                 consistency,
                 serial_consistency,
-                values: &serialized_values,
+                values: serialized_values,
                 page_size: query.get_page_size(),
                 paging_state,
                 timestamp: query.get_timestamp(),
@@ -685,7 +686,7 @@ impl Connection {
             parameters: query::QueryParameters {
                 consistency,
                 serial_consistency,
-                values: &serialized_values,
+                values: serialized_values,
                 page_size: prepared_statement.get_page_size(),
                 timestamp: prepared_statement.get_timestamp(),
                 paging_state,
@@ -760,16 +761,8 @@ impl Connection {
         consistency: Consistency,
         serial_consistency: Option<SerialConsistency>,
     ) -> Result<QueryResult, QueryError> {
-        let statements_iter = batch.statements.iter().map(|s| match s {
-            BatchStatement::Query(q) => batch::BatchStatement::Query { text: &q.contents },
-            BatchStatement::PreparedStatement(s) => {
-                batch::BatchStatement::Prepared { id: s.get_id() }
-            }
-        });
-
         let batch_frame = batch::Batch {
-            statements_count: statements_iter.len(),
-            statements: statements_iter,
+            statements: Cow::Borrowed(&batch.statements),
             values,
             batch_type: batch.get_type(),
             consistency,
@@ -879,7 +872,7 @@ impl Connection {
 
     async fn send_request(
         &self,
-        request: &impl Request,
+        request: &impl SerializableRequest,
         compress: bool,
         tracing: bool,
     ) -> Result<QueryResponse, QueryError> {
diff --git a/scylla/src/transport/iterator.rs b/scylla/src/transport/iterator.rs
index 2557c6a2..f132f12d 100644
--- a/scylla/src/transport/iterator.rs
+++ b/scylla/src/transport/iterator.rs
@@ -151,6 +151,12 @@ impl RowIterator {
             .serial_consistency
             .unwrap_or(execution_profile.serial_consistency);
 
+        let routing_info = RoutingInfo {
+            consistency,
+            serial_consistency,
+            ..Default::default()
+        };
+
         let retry_session = query
             .get_retry_policy()
             .map(|rp| &**rp)
@@ -188,7 +194,7 @@ impl RowIterator {
                 sender: sender.into(),
                 choose_connection,
                 page_query,
-                statement_info: RoutingInfo::default(),
+                statement_info: routing_info,
                 query_is_idempotent: query.config.is_idempotent,
                 query_consistency: consistency,
                 retry_session,
@@ -237,7 +243,7 @@ impl RowIterator {
         let worker_task = async move {
             let statement_info = RoutingInfo {
                 consistency,
-                serial_consistency: config.prepared.get_serial_consistency(),
+                serial_consistency,
                 token: config.token,
                 keyspace: config.prepared.get_keyspace_name(),
                 is_confirmed_lwt: config.prepared.is_confirmed_lwt(),
diff --git a/scylla/src/transport/session.rs b/scylla/src/transport/session.rs
index 0f8ce566..9e04f6a7 100644
--- a/scylla/src/transport/session.rs
+++ b/scylla/src/transport/session.rs
@@ -27,7 +27,6 @@ use std::fmt::Display;
 use std::future::Future;
 use std::io;
 use std::net::SocketAddr;
-use std::num::NonZeroU32;
 use std::str::FromStr;
 use std::sync::atomic::AtomicUsize;
 use std::sync::atomic::Ordering;
@@ -60,7 +59,7 @@ use crate::prepared_statement::{PartitionKeyError, PreparedStatement};
 use crate::query::Query;
 use crate::routing::Token;
 use crate::statement::{Consistency, SerialConsistency};
-use crate::tracing::{TracingEvent, TracingInfo};
+use crate::tracing::{GetTracingConfig, TracingEvent, TracingInfo};
 use crate::transport::cluster::{Cluster, ClusterData, ClusterNeatDebug};
 use crate::transport::connection::{Connection, ConnectionConfig, VerifiedKeyspaceName};
 use crate::transport::connection_pool::PoolConfig;
@@ -134,9 +133,6 @@ pub struct Session {
     auto_await_schema_agreement_timeout: Option<Duration>,
     refresh_metadata_on_auto_schema_agreement: bool,
     keyspace_name: ArcSwapOption<String>,
-    tracing_info_fetch_attempts: NonZeroU32,
-    tracing_info_fetch_interval: Duration,
-    tracing_info_fetch_consistency: Consistency,
 }
 
 /// This implementation deliberately omits some details from Cluster in order
@@ -242,22 +238,6 @@ pub struct SessionConfig {
     /// Please do performance measurements before committing to disabling
     /// this option.
     pub enable_write_coalescing: bool,
-
-    /// Number of attempts to fetch [`TracingInfo`]
-    /// in [`Session::get_tracing_info`]. Tracing info
-    /// might not be available immediately on queried node - that's why
-    /// the driver performs a few attempts with sleeps in between.
-    pub tracing_info_fetch_attempts: NonZeroU32,
-
-    /// Delay between attempts to fetch [`TracingInfo`]
-    /// in [`Session::get_tracing_info`]. Tracing info
-    /// might not be available immediately on queried node - that's why
-    /// the driver performs a few attempts with sleeps in between.
-    pub tracing_info_fetch_interval: Duration,
-
-    /// Consistency level of fetching [`TracingInfo`]
-    /// in [`Session::get_tracing_info`].
-    pub tracing_info_fetch_consistency: Consistency,
 }
 
 /// Describes database server known on Session startup.
@@ -317,9 +297,6 @@ impl SessionConfig {
             #[cfg(feature = "cloud")]
             cloud_config: None,
             enable_write_coalescing: true,
-            tracing_info_fetch_attempts: NonZeroU32::new(5).unwrap(),
-            tracing_info_fetch_interval: Duration::from_millis(3),
-            tracing_info_fetch_consistency: Consistency::One,
         }
     }
 
@@ -570,9 +547,6 @@ impl Session {
             refresh_metadata_on_auto_schema_agreement: config
                 .refresh_metadata_on_auto_schema_agreement,
             keyspace_name: ArcSwapOption::default(), // will be set by use_keyspace
-            tracing_info_fetch_attempts: config.tracing_info_fetch_attempts,
-            tracing_info_fetch_interval: config.tracing_info_fetch_interval,
-            tracing_info_fetch_consistency: config.tracing_info_fetch_consistency,
         };
 
         if let Some(keyspace_name) = config.used_keyspace {
@@ -653,12 +627,30 @@ impl Session {
         let query: Query = query.into();
         let serialized_values = values.serialized()?;
 
+        let execution_profile = query
+            .get_execution_profile_handle()
+            .unwrap_or_else(|| self.get_default_execution_profile_handle())
+            .access();
+
+        let statement_info = RoutingInfo {
+            consistency: query
+                .config
+                .consistency
+                .unwrap_or(execution_profile.consistency),
+            serial_consistency: query
+                .config
+                .serial_consistency
+                .unwrap_or(execution_profile.serial_consistency),
+            ..Default::default()
+        };
+
         let span = RequestSpan::new_query(&query.contents, serialized_values.size());
         let run_query_result = self
             .run_query(
-                RoutingInfo::default(),
+                statement_info,
                 &query.config,
                 query.get_retry_policy().map(|rp| &**rp),
+                execution_profile,
                 |node: Arc<Node>| async move { node.random_connection().await },
                 |connection: Arc<Connection>,
                  consistency: Consistency,
@@ -971,11 +963,20 @@ impl Session {
             .as_ref()
             .map(|pk| prepared.get_partitioner_name().hash(pk));
 
+        let execution_profile = prepared
+            .get_execution_profile_handle()
+            .unwrap_or_else(|| self.get_default_execution_profile_handle())
+            .access();
+
         let statement_info = RoutingInfo {
             consistency: prepared
-                .get_consistency()
-                .unwrap_or(self.default_execution_profile_handle.access().consistency),
-            serial_consistency: prepared.get_serial_consistency(),
+                .config
+                .consistency
+                .unwrap_or(execution_profile.consistency),
+            serial_consistency: prepared
+                .config
+                .serial_consistency
+                .unwrap_or(execution_profile.serial_consistency),
             token,
             keyspace: prepared.get_keyspace_name(),
             is_confirmed_lwt: prepared.is_confirmed_lwt(),
@@ -1003,6 +1004,7 @@ impl Session {
                 statement_info,
                 &prepared.config,
                 prepared.get_retry_policy().map(|rp| &**rp),
+                execution_profile,
                 |node: Arc<Node>| async move {
                     match token {
                         Some(token) => node.connection_for_token(token).await,
@@ -1173,19 +1175,37 @@ impl Session {
         // Extract first serialized_value
         let first_serialized_value = values.batch_values_iter().next_serialized().transpose()?;
         let first_serialized_value = first_serialized_value.as_deref();
+
+        let execution_profile = batch
+            .get_execution_profile_handle()
+            .unwrap_or_else(|| self.get_default_execution_profile_handle())
+            .access();
+
+        let consistency = batch
+            .config
+            .consistency
+            .unwrap_or(execution_profile.consistency);
+
+        let serial_consistency = batch
+            .config
+            .serial_consistency
+            .unwrap_or(execution_profile.serial_consistency);
+
         let statement_info = match (first_serialized_value, batch.statements.first()) {
             (Some(first_serialized_value), Some(BatchStatement::PreparedStatement(ps))) => {
                 RoutingInfo {
-                    consistency: batch
-                        .get_consistency()
-                        .unwrap_or(self.default_execution_profile_handle.access().consistency),
-                    serial_consistency: batch.get_serial_consistency(),
+                    consistency,
+                    serial_consistency,
                     token: self.calculate_token(ps, first_serialized_value)?,
                     keyspace: ps.get_keyspace_name(),
                     is_confirmed_lwt: false,
                 }
             }
-            _ => RoutingInfo::default(),
+            _ => RoutingInfo {
+                consistency,
+                serial_consistency,
+                ..Default::default()
+            },
         };
         let first_value_token = statement_info.token;
 
@@ -1201,6 +1221,7 @@ impl Session {
                 statement_info,
                 &batch.config,
                 batch.get_retry_policy().map(|rp| &**rp),
+                execution_profile,
                 |node: Arc<Node>| async move {
                     match first_value_token {
                         Some(first_value_token) => {
@@ -1370,24 +1391,8 @@ impl Session {
     /// See [the book](https://rust-driver.docs.scylladb.com/stable/tracing/tracing.html)
     /// for more information about query tracing
     pub async fn get_tracing_info(&self, tracing_id: &Uuid) -> Result<TracingInfo, QueryError> {
-        // tracing_info_fetch_attempts is NonZeroU32 so at least one attempt will be made
-        for _ in 0..self.tracing_info_fetch_attempts.get() {
-            let current_try: Option<TracingInfo> = self
-                .try_getting_tracing_info(tracing_id, Some(self.tracing_info_fetch_consistency))
-                .await?;
-
-            match current_try {
-                Some(tracing_info) => return Ok(tracing_info),
-                None => tokio::time::sleep(self.tracing_info_fetch_interval).await,
-            };
-        }
-
-        Err(QueryError::ProtocolError(
-            "All tracing queries returned an empty result, \
-            maybe the trace information didn't propagate yet. \
-            Consider configuring Session with \
-            a longer fetch interval (tracing_info_fetch_interval)",
-        ))
+        self.get_tracing_info_custom(tracing_id, &GetTracingConfig::default())
+            .await
     }
 
     /// Gets the name of the keyspace that is currently set, or `None` if no
@@ -1405,6 +1410,35 @@ impl Session {
         self.keyspace_name.load_full()
     }
 
+    /// Queries tracing info with custom retry settings.\
+    /// Tracing info might not be available immediately on queried node -
+    /// that's why the driver performs a few attempts with sleeps in between.
+    /// [`GetTracingConfig`] allows to specify a custom querying strategy.
+    pub async fn get_tracing_info_custom(
+        &self,
+        tracing_id: &Uuid,
+        config: &GetTracingConfig,
+    ) -> Result<TracingInfo, QueryError> {
+        // config.attempts is NonZeroU32 so at least one attempt will be made
+        for _ in 0..config.attempts.get() {
+            let current_try: Option<TracingInfo> = self
+                .try_getting_tracing_info(tracing_id, Some(config.consistency))
+                .await?;
+
+            match current_try {
+                Some(tracing_info) => return Ok(tracing_info),
+                None => tokio::time::sleep(config.interval).await,
+            };
+        }
+
+        Err(QueryError::ProtocolError(
+            "All tracing queries returned an empty result, \
+            maybe information didn't reach this node yet. \
+            Consider using get_tracing_info_custom with \
+            bigger interval in GetTracingConfig",
+        ))
+    }
+
     // Tries getting the tracing info
     // If the queries return 0 rows then returns None - the information didn't reach this node yet
     // If there is some other error returns this error
@@ -1471,6 +1505,20 @@ impl Session {
         Ok(Some(tracing_info))
     }
 
+    // Returns which replicas are likely to take part in handling the query.
+    // If a list of replicas cannot be easily narrowed down, all available replicas
+    // will be returned.
+    pub fn estimate_replicas_for_query(&self, statement: &RoutingInfo) -> Vec<Arc<Node>> {
+        let cluster_data = self.cluster.get_data();
+        match statement.token {
+            Some(token) => {
+                let cluster_data = self.cluster.get_data();
+                cluster_data.get_token_endpoints(statement.keyspace.unwrap_or(""), token)
+            }
+            None => cluster_data.get_nodes_info().to_owned(),
+        }
+    }
+
     // This method allows to easily run a query using load balancing, retry policy etc.
     // Requires some information about the query and two closures
     // First closure is used to choose a connection
@@ -1483,11 +1531,13 @@ impl Session {
     // On success this query's result is returned
     // I tried to make this closures take a reference instead of an Arc but failed
     // maybe once async closures get stabilized this can be fixed
+    #[allow(clippy::too_many_arguments)] // <-- remove this once retry policy is put into StatementConfig
     async fn run_query<'a, ConnFut, QueryFut, ResT>(
         &'a self,
         statement_info: RoutingInfo<'a>,
         statement_config: &'a StatementConfig,
         statement_retry_policy: Option<&dyn RetryPolicy>,
+        execution_profile: Arc<ExecutionProfileInner>,
         choose_connection: impl Fn(Arc<Node>) -> ConnFut,
         do_query: impl Fn(Arc<Connection>, Consistency, &ExecutionProfileInner) -> QueryFut,
         request_span: &'a RequestSpan,
@@ -1503,12 +1553,6 @@ impl Session {
                 .as_ref()
                 .map(|hl| (&**hl, hl.log_query_start()));
 
-        let execution_profile = statement_config
-            .execution_profile_handle
-            .as_ref()
-            .unwrap_or_else(|| self.get_default_execution_profile_handle())
-            .access();
-
         let load_balancer = &execution_profile.load_balancing_policy;
 
         let runner = async {
@@ -1575,7 +1619,7 @@ impl Session {
                             &execution_profile,
                             ExecuteQueryContext {
                                 is_idempotent: statement_config.is_idempotent,
-                                consistency: statement_config.consistency,
+                                consistency_set_on_statement: statement_config.consistency,
                                 retry_session: retry_policy.new_session(),
                                 history_data,
                                 query_info: &statement_info,
@@ -1611,7 +1655,7 @@ impl Session {
                         &execution_profile,
                         ExecuteQueryContext {
                             is_idempotent: statement_config.is_idempotent,
-                            consistency: statement_config.consistency,
+                            consistency_set_on_statement: statement_config.consistency,
                             retry_session: retry_policy.new_session(),
                             history_data,
                             query_info: &statement_info,
@@ -1666,8 +1710,9 @@ impl Session {
         ResT: AllowedRunQueryResTType,
     {
         let mut last_error: Option<QueryError> = None;
-        let mut current_consistency: Consistency =
-            context.consistency.unwrap_or(execution_profile.consistency);
+        let mut current_consistency: Consistency = context
+            .consistency_set_on_statement
+            .unwrap_or(execution_profile.consistency);
 
         'nodes_in_plan: for node in query_plan {
             let span = trace_span!("Executing query", node = %node.address);
@@ -1742,7 +1787,9 @@ impl Session {
                     error: the_error,
                     is_idempotent: context.is_idempotent,
                     consistency: LegacyConsistency::Regular(
-                        context.consistency.unwrap_or(execution_profile.consistency),
+                        context
+                            .consistency_set_on_statement
+                            .unwrap_or(execution_profile.consistency),
                     ),
                 };
 
@@ -1813,6 +1860,7 @@ impl Session {
                 info,
                 &config,
                 None, // No specific retry policy needed for schema agreement
+                self.get_default_execution_profile_handle().access(),
                 |node: Arc<Node>| async move { node.random_connection().await },
                 do_query,
                 &span,
@@ -1983,7 +2031,7 @@ impl AllowedRunQueryResTType for NonErrorQueryResponse {}
 
 struct ExecuteQueryContext<'a> {
     is_idempotent: bool,
-    consistency: Option<Consistency>,
+    consistency_set_on_statement: Option<Consistency>,
     retry_session: Box<dyn RetrySession>,
     history_data: Option<HistoryData<'a>>,
     query_info: &'a load_balancing::RoutingInfo<'a>,
diff --git a/scylla/src/transport/session_builder.rs b/scylla/src/transport/session_builder.rs
index b09dd932..aa88ce5f 100644
--- a/scylla/src/transport/session_builder.rs
+++ b/scylla/src/transport/session_builder.rs
@@ -10,13 +10,11 @@ use crate::cloud::{CloudConfig, CloudConfigError};
 #[cfg(feature = "cloud")]
 use crate::ExecutionProfile;
 
-use crate::statement::Consistency;
 use crate::transport::connection_pool::PoolSize;
 use crate::transport::host_filter::HostFilter;
 use std::borrow::Borrow;
 use std::marker::PhantomData;
 use std::net::SocketAddr;
-use std::num::NonZeroU32;
 #[cfg(feature = "cloud")]
 use std::path::Path;
 use std::sync::Arc;
@@ -830,77 +828,6 @@ impl<K: SessionBuilderKind> GenericSessionBuilder<K> {
         self.config.refresh_metadata_on_auto_schema_agreement = refresh_metadata;
         self
     }
-
-    /// Set the number of attempts to fetch [TracingInfo](crate::tracing::TracingInfo)
-    /// in [`Session::get_tracing_info`].
-    /// The default is 5 attempts.
-    ///
-    /// Tracing info might not be available immediately on queried node - that's why
-    /// the driver performs a few attempts with sleeps in between.
-    ///
-    /// # Example
-    /// ```
-    /// # use scylla::{Session, SessionBuilder};
-    /// # use std::num::NonZeroU32;
-    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
-    /// let session: Session = SessionBuilder::new()
-    ///     .known_node("127.0.0.1:9042")
-    ///     .tracing_info_fetch_attempts(NonZeroU32::new(10).unwrap())
-    ///     .build()
-    ///     .await?;
-    /// # Ok(())
-    /// # }
-    /// ```
-    pub fn tracing_info_fetch_attempts(mut self, attempts: NonZeroU32) -> Self {
-        self.config.tracing_info_fetch_attempts = attempts;
-        self
-    }
-
-    /// Set the delay between attempts to fetch [TracingInfo](crate::tracing::TracingInfo)
-    /// in [`Session::get_tracing_info`].
-    /// The default is 3 milliseconds.
-    ///
-    /// Tracing info might not be available immediately on queried node - that's why
-    /// the driver performs a few attempts with sleeps in between.
-    ///
-    /// # Example
-    /// ```
-    /// # use scylla::{Session, SessionBuilder};
-    /// # use std::time::Duration;
-    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
-    /// let session: Session = SessionBuilder::new()
-    ///     .known_node("127.0.0.1:9042")
-    ///     .tracing_info_fetch_interval(Duration::from_millis(50))
-    ///     .build()
-    ///     .await?;
-    /// # Ok(())
-    /// # }
-    /// ```
-    pub fn tracing_info_fetch_interval(mut self, interval: Duration) -> Self {
-        self.config.tracing_info_fetch_interval = interval;
-        self
-    }
-
-    /// Set the consistency level of fetching [TracingInfo](crate::tracing::TracingInfo)
-    /// in [`Session::get_tracing_info`].
-    /// The default is [`Consistency::One`].
-    ///
-    /// # Example
-    /// ```
-    /// # use scylla::{Session, SessionBuilder, statement::Consistency};
-    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
-    /// let session: Session = SessionBuilder::new()
-    ///     .known_node("127.0.0.1:9042")
-    ///     .tracing_info_fetch_consistency(Consistency::One)
-    ///     .build()
-    ///     .await?;
-    /// # Ok(())
-    /// # }
-    /// ```
-    pub fn tracing_info_fetch_consistency(mut self, consistency: Consistency) -> Self {
-        self.config.tracing_info_fetch_consistency = consistency;
-        self
-    }
 }
 
 /// Creates a [`SessionBuilder`] with default configuration, same as [`SessionBuilder::new`]


Test patch:
diff --git a/scylla/src/transport/session_test.rs b/scylla/src/transport/session_test.rs
index 35e2f7bf..994e0e34 100644
--- a/scylla/src/transport/session_test.rs
+++ b/scylla/src/transport/session_test.rs
@@ -7,7 +7,7 @@ use crate::query::Query;
 use crate::retry_policy::{QueryInfo, RetryDecision, RetryPolicy, RetrySession};
 use crate::routing::Token;
 use crate::statement::Consistency;
-use crate::tracing::TracingInfo;
+use crate::tracing::{GetTracingConfig, TracingInfo};
 use crate::transport::cluster::Datacenter;
 use crate::transport::errors::{BadKeyspaceName, BadQuery, DbError, QueryError};
 use crate::transport::partitioner::{Murmur3Partitioner, Partitioner, PartitionerName};
@@ -29,8 +29,10 @@ use itertools::Itertools;
 use scylla_cql::frame::value::Value;
 use std::collections::BTreeSet;
 use std::collections::{BTreeMap, HashMap};
+use std::num::NonZeroU32;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::Arc;
+use std::time::Duration;
 use tokio::net::TcpListener;
 use uuid::Uuid;
 
@@ -561,6 +563,14 @@ async fn test_token_awareness() {
         .unwrap();
     prepared_statement.set_tracing(true);
 
+    // The Cassandra cluster in CI is quite slow, so we use a quite forgiving
+    // configuration here (wait at most 10 seconds)
+    let get_tracing_config = GetTracingConfig {
+        attempts: NonZeroU32::new(50).unwrap(),
+        interval: Duration::from_millis(200),
+        consistency: Consistency::One,
+    };
+
     // The default policy should be token aware
     for size in 1..50usize {
         let key = vec!['a'; size].into_iter().collect::<String>();
@@ -569,7 +579,7 @@ async fn test_token_awareness() {
         // Execute a query and observe tracing info
         let res = session.execute(&prepared_statement, values).await.unwrap();
         let tracing_info = session
-            .get_tracing_info(res.tracing_id.as_ref().unwrap())
+            .get_tracing_info_custom(res.tracing_id.as_ref().unwrap(), &get_tracing_config)
             .await
             .unwrap();
 
@@ -582,7 +592,10 @@ async fn test_token_awareness() {
             .await
             .unwrap();
         let tracing_id = iter.get_tracing_ids()[0];
-        let tracing_info = session.get_tracing_info(&tracing_id).await.unwrap();
+        let tracing_info = session
+            .get_tracing_info_custom(&tracing_id, &get_tracing_config)
+            .await
+            .unwrap();
 
         // Again, verify that only one node was involved
         assert_eq!(tracing_info.nodes().len(), 1);
@@ -973,8 +986,21 @@ async fn test_get_tracing_info(session: &Session, ks: String) {
     let traced_query_result: QueryResult = session.query(traced_query, &[]).await.unwrap();
     let tracing_id: Uuid = traced_query_result.tracing_id.unwrap();
 
+    // The reason why we enable so long waiting for TracingInfo is... Cassandra. (Yes, again.)
+    // In Cassandra Java Driver, the wait time for tracing info is 10 seconds, so here we do the same.
+    // However, as Scylla usually gets TracingInfo ready really fast (our default interval is hence 3ms),
+    // we stick to a not-so-much-terribly-long interval here.
+    let get_tracing_config = GetTracingConfig {
+        attempts: NonZeroU32::new(50).unwrap(),
+        interval: Duration::from_millis(200),
+        consistency: Consistency::One,
+    };
+
     // Getting tracing info from session using this uuid works
-    let tracing_info: TracingInfo = session.get_tracing_info(&tracing_id).await.unwrap();
+    let tracing_info: TracingInfo = session
+        .get_tracing_info_custom(&tracing_id, &get_tracing_config)
+        .await
+        .unwrap();
     assert!(!tracing_info.events.is_empty());
     assert!(!tracing_info.nodes().is_empty());
 }
diff --git a/scylla/src/utils/test_utils.rs b/scylla/src/utils/test_utils.rs
index 7d3f079c..aea5ed27 100644
--- a/scylla/src/utils/test_utils.rs
+++ b/scylla/src/utils/test_utils.rs
@@ -1,14 +1,13 @@
-#[cfg(test)]
-use crate::transport::session_builder::{GenericSessionBuilder, SessionBuilderKind};
-#[cfg(test)]
-use crate::Session;
-#[cfg(test)]
-use std::{num::NonZeroU32, time::Duration};
 use std::{
     sync::atomic::{AtomicUsize, Ordering},
     time::{SystemTime, UNIX_EPOCH},
 };
 
+#[cfg(test)]
+use crate::transport::session_builder::{GenericSessionBuilder, SessionBuilderKind};
+#[cfg(test)]
+use crate::Session;
+
 static UNIQUE_COUNTER: AtomicUsize = AtomicUsize::new(0);
 
 pub fn unique_keyspace_name() -> String {
@@ -61,34 +60,27 @@ pub(crate) async fn supports_feature(session: &Session, feature: &str) -> bool {
 // connect to localhost.
 #[cfg(test)]
 pub fn create_new_session_builder() -> GenericSessionBuilder<impl SessionBuilderKind> {
-    let session_builder = {
-        #[cfg(not(scylla_cloud_tests))]
-        {
-            use crate::SessionBuilder;
-
-            let uri = std::env::var("SCYLLA_URI").unwrap_or_else(|_| "127.0.0.1:9042".to_string());
-
-            SessionBuilder::new().known_node(uri)
-        }
+    #[cfg(not(scylla_cloud_tests))]
+    {
+        use crate::transport::session_builder::DefaultMode;
+        use crate::SessionBuilder;
 
-        #[cfg(scylla_cloud_tests)]
-        {
-            use crate::transport::session_builder::CloudMode;
-            use crate::CloudSessionBuilder;
-            use std::path::Path;
+        let uri = std::env::var("SCYLLA_URI").unwrap_or_else(|_| "127.0.0.1:9042".to_string());
+        let session_builder: GenericSessionBuilder<DefaultMode> =
+            SessionBuilder::new().known_node(uri);
+        session_builder
+    }
 
-            std::env::var("CLOUD_CONFIG_PATH")
-                .map(|config_path| CloudSessionBuilder::new(Path::new(&config_path)))
-                .expect("Failed to initialize CloudSessionBuilder")
-                .expect("CLOUD_CONFIG_PATH environment variable is missing")
-        }
-    };
+    #[cfg(scylla_cloud_tests)]
+    {
+        use crate::transport::session_builder::CloudMode;
+        use crate::CloudSessionBuilder;
+        use std::path::Path;
 
-    // The reason why we enable so long waiting for TracingInfo is... Cassandra. (Yes, again.)
-    // In Cassandra Java Driver, the wait time for tracing info is 10 seconds, so here we do the same.
-    // However, as Scylla usually gets TracingInfo ready really fast (our default interval is hence 3ms),
-    // we stick to a not-so-much-terribly-long interval here.
-    session_builder
-        .tracing_info_fetch_attempts(NonZeroU32::new(50).unwrap())
-        .tracing_info_fetch_interval(Duration::from_millis(200))
+        let session_builder: GenericSessionBuilder<CloudMode> = std::env::var("CLOUD_CONFIG_PATH")
+            .map(|config_path| CloudSessionBuilder::new(Path::new(&config_path)))
+            .unwrap()
+            .expect("failed to create a session in cloud mode");
+        session_builder
+    }
 }
diff --git a/scylla/tests/integration/consistency.rs b/scylla/tests/integration/consistency.rs
new file mode 100644
index 00000000..41c32c73
--- /dev/null
+++ b/scylla/tests/integration/consistency.rs
@@ -0,0 +1,454 @@
+use crate::utils::test_with_3_node_cluster;
+
+use scylla::execution_profile::{ExecutionProfileBuilder, ExecutionProfileHandle};
+use scylla::load_balancing::{DefaultPolicy, LoadBalancingPolicy, RoutingInfo};
+use scylla::prepared_statement::PreparedStatement;
+use scylla::retry_policy::FallthroughRetryPolicy;
+use scylla::routing::Token;
+use scylla::test_utils::unique_keyspace_name;
+use scylla::transport::session::Session;
+use tokio::sync::mpsc::{self, UnboundedReceiver, UnboundedSender};
+
+use scylla::statement::batch::BatchStatement;
+use scylla::statement::query::Query;
+use scylla::{
+    batch::{Batch, BatchType},
+    statement::SerialConsistency,
+};
+use scylla::{ExecutionProfile, SessionBuilder};
+use scylla_cql::Consistency;
+use scylla_proxy::ShardAwareness;
+use scylla_proxy::{
+    Condition, ProxyError, Reaction, RequestFrame, RequestOpcode, RequestReaction, RequestRule,
+    WorkerError,
+};
+use std::sync::Arc;
+use std::time::Duration;
+
+fn consistencies() -> impl Iterator<Item = Consistency> {
+    [
+        Consistency::All,
+        Consistency::Any,
+        Consistency::EachQuorum,
+        Consistency::LocalOne,
+        Consistency::LocalQuorum,
+        Consistency::One,
+        Consistency::Quorum,
+        Consistency::Three,
+        Consistency::Two,
+    ]
+    .into_iter()
+}
+
+fn serial_consistencies() -> impl Iterator<Item = SerialConsistency> {
+    [SerialConsistency::Serial, SerialConsistency::LocalSerial].into_iter()
+}
+
+// Every consistency and every serial consistency is yielded at least once.
+// These are NOT all combinations of those.
+fn pairs_of_all_consistencies() -> impl Iterator<Item = (Consistency, Option<SerialConsistency>)> {
+    let consistencies = consistencies();
+    let serial_consistencies = serial_consistencies()
+        .map(Some)
+        .chain(std::iter::repeat(None));
+    consistencies.zip(serial_consistencies)
+}
+
+const CREATE_TABLE_STR: &str = "CREATE TABLE consistency_tests (a int, b int, PRIMARY KEY (a, b))";
+const QUERY_STR: &str = "INSERT INTO consistency_tests (a, b) VALUES (?, 1)";
+
+async fn create_schema(session: &Session, ks: &str) {
+    session.query(format!("CREATE KEYSPACE IF NOT EXISTS {} WITH REPLICATION = {{'class' : 'NetworkTopologyStrategy', 'replication_factor' : 3}}", ks), &[]).await.unwrap();
+    session.use_keyspace(ks, false).await.unwrap();
+
+    session.query(CREATE_TABLE_STR, &[]).await.unwrap();
+}
+
+// The following functions perform a request with consistencies set directly on a statement.
+async fn query_consistency_set_directly(
+    session: &Session,
+    query: &Query,
+    c: Consistency,
+    sc: Option<SerialConsistency>,
+) {
+    let mut query = query.clone();
+    query.set_consistency(c);
+    query.set_serial_consistency(sc);
+    session.query(query.clone(), (1,)).await.unwrap();
+    session.query_iter(query, (1,)).await.unwrap();
+}
+
+async fn execute_consistency_set_directly(
+    session: &Session,
+    prepared: &PreparedStatement,
+    c: Consistency,
+    sc: Option<SerialConsistency>,
+) {
+    let mut prepared = prepared.clone();
+    prepared.set_consistency(c);
+    prepared.set_serial_consistency(sc);
+    session.execute(&prepared, (1,)).await.unwrap();
+    session.execute_iter(prepared, (1,)).await.unwrap();
+}
+
+async fn batch_consistency_set_directly(
+    session: &Session,
+    batch: &Batch,
+    c: Consistency,
+    sc: Option<SerialConsistency>,
+) {
+    let mut batch = batch.clone();
+    batch.set_consistency(c);
+    batch.set_serial_consistency(sc);
+    session.batch(&batch, ((1,),)).await.unwrap();
+}
+
+// The following functions perform a request with consistencies set on a per-statement execution profile.
+async fn query_consistency_set_on_exec_profile(
+    session: &Session,
+    query: &Query,
+    profile: ExecutionProfileHandle,
+) {
+    let mut query = query.clone();
+    query.set_execution_profile_handle(Some(profile));
+    session.query(query.clone(), (1,)).await.unwrap();
+    session.query_iter(query, (1,)).await.unwrap();
+}
+
+async fn execute_consistency_set_on_exec_profile(
+    session: &Session,
+    prepared: &PreparedStatement,
+    profile: ExecutionProfileHandle,
+) {
+    let mut prepared = prepared.clone();
+    prepared.set_execution_profile_handle(Some(profile));
+    session.execute(&prepared, (1,)).await.unwrap();
+    session.execute_iter(prepared, (1,)).await.unwrap();
+}
+
+async fn batch_consistency_set_on_exec_profile(
+    session: &Session,
+    batch: &Batch,
+    profile: ExecutionProfileHandle,
+) {
+    let mut batch = batch.clone();
+    batch.set_execution_profile_handle(Some(profile));
+    session.batch(&batch, ((1,),)).await.unwrap();
+}
+
+// For all consistencies (as defined by `pairs_of_all_consistencies()`) and every method of setting consistencies
+// (directly on statement, on per-statement exec profile, on default per-session exec profile)
+// performs a request and calls `check_consistencies()` asserting function with `rx`, which is a generic way
+// of input for assertions.
+// `check_consistencies()` does not simply use &mut Rx, because then we enter the atrocious world of higher-order lifetimes
+// whose validity the compiler can not yet prove.
+async fn check_for_all_consistencies_and_setting_options<
+    Fut1: std::future::Future<Output = Rx>,
+    Fut2: std::future::Future<Output = Rx>,
+    Rx,
+>(
+    session_builder: SessionBuilder,
+    base_for_every_profile: ExecutionProfileBuilder,
+    ks: &str,
+    mut rx: Rx,
+    after_session_init: impl Fn(Rx) -> Fut1,
+    check_consistencies: impl Fn(Consistency, Option<SerialConsistency>, Rx) -> Fut2,
+) {
+    let session = session_builder
+        .clone()
+        .default_execution_profile_handle(base_for_every_profile.clone().build().into_handle())
+        .build()
+        .await
+        .unwrap();
+    create_schema(&session, ks).await;
+    rx = after_session_init(rx).await;
+
+    // We will be using these requests:
+    let query = Query::from(QUERY_STR);
+    let prepared = session.prepare(QUERY_STR).await.unwrap();
+    let batch = Batch::new_with_statements(
+        BatchType::Logged,
+        vec![BatchStatement::Query(Query::from(QUERY_STR))],
+    );
+
+    for (consistency, serial_consistency) in pairs_of_all_consistencies() {
+        // Some checks are double, because both non-paged and paged executions are done.
+        // (queries and prepared statements are double, batches are single)
+
+        // Set directly
+        query_consistency_set_directly(&session, &query, consistency, serial_consistency).await;
+        rx = check_consistencies(consistency, serial_consistency, rx).await;
+        rx = check_consistencies(consistency, serial_consistency, rx).await;
+
+        execute_consistency_set_directly(&session, &prepared, consistency, serial_consistency)
+            .await;
+        rx = check_consistencies(consistency, serial_consistency, rx).await;
+        rx = check_consistencies(consistency, serial_consistency, rx).await;
+
+        batch_consistency_set_directly(&session, &batch, consistency, serial_consistency).await;
+        rx = check_consistencies(consistency, serial_consistency, rx).await;
+
+        // Set on an exec profile
+        let handle = base_for_every_profile
+            .clone()
+            .consistency(consistency)
+            .serial_consistency(serial_consistency)
+            .build()
+            .into_handle();
+        query_consistency_set_on_exec_profile(&session, &query, handle.clone()).await;
+        rx = check_consistencies(consistency, serial_consistency, rx).await;
+        rx = check_consistencies(consistency, serial_consistency, rx).await;
+
+        execute_consistency_set_on_exec_profile(&session, &prepared, handle.clone()).await;
+        rx = check_consistencies(consistency, serial_consistency, rx).await;
+        rx = check_consistencies(consistency, serial_consistency, rx).await;
+
+        batch_consistency_set_on_exec_profile(&session, &batch, handle.clone()).await;
+        rx = check_consistencies(consistency, serial_consistency, rx).await;
+
+        // Set on session's default exec profile
+        let session_with_consistencies = session_builder
+            .clone()
+            .default_execution_profile_handle(handle)
+            .build()
+            .await
+            .unwrap();
+        session_with_consistencies
+            .use_keyspace(ks, true)
+            .await
+            .unwrap();
+        rx = after_session_init(rx).await;
+
+        session_with_consistencies
+            .query(QUERY_STR, (1,))
+            .await
+            .unwrap();
+        rx = check_consistencies(consistency, serial_consistency, rx).await;
+        session_with_consistencies
+            .query_iter(QUERY_STR, (1,))
+            .await
+            .unwrap();
+        rx = check_consistencies(consistency, serial_consistency, rx).await;
+
+        session_with_consistencies
+            .execute(&prepared, (1,))
+            .await
+            .unwrap();
+        rx = check_consistencies(consistency, serial_consistency, rx).await;
+        session_with_consistencies
+            .execute_iter(prepared.clone(), (1,))
+            .await
+            .unwrap();
+        rx = check_consistencies(consistency, serial_consistency, rx).await;
+
+        session_with_consistencies
+            .batch(&batch, ((1,),))
+            .await
+            .unwrap();
+        rx = check_consistencies(consistency, serial_consistency, rx).await;
+    }
+}
+
+// Checks that the expected consistency and serial_consistency are set
+//  in the CQL request frame.
+#[tokio::test]
+#[ntest::timeout(60000)]
+#[cfg(not(scylla_cloud_tests))]
+async fn consistency_is_correctly_set_in_cql_requests() {
+    let res = test_with_3_node_cluster(
+        ShardAwareness::QueryNode,
+        |proxy_uris, translation_map, mut running_proxy| async move {
+            let request_rule = |tx| {
+                RequestRule(
+                    Condition::or(
+                        Condition::RequestOpcode(RequestOpcode::Execute),
+                        Condition::or(
+                            Condition::RequestOpcode(RequestOpcode::Batch),
+                            Condition::and(
+                                Condition::RequestOpcode(RequestOpcode::Query),
+                                Condition::BodyContainsCaseSensitive(Box::new(
+                                    *b"INTO consistency_tests",
+                                )),
+                            ),
+                        ),
+                    ),
+                    RequestReaction::noop().with_feedback_when_performed(tx),
+                )
+            };
+
+            let (request_tx, request_rx) = mpsc::unbounded_channel();
+            for running_node in running_proxy.running_nodes.iter_mut() {
+                running_node.change_request_rules(Some(vec![request_rule(request_tx.clone())]));
+            }
+
+            let fallthrough_exec_profile_builder =
+                ExecutionProfile::builder().retry_policy(Box::new(FallthroughRetryPolicy));
+
+            let translation_map = Arc::new(translation_map);
+
+            // DB preparation phase
+            let ks = unique_keyspace_name();
+            let session_builder = SessionBuilder::new()
+                .known_node(proxy_uris[0].as_str())
+                .keepalive_interval(Duration::from_secs(10000))
+                .address_translator(translation_map.clone());
+
+            async fn check_consistencies(
+                consistency: Consistency,
+                serial_consistency: Option<SerialConsistency>,
+                mut request_rx: UnboundedReceiver<RequestFrame>,
+            ) -> UnboundedReceiver<RequestFrame> {
+                let request_frame = request_rx.recv().await.unwrap();
+                let deserialized_request = request_frame.deserialize().unwrap();
+                assert_eq!(deserialized_request.get_consistency().unwrap(), consistency);
+                assert_eq!(
+                    deserialized_request.get_serial_consistency().unwrap(),
+                    serial_consistency
+                );
+                request_rx
+            }
+
+            check_for_all_consistencies_and_setting_options(
+                session_builder,
+                fallthrough_exec_profile_builder,
+                &ks,
+                request_rx,
+                |rx| async move { rx },
+                check_consistencies,
+            )
+            .await;
+
+            running_proxy
+        },
+    )
+    .await;
+
+    match res {
+        Ok(()) => (),
+        Err(ProxyError::Worker(WorkerError::DriverDisconnected(_))) => (),
+        Err(err) => panic!("{}", err),
+    }
+}
+
+#[derive(Debug, Clone)]
+pub(crate) struct OwnedRoutingInfo {
+    pub consistency: Consistency,
+    pub serial_consistency: Option<SerialConsistency>,
+
+    #[allow(unused)]
+    pub keyspace: Option<String>,
+    #[allow(unused)]
+    pub token: Option<Token>,
+    #[allow(unused)]
+    pub is_confirmed_lwt: bool,
+}
+
+impl OwnedRoutingInfo {
+    fn from(info: RoutingInfo) -> Self {
+        let RoutingInfo {
+            consistency,
+            serial_consistency,
+            token,
+            keyspace,
+            is_confirmed_lwt,
+        } = info;
+        Self {
+            consistency,
+            serial_consistency,
+            token,
+            keyspace: keyspace.map(ToOwned::to_owned),
+            is_confirmed_lwt,
+        }
+    }
+}
+
+// Wraps over DefaultPolicy to provide access to RoutingInfo fed into the load balancer.
+#[derive(Debug)]
+struct RoutingInfoReportingWrapper {
+    wrapped: Arc<dyn LoadBalancingPolicy>,
+    routing_info_tx: UnboundedSender<OwnedRoutingInfo>,
+}
+
+impl LoadBalancingPolicy for RoutingInfoReportingWrapper {
+    fn pick<'a>(
+        &'a self,
+        query: &'a RoutingInfo,
+        cluster: &'a scylla::transport::ClusterData,
+    ) -> Option<scylla::transport::NodeRef<'a>> {
+        self.routing_info_tx
+            .send(OwnedRoutingInfo::from(query.clone()))
+            .unwrap();
+        self.wrapped.pick(query, cluster)
+    }
+
+    fn fallback<'a>(
+        &'a self,
+        query: &'a RoutingInfo,
+        cluster: &'a scylla::transport::ClusterData,
+    ) -> scylla::load_balancing::FallbackPlan<'a> {
+        self.routing_info_tx
+            .send(OwnedRoutingInfo::from(query.clone()))
+            .unwrap();
+        self.wrapped.fallback(query, cluster)
+    }
+
+    fn name(&self) -> String {
+        "RoutingInfoReportingWrapper".to_owned()
+    }
+}
+
+// Checks that the expected consistency and serial_consistency are set
+// in the RoutingInfo that is exposed to the load balancer.
+#[tokio::test]
+#[ntest::timeout(60000)]
+#[cfg(not(scylla_cloud_tests))]
+async fn consistency_is_correctly_set_in_routing_info() {
+    let uri = std::env::var("SCYLLA_URI").unwrap_or_else(|_| "127.0.0.1:9042".to_string());
+    let ks = unique_keyspace_name();
+
+    let (routing_info_tx, routing_info_rx) = mpsc::unbounded_channel();
+
+    let reporting_load_balancer = RoutingInfoReportingWrapper {
+        wrapped: DefaultPolicy::builder().build(),
+        routing_info_tx,
+    };
+
+    let exec_profile_builder = ExecutionProfile::builder()
+        .retry_policy(Box::new(FallthroughRetryPolicy))
+        .load_balancing_policy(Arc::new(reporting_load_balancer));
+
+    // DB preparation phase
+    let session_builder = SessionBuilder::new()
+        .known_node(uri.as_str())
+        .keepalive_interval(Duration::from_secs(10000))
+        .default_execution_profile_handle(exec_profile_builder.clone().build().into_handle());
+
+    async fn on_session_init(
+        mut routing_info_rx: UnboundedReceiver<OwnedRoutingInfo>,
+    ) -> UnboundedReceiver<OwnedRoutingInfo> {
+        // Clear RoutingInfo got from DB preparation (keyspace & table creation)
+        while routing_info_rx.try_recv().is_ok() {}
+        routing_info_rx
+    }
+
+    async fn check_consistencies(
+        consistency: Consistency,
+        serial_consistency: Option<SerialConsistency>,
+        mut routing_info_rx: UnboundedReceiver<OwnedRoutingInfo>,
+    ) -> UnboundedReceiver<OwnedRoutingInfo> {
+        let info = routing_info_rx.recv().await.unwrap();
+        assert_eq!(info.consistency, consistency);
+        assert_eq!(info.serial_consistency, serial_consistency);
+        routing_info_rx
+    }
+
+    check_for_all_consistencies_and_setting_options(
+        session_builder,
+        exec_profile_builder,
+        &ks,
+        routing_info_rx,
+        on_session_init,
+        check_consistencies,
+    )
+    .await;
+}
diff --git a/scylla/tests/integration/main.rs b/scylla/tests/integration/main.rs
index f44f3598..c2cb8fc6 100644
--- a/scylla/tests/integration/main.rs
+++ b/scylla/tests/integration/main.rs
@@ -1,3 +1,4 @@
+mod consistency;
 mod execution_profiles;
 mod hygiene;
 mod lwt_optimisation;