PreparedStatement: Introduce mutable metadata

This introduces second ResultMetadata field to PreparedStatement:
current_result_metadata. It is stored in ArcSwap, and updateable via new
`update_current_result_metadata` method.
The original field (`initial_result_metadata`) is kept. It can't be
removed because of backwards compatibility. The only way it is exposed
is public `get_result_set_col_specs` method, which will become
deprecated in future commit. Internal getter (`get_result_metadata`) is
renamed to `get_current_result_metadata`, and code in connection.rs
adjusted. Because of how the code is structured and all the preparation
in previous PRs, this is enough to cover all cases (_iter, retries etc).

Currently the update method is unused, intentionally. It will become
used when we implement Scylla's metadata extension.

Author: Lorak-mmk

scylla/src/network/connection.rs

@@ -930,14 +930,14 @@ impl Connection {
 
         let cached_metadata = prepared_statement
             .get_use_cached_result_metadata()
-            .then(|| prepared_statement.get_result_metadata());
+            .then(|| prepared_statement.get_current_result_metadata());
 
         let query_response = self
             .send_request(
                 &execute_frame,
                 true,
                 prepared_statement.config.tracing,
-                cached_metadata,
+                cached_metadata.as_ref(),
             )
             .await?;
 
@@ -967,7 +967,7 @@ impl Connection {
                         &execute_frame,
                         true,
                         prepared_statement.config.tracing,
-                        cached_metadata,
+                        cached_metadata.as_ref(),
                     )
                     .await?;
 

scylla/src/statement/prepared.rs

@@ -1,6 +1,7 @@
 //! Defines the [`PreparedStatement`] type, which represents a statement
 //! that has been prepared in advance on the server.
 
+use arc_swap::{ArcSwap, Guard};
 use bytes::{Bytes, BytesMut};
 use scylla_cql::frame::response::result::{
     ColumnSpec, PartitionKeyIndex, ResultMetadata, TableSpec,
@@ -187,6 +188,7 @@ struct PreparedStatementSharedData {
     id: Bytes,
     metadata: PreparedMetadata,
     initial_result_metadata: Arc<ResultMetadata<'static>>,
+    current_result_metadata: ArcSwap<ResultMetadata<'static>>,
     statement: String,
     is_confirmed_lwt: bool,
 }
@@ -203,6 +205,18 @@ impl Clone for PreparedStatement {
     }
 }
 
+/// Stores a snapshot of current result metadata column specs.
+pub struct ColumnSpecsGuard {
+    result: Guard<Arc<ResultMetadata<'static>>>,
+}
+
+impl ColumnSpecsGuard {
+    /// Retrieves current result metadata column specs.
+    pub fn get(&self) -> ColumnSpecs<'_, 'static> {
+        ColumnSpecs::new(self.result.col_specs())
+    }
+}
+
 impl PreparedStatement {
     fn new(
         id: Bytes,
@@ -217,7 +231,8 @@ impl PreparedStatement {
             shared: Arc::new(PreparedStatementSharedData {
                 id,
                 metadata,
-                initial_result_metadata: result_metadata,
+                initial_result_metadata: Arc::clone(&result_metadata),
+                current_result_metadata: ArcSwap::from(result_metadata),
                 statement,
                 is_confirmed_lwt: is_lwt,
             }),
@@ -507,8 +522,18 @@ impl PreparedStatement {
     }
 
     /// Access metadata about the result of prepared statement returned by the database
-    pub(crate) fn get_result_metadata(&self) -> &Arc<ResultMetadata<'static>> {
-        &self.shared.initial_result_metadata
+    pub(crate) fn get_current_result_metadata(&self) -> Arc<ResultMetadata<'static>> {
+        self.shared.current_result_metadata.load_full()
+    }
+
+    /// Update metadata about the result of prepared statement.
+    // Will be used when we implement support for metadata id extension.
+    #[allow(dead_code)]
+    pub(crate) fn update_current_result_metadata(
+        &self,
+        new_metadata: Arc<ResultMetadata<'static>>,
+    ) {
+        self.shared.current_result_metadata.store(new_metadata);
     }
 
     /// Access column specifications of the result set returned after the execution of this statement