Add an internal only BoundStatement

this struct keeps track of a PreparedStatement and
SerializedValues

Author: nrxus

scylla/src/client/pager.rs

@@ -18,7 +18,6 @@ use scylla_cql::frame::request::query::PagingState;
 use scylla_cql::frame::response::result::RawMetadataAndRawRows;
 use scylla_cql::frame::response::NonErrorResponse;
 use scylla_cql::frame::types::SerialConsistency;
-use scylla_cql::serialize::row::SerializedValues;
 use scylla_cql::Consistency;
 use std::result::Result;
 use thiserror::Error;
@@ -38,7 +37,8 @@ use crate::policies::load_balancing::{self, LoadBalancingPolicy, RoutingInfo};
 use crate::policies::retry::{RequestInfo, RetryDecision, RetrySession};
 use crate::response::query_result::ColumnSpecs;
 use crate::response::{NonErrorQueryResponse, QueryResponse};
-use crate::statement::prepared::{PartitionKeyError, PreparedStatement};
+use crate::statement::bound::BoundStatement;
+use crate::statement::prepared::PartitionKeyError;
 use crate::statement::unprepared::Statement;
 use tracing::{trace, trace_span, warn, Instrument};
 use uuid::Uuid;
@@ -63,8 +63,7 @@ struct ReceivedPage {
 }
 
 pub(crate) struct PreparedPagerConfig {
-    pub(crate) prepared: PreparedStatement,
-    pub(crate) values: SerializedValues,
+    pub(crate) bound: BoundStatement<'static>,
     pub(crate) execution_profile: Arc<ExecutionProfileInner>,
     pub(crate) cluster_state: Arc<ClusterState>,
     #[cfg(feature = "metrics")]
@@ -768,26 +767,30 @@ impl QueryPager {
         let (sender, receiver) = mpsc::channel::<Result<ReceivedPage, NextPageError>>(1);
 
         let consistency = config
+            .bound
             .prepared
             .config
             .consistency
             .unwrap_or(config.execution_profile.consistency);
         let serial_consistency = config
+            .bound
             .prepared
             .config
             .serial_consistency
             .unwrap_or(config.execution_profile.serial_consistency);
 
-        let page_size = config.prepared.get_validated_page_size();
+        let page_size = config.bound.prepared.get_validated_page_size();
 
         let load_balancing_policy = Arc::clone(
             config
+                .bound
                 .prepared
                 .get_load_balancing_policy()
                 .unwrap_or(&config.execution_profile.load_balancing_policy),
         );
 
         let retry_session = config
+            .bound
             .prepared
             .get_retry_policy()
             .map(|rp| &**rp)
@@ -796,14 +799,7 @@ impl QueryPager {
 
         let parent_span = tracing::Span::current();
         let worker_task = async move {
-            let prepared_ref = &config.prepared;
-            let values_ref = &config.values;
-
-            let (partition_key, token) = match prepared_ref
-                .extract_partition_key_and_calculate_token(
-                    prepared_ref.get_partitioner_name(),
-                    values_ref,
-                ) {
+            let (partition_key, token) = match config.bound.pk_and_token() {
                 Ok(res) => res.unzip(),
                 Err(err) => {
                     let (proof, _res) = ProvingSender::from(sender)
@@ -813,22 +809,22 @@ impl QueryPager {
                 }
             };
 
-            let table_spec = config.prepared.get_table_spec();
+            let table_spec = config.bound.prepared.get_table_spec();
             let statement_info = RoutingInfo {
                 consistency,
                 serial_consistency,
                 token,
                 table: table_spec,
-                is_confirmed_lwt: config.prepared.is_confirmed_lwt(),
+                is_confirmed_lwt: config.bound.prepared.is_confirmed_lwt(),
             };
 
+            let statement = &config.bound;
             let page_query = |connection: Arc<Connection>,
                               consistency: Consistency,
                               paging_state: PagingState| async move {
                 connection
                     .execute_raw_with_consistency(
-                        prepared_ref,
-                        values_ref,
+                        statement,
                         consistency,
                         serial_consistency,
                         Some(page_size),
@@ -837,7 +833,7 @@ impl QueryPager {
                     .await
             };
 
-            let serialized_values_size = config.values.buffer_size();
+            let serialized_values_size = config.bound.values.buffer_size();
 
             let replicas: Option<smallvec::SmallVec<[_; 8]>> =
                 if let (Some(table_spec), Some(token)) =
@@ -870,14 +866,14 @@ impl QueryPager {
                 sender: sender.into(),
                 page_query,
                 statement_info,
-                query_is_idempotent: config.prepared.config.is_idempotent,
+                query_is_idempotent: config.bound.prepared.config.is_idempotent,
                 query_consistency: consistency,
                 load_balancing_policy,
                 retry_session,
                 #[cfg(feature = "metrics")]
                 metrics: config.metrics,
                 paging_state: PagingState::start(),
-                history_listener: config.prepared.config.history_listener.clone(),
+                history_listener: config.bound.prepared.config.history_listener.clone(),
                 current_request_id: None,
                 current_attempt_id: None,
                 parent_span,
@@ -920,23 +916,21 @@ impl QueryPager {
     }
 
     pub(crate) async fn new_for_connection_execute_iter(
-        prepared: PreparedStatement,
-        values: SerializedValues,
+        bound: BoundStatement<'static>,
         connection: Arc<Connection>,
         consistency: Consistency,
         serial_consistency: Option<SerialConsistency>,
     ) -> Result<Self, NextPageError> {
         let (sender, receiver) = mpsc::channel::<Result<ReceivedPage, NextPageError>>(1);
 
-        let page_size = prepared.get_validated_page_size();
+        let page_size = bound.prepared.get_validated_page_size();
 
         let worker_task = async move {
             let worker = SingleConnectionPagerWorker {
                 sender: sender.into(),
                 fetcher: |paging_state| {
                     connection.execute_raw_with_consistency(
-                        &prepared,
-                        &values,
+                        &bound,
                         consistency,
                         serial_consistency,
                         Some(page_size),

scylla/src/client/session.rs

@@ -37,6 +37,7 @@ use crate::routing::partitioner::PartitionerName;
 use crate::routing::{Shard, ShardAwarePortRange};
 use crate::statement::batch::batch_values;
 use crate::statement::batch::{Batch, BatchStatement};
+use crate::statement::bound::BoundStatement;
 use crate::statement::prepared::{PartitionKeyError, PreparedStatement};
 use crate::statement::unprepared::Statement;
 use crate::statement::{Consistency, PageSize, StatementConfig};
@@ -46,7 +47,7 @@ use futures::future::try_join_all;
 use itertools::Itertools;
 use scylla_cql::frame::response::NonErrorResponse;
 use scylla_cql::serialize::batch::BatchValues;
-use scylla_cql::serialize::row::{SerializeRow, SerializedValues};
+use scylla_cql::serialize::row::SerializeRow;
 use std::borrow::Borrow;
 use std::future::Future;
 use std::net::{IpAddr, SocketAddr};
@@ -637,7 +638,8 @@ impl Session {
         prepared: &PreparedStatement,
         values: impl SerializeRow,
     ) -> Result<QueryResult, ExecutionError> {
-        self.do_execute_unpaged(prepared, values).await
+        let bound = prepared.bind(&values)?;
+        self.do_execute_unpaged(&bound).await
     }
 
     /// Executes a prepared statement, restricting results to single page.
@@ -702,8 +704,8 @@ impl Session {
         values: impl SerializeRow,
         paging_state: PagingState,
     ) -> Result<(QueryResult, PagingStateResponse), ExecutionError> {
-        self.do_execute_single_page(prepared, values, paging_state)
-            .await
+        let bound = prepared.bind(&values)?;
+        self.do_execute_single_page(&bound, paging_state).await
     }
 
     /// Execute a prepared statement with paging.\
@@ -750,7 +752,8 @@ impl Session {
         prepared: impl Into<PreparedStatement>,
         values: impl SerializeRow,
     ) -> Result<QueryPager, PagerExecutionError> {
-        self.do_execute_iter(prepared.into(), values).await
+        let bound = prepared.into().into_bind(&values)?;
+        self.do_execute_iter(bound).await
     }
 
     /// Execute a batch statement\
@@ -1083,12 +1086,11 @@ impl Session {
                                 .and_then(QueryResponse::into_non_error_query_response)
                         } else {
                             let prepared = connection.prepare(statement).await?;
-                            let serialized = prepared.serialize_values(values_ref)?;
-                            span_ref.record_request_size(serialized.buffer_size());
+                            let bound = prepared.bind(values_ref)?;
+                            span_ref.record_request_size(bound.values.buffer_size());
                             connection
                                 .execute_raw_with_consistency(
-                                    &prepared,
-                                    &serialized,
+                                    &bound,
                                     consistency,
                                     serial_consistency,
                                     page_size,
@@ -1181,11 +1183,9 @@ impl Session {
             // Making QueryPager::new_for_query work with values is too hard (if even possible)
             // so instead of sending one prepare to a specific connection on each iterator query,
             // we fully prepare a statement beforehand.
-            let prepared = self.prepare(statement).await?;
-            let values = prepared.serialize_values(&values)?;
+            let bound = self.prepare(statement).await?.into_bind(&values)?;
             QueryPager::new_for_prepared_statement(PreparedPagerConfig {
-                prepared,
-                values,
+                bound,
                 execution_profile,
                 cluster_state: self.cluster.get_state(),
                 #[cfg(feature = "metrics")]
@@ -1293,13 +1293,9 @@ impl Session {
 
     async fn do_execute_unpaged(
         &self,
-        prepared: &PreparedStatement,
-        values: impl SerializeRow,
+        bound: &BoundStatement<'_>,
     ) -> Result<QueryResult, ExecutionError> {
-        let serialized_values = prepared.serialize_values(&values)?;
-        let (result, paging_state) = self
-            .execute(prepared, &serialized_values, None, PagingState::start())
-            .await?;
+        let (result, paging_state) = self.execute(bound, None, PagingState::start()).await?;
         if !paging_state.finished() {
             error!("Unpaged prepared query returned a non-empty paging state! This is a driver-side or server-side bug.");
             return Err(ExecutionError::LastAttemptError(
@@ -1311,14 +1307,11 @@ impl Session {
 
     async fn do_execute_single_page(
         &self,
-        prepared: &PreparedStatement,
-        values: impl SerializeRow,
+        bound: &BoundStatement<'_>,
         paging_state: PagingState,
     ) -> Result<(QueryResult, PagingStateResponse), ExecutionError> {
-        let serialized_values = prepared.serialize_values(&values)?;
-        let page_size = prepared.get_validated_page_size();
-        self.execute(prepared, &serialized_values, Some(page_size), paging_state)
-            .await
+        let page_size = bound.prepared.get_validated_page_size();
+        self.execute(bound, Some(page_size), paging_state).await
     }
 
     /// Sends a prepared request to the database, optionally continuing from a saved point.
@@ -1333,44 +1326,45 @@ impl Session {
     /// should be made.
     async fn execute(
         &self,
-        prepared: &PreparedStatement,
-        serialized_values: &SerializedValues,
+        bound: &BoundStatement<'_>,
         page_size: Option<PageSize>,
         paging_state: PagingState,
     ) -> Result<(QueryResult, PagingStateResponse), ExecutionError> {
-        let values_ref = &serialized_values;
         let paging_state_ref = &paging_state;
 
-        let (partition_key, token) = prepared
-            .extract_partition_key_and_calculate_token(prepared.get_partitioner_name(), values_ref)
+        let (partition_key, token) = bound
+            .pk_and_token()
             .map_err(PartitionKeyError::into_execution_error)?
             .unzip();
 
-        let execution_profile = prepared
+        let execution_profile = bound
+            .prepared
             .get_execution_profile_handle()
             .unwrap_or_else(|| self.get_default_execution_profile_handle())
             .access();
 
-        let table_spec = prepared.get_table_spec();
+        let table_spec = bound.prepared.get_table_spec();
 
         let statement_info = RoutingInfo {
-            consistency: prepared
+            consistency: bound
+                .prepared
                 .config
                 .consistency
                 .unwrap_or(execution_profile.consistency),
-            serial_consistency: prepared
+            serial_consistency: bound
+                .prepared
                 .config
                 .serial_consistency
                 .unwrap_or(execution_profile.serial_consistency),
             token,
             table: table_spec,
-            is_confirmed_lwt: prepared.is_confirmed_lwt(),
+            is_confirmed_lwt: bound.prepared.is_confirmed_lwt(),
         };
 
         let span = RequestSpan::new_prepared(
             partition_key.as_ref().map(|pk| pk.iter()),
             token,
-            serialized_values.buffer_size(),
+            bound.values.buffer_size(),
         );
 
         if !span.span().is_disabled() {
@@ -1384,20 +1378,20 @@ impl Session {
         let run_request_result: RunRequestResult<NonErrorQueryResponse> = self
             .run_request(
                 statement_info,
-                &prepared.config,
+                &bound.prepared.config,
                 execution_profile,
                 |connection: Arc<Connection>,
                  consistency: Consistency,
                  execution_profile: &ExecutionProfileInner| {
-                    let serial_consistency = prepared
+                    let serial_consistency = bound
+                        .prepared
                         .config
                         .serial_consistency
                         .unwrap_or(execution_profile.serial_consistency);
                     async move {
                         connection
                             .execute_raw_with_consistency(
-                                prepared,
-                                values_ref,
+                                bound,
                                 consistency,
                                 serial_consistency,
                                 page_size,
@@ -1432,19 +1426,16 @@ impl Session {
 
     async fn do_execute_iter(
         &self,
-        prepared: PreparedStatement,
-        values: impl SerializeRow,
+        bound: BoundStatement<'static>,
     ) -> Result<QueryPager, PagerExecutionError> {
-        let serialized_values = prepared.serialize_values(&values)?;
-
-        let execution_profile = prepared
+        let execution_profile = bound
+            .prepared
             .get_execution_profile_handle()
             .unwrap_or_else(|| self.get_default_execution_profile_handle())
             .access();
 
         QueryPager::new_for_prepared_statement(PreparedPagerConfig {
-            prepared,
-            values: serialized_values,
+            bound,
             execution_profile,
             cluster_state: self.cluster.get_state(),
             #[cfg(feature = "metrics")]