scylla_cql/frame/request/
mod.rs

1//! CQL requests sent by the client.
2
3pub mod auth_response;
4pub mod batch;
5pub mod execute;
6pub mod options;
7pub mod prepare;
8pub mod query;
9pub mod register;
10pub mod startup;
11
12use batch::BatchTypeParseError;
13use thiserror::Error;
14
15use crate::Consistency;
16use crate::frame::protocol_features::ProtocolFeatures;
17use crate::frame::request::execute::ExecuteV2;
18use crate::serialize::row::SerializedValues;
19use bytes::Bytes;
20
21pub use auth_response::AuthResponse;
22pub use batch::Batch;
23#[expect(deprecated)]
24pub use execute::Execute;
25pub use options::Options;
26pub use prepare::Prepare;
27pub use query::Query;
28pub use startup::Startup;
29
30use self::batch::BatchStatement;
31
32use super::TryFromPrimitiveError;
33use super::frame_errors::{CqlRequestSerializationError, LowLevelDeserializationError};
34use super::types::SerialConsistency;
35
36// Re-export from scylla-cql-core for backward compatibility.
37pub use scylla_cql_core::frame::request::CqlRequestKind;
38
39/// Opcode of a request, used to identify the request type in a CQL frame.
40#[derive(Debug, Copy, Clone, PartialEq, Eq, PartialOrd, Ord)]
41#[repr(u8)]
42pub enum RequestOpcode {
43    /// See [CqlRequestKind::Startup].
44    Startup = 0x01,
45    /// See [CqlRequestKind::Options].
46    Options = 0x05,
47    /// See [CqlRequestKind::Query].
48    Query = 0x07,
49    /// See [CqlRequestKind::Prepare].
50    Prepare = 0x09,
51    /// See [CqlRequestKind::Execute].
52    Execute = 0x0A,
53    /// See [CqlRequestKind::Register].
54    Register = 0x0B,
55    /// See [CqlRequestKind::Batch].
56    Batch = 0x0D,
57    /// See [CqlRequestKind::AuthResponse].
58    AuthResponse = 0x0F,
59}
60
61impl TryFrom<u8> for RequestOpcode {
62    type Error = TryFromPrimitiveError<u8>;
63
64    fn try_from(value: u8) -> Result<Self, Self::Error> {
65        match value {
66            0x01 => Ok(Self::Startup),
67            0x05 => Ok(Self::Options),
68            0x07 => Ok(Self::Query),
69            0x09 => Ok(Self::Prepare),
70            0x0A => Ok(Self::Execute),
71            0x0B => Ok(Self::Register),
72            0x0D => Ok(Self::Batch),
73            0x0F => Ok(Self::AuthResponse),
74            _ => Err(TryFromPrimitiveError::new("RequestOpcode", value)),
75        }
76    }
77}
78
79/// Requests that can be serialized into a CQL frame.
80pub trait SerializableRequest {
81    /// Opcode of the request, used to identify the request type in the CQL frame.
82    const OPCODE: RequestOpcode;
83
84    /// Serializes the request into the provided buffer.
85    fn serialize(&self, buf: &mut Vec<u8>) -> Result<(), CqlRequestSerializationError>;
86
87    /// Serializes the request into a heap-allocated `Bytes` object.
88    fn to_bytes(&self) -> Result<Bytes, CqlRequestSerializationError> {
89        let mut v = Vec::new();
90        self.serialize(&mut v)?;
91        Ok(v.into())
92    }
93}
94
95/// Requests that can be deserialized from a CQL frame.
96///
97/// Not intended for driver's direct usage (as driver has no interest in deserialising CQL requests),
98/// but very useful for testing (e.g. asserting that the sent requests have proper parameters set).
99pub trait DeserializableRequest: SerializableRequest + Sized {
100    /// Deserializes the request from the provided buffer.
101    /// Use [DeserializableRequest::deserialize_with_features] instead, because some frame types
102    /// require knowing protocol features for correct deserialization.
103    #[deprecated(since = "1.4.0", note = "Use deserialize_with_features instead")]
104    fn deserialize(buf: &mut &[u8]) -> Result<Self, RequestDeserializationError>;
105
106    fn deserialize_with_features(
107        buf: &mut &[u8],
108        #[allow(unused_variables)] features: &ProtocolFeatures,
109    ) -> Result<Self, RequestDeserializationError> {
110        #[expect(deprecated)]
111        Self::deserialize(buf)
112    }
113}
114
115/// An error type returned by [`DeserializableRequest::deserialize`].
116/// This is not intended for driver's direct usage. It's a testing utility,
117/// mainly used by `scylla-proxy` crate.
118#[doc(hidden)]
119#[derive(Debug, Error)]
120pub enum RequestDeserializationError {
121    #[error("Low level deser error: {0}")]
122    LowLevelDeserialization(#[from] LowLevelDeserializationError),
123    #[error("Io error: {0}")]
124    IoError(#[from] std::io::Error),
125    #[error("Specified flags are not recognised: {:02x}", flags)]
126    UnknownFlags { flags: u8 },
127    #[error("Named values in frame are currently unsupported")]
128    NamedValuesUnsupported,
129    #[error("Expected SerialConsistency, got regular Consistency: {0}")]
130    ExpectedSerialConsistency(Consistency),
131    #[error(transparent)]
132    BatchTypeParse(#[from] BatchTypeParseError),
133    #[error("Unexpected batch statement kind: {0}")]
134    UnexpectedBatchStatementKind(u8),
135}
136
137/// A CQL request that can be sent to the server.
138#[non_exhaustive] // TODO: add remaining request types
139#[deprecated(
140    since = "1.4.0",
141    note = "Does not support Scylla metadata id extension. Use RequestV2 instead."
142)]
143pub enum Request<'r> {
144    /// QUERY request, used to execute a single unprepared statement.
145    Query(Query<'r>),
146    /// EXECUTE request, used to execute a single prepared statement.
147    #[expect(deprecated)]
148    Execute(Execute<'r>),
149    /// BATCH request, used to execute a batch of (prepared, unprepared, or mix of both)
150    /// statements.
151    Batch(Batch<'r, BatchStatement<'r>, Vec<SerializedValues>>),
152}
153
154#[expect(deprecated)]
155impl Request<'_> {
156    /// Deserializes the request from the provided buffer.
157    pub fn deserialize(
158        buf: &mut &[u8],
159        opcode: RequestOpcode,
160    ) -> Result<Self, RequestDeserializationError> {
161        match opcode {
162            RequestOpcode::Query => Query::deserialize(buf).map(Self::Query),
163            RequestOpcode::Execute => Execute::deserialize(buf).map(Self::Execute),
164            RequestOpcode::Batch => Batch::deserialize(buf).map(Self::Batch),
165            _ => unimplemented!(
166                "Deserialization of opcode {:?} is not yet supported",
167                opcode
168            ),
169        }
170    }
171
172    /// Retrieves consistency from request frame, if present.
173    pub fn get_consistency(&self) -> Option<Consistency> {
174        match self {
175            Request::Query(q) => Some(q.parameters.consistency),
176            Request::Execute(e) => Some(e.parameters.consistency),
177            Request::Batch(b) => Some(b.consistency),
178            #[expect(unreachable_patterns)] // until other opcodes are supported
179            _ => None,
180        }
181    }
182
183    /// Retrieves serial consistency from request frame.
184    pub fn get_serial_consistency(&self) -> Option<Option<SerialConsistency>> {
185        match self {
186            Request::Query(q) => Some(q.parameters.serial_consistency),
187            Request::Execute(e) => Some(e.parameters.serial_consistency),
188            Request::Batch(b) => Some(b.serial_consistency),
189            #[expect(unreachable_patterns)] // until other opcodes are supported
190            _ => None,
191        }
192    }
193}
194
195/// A CQL request that can be sent to the server.
196#[non_exhaustive] // TODO: add remaining request types
197pub enum RequestV2<'r> {
198    /// QUERY request, used to execute a single unprepared statement.
199    Query(Query<'r>),
200    /// EXECUTE request, used to execute a single prepared statement.
201    Execute(ExecuteV2<'r>),
202    /// BATCH request, used to execute a batch of (prepared, unprepared, or mix of both)
203    /// statements.
204    Batch(Batch<'r, BatchStatement<'r>, Vec<SerializedValues>>),
205}
206
207impl RequestV2<'_> {
208    /// Deserializes the request from the provided buffer.
209    pub fn deserialize(
210        buf: &mut &[u8],
211        opcode: RequestOpcode,
212        features: &ProtocolFeatures,
213    ) -> Result<Self, RequestDeserializationError> {
214        match opcode {
215            RequestOpcode::Query => {
216                Query::deserialize_with_features(buf, features).map(Self::Query)
217            }
218            RequestOpcode::Execute => {
219                ExecuteV2::deserialize_with_features(buf, features).map(Self::Execute)
220            }
221            RequestOpcode::Batch => {
222                Batch::deserialize_with_features(buf, features).map(Self::Batch)
223            }
224            _ => unimplemented!(
225                "Deserialization of opcode {:?} is not yet supported",
226                opcode
227            ),
228        }
229    }
230
231    /// Retrieves consistency from request frame, if present.
232    pub fn get_consistency(&self) -> Option<Consistency> {
233        match self {
234            Self::Query(q) => Some(q.parameters.consistency),
235            Self::Execute(e) => Some(e.parameters.consistency),
236            Self::Batch(b) => Some(b.consistency),
237            #[expect(unreachable_patterns)] // until other opcodes are supported
238            _ => None,
239        }
240    }
241
242    /// Retrieves serial consistency from request frame.
243    pub fn get_serial_consistency(&self) -> Option<Option<SerialConsistency>> {
244        match self {
245            Self::Query(q) => Some(q.parameters.serial_consistency),
246            Self::Execute(e) => Some(e.parameters.serial_consistency),
247            Self::Batch(b) => Some(b.serial_consistency),
248            #[expect(unreachable_patterns)] // until other opcodes are supported
249            _ => None,
250        }
251    }
252}
253
254#[cfg(test)]
255mod tests {
256    use std::{borrow::Cow, ops::Deref};
257
258    use bytes::Bytes;
259
260    use super::query::PagingState;
261    use crate::Consistency;
262    use crate::frame::protocol_features::ProtocolFeatures;
263    use crate::frame::request::batch::{Batch, BatchStatement, BatchType};
264    #[expect(deprecated)]
265    use crate::frame::request::execute::Execute;
266    use crate::frame::request::execute::ExecuteV2;
267    use crate::frame::request::query::{Query, QueryParameters};
268    use crate::frame::request::{DeserializableRequest, SerializableRequest};
269    use crate::frame::response::result::{ColumnType, NativeType};
270    use crate::frame::types::{self, SerialConsistency};
271    use crate::serialize::row::SerializedValues;
272
273    #[test]
274    fn request_ser_de_identity() {
275        // Query
276        let contents = Cow::Borrowed("SELECT host_id from system.peers");
277        let parameters = QueryParameters {
278            consistency: Consistency::All,
279            serial_consistency: Some(SerialConsistency::Serial),
280            timestamp: None,
281            page_size: Some(323),
282            paging_state: PagingState::new_from_raw_bytes(&[2_u8, 1, 3, 7] as &[u8]),
283            skip_metadata: false,
284            values: {
285                let mut vals = SerializedValues::new();
286                vals.add_value(&2137, &ColumnType::Native(NativeType::Int))
287                    .unwrap();
288                Cow::Owned(vals)
289            },
290        };
291        let query = Query {
292            contents,
293            parameters,
294        };
295
296        {
297            let mut buf = Vec::new();
298            query.serialize(&mut buf).unwrap();
299
300            let query_deserialized =
301                Query::deserialize_with_features(&mut &buf[..], &Default::default()).unwrap();
302            assert_eq!(&query_deserialized, &query);
303        }
304
305        // Legacy Execute
306        let id: Bytes = vec![2, 4, 5, 2, 6, 7, 3, 1].into();
307        let parameters = QueryParameters {
308            consistency: Consistency::Any,
309            serial_consistency: None,
310            timestamp: Some(3423434),
311            page_size: None,
312            paging_state: PagingState::start(),
313            skip_metadata: false,
314            values: {
315                let mut vals = SerializedValues::new();
316                vals.add_value(&42, &ColumnType::Native(NativeType::Int))
317                    .unwrap();
318                vals.add_value(&2137, &ColumnType::Native(NativeType::Int))
319                    .unwrap();
320                Cow::Owned(vals)
321            },
322        };
323
324        #[expect(deprecated)]
325        let execute = Execute {
326            id,
327            parameters: parameters.clone(),
328        };
329        {
330            let mut buf = Vec::new();
331            execute.serialize(&mut buf).unwrap();
332
333            #[expect(deprecated)]
334            let execute_deserialized =
335                Execute::deserialize_with_features(&mut &buf[..], &Default::default()).unwrap();
336            assert_eq!(&execute_deserialized, &execute);
337        }
338
339        // New Execute
340        let id = [2, 4, 5, 2, 6, 7, 3, 1].as_slice().into();
341        let result_metadata_id = Some([2, 4, 5, 2, 6, 7, 3, 1].as_slice().into());
342        let execute_with_id = ExecuteV2 {
343            id,
344            result_metadata_id,
345            parameters,
346            tablet_version_block: None,
347        };
348        {
349            let mut buf = Vec::new();
350            execute_with_id.serialize(&mut buf).unwrap();
351
352            let features = {
353                let mut default = ProtocolFeatures::default();
354                default.scylla_metadata_id_supported = true;
355                default
356            };
357            let execute_deserialized =
358                ExecuteV2::deserialize_with_features(&mut &buf[..], &features).unwrap();
359            assert_eq!(&execute_deserialized, &execute_with_id);
360        }
361
362        // Batch
363        let statements = vec![
364            BatchStatement::Query {
365                text: query.contents,
366            },
367            BatchStatement::Prepared {
368                id: Cow::Borrowed(execute_with_id.id.as_ref()),
369            },
370        ];
371        let batch = Batch {
372            statements: Cow::Owned(statements),
373            batch_type: BatchType::Logged,
374            consistency: Consistency::EachQuorum,
375            serial_consistency: Some(SerialConsistency::LocalSerial),
376            timestamp: Some(32432),
377
378            // Not execute's values, because named values are not supported in batches.
379            values: vec![
380                query.parameters.values.deref().clone(),
381                query.parameters.values.deref().clone(),
382            ],
383        };
384        {
385            let mut buf = Vec::new();
386            batch.serialize(&mut buf).unwrap();
387
388            let batch_deserialized =
389                Batch::deserialize_with_features(&mut &buf[..], &Default::default()).unwrap();
390            assert_eq!(&batch_deserialized, &batch);
391        }
392    }
393
394    #[test]
395    fn deser_rejects_unknown_flags() {
396        // Query
397        let contents = Cow::Borrowed("SELECT host_id from system.peers");
398        let parameters = QueryParameters {
399            consistency: Default::default(),
400            serial_consistency: Some(SerialConsistency::LocalSerial),
401            timestamp: None,
402            page_size: None,
403            paging_state: PagingState::start(),
404            skip_metadata: false,
405            values: Cow::Borrowed(SerializedValues::EMPTY),
406        };
407        let query = Query {
408            contents: contents.clone(),
409            parameters,
410        };
411
412        {
413            let mut buf = Vec::new();
414            query.serialize(&mut buf).unwrap();
415
416            // Sanity check: query deserializes to the equivalent.
417            let query_deserialized =
418                Query::deserialize_with_features(&mut &buf[..], &Default::default()).unwrap();
419            assert_eq!(&query_deserialized.contents, &query.contents);
420            assert_eq!(&query_deserialized.parameters, &query.parameters);
421
422            // Now modify flags by adding an unknown one.
423            // Find flags in buffer:
424            let mut buf_ptr = buf.as_slice();
425            let serialised_contents = types::read_long_string(&mut buf_ptr).unwrap();
426            assert_eq!(serialised_contents, contents);
427
428            // Now buf_ptr points at consistency.
429            let consistency = types::read_consistency(&mut buf_ptr).unwrap();
430            assert_eq!(consistency, Consistency::default());
431
432            // Now buf_ptr points at flags, but it is immutable. Get mutable reference into the buffer.
433            let flags_idx = buf.len() - buf_ptr.len();
434            let flags_mut = &mut buf[flags_idx];
435
436            // This assumes that the following flag is unknown, which is true at the time of writing this test.
437            *flags_mut |= 0x80;
438
439            // Unknown flag should lead to frame rejection, as unknown flags can be new protocol extensions
440            // leading to different semantics.
441            let _parse_error =
442                Query::deserialize_with_features(&mut &buf[..], &Default::default()).unwrap_err();
443        }
444
445        // Batch
446        let statements = vec![BatchStatement::Query {
447            text: query.contents,
448        }];
449        let batch = Batch {
450            statements: Cow::Owned(statements),
451            batch_type: BatchType::Logged,
452            consistency: Consistency::EachQuorum,
453            serial_consistency: None,
454            timestamp: None,
455
456            values: vec![query.parameters.values.deref().clone()],
457        };
458        {
459            let mut buf = Vec::new();
460            batch.serialize(&mut buf).unwrap();
461
462            // Sanity check: batch deserializes to the equivalent.
463            let batch_deserialized =
464                Batch::deserialize_with_features(&mut &buf[..], &Default::default()).unwrap();
465            assert_eq!(batch, batch_deserialized);
466
467            // Now modify flags by adding an unknown one.
468            // There are no timestamp nor serial consistency, so flags are the last byte in the buf.
469            let buf_len = buf.len();
470            let flags_mut = &mut buf[buf_len - 1];
471            // This assumes that the following flag is unknown, which is true at the time of writing this test.
472            *flags_mut |= 0x80;
473
474            // Unknown flag should lead to frame rejection, as unknown flags can be new protocol extensions
475            // leading to different semantics.
476            let _parse_error =
477                Batch::deserialize_with_features(&mut &buf[..], &Default::default()).unwrap_err();
478        }
479    }
480}
scylla_cql/frame/request/mod.rs

scylla_cql/frame/request/
mod.rs
