Repository: scylladb/scylla-rust-driver
Pull request #180 (2021-02-27 16:54:18)

Retry Policy
## Description
This PR adds support for `RetryPolicy` - when a query fails this `RetryPolicy` decides whether we should retry on this node, next node or don't retry at all.

All `RetryPolicies` implement the trait `RetryPolicy`:
```rust
pub trait RetryPolicy {
    fn decide_should_retry(&mut self, query_info: QueryInfo) -> RetryDecision;

    // Reset before using for the next query
    fn reset(&mut self);

    fn clone_boxed(&self) -> Box<dyn RetryPolicy + Send + Sync>;
}

/// Information about a failed query
pub struct QueryInfo<'a> {
    /// The error with which the query failed
    pub error: &'a QueryError,
    /// 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
    pub is_idempotent: bool,
    /// Consistency with which the query failed
    pub consistency: Consistency,
}

#[derive(Debug, PartialEq, Eq)]
pub enum RetryDecision {
    RetrySameNode,
    RetryNextNode,
    DontRetry,
}
```
There are two retry policies implemented:
* `FallthroughRetryPolicy` - never retries, retruns errors straight to the user
* `DefaultRetryPolicy` - policy used by default, behaviour based on [DataStax Java Driver](https://docs.datastax.com/en/developer/java-driver/4.10/manual/core/retries/)

The user can implement their own `RetryPolicy` by implementing this trait.
iIt is also possible to use a different `RetryPolicy` for a single query.

In [DataStax Java Driver](https://docs.datastax.com/en/developer/java-driver/4.10/manual/core/retries/) there is also a `ConsistencyDowngradingRetryPolicy` - a policy that tries to perform the query with specified consistency but if it fails because consistency requirements can't be met, it chooses a lower consistency so that the query can be finished.

[DataStax C/C++ Driver](https://docs.datastax.com/en/developer/cpp-driver/2.15/topics/configuration/retry_policies/) deprecated this policy and asks users not to use it in new applications. Their argumentation is that if the query would be valid with lower consistency then it shouldn't be preformed with higher consistency in the first place.

I didn't implement this `RetryPolicy` because I'm not sure if we want to have it at all.

I didn't add any new metrics, should there be something new? Do we need a number of retries or something else?

## Notes
When parsing `DBError` I assume that fields are initialized in the order in which they are specified.
I belive this is true, there are some tests in `rust` repository for struct evaluation order: ([1](https://github.com/rust-lang/rust/blob/master/src/test/ui/structs-enums/struct-order-of-eval-1.rs) [2](https://github.com/rust-lang/rust/blob/master/src/test/ui/structs-enums/struct-order-of-eval-2.rs) [3](https://github.com/rust-lang/rust/blob/master/src/test/ui/structs-enums/struct-order-of-eval-3.rs) [4](https://github.com/rust-lang/rust/blob/master/src/test/ui/structs-enums/struct-order-of-eval-4.rs))
If this is a concern I can change it to explicit separate parsing but the code will be longer and uglier.

## Fixed issues: 
Fixes: #41


## Pre-review 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.
- [x] I added relevant tests for new features and bug fixes.
- [x] All commits compile, pass static checks and pass test.
- [x] PR description sums up the changes and reasons why they should be introduced.
- [x] I added appropriate `Fixes:` annotations to PR description.


cvybhu: @psarna @piodul please review
piodul: As for the evaluation order of the struct fields, it was recently specified in the rust language reference:  https://github.com/rust-lang/reference/blob/master/src/expressions.md#evaluation-order-of-operands:

    Expressions taking multiple operands are evaluated left to right as written in the source code.


cvybhu: > 
> Maybe we could split the responsibility into a "factory" object and a "session" object? E.g. `RetryPolicy` and `RetrySession`. The latter would have the same interface as `RetryPolicy` has now, while the former would only be used to create session objects.
> 

I will try it out. This will probably be a little less efficient because with current design we could directly move Query's retry policy which required a single allocation. With this split design we would probably make Query keep `Box<dyn RetryPolicy>` which would be used to create `Box<dyn RetrySession>`. 
We could make it keep `Box<dyn RetrySession>` but this would be inconsistent with `Batch` and `Prepared` which are taken by reference and have to keep a `Box<dyn RetryPolicy>`. This seems a bit ugly API wise.

Maybe Query could be generic over `RetryPolicy` type - this would allow to remove most `Box`es I think.

I will try doing something like this and see how it turns out


cvybhu: Okay @piodul all fixed and ready
I split `RetryPolicy` into `RetryPolicy` and `RetrySession`:
```rust
/// Information about a failed query
pub struct QueryInfo<'a> {
    /// The error with which the query failed
    pub error: &'a QueryError,
    /// 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
    pub is_idempotent: bool,
    /// Consistency with which the query failed
    pub consistency: Consistency,
}

#[derive(Debug, PartialEq, Eq)]
pub enum RetryDecision {
    RetrySameNode,
    RetryNextNode,
    DontRetry,
}

/// Specifies a policy used to decide when to retry a query
pub trait RetryPolicy {
    /// Called for each new query, starts a session of deciding about retries
    fn new_session(&self) -> Box<dyn RetrySession + Send + Sync>;

    /// Used to clone this RetryPolicy
    fn clone_boxed(&self) -> Box<dyn RetryPolicy + Send + Sync>;
}

/// Used throughout a single query to decide when to retry it
/// After this query is finished it is destroyed or reset
pub trait RetrySession {
    /// Called after the query failed - decide what to do next
    fn decide_should_retry(&mut self, query_info: QueryInfo) -> RetryDecision;

    /// Reset before using for a new query
    fn reset(&mut self);
}
```

Patch:
diff --git a/.github/workflows/cassandra.yml b/.github/workflows/cassandra.yml
deleted file mode 100644
index 54e7a75c..00000000
--- a/.github/workflows/cassandra.yml
+++ /dev/null
@@ -1,28 +0,0 @@
-# This workflow runs all tests on cassandra to ensure full compatibility
-name: Cassandra tests
-
-on:
-  push:
-    branches: [ main ]
-  pull_request:
-    branches: [ main ]
-
-env:
-  CARGO_TERM_COLOR: always
-
-jobs:
-  build:
-    runs-on: ubuntu-latest
-    services:
-      cassandra:
-        image: cassandra
-        ports:
-          - 9042:9042
-    steps:
-    - uses: actions/checkout@v2
-      # A separate step for building to separate measuring time of compilation and testing
-    - name: Build the project
-      run: cargo build --verbose --tests
-    - name: Run tests on cassandra
-      # test threads must be one because else database tests will run in parallel and will result in flaky tests
-      run: cargo test --verbose -- --test-threads=1
diff --git a/scylla/src/frame/response/error.rs b/scylla/src/frame/response/error.rs
index a5f0ea02..8f195faf 100644
--- a/scylla/src/frame/response/error.rs
+++ b/scylla/src/frame/response/error.rs
@@ -1,10 +1,11 @@
 use crate::frame::frame_errors::ParseError;
 use crate::frame::types;
-use crate::transport::errors::{DBError, QueryError};
+use crate::transport::errors::{DBError, QueryError, WriteType};
+use byteorder::ReadBytesExt;
 
 #[derive(Debug)]
 pub struct Error {
-    pub code: i32,
+    pub error: DBError,
     pub reason: String,
 }
 
@@ -13,12 +14,301 @@ impl Error {
         let code = types::read_int(buf)?;
         let reason = types::read_string(buf)?.to_owned();
 
-        Ok(Error { code, reason })
+        let error: DBError = match code {
+            0x0000 => DBError::ServerError,
+            0x000A => DBError::ProtocolError,
+            0x0100 => DBError::AuthenticationError,
+            0x1000 => DBError::Unavailable {
+                consistency: types::read_consistency(buf)?,
+                required: types::read_int(buf)?,
+                alive: types::read_int(buf)?,
+            },
+            0x1001 => DBError::Overloaded,
+            0x1002 => DBError::IsBootstrapping,
+            0x1003 => DBError::TruncateError,
+            0x1100 => DBError::WriteTimeout {
+                consistency: types::read_consistency(buf)?,
+                received: types::read_int(buf)?,
+                required: types::read_int(buf)?,
+                write_type: WriteType::from(types::read_string(buf)?),
+            },
+            0x1200 => DBError::ReadTimeout {
+                consistency: types::read_consistency(buf)?,
+                received: types::read_int(buf)?,
+                required: types::read_int(buf)?,
+                data_present: buf.read_u8()? != 0,
+            },
+            0x1300 => DBError::ReadFailure {
+                consistency: types::read_consistency(buf)?,
+                received: types::read_int(buf)?,
+                required: types::read_int(buf)?,
+                numfailures: types::read_int(buf)?,
+                data_present: buf.read_u8()? != 0,
+            },
+            0x1400 => DBError::FunctionFailure {
+                keyspace: types::read_string(buf)?.to_string(),
+                function: types::read_string(buf)?.to_string(),
+                arg_types: types::read_string_list(buf)?,
+            },
+            0x1500 => DBError::WriteFailure {
+                consistency: types::read_consistency(buf)?,
+                received: types::read_int(buf)?,
+                required: types::read_int(buf)?,
+                numfailures: types::read_int(buf)?,
+                write_type: WriteType::from(types::read_string(buf)?),
+            },
+            0x2000 => DBError::SyntaxError,
+            0x2100 => DBError::Unauthorized,
+            0x2200 => DBError::Invalid,
+            0x2300 => DBError::ConfigError,
+            0x2400 => DBError::AlreadyExists {
+                keyspace: types::read_string(buf)?.to_string(),
+                table: types::read_string(buf)?.to_string(),
+            },
+            0x2500 => DBError::Unprepared,
+            _ => DBError::Other(code),
+        };
+
+        Ok(Error { error, reason })
     }
 }
 
 impl Into<QueryError> for Error {
     fn into(self) -> QueryError {
-        QueryError::DBError(DBError::ErrorMsg(self.code, self.reason))
+        QueryError::DBError(self.error, self.reason)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::Error;
+    use crate::statement::Consistency;
+    use crate::transport::errors::{DBError, WriteType};
+    use std::convert::TryInto;
+
+    // Serializes the beginning of an ERROR response - error code and message
+    // All custom data depending on the error type is appended after these bytes
+    fn make_error_request_bytes(error_code: i32, message: &str) -> Vec<u8> {
+        let mut bytes: Vec<u8> = Vec::new();
+        let message_len: u16 = message.len().try_into().unwrap();
+
+        bytes.extend(&error_code.to_be_bytes());
+        bytes.extend(&message_len.to_be_bytes());
+        bytes.extend(message.as_bytes());
+
+        bytes
+    }
+
+    // Tests deserialization of all errors without and additional data
+    #[test]
+    fn deserialize_simple_errors() {
+        let simple_error_mappings: [(i32, DBError); 12] = [
+            (0x0000, DBError::ServerError),
+            (0x000A, DBError::ProtocolError),
+            (0x0100, DBError::AuthenticationError),
+            (0x1001, DBError::Overloaded),
+            (0x1002, DBError::IsBootstrapping),
+            (0x1003, DBError::TruncateError),
+            (0x2000, DBError::SyntaxError),
+            (0x2100, DBError::Unauthorized),
+            (0x2200, DBError::Invalid),
+            (0x2300, DBError::ConfigError),
+            (0x2500, DBError::Unprepared),
+            (0x1234, DBError::Other(0x1234)),
+        ];
+
+        for (error_code, expected_error) in &simple_error_mappings {
+            let bytes: Vec<u8> = make_error_request_bytes(*error_code, "simple message");
+            let error: Error = Error::deserialize(&mut bytes.as_slice()).unwrap();
+            assert_eq!(error.error, *expected_error);
+            assert_eq!(error.reason, "simple message");
+        }
+    }
+
+    #[test]
+    fn deserialize_unavailable() {
+        let mut bytes = make_error_request_bytes(0x1000, "message 2");
+        bytes.extend(&1_i16.to_be_bytes());
+        bytes.extend(&2_i32.to_be_bytes());
+        bytes.extend(&3_i32.to_be_bytes());
+
+        let error: Error = Error::deserialize(&mut bytes.as_slice()).unwrap();
+
+        assert_eq!(
+            error.error,
+            DBError::Unavailable {
+                consistency: Consistency::One,
+                required: 2,
+                alive: 3,
+            }
+        );
+        assert_eq!(error.reason, "message 2");
+    }
+
+    #[test]
+    fn deserialize_write_timeout() {
+        let mut bytes = make_error_request_bytes(0x1100, "message 2");
+        bytes.extend(&0x0004_i16.to_be_bytes());
+        bytes.extend(&(-5_i32).to_be_bytes());
+        bytes.extend(&100_i32.to_be_bytes());
+
+        let write_type_str = "SIMPLE";
+        let write_type_str_len: u16 = write_type_str.len().try_into().unwrap();
+        bytes.extend(&write_type_str_len.to_be_bytes());
+        bytes.extend(write_type_str.as_bytes());
+
+        let error: Error = Error::deserialize(&mut bytes.as_slice()).unwrap();
+
+        assert_eq!(
+            error.error,
+            DBError::WriteTimeout {
+                consistency: Consistency::Quorum,
+                received: -5, // Allow negative values when they don't make sense, it's better than crashing with ProtocolError
+                required: 100,
+                write_type: WriteType::Simple,
+            }
+        );
+        assert_eq!(error.reason, "message 2");
+    }
+
+    #[test]
+    fn deserialize_read_timeout() {
+        let mut bytes = make_error_request_bytes(0x1200, "message 2");
+        bytes.extend(&0x0002_i16.to_be_bytes());
+        bytes.extend(&8_i32.to_be_bytes());
+        bytes.extend(&32_i32.to_be_bytes());
+        bytes.push(0_u8);
+
+        let error: Error = Error::deserialize(&mut bytes.as_slice()).unwrap();
+
+        assert_eq!(
+            error.error,
+            DBError::ReadTimeout {
+                consistency: Consistency::Two,
+                received: 8,
+                required: 32,
+                data_present: false,
+            }
+        );
+        assert_eq!(error.reason, "message 2");
+    }
+
+    #[test]
+    fn deserialize_read_failure() {
+        let mut bytes = make_error_request_bytes(0x1300, "message 2");
+        bytes.extend(&0x0003_i16.to_be_bytes());
+        bytes.extend(&4_i32.to_be_bytes());
+        bytes.extend(&5_i32.to_be_bytes());
+        bytes.extend(&6_i32.to_be_bytes());
+        bytes.push(123_u8); // Any non-zero value means data_present is true
+
+        let error: Error = Error::deserialize(&mut bytes.as_slice()).unwrap();
+
+        assert_eq!(
+            error.error,
+            DBError::ReadFailure {
+                consistency: Consistency::Three,
+                received: 4,
+                required: 5,
+                numfailures: 6,
+                data_present: true,
+            }
+        );
+        assert_eq!(error.reason, "message 2");
+    }
+
+    #[test]
+    fn deserialize_function_failure() {
+        let mut bytes = make_error_request_bytes(0x1400, "message 2");
+
+        let keyspace_name: &str = "keyspace_name";
+        let keyspace_name_len: u16 = keyspace_name.len().try_into().unwrap();
+
+        let function_name: &str = "function_name";
+        let function_name_len: u16 = function_name.len().try_into().unwrap();
+
+        let type1: &str = "type1";
+        let type1_len: u16 = type1.len().try_into().unwrap();
+
+        let type2: &str = "type2";
+        let type2_len: u16 = type1.len().try_into().unwrap();
+
+        bytes.extend(&keyspace_name_len.to_be_bytes());
+        bytes.extend(keyspace_name.as_bytes());
+        bytes.extend(&function_name_len.to_be_bytes());
+        bytes.extend(function_name.as_bytes());
+        bytes.extend(&2_i16.to_be_bytes());
+        bytes.extend(&type1_len.to_be_bytes());
+        bytes.extend(type1.as_bytes());
+        bytes.extend(&type2_len.to_be_bytes());
+        bytes.extend(type2.as_bytes());
+
+        let error: Error = Error::deserialize(&mut bytes.as_slice()).unwrap();
+
+        assert_eq!(
+            error.error,
+            DBError::FunctionFailure {
+                keyspace: "keyspace_name".to_string(),
+                function: "function_name".to_string(),
+                arg_types: vec!["type1".to_string(), "type2".to_string()]
+            }
+        );
+        assert_eq!(error.reason, "message 2");
+    }
+
+    #[test]
+    fn deserialize_write_failure() {
+        let mut bytes = make_error_request_bytes(0x1500, "message 2");
+
+        bytes.extend(&0x0000_i16.to_be_bytes());
+        bytes.extend(&2_i32.to_be_bytes());
+        bytes.extend(&4_i32.to_be_bytes());
+        bytes.extend(&8_i32.to_be_bytes());
+
+        let write_type_str = "COUNTER";
+        let write_type_str_len: u16 = write_type_str.len().try_into().unwrap();
+        bytes.extend(&write_type_str_len.to_be_bytes());
+        bytes.extend(write_type_str.as_bytes());
+
+        let error: Error = Error::deserialize(&mut bytes.as_slice()).unwrap();
+
+        assert_eq!(
+            error.error,
+            DBError::WriteFailure {
+                consistency: Consistency::Any,
+                received: 2,
+                required: 4,
+                numfailures: 8,
+                write_type: WriteType::Counter,
+            }
+        );
+        assert_eq!(error.reason, "message 2");
+    }
+
+    #[test]
+    fn deserialize_already_exists() {
+        let mut bytes = make_error_request_bytes(0x2400, "message 2");
+
+        let keyspace_name: &str = "keyspace_name";
+        let keyspace_name_len: u16 = keyspace_name.len().try_into().unwrap();
+
+        let table_name: &str = "table_name";
+        let table_name_len: u16 = table_name.len().try_into().unwrap();
+
+        bytes.extend(&keyspace_name_len.to_be_bytes());
+        bytes.extend(keyspace_name.as_bytes());
+        bytes.extend(&table_name_len.to_be_bytes());
+        bytes.extend(table_name.as_bytes());
+
+        let error: Error = Error::deserialize(&mut bytes.as_slice()).unwrap();
+
+        assert_eq!(
+            error.error,
+            DBError::AlreadyExists {
+                keyspace: "keyspace_name".to_string(),
+                table: "table_name".to_string(),
+            }
+        );
+        assert_eq!(error.reason, "message 2");
     }
 }
diff --git a/scylla/src/frame/types.rs b/scylla/src/frame/types.rs
index 0463b6ec..04be5144 100644
--- a/scylla/src/frame/types.rs
+++ b/scylla/src/frame/types.rs
@@ -32,6 +32,12 @@ impl Default for Consistency {
     }
 }
 
+impl std::fmt::Display for Consistency {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{:?}", self)
+    }
+}
+
 impl From<std::num::TryFromIntError> for ParseError {
     fn from(_err: std::num::TryFromIntError) -> Self {
         ParseError::BadData("Integer conversion out of range".to_string())
diff --git a/scylla/src/statement/batch.rs b/scylla/src/statement/batch.rs
index 03b37dcb..1b0c3780 100644
--- a/scylla/src/statement/batch.rs
+++ b/scylla/src/statement/batch.rs
@@ -2,16 +2,18 @@ use crate::statement::{prepared_statement::PreparedStatement, query::Query};
 
 pub use super::Consistency;
 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 {
     statements: Vec<BatchStatement>,
     batch_type: BatchType,
-    consistency: Consistency,
-    serial_consistency: Option<Consistency>,
+    pub consistency: Consistency,
+    pub serial_consistency: Option<Consistency>,
+    pub is_idempotent: bool,
+    pub retry_policy: Option<Box<dyn RetryPolicy + Send + Sync>>,
 }
 
 impl Batch {
@@ -59,6 +61,31 @@ impl Batch {
     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
+    }
 }
 
 impl Default for Batch {
@@ -68,6 +95,8 @@ impl Default for Batch {
             batch_type: BatchType::Logged,
             consistency: Default::default(),
             serial_consistency: None,
+            is_idempotent: false,
+            retry_policy: None,
         }
     }
 }
@@ -96,3 +125,19 @@ 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()),
+        }
+    }
+}
diff --git a/scylla/src/statement/prepared_statement.rs b/scylla/src/statement/prepared_statement.rs
index 7bb4794a..07004786 100644
--- a/scylla/src/statement/prepared_statement.rs
+++ b/scylla/src/statement/prepared_statement.rs
@@ -1,19 +1,21 @@
 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;
 
 /// Represents a statement prepared on the server.
