Repository: scylladb/scylla-rust-driver
Pull request #228 (2021-04-25 20:14:35)

Speculative execution support
## Initial speculative execution support

* Added speculative execution to non-iter session methods
* Refactored `run_query` session method to take less parameters (adding more caused clippy to freak out)
* Refactored `statement` module, to reduce code duplication (by introducing `StatementConfig` structure)
* Removed `MetricsView` and restricted `Metrics` mutating methods visibility (this was needed to simplify metrics infrastructure used by `PercentileSpeculativeExecutionPolicy`)
* Added two speculative execution policies - Simple and Percentile
* Documented & created an example

## Checklist
<!--
    Make sure you took care of the issues on the list.
    Put 'x' into those boxes which apply.
    You can also create the PR now and click on all relevant checkboxes.
    See CONTRIBUTING.md for more details.
-->

- [x] I have split my patch into logically separate commits.
- [x] All commit messages clearly explain what they change and why.
- [ ] I added relevant tests for new features and bug fixes.
- [x] All commits compile, pass static checks and pass test.
- [x] PR description sums up the changes and reasons why they should be introduced.
- [x] I added appropriate `Fixes:` annotations to PR description.

Fixes: #213


havaker: @psarna @piodul please review
psarna: Looks fine overall, **but** c3e4299f36453ed0206f7e7bb2c57152041397a2 sneakily introduces a non-backward-compatible change in the API, namely the need to type `stmt.config.set_xxx()` instead of `stmt.set_xxx()` we had previously. In this particular case I think that duplication in the code is better than forcing the clients to use a less ergonomic API. You could also combine the two approaches:
 1. create a new abstraction layer - StatementConfig
 2. add syntactic sugar to all statement types, which uses StatementConfig underneath, but allows clients to call `.set_xxx()` instead of `.config.set_xxx()` on all statement types

That way we also maintain backward compatibility. It's not strictly required for us to do so, because we're still at release 0.1.0, but it's good practice (and as a user I personally strongly prefer the previous API). /cc @piodul

Patch:
diff --git a/docs/source/SUMMARY.md b/docs/source/SUMMARY.md
index 9cc10d3a..5e09f2dd 100644
--- a/docs/source/SUMMARY.md
+++ b/docs/source/SUMMARY.md
@@ -48,6 +48,10 @@
     - [Fallthrough retry policy](retry-policy/fallthrough.md)
     - [Default retry policy](retry-policy/default.md)
 
+- [Speculative execution](speculative-execution/speculative.md)
+    - [Simple](speculative-execution/simple.md)
+    - [Latency Percentile](speculative-execution/percentile.md)
+
 - [Driver metrics](metrics/metrics.md)
 
 - [Logging](logging/logging.md)
@@ -56,4 +60,4 @@
     - [Tracing a simple/prepared query](tracing/simple-prepared.md)
     - [Tracing a batch query](tracing/batch.md)
     - [Tracing a paged query](tracing/paged.md)
-    - [Tracing `Session::prepare`](tracing/prepare.md)
\ No newline at end of file
+    - [Tracing `Session::prepare`](tracing/prepare.md)
diff --git a/docs/source/queries/prepared.md b/docs/source/queries/prepared.md
index 66a2998b..de8423ec 100644
--- a/docs/source/queries/prepared.md
+++ b/docs/source/queries/prepared.md
@@ -112,4 +112,4 @@ session.execute(&good_prepared, (12345, 54321)).await?;
 
 # Ok(())
 # }
-```
\ No newline at end of file
+```
diff --git a/docs/source/retry-policy/default.md b/docs/source/retry-policy/default.md
index a80684ed..c57ac88c 100644
--- a/docs/source/retry-policy/default.md
+++ b/docs/source/retry-policy/default.md
@@ -63,4 +63,4 @@ let to_insert: i32 = 12345;
 session.execute(&prepared, (to_insert,)).await?;
 # Ok(())
 # }
-```
\ No newline at end of file
+```
diff --git a/docs/source/retry-policy/fallthrough.md b/docs/source/retry-policy/fallthrough.md
index 917a4109..a965d0cb 100644
--- a/docs/source/retry-policy/fallthrough.md
+++ b/docs/source/retry-policy/fallthrough.md
@@ -62,4 +62,4 @@ let to_insert: i32 = 12345;
 session.execute(&prepared, (to_insert,)).await?;
 # Ok(())
 # }
-```
\ No newline at end of file
+```
diff --git a/docs/source/retry-policy/retry-policy.md b/docs/source/retry-policy/retry-policy.md
index 69822e78..23f19601 100644
--- a/docs/source/retry-policy/retry-policy.md
+++ b/docs/source/retry-policy/retry-policy.md
@@ -48,4 +48,4 @@ prepared.set_is_idempotent(true);
    fallthrough
    default
 
-```
\ No newline at end of file
+```
diff --git a/docs/source/speculative-execution/percentile.md b/docs/source/speculative-execution/percentile.md
new file mode 100644
index 00000000..955ab702
--- /dev/null
+++ b/docs/source/speculative-execution/percentile.md
@@ -0,0 +1,34 @@
+# Percentile speculative execution
+
+This policy has access to `Metrics` shared with session, and triggers
+speculative execution when the request to the current host is above a
+given percentile.
+
+
+### Example
+To use this policy in `Session`:
+```rust
+# extern crate scylla;
+# use scylla::Session;
+# use std::error::Error;
+# async fn check_only_compiles() -> Result<(), Box<dyn Error>> {
+use std::{sync::Arc, time::Duration};
+use scylla::{
+    Session,
+    SessionBuilder,
+    speculative_execution::PercentileSpeculativeExecutionPolicy,
+};
+
+let policy = PercentileSpeculativeExecutionPolicy  {
+    max_retry_count: 3,
+    percentile: 99.0,
+};
+
+let session: Session = SessionBuilder::new()
+    .known_node("127.0.0.1:9042")
+    .speculative_execution(Arc::new(policy))
+    .build()
+    .await?;
+# Ok(())
+# }
+```
diff --git a/docs/source/speculative-execution/simple.md b/docs/source/speculative-execution/simple.md
new file mode 100644
index 00000000..29df32ab
--- /dev/null
+++ b/docs/source/speculative-execution/simple.md
@@ -0,0 +1,34 @@
+# Simple speculative execution
+
+The easiest speculative execution policy available.  It starts another
+execution of a query after constant delay of `retry_interval` and does at most
+`max_retry_count` speculative query executions (not counting the first,
+non-speculative one).
+
+### Example
+To use this policy in `Session`:
+```rust
+# extern crate scylla;
+# use scylla::Session;
+# use std::error::Error;
+# async fn check_only_compiles() -> Result<(), Box<dyn Error>> {
+use std::{sync::Arc, time::Duration};
+use scylla::{
+    Session,
+    SessionBuilder,
+    speculative_execution::SimpleSpeculativeExecutionPolicy
+};
+
+let policy = SimpleSpeculativeExecutionPolicy {
+    max_retry_count: 3,
+    retry_interval: Duration::from_millis(100),
+};
+
+let session: Session = SessionBuilder::new()
+    .known_node("127.0.0.1:9042")
+    .speculative_execution(Arc::new(policy))
+    .build()
+    .await?;
+# Ok(())
+# }
+```
diff --git a/docs/source/speculative-execution/speculative.md b/docs/source/speculative-execution/speculative.md
new file mode 100644
index 00000000..fda2e754
--- /dev/null
+++ b/docs/source/speculative-execution/speculative.md
@@ -0,0 +1,16 @@
+# Speculative execution
+
+Speculative query execution is an optimization technique where a driver
+pre-emptively starts a second execution of a query against another node,
+before the first node has replied.
+
+There are multiple speculative execution strategies that the driver can use.
+Speculative execution can be configured for the whole whole `Session` during
+its creation.
+
+Available speculative execution strategies:
+* [Simple](simple.md)
+* [Latency Percentile](percentile.md)
+
+Speculative execution is not enabled by default, and currently only
+non-iter session methods use it.
diff --git a/docs/source/tracing/batch.md b/docs/source/tracing/batch.md
index d20e4f9a..840df069 100644
--- a/docs/source/tracing/batch.md
+++ b/docs/source/tracing/batch.md
@@ -29,4 +29,4 @@ if let Some(id) = tracing_id {
 }
 # Ok(())
 # }
-```
\ No newline at end of file
+```
diff --git a/docs/source/tracing/paged.md b/docs/source/tracing/paged.md
index d5a2fa1b..cdbcb995 100644
--- a/docs/source/tracing/paged.md
+++ b/docs/source/tracing/paged.md
@@ -78,4 +78,4 @@ for id in tracing_ids {
 }
 # Ok(())
 # }
-```
\ No newline at end of file
+```
diff --git a/docs/source/tracing/prepare.md b/docs/source/tracing/prepare.md
index de6697c3..a520b02a 100644
--- a/docs/source/tracing/prepare.md
+++ b/docs/source/tracing/prepare.md
@@ -31,4 +31,4 @@ for id in tracing_ids {
 }
 # Ok(())
 # }
-```
\ No newline at end of file
+```
diff --git a/docs/source/tracing/simple-prepared.md b/docs/source/tracing/simple-prepared.md
index 966b480e..782c98f2 100644
--- a/docs/source/tracing/simple-prepared.md
+++ b/docs/source/tracing/simple-prepared.md
@@ -61,4 +61,4 @@ if let Some(id) = tracing_id {
 }
 # Ok(())
 # }
