netget/server/clickhouse/
wire.rs

1//! ClickHouse native TCP protocol, at protocol revision 54429 (the first with settings
2//! serialized as strings, and before the later revision-dependent query fields): hello,
3//! query, data blocks with typed columns, compressed frames (LZ4, CityHash128 v1.0.2
4//! checksums), exceptions, progress and end of stream. Shared by server and client.
5//!
6//! The protocol has no frame lengths, so every read is bounded where it happens: strings,
7//! column and row counts, each compressed frame, and the total bytes one block may consume.
8use anyhow::{bail, ensure, Context, Result};
9use serde_json::{json, Value};
10use tokio::io::{AsyncRead, AsyncReadExt};
11
12/// The protocol revision NetGet speaks in both roles; the peer uses the minimum of the two.
13pub const REVISION: u64 = 54429;
14pub const VERSION_MAJOR: u64 = 24;
15pub const VERSION_MINOR: u64 = 8;
16pub const VERSION_PATCH: u64 = 0;
17/// Longest string field (a query, a password, a String value).
18pub const MAX_STRING: usize = 1024 * 1024;
19/// Bytes one block may consume, compressed frames included.
20pub const MAX_BLOCK_BYTES: usize = 16 * 1024 * 1024;
21pub const MAX_COLUMNS: usize = 1000;
22pub const MAX_ROWS: usize = 100_000;
23/// Settings (or other name/value pairs) in one query.
24pub const MAX_SETTINGS: usize = 1000;
25/// Deadline for packets that must follow at once (a query's external-tables terminator).
26pub const IO_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
27
28pub mod client_packet {
29    pub const HELLO: u64 = 0;
30    pub const QUERY: u64 = 1;
31    pub const DATA: u64 = 2;
32    pub const CANCEL: u64 = 3;
33    pub const PING: u64 = 4;
34}
35
36pub mod server_packet {
37    pub const HELLO: u64 = 0;
38    pub const DATA: u64 = 1;
39    pub const EXCEPTION: u64 = 2;
40    pub const PROGRESS: u64 = 3;
41    pub const PONG: u64 = 4;
42    pub const END_OF_STREAM: u64 = 5;
43    pub const PROFILE_INFO: u64 = 6;
44    pub const TOTALS: u64 = 7;
45    pub const EXTREMES: u64 = 8;
46    pub const LOG: u64 = 10;
47    pub const TABLE_COLUMNS: u64 = 11;
48}
49
50pub fn put_varuint(out: &mut Vec<u8>, mut v: u64) {
51    loop {
52        let b = (v & 0x7f) as u8;
53        v >>= 7;
54        if v == 0 {
55            out.push(b);
56            return;
57        }
58        out.push(b | 0x80);
59    }
60}
61
62pub fn put_string(out: &mut Vec<u8>, s: &[u8]) {
63    put_varuint(out, s.len() as u64);
64    out.extend_from_slice(s);
65}
66
67/// Bytes read from the peer, from the socket directly or through compressed frames, against
68/// a budget.
69pub struct Source<'a, R> {
70    inner: &'a mut R,
71    compressed: bool,
72    buf: Vec<u8>,
73    pos: usize,
74    budget: usize,
75}
76
77impl<'a, R: AsyncRead + Unpin> Source<'a, R> {
78    pub fn plain(inner: &'a mut R) -> Self {
79        Self::new(inner, false)
80    }
81    pub fn new(inner: &'a mut R, compressed: bool) -> Self {
82        Self {
83            inner,
84            compressed,
85            buf: Vec::new(),
86            pos: 0,
87            budget: MAX_BLOCK_BYTES,
88        }
89    }
90
91    fn spend(&mut self, n: usize) -> Result<()> {
92        ensure!(
93            n <= self.budget,
94            "ClickHouse block exceeds {MAX_BLOCK_BYTES} bytes"
95        );
96        self.budget -= n;
97        Ok(())
98    }
99
100    async fn refill(&mut self) -> Result<()> {
101        let mut header = [0u8; 25];
102        self.inner.read_exact(&mut header).await?;
103        let method = header[16];
104        let compressed_size = u32::from_le_bytes(header[17..21].try_into()?) as usize;
105        let size = u32::from_le_bytes(header[21..25].try_into()?) as usize;
106        ensure!(
107            compressed_size >= 9,
108            "compressed frame smaller than its header"
109        );
110        self.spend(compressed_size + 16)?;
111        ensure!(
112            size <= self.budget,
113            "compressed frame decompresses past the block bound"
114        );
115        let mut frame = Vec::with_capacity(compressed_size);
116        frame.extend_from_slice(&header[16..25]);
117        frame.resize(compressed_size, 0);
118        self.inner.read_exact(&mut frame[9..]).await?;
119        let sum = naive_cityhash::cityhash128(&frame);
120        let mut expected = [0u8; 16];
121        expected[..8].copy_from_slice(&sum.lo.to_le_bytes());
122        expected[8..].copy_from_slice(&sum.hi.to_le_bytes());
123        ensure!(
124            expected == header[..16],
125            "compressed frame checksum mismatch"
126        );
127        let data = match method {
128            0x02 => frame[9..].to_vec(),
129            0x82 => lz4_flex::block::decompress(&frame[9..], size)
130                .map_err(|e| anyhow::anyhow!("LZ4 frame: {e}"))?,
131            other => bail!("compression method 0x{other:02x} is not supported (LZ4 or none)"),
132        };
133        ensure!(
134            data.len() == size,
135            "compressed frame decompressed to the wrong size"
136        );
137        self.buf.drain(..self.pos);
138        self.pos = 0;
139        self.buf.extend_from_slice(&data);
140        Ok(())
141    }
142
143    pub async fn bytes(&mut self, n: usize) -> Result<Vec<u8>> {
144        if self.compressed {
145            while self.buf.len() - self.pos < n {
146                self.refill().await?;
147            }
148            let out = self.buf[self.pos..self.pos + n].to_vec();
149            self.pos += n;
150            Ok(out)
151        } else {
152            self.spend(n)?;
153            let mut out = vec![0u8; n];
154            self.inner.read_exact(&mut out).await?;
155            Ok(out)
156        }
157    }
158    pub async fn u8(&mut self) -> Result<u8> {
159        Ok(self.bytes(1).await?[0])
160    }
161    pub async fn varuint(&mut self) -> Result<u64> {
162        let mut v = 0u64;
163        for i in 0..10 {
164            let b = self.u8().await?;
165            v |= u64::from(b & 0x7f) << (7 * i);
166            if b & 0x80 == 0 {
167                return Ok(v);
168            }
169        }
170        bail!("VarUInt longer than 10 bytes")
171    }
172    pub async fn string(&mut self) -> Result<String> {
173        let b = self.raw_string().await?;
174        Ok(String::from_utf8_lossy(&b).to_string())
175    }
176    pub async fn raw_string(&mut self) -> Result<Vec<u8>> {
177        let n = self.varuint().await? as usize;
178        ensure!(n <= MAX_STRING, "string of {n} bytes exceeds {MAX_STRING}");
179        self.bytes(n).await
180    }
181    /// Whether every compressed byte read so far has been consumed (a block ends on a frame).
182    pub fn drained(&self) -> bool {
183        self.pos == self.buf.len()
184    }
185}
186
187/// The first packet type of a message, waiting at most `idle`. `None` on a clean EOF.
188pub async fn read_packet_type<R: AsyncRead + Unpin>(
189    r: &mut R,
190    idle: std::time::Duration,
191) -> Result<Option<u64>> {
192    let mut first = [0u8; 1];
193    match tokio::time::timeout(idle, r.read(&mut first)).await {
194        Ok(Ok(0)) => return Ok(None),
195        Ok(Ok(_)) => {}
196        Ok(Err(e)) => return Err(e.into()),
197        Err(_) => bail!("ClickHouse peer idle for {}s", idle.as_secs()),
198    }
199    let mut v = u64::from(first[0] & 0x7f);
200    let mut b = first[0];
201    let mut shift = 7;
202    while b & 0x80 != 0 {
203        ensure!(shift < 64, "VarUInt longer than 10 bytes");
204        b = r.read_u8().await?;
205        v |= u64::from(b & 0x7f) << shift;
206        shift += 7;
207    }
208    Ok(Some(v))
209}
210
211/// A column type NetGet can encode and decode.
212#[derive(Debug, Clone, PartialEq, Eq)]
213pub enum ColType {
214    UInt8,
215    UInt16,
216    UInt32,
217    UInt64,
218    Int8,
219    Int16,
220    Int32,
221    Int64,
222    Float32,
223    Float64,
224    Bool,
225    String,
226    Date,
227    DateTime,
228    Nullable(Box<ColType>),
229}
230
231impl ColType {
232    pub fn parse(t: &str) -> Result<Self> {
233        let t = t.trim();
234        if let Some(inner) = t
235            .strip_prefix("Nullable(")
236            .and_then(|r| r.strip_suffix(')'))
237        {
238            let inner = Self::parse(inner)?;
239            ensure!(
240                !matches!(inner, ColType::Nullable(_)),
241                "Nullable cannot nest"
242            );
243            return Ok(ColType::Nullable(Box::new(inner)));
244        }
245        if t.starts_with("DateTime(") && !t.starts_with("DateTime64") {
246            return Ok(ColType::DateTime);
247        }
248        Ok(match t {
249            "UInt8" => ColType::UInt8,
250            "UInt16" => ColType::UInt16,
251            "UInt32" => ColType::UInt32,
252            "UInt64" => ColType::UInt64,
253            "Int8" => ColType::Int8,
254            "Int16" => ColType::Int16,
255            "Int32" => ColType::Int32,
256            "Int64" => ColType::Int64,
257            "Float32" => ColType::Float32,
258            "Float64" => ColType::Float64,
259            "Bool" => ColType::Bool,
260            "String" => ColType::String,
261            "Date" => ColType::Date,
262            "DateTime" => ColType::DateTime,
263            other => bail!("column type {other} is not supported (integers, floats, Bool, String, Date, DateTime, Nullable of those)"),
264        })
265    }
266
267    fn width(&self) -> usize {
268        match self {
269            ColType::UInt8 | ColType::Int8 | ColType::Bool => 1,
270            ColType::UInt16 | ColType::Int16 | ColType::Date => 2,
271            ColType::UInt32 | ColType::Int32 | ColType::Float32 | ColType::DateTime => 4,
272            ColType::UInt64 | ColType::Int64 | ColType::Float64 => 8,
273            ColType::String | ColType::Nullable(_) => 0,
274        }
275    }
276
277    /// Encode one non-null value of this type from JSON.
278    fn encode(&self, v: &Value, out: &mut Vec<u8>) -> Result<()> {
279        let int = |v: &Value| -> Result<i128> {
280            v.as_i64()
281                .map(i128::from)
282                .or_else(|| v.as_u64().map(i128::from))
283                .or_else(|| v.as_str().and_then(|s| s.parse().ok()))
284                .context("expected an integer")
285        };
286        let ranged = |lo: i128, hi: i128| -> Result<i128> {
287            let n = int(v)?;
288            ensure!((lo..=hi).contains(&n), "{n} out of range for this column");
289            Ok(n)
290        };
291        match self {
292            ColType::UInt8 => out.push(ranged(0, u8::MAX.into())? as u8),
293            ColType::UInt16 => out.extend((ranged(0, u16::MAX.into())? as u16).to_le_bytes()),
294            ColType::UInt32 => out.extend((ranged(0, u32::MAX.into())? as u32).to_le_bytes()),
295            ColType::UInt64 => out.extend((ranged(0, u64::MAX.into())? as u64).to_le_bytes()),
296            ColType::Int8 => {
297                out.extend((ranged(i8::MIN.into(), i8::MAX.into())? as i8).to_le_bytes())
298            }
299            ColType::Int16 => {
300                out.extend((ranged(i16::MIN.into(), i16::MAX.into())? as i16).to_le_bytes())
301            }
302            ColType::Int32 => {
303                out.extend((ranged(i32::MIN.into(), i32::MAX.into())? as i32).to_le_bytes())
304            }
305            ColType::Int64 => {
306                out.extend((ranged(i64::MIN.into(), i64::MAX.into())? as i64).to_le_bytes())
307            }
308            ColType::Float32 => {
309                out.extend((v.as_f64().context("expected a number")? as f32).to_le_bytes())
310            }
311            ColType::Float64 => out.extend(v.as_f64().context("expected a number")?.to_le_bytes()),
312            ColType::Bool => out.push(u8::from(v.as_bool().context("expected a boolean")?)),
313            ColType::String => {
314                let s = match v {
315                    Value::String(s) => s.clone(),
316                    other => other.to_string(),
317                };
318                ensure!(s.len() <= MAX_STRING, "string value too long");
319                put_string(out, s.as_bytes());
320            }
321            ColType::Date => {
322                let d = chrono::NaiveDate::parse_from_str(
323                    v.as_str().context("expected YYYY-MM-DD")?,
324                    "%Y-%m-%d",
325                )?;
326                let days =
327                    (d - chrono::NaiveDate::from_ymd_opt(1970, 1, 1).context("epoch")?).num_days();
328                ensure!(
329                    (0..=i64::from(u16::MAX)).contains(&days),
330                    "date out of range"
331                );
332                out.extend((days as u16).to_le_bytes());
333            }
334            ColType::DateTime => {
335                let t = chrono::NaiveDateTime::parse_from_str(
336                    v.as_str().context("expected YYYY-MM-DD hh:mm:ss")?,
337                    "%Y-%m-%d %H:%M:%S",
338                )?;
339                let secs = t.and_utc().timestamp();
340                ensure!(
341                    (0..=i64::from(u32::MAX)).contains(&secs),
342                    "datetime out of range"
343                );
344                out.extend((secs as u32).to_le_bytes());
345            }
346            ColType::Nullable(_) => bail!("nested Nullable"),
347        }
348        Ok(())
349    }
350
351    fn default_bytes(&self, out: &mut Vec<u8>) {
352        match self {
353            ColType::String => out.push(0),
354            other => out.extend(std::iter::repeat_n(0, other.width())),
355        }
356    }
357
358    /// Encode a whole column of `rows` values.
359    pub fn encode_column(&self, values: &[&Value]) -> Result<Vec<u8>> {
360        let mut out = Vec::new();
361        match self {
362            ColType::Nullable(inner) => {
363                for v in values {
364                    out.push(u8::from(v.is_null()));
365                }
366                for v in values {
367                    if v.is_null() {
368                        inner.default_bytes(&mut out);
369                    } else {
370                        inner.encode(v, &mut out)?;
371                    }
372                }
373            }
374            other => {
375                for v in values {
376                    ensure!(!v.is_null(), "null in a non-Nullable column");
377                    other.encode(v, &mut out)?;
378                }
379            }
380        }
381        Ok(out)
382    }
383
384    async fn decode_value<R: AsyncRead + Unpin>(&self, s: &mut Source<'_, R>) -> Result<Value> {
385        let w = self.width();
386        Ok(match self {
387            ColType::String => {
388                let b = s.raw_string().await?;
389                match String::from_utf8(b) {
390                    Ok(t) => json!(t),
391                    Err(e) => json!(e
392                        .into_bytes()
393                        .iter()
394                        .map(|x| format!("{x:02x}"))
395                        .collect::<String>()),
396                }
397            }
398            _ => {
399                let b = s.bytes(w).await?;
400                match self {
401                    ColType::UInt8 => json!(b[0]),
402                    ColType::UInt16 => json!(u16::from_le_bytes(b[..2].try_into()?)),
403                    ColType::UInt32 => json!(u32::from_le_bytes(b[..4].try_into()?)),
404                    ColType::UInt64 => json!(u64::from_le_bytes(b[..8].try_into()?)),
405                    ColType::Int8 => json!(b[0] as i8),
406                    ColType::Int16 => json!(i16::from_le_bytes(b[..2].try_into()?)),
407                    ColType::Int32 => json!(i32::from_le_bytes(b[..4].try_into()?)),
408                    ColType::Int64 => json!(i64::from_le_bytes(b[..8].try_into()?)),
409                    ColType::Float32 => json!(f32::from_le_bytes(b[..4].try_into()?)),
410                    ColType::Float64 => json!(f64::from_le_bytes(b[..8].try_into()?)),
411                    ColType::Bool => json!(b[0] != 0),
412                    ColType::Date => {
413                        let days = u16::from_le_bytes(b[..2].try_into()?);
414                        let d = chrono::NaiveDate::from_ymd_opt(1970, 1, 1).context("epoch")?
415                            + chrono::Duration::days(i64::from(days));
416                        json!(d.format("%Y-%m-%d").to_string())
417                    }
418                    ColType::DateTime => {
419                        let secs = u32::from_le_bytes(b[..4].try_into()?);
420                        let t = chrono::DateTime::from_timestamp(i64::from(secs), 0)
421                            .context("timestamp")?;
422                        json!(t.format("%Y-%m-%d %H:%M:%S").to_string())
423                    }
424                    _ => unreachable!("handled above"),
425                }
426            }
427        })
428    }
429
430    async fn decode_column<R: AsyncRead + Unpin>(
431        &self,
432        s: &mut Source<'_, R>,
433        rows: usize,
434    ) -> Result<Vec<Value>> {
435        let mut out = Vec::with_capacity(rows.min(1024));
436        match self {
437            ColType::Nullable(inner) => {
438                let nulls = s.bytes(rows).await?;
439                for null in nulls {
440                    let v = Box::pin(inner.decode_value(s)).await?;
441                    out.push(if null != 0 { Value::Null } else { v });
442                }
443            }
444            other => {
445                for _ in 0..rows {
446                    out.push(Box::pin(other.decode_value(s)).await?);
447                }
448            }
449        }
450        Ok(out)
451    }
452}
453
454/// A block: named, typed columns and rows (row-major, as the handler sees them).
455#[derive(Debug, Clone, PartialEq, Default)]
456pub struct Block {
457    pub columns: Vec<(String, String)>,
458    pub rows: Vec<Vec<Value>>,
459}
460
461impl Block {
462    pub fn is_empty(&self) -> bool {
463        self.columns.is_empty() && self.rows.is_empty()
464    }
465
466    /// Check a block the handler supplied: types NetGet can encode, rows of the right width.
467    pub fn from_json(columns: &Value, rows: &Value) -> Result<Self> {
468        let cols = columns.as_array().context("columns must be an array")?;
469        ensure!(cols.len() <= MAX_COLUMNS, "at most {MAX_COLUMNS} columns");
470        let mut columns = Vec::with_capacity(cols.len());
471        for c in cols {
472            let name = c["name"].as_str().context("each column needs a name")?;
473            let t = c["type"].as_str().context("each column needs a type")?;
474            ColType::parse(t)?;
475            columns.push((name.to_string(), t.to_string()));
476        }
477        let rows: Vec<Vec<Value>> = match rows {
478            Value::Null => Vec::new(),
479            Value::Array(rs) => {
480                ensure!(rs.len() <= MAX_ROWS, "at most {MAX_ROWS} rows");
481                rs.iter()
482                    .map(|r| {
483                        let r = r.as_array().context("each row must be an array")?;
484                        ensure!(
485                            r.len() == columns.len(),
486                            "a row has {} values for {} columns",
487                            r.len(),
488                            columns.len()
489                        );
490                        Ok(r.clone())
491                    })
492                    .collect::<Result<_>>()?
493            }
494            _ => bail!("rows must be an array of arrays"),
495        };
496        let block = Block { columns, rows };
497        block.encode_body()?;
498        Ok(block)
499    }
500
501    /// BlockInfo, counts, then each column's name, type and data.
502    pub fn encode_body(&self) -> Result<Vec<u8>> {
503        let mut out = Vec::new();
504        put_varuint(&mut out, 1);
505        out.push(0); // is_overflows
506        put_varuint(&mut out, 2);
507        out.extend((-1i32).to_le_bytes()); // bucket_num
508        put_varuint(&mut out, 0);
509        put_varuint(&mut out, self.columns.len() as u64);
510        put_varuint(&mut out, self.rows.len() as u64);
511        for (i, (name, t)) in self.columns.iter().enumerate() {
512            put_string(&mut out, name.as_bytes());
513            put_string(&mut out, t.as_bytes());
514            let values: Vec<&Value> = self.rows.iter().map(|r| &r[i]).collect();
515            out.extend(
516                ColType::parse(t)?
517                    .encode_column(&values)
518                    .with_context(|| format!("column {name}"))?,
519            );
520        }
521        Ok(out)
522    }
523
524    pub async fn decode<R: AsyncRead + Unpin>(s: &mut Source<'_, R>) -> Result<Self> {
525        loop {
526            match s.varuint().await? {
527                0 => break,
528                1 => {
529                    s.u8().await?;
530                }
531                2 => {
532                    s.bytes(4).await?;
533                }
534                other => bail!("unknown BlockInfo field {other}"),
535            }
536        }
537        let ncols = s.varuint().await? as usize;
538        let nrows = s.varuint().await? as usize;
539        ensure!(ncols <= MAX_COLUMNS, "block of {ncols} columns");
540        ensure!(nrows <= MAX_ROWS, "block of {nrows} rows");
541        let mut columns = Vec::with_capacity(ncols);
542        let mut data = Vec::with_capacity(ncols);
543        for _ in 0..ncols {
544            let name = s.string().await?;
545            let t = s.string().await?;
546            let col = ColType::parse(&t)?;
547            data.push(col.decode_column(s, nrows).await?);
548            columns.push((name, t));
549        }
550        let rows = (0..nrows)
551            .map(|r| data.iter().map(|c| c[r].clone()).collect())
552            .collect();
553        Ok(Block { columns, rows })
554    }
555
556    pub fn columns_json(&self) -> Value {
557        Value::Array(
558            self.columns
559                .iter()
560                .map(|(n, t)| json!({"name": n, "type": t}))
561                .collect(),
562        )
563    }
564}
565
566/// One compressed frame (LZ4) around `data`, with its CityHash128 checksum.
567pub fn compress(data: &[u8]) -> Vec<u8> {
568    let body = lz4_flex::block::compress(data);
569    let mut frame = Vec::with_capacity(body.len() + 25);
570    frame.push(0x82);
571    frame.extend(((body.len() + 9) as u32).to_le_bytes());
572    frame.extend((data.len() as u32).to_le_bytes());
573    frame.extend_from_slice(&body);
574    let sum = naive_cityhash::cityhash128(&frame);
575    let mut out = Vec::with_capacity(frame.len() + 16);
576    out.extend(sum.lo.to_le_bytes());
577    out.extend(sum.hi.to_le_bytes());
578    out.extend(frame);
579    out
580}
581
582/// A Data packet (server 1 / client 2): table name outside, the block inside, compressed when
583/// the query asked for it.
584pub fn data_packet(kind: u64, block: &Block, compressed: bool) -> Result<Vec<u8>> {
585    let mut out = Vec::new();
586    put_varuint(&mut out, kind);
587    put_string(&mut out, b"");
588    let body = block.encode_body()?;
589    if compressed {
590        out.extend(compress(&body));
591    } else {
592        out.extend(body);
593    }
594    Ok(out)
595}
596
597/// An Exception packet.
598pub fn exception(code: i32, message: &str) -> Vec<u8> {
599    let mut out = Vec::new();
600    put_varuint(&mut out, server_packet::EXCEPTION);
601    out.extend(code.to_le_bytes());
602    put_string(&mut out, b"DB::Exception");
603    put_string(&mut out, format!("DB::Exception: {message}").as_bytes());
604    put_string(&mut out, b"");
605    out.push(0);
606    out
607}
608
609pub fn progress(rows: u64, bytes: u64, written_rows: u64) -> Vec<u8> {
610    let mut out = Vec::new();
611    put_varuint(&mut out, server_packet::PROGRESS);
612    put_varuint(&mut out, rows);
613    put_varuint(&mut out, bytes);
614    put_varuint(&mut out, rows);
615    put_varuint(&mut out, written_rows);
616    put_varuint(&mut out, 0);
617    out
618}
619
620/// The client's Query packet fields at revision 54429.
621#[derive(Debug, Clone, Default)]
622pub struct Query {
623    pub query_id: String,
624    pub client_name: String,
625    pub settings: Vec<(String, String)>,
626    pub stage: u64,
627    pub compression: bool,
628    pub query: String,
629}
630
631/// Read a Query packet (after its type) at `revision` (at most 54429).
632pub async fn read_query<R: AsyncRead + Unpin>(
633    s: &mut Source<'_, R>,
634    revision: u64,
635) -> Result<Query> {
636    let mut q = Query {
637        query_id: s.string().await?,
638        ..Default::default()
639    };
640    if revision >= 54032 {
641        let kind = s.u8().await?;
642        if kind != 0 {
643            for _ in 0..3 {
644                s.string().await?; // initial user, query id, address
645            }
646            let interface = s.u8().await?;
647            ensure!(interface == 1, "client interface {interface} is not TCP");
648            s.string().await?; // os user
649            s.string().await?; // hostname
650            q.client_name = s.string().await?;
651            s.varuint().await?;
652            s.varuint().await?;
653            s.varuint().await?;
654            if revision >= 54060 {
655                s.string().await?; // quota key
656            }
657            if revision >= 54401 {
658                s.varuint().await?; // version patch
659            }
660        }
661    }
662    for _ in 0..=MAX_SETTINGS {
663        let name = s.string().await?;
664        if name.is_empty() {
665            q.stage = s.varuint().await?;
666            q.compression = s.varuint().await? != 0;
667            q.query = s.string().await?;
668            return Ok(q);
669        }
670        s.varuint().await?; // flags
671        let value = s.string().await?;
672        q.settings.push((name, value));
673    }
674    bail!("more than {MAX_SETTINGS} settings")
675}
676
677/// The client's Query packet, as NetGet's client sends it at revision 54429.
678pub fn query_packet(query: &str, compression: bool) -> Vec<u8> {
679    let mut out = Vec::new();
680    put_varuint(&mut out, client_packet::QUERY);
681    put_string(&mut out, b"");
682    out.push(1); // initial query
683    put_string(&mut out, b"");
684    put_string(&mut out, b"");
685    put_string(&mut out, b"0.0.0.0:0");
686    out.push(1); // TCP
687    put_string(&mut out, b"netget");
688    put_string(&mut out, b"netget");
689    put_string(&mut out, b"NetGet");
690    put_varuint(&mut out, VERSION_MAJOR);
691    put_varuint(&mut out, VERSION_MINOR);
692    put_varuint(&mut out, REVISION);
693    put_string(&mut out, b"");
694    put_varuint(&mut out, VERSION_PATCH);
695    put_string(&mut out, b""); // end of settings
696    put_varuint(&mut out, 2); // stage: complete
697    put_varuint(&mut out, u64::from(compression));
698    put_string(&mut out, query.as_bytes());
699    out
700}
701
702/// A decoded Exception packet (after its type): code and message, nested ones joined.
703pub async fn read_exception<R: AsyncRead + Unpin>(s: &mut Source<'_, R>) -> Result<(i32, String)> {
704    let mut first: Option<(i32, String)> = None;
705    for _ in 0..16 {
706        let code = i32::from_le_bytes(
707            s.bytes(4)
708                .await?
709                .try_into()
710                .map_err(|_| anyhow::anyhow!("code"))?,
711        );
712        s.string().await?; // name
713        let message = s.string().await?;
714        s.string().await?; // stack trace
715        let nested = s.u8().await? != 0;
716        if first.is_none() {
717            first = Some((code, message));
718        }
719        if !nested {
720            return first.context("exception");
721        }
722    }
723    bail!("exception nested too deeply")
724}
netget/server/clickhouse/wire.rs

netget/server/clickhouse/
wire.rs