-#[derive(Debug, Clone)]
 pub struct PreparedStatement {
     id: Bytes,
     metadata: PreparedMetadata,
     statement: String,
     page_size: Option<i32>,
-    consistency: Consistency,
-    serial_consistency: Option<Consistency>,
+    pub consistency: Consistency,
+    pub serial_consistency: Option<Consistency>,
+    pub is_idempotent: bool,
+    pub retry_policy: Option<Box<dyn RetryPolicy + Send + Sync>>,
 }
 
 impl PreparedStatement {
@@ -25,6 +27,8 @@ impl PreparedStatement {
             page_size: None,
             consistency: Default::default(),
             serial_consistency: None,
+            is_idempotent: false,
+            retry_policy: None,
         }
     }
 
@@ -74,6 +78,31 @@ impl PreparedStatement {
         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
+    }
+
     /// Computes the partition key of the target table from given values
     /// Partition keys have a specific serialization rules.
     /// Ref: https://github.com/scylladb/scylla/blob/40adf38915b6d8f5314c621a94d694d172360833/compound_compat.hh#L33-L47
@@ -145,3 +174,21 @@ 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(),
+            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()),
+        }
+    }
+}
diff --git a/scylla/src/statement/query.rs b/scylla/src/statement/query.rs
index 4eb9dea0..97ba6581 100644
--- a/scylla/src/statement/query.rs
+++ b/scylla/src/statement/query.rs
@@ -1,14 +1,16 @@
 use super::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 {
     contents: String,
     page_size: Option<i32>,
-    consistency: Consistency,
-    serial_consistency: Option<Consistency>,
+    pub consistency: Consistency,
+    pub serial_consistency: Option<Consistency>,
+    pub is_idempotent: bool,
+    pub retry_policy: Option<Box<dyn RetryPolicy + Send + Sync>>,
 }
 
 impl Query {
@@ -19,6 +21,8 @@ impl Query {
             page_size: None,
             consistency: Default::default(),
             serial_consistency: None,
+            is_idempotent: false,
+            retry_policy: None,
         }
     }
 
@@ -64,6 +68,31 @@ impl Query {
     pub fn get_serial_consistency(&self) -> Option<Consistency> {
         self.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  
+    /// 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
+    }
 }
 
 impl From<String> for Query {
@@ -77,3 +106,19 @@ 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()),
+        }
+    }
+}
diff --git a/scylla/src/transport/connection.rs b/scylla/src/transport/connection.rs
index d39d6640..bd15df0c 100644
--- a/scylla/src/transport/connection.rs
+++ b/scylla/src/transport/connection.rs
@@ -15,7 +15,7 @@ use std::{
     net::{Ipv4Addr, Ipv6Addr},
 };
 
