questdb/egress/wire/
msg_kind.rs

1/*******************************************************************************
2 *     ___                  _   ____  ____
3 *    / _ \ _   _  ___  ___| |_|  _ \| __ )
4 *   | | | | | | |/ _ \/ __| __| | | |  _ \
5 *   | |_| | |_| |  __/\__ \ |_| |_| | |_) |
6 *    \__\_\\__,_|\___||___/\__|____/|____/
7 *
8 *  Copyright (c) 2014-2019 Appsicle
9 *  Copyright (c) 2019-2025 QuestDB
10 *
11 *  Licensed under the Apache License, Version 2.0 (the "License");
12 *  you may not use this file except in compliance with the License.
13 *  You may obtain a copy of the License at
14 *
15 *  http://www.apache.org/licenses/LICENSE-2.0
16 *
17 *  Unless required by applicable law or agreed to in writing, software
18 *  distributed under the License is distributed on an "AS IS" BASIS,
19 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
20 *  See the License for the specific language governing permissions and
21 *  limitations under the License.
22 *
23 ******************************************************************************/
24
25//! Message kind discriminator (first byte of frame payload).
26//!
27//! ABI-stable: variants append-only, never reorder.
28
29use crate::error::{Result, fmt};
30
31/// Message kind code (uint8). `repr(u8)` keeps wire transcoding trivial.
32///
33/// `#[non_exhaustive]` because the QWP message-kind table is
34/// append-only across protocol revisions.
35#[repr(u8)]
36#[derive(Debug, Copy, Clone, PartialEq, Eq, Hash)]
37#[non_exhaustive]
38pub enum MsgKind {
39    /// Client → Server: initiate cursor with SQL + binds.
40    QueryRequest = 0x10,
41    /// Server → Client: one table block of results.
42    ResultBatch = 0x11,
43    /// Server → Client: successful stream termination.
44    ResultEnd = 0x12,
45    /// Server → Client: failure at any lifecycle point.
46    QueryError = 0x13,
47    /// Client → Server: request query termination.
48    Cancel = 0x14,
49    /// Client → Server: extend byte-credit window.
50    Credit = 0x15,
51    /// Server → Client: non-SELECT acknowledgement.
52    ExecDone = 0x16,
53    /// Server → Client: clear connection caches.
54    CacheReset = 0x17,
55    /// Server → Client: role + cluster identity.
56    ServerInfo = 0x18,
57}
58
59impl MsgKind {
60    /// Parse a wire byte into a known kind.
61    pub fn from_u8(byte: u8) -> Result<Self> {
62        Ok(match byte {
63            0x10 => MsgKind::QueryRequest,
64            0x11 => MsgKind::ResultBatch,
65            0x12 => MsgKind::ResultEnd,
66            0x13 => MsgKind::QueryError,
67            0x14 => MsgKind::Cancel,
68            0x15 => MsgKind::Credit,
69            0x16 => MsgKind::ExecDone,
70            0x17 => MsgKind::CacheReset,
71            0x18 => MsgKind::ServerInfo,
72            other => return Err(fmt!(ProtocolError, "unknown msg_kind 0x{:02X}", other)),
73        })
74    }
75
76    /// Wire byte for this kind.
77    pub fn as_u8(self) -> u8 {
78        self as u8
79    }
80}
81
82/// QWP status codes carried by `QUERY_ERROR` (and surfaced to clients).
83///
84/// `#[non_exhaustive]` because the status table is append-only across
85/// protocol revisions.
86#[repr(u8)]
87#[derive(Debug, Copy, Clone, PartialEq, Eq, Hash)]
88#[non_exhaustive]
89pub enum StatusCode {
90    SchemaMismatch = 0x03,
91    ParseError = 0x05,
92    InternalError = 0x06,
93    SecurityError = 0x08,
94    Cancelled = 0x0A,
95    LimitExceeded = 0x0B,
96}
97
98impl StatusCode {
99    pub fn from_u8(byte: u8) -> Result<Self> {
100        Ok(match byte {
101            0x03 => StatusCode::SchemaMismatch,
102            0x05 => StatusCode::ParseError,
103            0x06 => StatusCode::InternalError,
104            0x08 => StatusCode::SecurityError,
105            0x0A => StatusCode::Cancelled,
106            0x0B => StatusCode::LimitExceeded,
107            other => {
108                return Err(fmt!(
109                    ProtocolError,
110                    "unknown QWP status code 0x{:02X}",
111                    other
112                ));
113            }
114        })
115    }
116
117    pub fn as_u8(self) -> u8 {
118        self as u8
119    }
120}
121
122#[cfg(test)]
123mod tests {
124    use super::*;
125
126    #[test]
127    fn msg_kind_roundtrip() {
128        for &k in &[
129            MsgKind::QueryRequest,
130            MsgKind::ResultBatch,
131            MsgKind::ResultEnd,
132            MsgKind::QueryError,
133            MsgKind::Cancel,
134            MsgKind::Credit,
135            MsgKind::ExecDone,
136            MsgKind::CacheReset,
137            MsgKind::ServerInfo,
138        ] {
139            let b = k.as_u8();
140            assert_eq!(MsgKind::from_u8(b).unwrap(), k);
141        }
142    }
143
144    #[test]
145    fn unknown_msg_kind_rejected() {
146        assert!(MsgKind::from_u8(0x00).is_err());
147        assert!(MsgKind::from_u8(0xFF).is_err());
148        assert!(MsgKind::from_u8(0x09).is_err());
149    }
150
151    #[test]
152    fn status_code_roundtrip() {
153        for &s in &[
154            StatusCode::SchemaMismatch,
155            StatusCode::ParseError,
156            StatusCode::InternalError,
157            StatusCode::SecurityError,
158            StatusCode::Cancelled,
159            StatusCode::LimitExceeded,
160        ] {
161            assert_eq!(StatusCode::from_u8(s.as_u8()).unwrap(), s);
162        }
163    }
164}
questdb/egress/wire/msg_kind.rs

questdb/egress/wire/
msg_kind.rs