-```
\ No newline at end of file
+```
diff --git a/examples/Cargo.toml b/examples/Cargo.toml
index a812225d..b91cb8a6 100644
--- a/examples/Cargo.toml
+++ b/examples/Cargo.toml
@@ -68,3 +68,6 @@ path = "schema_agreement.rs"
 name = "clone"
 path = "clone.rs"
 
+[[example]]
+name = "speculative-execution"
+path = "speculative-execution.rs"
diff --git a/examples/speculative-execution.rs b/examples/speculative-execution.rs
new file mode 100644
index 00000000..3260cc04
--- /dev/null
+++ b/examples/speculative-execution.rs
@@ -0,0 +1,42 @@
+use scylla::{
+    speculative_execution::PercentileSpeculativeExecutionPolicy, Session, SessionBuilder,
+};
+
+use anyhow::Result;
+use std::{env, sync::Arc};
+
+#[tokio::main]
+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 speculative = PercentileSpeculativeExecutionPolicy {
+        max_retry_count: 2,
+        percentile: 99.0,
+    };
+
+    let session: Session = SessionBuilder::new()
+        .known_node(uri)
+        .speculative_execution(Arc::new(speculative))
+        .build()
+        .await?;
+
+    session.query("CREATE KEYSPACE IF NOT EXISTS ks WITH REPLICATION = {'class' : 'SimpleStrategy', 'replication_factor' : 1}", &[]).await?;
+
+    session
+        .query(
+            "CREATE TABLE IF NOT EXISTS ks.t (a int, b int, c text, primary key (a, b))",
+            &[],
+        )
+        .await?;
+
+    let mut select_stmt = session.prepare("SELECT a, b, c FROM ks.t").await?;
+
+    // This will allow for speculative execution
+    select_stmt.set_is_idempotent(true);
+
+    // This will trigger speculative execution
+    session.execute(&select_stmt, &[]).await?;
+
+    Ok(())
+}
diff --git a/scylla/src/lib.rs b/scylla/src/lib.rs
index 1cdf3c56..2f6e4b31 100644
--- a/scylla/src/lib.rs
+++ b/scylla/src/lib.rs
@@ -111,4 +111,8 @@ pub use transport::connection::{BatchResult, QueryResult};
 pub use transport::session::{IntoTypedRows, Session, SessionConfig};
 pub use transport::session_builder::SessionBuilder;
 
-pub use transport::metrics::MetricsView;
+pub use transport::load_balancing;
+pub use transport::retry_policy;
+pub use transport::speculative_execution;
+
+pub use transport::metrics::Metrics;
diff --git a/scylla/src/statement/batch.rs b/scylla/src/statement/batch.rs
index 9d678160..9fc7e876 100644
--- a/scylla/src/statement/batch.rs
+++ b/scylla/src/statement/batch.rs
@@ -1,20 +1,19 @@
 use crate::statement::{prepared_statement::PreparedStatement, query::Query};
+use crate::transport::retry_policy::RetryPolicy;
 
 pub use super::Consistency;
+use super::StatementConfig;
 pub use crate::frame::request::batch::BatchType;
-use crate::transport::retry_policy::RetryPolicy;
 
 /// CQL batch statement.
 ///
 /// This represents a CQL batch that can be executed on a server.