-use super::errors::{BadKeyspaceName, BadQuery, QueryError};
+use super::errors::{BadKeyspaceName, BadQuery, DBError, QueryError};
 
 use crate::batch::{Batch, BatchStatement};
 use crate::frame::{
@@ -153,7 +153,16 @@ impl Connection {
         query: impl Into<Query>,
         values: impl ValueList,
     ) -> Result<Option<Vec<result::Row>>, QueryError> {
-        let result = self.query(&query.into(), values, None).await?;
+        let query: Query = query.into();
+        self.query_single_page_by_ref(&query, &values).await
+    }
+
+    pub async fn query_single_page_by_ref(
+        &self,
+        query: &Query,
+        values: &impl ValueList,
+    ) -> Result<Option<Vec<result::Row>>, QueryError> {
+        let result = self.query(query, values, None).await?;
         match result {
             Response::Error(err) => Err(err.into()),
             Response::Result(result::Result::Rows(rs)) => Ok(Some(rs.rows)),
@@ -186,6 +195,23 @@ impl Connection {
         self.send_request(&query_frame, true).await
     }
 
+    pub async fn execute_single_page(
+        &self,
+        prepared_statement: &PreparedStatement,
+        values: impl ValueList,
+    ) -> Result<Option<Vec<result::Row>>, QueryError> {
+        let response = self.execute(prepared_statement, values, None).await?;
+
+        match response {
+            Response::Error(err) => Err(err.into()),
+            Response::Result(result::Result::Rows(rs)) => Ok(Some(rs.rows)),
+            Response::Result(_) => Ok(None),
+            _ => Err(QueryError::ProtocolError(
+                "EXECUTE: Unexpected server response",
+            )),
+        }
+    }
+
     pub async fn execute(
         &self,
         prepared_statement: &PreparedStatement,
@@ -205,14 +231,28 @@ impl Connection {
             },
         };
 
-        self.send_request(&execute_frame, true).await
+        let response = self.send_request(&execute_frame, true).await?;
+
+        if let Response::Error(err) = &response {
+            if err.error == DBError::Unprepared {
+                // Repreparation of a statement is needed
+                let reprepared = self.prepare(prepared_statement.get_statement()).await?;
+                // Reprepared statement should keep its id - it's the md5 sum
+                // of statement contents
+                if reprepared.get_id() != prepared_statement.get_id() {
+                    return Err(QueryError::ProtocolError(
+                        "Prepared statement Id changed, md5 sum should stay the same",
+                    ));
+                }
+
+                return self.send_request(&execute_frame, true).await;
+            }
+        }
+
+        Ok(response)
     }
 
-    pub async fn batch(
-        &self,
-        batch: &Batch,
-        values: impl BatchValues,
-    ) -> Result<Response, QueryError> {
+    pub async fn batch(&self, batch: &Batch, values: impl BatchValues) -> Result<(), QueryError> {
         let statements_count = batch.get_statements().len();
         if statements_count != values.len() {
             return Err(QueryError::BadQuery(BadQuery::ValueLenMismatch(
@@ -239,7 +279,15 @@ impl Connection {
             serial_consistency: batch.get_serial_consistency(),
         };
 
-        self.send_request(&batch_frame, true).await
+        let response = self.send_request(&batch_frame, true).await?;
+
+        match response {
+            Response::Error(err) => Err(err.into()),
+            Response::Result(_) => Ok(()),
+            _ => Err(QueryError::ProtocolError(
+                "BATCH: Unexpected server response",
+            )),
+        }
     }
 
     pub async fn use_keyspace(
diff --git a/scylla/src/transport/errors.rs b/scylla/src/transport/errors.rs
index 1d20f999..4fc1fec1 100644
--- a/scylla/src/transport/errors.rs
+++ b/scylla/src/transport/errors.rs
@@ -1,5 +1,6 @@
 use crate::frame::frame_errors::{FrameError, ParseError};
 use crate::frame::value::SerializeValuesError;
+use crate::statement::Consistency;
 use std::io::ErrorKind;
 use std::sync::Arc;
 use thiserror::Error;
@@ -7,9 +8,9 @@ use thiserror::Error;
 /// Error that occured during query execution
 #[derive(Error, Debug, Clone)]
 pub enum QueryError {
-    /// Database sent a response containing some error
-    #[error(transparent)]
-    DBError(#[from] DBError),
+    /// Database sent a response containing some error with a message
+    #[error("Database returned an error: {0}, Error message: {1}")]
+    DBError(DBError, String),
 
     /// Caller passed an invalid query
     #[error(transparent)]
@@ -24,15 +25,189 @@ pub enum QueryError {
     ProtocolError(&'static str),
 }
 
-/// An error sent from database in response to a query
-#[derive(Error, Debug, Clone)]
-#[error("Database response contains an error")]
+/// An error sent from the database in response to a query
+/// as described in the [specification](https://github.com/apache/cassandra/blob/5ed5e84613ef0e9664a774493db7d2604e3596e0/doc/native_protocol_v4.spec#L1029)  
+#[derive(Error, Debug, Clone, PartialEq, Eq)]
 pub enum DBError {
-    // TODO develop this while implementing retry policies
-    // to react differently to each error.
-    // Make it fit what we encounter.
-    #[error("Response is an error message: code: {0} message: {1}")]
-    ErrorMsg(i32, String),
+    /// The submitted query has a syntax error
+    #[error("The submitted query has a syntax error")]
+    SyntaxError,
+
+    /// The query is syntatically correct but invalid
+    #[error("The query is syntatically correct but invalid")]
+    Invalid,
+
+    /// Attempted to create a keyspace or a table that was already existing
+    #[error(
+        "Attempted to create a keyspace or a table that was already existing \
+        (keyspace: {keyspace}, table: {table})"
+    )]
+    AlreadyExists {
+        /// Created keyspace name or name of the keyspace in which table was created
+        keyspace: String,
+        /// Name of the table created, in case of keyspace creation it's an empty string
+        table: String,
+    },
+
+    /// User defined function failed during execution
+    #[error(
+        "User defined function failed during execution \
+        (keyspace: {keyspace}, function: {function}, arg_types: {arg_types:?})"
+    )]
+    FunctionFailure {
+        /// Keyspace of the failed function
+        keyspace: String,
+        /// Name of the failed function
+        function: String,
+        /// Types of arguments passed to the function
+        arg_types: Vec<String>,
+    },
+
+    /// Authentication failed - bad credentials
+    #[error("Authentication failed - bad credentials")]
+    AuthenticationError,
+
+    /// The logged user doesn't have the right to perform the query
+    #[error("The logged user doesn't have the right to perform the query")]
+    Unauthorized,
+
+    /// The query is invalid because of some configuration issue
+    #[error("The query is invalid because of some configuration issue")]
+    ConfigError,
+
+    /// Not enough nodes are alive to satisfy required consistency level
+    #[error(
+        "Not enough nodes are alive to satisfy required consistency level \
+        (consistency: {consistency}, required: {required}, alive: {alive})"
+    )]
+    Unavailable {
+        /// Consistency level of the query
+        consistency: Consistency,
+        /// Number of nodes required to be alive to satisfy required consistency level
+        required: i32,
+        /// Found number of active nodes
+        alive: i32,
+    },
+
+    /// The request cannot be processed because the coordinator node is overloaded
+    #[error("The request cannot be processed because the coordinator node is overloaded")]
+    Overloaded,
+
+    /// The coordinator node is still bootstrapping
+    #[error("The coordinator node is still bootstrapping")]
+    IsBootstrapping,
+
+    /// Error during truncate operation
+    #[error("Error during truncate operation")]
+    TruncateError,
+
+    /// Not enough nodes responded to the read request in time to satisfy required consistency level
+    #[error("Not enough nodes responded to the read request in time to satisfy required consistency level \
+            (consistency: {consistency}, received: {received}, required: {required}, data_present: {data_present})")]
+    ReadTimeout {
+        /// Consistency level of the query
+        consistency: Consistency,
+        /// Number of nodes that responded to the read request
+        received: i32,
+        /// Number of nodes required to respond to satisfy required consistency level
+        required: i32,
+        /// Replica that was asked for data has responded
+        data_present: bool,
+    },
+
+    /// Not enough nodes responded to the write request in time to satisfy required consistency level
+    #[error("Not enough nodes responded to the write request in time to satisfy required consistency level \
+            (consistency: {consistency}, received: {received}, required: {required}, write_type: {write_type})")]
+    WriteTimeout {
+        /// Consistency level of the query
+        consistency: Consistency,
+        /// Number of nodes that responded to the write request
+        received: i32,
+        /// Number of nodes required to respond to satisfy required consistency level
+        required: i32,
+        /// Type of write operation requested
+        write_type: WriteType,
+    },
+
+    /// A non-timeout error during a read request
+    #[error(
+        "A non-timeout error during a read request \
+        (consistency: {consistency}, received: {received}, required: {required}, \
+        numfailures: {numfailures}, data_present: {data_present})"
+    )]
+    ReadFailure {
+        /// Consistency level of the query
+        consistency: Consistency,
+        /// Number of nodes that responded to the read request
+        received: i32,
+        /// Number of nodes required to respond to satisfy required consistency level
+        required: i32,
+        /// Number of nodes that experience a failure while executing the request
+        numfailures: i32,
+        /// Replica that was asked for data has responded
+        data_present: bool,
+    },
+
+    /// A non-timeout error during a write request
+    #[error(
+        "A non-timeout error during a write request \
+        (consistency: {consistency}, received: {received}, required: {required}, \
+        numfailures: {numfailures}, write_type: {write_type}"
+    )]
+    WriteFailure {
+        /// Consistency level of the query
+        consistency: Consistency,
+        /// Number of ndoes that responded to the read request
+        received: i32,
+        /// Number of nodes required to respond to satisfy required consistency level
+        required: i32,
+        /// Number of nodes that experience a failure while executing the request
+        numfailures: i32,
+        /// Type of write operation requested
+        write_type: WriteType,
+    },
+
+    /// Tried to execute a prepared statement that is not prepared. Driver shoud prepare it again
+    #[error(
+        "Tried to execute a prepared statement that is not prepared. Driver shoud prepare it again"
+    )]
+    Unprepared,
+
+    /// Internal server error. This indicates a server-side bug
+    #[error("Internal server error. This indicates a server-side bug")]
+    ServerError,
+
+    /// Invalid protocol message received from the driver
+    #[error("Invalid protocol message received from the driver")]
+    ProtocolError,
+
+    /// Other error code not specified in the specification
+    #[error("Other error not specified in the specification. Error code: {0}")]
+    Other(i32),
+}
+
+// Type of write operation requested
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum WriteType {
+    /// Non-batched non-counter write
+    Simple,
+    /// Logged batch write. If this type is received, it means the batch log has been succesfully written
+    /// (otherwise BatchLog type would be present)
+    Batch,
+    /// Unlogged batch. No batch log write has been attempted.
+    UnloggedBatch,
+    /// Counter write (batched or not)
+    Counter,
+    /// Timeout occured during the write to the batch log when a logged batch was requested
+    BatchLog,
+    /// Timeout occured during Compare And Set write/update
+    CAS,
+    /// Write involves VIEW update and failure to acquire local view(MV) lock for key within timeout
+    View,
+    /// Timeout occured  when a cdc_total_space_in_mb is exceeded when doing a write to data tracked by cdc
+    CDC,
+    /// Other type not specified in the specification
+    Other(String),
 }
 
 /// Error caused by caller creating an invalid query
