Return Query Error for non existent DC

scylla-cql/src/errors.rs

@@ -46,6 +46,10 @@ pub enum QueryError {
     #[error("Request timeout: {0}")]
     RequestTimeout(String),
 
+    /// Empty Query Plan
+    #[error("Load balancing policy returned empty query plan. It can happen when the driver is provided with non-existing datacenter name")]
+    EmptyQueryPlan,
+
     /// Address translation failed
     #[error("Address translation failed: {0}")]
     TranslationError(#[from] TranslationError),
@@ -404,6 +408,10 @@ pub enum NewSessionError {
     #[error("Client timeout: {0}")]
     RequestTimeout(String),
 
+    /// Empty Query Plan
+    #[error("Load balancing policy returned empty query plan. It can happen when the driver is provided with non-existing datacenter name")]
+    EmptyQueryPlan,
+
     /// Address translation failed
     #[error("Address translation failed: {0}")]
     TranslationError(#[from] TranslationError),
@@ -482,6 +490,7 @@ impl From<QueryError> for NewSessionError {
             QueryError::UnableToAllocStreamId => NewSessionError::UnableToAllocStreamId,
             QueryError::RequestTimeout(msg) => NewSessionError::RequestTimeout(msg),
             QueryError::TranslationError(e) => NewSessionError::TranslationError(e),
+            QueryError::EmptyQueryPlan => NewSessionError::EmptyQueryPlan,
         }
     }
 }

scylla/src/transport/load_balancing/default.rs

@@ -2414,6 +2414,7 @@ mod latency_awareness {
                 | QueryError::IoError(_)
                 | QueryError::ProtocolError(_)
                 | QueryError::TimeoutError
+                | QueryError::EmptyQueryPlan
                 | QueryError::RequestTimeout(_) => true,
             }
         }

scylla/src/transport/session.rs

@@ -1666,7 +1666,10 @@ impl Session {
             .consistency_set_on_statement
             .unwrap_or(execution_profile.consistency);
 
+        let mut query_plan_is_empty = true;
+
         'nodes_in_plan: for node in query_plan {
+            query_plan_is_empty = false;
             let span = trace_span!("Executing query", node = %node.address);
             'same_node_retries: loop {
                 trace!(parent: &span, "Execution started");
@@ -1769,6 +1772,10 @@ impl Session {
             }
         }
 
+        if query_plan_is_empty {
+            return Some(Err(QueryError::EmptyQueryPlan));
+        }
+
         last_error.map(Result::Err)
     }
 

scylla/src/transport/session_test.rs

@@ -2,6 +2,7 @@ use crate as scylla;
 use crate::batch::{Batch, BatchStatement};
 use crate::frame::response::result::Row;
 use crate::frame::value::ValueList;
+use crate::load_balancing::DefaultPolicy;
 use crate::prepared_statement::PreparedStatement;
 use crate::query::Query;
 use crate::retry_policy::{QueryInfo, RetryDecision, RetryPolicy, RetrySession};
@@ -2857,3 +2858,33 @@ async fn test_manual_primary_key_computation() {
         .await;
     }
 }
+
+#[tokio::test]
+async fn test_non_existent_dc_return_correct_error() {
+    let ks = "iot";
+
+    let host = "127.0.0.1";
+    let dc = "non existent dc";
+
+    let default_policy = DefaultPolicy::builder()
+        .prefer_datacenter(dc.to_string())
+        .build();
+
+    let profile = ExecutionProfile::builder()
+        .load_balancing_policy(default_policy)
+        .build();
+
+    let handle = profile.into_handle();
+
+    let session: Session = SessionBuilder::new()
+        .known_node(host)
+        .default_execution_profile_handle(handle)
+        .build()
+        .await
+        .expect("cannot create session");
+
+    let ks_stmt = format!("CREATE KEYSPACE IF NOT EXISTS {} WITH replication = {{'class': 'NetworkTopologyStrategy', '{}': 1}}", ks, dc);
+    let query_result = session.query(ks_stmt, &[]).await;
+
+    assert_matches!(query_result.unwrap_err(), QueryError::EmptyQueryPlan)
+}