clickhouse_c/
client.rs

1//! Blocking ClickHouse native protocol client.
2//!
3//! Caller supplies a connected [`Io`] transport. Crate provides
4//! [`PosixIo`](crate::PosixIo) for Unix file descriptors and
5//! `tls::TlsIo` for rustls connections.
6
7use core::ffi::c_char;
8use core::pin::Pin;
9use core::ptr::NonNull;
10use core::slice;
11use core::time::Duration;
12use std::ffi::CString;
13
14use crate::alloc::Allocator;
15use crate::block::Block;
16use crate::builder::BlockBuilder;
17use crate::codec::{Codec, Compression};
18use crate::error::{Error, ErrorKind, Result, check};
19use crate::io::Io;
20use crate::query::{QueryOpts, RawQueryOpts, cstring};
21use crate::sys;
22
23/// Client settings sent during Hello handshake.
24///
25/// String values are copied and null-terminated during connection. Interior
26/// null bytes return [`ErrorKind::Usage`](crate::ErrorKind::Usage).
27#[derive(Clone, Debug, Default)]
28pub struct ClientOpts {
29    client_name: Option<String>,
30    database: Option<String>,
31    user: Option<String>,
32    password: Option<String>,
33    /// Client version reported in `system.query_log`. Default is 0.0.0.
34    pub client_version_major: u64,
35    pub client_version_minor: u64,
36    pub client_version_patch: u64,
37    pub compression: Compression,
38    /// Read buffer size in bytes. Zero selects clickhouse-c 8 KiB default.
39    pub read_buffer_bytes: usize,
40}
41
42impl ClientOpts {
43    /// Creates settings for `default` user and database, empty password, and
44    /// no compression.
45    pub fn new() -> Self {
46        Self::default()
47    }
48
49    /// Sets client name. Default is `"clickhouse-c"`.
50    pub fn client_name(mut self, s: &str) -> Self {
51        self.client_name = Some(s.to_owned());
52        self
53    }
54    /// Sets default database for unqualified table names.
55    pub fn database(mut self, s: &str) -> Self {
56        self.database = Some(s.to_owned());
57        self
58    }
59    /// Sets account name used for authentication.
60    pub fn user(mut self, s: &str) -> Self {
61        self.user = Some(s.to_owned());
62        self
63    }
64    /// Sets authentication password.
65    ///
66    /// Native protocol sends password as clear text inside Hello message. Use
67    /// TLS when transport is not trusted.
68    pub fn password(mut self, s: &str) -> Self {
69        self.password = Some(s.to_owned());
70        self
71    }
72
73    /// Sets reported client version.
74    pub fn client_version(mut self, major: u64, minor: u64, patch: u64) -> Self {
75        self.client_version_major = major;
76        self.client_version_minor = minor;
77        self.client_version_patch = patch;
78        self
79    }
80
81    /// Sets compression algorithm.
82    ///
83    /// Compressed connections require a matching [`Codec`] in [`Client::init`].
84    pub fn compression(mut self, compression: Compression) -> Self {
85        self.compression = compression;
86        self
87    }
88
89    pub(crate) fn to_raw(&self, codec: Option<*const sys::chc_codec>) -> Result<RawClientOpts> {
90        let mut owned = Vec::with_capacity(4);
91        let mut field = |label, value: &Option<String>| -> Result<*const c_char> {
92            let Some(value) = value else {
93                return Ok(core::ptr::null());
94            };
95            owned.push(cstring(label, value)?);
96            Ok(owned.last().expect("just pushed").as_ptr())
97        };
98        let client_name = field("client name", &self.client_name)?;
99        let database = field("database", &self.database)?;
100        let user = field("user", &self.user)?;
101        let password = field("password", &self.password)?;
102
103        Ok(RawClientOpts {
104            _owned: owned,
105            raw: sys::chc_client_opts {
106                client_name,
107                client_version_major: self.client_version_major,
108                client_version_minor: self.client_version_minor,
109                client_version_patch: self.client_version_patch,
110                database,
111                user,
112                password,
113                compression: self.compression as i32,
114                codec: codec.unwrap_or(core::ptr::null()),
115                read_buffer_bytes: self.read_buffer_bytes,
116            },
117        })
118    }
119
120    pub(crate) fn validate_codec(&self, codec: Option<Pin<&Codec>>) -> Result<()> {
121        if self.compression == Compression::None {
122            return Ok(());
123        }
124        let codec = codec.ok_or_else(|| {
125            Error::new(
126                ErrorKind::Usage,
127                format!("{:?} compression requires a codec", self.compression),
128            )
129        })?;
130        if codec.supports(self.compression) {
131            Ok(())
132        } else {
133            Err(Error::new(
134                ErrorKind::Usage,
135                format!("codec does not support {:?} compression", self.compression),
136            ))
137        }
138    }
139}
140
141/// Owns raw client options and null-terminated strings referenced by them.
142pub(crate) struct RawClientOpts {
143    _owned: Vec<CString>,
144    raw: sys::chc_client_opts,
145}
146
147impl RawClientOpts {
148    #[inline]
149    pub(crate) fn as_ptr(&self) -> *const sys::chc_client_opts {
150        &self.raw
151    }
152}
153
154/// Server information received during Hello handshake.
155#[derive(Debug, Clone)]
156pub struct ServerInfo {
157    pub name: String,
158    pub timezone: String,
159    pub display_name: String,
160    pub version_major: u64,
161    pub version_minor: u64,
162    pub version_patch: u64,
163    pub revision: u64,
164}
165
166impl ServerInfo {
167    pub(crate) fn from_raw(raw: &sys::chc_server_info) -> Self {
168        Self {
169            name: cstr_array_to_string(&raw.name),
170            timezone: cstr_array_to_string(&raw.timezone),
171            display_name: cstr_array_to_string(&raw.display_name),
172            version_major: raw.version_major,
173            version_minor: raw.version_minor,
174            version_patch: raw.version_patch,
175            revision: raw.revision,
176        }
177    }
178}
179
180fn cstr_array_to_string(buf: &[c_char]) -> String {
181    let end = buf.iter().position(|&b| b == 0).unwrap_or(buf.len());
182    let bytes: &[u8] = unsafe { slice::from_raw_parts(buf.as_ptr().cast::<u8>(), end) };
183    String::from_utf8_lossy(bytes).into_owned()
184}
185
186/// Active blocking ClickHouse connection.
187///
188/// Client owns C connection, I/O transport, and optional compression codec.
189pub struct Client<'fd> {
190    raw: NonNull<sys::chc_client>,
191    // C connection retains allocator address until close
192    alloc: Box<Allocator>,
193    _codec: Option<Pin<Box<Codec>>>,
194    // C connection retains callback pointer into pinned transport
195    io: Pin<Box<dyn Io + Send + 'fd>>,
196}
197
198impl<'fd> Client<'fd> {
199    /// Creates client and completes Hello handshake using supplied transport.
200    ///
201    /// Method takes ownership of `io` and `codec`. `io` can be
202    /// [`PosixIo`](crate::PosixIo), `tls::TlsIo`, or custom [`Io`]
203    /// implementation.
204    ///
205    /// `codec` can be `None` only when compression is disabled.
206    ///
207    /// Server rejection returns [`ErrorKind::Server`] carrying exception code,
208    /// class, and untruncated message.
209    ///
210    /// Lifetime `'fd` prevents client from outliving a borrowed file
211    /// descriptor:
212    ///
213    /// ```compile_fail
214    /// use clickhouse_c::{Allocator, Client, ClientOpts, PosixIo};
215    /// use std::net::TcpStream;
216    /// use std::os::fd::AsFd;
217    ///
218    /// fn build() -> clickhouse_c::Result<Client<'static>> {
219    ///     let sock = TcpStream::connect("localhost:9000")?;
220    ///     let io = PosixIo::new(sock.as_fd());
221    ///     // Borrowed socket cannot produce Client<'static>.
222    ///     Client::init(&ClientOpts::new(), Allocator::stdlib(), io, None)
223    /// }
224    /// ```
225    pub fn init<I: Io + Send + 'fd>(
226        opts: &ClientOpts,
227        alloc: Allocator,
228        mut io: Pin<Box<I>>,
229        codec: Option<Pin<Box<Codec>>>,
230    ) -> Result<Self> {
231        opts.validate_codec(codec.as_ref().map(|codec| codec.as_ref()))?;
232        let codec_ptr = codec.as_ref().map(|c| c.as_ref().as_ptr());
233        let raw_opts = opts.to_raw(codec_ptr)?;
234        let alloc = Box::new(alloc);
235        let mut out: *mut sys::chc_client = core::ptr::null_mut();
236        let mut exc: *mut sys::chc_exception = core::ptr::null_mut();
237        let mut err = sys::chc_err::zeroed();
238        let rc = unsafe {
239            sys::chc_client_init(
240                &mut out,
241                raw_opts.as_ptr(),
242                alloc.as_ptr(),
243                io.as_mut().io_ptr(),
244                &mut exc,
245                &mut err,
246            )
247        };
248        if let Some(e) = take_handshake_exception(exc, *alloc) {
249            return Err(e);
250        }
251        check(rc, &err)?;
252        Ok(Self {
253            raw: NonNull::new(out).expect("chc_client_init returned OK with NULL"),
254            alloc,
255            _codec: codec,
256            io,
257        })
258    }
259
260    /// Returns server information received during handshake.
261    pub fn server_info(&self) -> Option<ServerInfo> {
262        let p = unsafe { sys::chc_client_server_info(self.raw.as_ptr().cast_const()) };
263        (!p.is_null()).then(|| ServerInfo::from_raw(unsafe { &*p }))
264    }
265
266    /// Sets transport read timeout.
267    ///
268    /// [`PosixIo`](crate::PosixIo) uses an absolute deadline. Set timeout
269    /// again before each operation that requires a fresh deadline.
270    pub fn set_read_timeout(&mut self, timeout: Option<Duration>) -> Result<()> {
271        self.io.as_mut().set_read_timeout(timeout)
272    }
273
274    /// Sends a query without settings or parameters.
275    ///
276    /// Server profile supplies all query settings. Use
277    /// [`QuerySetting::TEXT_TYPE_NAMES`](crate::QuerySetting::TEXT_TYPE_NAMES)
278    /// with [`send_query_with`](Self::send_query_with) when profile may enable
279    /// binary type names.
280    pub fn send_query(&mut self, sql: &str, query_id: Option<&str>) -> Result<()> {
281        let (qid, qid_len) = query_id
282            .map(|q| (q.as_ptr().cast::<c_char>(), q.len()))
283            .unwrap_or((core::ptr::null(), 0));
284        let mut err = sys::chc_err::zeroed();
285        let rc = unsafe {
286            sys::chc_client_send_query(
287                self.raw.as_ptr(),
288                sql.as_ptr().cast::<c_char>(),
289                sql.len(),
290                qid,
291                qid_len,
292                &mut err,
293            )
294        };
295        check(rc, &err)
296    }
297
298    /// Sends a query with settings and `{name:Type}` parameters.
299    ///
300    /// ```no_run
301    /// use clickhouse_c::{QueryOpts, QueryParam, QuerySetting};
302    /// # fn run(client: &mut clickhouse_c::Client<'_>) -> clickhouse_c::Result<()> {
303    /// let settings = [
304    ///     QuerySetting::TEXT_TYPE_NAMES,
305    ///     QuerySetting::new("max_block_size", "8192"),
306    /// ];
307    /// let params = [QueryParam::new("cutoff", "'100'")];
308    /// client.send_query_with(
309    ///     "SELECT number FROM numbers(1000) WHERE number > {cutoff:UInt64}",
310    ///     &QueryOpts::new().settings(&settings).params(&params),
311    /// )?;
312    /// # Ok(())
313    /// # }
314    /// ```
315    pub fn send_query_with(&mut self, sql: &str, opts: &QueryOpts<'_>) -> Result<()> {
316        let raw_opts = RawQueryOpts::new(opts)?;
317        let mut err = sys::chc_err::zeroed();
318        let rc = unsafe {
319            sys::chc_client_send_query_ex(
320                self.raw.as_ptr(),
321                sql.as_ptr().cast::<c_char>(),
322                sql.len(),
323                raw_opts.as_ptr(),
324                &mut err,
325            )
326        };
327        check(rc, &err)
328    }
329
330    /// Sends a Data block.
331    ///
332    /// `None` sends empty block that ends INSERT input.
333    pub fn send_data(&mut self, builder: Option<&BlockBuilder<'_>>) -> Result<()> {
334        let bb_ptr = builder.map(|b| b.as_ptr()).unwrap_or(core::ptr::null());
335        let mut err = sys::chc_err::zeroed();
336        let rc = unsafe { sys::chc_client_send_data(self.raw.as_ptr(), bb_ptr, &mut err) };
337        check(rc, &err)
338    }
339
340    /// Sends protocol Cancel packet for active query.
341    ///
342    /// Continue receiving events until [`Event::EndOfStream`] because packets
343    /// already sent by server can still arrive. Use [`CancelToken`](crate::CancelToken)
344    /// to cancel local reads without sending a packet.
345    pub fn send_cancel(&mut self) -> Result<()> {
346        let mut err = sys::chc_err::zeroed();
347        let rc = unsafe { sys::chc_client_send_cancel(self.raw.as_ptr(), &mut err) };
348        check(rc, &err)
349    }
350
351    /// Sends Ping packet. Server responds with [`Event::Pong`].
352    pub fn send_ping(&mut self) -> Result<()> {
353        let mut err = sys::chc_err::zeroed();
354        let rc = unsafe { sys::chc_client_send_ping(self.raw.as_ptr(), &mut err) };
355        check(rc, &err)
356    }
357
358    /// Reads next server event and blocks until complete packet arrives.
359    ///
360    /// Returned event owns block or exception payload.
361    pub fn recv_event(&mut self) -> Result<Event> {
362        let mut raw = sys::chc_packet::zeroed();
363        let mut err = sys::chc_err::zeroed();
364        let rc = unsafe { sys::chc_client_recv_packet(self.raw.as_ptr(), &mut raw, &mut err) };
365        if let Err(e) = check(rc, &err) {
366            unsafe { sys::chc_packet_clear(self.raw.as_ptr(), &mut raw) };
367            return Err(e);
368        }
369        let event = Event::from_raw(&mut raw, *self.alloc);
370        unsafe { sys::chc_packet_clear(self.raw.as_ptr(), &mut raw) };
371        event
372    }
373}
374
375impl<'fd> Drop for Client<'fd> {
376    fn drop(&mut self) {
377        unsafe { sys::chc_client_close(self.raw.as_ptr()) };
378    }
379}
380
381unsafe impl<'fd> Send for Client<'fd> {}
382
383/// Exception returned by ClickHouse server.
384pub struct Exception {
385    raw: NonNull<sys::chc_exception>,
386    alloc: Allocator,
387}
388
389impl Exception {
390    /// SAFETY: caller must own `raw`, and `alloc` must match its allocator
391    pub(crate) unsafe fn from_raw(raw: NonNull<sys::chc_exception>, alloc: Allocator) -> Self {
392        Self { raw, alloc }
393    }
394
395    /// Returns ClickHouse error code from `system.errors`.
396    pub fn code(&self) -> i32 {
397        unsafe { (*self.raw.as_ptr()).code }
398    }
399
400    /// Returns exception class name without UTF-8 validation.
401    pub fn name(&self) -> &[u8] {
402        let r = unsafe { self.raw.as_ref() };
403        cstr_bytes(r.name, r.name_len)
404    }
405
406    /// Returns exception message without UTF-8 validation.
407    pub fn display_text(&self) -> &[u8] {
408        let r = unsafe { self.raw.as_ref() };
409        cstr_bytes(r.display_text, r.display_text_len)
410    }
411
412    /// Returns server stack trace without UTF-8 validation.
413    ///
414    /// Value is empty unless query requested a stack trace.
415    pub fn stack_trace(&self) -> &[u8] {
416        let r = unsafe { self.raw.as_ref() };
417        cstr_bytes(r.stack_trace, r.stack_trace_len)
418    }
419}
420
421impl Drop for Exception {
422    fn drop(&mut self) {
423        unsafe { sys::chc_exception_free(self.raw.as_ptr(), self.alloc.as_ptr()) };
424    }
425}
426
427impl core::fmt::Debug for Exception {
428    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
429        f.debug_struct("Exception")
430            .field("code", &self.code())
431            .field("name", &String::from_utf8_lossy(self.name()))
432            .field(
433                "display_text",
434                &String::from_utf8_lossy(self.display_text()),
435            )
436            .field("stack_trace", &String::from_utf8_lossy(self.stack_trace()))
437            .finish()
438    }
439}
440
441impl core::fmt::Display for Exception {
442    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
443        write!(
444            f,
445            "{} (code {}): {}",
446            String::from_utf8_lossy(self.name()),
447            self.code(),
448            String::from_utf8_lossy(self.display_text()),
449        )
450    }
451}
452
453impl std::error::Error for Exception {}
454
455unsafe impl Send for Exception {}
456
457impl From<Exception> for Error {
458    fn from(exc: Exception) -> Self {
459        Self {
460            kind: ErrorKind::Server,
461            server_code: exc.code(),
462            message: String::from_utf8_lossy(exc.display_text()).into_owned(),
463            server_name: String::from_utf8_lossy(exc.name()).into_owned(),
464        }
465    }
466}
467
468/// Converts handshake rejection into an error.
469///
470/// clickhouse-c leaves `err` empty for handshake rejection and transfers
471/// exception ownership instead.
472pub(crate) fn take_handshake_exception(
473    exc: *mut sys::chc_exception,
474    alloc: Allocator,
475) -> Option<Error> {
476    // SAFETY: handshake transferred ownership of exception allocated by alloc
477    NonNull::new(exc).map(|p| unsafe { Exception::from_raw(p, alloc) }.into())
478}
479
480fn cstr_bytes<'a>(ptr: *mut c_char, len: usize) -> &'a [u8] {
481    if ptr.is_null() || len == 0 {
482        return &[];
483    }
484    debug_assert!(
485        len <= isize::MAX as usize,
486        "clickhouse-c published exception field len = {len}",
487    );
488    unsafe { slice::from_raw_parts(ptr.cast::<u8>(), len) }
489}
490
491/// Server packet kind.
492///
493/// Hello packets are handled during [`Client::init`] and do not have a variant.
494#[derive(Debug, Clone, Copy, PartialEq, Eq)]
495#[repr(i32)]
496pub enum PacketKind {
497    Data = sys::CHC_PKT_DATA,
498    Exception = sys::CHC_PKT_EXCEPTION,
499    Progress = sys::CHC_PKT_PROGRESS,
500    Pong = sys::CHC_PKT_PONG,
501    EndOfStream = sys::CHC_PKT_END_OF_STREAM,
502    ProfileInfo = sys::CHC_PKT_PROFILE_INFO,
503    Totals = sys::CHC_PKT_TOTALS,
504    Extremes = sys::CHC_PKT_EXTREMES,
505    Log = sys::CHC_PKT_LOG,
506    TableColumns = sys::CHC_PKT_TABLE_COLUMNS,
507    ProfileEvents = sys::CHC_PKT_PROFILE_EVENTS,
508    TimezoneUpdate = sys::CHC_PKT_TIMEZONE_UPDATE,
509}
510
511impl PacketKind {
512    pub(crate) fn from_raw(k: sys::chc_packet_kind) -> Option<Self> {
513        Some(match k {
514            sys::CHC_PKT_DATA => Self::Data,
515            sys::CHC_PKT_EXCEPTION => Self::Exception,
516            sys::CHC_PKT_PROGRESS => Self::Progress,
517            sys::CHC_PKT_PONG => Self::Pong,
518            sys::CHC_PKT_END_OF_STREAM => Self::EndOfStream,
519            sys::CHC_PKT_PROFILE_INFO => Self::ProfileInfo,
520            sys::CHC_PKT_TOTALS => Self::Totals,
521            sys::CHC_PKT_EXTREMES => Self::Extremes,
522            sys::CHC_PKT_LOG => Self::Log,
523            sys::CHC_PKT_TABLE_COLUMNS => Self::TableColumns,
524            sys::CHC_PKT_PROFILE_EVENTS => Self::ProfileEvents,
525            sys::CHC_PKT_TIMEZONE_UPDATE => Self::TimezoneUpdate,
526            _ => return None,
527        })
528    }
529}
530
531/// Event received from server.
532///
533/// Event owns any block or exception payload.
534pub enum Event {
535    /// Result block or expected INSERT structure.
536    Data(Block),
537    /// Row produced by `WITH TOTALS`.
538    Totals(Block),
539    /// Minimum and maximum rows produced by `WITH EXTREMES`.
540    Extremes(Block),
541    /// Server log rows requested by `send_logs_level`.
542    Log(Block),
543    /// Per-query profile event counters.
544    ProfileEvents(Block),
545    /// Server exception that ends current query.
546    Exception(Exception),
547    /// Incremental read and write counters.
548    Progress(Progress),
549    /// Row and byte totals sent near query completion.
550    ProfileInfo(ProfileInfo),
551    /// Response to [`Client::send_ping`].
552    Pong,
553    /// Query completion marker.
554    EndOfStream,
555    /// INSERT target metadata. Payload is not decoded. Following Data block
556    /// contains same structure.
557    TableColumns,
558    /// Server timezone changed. Read updated value from client server info.
559    TimezoneUpdate,
560}
561
562impl Event {
563    /// Converts received C packet and takes ownership of its payload.
564    pub(crate) fn from_raw(raw: &mut sys::chc_packet, alloc: Allocator) -> Result<Self> {
565        let Some(kind) = PacketKind::from_raw(raw.kind) else {
566            return Err(Error::new(
567                ErrorKind::Protocol,
568                format!("unknown server packet {}", raw.kind),
569            ));
570        };
571        Ok(match kind {
572            PacketKind::Data => Self::Data(take_block(raw, alloc)?),
573            PacketKind::Totals => Self::Totals(take_block(raw, alloc)?),
574            PacketKind::Extremes => Self::Extremes(take_block(raw, alloc)?),
575            PacketKind::Log => Self::Log(take_block(raw, alloc)?),
576            PacketKind::ProfileEvents => Self::ProfileEvents(take_block(raw, alloc)?),
577            PacketKind::Exception => Self::Exception(take_exception(raw, alloc)?),
578            PacketKind::Progress => {
579                // SAFETY: packet kind selects progress union member
580                Self::Progress(Progress::from_raw(unsafe { &raw.payload.progress }))
581            }
582            PacketKind::ProfileInfo => {
583                // SAFETY: packet kind selects profile union member
584                Self::ProfileInfo(ProfileInfo::from_raw(unsafe { &raw.payload.profile }))
585            }
586            PacketKind::Pong => Self::Pong,
587            PacketKind::EndOfStream => Self::EndOfStream,
588            PacketKind::TableColumns => Self::TableColumns,
589            PacketKind::TimezoneUpdate => Self::TimezoneUpdate,
590        })
591    }
592}
593
594fn take_block(raw: &mut sys::chc_packet, alloc: Allocator) -> Result<Block> {
595    // SAFETY: caller matched block packet kind
596    let p = unsafe { raw.payload.block };
597    raw.payload.block = core::ptr::null_mut();
598    // SAFETY: allocator belongs to client that received block
599    unsafe { Block::from_raw(p, alloc) }
600        .ok_or_else(|| Error::new(ErrorKind::Protocol, "block packet missing block"))
601}
602
603fn take_exception(raw: &mut sys::chc_packet, alloc: Allocator) -> Result<Exception> {
604    // SAFETY: caller matched exception packet kind
605    let p = NonNull::new(unsafe { raw.payload.exception })
606        .ok_or_else(|| Error::new(ErrorKind::Protocol, "exception packet missing exception"))?;
607    raw.payload.exception = core::ptr::null_mut();
608    // SAFETY: allocator belongs to client that received exception
609    Ok(unsafe { Exception::from_raw(p, alloc) })
610}
611
612/// Incremental query counters. Each packet contains a delta.
613#[derive(Debug, Clone, Copy, PartialEq, Eq)]
614pub struct Progress {
615    pub rows: u64,
616    pub bytes: u64,
617    pub total_rows: u64,
618    pub total_bytes: u64,
619    pub written_rows: u64,
620    pub written_bytes: u64,
621    pub elapsed_ns: u64,
622}
623
624impl Progress {
625    fn from_raw(raw: &sys::chc_packet_progress) -> Self {
626        Self {
627            rows: raw.rows,
628            bytes: raw.bytes,
629            total_rows: raw.total_rows,
630            total_bytes: raw.total_bytes,
631            written_rows: raw.written_rows,
632            written_bytes: raw.written_bytes,
633            elapsed_ns: raw.elapsed_ns,
634        }
635    }
636}
637
638/// Query totals reported near completion.
639#[derive(Debug, Clone, Copy, PartialEq, Eq)]
640pub struct ProfileInfo {
641    pub rows: u64,
642    pub blocks: u64,
643    pub bytes: u64,
644    pub rows_before_limit: u64,
645    pub applied_limit: bool,
646    pub calculated_rows_before_limit: bool,
647}
648
649impl ProfileInfo {
650    fn from_raw(raw: &sys::chc_packet_profile) -> Self {
651        Self {
652            rows: raw.rows,
653            blocks: raw.blocks,
654            bytes: raw.bytes,
655            rows_before_limit: raw.rows_before_limit,
656            applied_limit: raw.applied_limit != 0,
657            calculated_rows_before_limit: raw.calculated_rows_before_limit != 0,
658        }
659    }
660}
661
662#[cfg(test)]
663mod tests {
664    use std::ffi::CStr;
665
666    use core::ffi::c_char;
667
668    use super::{ClientOpts, Event, PacketKind, cstr_bytes};
669    use crate::sys;
670    use crate::{Allocator, Compression, ErrorKind};
671
672    /// Compression always requires a codec, including builds without codecs
673    #[test]
674    fn compression_without_a_codec_is_a_usage_error() {
675        let err = ClientOpts::new()
676            .compression(Compression::Lz4)
677            .validate_codec(None)
678            .expect_err("missing codec");
679        assert_eq!(err.kind, ErrorKind::Usage);
680    }
681
682    #[cfg(feature = "lz4")]
683    #[test]
684    fn compression_requires_matching_codec() {
685        use crate::Codec;
686
687        let codec = Codec::lz4();
688        let mismatch = ClientOpts::new()
689            .compression(Compression::Zstd)
690            .validate_codec(Some(codec.as_ref()))
691            .expect_err("mismatched codec");
692        assert_eq!(mismatch.kind, ErrorKind::Usage);
693    }
694
695    #[cfg(feature = "lz4")]
696    #[test]
697    fn matching_codec_passes_validation() {
698        use crate::Codec;
699
700        let codec = Codec::lz4();
701        ClientOpts::new()
702            .compression(Compression::Lz4)
703            .validate_codec(Some(codec.as_ref()))
704            .expect("lz4 codec for lz4 compression");
705    }
706
707    // Hello fields reach C as null-terminated copies
708    #[test]
709    fn handshake_fields_reach_raw_opts() {
710        let opts = ClientOpts::new()
711            .client_name("probe")
712            .database("db")
713            .user("reader")
714            .password("secret")
715            .client_version(1, 2, 3);
716        let raw = opts.to_raw(None).expect("no interior NUL");
717        let field = |p: *const core::ffi::c_char| unsafe { CStr::from_ptr(p) }.to_owned();
718        let raw = unsafe { &*raw.as_ptr() };
719        assert_eq!(field(raw.client_name).to_bytes(), b"probe");
720        assert_eq!(field(raw.database).to_bytes(), b"db");
721        assert_eq!(field(raw.user).to_bytes(), b"reader");
722        assert_eq!(field(raw.password).to_bytes(), b"secret");
723        assert_eq!(
724            (
725                raw.client_version_major,
726                raw.client_version_minor,
727                raw.client_version_patch,
728            ),
729            (1, 2, 3),
730        );
731        assert!(raw.codec.is_null());
732    }
733
734    #[test]
735    fn default_opts_leave_every_string_null() {
736        let raw = ClientOpts::new().to_raw(None).expect("no strings");
737        let raw = unsafe { &*raw.as_ptr() };
738        assert!(raw.client_name.is_null());
739        assert!(raw.database.is_null());
740        assert!(raw.user.is_null());
741        assert!(raw.password.is_null());
742    }
743
744    // Packet kinds added to C API must not convert to an adjacent variant
745    #[test]
746    fn unknown_packet_kind_is_none() {
747        assert!(PacketKind::from_raw(i32::MAX).is_none());
748        assert!(PacketKind::from_raw(-1).is_none());
749    }
750
751    #[test]
752    fn unknown_packet_is_a_protocol_error() {
753        let mut raw = sys::chc_packet::zeroed();
754        raw.kind = i32::MAX;
755        let err = Event::from_raw(&mut raw, Allocator::stdlib())
756            .err()
757            .expect("unknown packet kind accepted");
758        assert_eq!(err.kind, ErrorKind::Protocol);
759        assert!(err.message.contains("unknown server packet"), "{err}");
760    }
761
762    // Every packet kind, with whether C hands over a payload pointer
763    const KINDS: [(sys::chc_packet_kind, bool); 12] = [
764        (sys::CHC_PKT_DATA, true),
765        (sys::CHC_PKT_TOTALS, true),
766        (sys::CHC_PKT_EXTREMES, true),
767        (sys::CHC_PKT_LOG, true),
768        (sys::CHC_PKT_PROFILE_EVENTS, true),
769        (sys::CHC_PKT_EXCEPTION, true),
770        (sys::CHC_PKT_PROGRESS, false),
771        (sys::CHC_PKT_PROFILE_INFO, false),
772        (sys::CHC_PKT_PONG, false),
773        (sys::CHC_PKT_END_OF_STREAM, false),
774        (sys::CHC_PKT_TABLE_COLUMNS, false),
775        (sys::CHC_PKT_TIMEZONE_UPDATE, false),
776    ];
777
778    // A payload-carrying kind whose payload C left null must not convert, and
779    // every other kind must convert to its own variant
780    #[test]
781    fn each_packet_kind_converts_to_its_variant() {
782        for (kind, carries_payload) in KINDS {
783            let mut raw = sys::chc_packet::zeroed();
784            raw.kind = kind;
785            let event = Event::from_raw(&mut raw, Allocator::stdlib());
786            assert_eq!(
787                matches!(&event, Err(err) if err.kind == ErrorKind::Protocol),
788                carries_payload,
789                "kind {kind}",
790            );
791            assert_eq!(matches!(event, Ok(Event::Pong)), kind == sys::CHC_PKT_PONG);
792            assert_eq!(
793                matches!(event, Ok(Event::EndOfStream)),
794                kind == sys::CHC_PKT_END_OF_STREAM,
795            );
796            assert_eq!(
797                matches!(event, Ok(Event::TableColumns)),
798                kind == sys::CHC_PKT_TABLE_COLUMNS,
799            );
800            assert_eq!(
801                matches!(event, Ok(Event::TimezoneUpdate)),
802                kind == sys::CHC_PKT_TIMEZONE_UPDATE,
803            );
804            assert_eq!(
805                matches!(event, Ok(Event::Progress(_))),
806                kind == sys::CHC_PKT_PROGRESS,
807            );
808            assert_eq!(
809                matches!(event, Ok(Event::ProfileInfo(_))),
810                kind == sys::CHC_PKT_PROFILE_INFO,
811            );
812        }
813    }
814
815    #[test]
816    fn every_c_packet_kind_has_a_variant() {
817        for (kind, _) in KINDS {
818            assert!(PacketKind::from_raw(kind).is_some(), "kind {kind}");
819        }
820    }
821
822    // Exception fields C left empty must not produce a slice over null
823    #[test]
824    fn empty_exception_fields_read_as_empty_slices() {
825        assert!(cstr_bytes(core::ptr::null_mut(), 7).is_empty());
826        let mut byte = b'x' as c_char;
827        assert!(cstr_bytes(&mut byte, 0).is_empty());
828    }
829}
clickhouse_c/client.rs

clickhouse_c/
client.rs