@@ -68,9 +243,9 @@ pub enum NewSessionError {
     #[error("Empty known nodes list")]
     EmptyKnownNodesList,
 
-    /// Database sent a response containing some error
-    #[error(transparent)]
-    DBError(#[from] DBError),
+    /// Database sent a response containing some error with a message
+    #[error("Database returned an error: {0}, Error message: {1}")]
+    DBError(DBError, String),
 
     /// Caller passed an invalid query
     #[error(transparent)]
@@ -101,6 +276,12 @@ pub enum BadKeyspaceName {
     IllegalCharacter(String, char),
 }
 
+impl std::fmt::Display for WriteType {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{:?}", self)
+    }
+}
+
 impl From<std::io::Error> for QueryError {
     fn from(io_error: std::io::Error) -> QueryError {
         QueryError::IOError(Arc::new(io_error))
@@ -134,7 +315,7 @@ impl From<std::io::Error> for NewSessionError {
 impl From<QueryError> for NewSessionError {
     fn from(query_error: QueryError) -> NewSessionError {
         match query_error {
-            QueryError::DBError(e) => NewSessionError::DBError(e),
+            QueryError::DBError(e, msg) => NewSessionError::DBError(e, msg),
             QueryError::BadQuery(e) => NewSessionError::BadQuery(e),
             QueryError::IOError(e) => NewSessionError::IOError(e),
             QueryError::ProtocolError(m) => NewSessionError::ProtocolError(m),
@@ -162,3 +343,79 @@ impl QueryError {
         false
     }
 }
+
+impl From<&str> for WriteType {
+    fn from(write_type_str: &str) -> WriteType {
+        match write_type_str {
+            "SIMPLE" => WriteType::Simple,
+            "BATCH" => WriteType::Batch,
+            "UNLOGGED_BATCH" => WriteType::UnloggedBatch,
+            "COUNTER" => WriteType::Counter,
+            "BATCH_LOG" => WriteType::BatchLog,
+            "CAS" => WriteType::CAS,
+            "VIEW" => WriteType::View,
+            "CDC" => WriteType::CDC,
+            _ => WriteType::Other(write_type_str.to_string()),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{DBError, QueryError, WriteType};
+    use crate::statement::Consistency;
+
+    #[test]
+    fn write_type_from_str() {
+        let test_cases: [(&str, WriteType); 9] = [
+            ("SIMPLE", WriteType::Simple),
+            ("BATCH", WriteType::Batch),
+            ("UNLOGGED_BATCH", WriteType::UnloggedBatch),
+            ("COUNTER", WriteType::Counter),
+            ("BATCH_LOG", WriteType::BatchLog),
+            ("CAS", WriteType::CAS),
+            ("VIEW", WriteType::View),
+            ("CDC", WriteType::CDC),
+            ("SOMEOTHER", WriteType::Other("SOMEOTHER".to_string())),
+        ];
+
+        for (write_type_str, expected_write_type) in &test_cases {
+            let write_type = WriteType::from(*write_type_str);
+            assert_eq!(write_type, *expected_write_type);
+        }
+    }
+
+    // A test to check that displaying DBError and QueryError::DBError works as expected
+    // - displays error description
+    // - displays error parameters
+    // - displays error message
+    // - indented multiline strings dont cause whitespace gaps
+    #[test]
+    fn dberror_full_info() {
+        // Test that DBError::Unavailable is displayed correctly
+        let db_error = DBError::Unavailable {
+            consistency: Consistency::Three,
+            required: 3,
+            alive: 2,
+        };
+
+        let db_error_displayed: String = format!("{}", db_error);
+
+        let mut expected_dberr_msg =
+            "Not enough nodes are alive to satisfy required consistency level ".to_string();
+        expected_dberr_msg += "(consistency: Three, required: 3, alive: 2)";
+
+        assert_eq!(db_error_displayed, expected_dberr_msg);
+
+        // Test that QueryError::DBError::(DBError::Unavailable) is displayed correctly
+        let query_error =
+            QueryError::DBError(db_error, "a message about unavailable error".to_string());
+        let query_error_displayed: String = format!("{}", query_error);
+
+        let mut expected_querr_msg = "Database returned an error: ".to_string();
+        expected_querr_msg += &expected_dberr_msg;
+        expected_querr_msg += ", Error message: a message about unavailable error";
+
+        assert_eq!(query_error_displayed, expected_querr_msg);
+    }
+}
diff --git a/scylla/src/transport/iterator.rs b/scylla/src/transport/iterator.rs
index 8814b720..4d776c20 100644
--- a/scylla/src/transport/iterator.rs
+++ b/scylla/src/transport/iterator.rs
@@ -3,11 +3,10 @@ use std::mem;
 use std::pin::Pin;
 use std::sync::Arc;
 use std::task::{Context, Poll};
-use std::time::Instant;
 
 use bytes::Bytes;
 use futures::Stream;
-use std::result::Result as StdResult;
+use std::result::Result;
 use thiserror::Error;
 use tokio::sync::mpsc;
 
@@ -16,23 +15,30 @@ use crate::cql_to_rust::{FromRow, FromRowError};
 
 use crate::frame::{
     response::{
-        result::{Result, Row, Rows},
+        result,
+        result::{Row, Rows},
         Response,
     },
     value::SerializedValues,
 };
+use crate::routing::Token;
+use crate::statement::Consistency;
 use crate::statement::{prepared_statement::PreparedStatement, query::Query};
+use crate::transport::cluster::ClusterData;
 use crate::transport::connection::Connection;
+use crate::transport::load_balancing::{LoadBalancingPolicy, Statement};
 use crate::transport::metrics::Metrics;
+use crate::transport::node::Node;
+use crate::transport::retry_policy::{QueryInfo, RetryDecision, RetrySession};
 
 pub struct RowIterator {
     current_row_idx: usize,
     current_page: Rows,
-    page_receiver: mpsc::Receiver<StdResult<Rows, QueryError>>,
+    page_receiver: mpsc::Receiver<Result<Rows, QueryError>>,
 }
 
 impl Stream for RowIterator {
-    type Item = StdResult<Row, QueryError>;
+    type Item = Result<Row, QueryError>;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         let mut s = self.as_mut();
@@ -72,67 +78,96 @@ impl RowIterator {
     }
 
     pub(crate) fn new_for_query(
-        conn: Arc<Connection>,
         query: Query,
         values: SerializedValues,
+        retry_session: Box<dyn RetrySession + Send + Sync>,
+        load_balancer: Arc<dyn LoadBalancingPolicy>,
+        cluster_data: Arc<ClusterData>,
         metrics: Arc<Metrics>,
     ) -> RowIterator {
-        let metrics_copy = metrics.clone();
-        Self::new_with_worker(
-            |mut helper| async move {
-                let mut paging_state = None;
-                loop {
-                    let now = Instant::now();
-                    metrics_copy.inc_total_paged_queries();
-                    let rows = conn.query(&query, &values, paging_state).await;
-                    let _ = metrics_copy.log_query_latency(now.elapsed().as_millis() as u64);
-                    paging_state = helper.handle_response(rows).await;
-                    if paging_state.is_none() {
-                        break;
-                    }
-                }
-            },
-            metrics,
-        )
+        let (sender, receiver) = mpsc::channel(1);
+
+        let worker_task = async move {
+            let query_ref = &query;
+            let values_ref = &values;
+
+            let choose_connection = |node: Arc<Node>| async move { node.random_connection().await };
+
+            let page_query = |connection: Arc<Connection>, paging_state: Option<Bytes>| async move {
+                connection.query(query_ref, values_ref, paging_state).await
+            };
+
+            let worker = RowIteratorWorker {
+                sender,
+                choose_connection,
+                page_query,
+                statement_info: Statement::default(),
+                query_is_idempotent: query.is_idempotent,
+                query_consistency: query.consistency,
+                retry_session,
+                load_balancer,
+                metrics,
+                paging_state: None,
+            };
+
+            worker.work(cluster_data).await;
+        };
+
+        tokio::task::spawn(worker_task);
+
+        RowIterator {
+            current_row_idx: 0,
+            current_page: Default::default(),
+            page_receiver: receiver,
+        }
     }
 
     pub(crate) fn new_for_prepared_statement(
-        conn: Arc<Connection>,
-        prepared_statement: PreparedStatement,
+        prepared: PreparedStatement,
         values: SerializedValues,
+        token: Token,
+        retry_session: Box<dyn RetrySession + Send + Sync>,
+        load_balancer: Arc<dyn LoadBalancingPolicy>,
+        cluster_data: Arc<ClusterData>,
         metrics: Arc<Metrics>,
     ) -> RowIterator {
-        let metrics_copy = metrics.clone();
-        Self::new_with_worker(
-            |mut helper| async move {
-                let mut paging_state = None;
-                loop {
-                    let now = Instant::now();
-                    metrics_copy.inc_total_paged_queries();
-                    let rows = conn
-                        .execute(&prepared_statement, &values, paging_state)
-                        .await;
-                    let _ = metrics_copy.log_query_latency(now.elapsed().as_millis() as u64);
-                    paging_state = helper.handle_response(rows).await;
-                    if paging_state.is_none() {
-                        break;
-                    }
-                }
-            },
-            metrics,
-        )
-    }
-
-    fn new_with_worker<F, G>(worker: F, metrics: Arc<Metrics>) -> RowIterator
-    where
-        F: FnOnce(WorkerHelper) -> G,
-        G: Future<Output = ()> + Send + 'static,
-    {
-        // TODO: How many pages in flight do we allow?
         let (sender, receiver) = mpsc::channel(1);
-        let helper = WorkerHelper::new(sender, metrics);
 
-        tokio::task::spawn(worker(helper));
+        let statement_info = Statement {
+            token: Some(token),
+            keyspace: None,
+        };
+
+        let worker_task = async move {
+            let prepared_ref = &prepared;
+            let values_ref = &values;
+
+            let choose_connection =
+                |node: Arc<Node>| async move { node.connection_for_token(token).await };
+
+            let page_query = |connection: Arc<Connection>, paging_state: Option<Bytes>| async move {
+                connection
+                    .execute(prepared_ref, values_ref, paging_state)
+                    .await
+            };
+
+            let worker = RowIteratorWorker {
+                sender,
+                choose_connection,
+                page_query,
+                statement_info,
+                query_is_idempotent: prepared.is_idempotent,
+                query_consistency: prepared.consistency,
+                retry_session,
+                load_balancer,
+                metrics,
+                paging_state: None,
+            };
+
+            worker.work(cluster_data).await;
+        };
+
+        tokio::task::spawn(worker_task);
 
         RowIterator {
             current_row_idx: 0,
@@ -145,49 +180,132 @@ impl RowIterator {
         self.current_row_idx >= self.current_page.rows.len()
     }
 }
-struct WorkerHelper {
-    sender: mpsc::Sender<StdResult<Rows, QueryError>>,
+
+// RowIteratorWorker works in the background to fetch pages
+// RowIterator receives them through a channel
+struct RowIteratorWorker<'a, ConnFunc, QueryFunc> {
+    sender: mpsc::Sender<Result<Rows, QueryError>>,
+
+    // Closure used to choose a connection from a node
+    // AsyncFn(Arc<Node>) -> Result<Arc<Connection>, QueryError>
+    choose_connection: ConnFunc,
+
+    // Closure used to perform a single page query
+    // AsyncFn(Arc<Connection>, Option<Bytes>) -> Result<Response, QueryError>
+    page_query: QueryFunc,
+
+    statement_info: Statement<'a>,
+    query_is_idempotent: bool,
+    query_consistency: Consistency,
+
+    retry_session: Box<dyn RetrySession + Send + Sync>,
+    load_balancer: Arc<dyn LoadBalancingPolicy>,
     metrics: Arc<Metrics>,
+
+    paging_state: Option<Bytes>,
 }
 
-impl WorkerHelper {
-    fn new(sender: mpsc::Sender<StdResult<Rows, QueryError>>, metrics: Arc<Metrics>) -> Self {
-        Self { sender, metrics }
-    }
+impl<ConnFunc, ConnFut, QueryFunc, QueryFut> RowIteratorWorker<'_, ConnFunc, QueryFunc>
+where
+    ConnFunc: Fn(Arc<Node>) -> ConnFut,
+    ConnFut: Future<Output = Result<Arc<Connection>, QueryError>>,
+    QueryFunc: Fn(Arc<Connection>, Option<Bytes>) -> QueryFut,
+    QueryFut: Future<Output = Result<Response, QueryError>>,
+{
+    async fn work(mut self, cluster_data: Arc<ClusterData>) {
+        let query_plan = self.load_balancer.plan(&self.statement_info, &cluster_data);
 
-    async fn handle_response(
-        &mut self,
-        response: StdResult<Response, QueryError>,
-    ) -> Option<Bytes> {
-        match response {
-            Ok(Response::Result(Result::Rows(rows))) => {
-                let paging_state = rows.metadata.paging_state.clone();
-                if self.sender.send(Ok(rows)).await.is_err() {
-                    // TODO: Log error
-                    None
-                } else {
-                    paging_state
+        let mut last_error: QueryError =
+            QueryError::ProtocolError("Empty query plan - driver bug!");
+
+        'nodes_in_plan: for node in query_plan {
+            // For each node in the plan choose a connection to use
+            // This connection will be reused for same node retries to preserve paging cache on the shard
+            let connection: Arc<Connection> = match (self.choose_connection)(node).await {
+                Ok(connection) => connection,
+                Err(e) => {
+                    last_error = e;
+                    // Broken connection doesn't count as a failed query, don't log in metrics
+                    continue 'nodes_in_plan;
                 }
+            };
+
+            'same_node_retries: loop {
+                // Query pages until an error occurs
+                let queries_result: Result<(), QueryError> = self.query_pages(&connection).await;
+
+                last_error = match queries_result {
+                    Ok(()) => return,
+                    Err(error) => error,
+                };
+
+                // Use retry policy to decide what to do next
+                let query_info = QueryInfo {
+                    error: &last_error,
+                    is_idempotent: self.query_is_idempotent,
+                    consistency: self.query_consistency,
+                };
+
+                match self.retry_session.decide_should_retry(query_info) {
+                    RetryDecision::RetrySameNode => {
+                        self.metrics.inc_retries_num();
+                        continue 'same_node_retries;
+                    }
+                    RetryDecision::RetryNextNode => {
+                        self.metrics.inc_retries_num();
+                        continue 'nodes_in_plan;
+                    }
+                    RetryDecision::DontRetry => break 'nodes_in_plan,
+                };
             }
-            Ok(Response::Error(err)) => {
-                self.metrics.inc_failed_paged_queries();
-                let _ = self.sender.send(Err(err.into())).await;
-                None
-            }
-            Ok(_) => {
-                self.metrics.inc_failed_paged_queries();
-                let _ = self
-                    .sender
-                    .send(Err(QueryError::ProtocolError(
+        }
+
+        // Send last_error to RowIterator - query failed fully
+        let _ = self.sender.send(Err(last_error)).await;
+    }
+
+    // Given a working connection query as many pages as possible until the first error
+    async fn query_pages(&mut self, connection: &Arc<Connection>) -> Result<(), QueryError> {
+        loop {
+            self.metrics.inc_total_paged_queries();
+            let query_start = std::time::Instant::now();
+
+            let response: Response =
+                (self.page_query)(connection.clone(), self.paging_state.clone()).await?;
+
+            match response {
+                Response::Result(result::Result::Rows(mut rows)) => {
+                    let _ = self
+                        .metrics
+                        .log_query_latency(query_start.elapsed().as_millis() as u64);
+
+                    self.paging_state = rows.metadata.paging_state.take();
+
+                    // Send next page to RowIterator
+                    if self.sender.send(Ok(rows)).await.is_err() {
+                        // channel was closed, RowIterator was dropped - should shutdown
+                        return Ok(());
+                    }
+
+                    if self.paging_state.is_none() {
+                        // Reached the last query, shutdown
+                        return Ok(());
+                    }
+
+                    // Query succeded, reset retry policy for future retries
+                    self.retry_session.reset();
+                }
+                Response::Error(err) => {
+                    self.metrics.inc_failed_paged_queries();
+                    return Err(err.into());
+                }
+                _ => {
+                    self.metrics.inc_failed_paged_queries();
+
+                    return Err(QueryError::ProtocolError(
                         "Unexpected response to next page query",
-                    )))
-                    .await;
-                None
-            }
-            Err(err) => {
-                self.metrics.inc_failed_paged_queries();
-                let _ = self.sender.send(Err(err)).await;
-                None
+                    ));
+                }
             }
         }
     }
@@ -211,12 +329,12 @@ pub enum NextRowError {
 }
 
 impl<RowT: FromRow> Stream for TypedRowIterator<RowT> {
-    type Item = StdResult<RowT, NextRowError>;
+    type Item = Result<RowT, NextRowError>;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         let mut s = self.as_mut();
 
-        let next_elem: Option<StdResult<Row, QueryError>> =
+        let next_elem: Option<Result<Row, QueryError>> =
             match Pin::new(&mut s.row_iterator).poll_next(cx) {
                 Poll::Ready(next_elem) => next_elem,
                 Poll::Pending => return Poll::Pending,
diff --git a/scylla/src/transport/load_balancing/dc_aware_round_robin.rs b/scylla/src/transport/load_balancing/dc_aware_round_robin.rs
index e0652253..bd1e5b35 100644
--- a/scylla/src/transport/load_balancing/dc_aware_round_robin.rs
+++ b/scylla/src/transport/load_balancing/dc_aware_round_robin.rs
@@ -55,7 +55,7 @@ impl LoadBalancingPolicy for DCAwareRoundRobinPolicy {
         &self,
         _statement: &Statement,
         cluster: &'a ClusterData,
-    ) -> Box<dyn Iterator<Item = Arc<Node>> + 'a> {
+    ) -> Box<dyn Iterator<Item = Arc<Node>> + Send + Sync + 'a> {
         let index = self.index.fetch_add(1, ORDER_TYPE);
 
         let local_nodes = self.retrieve_local_nodes(cluster);
@@ -78,7 +78,10 @@ impl LoadBalancingPolicy for DCAwareRoundRobinPolicy {
 }
 
 impl ChildLoadBalancingPolicy for DCAwareRoundRobinPolicy {
-    fn apply_child_policy(&self, plan: Vec<Arc<Node>>) -> Box<dyn Iterator<Item = Arc<Node>>> {
+    fn apply_child_policy(
+        &self,
+        plan: Vec<Arc<Node>>,
+    ) -> Box<dyn Iterator<Item = Arc<Node>> + Send + Sync> {
         let index = self.index.fetch_add(1, ORDER_TYPE);
 
         let (local_nodes, remote_nodes): (Vec<_>, Vec<_>) = plan
diff --git a/scylla/src/transport/load_balancing/mod.rs b/scylla/src/transport/load_balancing/mod.rs
index 306e4814..7e9d0921 100644
--- a/scylla/src/transport/load_balancing/mod.rs
+++ b/scylla/src/transport/load_balancing/mod.rs
@@ -24,7 +24,7 @@ pub trait LoadBalancingPolicy: Send + Sync {
         &self,
         statement: &Statement,
         cluster: &'a ClusterData,
-    ) -> Box<dyn Iterator<Item = Arc<Node>> + 'a>;
+    ) -> Box<dyn Iterator<Item = Arc<Node>> + Send + Sync + 'a>;
 
     /// Returns name of load balancing policy
     fn name(&self) -> String;
@@ -34,7 +34,10 @@ pub trait LoadBalancingPolicy: Send + Sync {
 ///
 /// For example, this enables RoundRobinPolicy to process plan made by TokenAwarePolicy.
 pub trait ChildLoadBalancingPolicy: LoadBalancingPolicy {
-    fn apply_child_policy(&self, plan: Vec<Arc<Node>>) -> Box<dyn Iterator<Item = Arc<Node>>>;
+    fn apply_child_policy(
+        &self,
+        plan: Vec<Arc<Node>>,
+    ) -> Box<dyn Iterator<Item = Arc<Node>> + Send + Sync>;
 }
 
 // Does safe modulo
@@ -63,6 +66,15 @@ fn slice_rotated_left<'a, T>(slice: &'a [T], mid: usize) -> impl Iterator<Item =
     begin.iter().chain(end.iter())
 }
 
+impl Default for Statement<'_> {
+    fn default() -> Self {
+        Statement {
+            token: None,
+            keyspace: None,
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
diff --git a/scylla/src/transport/load_balancing/round_robin.rs b/scylla/src/transport/load_balancing/round_robin.rs
index 77f03496..002c5b42 100644
--- a/scylla/src/transport/load_balancing/round_robin.rs
+++ b/scylla/src/transport/load_balancing/round_robin.rs
@@ -32,7 +32,7 @@ impl LoadBalancingPolicy for RoundRobinPolicy {
         &self,
         _statement: &Statement,
         cluster: &'a ClusterData,
-    ) -> Box<dyn Iterator<Item = Arc<Node>> + 'a> {
+    ) -> Box<dyn Iterator<Item = Arc<Node>> + Send + Sync + 'a> {
         let index = self.index.fetch_add(1, ORDER_TYPE);
 
         let nodes_count = cluster.all_nodes.len();
@@ -48,7 +48,10 @@ impl LoadBalancingPolicy for RoundRobinPolicy {
 }
 
 impl ChildLoadBalancingPolicy for RoundRobinPolicy {
-    fn apply_child_policy(&self, mut plan: Vec<Arc<Node>>) -> Box<dyn Iterator<Item = Arc<Node>>> {
+    fn apply_child_policy(
+        &self,
+        mut plan: Vec<Arc<Node>>,
+    ) -> Box<dyn Iterator<Item = Arc<Node>> + Send + Sync> {
         let index = self.index.fetch_add(1, ORDER_TYPE);
 
         let len = plan.len(); // borrow checker forces making such a variable
diff --git a/scylla/src/transport/load_balancing/token_aware.rs b/scylla/src/transport/load_balancing/token_aware.rs
index e6c820f0..3993b5e2 100644
--- a/scylla/src/transport/load_balancing/token_aware.rs
+++ b/scylla/src/transport/load_balancing/token_aware.rs
@@ -107,7 +107,7 @@ impl LoadBalancingPolicy for TokenAwarePolicy {
         &self,
         statement: &Statement,
         cluster: &'a ClusterData,
-    ) -> Box<dyn Iterator<Item = Arc<Node>> + 'a> {
+    ) -> Box<dyn Iterator<Item = Arc<Node>> + Send + Sync + 'a> {
         match statement.token {
             Some(token) => {
                 let keyspace = statement.keyspace.and_then(|k| cluster.keyspaces.get(k));
@@ -407,7 +407,7 @@ mod tests {
             &self,
             _: &Statement,
             _: &'a ClusterData,
-        ) -> Box<dyn Iterator<Item = Arc<Node>> + 'a> {
+        ) -> Box<dyn Iterator<Item = Arc<Node>> + Send + Sync + 'a> {
             let empty_node_list: Vec<Arc<Node>> = Vec::new();
 
             Box::new(empty_node_list.into_iter())
@@ -419,7 +419,10 @@ mod tests {
     }
 
     impl ChildLoadBalancingPolicy for DumbPolicy {
-        fn apply_child_policy(&self, plan: Vec<Arc<Node>>) -> Box<dyn Iterator<Item = Arc<Node>>> {
+        fn apply_child_policy(
+            &self,
+            plan: Vec<Arc<Node>>,
+        ) -> Box<dyn Iterator<Item = Arc<Node>> + Send + Sync> {
             Box::new(plan.into_iter())
         }
     }
diff --git a/scylla/src/transport/metrics.rs b/scylla/src/transport/metrics.rs
index c8684884..8dc75b0c 100644
--- a/scylla/src/transport/metrics.rs
+++ b/scylla/src/transport/metrics.rs
@@ -34,6 +34,7 @@ pub struct Metrics {
     queries_num: AtomicU64,
     errors_iter_num: AtomicU64,
     queries_iter_num: AtomicU64,
+    retries_num: AtomicU64,
     histogram: Arc<Mutex<Histogram>>,
 }
 
@@ -44,6 +45,7 @@ impl Metrics {
             queries_num: AtomicU64::new(0),
             errors_iter_num: AtomicU64::new(0),
             queries_iter_num: AtomicU64::new(0),
+            retries_num: AtomicU64::new(0),
             histogram: Arc::new(Mutex::new(Histogram::new())),
         }
     }
@@ -69,6 +71,11 @@ impl Metrics {
         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) {
+        self.retries_num.fetch_add(1, ORDER_TYPE);
+    }
+
     /// Saves to histogram latency of completing single query.
     /// For paged queries it should log latency for every page.
     ///
@@ -115,6 +122,11 @@ impl Metrics {
     pub fn get_queries_iter_num(&self) -> u64 {
         self.queries_iter_num.load(ORDER_TYPE)
     }
+
+    /// Returns counter measuring how many times a retry policy has decided to retry a query
+    pub fn get_retries_num(&self) -> u64 {
+        self.retries_num.load(ORDER_TYPE)
+    }
 }
 
 pub struct MetricsView {
@@ -158,4 +170,9 @@ impl MetricsView {
     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 6faf6523..cbaac27f 100644
--- a/scylla/src/transport/mod.rs
+++ b/scylla/src/transport/mod.rs
@@ -3,6 +3,7 @@ pub mod connection;
 mod connection_keeper;
 pub mod load_balancing;
 mod node;
+pub mod retry_policy;
 pub mod session;
 pub mod session_builder;
 mod topology;
diff --git a/scylla/src/transport/node.rs b/scylla/src/transport/node.rs
index c3a43277..ffd2dac0 100644
--- a/scylla/src/transport/node.rs
+++ b/scylla/src/transport/node.rs
@@ -1,5 +1,5 @@
 /// Node represents a cluster node along with it's data and connections
-use crate::routing::{Shard, ShardInfo, Token};
+use crate::routing::{ShardInfo, Token};
 use crate::transport::connection::VerifiedKeyspaceName;
 use crate::transport::connection::{Connection, ConnectionConfig};
 use crate::transport::connection_keeper::{ConnectionKeeper, ShardInfoSender};
@@ -8,6 +8,7 @@ use futures::future::join_all;
 
 use futures::{future::RemoteHandle, FutureExt};
 use rand::Rng;
+use std::convert::TryInto;
 use std::hash::{Hash, Hasher};
 use std::net::SocketAddr;
 use std::sync::{Arc, RwLock};
@@ -113,6 +114,7 @@ impl Node {
     }
 
     /// Get connection which should be used to connect using given token
+    /// If this connection is broken get any random connection to this Node
     pub async fn connection_for_token(&self, token: Token) -> Result<Arc<Connection>, QueryError> {
         let connections: Arc<NodeConnections> = self.connections.read().unwrap().clone();
 
@@ -122,10 +124,11 @@ impl Node {
                 shard_info,
                 shard_conns,
             } => {
-                let shard: Shard = shard_info.shard_of(token);
-                // It's guaranteed by NodeWorker that for each shard < shard_info.nr_shards
-                // there is an entry for this shard in connections map
-                shard_conns[shard as usize].get_connection().await
+                let shard: u16 = shard_info
+                    .shard_of(token)
+                    .try_into()
+                    .expect("Shard number doesn't fit in u16");
+                Self::connection_for_shard(shard, shard_info.nr_shards, shard_conns).await
             }
         }
     }
@@ -140,13 +143,40 @@ impl Node {
                 shard_info,
                 shard_conns,
             } => {
-                let shard = rand::thread_rng().gen_range(0..shard_info.nr_shards as u32);
-                // It's guaranteed by NodeWorker that shard_conns.len() == shard_info.nr_shards
-                shard_conns[shard as usize].get_connection().await
+                let shard: u16 = rand::thread_rng().gen_range(0..shard_info.nr_shards);
+                Self::connection_for_shard(shard, shard_info.nr_shards, shard_conns).await
             }
         }
     }
 
+    // Tries to get a connection to given shard, if it's broken returns any working connection
+    async fn connection_for_shard(
+        shard: u16,
+        nr_shards: u16,
+        shard_conns: &[ConnectionKeeper],
+    ) -> Result<Arc<Connection>, QueryError> {
+        // Try getting the desired connection
+        let mut last_error: QueryError = match shard_conns[shard as usize].get_connection().await {
+            Ok(connection) => return Ok(connection),
+            Err(e) => e,
+        };
+
+        // If this fails try getting any other in random order
+        let mut shards_to_try: Vec<u16> = (shard..nr_shards).chain(0..shard).skip(1).collect();
+
+        while !shards_to_try.is_empty() {
+            let idx = rand::thread_rng().gen_range(0..shards_to_try.len());
+            let shard = shards_to_try.swap_remove(idx);
+
+            match shard_conns[shard as usize].get_connection().await {
+                Ok(conn) => return Ok(conn),
+                Err(e) => last_error = e,
+            }
+        }
+
+        return Err(last_error);
+    }
+
     pub async fn use_keyspace(
         &self,
         keyspace_name: VerifiedKeyspaceName,
diff --git a/scylla/src/transport/retry_policy.rs b/scylla/src/transport/retry_policy.rs
new file mode 100644
index 00000000..fe01652b
--- /dev/null
+++ b/scylla/src/transport/retry_policy.rs
@@ -0,0 +1,500 @@
+use crate::statement::Consistency;
+use crate::transport::errors::{DBError, QueryError, WriteType};
+
+/// Information about a failed query
+pub struct QueryInfo<'a> {
+    /// The error with which the query failed
+    pub error: &'a QueryError,
+    /// 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
+    pub is_idempotent: bool,
+    /// Consistency with which the query failed
+    pub consistency: Consistency,
+}
+
+#[derive(Debug, PartialEq, Eq)]
+pub enum RetryDecision {
+    RetrySameNode,
+    RetryNextNode,
+    DontRetry,
+}
+
+/// Specifies a policy used to decide when to retry a query
+pub trait RetryPolicy {
+    /// Called for each new query, starts a session of deciding about retries
+    fn new_session(&self) -> Box<dyn RetrySession + Send + Sync>;
+
+    /// Used to clone this RetryPolicy
+    fn clone_boxed(&self) -> Box<dyn RetryPolicy + Send + Sync>;
+}
+
+/// Used throughout a single query to decide when to retry it
+/// After this query is finished it is destroyed or reset
+pub trait RetrySession {
+    /// Called after the query failed - decide what to do next
+    fn decide_should_retry(&mut self, query_info: QueryInfo) -> RetryDecision;
+
+    /// Reset before using for a new query
+    fn reset(&mut self);
+}
+
+/// Forwards all errors directly to the user, never retries
+pub struct FallthroughRetryPolicy;
+pub struct FallthroughRetrySession;
+
+impl FallthroughRetryPolicy {
+    pub fn new() -> FallthroughRetryPolicy {
+        FallthroughRetryPolicy
+    }
+}
+
+impl Default for FallthroughRetryPolicy {
+    fn default() -> FallthroughRetryPolicy {
+        FallthroughRetryPolicy
+    }
+}
+
+impl RetryPolicy for FallthroughRetryPolicy {
+    fn new_session(&self) -> Box<dyn RetrySession + Send + Sync> {
+        Box::new(FallthroughRetrySession)
+    }
+
+    fn clone_boxed(&self) -> Box<dyn RetryPolicy + Send + Sync> {
+        Box::new(FallthroughRetryPolicy)
+    }
+}
+
+impl RetrySession for FallthroughRetrySession {
+    fn decide_should_retry(&mut self, _query_info: QueryInfo) -> RetryDecision {
+        RetryDecision::DontRetry
+    }
+
+    fn reset(&mut self) {}
+}
+
+/// Default retry policy - retries when there is a high chance that a retry might help.  
+/// Behaviour based on [DataStax Java Driver](https://docs.datastax.com/en/developer/java-driver/4.10/manual/core/retries/)
+pub struct DefaultRetryPolicy;
+
+impl DefaultRetryPolicy {
+    pub fn new() -> DefaultRetryPolicy {
+        DefaultRetryPolicy
+    }
+}
+
+impl Default for DefaultRetryPolicy {
+    fn default() -> DefaultRetryPolicy {
+        DefaultRetryPolicy::new()
+    }
+}
+
+impl RetryPolicy for DefaultRetryPolicy {
+    fn new_session(&self) -> Box<dyn RetrySession + Send + Sync> {
+        Box::new(DefaultRetrySession::new())
+    }
+
+    fn clone_boxed(&self) -> Box<dyn RetryPolicy + Send + Sync> {
+        Box::new(DefaultRetryPolicy)
+    }
+}
+
+pub struct DefaultRetrySession {
+    was_unavailable_retry: bool,
+    was_read_timeout_retry: bool,
+    was_write_timeout_retry: bool,
+}
+
+impl DefaultRetrySession {
+    pub fn new() -> DefaultRetrySession {
+        DefaultRetrySession {
+            was_unavailable_retry: false,
+            was_read_timeout_retry: false,
+            was_write_timeout_retry: false,
+        }
+    }
+}
+
+impl Default for DefaultRetrySession {
+    fn default() -> DefaultRetrySession {
+        DefaultRetrySession::new()
+    }
+}
+
+impl RetrySession for DefaultRetrySession {
+    fn decide_should_retry(&mut self, query_info: QueryInfo) -> RetryDecision {
+        match query_info.error {
+            // Basic errors - there are some problems on this node
+            // Retry on a different one if possible
+            QueryError::IOError(_)
+            | QueryError::DBError(DBError::Overloaded, _)
+            | QueryError::DBError(DBError::ServerError, _)
+            | QueryError::DBError(DBError::TruncateError, _) => {
+                if query_info.is_idempotent {
+                    RetryDecision::RetryNextNode
+                } else {
+                    RetryDecision::DontRetry
+                }
+            }
+            // Unavailable - the current node believes that not enough nodes
+            // are alive to satisfy specified consistency requirements.
+            // Maybe this node has network problems - try a different one.
+            // Perform at most one retry - it's unlikely that two nodes
+            // have network problems at the same time
+            QueryError::DBError(DBError::Unavailable { .. }, _) => {
+                if !self.was_unavailable_retry {
+                    self.was_unavailable_retry = true;
+                    RetryDecision::RetryNextNode
+                } else {
+                    RetryDecision::DontRetry
+                }
+            }
+            // ReadTimeout - coordinator didn't receive enough replies in time.
+            // Retry at most once and only if there were actually enough replies
+            // to satisfy consistency but they were all just checksums (data_present == true).
+            // This happens when the coordinator picked replicas that were overloaded/dying.
+            // Retried request should have some useful response because the node will detect
+            // that these replicas are dead.
+            QueryError::DBError(
+                DBError::ReadTimeout {
+                    received,
+                    required,
+                    data_present,
+                    ..
+                },
+                _,
+            ) => {
+                if !self.was_read_timeout_retry && received >= required && *data_present {
+                    self.was_read_timeout_retry = true;
+                    RetryDecision::RetrySameNode
+                } else {
+                    RetryDecision::DontRetry
+                }
+            }
+            // Write timeout - coordinator didn't receive enough replies in time.
+            // Retry at most once and only for BatchLog write.
+            // Coordinator probably didn't detect the nodes as dead.
+            // By the time we retry they should be detected as dead.
+            QueryError::DBError(DBError::WriteTimeout { write_type, .. }, _) => {
+                if !self.was_write_timeout_retry
+                    && query_info.is_idempotent
+                    && *write_type == WriteType::BatchLog
+                {
+                    self.was_write_timeout_retry = true;
+                    RetryDecision::RetrySameNode
+                } else {
+                    RetryDecision::DontRetry
+                }
+            }
+            // The node is still bootstrapping it can't execute the query, we should try another one
+            QueryError::DBError(DBError::IsBootstrapping, _) => RetryDecision::RetryNextNode,
+            // In all other cases propagate the error to the user
+            _ => RetryDecision::DontRetry,
+        }
+    }
+
+    fn reset(&mut self) {
+        *self = DefaultRetrySession::new();
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{DefaultRetryPolicy, QueryInfo, RetryDecision, RetryPolicy};
+    use crate::statement::Consistency;
+    use crate::transport::errors::{BadQuery, DBError, QueryError, WriteType};
+    use std::io::ErrorKind;
+    use std::sync::Arc;
+
+    fn make_query_info(error: &QueryError, is_idempotent: bool) -> QueryInfo<'_> {
+        QueryInfo {
+            error,
+            is_idempotent,
+            consistency: Consistency::One,
+        }
+    }
+
+    // Asserts that default policy never retries for this Error
+    fn default_policy_assert_never_retries(error: QueryError) {
+        let mut policy = DefaultRetryPolicy::new().new_session();
+        assert_eq!(
+            policy.decide_should_retry(make_query_info(&error, false)),
+            RetryDecision::DontRetry
+        );
+
+        let mut policy = DefaultRetryPolicy::new().new_session();
+        assert_eq!(
+            policy.decide_should_retry(make_query_info(&error, true)),
+            RetryDecision::DontRetry
+        );
+    }
+
+    #[test]
+    fn default_never_retries() {
+        let never_retried_dberrors = vec![
+            DBError::SyntaxError,
+            DBError::Invalid,
+            DBError::AlreadyExists {
+                keyspace: String::new(),
+                table: String::new(),
+            },
+            DBError::FunctionFailure {
+                keyspace: String::new(),
+                function: String::new(),
+                arg_types: vec![],
+            },
+            DBError::AuthenticationError,
+            DBError::Unauthorized,
+            DBError::ConfigError,
+            DBError::ReadFailure {
+                consistency: Consistency::Two,
+                received: 2,
+                required: 1,
+                numfailures: 1,
+                data_present: false,
+            },
+            DBError::WriteFailure {
+                consistency: Consistency::Two,
+                received: 1,
+                required: 2,
+                numfailures: 1,
+                write_type: WriteType::BatchLog,
+            },
+            DBError::Unprepared,
+            DBError::ProtocolError,
+            DBError::Other(0x124816),
+        ];
+
+        for dberror in never_retried_dberrors {
+            default_policy_assert_never_retries(QueryError::DBError(dberror, String::new()));
+        }
+
+        default_policy_assert_never_retries(QueryError::BadQuery(BadQuery::ValueLenMismatch(1, 2)));
+        default_policy_assert_never_retries(QueryError::ProtocolError("test"));
+    }
+
+    // Asserts that for this error policy retries on next on idempotent queries only
+    fn default_policy_assert_idempotent_next(error: QueryError) {
+        let mut policy = DefaultRetryPolicy::new().new_session();
+        assert_eq!(
+            policy.decide_should_retry(make_query_info(&error, false)),
+            RetryDecision::DontRetry
+        );
+
+        let mut policy = DefaultRetryPolicy::new().new_session();
+        assert_eq!(
+            policy.decide_should_retry(make_query_info(&error, true)),
+            RetryDecision::RetryNextNode
+        );
+    }
+
+    #[test]
+    fn default_idempotent_next_retries() {
+        let idempotent_next_errors = vec![
+            QueryError::DBError(DBError::Overloaded, String::new()),
+            QueryError::DBError(DBError::TruncateError, String::new()),
+            QueryError::DBError(DBError::ServerError, String::new()),
+            QueryError::IOError(Arc::new(std::io::Error::new(ErrorKind::Other, "test"))),
+        ];
+
+        for error in idempotent_next_errors {
+            default_policy_assert_idempotent_next(error);
+        }
+    }
+
+    // Always retry on next node if current one is bootstrapping
+    #[test]
+    fn default_bootstrapping() {
+        let error = QueryError::DBError(DBError::IsBootstrapping, String::new());
+
+        let mut policy = DefaultRetryPolicy::new().new_session();
+        assert_eq!(
+            policy.decide_should_retry(make_query_info(&error, false)),
+            RetryDecision::RetryNextNode
+        );
+
+        let mut policy = DefaultRetryPolicy::new().new_session();
+        assert_eq!(
+            policy.decide_should_retry(make_query_info(&error, true)),
+            RetryDecision::RetryNextNode
+        );
+    }
+
+    // On Unavailable error we retry one time no matter the idempotence
+    #[test]
+    fn default_unavailable() {
+        let error = QueryError::DBError(
+            DBError::Unavailable {
+                consistency: Consistency::Two,
+                required: 2,
+                alive: 1,
+            },
+            String::new(),
+        );
+
+        let mut policy_not_idempotent = DefaultRetryPolicy::new().new_session();
+        assert_eq!(
+            policy_not_idempotent.decide_should_retry(make_query_info(&error, false)),
+            RetryDecision::RetryNextNode
+        );
+        assert_eq!(
+            policy_not_idempotent.decide_should_retry(make_query_info(&error, false)),
+            RetryDecision::DontRetry
+        );
+
+        let mut policy_idempotent = DefaultRetryPolicy::new().new_session();
+        assert_eq!(
+            policy_idempotent.decide_should_retry(make_query_info(&error, true)),
+            RetryDecision::RetryNextNode
+        );
+        assert_eq!(
+            policy_idempotent.decide_should_retry(make_query_info(&error, true)),
+            RetryDecision::DontRetry
+        );
+    }
+
+    // On ReadTimeout we retry one time if there were enough responses and the data was present no matter the idempotence
+    #[test]
+    fn default_read_timeout() {
+        // Enough responses and data_present == true
+        let enough_responses_with_data = QueryError::DBError(
+            DBError::ReadTimeout {
+                consistency: Consistency::Two,
+                received: 2,
+                required: 2,
+                data_present: true,
+            },
+            String::new(),
+        );
+
+        // Not idempotent
+        let mut policy = DefaultRetryPolicy::new().new_session();
+        assert_eq!(
+            policy.decide_should_retry(make_query_info(&enough_responses_with_data, false)),
+            RetryDecision::RetrySameNode
+        );
+        assert_eq!(
+            policy.decide_should_retry(make_query_info(&enough_responses_with_data, false)),
+            RetryDecision::DontRetry
+        );
+
+        // Idempotent
+        let mut policy = DefaultRetryPolicy::new().new_session();
+        assert_eq!(
+            policy.decide_should_retry(make_query_info(&enough_responses_with_data, true)),
+            RetryDecision::RetrySameNode
+        );
+        assert_eq!(
+            policy.decide_should_retry(make_query_info(&enough_responses_with_data, true)),
+            RetryDecision::DontRetry
+        );
+
+        // Enough responses but data_present == false
+        let enough_responses_no_data = QueryError::DBError(
+            DBError::ReadTimeout {
+                consistency: Consistency::Two,
+                received: 2,
+                required: 2,
+                data_present: false,
+            },
+            String::new(),
+        );
+
+        // Not idempotent
+        let mut policy = DefaultRetryPolicy::new().new_session();
+        assert_eq!(
+            policy.decide_should_retry(make_query_info(&enough_responses_no_data, false)),
+            RetryDecision::DontRetry
+        );
+
+        // Idempotent
+        let mut policy = DefaultRetryPolicy::new().new_session();
+        assert_eq!(
+            policy.decide_should_retry(make_query_info(&enough_responses_no_data, true)),
+            RetryDecision::DontRetry
+        );
+
+        // Not enough responses, data_present == true
+        let not_enough_responses_with_data = QueryError::DBError(
+            DBError::ReadTimeout {
+                consistency: Consistency::Two,
+                received: 1,
+                required: 2,
+                data_present: true,
+            },
+            String::new(),
+        );
+
+        // Not idempotent
+        let mut policy = DefaultRetryPolicy::new().new_session();
+        assert_eq!(
+            policy.decide_should_retry(make_query_info(&not_enough_responses_with_data, false)),
+            RetryDecision::DontRetry
+        );
+
+        // Idempotent
+        let mut policy = DefaultRetryPolicy::new().new_session();
+        assert_eq!(
+            policy.decide_should_retry(make_query_info(&not_enough_responses_with_data, true)),
+            RetryDecision::DontRetry
+        );
+    }
+
+    // WriteTimeout will retry once when the query is idempotent and write_type == BatchLog
+    #[test]
+    fn default_write_timeout() {
+        // WriteType == BatchLog
+        let good_write_type = QueryError::DBError(
+            DBError::WriteTimeout {
+                consistency: Consistency::Two,
+                received: 1,
+                required: 2,
+                write_type: WriteType::BatchLog,
+            },
+            String::new(),
+        );
+
+        // Not idempotent
+        let mut policy = DefaultRetryPolicy::new().new_session();
+        assert_eq!(
+            policy.decide_should_retry(make_query_info(&good_write_type, false)),
+            RetryDecision::DontRetry
+        );
+
+        // Idempotent
+        let mut policy = DefaultRetryPolicy::new().new_session();
+        assert_eq!(
+            policy.decide_should_retry(make_query_info(&good_write_type, true)),
+            RetryDecision::RetrySameNode
+        );
+        assert_eq!(
+            policy.decide_should_retry(make_query_info(&good_write_type, true)),
+            RetryDecision::DontRetry
+        );
+
+        // WriteType != BatchLog
+        let bad_write_type = QueryError::DBError(
+            DBError::WriteTimeout {
+                consistency: Consistency::Two,
+                received: 4,
+                required: 2,
+                write_type: WriteType::Simple,
+            },
+            String::new(),
+        );
+
+        // Not idempotent
+        let mut policy = DefaultRetryPolicy::new().new_session();
+        assert_eq!(
+            policy.decide_should_retry(make_query_info(&bad_write_type, false)),
+            RetryDecision::DontRetry
+        );
+
+        // Idempotent
+        let mut policy = DefaultRetryPolicy::new().new_session();
+        assert_eq!(
+            policy.decide_should_retry(make_query_info(&bad_write_type, true)),
+            RetryDecision::DontRetry
+        );
+    }
+}
diff --git a/scylla/src/transport/session.rs b/scylla/src/transport/session.rs
index 1b9e4a95..e2dcbc66 100644
--- a/scylla/src/transport/session.rs
+++ b/scylla/src/transport/session.rs
@@ -1,7 +1,7 @@
 use futures::future::join_all;
+use std::future::Future;
 use std::net::SocketAddr;
 use std::sync::Arc;
-use std::time::Instant;
 use tokio::net::lookup_host;
 use tracing::warn;
 
@@ -10,24 +10,26 @@ 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::response::Response;
 use crate::frame::value::{BatchValues, SerializedValues, ValueList};
 use crate::prepared_statement::{PartitionKeyError, PreparedStatement};
 use crate::query::Query;
 use crate::routing::{murmur3_token, Token};
+use crate::statement::Consistency;
 use crate::transport::{
     cluster::Cluster,
-    connection::{ConnectionConfig, VerifiedKeyspaceName},
+    connection::{Connection, ConnectionConfig, VerifiedKeyspaceName},
     iterator::RowIterator,
     load_balancing::{LoadBalancingPolicy, RoundRobinPolicy, Statement, TokenAwarePolicy},
     metrics::{Metrics, MetricsView},
     node::Node,
+    retry_policy::{DefaultRetryPolicy, QueryInfo, RetryDecision, RetryPolicy, RetrySession},
     Compression,
 };
 
 pub struct Session {
     cluster: Cluster,
-    load_balancer: Box<dyn LoadBalancingPolicy>,
+    load_balancer: Arc<dyn LoadBalancingPolicy>,
+    retry_policy: Box<dyn RetryPolicy + Send + Sync>,
 
     metrics: Arc<Metrics>,
 }
@@ -47,10 +49,12 @@ pub struct SessionConfig {
     pub tcp_nodelay: bool,
 
     /// Load balancing policy used by Session
-    pub load_balancing: Box<dyn LoadBalancingPolicy>,
+    pub load_balancing: Arc<dyn LoadBalancingPolicy>,
 
     pub used_keyspace: Option<String>,
     pub keyspace_case_sensitive: bool,
+
+    pub retry_policy: Box<dyn RetryPolicy + Send + Sync>,
     /*
     These configuration options will be added in the future:
 
@@ -62,8 +66,6 @@ pub struct SessionConfig {
 
     pub tcp_keepalive: bool,
 
-    pub retry_policy: Option<String>,
-
     pub default_consistency: Option<String>,
     */
 }
@@ -91,9 +93,10 @@ impl SessionConfig {
             known_nodes: Vec::new(),
             compression: None,
             tcp_nodelay: false,
-            load_balancing: Box::new(TokenAwarePolicy::new(Box::new(RoundRobinPolicy::new()))),
+            load_balancing: Arc::new(TokenAwarePolicy::new(Box::new(RoundRobinPolicy::new()))),
             used_keyspace: None,
             keyspace_case_sensitive: false,
+            retry_policy: Box::new(DefaultRetryPolicy),
         }
     }
 
@@ -239,6 +242,7 @@ impl Session {
         let session = Session {
             cluster,
             load_balancer: config.load_balancing,
+            retry_policy: config.retry_policy,
             metrics,
         };
 
@@ -266,23 +270,9 @@ impl Session {
         query: impl Into<Query>,
         values: impl ValueList,
     ) -> Result<Option<Vec<result::Row>>, QueryError> {
-        let now = Instant::now();
-        self.metrics.inc_total_nonpaged_queries();
-        let result = self.query_no_metrics(query, values).await;
-        match &result {
-            Ok(_) => self.log_latency(now.elapsed().as_millis() as u64),
-            Err(_) => self.metrics.inc_failed_nonpaged_queries(),
-        };
-        result
-    }
-
-    async fn query_no_metrics(
-        &self,
-        query: impl Into<Query>,
-        values: impl ValueList,
-    ) -> Result<Option<Vec<result::Row>>, QueryError> {
-        let query = query.into();
-        let query_text = query.get_contents();
+        let query: Query = query.into();
+        let query_text: &str = query.get_contents();
+        let serialized_values = values.serialized();
 
         // In case the user tried doing session.query("use keyspace ks") run session::use_keyspace
         if query_is_setting_keyspace(query_text) {
@@ -298,16 +288,28 @@ impl Session {
                 .map(|_| None);
         }
 
-        let statement_info = Statement {
-            token: None,
-            keyspace: None,
+        let retry_session = match &query.retry_policy {
+            Some(policy) => policy.new_session(),
+            None => self.retry_policy.new_session(),
         };
-        let node = self.load_balancing_plan(statement_info);
 
-        node.random_connection()
-            .await?
-            .query_single_page(query, values)
-            .await
+        // Needed to avoid moving query and values into async move block
+        let query_ref: &Query = &query;
+        let values_ref = &serialized_values;
+
+        self.run_query(
+            Statement::default(),
+            query.is_idempotent,
+            query.consistency,
+            retry_session,
+            |node: Arc<Node>| async move { node.random_connection().await },
+            |connection: Arc<Connection>| async move {
+                connection
+                    .query_single_page_by_ref(query_ref, values_ref)
+                    .await
+            },
+        )
+        .await
     }
 
     pub async fn query_iter(
@@ -315,18 +317,20 @@ impl Session {
         query: impl Into<Query>,
         values: impl ValueList,
     ) -> Result<RowIterator, QueryError> {
+        let query: Query = query.into();
         let serialized_values = values.serialized()?;
 
-        let statement_info = Statement {
-            token: None,
-            keyspace: None,
+        let retry_session = match &query.retry_policy {
+            Some(policy) => policy.new_session(),
+            None => self.retry_policy.new_session(),
         };
-        let node = self.load_balancing_plan(statement_info);
 
         Ok(RowIterator::new_for_query(
-            node.random_connection().await?,
-            query.into(),
+            query,
             serialized_values.into_owned(),
+            retry_session,
+            self.load_balancer.clone(),
+            self.cluster.get_data(),
             self.metrics.clone(),
         ))
     }
@@ -383,24 +387,8 @@ impl Session {
         prepared: &PreparedStatement,
         values: impl ValueList,
     ) -> Result<Option<Vec<result::Row>>, QueryError> {
-        let now = Instant::now();
-        self.metrics.inc_total_nonpaged_queries();
-        let result = self.execute_no_metrics(prepared, values).await;
-        match &result {
-            Ok(_) => self.log_latency(now.elapsed().as_millis() as u64),
-            Err(_) => self.metrics.inc_failed_nonpaged_queries(),
-        };
-        result
-    }
-
-    async fn execute_no_metrics(
-        &self,
-        prepared: &PreparedStatement,
-        values: impl ValueList,
-    ) -> Result<Option<Vec<result::Row>>, QueryError> {
-        // FIXME: Prepared statement ids are local to a node, so we must make sure
-        // that prepare() sends to all nodes and keeps all ids.
         let serialized_values = values.serialized()?;
+        let values_ref = &serialized_values;
 
         let token = calculate_token(prepared, &serialized_values)?;
 
@@ -408,47 +396,23 @@ impl Session {
             token: Some(token),
             keyspace: prepared.get_keyspace_name(),
         };
-        let node = self.load_balancing_plan(statement_info);
-
-        let connection = node.connection_for_token(token).await?;
-        let result = connection
-            .execute(prepared, &serialized_values, None)
-            .await?;
-        match result {
-            Response::Error(err) => {
-                match err.code {
-                    9472 => {
-                        // Repreparation of a statement is needed
-                        let reprepared = connection.prepare(prepared.get_statement()).await?;
-                        // Reprepared statement should keep its id - it's the md5 sum
-                        // of statement contents
-                        if reprepared.get_id() != prepared.get_id() {
-                            return Err(QueryError::ProtocolError(
-                                "Prepared statement Id changed, md5 sum should stay the same",
-                            ));
-                        }
-
-                        let result = connection
-                            .execute(prepared, &serialized_values, None)
-                            .await?;
-                        match result {
-                            Response::Error(err) => Err(err.into()),
-                            Response::Result(result::Result::Rows(rs)) => Ok(Some(rs.rows)),
-                            Response::Result(_) => Ok(None),
-                            _ => Err(QueryError::ProtocolError(
-                                "EXECUTE: Unexpected server response",
-                            )),
-                        }
-                    }
-                    _ => Err(err.into()),
-                }
-            }
-            Response::Result(result::Result::Rows(rs)) => Ok(Some(rs.rows)),
-            Response::Result(_) => Ok(None),
-            _ => Err(QueryError::ProtocolError(
-                "EXECUTE: Unexpected server response",
-            )),
-        }
+
+        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,
+            |node: Arc<Node>| async move { node.connection_for_token(token).await },
+            |connection: Arc<Connection>| async move {
+                connection.execute_single_page(prepared, values_ref).await
+            },
+        )
+        .await
     }
 
     pub async fn execute_iter(
@@ -456,18 +420,23 @@ impl Session {
         prepared: impl Into<PreparedStatement>,
         values: impl ValueList,
     ) -> Result<RowIterator, QueryError> {
+        let prepared: PreparedStatement = prepared.into();
         let serialized_values = values.serialized()?;
 
-        let statement_info = Statement {
-            token: None,
-            keyspace: None,
+        let token = calculate_token(&prepared, &serialized_values)?;
+
+        let retry_session = match &prepared.retry_policy {
+            Some(policy) => policy.new_session(),
+            None => self.retry_policy.new_session(),
         };
-        let node = self.load_balancing_plan(statement_info);
 
         Ok(RowIterator::new_for_prepared_statement(
-            node.random_connection().await?,
-            prepared.into(),
+            prepared,
             serialized_values.into_owned(),
+            token,
+            retry_session,
+            self.load_balancer.clone(),
+            self.cluster.get_data(),
             self.metrics.clone(),
         ))
     }
@@ -478,26 +447,22 @@ impl Session {
     /// * `batch` - batch to be performed
     /// * `values` - values bound to the query
     pub async fn batch(&self, batch: &Batch, values: impl BatchValues) -> Result<(), QueryError> {
-        // FIXME: Prepared statement ids are local to a node
-        // this method does not handle this
-        let statement_info = Statement {
-            token: None,
-            keyspace: None,
+        let values_ref = &values;
+
+        let retry_session = match &batch.retry_policy {
+            Some(policy) => policy.new_session(),
+            None => self.retry_policy.new_session(),
         };
-        let node = self.load_balancing_plan(statement_info);
-
-        let response = node
-            .random_connection()
-            .await?
-            .batch(&batch, values)
-            .await?;
-        match response {
-            Response::Error(err) => Err(err.into()),
-            Response::Result(_) => Ok(()),
-            _ => Err(QueryError::ProtocolError(
-                "BATCH: Unexpected server response",
-            )),
-        }
+
+        self.run_query(
+            Statement::default(),
+            batch.is_idempotent,
+            batch.consistency,
+            retry_session,
+            |node: Arc<Node>| async move { node.random_connection().await },
+            |connection: Arc<Connection>| async move { connection.batch(batch, values_ref).await },
+        )
+        .await
     }
 
     /// Sends `USE <keyspace_name>` request on all connections  
@@ -556,17 +521,88 @@ impl Session {
         MetricsView::new(self.metrics.clone())
     }
 
-    fn log_latency(&self, latency: u64) {
-        let _ = self // silent fail if mutex is poisoned
-            .metrics
-            .log_query_latency(latency);
-    }
+    // 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
+    // - query will use node.random_connection()
+    // - execute will use node.connection_for_token()
+    // The second closure is used to do the query itself on a connection
+    // - query will use connection.query()
+    // - execute will use connection.execute()
+    // If this query closure fails with some errors retry policy is used to perform retries
+    // 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
+    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>,
+        choose_connection: impl Fn(Arc<Node>) -> ConnFut,
+        do_query: impl Fn(Arc<Connection>) -> QueryFut,
+    ) -> Result<ResT, QueryError>
+    where
+        ConnFut: Future<Output = Result<Arc<Connection>, QueryError>>,
+        QueryFut: Future<Output = Result<ResT, QueryError>>,
+    {
+        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!");
+
+        '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;
+                        // Broken connection doesn't count as a failed query, don't log in metrics
+                        continue 'nodes_in_plan;
+                    }
+                };
+
+                self.metrics.inc_total_nonpaged_queries();
+                let query_start = std::time::Instant::now();
 
-    fn load_balancing_plan(&self, statement_info: Statement) -> Arc<Node> {
-        let cluster_info = self.cluster.get_data();
-        let mut plan = self.load_balancer.plan(&statement_info, &cluster_info);
+                let query_result: Result<ResT, QueryError> = do_query(connection).await;
+
+                last_error = match query_result {
+                    Ok(response) => {
+                        let _ = self
+                            .metrics
+                            .log_query_latency(query_start.elapsed().as_millis() as u64);
+                        return Ok(response);
+                    }
+                    Err(e) => {
+                        self.metrics.inc_failed_nonpaged_queries();
+                        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,
+                };
+
+                match retry_session.decide_should_retry(query_info) {
+                    RetryDecision::RetrySameNode => {
+                        self.metrics.inc_retries_num();
+                        continue 'same_node_retries;
+                    }
+                    RetryDecision::RetryNextNode => {
+                        self.metrics.inc_retries_num();
+                        continue 'nodes_in_plan;
+                    }
+                    RetryDecision::DontRetry => return Err(last_error),
+                };
+            }
+        }
 
-        plan.next().unwrap() // Plan returned by load balancing policies should never be empty
+        return Err(last_error);
     }
 }
 
diff --git a/scylla/src/transport/session_builder.rs b/scylla/src/transport/session_builder.rs
index 734f3aa4..0cce157d 100644
--- a/scylla/src/transport/session_builder.rs
+++ b/scylla/src/transport/session_builder.rs
@@ -2,7 +2,9 @@ use super::errors::NewSessionError;
 use super::load_balancing::LoadBalancingPolicy;
 use super::session::{Session, SessionConfig};
 use super::Compression;
+use crate::transport::retry_policy::RetryPolicy;
 use std::net::SocketAddr;
+use std::sync::Arc;
 
 /// SessionBuilder is used to create new Session instances
 /// # Example
@@ -186,20 +188,44 @@ impl SessionBuilder {
     /// # use scylla::{Session, SessionBuilder};
     /// # use scylla::transport::Compression;
     /// # use scylla::transport::load_balancing::RoundRobinPolicy;
+    /// # use std::sync::Arc;
     /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
     /// let session: Session = SessionBuilder::new()
     ///     .known_node("127.0.0.1:9042")
-    ///     .load_balancing(Box::new(RoundRobinPolicy::new()))
+    ///     .load_balancing(Arc::new(RoundRobinPolicy::new()))
     ///     .build()
     ///     .await?;
     /// # Ok(())
     /// # }
     /// ```
-    pub fn load_balancing(mut self, policy: Box<dyn LoadBalancingPolicy>) -> Self {
+    pub fn load_balancing(mut self, policy: Arc<dyn LoadBalancingPolicy>) -> Self {
         self.config.load_balancing = 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`]
+    ///
+    /// # Example
+    /// ```
+    /// # use scylla::{Session, SessionBuilder};
+    /// # use scylla::transport::Compression;
+    /// use scylla::transport::retry_policy::DefaultRetryPolicy;
+    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
+    /// let session: Session = SessionBuilder::new()
+    ///     .known_node("127.0.0.1:9042")
+    ///     .retry_policy(Box::new(DefaultRetryPolicy::new()))
+    ///     .build()
+    ///     .await?;
+    /// # Ok(())
+    /// # }
+    /// ```
+    pub fn retry_policy(mut self, retry_policy: Box<dyn RetryPolicy + Send + Sync>) -> Self {
+        self.config.retry_policy = retry_policy;
+        self
+    }
+
     /// Builds the Session after setting all the options
     ///
     /// # Example
@@ -234,6 +260,7 @@ mod tests {
     use crate::transport::session::KnownNode;
     use crate::transport::Compression;
     use std::net::{IpAddr, Ipv4Addr, SocketAddr};
+    use std::sync::Arc;
 
     #[test]
     fn default_session_builder() {
@@ -334,7 +361,7 @@ mod tests {
             "TokenAwarePolicy{child_policy: RoundRobinPolicy}".to_string()
         );
 
-        builder = builder.load_balancing(Box::new(RoundRobinPolicy::new()));
+        builder = builder.load_balancing(Arc::new(RoundRobinPolicy::new()));
         assert_eq!(
             builder.config.load_balancing.name(),
             "RoundRobinPolicy".to_string()
@@ -370,7 +397,7 @@ mod tests {
         builder = builder.known_nodes_addr(&[addr1, addr2]);
         builder = builder.compression(Some(Compression::Snappy));
         builder = builder.tcp_nodelay(true);
-        builder = builder.load_balancing(Box::new(RoundRobinPolicy::new()));
+        builder = builder.load_balancing(Arc::new(RoundRobinPolicy::new()));
         builder = builder.use_keyspace("ks_name", true);
 
         assert_eq!(


Test patch:
diff --git a/scylla/src/transport/session_test.rs b/scylla/src/transport/session_test.rs
index 776ba26f..4ce83f8e 100644
--- a/scylla/src/transport/session_test.rs
+++ b/scylla/src/transport/session_test.rs
@@ -1,6 +1,6 @@
 use crate::frame::value::ValueList;
 use crate::routing::hash3_x64_128;
-use crate::transport::errors::{BadKeyspaceName, BadQuery, QueryError};
+use crate::transport::errors::{BadKeyspaceName, BadQuery, DBError, QueryError};
 use crate::{IntoTypedRows, Session, SessionBuilder};
 
 #[tokio::test]
@@ -542,3 +542,58 @@ async fn test_fetch_system_keyspace() {
 
     session.execute(&prepared_statement, &[]).await.unwrap();
 }
+
+// Test that some Database Errors are parsed correctly
+#[tokio::test]
+async fn test_db_errors() {
+    let uri = std::env::var("SCYLLA_URI").unwrap_or_else(|_| "127.0.0.1:9042".to_string());
+    let session = SessionBuilder::new().known_node(uri).build().await.unwrap();
+
+    // SyntaxError on bad query
+    assert!(matches!(
+        session.query("gibberish", &[]).await,
+        Err(QueryError::DBError(DBError::SyntaxError, _))
+    ));
+
+    // AlreadyExists when creating a keyspace for the second time
+    session.query("CREATE KEYSPACE IF NOT EXISTS db_errors_ks WITH REPLICATION = {'class' : 'SimpleStrategy', 'replication_factor' : 1}", &[]).await.unwrap();
+
+    let create_keyspace_res = session.query("CREATE KEYSPACE db_errors_ks WITH REPLICATION = {'class' : 'SimpleStrategy', 'replication_factor' : 1}", &[]).await;
+    let keyspace_exists_error: DBError = match create_keyspace_res {
+        Err(QueryError::DBError(e, _)) => e,
+        _ => panic!("Second CREATE KEYSPACE didn't return an error!"),
+    };
+
+    assert_eq!(
+        keyspace_exists_error,
+        DBError::AlreadyExists {
+            keyspace: "db_errors_ks".to_string(),
+            table: "".to_string()
+        }
+    );
+
+    // AlreadyExists when creating a table for the second time
+    session
+        .query(
+            "CREATE TABLE IF NOT EXISTS db_errors_ks.tab (a text primary key)",
+            &[],
+        )
+        .await
+        .unwrap();
+
+    let create_table_res = session
+        .query("CREATE TABLE db_errors_ks.tab (a text primary key)", &[])
+        .await;
+    let create_tab_error: DBError = match create_table_res {
+        Err(QueryError::DBError(e, _)) => e,
+        _ => panic!("Second CREATE TABLE didn't return an error!"),
+    };
+
+    assert_eq!(
+        create_tab_error,
+        DBError::AlreadyExists {
+            keyspace: "db_errors_ks".to_string(),
+            table: "tab".to_string()
+        }
+    );
+}