+#[derive(Clone)]
 pub struct Batch {
+    pub(crate) config: StatementConfig,
+
     statements: Vec<BatchStatement>,
     batch_type: BatchType,
-    pub consistency: Consistency,
-    pub serial_consistency: Option<Consistency>,
-    pub is_idempotent: bool,
-    pub retry_policy: Option<Box<dyn RetryPolicy + Send + Sync>>,
-    pub tracing: bool,
 }
 
 impl Batch {
@@ -43,61 +42,61 @@ impl Batch {
 
     /// Sets the consistency to be used when executing this batch.
     pub fn set_consistency(&mut self, c: Consistency) {
-        self.consistency = c;
+        self.config.consistency = c;
     }
 
     /// Gets the consistency to be used when executing this batch.
     pub fn get_consistency(&self) -> Consistency {
-        self.consistency
+        self.config.consistency
     }
 
     /// Sets the serial consistency to be used when executing this batch.
     /// (Ignored unless the batch is an LWT)
     pub fn set_serial_consistency(&mut self, sc: Option<Consistency>) {
-        self.serial_consistency = sc;
+        self.config.serial_consistency = sc;
     }
 
     /// Gets the serial consistency to be used when executing this batch.
     /// (Ignored unless the batch is an LWT)
     pub fn get_serial_consistency(&self) -> Option<Consistency> {
-        self.serial_consistency
+        self.config.serial_consistency
     }
 
-    /// Sets the idempotence of this statement  
-    /// A query is idempotent if it can be applied multiple times without changing the result of the initial application  
-    /// If set to `true` we can be sure that it is idempotent  
-    /// If set to `false` it is unknown whether it is idempotent  
+    /// Sets the idempotence of this statement
+    /// A query is idempotent if it can be applied multiple times without changing the result of the initial application
+    /// If set to `true` we can be sure that it is idempotent
+    /// If set to `false` it is unknown whether it is idempotent
     /// This is used in [`RetryPolicy`] to decide if retrying a query is safe
     pub fn set_is_idempotent(&mut self, is_idempotent: bool) {
-        self.is_idempotent = is_idempotent;
+        self.config.is_idempotent = is_idempotent;
     }
 
     /// Gets the idempotence of this statement
     pub fn get_is_idempotent(&self) -> bool {
-        self.is_idempotent
+        self.config.is_idempotent
     }
 
-    /// Sets a custom [`RetryPolicy`] to be used with this statement  
+    /// Sets a custom [`RetryPolicy`] to be used with this statement
     /// By default Session's retry policy is used, this allows to use a custom retry policy
-    pub fn set_retry_policy(&mut self, retry_policy: Box<dyn RetryPolicy + Send + Sync>) {
-        self.retry_policy = Some(retry_policy);
+    pub fn set_retry_policy(&mut self, retry_policy: Box<dyn RetryPolicy>) {
+        self.config.retry_policy = Some(retry_policy);
     }
 
     /// Gets custom [`RetryPolicy`] used by this statement
-    pub fn get_retry_policy(&self) -> &Option<Box<dyn RetryPolicy + Send + Sync>> {
-        &self.retry_policy
+    pub fn get_retry_policy(&self) -> &Option<Box<dyn RetryPolicy>> {
+        &self.config.retry_policy
     }
 
-    /// Enable or disable CQL Tracing for this batch  
+    /// Enable or disable CQL Tracing for this batch
     /// If enabled session.batch() will return a BatchResult containing tracing_id
     /// which can be used to query tracing information about the execution of this query
     pub fn set_tracing(&mut self, should_trace: bool) {
-        self.tracing = should_trace;
+        self.config.tracing = should_trace;
     }
 
     /// Gets whether tracing is enabled for this batch
     pub fn get_tracing(&self) -> bool {
-        self.tracing
+        self.config.tracing
     }
 }
 
@@ -106,11 +105,7 @@ impl Default for Batch {
         Self {
             statements: Vec::new(),
             batch_type: BatchType::Logged,
-            consistency: Default::default(),
-            serial_consistency: None,
-            is_idempotent: false,
-            retry_policy: None,
-            tracing: false,
+            config: Default::default(),
         }
     }
 }
@@ -139,20 +134,3 @@ impl From<PreparedStatement> for BatchStatement {
         BatchStatement::PreparedStatement(p)
     }
 }
-
-impl Clone for Batch {
-    fn clone(&self) -> Batch {
-        Batch {
-            statements: self.statements.clone(),
-            batch_type: self.batch_type,
-            consistency: self.consistency,
-            serial_consistency: self.serial_consistency,
-            is_idempotent: self.is_idempotent,
-            retry_policy: self
-                .retry_policy
-                .as_ref()
-                .map(|policy| policy.clone_boxed()),
-            tracing: self.tracing,
-        }
-    }
-}
diff --git a/scylla/src/statement/mod.rs b/scylla/src/statement/mod.rs
index cc51e655..3f8b8b3b 100644
--- a/scylla/src/statement/mod.rs
+++ b/scylla/src/statement/mod.rs
@@ -1,5 +1,51 @@
+use std::sync::Arc;
+
+use crate::transport::retry_policy::RetryPolicy;
+use crate::transport::speculative_execution::SpeculativeExecutionPolicy;
+
 pub mod batch;
 pub mod prepared_statement;
 pub mod query;
 
 pub use crate::frame::types::Consistency;
+
+pub struct StatementConfig {
+    pub consistency: Consistency,
+    pub serial_consistency: Option<Consistency>,
+
+    pub is_idempotent: bool,
+
+    pub retry_policy: Option<Box<dyn RetryPolicy>>,
+    pub speculative_execution_policy: Option<Arc<dyn SpeculativeExecutionPolicy>>,
+
+    pub tracing: bool,
+}
+
+impl Default for StatementConfig {
+    fn default() -> Self {
+        Self {
+            consistency: Default::default(),
+            serial_consistency: None,
+            is_idempotent: false,
+            retry_policy: None,
+            speculative_execution_policy: None,
+            tracing: false,
+        }
+    }
+}
+
+impl Clone for StatementConfig {
+    fn clone(&self) -> Self {
+        Self {
+            consistency: self.consistency,
+            serial_consistency: self.serial_consistency,
+            is_idempotent: self.is_idempotent,
+            retry_policy: self
+                .retry_policy
+                .as_ref()
+                .map(|policy| policy.clone_boxed()),
+            speculative_execution_policy: self.speculative_execution_policy.clone(),
+            tracing: self.tracing,
+        }
+    }
+}
diff --git a/scylla/src/statement/prepared_statement.rs b/scylla/src/statement/prepared_statement.rs
index c2fa018d..6eae11fb 100644
--- a/scylla/src/statement/prepared_statement.rs
+++ b/scylla/src/statement/prepared_statement.rs
@@ -1,24 +1,24 @@
-use super::Consistency;
-use crate::frame::response::result::PreparedMetadata;
-use crate::frame::value::SerializedValues;
-use crate::transport::retry_policy::RetryPolicy;
 use bytes::{BufMut, Bytes, BytesMut};
 use std::convert::TryInto;
 use thiserror::Error;
 use uuid::Uuid;
 
+use super::StatementConfig;
+use crate::frame::response::result::PreparedMetadata;
+use crate::frame::types::Consistency;
+use crate::frame::value::SerializedValues;
+use crate::transport::retry_policy::RetryPolicy;
+
 /// Represents a statement prepared on the server.
+#[derive(Clone)]
 pub struct PreparedStatement {
+    pub(crate) config: StatementConfig,
+    pub prepare_tracing_ids: Vec<Uuid>,
+
     id: Bytes,
     metadata: PreparedMetadata,
     statement: String,
-    pub prepare_tracing_ids: Vec<Uuid>,
     page_size: Option<i32>,
-    pub consistency: Consistency,
-    pub serial_consistency: Option<Consistency>,
-    pub is_idempotent: bool,
-    pub retry_policy: Option<Box<dyn RetryPolicy + Send + Sync>>,
-    pub tracing: bool,
 }
 
 impl PreparedStatement {
@@ -34,11 +34,7 @@ impl PreparedStatement {
             statement,
             prepare_tracing_ids: Vec::new(),
             page_size,
-            consistency: Default::default(),
-            serial_consistency: None,
-            is_idempotent: false,
-            retry_policy: None,
-            tracing: false,
+            config: Default::default(),
         }
     }
 
@@ -66,65 +62,6 @@ impl PreparedStatement {
         self.page_size
     }
 
-    /// Sets the consistency to be used when executing this statement.
-    pub fn set_consistency(&mut self, c: Consistency) {
-        self.consistency = c;
-    }
-
-    /// Gets the consistency to be used when executing this statement.
-    pub fn get_consistency(&self) -> Consistency {
-        self.consistency
-    }
-
-    /// Sets the serial consistency to be used when executing this statement.
-    /// (Ignored unless the statement is an LWT)
-    pub fn set_serial_consistency(&mut self, sc: Option<Consistency>) {
-        self.serial_consistency = sc;
-    }
-
-    /// Gets the serial consistency to be used when executing this statement.
-    /// (Ignored unless the statement is an LWT)
-    pub fn get_serial_consistency(&self) -> Option<Consistency> {
-        self.serial_consistency
-    }
-
-    /// Sets the idempotence of this statement  
-    /// A query is idempotent if it can be applied multiple times without changing the result of the initial application  
-    /// If set to `true` we can be sure that it is idempotent  
-    /// If set to `false` it is unknown whether it is idempotent  
-    /// This is used in [`RetryPolicy`] to decide if retrying a query is safe
-    pub fn set_is_idempotent(&mut self, is_idempotent: bool) {
-        self.is_idempotent = is_idempotent;
-    }
-
-    /// Gets the idempotence of this statement
-    pub fn get_is_idempotent(&self) -> bool {
-        self.is_idempotent
-    }
-
-    /// Sets a custom [`RetryPolicy`] to be used with this statement  
-    /// By default Session's retry policy is used, this allows to use a custom retry policy
-    pub fn set_retry_policy(&mut self, retry_policy: Box<dyn RetryPolicy + Send + Sync>) {
-        self.retry_policy = Some(retry_policy);
-    }
-
-    /// Gets custom [`RetryPolicy`] used by this statement
-    pub fn get_retry_policy(&self) -> &Option<Box<dyn RetryPolicy + Send + Sync>> {
-        &self.retry_policy
-    }
-
-    /// Enable or disable CQL Tracing for this prepared statement  
-    /// If enabled session.execute() will return QueryResult containing tracing_id
-    /// which can be used to query tracing information about the execution of this statement
-    pub fn set_tracing(&mut self, should_trace: bool) {
-        self.tracing = should_trace;
-    }
-
-    /// Gets whether tracing is enabled for this prepared statement
-    pub fn get_tracing(&self) -> bool {
-        self.tracing
-    }
-
     /// Gets tracing ids of queries used to prepare this statement
     pub fn get_prepare_tracing_ids(&self) -> &[Uuid] {
         &self.prepare_tracing_ids
@@ -192,6 +129,65 @@ impl PreparedStatement {
             .first()
             .map(|col_spec| col_spec.table_spec.ks_name.as_str())
     }
+
+    /// Sets the consistency to be used when executing this batch.
+    pub fn set_consistency(&mut self, c: Consistency) {
+        self.config.consistency = c;
+    }
+
+    /// Gets the consistency to be used when executing this batch.
+    pub fn get_consistency(&self) -> Consistency {
+        self.config.consistency
+    }
+
+    /// Sets the serial consistency to be used when executing this batch.
+    /// (Ignored unless the batch is an LWT)
+    pub fn set_serial_consistency(&mut self, sc: Option<Consistency>) {
+        self.config.serial_consistency = sc;
+    }
+
+    /// Gets the serial consistency to be used when executing this batch.
+    /// (Ignored unless the batch is an LWT)
+    pub fn get_serial_consistency(&self) -> Option<Consistency> {
+        self.config.serial_consistency
+    }
+
+    /// Sets the idempotence of this statement
+    /// A query is idempotent if it can be applied multiple times without changing the result of the initial application
+    /// If set to `true` we can be sure that it is idempotent
+    /// If set to `false` it is unknown whether it is idempotent
+    /// This is used in [`RetryPolicy`] to decide if retrying a query is safe
+    pub fn set_is_idempotent(&mut self, is_idempotent: bool) {
+        self.config.is_idempotent = is_idempotent;
+    }
+
+    /// Gets the idempotence of this statement
+    pub fn get_is_idempotent(&self) -> bool {
+        self.config.is_idempotent
+    }
+
+    /// Sets a custom [`RetryPolicy`] to be used with this statement
+    /// By default Session's retry policy is used, this allows to use a custom retry policy
+    pub fn set_retry_policy(&mut self, retry_policy: Box<dyn RetryPolicy>) {
+        self.config.retry_policy = Some(retry_policy);
+    }
+
+    /// Gets custom [`RetryPolicy`] used by this statement
+    pub fn get_retry_policy(&self) -> &Option<Box<dyn RetryPolicy>> {
+        &self.config.retry_policy
+    }
+
+    /// Enable or disable CQL Tracing for this batch
+    /// If enabled session.batch() will return a BatchResult containing tracing_id
+    /// which can be used to query tracing information about the execution of this query
+    pub fn set_tracing(&mut self, should_trace: bool) {
+        self.config.tracing = should_trace;
+    }
+
+    /// Gets whether tracing is enabled for this batch
+    pub fn get_tracing(&self) -> bool {
+        self.config.tracing
+    }
 }
 
 #[derive(Debug, Error, PartialEq, Eq, PartialOrd, Ord)]
@@ -201,23 +197,3 @@ pub enum PartitionKeyError {
     #[error("Value bytes too long to create partition key, max 65 535 allowed! value.len(): {0}")]
     ValueTooLong(usize),
 }
-
-impl Clone for PreparedStatement {
-    fn clone(&self) -> PreparedStatement {
-        PreparedStatement {
-            id: self.id.clone(),
-            metadata: self.metadata.clone(),
-            statement: self.statement.clone(),
-            prepare_tracing_ids: self.prepare_tracing_ids.clone(),
-            page_size: self.page_size,
-            consistency: self.consistency,
-            serial_consistency: self.serial_consistency,
-            is_idempotent: self.is_idempotent,
-            retry_policy: self
-                .retry_policy
-                .as_ref()
-                .map(|policy| policy.clone_boxed()),
-            tracing: self.tracing,
-        }
-    }
-}
diff --git a/scylla/src/statement/query.rs b/scylla/src/statement/query.rs
index ac5c206d..6cc8fd59 100644
--- a/scylla/src/statement/query.rs
+++ b/scylla/src/statement/query.rs
@@ -1,17 +1,16 @@
-use super::Consistency;
+use super::StatementConfig;
+use crate::frame::types::Consistency;
 use crate::transport::retry_policy::RetryPolicy;
 
 /// CQL query statement.
 ///
 /// This represents a CQL query that can be executed on a server.
+#[derive(Clone)]
 pub struct Query {
+    pub(crate) config: StatementConfig,
+
     contents: String,
     page_size: Option<i32>,
-    pub consistency: Consistency,
-    pub serial_consistency: Option<Consistency>,
-    pub is_idempotent: bool,
-    pub retry_policy: Option<Box<dyn RetryPolicy + Send + Sync>>,
-    pub tracing: bool,
 }
 
 impl Query {
@@ -20,11 +19,7 @@ impl Query {
         Self {
             contents,
             page_size: None,
-            consistency: Default::default(),
-            serial_consistency: None,
-            is_idempotent: false,
-            retry_policy: None,
-            tracing: false,
+            config: Default::default(),
         }
     }
 
@@ -55,63 +50,63 @@ impl Query {
         self.page_size
     }
 
-    /// Sets the consistency to be used when executing this query.
+    /// Sets the consistency to be used when executing this batch.
     pub fn set_consistency(&mut self, c: Consistency) {
-        self.consistency = c;
+        self.config.consistency = c;
     }
 
-    /// Gets the consistency to be used when executing this query.
+    /// Gets the consistency to be used when executing this batch.
     pub fn get_consistency(&self) -> Consistency {
-        self.consistency
+        self.config.consistency
     }
 
-    /// Sets the serial consistency to be used when executing this query.
-    /// (Ignored unless the query is an LWT)
+    /// Sets the serial consistency to be used when executing this batch.
+    /// (Ignored unless the batch is an LWT)
     pub fn set_serial_consistency(&mut self, sc: Option<Consistency>) {
-        self.serial_consistency = sc;
+        self.config.serial_consistency = sc;
     }
 
-    /// Gets the serial consistency to be used when executing this query.
-    /// (Ignored unless the query is an LWT)
+    /// Gets the serial consistency to be used when executing this batch.
+    /// (Ignored unless the batch is an LWT)
     pub fn get_serial_consistency(&self) -> Option<Consistency> {
-        self.serial_consistency
+        self.config.serial_consistency
     }
 
-    /// Sets the idempotence of this query  
-    /// A query is idempotent if it can be applied multiple times without changing the result of the initial application  
-    /// If set to `true` we can be sure that it is idempotent  
-    /// If set to `false` it is unknown whether it is idempotent  
+    /// Sets the idempotence of this statement
+    /// A query is idempotent if it can be applied multiple times without changing the result of the initial application
+    /// If set to `true` we can be sure that it is idempotent
+    /// If set to `false` it is unknown whether it is idempotent
     /// This is used in [`RetryPolicy`] to decide if retrying a query is safe
     pub fn set_is_idempotent(&mut self, is_idempotent: bool) {
-        self.is_idempotent = is_idempotent;
+        self.config.is_idempotent = is_idempotent;
     }
 
     /// Gets the idempotence of this statement
     pub fn get_is_idempotent(&self) -> bool {
-        self.is_idempotent
+        self.config.is_idempotent
     }
 
-    /// Sets a custom [`RetryPolicy`] to be used with this statement  
+    /// Sets a custom [`RetryPolicy`] to be used with this statement
     /// By default Session's retry policy is used, this allows to use a custom retry policy
-    pub fn set_retry_policy(&mut self, retry_policy: Box<dyn RetryPolicy + Send + Sync>) {
-        self.retry_policy = Some(retry_policy);
+    pub fn set_retry_policy(&mut self, retry_policy: Box<dyn RetryPolicy>) {
+        self.config.retry_policy = Some(retry_policy);
     }
 
     /// Gets custom [`RetryPolicy`] used by this statement
-    pub fn get_retry_policy(&self) -> &Option<Box<dyn RetryPolicy + Send + Sync>> {
-        &self.retry_policy
+    pub fn get_retry_policy(&self) -> &Option<Box<dyn RetryPolicy>> {
+        &self.config.retry_policy
     }
 
-    /// Enable or disable CQL Tracing for this query  
-    /// If enabled session.query() will return QueryResult containing tracing_id
+    /// Enable or disable CQL Tracing for this batch
+    /// If enabled session.batch() will return a BatchResult containing tracing_id
     /// which can be used to query tracing information about the execution of this query
     pub fn set_tracing(&mut self, should_trace: bool) {
-        self.tracing = should_trace;
+        self.config.tracing = should_trace;
     }
 
-    /// Gets whether tracing is enabled for this query
+    /// Gets whether tracing is enabled for this batch
     pub fn get_tracing(&self) -> bool {
-        self.tracing
+        self.config.tracing
     }
 }
 
@@ -126,20 +121,3 @@ impl<'a> From<&'a str> for Query {
         Query::new(s.to_owned())
     }
 }
-
-impl Clone for Query {
-    fn clone(&self) -> Query {
-        Query {
-            contents: self.contents.clone(),
-            page_size: self.page_size,
-            consistency: self.consistency,
-            serial_consistency: self.serial_consistency,
-            is_idempotent: self.is_idempotent,
-            retry_policy: self
-                .retry_policy
-                .as_ref()
-                .map(|policy| policy.clone_boxed()),
-            tracing: self.tracing,
-        }
-    }
-}
diff --git a/scylla/src/transport/cluster.rs b/scylla/src/transport/cluster.rs
index 9178ecad..726033fc 100644
--- a/scylla/src/transport/cluster.rs
+++ b/scylla/src/transport/cluster.rs
@@ -432,7 +432,7 @@ impl ClusterWorker {
         }
 
         // We can unwrap io_error because use_keyspace_futures must be nonempty
-        return Err(QueryError::IoError(io_error.unwrap()));
+        Err(QueryError::IoError(io_error.unwrap()))
     }
 
     async fn perform_refresh(&mut self) -> Result<(), QueryError> {
diff --git a/scylla/src/transport/connection.rs b/scylla/src/transport/connection.rs
index 1bbeff41..25e73385 100644
--- a/scylla/src/transport/connection.rs
+++ b/scylla/src/transport/connection.rs
@@ -228,7 +228,7 @@ impl Connection {
                     query: &query.get_contents(),
                 },
                 true,
-                query.tracing,
+                query.config.tracing,
             )
             .await?;
 
@@ -310,7 +310,8 @@ impl Connection {
             },
         };
 
