result: deser_table_spec avoids allocations

deser_table_spec() now does not allocate and returns the borrowed
version of TableSpec. Moreover, now it verifies that table specs are
identical across columns, and returns an error if they aren't.

scylla-cql/src/frame/frame_errors.rs

@@ -11,6 +11,7 @@ pub use super::request::{
     startup::StartupSerializationError,
 };
 
+use super::response::result::TableSpec;
 use super::response::CqlResponseKind;
 use super::TryFromPrimitiveError;
 use crate::types::deserialize::DeserializationError;
@@ -353,6 +354,8 @@ pub enum TableSpecParseError {
     MalformedKeyspaceName(LowLevelDeserializationError),
     #[error("Malformed table name: {0}")]
     MalformedTableName(LowLevelDeserializationError),
+    #[error("TableSpec differs across columns - got specs: {0:?} and {1:?}")]
+    TableSpecDiffersAcrossColumns(TableSpec<'static>, TableSpec<'static>),
 }
 
 /// An error type returned when deserialization

scylla-cql/src/frame/response/result.rs

@@ -619,14 +619,30 @@ pub enum Result {
     SchemaChange(SchemaChange),
 }
 
-fn deser_table_spec(buf: &mut &[u8]) -> StdResult<TableSpec<'static>, TableSpecParseError> {
-    let ks_name = types::read_string(buf)
-        .map_err(TableSpecParseError::MalformedKeyspaceName)?
-        .to_owned();
-    let table_name = types::read_string(buf)
-        .map_err(TableSpecParseError::MalformedTableName)?
-        .to_owned();
-    Ok(TableSpec::owned(ks_name, table_name))
+// Only allocates a new `TableSpec` if one is not yet given.
+fn deser_table_spec<'frame: 'spec, 'spec>(
+    buf: &mut &'frame [u8],
+    known_spec: Option<TableSpec<'spec>>,
+) -> StdResult<TableSpec<'spec>, TableSpecParseError> {
+    let ks_name = types::read_string(buf).map_err(TableSpecParseError::MalformedKeyspaceName)?;
+    let table_name = types::read_string(buf).map_err(TableSpecParseError::MalformedTableName)?;
+
+    match known_spec {
+        None => Ok(TableSpec::borrowed(ks_name, table_name)),
+        Some(table_spec) => {
+            // We assume that for each column, table spec is the same.
+            // As this is not guaranteed by the CQL protocol specification but only by how
+            // Cassandra and ScyllaDB work (no support for joins), we perform a sanity check here.
+            if table_spec.table_name == table_name && table_spec.ks_name == ks_name {
+                Ok(table_spec)
+            } else {
+                Err(TableSpecParseError::TableSpecDiffersAcrossColumns(
+                    table_spec.into_owned(),
+                    TableSpec::owned(ks_name.to_owned(), table_name.to_owned()),
+                ))
+            }
+        }
+    }
 }
 
 macro_rules! generate_deser_type {
@@ -730,21 +746,30 @@ fn mk_col_spec_parse_error(
 
 fn deser_col_specs(
     buf: &mut &[u8],
-    global_table_spec: &Option<TableSpec<'static>>,
+    global_table_spec: Option<TableSpec<'static>>,
     col_count: usize,
 ) -> StdResult<Vec<ColumnSpec<'static>>, ColumnSpecParseError> {
+    let global_table_spec_provided = global_table_spec.is_some();
+    let mut table_spec = global_table_spec;
+
     let mut col_specs = Vec::with_capacity(col_count);
     for col_idx in 0..col_count {
-        let table_spec = if let Some(spec) = global_table_spec {
-            spec.clone()
-        } else {
-            deser_table_spec(buf).map_err(|err| mk_col_spec_parse_error(col_idx, err))?
-        };
+        if !global_table_spec_provided {
+            table_spec = Some(
+                deser_table_spec(buf, table_spec)
+                    .map_err(|err| mk_col_spec_parse_error(col_idx, err))?
+                    .to_owned(),
+            );
+        }
         let name = types::read_string(buf)
             .map_err(|err| mk_col_spec_parse_error(col_idx, err))?
             .to_owned();
         let typ = deser_type_owned(buf).map_err(|err| mk_col_spec_parse_error(col_idx, err))?;
-        col_specs.push(ColumnSpec::owned(name, typ, table_spec));
+        col_specs.push(ColumnSpec::owned(
+            name,
+            typ,
+            table_spec.as_ref().unwrap().clone(),
+        ));
     }
     Ok(col_specs)
 }
@@ -769,13 +794,11 @@ fn deser_result_metadata(
     let col_specs = if no_metadata {
         vec![]
     } else {
-        let global_table_spec = if global_tables_spec {
-            Some(deser_table_spec(buf)?)
-        } else {
-            None
-        };
+        let global_table_spec = global_tables_spec
+            .then(|| deser_table_spec(buf, None).map(TableSpec::into_owned))
+            .transpose()?;
 
-        deser_col_specs(buf, &global_table_spec, col_count)?
+        deser_col_specs(buf, global_table_spec, col_count)?
     };
 
     let metadata = ResultMetadata {
@@ -809,13 +832,11 @@ fn deser_prepared_metadata(
     }
     pk_indexes.sort_unstable_by_key(|pki| pki.index);
 
-    let global_table_spec = if global_tables_spec {
-        Some(deser_table_spec(buf)?)
-    } else {
-        None
-    };
+    let global_table_spec = global_tables_spec
+        .then(|| deser_table_spec(buf, None).map(TableSpec::into_owned))
+        .transpose()?;
 
-    let col_specs = deser_col_specs(buf, &global_table_spec, col_count)?;
+    let col_specs = deser_col_specs(buf, global_table_spec, col_count)?;
 
     Ok(PreparedMetadata {
         flags,