cass_prepared: hold info about col data types

Until now, we would not hold the information about column data types
from PreparedMetadata. We need to hold this information to perform
a typecheck during binding values to statement.

We could construct the data types from column specs each time
we bind a value. However, CassDataType might be a heavy nested
object, and so I decided to cache it in CassPrepared.

Author: muzarski

scylla-rust-wrapper/src/batch.rs

@@ -165,7 +165,7 @@ pub unsafe extern "C" fn cass_batch_add_statement(
 
     match &statement.statement {
         Statement::Simple(q) => state.batch.append_statement(q.query.clone()),
-        Statement::Prepared(p) => state.batch.append_statement((**p).clone()),
+        Statement::Prepared(p) => state.batch.append_statement(p.statement.clone()),
     };
 
     state.bound_values.push(statement.bound_values.clone());

scylla-rust-wrapper/src/future.rs

@@ -8,7 +8,6 @@ use crate::types::*;
 use crate::uuid::CassUuid;
 use crate::RUNTIME;
 use futures::future;
-use scylla::prepared_statement::PreparedStatement;
 use std::future::Future;
 use std::mem;
 use std::os::raw::c_void;
@@ -20,7 +19,7 @@ pub enum CassResultValue {
     Empty,
     QueryResult(Arc<CassResult>),
     QueryError(Arc<CassErrorResult>),
-    Prepared(Arc<PreparedStatement>),
+    Prepared(Arc<CassPrepared>),
 }
 
 type CassFutureError = (CassError, String);

scylla-rust-wrapper/src/prepared.rs

@@ -3,11 +3,32 @@ use std::sync::Arc;
 
 use crate::{
     argconv::*,
+    cass_types::{get_column_type, CassDataType},
     statement::{CassStatement, Statement},
 };
 use scylla::prepared_statement::PreparedStatement;
 
-pub type CassPrepared = PreparedStatement;
+#[derive(Debug, Clone)]
+pub struct CassPrepared {
+    // Data types of columns from PreparedMetadata.
+    pub variable_col_data_types: Vec<Arc<CassDataType>>,
+    pub statement: PreparedStatement,
+}
+
+impl CassPrepared {
+    pub fn new_from_prepared_statement(statement: PreparedStatement) -> Self {
+        let variable_col_data_types = statement
+            .get_variable_col_specs()
+            .iter()
+            .map(|col_spec| Arc::new(get_column_type(&col_spec.typ)))
+            .collect();
+
+        Self {
+            variable_col_data_types,
+            statement,
+        }
+    }
+}
 
 #[no_mangle]
 pub unsafe extern "C" fn cass_prepared_free(prepared_raw: *const CassPrepared) {
@@ -19,7 +40,7 @@ pub unsafe extern "C" fn cass_prepared_bind(
     prepared_raw: *const CassPrepared,
 ) -> *mut CassStatement {
     let prepared: Arc<_> = clone_arced(prepared_raw);
-    let bound_values_size = prepared.get_variable_col_specs().len();
+    let bound_values_size = prepared.statement.get_variable_col_specs().len();
 
     // cloning prepared statement's arc, because creating CassStatement should not invalidate
     // the CassPrepared argument

scylla-rust-wrapper/src/session.rs

@@ -8,6 +8,7 @@ use crate::exec_profile::{CassExecProfile, ExecProfileName, PerStatementExecProf
 use crate::future::{CassFuture, CassFutureResult, CassResultValue};
 use crate::metadata::create_table_metadata;
 use crate::metadata::{CassKeyspaceMeta, CassMaterializedViewMeta, CassSchemaMeta};
+use crate::prepared::CassPrepared;
 use crate::query_result::Value::{CollectionValue, RegularValue};
 use crate::query_result::{CassResult, CassResultData, CassRow, CassValue, Collection, Value};
 use crate::statement::CassStatement;
@@ -279,9 +280,9 @@ pub unsafe extern "C" fn cass_session_execute(
 
         match &mut statement {
             Statement::Simple(query) => query.query.set_execution_profile_handle(handle),
-            Statement::Prepared(prepared) => {
-                Arc::make_mut(prepared).set_execution_profile_handle(handle)
-            }
+            Statement::Prepared(prepared) => Arc::make_mut(prepared)
+                .statement
+                .set_execution_profile_handle(handle),
         }
 
         let query_res: Result<(QueryResult, PagingStateResponse), QueryError> = match statement {
@@ -300,11 +301,11 @@ pub unsafe extern "C" fn cass_session_execute(
             Statement::Prepared(prepared) => {
                 if paging_enabled {
                     session
-                        .execute_single_page(&prepared, bound_values, paging_state)
+                        .execute_single_page(&prepared.statement, bound_values, paging_state)
                         .await
                 } else {
                     session
-                        .execute_unpaged(&prepared, bound_values)
+                        .execute_unpaged(&prepared.statement, bound_values)
                         .await
                         .map(|result| (result, PagingStateResponse::NoMorePages))
                 }
@@ -499,7 +500,9 @@ pub unsafe extern "C" fn cass_session_prepare_from_existing(
             .await
             .map_err(|err| (CassError::from(&err), err.msg()))?;
 
-        Ok(CassResultValue::Prepared(Arc::new(prepared)))
+        Ok(CassResultValue::Prepared(Arc::new(
+            CassPrepared::new_from_prepared_statement(prepared),
+        )))
     })
 }
 
@@ -542,7 +545,9 @@ pub unsafe extern "C" fn cass_session_prepare_n(
         // Set Cpp Driver default configuration for queries:
         prepared.set_consistency(Consistency::One);
 
-        Ok(CassResultValue::Prepared(Arc::new(prepared)))
+        Ok(CassResultValue::Prepared(Arc::new(
+            CassPrepared::new_from_prepared_statement(prepared),
+        )))
     })
 }
 

scylla-rust-wrapper/src/statement.rs

@@ -1,6 +1,7 @@
 use crate::argconv::*;
 use crate::cass_error::CassError;
 use crate::exec_profile::PerStatementExecProfile;
+use crate::prepared::CassPrepared;
 use crate::query_result::CassResult;
 use crate::retry_policy::CassRetryPolicy;
 use crate::types::*;
@@ -9,7 +10,6 @@ use scylla::frame::types::Consistency;
 use scylla::frame::value::MaybeUnset;
 use scylla::frame::value::MaybeUnset::{Set, Unset};
 use scylla::query::Query;
-use scylla::statement::prepared_statement::PreparedStatement;
 use scylla::statement::SerialConsistency;
 use scylla::transport::{PagingState, PagingStateResponse};
 use std::collections::HashMap;
@@ -24,7 +24,7 @@ include!(concat!(env!("OUT_DIR"), "/cppdriver_data_query_error.rs"));
 pub enum Statement {
     Simple(SimpleQuery),
     // Arc is needed, because PreparedStatement is passed by reference to session.execute
-    Prepared(Arc<PreparedStatement>),
+    Prepared(Arc<CassPrepared>),
 }
 
 #[derive(Clone)]
@@ -83,6 +83,7 @@ impl CassStatement {
         match &self.statement {
             Statement::Prepared(prepared) => {
                 let indices: Vec<usize> = prepared
+                    .statement
                     .get_variable_col_specs()
                     .iter()
                     .enumerate()
@@ -185,7 +186,9 @@ pub unsafe extern "C" fn cass_statement_set_consistency(
     if let Some(consistency) = consistency_opt {
         match &mut ptr_to_ref_mut(statement).statement {
             Statement::Simple(inner) => inner.query.set_consistency(consistency),
-            Statement::Prepared(inner) => Arc::make_mut(inner).set_consistency(consistency),
+            Statement::Prepared(inner) => {
+                Arc::make_mut(inner).statement.set_consistency(consistency)
+            }
         }
     }
 
@@ -205,7 +208,7 @@ pub unsafe extern "C" fn cass_statement_set_paging_size(
         statement.paging_enabled = true;
         match &mut statement.statement {
             Statement::Simple(inner) => inner.query.set_page_size(page_size),
-            Statement::Prepared(inner) => Arc::make_mut(inner).set_page_size(page_size),
+            Statement::Prepared(inner) => Arc::make_mut(inner).statement.set_page_size(page_size),
         }
     }
 
@@ -253,7 +256,9 @@ pub unsafe extern "C" fn cass_statement_set_is_idempotent(
 ) -> CassError {
     match &mut ptr_to_ref_mut(statement_raw).statement {
         Statement::Simple(inner) => inner.query.set_is_idempotent(is_idempotent != 0),
-        Statement::Prepared(inner) => Arc::make_mut(inner).set_is_idempotent(is_idempotent != 0),
+        Statement::Prepared(inner) => Arc::make_mut(inner)
+            .statement
+            .set_is_idempotent(is_idempotent != 0),
     }
 
     CassError::CASS_OK
@@ -266,7 +271,7 @@ pub unsafe extern "C" fn cass_statement_set_tracing(
 ) -> CassError {
     match &mut ptr_to_ref_mut(statement_raw).statement {
         Statement::Simple(inner) => inner.query.set_tracing(enabled != 0),
-        Statement::Prepared(inner) => Arc::make_mut(inner).set_tracing(enabled != 0),
+        Statement::Prepared(inner) => Arc::make_mut(inner).statement.set_tracing(enabled != 0),
     }
 
     CassError::CASS_OK
@@ -288,9 +293,9 @@ pub unsafe extern "C" fn cass_statement_set_retry_policy(
 
     match &mut ptr_to_ref_mut(statement).statement {
         Statement::Simple(inner) => inner.query.set_retry_policy(maybe_arced_retry_policy),
-        Statement::Prepared(inner) => {
-            Arc::make_mut(inner).set_retry_policy(maybe_arced_retry_policy)
-        }
+        Statement::Prepared(inner) => Arc::make_mut(inner)
+            .statement
+            .set_retry_policy(maybe_arced_retry_policy),
     }
 
     CassError::CASS_OK
@@ -317,9 +322,9 @@ pub unsafe extern "C" fn cass_statement_set_serial_consistency(
 
     match &mut ptr_to_ref_mut(statement).statement {
         Statement::Simple(inner) => inner.query.set_serial_consistency(Some(consistency)),
-        Statement::Prepared(inner) => {
-            Arc::make_mut(inner).set_serial_consistency(Some(consistency))
-        }
+        Statement::Prepared(inner) => Arc::make_mut(inner)
+            .statement
+            .set_serial_consistency(Some(consistency)),
     }
 
     CassError::CASS_OK
@@ -349,7 +354,9 @@ pub unsafe extern "C" fn cass_statement_set_timestamp(
 ) -> CassError {
     match &mut ptr_to_ref_mut(statement).statement {
         Statement::Simple(inner) => inner.query.set_timestamp(Some(timestamp)),
-        Statement::Prepared(inner) => Arc::make_mut(inner).set_timestamp(Some(timestamp)),
+        Statement::Prepared(inner) => Arc::make_mut(inner)
+            .statement
+            .set_timestamp(Some(timestamp)),
     }
 
     CassError::CASS_OK