-        self.send_request(&query_frame, true, query.tracing).await
+        self.send_request(&query_frame, true, query.config.tracing)
+            .await
     }
 
     pub async fn execute_single_page(
@@ -344,7 +345,7 @@ impl Connection {
         };
 
         let query_response = self
-            .send_request(&execute_frame, true, prepared_statement.tracing)
+            .send_request(&execute_frame, true, prepared_statement.config.tracing)
             .await?;
 
         if let Response::Error(err) = &query_response.response {
@@ -361,7 +362,7 @@ impl Connection {
                 }
 
                 return self
-                    .send_request(&execute_frame, true, prepared_statement.tracing)
+                    .send_request(&execute_frame, true, prepared_statement.config.tracing)
                     .await;
             }
         }
@@ -400,7 +401,9 @@ impl Connection {
             serial_consistency: batch.get_serial_consistency(),
         };
 
-        let query_response = self.send_request(&batch_frame, true, batch.tracing).await?;
+        let query_response = self
+            .send_request(&batch_frame, true, batch.config.tracing)
+            .await?;
 
         match query_response.response {
             Response::Error(err) => Err(err.into()),
diff --git a/scylla/src/transport/connection_keeper.rs b/scylla/src/transport/connection_keeper.rs
index 799e65cb..616c29aa 100644
--- a/scylla/src/transport/connection_keeper.rs
+++ b/scylla/src/transport/connection_keeper.rs
@@ -277,10 +277,10 @@ impl ConnectionKeeperWorker {
         }
 
         // Tried all source ports for that shard, give up
-        return Err(QueryError::IoError(Arc::new(std::io::Error::new(
+        Err(QueryError::IoError(Arc::new(std::io::Error::new(
             std::io::ErrorKind::AddrInUse,
             "Could not find free source port for shard",
-        ))));
+        ))))
     }
 }
 
diff --git a/scylla/src/transport/iterator.rs b/scylla/src/transport/iterator.rs
index 0989cda8..e49242b5 100644
--- a/scylla/src/transport/iterator.rs
+++ b/scylla/src/transport/iterator.rs
@@ -98,7 +98,7 @@ impl RowIterator {
     pub(crate) fn new_for_query(
         query: Query,
         values: SerializedValues,
-        retry_session: Box<dyn RetrySession + Send + Sync>,
+        retry_session: Box<dyn RetrySession>,
         load_balancer: Arc<dyn LoadBalancingPolicy>,
         cluster_data: Arc<ClusterData>,
         metrics: Arc<Metrics>,
@@ -120,8 +120,8 @@ impl RowIterator {
                 choose_connection,
                 page_query,
                 statement_info: Statement::default(),
-                query_is_idempotent: query.is_idempotent,
-                query_consistency: query.consistency,
+                query_is_idempotent: query.config.is_idempotent,
+                query_consistency: query.config.consistency,
                 retry_session,
                 load_balancer,
                 metrics,
@@ -145,7 +145,7 @@ impl RowIterator {
         prepared: PreparedStatement,
         values: SerializedValues,
         token: Token,
-        retry_session: Box<dyn RetrySession + Send + Sync>,
+        retry_session: Box<dyn RetrySession>,
         load_balancer: Arc<dyn LoadBalancingPolicy>,
         cluster_data: Arc<ClusterData>,
         metrics: Arc<Metrics>,
@@ -175,8 +175,8 @@ impl RowIterator {
                 choose_connection,
                 page_query,
                 statement_info,
-                query_is_idempotent: prepared.is_idempotent,
-                query_consistency: prepared.consistency,
+                query_is_idempotent: prepared.config.is_idempotent,
+                query_consistency: prepared.config.consistency,
                 retry_session,
                 load_balancer,
                 metrics,
@@ -223,7 +223,7 @@ struct RowIteratorWorker<'a, ConnFunc, QueryFunc> {
     query_is_idempotent: bool,
     query_consistency: Consistency,
 
-    retry_session: Box<dyn RetrySession + Send + Sync>,
+    retry_session: Box<dyn RetrySession>,
     load_balancer: Arc<dyn LoadBalancingPolicy>,
     metrics: Arc<Metrics>,
 
diff --git a/scylla/src/transport/metrics.rs b/scylla/src/transport/metrics.rs
index 8dc75b0c..19cc6751 100644
--- a/scylla/src/transport/metrics.rs
+++ b/scylla/src/transport/metrics.rs
@@ -28,7 +28,7 @@ impl std::fmt::Display for MetricsError<'_> {
     }
 }
 
-#[derive(Default)]
+#[derive(Default, Debug)]
 pub struct Metrics {
     errors_num: AtomicU64,
     queries_num: AtomicU64,
@@ -51,28 +51,28 @@ impl Metrics {
     }
 
     /// Increments counter for errors that occured in nonpaged queries.
-    pub fn inc_failed_nonpaged_queries(&self) {
+    pub(crate) fn inc_failed_nonpaged_queries(&self) {
         self.errors_num.fetch_add(1, ORDER_TYPE);
     }
 
     /// Increments counter for nonpaged queries.
-    pub fn inc_total_nonpaged_queries(&self) {
+    pub(crate) fn inc_total_nonpaged_queries(&self) {
         self.queries_num.fetch_add(1, ORDER_TYPE);
     }
 
     /// Increments counter for errors that occured in paged queries.
-    pub fn inc_failed_paged_queries(&self) {
+    pub(crate) fn inc_failed_paged_queries(&self) {
         self.errors_iter_num.fetch_add(1, ORDER_TYPE);
     }
 
     /// Increments counter for page queries in paged queries.
     /// If query_iter would return 4 pages then this counter should be incremented 4 times.
-    pub fn inc_total_paged_queries(&self) {
+    pub(crate) fn inc_total_paged_queries(&self) {
         self.queries_iter_num.fetch_add(1, ORDER_TYPE);
     }
 
     /// Increments counter measuring how many times a retry policy has decided to retry a query
-    pub fn inc_retries_num(&self) {
+    pub(crate) fn inc_retries_num(&self) {
         self.retries_num.fetch_add(1, ORDER_TYPE);
     }
 
@@ -82,15 +82,15 @@ impl Metrics {
     /// # Arguments
     ///
     /// * `latency` - time in milliseconds that should be logged
-    pub fn log_query_latency(&self, latency: u64) -> Result<(), MetricsError> {
-        let mut histogram_unlocked = self.histogram.lock()?;
+    pub(crate) fn log_query_latency(&self, latency: u64) -> Result<(), MetricsError> {
+        let mut histogram_unlocked = self.histogram.lock().unwrap();
         histogram_unlocked.increment(latency)?;
         Ok(())
     }
 
     /// Returns average latency in milliseconds
     pub fn get_latency_avg_ms(&self) -> Result<u64, MetricsError> {
-        let histogram_unlocked = self.histogram.lock()?;
+        let histogram_unlocked = self.histogram.lock().unwrap();
         Ok(histogram_unlocked.mean()?)
     }
 
@@ -99,7 +99,7 @@ impl Metrics {
     ///
     /// * `percentile` - float value (0.0 - 100.0)
     pub fn get_latency_percentile_ms(&self, percentile: f64) -> Result<u64, MetricsError> {
-        let histogram_unlocked = self.histogram.lock()?;
+        let histogram_unlocked = self.histogram.lock().unwrap();
         Ok(histogram_unlocked.percentile(percentile)?)
     }
 
@@ -128,51 +128,3 @@ impl Metrics {
         self.retries_num.load(ORDER_TYPE)
     }
 }
-
-pub struct MetricsView {
-    metrics: Arc<Metrics>,
-}
-
-impl MetricsView {
-    pub fn new(metrics: Arc<Metrics>) -> Self {
-        Self { metrics }
-    }
-
-    /// Returns average latency in milliseconds
-    pub fn get_latency_avg_ms(&self) -> Result<u64, MetricsError> {
-        self.metrics.get_latency_avg_ms()
-    }
-
-    /// Returns latency from histogram for a given percentile
-    /// # Arguments
-    ///
-    /// * `percentile` - float value (0.0 - 100.0)
-    pub fn get_latency_percentile_ms(&self, percentile: f64) -> Result<u64, MetricsError> {
-        self.metrics.get_latency_percentile_ms(percentile)
-    }
-
-    /// Returns counter for errors occured in nonpaged queries
-    pub fn get_errors_num(&self) -> u64 {
-        self.metrics.get_errors_num()
-    }
-
-    /// Returns counter for nonpaged queries
-    pub fn get_queries_num(&self) -> u64 {
-        self.metrics.get_queries_num()
-    }
-
-    /// Returns counter for errors occured in paged queries
-    pub fn get_errors_iter_num(&self) -> u64 {
-        self.metrics.get_errors_iter_num()
-    }
-
-    /// Returns counter for pages requested in paged queries
-    pub fn get_queries_iter_num(&self) -> u64 {
-        self.metrics.get_queries_iter_num()
-    }
-
-    /// Returns counter measuring how many times a retry policy has decided to retry a query
-    pub fn get_retries_num(&self) -> u64 {
-        self.metrics.get_retries_num()
-    }
-}
diff --git a/scylla/src/transport/mod.rs b/scylla/src/transport/mod.rs
index 12f8462a..0eff9c9a 100644
--- a/scylla/src/transport/mod.rs
+++ b/scylla/src/transport/mod.rs
@@ -6,6 +6,7 @@ mod node;
 pub mod retry_policy;
 pub mod session;
 pub mod session_builder;
+pub mod speculative_execution;
 mod topology;
 
 pub mod errors;
diff --git a/scylla/src/transport/node.rs b/scylla/src/transport/node.rs
index ec100183..acda5d92 100644
--- a/scylla/src/transport/node.rs
+++ b/scylla/src/transport/node.rs
@@ -190,7 +190,7 @@ impl Node {
             }
         }
 
-        return Err(last_error);
+        Err(last_error)
     }
 
     pub async fn use_keyspace(
@@ -373,6 +373,6 @@ impl NodeWorker {
         }
 
         // We can unwrap io_error because use_keyspace_futures must be nonempty
-        return Err(QueryError::IoError(io_error.unwrap()));
+        Err(QueryError::IoError(io_error.unwrap()))
     }
 }
diff --git a/scylla/src/transport/retry_policy.rs b/scylla/src/transport/retry_policy.rs
index eb963cdc..36212bd5 100644
--- a/scylla/src/transport/retry_policy.rs
+++ b/scylla/src/transport/retry_policy.rs
@@ -25,23 +25,23 @@ pub enum RetryDecision {
 }
 
 /// Specifies a policy used to decide when to retry a query
-pub trait RetryPolicy {
+pub trait RetryPolicy: Send + Sync {
     /// Called for each new query, starts a session of deciding about retries
-    fn new_session(&self) -> Box<dyn RetrySession + Send + Sync>;
+    fn new_session(&self) -> Box<dyn RetrySession>;
 
     /// Used to clone this RetryPolicy
-    fn clone_boxed(&self) -> Box<dyn RetryPolicy + Send + Sync>;
+    fn clone_boxed(&self) -> Box<dyn RetryPolicy>;
 }
 
-impl Clone for Box<dyn RetryPolicy + Send + Sync> {
-    fn clone(&self) -> Box<dyn RetryPolicy + Send + Sync> {
+impl Clone for Box<dyn RetryPolicy> {
+    fn clone(&self) -> Box<dyn RetryPolicy> {
         self.clone_boxed()
     }
 }
 
 /// Used throughout a single query to decide when to retry it
 /// After this query is finished it is destroyed or reset
-pub trait RetrySession {
+pub trait RetrySession: Send + Sync {
     /// Called after the query failed - decide what to do next
     fn decide_should_retry(&mut self, query_info: QueryInfo) -> RetryDecision;
 
@@ -66,11 +66,11 @@ impl Default for FallthroughRetryPolicy {
 }
 
 impl RetryPolicy for FallthroughRetryPolicy {
-    fn new_session(&self) -> Box<dyn RetrySession + Send + Sync> {
+    fn new_session(&self) -> Box<dyn RetrySession> {
         Box::new(FallthroughRetrySession)
     }
 
-    fn clone_boxed(&self) -> Box<dyn RetryPolicy + Send + Sync> {
+    fn clone_boxed(&self) -> Box<dyn RetryPolicy> {
         Box::new(FallthroughRetryPolicy)
     }
 }
@@ -100,11 +100,11 @@ impl Default for DefaultRetryPolicy {
 }
 
 impl RetryPolicy for DefaultRetryPolicy {
-    fn new_session(&self) -> Box<dyn RetrySession + Send + Sync> {
+    fn new_session(&self) -> Box<dyn RetrySession> {
         Box::new(DefaultRetrySession::new())
     }
 
-    fn clone_boxed(&self) -> Box<dyn RetryPolicy + Send + Sync> {
+    fn clone_boxed(&self) -> Box<dyn RetryPolicy> {
         Box::new(DefaultRetryPolicy)
     }
 }
diff --git a/scylla/src/transport/session.rs b/scylla/src/transport/session.rs
index d8213e43..3d29fc19 100644
--- a/scylla/src/transport/session.rs
+++ b/scylla/src/transport/session.rs
@@ -13,8 +13,6 @@ use tracing::warn;
 use uuid::Uuid;
 
 use super::errors::{BadQuery, NewSessionError, QueryError};
-use crate::batch::Batch;
-use crate::cql_to_rust::FromRow;
 use crate::frame::response::cql_to_rust::FromRowError;
 use crate::frame::response::result;
 use crate::frame::value::{BatchValues, SerializedValues, ValueList};
@@ -28,11 +26,14 @@ use crate::transport::{
     connection::{BatchResult, Connection, ConnectionConfig, QueryResult, VerifiedKeyspaceName},
     iterator::RowIterator,
     load_balancing::{LoadBalancingPolicy, RoundRobinPolicy, Statement, TokenAwarePolicy},
-    metrics::{Metrics, MetricsView},
+    metrics::Metrics,
     node::Node,
     retry_policy::{DefaultRetryPolicy, QueryInfo, RetryDecision, RetryPolicy, RetrySession},
+    speculative_execution::SpeculativeExecutionPolicy,
     Compression,
 };
+use crate::{batch::Batch, statement::StatementConfig};
+use crate::{cql_to_rust::FromRow, transport::speculative_execution};
 
 #[cfg(feature = "ssl")]
 use openssl::ssl::SslContext;
@@ -41,8 +42,9 @@ use openssl::ssl::SslContext;
 pub struct Session {
     cluster: Cluster,
     load_balancer: Arc<dyn LoadBalancingPolicy>,
-    retry_policy: Box<dyn RetryPolicy + Send + Sync>,
     schema_agreement_interval: Duration,
+    retry_policy: Box<dyn RetryPolicy>,
+    speculative_execution_policy: Option<Arc<dyn SpeculativeExecutionPolicy>>,
 
     metrics: Arc<Metrics>,
 }
@@ -68,7 +70,8 @@ pub struct SessionConfig {
     pub used_keyspace: Option<String>,
     pub keyspace_case_sensitive: bool,
 
-    pub retry_policy: Box<dyn RetryPolicy + Send + Sync>,
+    pub retry_policy: Box<dyn RetryPolicy>,
+    pub speculative_execution_policy: Option<Arc<dyn SpeculativeExecutionPolicy>>,
 
     /// Provide our Session with TLS
     #[cfg(feature = "ssl")]
@@ -117,6 +120,7 @@ impl SessionConfig {
             used_keyspace: None,
             keyspace_case_sensitive: false,
             retry_policy: Box::new(DefaultRetryPolicy),
+            speculative_execution_policy: None,
             #[cfg(feature = "ssl")]
             ssl_context: None,
             auth_username: None,
@@ -286,14 +290,14 @@ impl Session {
 
         // Start the session
         let cluster = Cluster::new(&node_addresses, config.get_connection_config()).await?;
-        let metrics = Arc::new(Metrics::new());
 
         let session = Session {
             cluster,
             load_balancer: config.load_balancing,
             retry_policy: config.retry_policy,
             schema_agreement_interval: config.schema_agreement_interval,
-            metrics,
+            speculative_execution_policy: config.speculative_execution_policy,
+            metrics: Arc::new(Metrics::new()),
         };
 
         if let Some(keyspace_name) = config.used_keyspace {
@@ -389,11 +393,6 @@ impl Session {
                 .map(|_| QueryResult::default());
         }
 
-        let retry_session = match &query.retry_policy {
-            Some(policy) => policy.new_session(),
-            None => self.retry_policy.new_session(),
-        };
-
         // Needed to avoid moving query and values into async move block
         let query_ref: &Query = &query;
         let values_ref = &serialized_values;
@@ -401,9 +400,7 @@ impl Session {
 
         self.run_query(
             Statement::default(),
-            query.is_idempotent,
-            query.consistency,
-            retry_session,
+            &query.config,
             |node: Arc<Node>| async move { node.random_connection().await },
             |connection: Arc<Connection>| async move {
                 connection
@@ -455,7 +452,7 @@ impl Session {
         let query: Query = query.into();
         let serialized_values = values.serialized()?;
 
-        let retry_session = match &query.retry_policy {
+        let retry_session = match &query.config.retry_policy {
             Some(policy) => policy.new_session(),
             None => self.retry_policy.new_session(),
         };
@@ -616,16 +613,9 @@ impl Session {
             keyspace: prepared.get_keyspace_name(),
         };
 
-        let retry_session = match &prepared.retry_policy {
-            Some(policy) => policy.new_session(),
-            None => self.retry_policy.new_session(),
-        };
-
         self.run_query(
             statement_info,
-            prepared.is_idempotent,
-            prepared.consistency,
-            retry_session,
+            &prepared.config,
             |node: Arc<Node>| async move { node.connection_for_token(token).await },
             |connection: Arc<Connection>| async move {
                 connection
@@ -687,7 +677,7 @@ impl Session {
 
         let token = calculate_token(&prepared, &serialized_values)?;
 
-        let retry_session = match &prepared.retry_policy {
+        let retry_session = match &prepared.config.retry_policy {
             Some(policy) => policy.new_session(),
             None => self.retry_policy.new_session(),
         };
@@ -750,16 +740,9 @@ impl Session {
     ) -> Result<BatchResult, QueryError> {
         let values_ref = &values;
 
-        let retry_session = match &batch.retry_policy {
-            Some(policy) => policy.new_session(),
-            None => self.retry_policy.new_session(),
-        };
-
         self.run_query(
             Statement::default(),
-            batch.is_idempotent,
-            batch.consistency,
-            retry_session,
+            &batch.config,
             |node: Arc<Node>| async move { node.random_connection().await },
             |connection: Arc<Connection>| async move { connection.batch(batch, values_ref).await },
         )
@@ -829,8 +812,8 @@ impl Session {
     /// Access metrics collected by the driver  
     /// Driver collects various metrics like number of queries or query latencies.
     /// They can be read using this method
-    pub fn get_metrics(&self) -> MetricsView {
-        MetricsView::new(self.metrics.clone())
+    pub fn get_metrics(&self) -> Arc<Metrics> {
+        self.metrics.clone()
     }
 
     /// Get [`TracingInfo`] of a traced query performed earlier
@@ -863,12 +846,12 @@ impl Session {
             };
         }
 
-        return Err(QueryError::ProtocolError(
+        Err(QueryError::ProtocolError(
             "All tracing queries returned an empty result, \
             maybe information didnt reach this node yet. \
             Consider using get_tracing_info_custom with \
             bigger interval in GetTracingConfig",
-        ));
+        ))
     }
 
     // Tries getting the tracing info
@@ -882,13 +865,13 @@ impl Session {
         // Query system_traces.sessions for TracingInfo
         let mut traces_session_query =
             Query::new(crate::tracing::TRACES_SESSION_QUERY_STR.to_string());
-        traces_session_query.consistency = consistency;
+        traces_session_query.config.consistency = consistency;
 
         // Query system_traces.events for TracingEvents
         let mut traces_events_query =
             Query::new(crate::tracing::TRACES_EVENTS_QUERY_STR.to_string());
-        traces_events_query.consistency = Consistency::One;
-        traces_events_query.consistency = consistency;
+        traces_events_query.config.consistency = Consistency::One;
+        traces_events_query.config.consistency = consistency;
 
         let (traces_session_res, traces_events_res) = tokio::try_join!(
             self.query(traces_session_query, (tracing_id,)),
@@ -953,9 +936,7 @@ impl Session {
     async fn run_query<'a, ConnFut, QueryFut, ResT>(
         &'a self,
         statement_info: Statement<'a>,
-        query_is_idempotent: bool,
-        query_consistency: Consistency,
-        mut retry_session: Box<dyn RetrySession + Send + Sync>,
+        statement_config: &StatementConfig,
         choose_connection: impl Fn(Arc<Node>) -> ConnFut,
         do_query: impl Fn(Arc<Connection>) -> QueryFut,
     ) -> Result<ResT, QueryError>
@@ -966,15 +947,102 @@ impl Session {
         let cluster_data = self.cluster.get_data();
         let query_plan = self.load_balancer.plan(&statement_info, &cluster_data);
 
-        let mut last_error: QueryError =
-            QueryError::ProtocolError("Empty query plan - driver bug!");
+        // If a speculative execution policy is used to run query, query_plan has to be shared
+        // between different async functions. This struct helps to wrap query_plan in mutex so it
+        // can be shared safely.
+        struct SharedPlan<I>
+        where
+            I: Iterator<Item = Arc<Node>>,
+        {
+            iter: std::sync::Mutex<I>,
+        }
+
+        impl<I> Iterator for &SharedPlan<I>
+        where
+            I: Iterator<Item = Arc<Node>>,
+        {
+            type Item = Arc<Node>;
+
+            fn next(&mut self) -> Option<Self::Item> {
+                self.iter.lock().unwrap().next()
+            }
+        }
+
+        let retry_policy = match &statement_config.retry_policy {
+            Some(policy) => policy,
+            None => &self.retry_policy,
+        };
+
+        let speculative_policy = statement_config
+            .speculative_execution_policy
+            .as_ref()
+            .or_else(|| self.speculative_execution_policy.as_ref());
+
+        match speculative_policy {
+            Some(speculative) if statement_config.is_idempotent => {
+                let shared_query_plan = SharedPlan {
+                    iter: std::sync::Mutex::new(query_plan),
+                };
+
+                let execute_query_generator = || {
+                    self.execute_query(
+                        &shared_query_plan,
+                        statement_config.is_idempotent,
+                        statement_config.consistency,
+                        retry_policy.new_session(),
+                        &choose_connection,
+                        &do_query,
+                    )
+                };
+
+                let context = speculative_execution::Context {
+                    metrics: self.metrics.clone(),
+                };
+
+                speculative_execution::execute(
+                    speculative.as_ref(),
+                    &context,
+                    execute_query_generator,
+                )
+                .await
+            }
+            _ => self
+                .execute_query(
+                    query_plan,
+                    statement_config.is_idempotent,
+                    statement_config.consistency,
+                    retry_policy.new_session(),
+                    &choose_connection,
+                    &do_query,
+                )
+                .await
+                .unwrap_or(Err(QueryError::ProtocolError(
+                    "Empty query plan - driver bug!",
+                ))),
+        }
+    }
+
+    async fn execute_query<ConnFut, QueryFut, ResT>(
+        &self,
+        query_plan: impl Iterator<Item = Arc<Node>>,
+        is_idempotent: bool,
+        consistency: Consistency,
+        mut retry_session: Box<dyn RetrySession>,
+        choose_connection: impl Fn(Arc<Node>) -> ConnFut,
+        do_query: impl Fn(Arc<Connection>) -> QueryFut,
+    ) -> Option<Result<ResT, QueryError>>
+    where
+        ConnFut: Future<Output = Result<Arc<Connection>, QueryError>>,
+        QueryFut: Future<Output = Result<ResT, QueryError>>,
+    {
+        let mut last_error: Option<QueryError> = None;
 
         'nodes_in_plan: for node in query_plan {
             'same_node_retries: loop {
                 let connection: Arc<Connection> = match choose_connection(node.clone()).await {
                     Ok(connection) => connection,
                     Err(e) => {
-                        last_error = e;
+                        last_error = Some(e);
                         // Broken connection doesn't count as a failed query, don't log in metrics
                         continue 'nodes_in_plan;
                     }
@@ -990,19 +1058,19 @@ impl Session {
                         let _ = self
                             .metrics
                             .log_query_latency(query_start.elapsed().as_millis() as u64);
-                        return Ok(response);
+                        return Some(Ok(response));
                     }
                     Err(e) => {
                         self.metrics.inc_failed_nonpaged_queries();
-                        e
+                        Some(e)
                     }
                 };
 
                 // Use retry policy to decide what to do next
                 let query_info = QueryInfo {
-                    error: &last_error,
-                    is_idempotent: query_is_idempotent,
-                    consistency: query_consistency,
+                    error: last_error.as_ref().unwrap(),
+                    is_idempotent,
+                    consistency,
                 };
 
                 match retry_session.decide_should_retry(query_info) {
@@ -1014,12 +1082,12 @@ impl Session {
                         self.metrics.inc_retries_num();
                         continue 'nodes_in_plan;
                     }
-                    RetryDecision::DontRetry => return Err(last_error),
+                    RetryDecision::DontRetry => return last_error.map(Result::Err),
                 };
             }
         }
 
-        return Err(last_error);
+        last_error.map(Result::Err)
     }
 
     pub async fn await_schema_agreement(&self) -> Result<(), QueryError> {
@@ -1045,11 +1113,16 @@ impl Session {
     where
         QueryFut: Future<Output = Result<ResT, QueryError>>,
     {
+        let info = Statement::default();
+        let config = StatementConfig {
+            is_idempotent: true,
+            serial_consistency: Some(Consistency::One),
+            ..Default::default()
+        };
+
         self.run_query(
-            Statement::default(),
-            true,
-            Consistency::One,
-            self.retry_policy.new_session(),
+            info,
+            &config,
             |node: Arc<Node>| async move { node.random_connection().await },
             do_query,
         )
diff --git a/scylla/src/transport/session_builder.rs b/scylla/src/transport/session_builder.rs
index f080f8ed..b3b176f3 100644
--- a/scylla/src/transport/session_builder.rs
+++ b/scylla/src/transport/session_builder.rs
@@ -3,6 +3,7 @@
 use super::errors::NewSessionError;
 use super::load_balancing::LoadBalancingPolicy;
 use super::session::{Session, SessionConfig};
+use super::speculative_execution::SpeculativeExecutionPolicy;
 use super::Compression;
 use crate::transport::retry_policy::RetryPolicy;
 use std::net::SocketAddr;
@@ -251,6 +252,39 @@ impl SessionBuilder {
         self
     }
 
+    /// Set the speculative execution policy
+    /// The default is None
+    /// # Example
+    /// ```
+    /// # extern crate scylla;
+    /// # use scylla::Session;
+    /// # use std::error::Error;
+    /// # async fn check_only_compiles() -> Result<(), Box<dyn Error>> {
+    /// use std::{sync::Arc, time::Duration};
+    /// use scylla::{
+    ///     Session,
+    ///     SessionBuilder,
+    ///     transport::speculative_execution::SimpleSpeculativeExecutionPolicy
+    /// };
+    ///
+    /// let policy = SimpleSpeculativeExecutionPolicy {
+    ///     max_retry_count: 3,
+    ///     retry_interval: Duration::from_millis(100),
+    /// };
+    ///
+    /// let session: Session = SessionBuilder::new()
+    ///     .known_node("127.0.0.1:9042")
+    ///     .speculative_execution(Arc::new(policy))
+    ///     .build()
+    ///     .await?;
+    /// # Ok(())
+    /// # }
+    /// ```
+    pub fn speculative_execution(mut self, policy: Arc<dyn SpeculativeExecutionPolicy>) -> Self {
+        self.config.speculative_execution_policy = Some(policy);
+        self
+    }
+
     /// Sets the [`RetryPolicy`] to use by default on queries
     /// The default is [DefaultRetryPolicy](crate::transport::retry_policy::DefaultRetryPolicy)
     /// It is possible to implement a custom retry policy by implementing the trait [`RetryPolicy`]
@@ -446,13 +480,13 @@ mod tests {
     #[test]
     fn tcp_nodelay() {
         let mut builder = SessionBuilder::new();
-        assert_eq!(builder.config.tcp_nodelay, true);
+        assert!(builder.config.tcp_nodelay);
 
         builder = builder.tcp_nodelay(false);
-        assert_eq!(builder.config.tcp_nodelay, false);
+        assert!(!builder.config.tcp_nodelay);
 
         builder = builder.tcp_nodelay(true);
-        assert_eq!(builder.config.tcp_nodelay, true);
+        assert!(builder.config.tcp_nodelay);
     }
 
     #[test]
@@ -474,15 +508,15 @@ mod tests {
     fn use_keyspace() {
         let mut builder = SessionBuilder::new();
         assert_eq!(builder.config.used_keyspace, None);
-        assert_eq!(builder.config.keyspace_case_sensitive, false);
+        assert!(!builder.config.keyspace_case_sensitive);
 
         builder = builder.use_keyspace("ks_name_1", true);
         assert_eq!(builder.config.used_keyspace, Some("ks_name_1".to_string()));
-        assert_eq!(builder.config.keyspace_case_sensitive, true);
+        assert!(builder.config.keyspace_case_sensitive);
 
         builder = builder.use_keyspace("ks_name_2", false);
         assert_eq!(builder.config.used_keyspace, Some("ks_name_2".to_string()));
-        assert_eq!(builder.config.keyspace_case_sensitive, false);
+        assert!(!builder.config.keyspace_case_sensitive);
     }
 
     #[test]
@@ -530,7 +564,7 @@ mod tests {
         );
 
         assert_eq!(builder.config.compression, Some(Compression::Snappy));
-        assert_eq!(builder.config.tcp_nodelay, true);
+        assert!(builder.config.tcp_nodelay);
         assert_eq!(
             builder.config.load_balancing.name(),
             "RoundRobinPolicy".to_string()
@@ -538,6 +572,6 @@ mod tests {
 
         assert_eq!(builder.config.used_keyspace, Some("ks_name".to_string()));
 
-        assert_eq!(builder.config.keyspace_case_sensitive, true);
+        assert!(builder.config.keyspace_case_sensitive);
     }
 }
diff --git a/scylla/src/transport/speculative_execution.rs b/scylla/src/transport/speculative_execution.rs
new file mode 100644
index 00000000..44e01f73
--- /dev/null
+++ b/scylla/src/transport/speculative_execution.rs
@@ -0,0 +1,143 @@
+use futures::{
+    future::FutureExt,
+    stream::{FuturesUnordered, StreamExt},
+};
+use std::{future::Future, sync::Arc, time::Duration};
+use tracing::warn;
+
+use super::{errors::QueryError, metrics::Metrics};
+
+/// Context is passed as an argument to `SpeculativeExecutionPolicy` methods
+pub struct Context {
+    pub metrics: Arc<Metrics>,
+}
+
+/// The policy that decides if the driver will send speculative queries to the
+/// next hosts when the current host takes too long to respond.
+pub trait SpeculativeExecutionPolicy: Send + Sync {
+    /// The maximum number of speculative executions that will be triggered
+    /// for a given request (does not include the initial request)
+    fn max_retry_count(&self, context: &Context) -> usize;
+
+    /// The delay between each speculative execution
+    fn retry_interval(&self, context: &Context) -> Duration;
+}
+
+/// A SpeculativeExecutionPolicy that schedules a given number of speculative
+/// executions, separated by a fixed delay.
+#[derive(Debug, Clone)]
+pub struct SimpleSpeculativeExecutionPolicy {
+    /// The maximum number of speculative executions that will be triggered
+    /// for a given request (does not include the initial request)
+    pub max_retry_count: usize,
+
+    /// The delay between each speculative execution
+    pub retry_interval: Duration,
+}
+
+/// A policy that triggers speculative executions when the request to the current
+/// host is above a given percentile.
+#[derive(Debug, Clone)]
+pub struct PercentileSpeculativeExecutionPolicy {
+    /// The maximum number of speculative executions that will be triggered
+    /// for a given request (does not include the initial request)
+    pub max_retry_count: usize,
+
+    /// The percentile that a request's latency must fall into to be considered
+    /// slow (ex: 99.0)
+    pub percentile: f64,
+}
+
+impl SpeculativeExecutionPolicy for SimpleSpeculativeExecutionPolicy {
+    fn max_retry_count(&self, _: &Context) -> usize {
+        self.max_retry_count
+    }
+
+    fn retry_interval(&self, _: &Context) -> Duration {
+        self.retry_interval
+    }
+}
+
+impl SpeculativeExecutionPolicy for PercentileSpeculativeExecutionPolicy {
+    fn max_retry_count(&self, _: &Context) -> usize {
+        self.max_retry_count
+    }
+
+    fn retry_interval(&self, context: &Context) -> Duration {
+        let interval = context.metrics.get_latency_percentile_ms(self.percentile);
+        let ms = match interval {
+            Ok(d) => d,
+            Err(e) => {
+                warn!(
+                    "Failed to get latency percentile ({}), defaulting to 100 ms",
+                    e
+                );
+                100
+            }
+        };
+        Duration::from_millis(ms)
+    }
+}
+
+// checks if a result created in a speculative execution branch can be ignored
+fn can_be_ignored<ResT>(result: &Result<ResT, QueryError>) -> bool {
+    match result {
+        Ok(_) => false,
+        Err(QueryError::IoError(_)) => true,
+        Err(QueryError::TimeoutError) => true,
+        _ => false,
+    }
+}
+
+const EMPTY_PLAN_ERROR: QueryError = QueryError::ProtocolError("Empty query plan - driver bug!");
+
+pub async fn execute<QueryFut, ResT>(
+    policy: &dyn SpeculativeExecutionPolicy,
+    context: &Context,
+    query_runner_generator: impl Fn() -> QueryFut,
+) -> Result<ResT, QueryError>
+where
+    QueryFut: Future<Output = Option<Result<ResT, QueryError>>>,
+{
+    let mut retries_remaining = policy.max_retry_count(context);
+    let retry_interval = policy.retry_interval(context);
+
+    let mut async_tasks = FuturesUnordered::new();
+    async_tasks.push(query_runner_generator());
+
+    let sleep = tokio::time::sleep(retry_interval).fuse();
+    tokio::pin!(sleep);
+
+    let mut last_error = None;
+    loop {
+        futures::select! {
+            _ = &mut sleep => {
+                if retries_remaining > 0 {
+                    async_tasks.push(query_runner_generator());
+                    retries_remaining -= 1;
+
+                    // reset the timeout
+                    sleep.set(tokio::time::sleep(retry_interval).fuse());
+                }
+            }
+            res = async_tasks.select_next_some() => {
+                match res {
+                    Some(r) => {
+                        if !can_be_ignored(&r) {
+                            return r;
+                        } else {
+                            last_error = Some(r)
+                        }
+                    },
+                    None =>  {
+                        if async_tasks.is_empty() && retries_remaining == 0 {
+                            return last_error.unwrap_or({
+                                Err(EMPTY_PLAN_ERROR)
+                            });
+                        }
+                    },
+                }
+            }
+        }
+    }
+}


Test patch:
diff --git a/scylla/src/frame/value_tests.rs b/scylla/src/frame/value_tests.rs
index c3120ea8..2fb52d1f 100644
--- a/scylla/src/frame/value_tests.rs
+++ b/scylla/src/frame/value_tests.rs
@@ -162,7 +162,7 @@ fn ref_value() {
 fn empty_serialized_values() {
     const EMPTY: SerializedValues = SerializedValues::new();
     assert_eq!(EMPTY.len(), 0);
-    assert_eq!(EMPTY.is_empty(), true);
+    assert!(EMPTY.is_empty());
     assert_eq!(EMPTY.iter().next(), None);
 
     let mut empty_request = Vec::<u8>::new();
@@ -173,13 +173,13 @@ fn empty_serialized_values() {
 #[test]
 fn serialized_values() {
     let mut values = SerializedValues::new();
-    assert_eq!(values.is_empty(), true);
+    assert!(values.is_empty());
 
     // Add first value
     values.add_value(&8_i8).unwrap();
     {
         assert_eq!(values.len(), 1);
-        assert_eq!(values.is_empty(), false);
+        assert!(!values.is_empty());
 
         let mut request = Vec::<u8>::new();
         values.write_to_request(&mut request);
@@ -192,7 +192,7 @@ fn serialized_values() {
     values.add_value(&16_i16).unwrap();
     {
         assert_eq!(values.len(), 2);
-        assert_eq!(values.is_empty(), false);
+        assert!(!values.is_empty());
 
         let mut request = Vec::<u8>::new();
         values.write_to_request(&mut request);
@@ -224,7 +224,7 @@ fn serialized_values() {
     // All checks for two values should still pass
     {
         assert_eq!(values.len(), 2);
-        assert_eq!(values.is_empty(), false);
+        assert!(!values.is_empty());
 
         let mut request = Vec::<u8>::new();
         values.write_to_request(&mut request);
diff --git a/scylla/src/transport/session_test.rs b/scylla/src/transport/session_test.rs
index 9862a7a6..0d73a2ac 100644
--- a/scylla/src/transport/session_test.rs
+++ b/scylla/src/transport/session_test.rs
@@ -688,7 +688,7 @@ async fn test_tracing_query(session: &Session) {
 
     // A query with tracing enabled has a tracing uuid in result
     let mut traced_query: Query = Query::new("SELECT * FROM test_tracing_ks.tab".to_string());
-    traced_query.tracing = true;
+    traced_query.config.tracing = true;
 
     let traced_query_result: QueryResult = session.query(traced_query, &[]).await.unwrap();
     assert!(traced_query_result.tracing_id.is_some());
@@ -715,7 +715,7 @@ async fn test_tracing_execute(session: &Session) {
         .await
         .unwrap();
 
-    traced_prepared.tracing = true;
+    traced_prepared.config.tracing = true;
 
     let traced_prepared_result: QueryResult = session.execute(&traced_prepared, &[]).await.unwrap();
     assert!(traced_prepared_result.tracing_id.is_some());
@@ -735,7 +735,7 @@ async fn test_tracing_prepare(session: &Session) {
 
     // Preparing a statement with tracing enabled has tracing uuids in result
     let mut to_prepare_traced = Query::new("SELECT * FROM test_tracing_ks.tab".to_string());
-    to_prepare_traced.tracing = true;
+    to_prepare_traced.config.tracing = true;
 
     let traced_prepared = session.prepare(to_prepare_traced).await.unwrap();
     assert!(!traced_prepared.prepare_tracing_ids.is_empty());
@@ -749,7 +749,7 @@ async fn test_tracing_prepare(session: &Session) {
 async fn test_get_tracing_info(session: &Session) {
     // A query with tracing enabled has a tracing uuid in result
     let mut traced_query: Query = Query::new("SELECT * FROM test_tracing_ks.tab".to_string());
-    traced_query.tracing = true;
+    traced_query.config.tracing = true;
 
     let traced_query_result: QueryResult = session.query(traced_query, &[]).await.unwrap();
     let tracing_id: Uuid = traced_query_result.tracing_id.unwrap();
@@ -776,7 +776,7 @@ async fn test_tracing_query_iter(session: &Session) {
 
     // A query with tracing enabled has a tracing ids in result
     let mut traced_query: Query = Query::new("SELECT * FROM test_tracing_ks.tab".to_string());
-    traced_query.tracing = true;
+    traced_query.config.tracing = true;
 
     let mut traced_row_iter = session.query_iter(traced_query, &[]).await.unwrap();
     while let Some(_row) = traced_row_iter.next().await {
@@ -817,7 +817,7 @@ async fn test_tracing_execute_iter(session: &Session) {
         .prepare("SELECT * FROM test_tracing_ks.tab")
         .await
         .unwrap();
-    traced_prepared.tracing = true;
+    traced_prepared.config.tracing = true;
 
     let mut traced_row_iter = session.execute_iter(traced_prepared, &[]).await.unwrap();
     while let Some(_row) = traced_row_iter.next().await {
@@ -846,7 +846,7 @@ async fn test_tracing_batch(session: &Session) {
     // Batch with tracing enabled has a tracing uuid in result
     let mut traced_batch: Batch = Default::default();
     traced_batch.append_statement("INSERT INTO test_tracing_ks.tab (a) VALUES('a')");
-    traced_batch.tracing = true;
+    traced_batch.config.tracing = true;
 
     let traced_batch_result: BatchResult = session.batch(&traced_batch, ((),)).await.unwrap();
     assert!(traced_batch_result.tracing_id.is_some());
@@ -857,7 +857,7 @@ async fn test_tracing_batch(session: &Session) {
 async fn assert_in_tracing_table(session: &Session, tracing_uuid: Uuid) {
     let mut traces_query =
         Query::new("SELECT * FROM system_traces.sessions WHERE session_id = ?".to_string());
-    traces_query.consistency = Consistency::One;
+    traces_query.config.consistency = Consistency::One;
 
     // Tracing info might not be immediately available
     // If rows are empty perform 8 retries with a 32ms wait in between