questdb/ingress/
buffer.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
25mod ilp;
26mod op_state;
27
28#[cfg(any(feature = "_sender-qwp-udp", feature = "_sender-qwp-ws"))]
29mod qwp;
30
31use crate::ingress::ndarr::ArrayElementSealed;
32use crate::ingress::{ArrayElement, DecimalView, NdArrayView, ProtocolVersion, Timestamp};
33use crate::{Error, error};
34use std::sync::atomic::{AtomicU64, Ordering};
35
36pub(crate) use self::ilp::Buffer as IlpBuffer;
37#[allow(unused_imports)]
38pub(crate) use self::ilp::F64Serializer;
39
40#[cfg(all(feature = "_sender-qwp-ws", feature = "arrow-ingress"))]
41pub(crate) use self::qwp::QWP_DECIMAL_MAX_SCALE;
42#[cfg(any(feature = "_sender-qwp-udp", feature = "_sender-qwp-ws"))]
43pub(crate) use self::qwp::QwpBuffer;
44#[cfg(feature = "_sender-qwp-udp")]
45pub(crate) use self::qwp::QwpSendScratch;
46#[cfg(feature = "_sender-qwp-ws")]
47pub(crate) use self::qwp::{
48    MAX_PERSISTED_SYMBOL_ENTRY_LEN, QwpWsColumnarBuffer, QwpWsEncodeScratch, SymbolGlobalDict,
49    SymbolGlobalDictMark, decode_qwp_varint,
50};
51// Test-only: lets the sender-level suites drive the connection dictionary's
52// cap-rejection path through a real `Sender` (see `TestDictCapGuard`).
53#[cfg(all(test, feature = "_sender-qwp-ws"))]
54pub(crate) use self::qwp::TestDictCapGuard;
55// `QwpWsSymbolHasher`'s only re-export consumer is the `arrow`-gated
56// `column_sender::arrow_batch`, so it is gated identically: a `_sender-qwp-ws`
57// build without `arrow` would otherwise carry an unused import.
58#[cfg(feature = "arrow-ingress")]
59pub(crate) use self::qwp::QwpWsSymbolHasher;
60
61static NEXT_BOOKMARK_ORIGIN: AtomicU64 = AtomicU64::new(1);
62
63/// Opaque rollback handle captured from a [`Buffer`].
64#[derive(Clone, Copy, Debug, Eq, PartialEq)]
65pub struct Bookmark {
66    origin: u64,
67    generation: u64,
68}
69
70impl Bookmark {
71    /// Construct a bookmark from raw parts.
72    ///
73    /// This is exposed for FFI interop. Application code should prefer
74    /// [`Buffer::bookmark`].
75    #[doc(hidden)]
76    pub const fn from_raw(origin: u64, generation: u64) -> Self {
77        Self { origin, generation }
78    }
79
80    /// Return the originating buffer namespace for this bookmark.
81    ///
82    /// This accessor is primarily intended for FFI wrappers.
83    #[doc(hidden)]
84    pub const fn origin(self) -> u64 {
85        self.origin
86    }
87
88    /// Return the generation number associated with this bookmark.
89    ///
90    /// This accessor is primarily intended for FFI wrappers.
91    #[doc(hidden)]
92    pub const fn generation(self) -> u64 {
93        self.generation
94    }
95}
96
97#[derive(Clone, Copy, Debug)]
98pub(super) struct BufferBookmarkMeta {
99    origin: u64,
100}
101
102impl BufferBookmarkMeta {
103    pub(super) fn new() -> Self {
104        Self {
105            origin: NEXT_BOOKMARK_ORIGIN.fetch_add(1, Ordering::Relaxed),
106        }
107    }
108
109    pub(super) const fn origin(self) -> u64 {
110        self.origin
111    }
112}
113
114#[derive(Clone, Copy, Debug)]
115pub(super) struct StoredBookmark<S: Copy> {
116    generation: u64,
117    state: Option<S>,
118}
119
120impl<S: Copy> StoredBookmark<S> {
121    pub(super) const fn new() -> Self {
122        Self {
123            generation: 0,
124            state: None,
125        }
126    }
127
128    pub(super) fn capture(&mut self, origin: u64, state: S) -> Bookmark {
129        self.generation = self.generation.wrapping_add(1);
130        if self.generation == 0 {
131            self.generation = 1;
132        }
133        self.state = Some(state);
134        Bookmark::from_raw(origin, self.generation)
135    }
136
137    pub(super) const fn current(&self) -> Option<S> {
138        self.state
139    }
140
141    pub(super) fn restore(&self, origin: u64, bookmark: Bookmark) -> crate::Result<S> {
142        if bookmark.origin() != origin {
143            return Err(error::fmt!(
144                InvalidApiCall,
145                "Can't rewind to the bookmark: Bookmark does not belong to this buffer."
146            ));
147        }
148        if self.state.is_none() && self.generation == 0 {
149            // This path is mainly defensive for forged or FFI-constructed
150            // bookmarks. Normal Rust callers cannot obtain a matching-origin
151            // bookmark without first capturing one, which advances generation.
152            return Err(error::fmt!(
153                InvalidApiCall,
154                "Can't rewind to the bookmark: No bookmark set."
155            ));
156        }
157        match self.state {
158            Some(state) if self.generation == bookmark.generation() => Ok(state),
159            _ => Err(error::fmt!(
160                InvalidApiCall,
161                "Can't rewind to the bookmark: Bookmark is stale."
162            )),
163        }
164    }
165
166    pub(super) fn clear_if_matches(&mut self, origin: u64, bookmark: Bookmark) -> bool {
167        if bookmark.origin() == 0 {
168            return false;
169        }
170        if bookmark.origin() != origin {
171            // `clear_bookmark()` is intentionally a no-op in release builds so
172            // cleanup paths stay idempotent, but we still want debug builds to
173            // catch obvious cross-buffer misuse early.
174            debug_assert_eq!(
175                bookmark.origin(),
176                origin,
177                "attempted to clear a bookmark from a different buffer"
178            );
179            return false;
180        }
181        if self.state.is_some() && self.generation == bookmark.generation() {
182            self.state = None;
183            return true;
184        }
185        false
186    }
187
188    pub(super) fn clear(&mut self) {
189        self.state = None;
190    }
191}
192
193/// A validated table name.
194///
195/// This type simply wraps a `&str`.
196///
197/// When you pass a `TableName` instead of a plain string to a [`Buffer`] method,
198/// it doesn't have to validate it again. This saves CPU cycles.
199#[derive(Clone, Copy)]
200pub struct TableName<'a> {
201    name: &'a str,
202}
203
204impl<'a> TableName<'a> {
205    /// Construct a validated table name.
206    pub fn new(name: &'a str) -> crate::Result<Self> {
207        if name.is_empty() {
208            return Err(error::fmt!(
209                InvalidName,
210                "Table names must have a non-zero length."
211            ));
212        }
213
214        let mut prev = '\0';
215        for (byte_idx, c) in name.char_indices() {
216            match c {
217                '.' if byte_idx == 0 || byte_idx + c.len_utf8() == name.len() || prev == '.' => {
218                    return Err(error::fmt!(
219                        InvalidName,
220                        concat!("Bad string {:?}: ", "Found invalid dot `.` at position {}."),
221                        name,
222                        byte_idx
223                    ));
224                }
225                '.' => {}
226                '?' | ',' | '\'' | '\"' | '\\' | '/' | ':' | ')' | '(' | '+' | '*' | '%' | '~'
227                | '\r' | '\n' | '\0' | '\u{0001}' | '\u{0002}' | '\u{0003}' | '\u{0004}'
228                | '\u{0005}' | '\u{0006}' | '\u{0007}' | '\u{0008}' | '\u{0009}' | '\u{000b}'
229                | '\u{000c}' | '\u{000e}' | '\u{000f}' | '\u{007f}' => {
230                    return Err(error::fmt!(
231                        InvalidName,
232                        concat!(
233                            "Bad string {:?}: ",
234                            "Table names can't contain ",
235                            "a {:?} character, which was found at ",
236                            "byte position {}."
237                        ),
238                        name,
239                        c,
240                        byte_idx
241                    ));
242                }
243                '\u{feff}' => {
244                    // Reject Unicode char 'ZERO WIDTH NO-BREAK SPACE',
245                    // aka UTF-8 BOM if it appears anywhere in the string.
246                    return Err(error::fmt!(
247                        InvalidName,
248                        concat!(
249                            "Bad string {:?}: ",
250                            "Table names can't contain ",
251                            "a UTF-8 BOM character, which was found at ",
252                            "byte position {}."
253                        ),
254                        name,
255                        byte_idx
256                    ));
257                }
258                _ => (),
259            }
260            prev = c;
261        }
262
263        Ok(Self { name })
264    }
265
266    /// Construct a table name without validating it.
267    ///
268    /// This breaks API encapsulation and is only intended for use
269    /// when the string was already previously validated.
270    ///
271    /// The QuestDB server will reject an invalid table name.
272    pub fn new_unchecked(name: &'a str) -> Self {
273        Self { name }
274    }
275}
276
277impl<'a> TryFrom<&'a str> for TableName<'a> {
278    type Error = Error;
279
280    fn try_from(name: &'a str) -> crate::Result<Self> {
281        Self::new(name)
282    }
283}
284
285impl AsRef<str> for TableName<'_> {
286    fn as_ref(&self) -> &str {
287        self.name
288    }
289}
290
291/// A validated column name.
292///
293/// This type simply wraps a `&str`.
294///
295/// When you pass a `ColumnName` instead of a plain string to a [`Buffer`] method,
296/// it doesn't have to validate it again. This saves CPU cycles.
297#[derive(Clone, Copy)]
298pub struct ColumnName<'a> {
299    name: &'a str,
300}
301
302impl<'a> ColumnName<'a> {
303    /// Construct a validated column name.
304    pub fn new(name: &'a str) -> crate::Result<Self> {
305        if name.is_empty() {
306            return Err(error::fmt!(
307                InvalidName,
308                "Column names must have a non-zero length."
309            ));
310        }
311
312        for (byte_idx, c) in name.char_indices() {
313            match c {
314                '?' | '.' | ',' | '\'' | '\"' | '\\' | '/' | ':' | ')' | '(' | '+' | '-' | '*'
315                | '%' | '~' | '\r' | '\n' | '\0' | '\u{0001}' | '\u{0002}' | '\u{0003}'
316                | '\u{0004}' | '\u{0005}' | '\u{0006}' | '\u{0007}' | '\u{0008}' | '\u{0009}'
317                | '\u{000b}' | '\u{000c}' | '\u{000e}' | '\u{000f}' | '\u{007f}' => {
318                    return Err(error::fmt!(
319                        InvalidName,
320                        concat!(
321                            "Bad string {:?}: ",
322                            "Column names can't contain ",
323                            "a {:?} character, which was found at ",
324                            "byte position {}."
325                        ),
326                        name,
327                        c,
328                        byte_idx
329                    ));
330                }
331                '\u{FEFF}' => {
332                    // Reject Unicode char 'ZERO WIDTH NO-BREAK SPACE',
333                    // aka UTF-8 BOM if it appears anywhere in the string.
334                    return Err(error::fmt!(
335                        InvalidName,
336                        concat!(
337                            "Bad string {:?}: ",
338                            "Column names can't contain ",
339                            "a UTF-8 BOM character, which was found at ",
340                            "byte position {}."
341                        ),
342                        name,
343                        byte_idx
344                    ));
345                }
346                _ => (),
347            }
348        }
349
350        Ok(Self { name })
351    }
352
353    /// Construct a column name without validating it.
354    ///
355    /// This breaks API encapsulation and is only intended for use
356    /// when the string was already previously validated.
357    ///
358    /// The QuestDB server will reject an invalid column name.
359    pub fn new_unchecked(name: &'a str) -> Self {
360        Self { name }
361    }
362}
363
364impl<'a> TryFrom<&'a str> for ColumnName<'a> {
365    type Error = Error;
366
367    fn try_from(name: &'a str) -> crate::Result<Self> {
368        Self::new(name)
369    }
370}
371
372impl AsRef<str> for ColumnName<'_> {
373    fn as_ref(&self) -> &str {
374        self.name
375    }
376}
377
378#[derive(Clone, Debug)]
379enum BufferInner {
380    Ilp(IlpBuffer),
381
382    #[cfg(any(feature = "_sender-qwp-udp", feature = "_sender-qwp-ws"))]
383    Qwp(Box<QwpBuffer>),
384
385    #[cfg(feature = "_sender-qwp-ws")]
386    QwpWs(Box<QwpWsColumnarBuffer>),
387}
388
389/// A reusable row buffer.
390///
391/// For ILP senders this exposes the existing byte-oriented buffer implementation.
392/// For QWP/UDP senders it dispatches to the QWP-specific row buffer.
393#[derive(Clone, Debug)]
394pub struct Buffer {
395    inner: BufferInner,
396}
397
398// Keep the public buffer movable and shareable across threads. In particular,
399// QWP/WebSocket's lazily maintained size hint must not silently weaken this
400// contract through interior-mutability choices such as `RefCell`.
401const _: fn() = || {
402    fn assert_send_sync<T: Send + Sync>() {}
403    assert_send_sync::<Buffer>();
404};
405
406impl Buffer {
407    /// Creates a new ILP buffer with default parameters.
408    pub fn new(protocol_version: ProtocolVersion) -> Self {
409        Self {
410            inner: BufferInner::Ilp(IlpBuffer::new(protocol_version)),
411        }
412    }
413
414    /// Creates a new ILP buffer with a custom maximum name length.
415    pub fn with_max_name_len(protocol_version: ProtocolVersion, max_name_len: usize) -> Self {
416        Self {
417            inner: BufferInner::Ilp(IlpBuffer::with_max_name_len(protocol_version, max_name_len)),
418        }
419    }
420
421    /// Creates a new ILP buffer that pre-allocates its byte storage to
422    /// `init_capacity` and accepts table / column names up to `max_name_len`.
423    /// The buffer is allowed to grow past `init_capacity`; it is purely a
424    /// starting-size hint to avoid early reallocations.
425    pub fn with_init_capacity_and_max_name_len(
426        protocol_version: ProtocolVersion,
427        init_capacity: usize,
428        max_name_len: usize,
429    ) -> Self {
430        Self {
431            inner: BufferInner::Ilp(IlpBuffer::with_init_capacity_and_max_name_len(
432                protocol_version,
433                init_capacity,
434                max_name_len,
435            )),
436        }
437    }
438
439    #[cfg(any(feature = "_sender-qwp-udp", feature = "_sender-qwp-ws"))]
440    pub fn new_qwp() -> Self {
441        Self::qwp_with_max_name_len(127)
442    }
443
444    #[cfg(any(feature = "_sender-qwp-udp", feature = "_sender-qwp-ws"))]
445    /// Like [`Buffer::new_qwp`] with an explicit maximum name length.
446    pub fn qwp_with_max_name_len(max_name_len: usize) -> Self {
447        Self {
448            inner: BufferInner::Qwp(Box::new(QwpBuffer::new(max_name_len))),
449        }
450    }
451
452    /// Creates a new QWP/WebSocket columnar buffer with a 127-byte name
453    /// length limit. Accepts the row-by-row `table` / `symbol` /
454    /// `column_*` / `at` API; consumed by [`Sender::flush`].
455    ///
456    /// [`Sender::flush`]: crate::ingress::Sender::flush
457    #[cfg(feature = "_sender-qwp-ws")]
458    pub fn new_qwp_ws() -> Self {
459        Self::qwp_ws_with_max_name_len(127)
460    }
461
462    /// Like [`Buffer::new_qwp_ws`] with an explicit maximum name length.
463    #[cfg(feature = "_sender-qwp-ws")]
464    pub fn qwp_ws_with_max_name_len(max_name_len: usize) -> Self {
465        Self {
466            inner: BufferInner::QwpWs(Box::new(QwpWsColumnarBuffer::new(max_name_len))),
467        }
468    }
469
470    #[cfg(feature = "_sync-sender")]
471    pub(crate) fn as_ilp(&self) -> Option<&IlpBuffer> {
472        match &self.inner {
473            BufferInner::Ilp(inner) => Some(inner),
474            #[cfg(any(feature = "_sender-qwp-udp", feature = "_sender-qwp-ws"))]
475            BufferInner::Qwp(_) => None,
476            #[cfg(feature = "_sender-qwp-ws")]
477            BufferInner::QwpWs(_) => None,
478        }
479    }
480
481    #[cfg(any(
482        feature = "_sender-qwp-udp",
483        all(test, feature = "_sender-qwp-ws", feature = "_sender-http")
484    ))]
485    pub(crate) fn as_qwp(&self) -> Option<&QwpBuffer> {
486        match &self.inner {
487            BufferInner::Ilp(_) => None,
488            BufferInner::Qwp(inner) => Some(inner.as_ref()),
489            #[cfg(feature = "_sender-qwp-ws")]
490            BufferInner::QwpWs(_) => None,
491        }
492    }
493
494    #[cfg(feature = "_sender-qwp-ws")]
495    pub(crate) fn as_qwp_ws(&self) -> Option<&QwpWsColumnarBuffer> {
496        match &self.inner {
497            BufferInner::Ilp(_) => None,
498            #[cfg(any(feature = "_sender-qwp-udp", feature = "_sender-qwp-ws"))]
499            BufferInner::Qwp(_) => None,
500            BufferInner::QwpWs(inner) => Some(inner.as_ref()),
501        }
502    }
503
504    /// Returns the protocol version associated with this buffer.
505    ///
506    /// For ILP buffers this is the ILP protocol version. For QWP/UDP buffers
507    /// this is the QWP datagram version, currently represented as
508    /// [`ProtocolVersion::V1`]. Interpret the value together with the buffer
509    /// transport; do not use it by itself for ILP feature gating.
510    pub fn protocol_version(&self) -> ProtocolVersion {
511        match &self.inner {
512            BufferInner::Ilp(inner) => inner.protocol_version(),
513            #[cfg(any(feature = "_sender-qwp-udp", feature = "_sender-qwp-ws"))]
514            BufferInner::Qwp(_) => ProtocolVersion::V1,
515            #[cfg(feature = "_sender-qwp-ws")]
516            BufferInner::QwpWs(_) => ProtocolVersion::V1,
517        }
518    }
519
520    /// Reserves capacity associated with `additional` more bytes of buffered data.
521    ///
522    /// For ILP buffers this reserves exact serialized-byte capacity. For
523    /// QWP/UDP buffers this is a heuristic prewarm of the internal arenas and
524    /// planner scratch used during datagram planning and encoding; it is not an
525    /// exact guarantee that [`Buffer::len`] can grow by `additional` bytes
526    /// without further allocation.
527    pub fn reserve(&mut self, additional: usize) {
528        match &mut self.inner {
529            BufferInner::Ilp(inner) => inner.reserve(additional),
530            #[cfg(any(feature = "_sender-qwp-udp", feature = "_sender-qwp-ws"))]
531            BufferInner::Qwp(inner) => inner.reserve(additional),
532            #[cfg(feature = "_sender-qwp-ws")]
533            BufferInner::QwpWs(inner) => inner.reserve(additional),
534        }
535    }
536
537    /// Returns the current buffered size.
538    ///
539    /// For ILP buffers this is the exact serialized byte count. For QWP/UDP
540    /// buffers this is the size hint used for flush planning. For QWP/WebSocket
541    /// buffers this is only a local size hint; symbol-ID remapping and replay
542    /// dictionary state can change the eventual frame size. The sender
543    /// enforces `max_buf_size` against the encoded replay message.
544    pub fn len(&self) -> usize {
545        match &self.inner {
546            BufferInner::Ilp(inner) => inner.len(),
547            #[cfg(any(feature = "_sender-qwp-udp", feature = "_sender-qwp-ws"))]
548            BufferInner::Qwp(inner) => inner.len(),
549            #[cfg(feature = "_sender-qwp-ws")]
550            BufferInner::QwpWs(inner) => inner.len(),
551        }
552    }
553
554    /// Returns the number of completed rows currently buffered.
555    ///
556    /// A row is counted only after [`Buffer::at`] or [`Buffer::at_now`] completes
557    /// it.
558    pub fn row_count(&self) -> usize {
559        match &self.inner {
560            BufferInner::Ilp(inner) => inner.row_count(),
561            #[cfg(any(feature = "_sender-qwp-udp", feature = "_sender-qwp-ws"))]
562            BufferInner::Qwp(inner) => inner.row_count(),
563            #[cfg(feature = "_sender-qwp-ws")]
564            BufferInner::QwpWs(inner) => inner.row_count(),
565        }
566    }
567
568    /// Returns whether the buffered batch is transactional.
569    ///
570    /// For ILP buffers this is `true` only while the buffer contains rows for
571    /// at most one table. QWP/UDP does not support transactional flushes, so
572    /// QWP buffers always return `false`.
573    pub fn transactional(&self) -> bool {
574        match &self.inner {
575            BufferInner::Ilp(inner) => inner.transactional(),
576            #[cfg(any(feature = "_sender-qwp-udp", feature = "_sender-qwp-ws"))]
577            BufferInner::Qwp(_) => false,
578            #[cfg(feature = "_sender-qwp-ws")]
579            BufferInner::QwpWs(_) => false,
580        }
581    }
582
583    /// Returns `true` if the buffer contains no committed or in-progress rows.
584    pub fn is_empty(&self) -> bool {
585        match &self.inner {
586            BufferInner::Ilp(inner) => inner.is_empty(),
587            #[cfg(any(feature = "_sender-qwp-udp", feature = "_sender-qwp-ws"))]
588            BufferInner::Qwp(inner) => inner.is_empty(),
589            #[cfg(feature = "_sender-qwp-ws")]
590            BufferInner::QwpWs(inner) => inner.is_empty(),
591        }
592    }
593
594    /// Returns the current retained-capacity hint for the buffer.
595    ///
596    /// For ILP buffers, this is byte capacity. For QWP/UDP buffers, this is an
597    /// implementation-defined retained-capacity hint and should not be
598    /// interpreted as exact byte capacity.
599    pub fn capacity(&self) -> usize {
600        match &self.inner {
601            BufferInner::Ilp(inner) => inner.capacity(),
602            #[cfg(any(feature = "_sender-qwp-udp", feature = "_sender-qwp-ws"))]
603            BufferInner::Qwp(inner) => inner.capacity(),
604            #[cfg(feature = "_sender-qwp-ws")]
605            BufferInner::QwpWs(inner) => inner.capacity(),
606        }
607    }
608
609    /// Returns the raw serialized ILP bytes currently stored in the buffer.
610    ///
611    /// QWP/UDP buffers build datagrams during flush, so this returns an empty
612    /// slice for QWP/UDP.
613    pub fn as_bytes(&self) -> &[u8] {
614        match &self.inner {
615            BufferInner::Ilp(inner) => inner.as_bytes(),
616            #[cfg(any(feature = "_sender-qwp-udp", feature = "_sender-qwp-ws"))]
617            BufferInner::Qwp(inner) => inner.as_bytes(),
618            #[cfg(feature = "_sender-qwp-ws")]
619            BufferInner::QwpWs(inner) => inner.as_bytes(),
620        }
621    }
622
623    /// Marks the current buffer state so it can later be restored with
624    /// [`Buffer::rewind_to_marker`].
625    ///
626    /// Setting a new marker replaces the currently stored rewind point,
627    /// including one established by [`Buffer::bookmark`].
628    pub fn set_marker(&mut self) -> crate::Result<()> {
629        match &mut self.inner {
630            BufferInner::Ilp(inner) => inner.set_marker(),
631            #[cfg(any(feature = "_sender-qwp-udp", feature = "_sender-qwp-ws"))]
632            BufferInner::Qwp(inner) => inner.set_marker(),
633            #[cfg(feature = "_sender-qwp-ws")]
634            BufferInner::QwpWs(inner) => inner.set_marker(),
635        }
636    }
637
638    /// Captures the current buffer state so it can later be restored with
639    /// [`Buffer::rewind_to_bookmark`].
640    ///
641    /// Capturing a new bookmark replaces the previous bookmark or marker.
642    pub fn bookmark(&mut self) -> crate::Result<Bookmark> {
643        match &mut self.inner {
644            BufferInner::Ilp(inner) => inner.bookmark(),
645            #[cfg(any(feature = "_sender-qwp-udp", feature = "_sender-qwp-ws"))]
646            BufferInner::Qwp(inner) => inner.bookmark(),
647            #[cfg(feature = "_sender-qwp-ws")]
648            BufferInner::QwpWs(inner) => inner.bookmark(),
649        }
650    }
651
652    /// Rewinds the buffer to the state referenced by `bookmark` and then
653    /// clears that bookmark.
654    pub fn rewind_to_bookmark(&mut self, bookmark: Bookmark) -> crate::Result<()> {
655        match &mut self.inner {
656            BufferInner::Ilp(inner) => inner.rewind_to_bookmark(bookmark),
657            #[cfg(any(feature = "_sender-qwp-udp", feature = "_sender-qwp-ws"))]
658            BufferInner::Qwp(inner) => inner.rewind_to_bookmark(bookmark),
659            #[cfg(feature = "_sender-qwp-ws")]
660            BufferInner::QwpWs(inner) => inner.rewind_to_bookmark(bookmark),
661        }
662    }
663
664    /// Clears `bookmark` if it is still the currently active bookmark.
665    pub fn clear_bookmark(&mut self, bookmark: Bookmark) {
666        match &mut self.inner {
667            BufferInner::Ilp(inner) => inner.clear_bookmark(bookmark),
668            #[cfg(any(feature = "_sender-qwp-udp", feature = "_sender-qwp-ws"))]
669            BufferInner::Qwp(inner) => inner.clear_bookmark(bookmark),
670            #[cfg(feature = "_sender-qwp-ws")]
671            BufferInner::QwpWs(inner) => inner.clear_bookmark(bookmark),
672        }
673    }
674
675    /// Rewinds the buffer to the currently stored rewind point and then clears
676    /// it.
677    ///
678    /// This may rewind a state established by either [`Buffer::set_marker`] or
679    /// [`Buffer::bookmark`].
680    ///
681    /// Returns an error if no rewind point is set.
682    pub fn rewind_to_marker(&mut self) -> crate::Result<()> {
683        match &mut self.inner {
684            BufferInner::Ilp(inner) => inner.rewind_to_marker(),
685            #[cfg(any(feature = "_sender-qwp-udp", feature = "_sender-qwp-ws"))]
686            BufferInner::Qwp(inner) => inner.rewind_to_marker(),
687            #[cfg(feature = "_sender-qwp-ws")]
688            BufferInner::QwpWs(inner) => inner.rewind_to_marker(),
689        }
690    }
691
692    /// Clears the current stored rewind point, including one established by
693    /// [`Buffer::bookmark`].
694    pub fn clear_marker(&mut self) {
695        match &mut self.inner {
696            BufferInner::Ilp(inner) => inner.clear_marker(),
697            #[cfg(any(feature = "_sender-qwp-udp", feature = "_sender-qwp-ws"))]
698            BufferInner::Qwp(inner) => inner.clear_marker(),
699            #[cfg(feature = "_sender-qwp-ws")]
700            BufferInner::QwpWs(inner) => inner.clear_marker(),
701        }
702    }
703
704    /// Clears the buffer contents and marker while retaining allocated capacity.
705    pub fn clear(&mut self) {
706        match &mut self.inner {
707            BufferInner::Ilp(inner) => inner.clear(),
708            #[cfg(any(feature = "_sender-qwp-udp", feature = "_sender-qwp-ws"))]
709            BufferInner::Qwp(inner) => inner.clear(),
710            #[cfg(feature = "_sender-qwp-ws")]
711            BufferInner::QwpWs(inner) => inner.clear(),
712        }
713    }
714
715    /// Validates that the buffer is ready to be flushed with
716    /// `crate::ingress::Sender::flush` or one of its variants.
717    ///
718    /// Returns an error when the current API call sequence is incomplete, such
719    /// as an unfinished row.
720    pub fn check_can_flush(&self) -> crate::Result<()> {
721        match &self.inner {
722            BufferInner::Ilp(inner) => inner.check_can_flush(),
723            #[cfg(any(feature = "_sender-qwp-udp", feature = "_sender-qwp-ws"))]
724            BufferInner::Qwp(inner) => inner.check_can_flush(),
725            #[cfg(feature = "_sender-qwp-ws")]
726            BufferInner::QwpWs(inner) => inner.check_can_flush(),
727        }
728    }
729
730    /// Starts a new row for `name`.
731    ///
732    /// Every row must begin with a table name. See [`Buffer`] for the full call
733    /// sequence.
734    #[inline(always)]
735    pub fn table<'a, N>(&mut self, name: N) -> crate::Result<&mut Self>
736    where
737        N: AsRef<str> + TryInto<TableName<'a>>,
738        Error: From<N::Error>,
739    {
740        match &mut self.inner {
741            BufferInner::Ilp(inner) => {
742                inner.table(name)?;
743            }
744            #[cfg(any(feature = "_sender-qwp-udp", feature = "_sender-qwp-ws"))]
745            BufferInner::Qwp(inner) => {
746                inner.table(name)?;
747            }
748            #[cfg(feature = "_sender-qwp-ws")]
749            BufferInner::QwpWs(inner) => {
750                inner.table(name)?;
751            }
752        }
753        Ok(self)
754    }
755
756    /// Adds a symbol column to the current row.
757    ///
758    /// All symbol columns must be recorded before any non-symbol columns.
759    ///
760    /// When the buffer is flushed over QWP/WebSocket, every distinct symbol
761    /// recorded here is interned into the *same* connection-scoped dictionary
762    /// the column/chunk API uses — capped at 2,000,000 entries and 256 MiB of
763    /// UTF-8 across the whole connection, not per buffer or per flush.
764    /// Exceeding it fails the flush with
765    /// [`SymbolDictFull`](crate::ErrorCode::SymbolDictFull), and the dictionary
766    /// is only reset by retiring the connection that owns it — which a full
767    /// dictionary now does automatically on return, so a pooled sender is dropped
768    /// (not recycled) and the next borrow gets a fresh one. Wait / commit first if
769    /// frames flushed earlier must not be lost; see
770    /// [`SymbolDictFull`](crate::ErrorCode::SymbolDictFull) for the per-API
771    /// list. ILP (TCP/HTTP) flushes carry no such dictionary and are unaffected.
772    #[inline(always)]
773    pub fn symbol<'a, N, S>(&mut self, name: N, value: S) -> crate::Result<&mut Self>
774    where
775        N: AsRef<str> + TryInto<ColumnName<'a>>,
776        S: AsRef<str>,
777        Error: From<N::Error>,
778    {
779        match &mut self.inner {
780            BufferInner::Ilp(inner) => {
781                inner.symbol(name, value)?;
782            }
783            #[cfg(any(feature = "_sender-qwp-udp", feature = "_sender-qwp-ws"))]
784            BufferInner::Qwp(inner) => {
785                inner.symbol(name, value)?;
786            }
787            #[cfg(feature = "_sender-qwp-ws")]
788            BufferInner::QwpWs(inner) => {
789                inner.symbol(name, value)?;
790            }
791        }
792        Ok(self)
793    }
794
795    /// Adds a symbol column if `value` is `Some`; otherwise leaves the row unchanged.
796    ///
797    /// See [`symbol`](Self::symbol) for the QWP/WebSocket connection-scoped
798    /// dictionary cap that applies to every symbol recorded this way.
799    pub fn symbol_opt<'a, N, S>(&mut self, name: N, value: Option<S>) -> crate::Result<&mut Self>
800    where
801        N: AsRef<str> + TryInto<ColumnName<'a>>,
802        S: AsRef<str>,
803        Error: From<N::Error>,
804    {
805        if let Some(value) = value {
806            self.symbol(name, value)
807        } else {
808            Ok(self)
809        }
810    }
811
812    /// Adds a boolean column to the current row.
813    pub fn column_bool<'a, N>(&mut self, name: N, value: bool) -> crate::Result<&mut Self>
814    where
815        N: AsRef<str> + TryInto<ColumnName<'a>>,
816        Error: From<N::Error>,
817    {
818        match &mut self.inner {
819            BufferInner::Ilp(inner) => {
820                inner.column_bool(name, value)?;
821            }
822            #[cfg(any(feature = "_sender-qwp-udp", feature = "_sender-qwp-ws"))]
823            BufferInner::Qwp(inner) => {
824                inner.column_bool(name, value)?;
825            }
826            #[cfg(feature = "_sender-qwp-ws")]
827            BufferInner::QwpWs(inner) => {
828                inner.column_bool(name, value)?;
829            }
830        }
831        Ok(self)
832    }
833
834    /// Adds a boolean column if `value` is `Some`; otherwise leaves the row unchanged.
835    pub fn column_bool_opt<'a, N>(
836        &mut self,
837        name: N,
838        value: Option<bool>,
839    ) -> crate::Result<&mut Self>
840    where
841        N: AsRef<str> + TryInto<ColumnName<'a>>,
842        Error: From<N::Error>,
843    {
844        if let Some(value) = value {
845            self.column_bool(name, value)
846        } else {
847            Ok(self)
848        }
849    }
850
851    /// Adds an integer column to the current row.
852    #[inline(always)]
853    pub fn column_i64<'a, N>(&mut self, name: N, value: i64) -> crate::Result<&mut Self>
854    where
855        N: AsRef<str> + TryInto<ColumnName<'a>>,
856        Error: From<N::Error>,
857    {
858        match &mut self.inner {
859            BufferInner::Ilp(inner) => {
860                inner.column_i64(name, value)?;
861            }
862            #[cfg(any(feature = "_sender-qwp-udp", feature = "_sender-qwp-ws"))]
863            BufferInner::Qwp(inner) => {
864                inner.column_i64(name, value)?;
865            }
866            #[cfg(feature = "_sender-qwp-ws")]
867            BufferInner::QwpWs(inner) => {
868                inner.column_i64(name, value)?;
869            }
870        }
871        Ok(self)
872    }
873
874    /// Adds an integer column if `value` is `Some`; otherwise leaves the row unchanged.
875    pub fn column_i64_opt<'a, N>(&mut self, name: N, value: Option<i64>) -> crate::Result<&mut Self>
876    where
877        N: AsRef<str> + TryInto<ColumnName<'a>>,
878        Error: From<N::Error>,
879    {
880        if let Some(value) = value {
881            self.column_i64(name, value)
882        } else {
883            Ok(self)
884        }
885    }
886
887    /// Adds an 8-bit signed integer column to the current row. QWP-only.
888    pub fn column_i8<'a, N>(&mut self, name: N, value: i8) -> crate::Result<&mut Self>
889    where
890        N: AsRef<str> + TryInto<ColumnName<'a>>,
891        Error: From<N::Error>,
892    {
893        let _ = (&name, &value);
894        match &mut self.inner {
895            BufferInner::Ilp(_) => Err(error::fmt!(
896                InvalidApiCall,
897                "column_i8 requires a QWP transport (ws:: or udp::)"
898            )),
899            #[cfg(any(feature = "_sender-qwp-udp", feature = "_sender-qwp-ws"))]
900            BufferInner::Qwp(inner) => {
901                inner.column_i8(name, value)?;
902                Ok(self)
903            }
904            #[cfg(feature = "_sender-qwp-ws")]
905            BufferInner::QwpWs(inner) => {
906                inner.column_i8(name, value)?;
907                Ok(self)
908            }
909        }
910    }
911
912    /// Adds an 8-bit signed integer column if `value` is `Some`. QWP-only.
913    pub fn column_i8_opt<'a, N>(&mut self, name: N, value: Option<i8>) -> crate::Result<&mut Self>
914    where
915        N: AsRef<str> + TryInto<ColumnName<'a>>,
916        Error: From<N::Error>,
917    {
918        if let Some(value) = value {
919            self.column_i8(name, value)
920        } else {
921            Ok(self)
922        }
923    }
924
925    /// Adds a 16-bit signed integer column to the current row. QWP-only.
926    pub fn column_i16<'a, N>(&mut self, name: N, value: i16) -> crate::Result<&mut Self>
927    where
928        N: AsRef<str> + TryInto<ColumnName<'a>>,
929        Error: From<N::Error>,
930    {
931        let _ = (&name, &value);
932        match &mut self.inner {
933            BufferInner::Ilp(_) => Err(error::fmt!(
934                InvalidApiCall,
935                "column_i16 requires a QWP transport (ws:: or udp::)"
936            )),
937            #[cfg(any(feature = "_sender-qwp-udp", feature = "_sender-qwp-ws"))]
938            BufferInner::Qwp(inner) => {
939                inner.column_i16(name, value)?;
940                Ok(self)
941            }
942            #[cfg(feature = "_sender-qwp-ws")]
943            BufferInner::QwpWs(inner) => {
944                inner.column_i16(name, value)?;
945                Ok(self)
946            }
947        }
948    }
949
950    /// Adds a 16-bit signed integer column if `value` is `Some`. QWP-only.
951    pub fn column_i16_opt<'a, N>(&mut self, name: N, value: Option<i16>) -> crate::Result<&mut Self>
952    where
953        N: AsRef<str> + TryInto<ColumnName<'a>>,
954        Error: From<N::Error>,
955    {
956        if let Some(value) = value {
957            self.column_i16(name, value)
958        } else {
959            Ok(self)
960        }
961    }
962
963    /// Adds a 32-bit signed integer column to the current row. QWP-only.
964    pub fn column_i32<'a, N>(&mut self, name: N, value: i32) -> crate::Result<&mut Self>
965    where
966        N: AsRef<str> + TryInto<ColumnName<'a>>,
967        Error: From<N::Error>,
968    {
969        let _ = (&name, &value);
970        match &mut self.inner {
971            BufferInner::Ilp(_) => Err(error::fmt!(
972                InvalidApiCall,
973                "column_i32 requires a QWP transport (ws:: or udp::)"
974            )),
975            #[cfg(any(feature = "_sender-qwp-udp", feature = "_sender-qwp-ws"))]
976            BufferInner::Qwp(inner) => {
977                inner.column_i32(name, value)?;
978                Ok(self)
979            }
980            #[cfg(feature = "_sender-qwp-ws")]
981            BufferInner::QwpWs(inner) => {
982                inner.column_i32(name, value)?;
983                Ok(self)
984            }
985        }
986    }
987
988    /// Adds a 32-bit signed integer column if `value` is `Some`. QWP-only.
989    pub fn column_i32_opt<'a, N>(&mut self, name: N, value: Option<i32>) -> crate::Result<&mut Self>
990    where
991        N: AsRef<str> + TryInto<ColumnName<'a>>,
992        Error: From<N::Error>,
993    {
994        if let Some(value) = value {
995            self.column_i32(name, value)
996        } else {
997            Ok(self)
998        }
999    }
1000
1001    /// Adds a 32-bit floating-point column to the current row. QWP-only.
1002    pub fn column_f32<'a, N>(&mut self, name: N, value: f32) -> crate::Result<&mut Self>
1003    where
1004        N: AsRef<str> + TryInto<ColumnName<'a>>,
1005        Error: From<N::Error>,
1006    {
1007        let _ = (&name, &value);
1008        match &mut self.inner {
1009            BufferInner::Ilp(_) => Err(error::fmt!(
1010                InvalidApiCall,
1011                "column_f32 requires a QWP transport (ws:: or udp::)"
1012            )),
1013            #[cfg(any(feature = "_sender-qwp-udp", feature = "_sender-qwp-ws"))]
1014            BufferInner::Qwp(inner) => {
1015                inner.column_f32(name, value)?;
1016                Ok(self)
1017            }
1018            #[cfg(feature = "_sender-qwp-ws")]
1019            BufferInner::QwpWs(inner) => {
1020                inner.column_f32(name, value)?;
1021                Ok(self)
1022            }
1023        }
1024    }
1025
1026    /// Adds a 32-bit floating-point column if `value` is `Some`. QWP-only.
1027    pub fn column_f32_opt<'a, N>(&mut self, name: N, value: Option<f32>) -> crate::Result<&mut Self>
1028    where
1029        N: AsRef<str> + TryInto<ColumnName<'a>>,
1030        Error: From<N::Error>,
1031    {
1032        if let Some(value) = value {
1033            self.column_f32(name, value)
1034        } else {
1035            Ok(self)
1036        }
1037    }
1038
1039    /// Adds a floating-point column to the current row.
1040    #[inline(always)]
1041    pub fn column_f64<'a, N>(&mut self, name: N, value: f64) -> crate::Result<&mut Self>
1042    where
1043        N: AsRef<str> + TryInto<ColumnName<'a>>,
1044        Error: From<N::Error>,
1045    {
1046        match &mut self.inner {
1047            BufferInner::Ilp(inner) => {
1048                inner.column_f64(name, value)?;
1049            }
1050            #[cfg(any(feature = "_sender-qwp-udp", feature = "_sender-qwp-ws"))]
1051            BufferInner::Qwp(inner) => {
1052                inner.column_f64(name, value)?;
1053            }
1054            #[cfg(feature = "_sender-qwp-ws")]
1055            BufferInner::QwpWs(inner) => {
1056                inner.column_f64(name, value)?;
1057            }
1058        }
1059        Ok(self)
1060    }
1061
1062    /// Adds a floating-point column if `value` is `Some`; otherwise leaves the row unchanged.
1063    pub fn column_f64_opt<'a, N>(&mut self, name: N, value: Option<f64>) -> crate::Result<&mut Self>
1064    where
1065        N: AsRef<str> + TryInto<ColumnName<'a>>,
1066        Error: From<N::Error>,
1067    {
1068        if let Some(value) = value {
1069            self.column_f64(name, value)
1070        } else {
1071            Ok(self)
1072        }
1073    }
1074
1075    /// Adds a string column to the current row.
1076    #[inline(always)]
1077    pub fn column_str<'a, N, S>(&mut self, name: N, value: S) -> crate::Result<&mut Self>
1078    where
1079        N: AsRef<str> + TryInto<ColumnName<'a>>,
1080        S: AsRef<str>,
1081        Error: From<N::Error>,
1082    {
1083        match &mut self.inner {
1084            BufferInner::Ilp(inner) => {
1085                inner.column_str(name, value)?;
1086            }
1087            #[cfg(any(feature = "_sender-qwp-udp", feature = "_sender-qwp-ws"))]
1088            BufferInner::Qwp(inner) => {
1089                inner.column_str(name, value)?;
1090            }
1091            #[cfg(feature = "_sender-qwp-ws")]
1092            BufferInner::QwpWs(inner) => {
1093                inner.column_str(name, value)?;
1094            }
1095        }
1096        Ok(self)
1097    }
1098
1099    /// Adds a string column if `value` is `Some`; otherwise leaves the row unchanged.
1100    pub fn column_str_opt<'a, N, S>(
1101        &mut self,
1102        name: N,
1103        value: Option<S>,
1104    ) -> crate::Result<&mut Self>
1105    where
1106        N: AsRef<str> + TryInto<ColumnName<'a>>,
1107        S: AsRef<str>,
1108        Error: From<N::Error>,
1109    {
1110        if let Some(value) = value {
1111            self.column_str(name, value)
1112        } else {
1113            Ok(self)
1114        }
1115    }
1116
1117    /// Adds a decimal column to the current row.
1118    ///
1119    /// Returns an error if the active protocol does not support decimal values.
1120    /// QWP/UDP accepts the same decimal input forms as ILP and encodes them as
1121    /// nullable DECIMAL256 columns on the wire.
1122    pub fn column_dec<'a, N, S>(&mut self, name: N, value: S) -> crate::Result<&mut Self>
1123    where
1124        N: AsRef<str> + TryInto<ColumnName<'a>>,
1125        S: TryInto<DecimalView<'a>>,
1126        Error: From<N::Error>,
1127        Error: From<S::Error>,
1128    {
1129        match &mut self.inner {
1130            BufferInner::Ilp(inner) => {
1131                inner.column_dec(name, value)?;
1132            }
1133            #[cfg(any(feature = "_sender-qwp-udp", feature = "_sender-qwp-ws"))]
1134            BufferInner::Qwp(inner) => {
1135                inner.column_dec(name, value)?;
1136            }
1137            #[cfg(feature = "_sender-qwp-ws")]
1138            BufferInner::QwpWs(inner) => {
1139                inner.column_dec(name, value)?;
1140            }
1141        }
1142        Ok(self)
1143    }
1144
1145    /// Adds a decimal column if `value` is `Some`; otherwise leaves the row unchanged.
1146    pub fn column_dec_opt<'a, N, S>(
1147        &mut self,
1148        name: N,
1149        value: Option<S>,
1150    ) -> crate::Result<&mut Self>
1151    where
1152        N: AsRef<str> + TryInto<ColumnName<'a>>,
1153        S: TryInto<DecimalView<'a>>,
1154        Error: From<N::Error>,
1155        Error: From<S::Error>,
1156    {
1157        if let Some(value) = value {
1158            self.column_dec(name, value)
1159        } else {
1160            Ok(self)
1161        }
1162    }
1163
1164    /// Adds a 64-bit decimal column to the current row. QWP-only.
1165    ///
1166    /// The unscaled magnitude (at the column's pinned scale) must fit a signed
1167    /// 64-bit integer; values that do not fit return `InvalidApiCall`.
1168    pub fn column_dec64<'a, N, S>(&mut self, name: N, value: S) -> crate::Result<&mut Self>
1169    where
1170        N: AsRef<str> + TryInto<ColumnName<'a>>,
1171        S: TryInto<DecimalView<'a>>,
1172        Error: From<N::Error>,
1173        Error: From<S::Error>,
1174    {
1175        let _ = &name;
1176        match &mut self.inner {
1177            BufferInner::Ilp(_) => {
1178                let _ = value.try_into().map_err(Error::from)?;
1179                Err(error::fmt!(
1180                    InvalidApiCall,
1181                    "column_dec64 requires a QWP transport (ws:: or udp::)"
1182                ))
1183            }
1184            #[cfg(any(feature = "_sender-qwp-udp", feature = "_sender-qwp-ws"))]
1185            BufferInner::Qwp(inner) => {
1186                inner.column_dec64(name, value)?;
1187                Ok(self)
1188            }
1189            #[cfg(feature = "_sender-qwp-ws")]
1190            BufferInner::QwpWs(inner) => {
1191                inner.column_dec64(name, value)?;
1192                Ok(self)
1193            }
1194        }
1195    }
1196
1197    /// Adds a 64-bit decimal column if `value` is `Some`. QWP-only.
1198    pub fn column_dec64_opt<'a, N, S>(
1199        &mut self,
1200        name: N,
1201        value: Option<S>,
1202    ) -> crate::Result<&mut Self>
1203    where
1204        N: AsRef<str> + TryInto<ColumnName<'a>>,
1205        S: TryInto<DecimalView<'a>>,
1206        Error: From<N::Error>,
1207        Error: From<S::Error>,
1208    {
1209        if let Some(value) = value {
1210            self.column_dec64(name, value)
1211        } else {
1212            Ok(self)
1213        }
1214    }
1215
1216    /// Adds a 128-bit decimal column to the current row. QWP-only.
1217    ///
1218    /// The unscaled magnitude (at the column's pinned scale) must fit a signed
1219    /// 128-bit integer; values that do not fit return `InvalidApiCall`.
1220    pub fn column_dec128<'a, N, S>(&mut self, name: N, value: S) -> crate::Result<&mut Self>
1221    where
1222        N: AsRef<str> + TryInto<ColumnName<'a>>,
1223        S: TryInto<DecimalView<'a>>,
1224        Error: From<N::Error>,
1225        Error: From<S::Error>,
1226    {
1227        let _ = &name;
1228        match &mut self.inner {
1229            BufferInner::Ilp(_) => {
1230                let _ = value.try_into().map_err(Error::from)?;
1231                Err(error::fmt!(
1232                    InvalidApiCall,
1233                    "column_dec128 requires a QWP transport (ws:: or udp::)"
1234                ))
1235            }
1236            #[cfg(any(feature = "_sender-qwp-udp", feature = "_sender-qwp-ws"))]
1237            BufferInner::Qwp(inner) => {
1238                inner.column_dec128(name, value)?;
1239                Ok(self)
1240            }
1241            #[cfg(feature = "_sender-qwp-ws")]
1242            BufferInner::QwpWs(inner) => {
1243                inner.column_dec128(name, value)?;
1244                Ok(self)
1245            }
1246        }
1247    }
1248
1249    /// Adds a 128-bit decimal column if `value` is `Some`. QWP-only.
1250    pub fn column_dec128_opt<'a, N, S>(
1251        &mut self,
1252        name: N,
1253        value: Option<S>,
1254    ) -> crate::Result<&mut Self>
1255    where
1256        N: AsRef<str> + TryInto<ColumnName<'a>>,
1257        S: TryInto<DecimalView<'a>>,
1258        Error: From<N::Error>,
1259        Error: From<S::Error>,
1260    {
1261        if let Some(value) = value {
1262            self.column_dec128(name, value)
1263        } else {
1264            Ok(self)
1265        }
1266    }
1267
1268    /// Adds a UUID column to the current row. QWP-only.
1269    ///
1270    /// Per spec, the wire encoding writes `lo` (8 bytes LE) followed by `hi`
1271    /// (8 bytes LE).
1272    pub fn column_uuid<'a, N>(&mut self, name: N, lo: u64, hi: u64) -> crate::Result<&mut Self>
1273    where
1274        N: AsRef<str> + TryInto<ColumnName<'a>>,
1275        Error: From<N::Error>,
1276    {
1277        let _ = &name;
1278        let _ = (lo, hi);
1279        match &mut self.inner {
1280            BufferInner::Ilp(_) => Err(error::fmt!(
1281                InvalidApiCall,
1282                "column_uuid requires a QWP transport (ws:: or udp::)"
1283            )),
1284            #[cfg(any(feature = "_sender-qwp-udp", feature = "_sender-qwp-ws"))]
1285            BufferInner::Qwp(inner) => {
1286                inner.column_uuid(name, lo, hi)?;
1287                Ok(self)
1288            }
1289            #[cfg(feature = "_sender-qwp-ws")]
1290            BufferInner::QwpWs(inner) => {
1291                inner.column_uuid(name, lo, hi)?;
1292                Ok(self)
1293            }
1294        }
1295    }
1296
1297    /// Adds a UUID column if `value` is `Some`. QWP-only.
1298    pub fn column_uuid_opt<'a, N>(
1299        &mut self,
1300        name: N,
1301        value: Option<(u64, u64)>,
1302    ) -> crate::Result<&mut Self>
1303    where
1304        N: AsRef<str> + TryInto<ColumnName<'a>>,
1305        Error: From<N::Error>,
1306    {
1307        if let Some((lo, hi)) = value {
1308            self.column_uuid(name, lo, hi)
1309        } else {
1310            Ok(self)
1311        }
1312    }
1313
1314    /// Adds a LONG256 column to the current row. QWP-only.
1315    ///
1316    /// `value` is the wire-format byte buffer: four 64-bit limbs encoded
1317    /// little-endian, least-significant limb first (32 bytes total).
1318    pub fn column_long256<'a, N>(&mut self, name: N, value: &[u8; 32]) -> crate::Result<&mut Self>
1319    where
1320        N: AsRef<str> + TryInto<ColumnName<'a>>,
1321        Error: From<N::Error>,
1322    {
1323        let _ = (&name, value);
1324        match &mut self.inner {
1325            BufferInner::Ilp(_) => Err(error::fmt!(
1326                InvalidApiCall,
1327                "column_long256 requires a QWP transport (ws:: or udp::)"
1328            )),
1329            #[cfg(any(feature = "_sender-qwp-udp", feature = "_sender-qwp-ws"))]
1330            BufferInner::Qwp(inner) => {
1331                inner.column_long256(name, value)?;
1332                Ok(self)
1333            }
1334            #[cfg(feature = "_sender-qwp-ws")]
1335            BufferInner::QwpWs(inner) => {
1336                inner.column_long256(name, value)?;
1337                Ok(self)
1338            }
1339        }
1340    }
1341
1342    /// Adds a LONG256 column if `value` is `Some`. QWP-only.
1343    pub fn column_long256_opt<'a, N>(
1344        &mut self,
1345        name: N,
1346        value: Option<&[u8; 32]>,
1347    ) -> crate::Result<&mut Self>
1348    where
1349        N: AsRef<str> + TryInto<ColumnName<'a>>,
1350        Error: From<N::Error>,
1351    {
1352        if let Some(value) = value {
1353            self.column_long256(name, value)
1354        } else {
1355            Ok(self)
1356        }
1357    }
1358
1359    /// Adds an IPv4 column to the current row. QWP-only.
1360    ///
1361    /// The wire encoding writes the 4 octets as `u32::from(addr).to_le_bytes()`,
1362    /// matching Rust's natural Ipv4Addr packing (octet 0 in the high byte).
1363    ///
1364    /// IPv4 (`0x18`) is part of the QWP v1 spec.
1365    pub fn column_ipv4<'a, N>(
1366        &mut self,
1367        name: N,
1368        value: std::net::Ipv4Addr,
1369    ) -> crate::Result<&mut Self>
1370    where
1371        N: AsRef<str> + TryInto<ColumnName<'a>>,
1372        Error: From<N::Error>,
1373    {
1374        let _ = (&name, value);
1375        let packed = u32::from(value);
1376        let _ = packed;
1377        match &mut self.inner {
1378            BufferInner::Ilp(_) => Err(error::fmt!(
1379                InvalidApiCall,
1380                "column_ipv4 requires a QWP transport (ws:: or udp::)"
1381            )),
1382            #[cfg(any(feature = "_sender-qwp-udp", feature = "_sender-qwp-ws"))]
1383            BufferInner::Qwp(inner) => {
1384                inner.column_ipv4(name, packed)?;
1385                Ok(self)
1386            }
1387            #[cfg(feature = "_sender-qwp-ws")]
1388            BufferInner::QwpWs(inner) => {
1389                inner.column_ipv4(name, packed)?;
1390                Ok(self)
1391            }
1392        }
1393    }
1394
1395    /// Adds an IPv4 column if `value` is `Some`. QWP-only.
1396    pub fn column_ipv4_opt<'a, N>(
1397        &mut self,
1398        name: N,
1399        value: Option<std::net::Ipv4Addr>,
1400    ) -> crate::Result<&mut Self>
1401    where
1402        N: AsRef<str> + TryInto<ColumnName<'a>>,
1403        Error: From<N::Error>,
1404    {
1405        if let Some(value) = value {
1406            self.column_ipv4(name, value)
1407        } else {
1408            Ok(self)
1409        }
1410    }
1411
1412    /// Adds a DATE column (milliseconds since the Unix epoch). QWP-only.
1413    pub fn column_date<'a, N>(&mut self, name: N, millis: i64) -> crate::Result<&mut Self>
1414    where
1415        N: AsRef<str> + TryInto<ColumnName<'a>>,
1416        Error: From<N::Error>,
1417    {
1418        let _ = (&name, &millis);
1419        match &mut self.inner {
1420            BufferInner::Ilp(_) => Err(error::fmt!(
1421                InvalidApiCall,
1422                "column_date requires a QWP transport (ws:: or udp::)"
1423            )),
1424            #[cfg(any(feature = "_sender-qwp-udp", feature = "_sender-qwp-ws"))]
1425            BufferInner::Qwp(inner) => {
1426                inner.column_date(name, millis)?;
1427                Ok(self)
1428            }
1429            #[cfg(feature = "_sender-qwp-ws")]
1430            BufferInner::QwpWs(inner) => {
1431                inner.column_date(name, millis)?;
1432                Ok(self)
1433            }
1434        }
1435    }
1436
1437    /// Adds a DATE column if `value` is `Some`. QWP-only.
1438    pub fn column_date_opt<'a, N>(
1439        &mut self,
1440        name: N,
1441        value: Option<i64>,
1442    ) -> crate::Result<&mut Self>
1443    where
1444        N: AsRef<str> + TryInto<ColumnName<'a>>,
1445        Error: From<N::Error>,
1446    {
1447        if let Some(value) = value {
1448            self.column_date(name, value)
1449        } else {
1450            Ok(self)
1451        }
1452    }
1453
1454    /// Adds a CHAR column (single UTF-16 code unit). QWP-only.
1455    pub fn column_char<'a, N>(&mut self, name: N, value: u16) -> crate::Result<&mut Self>
1456    where
1457        N: AsRef<str> + TryInto<ColumnName<'a>>,
1458        Error: From<N::Error>,
1459    {
1460        let _ = (&name, &value);
1461        match &mut self.inner {
1462            BufferInner::Ilp(_) => Err(error::fmt!(
1463                InvalidApiCall,
1464                "column_char requires a QWP transport (ws:: or udp::)"
1465            )),
1466            #[cfg(any(feature = "_sender-qwp-udp", feature = "_sender-qwp-ws"))]
1467            BufferInner::Qwp(inner) => {
1468                inner.column_char(name, value)?;
1469                Ok(self)
1470            }
1471            #[cfg(feature = "_sender-qwp-ws")]
1472            BufferInner::QwpWs(inner) => {
1473                inner.column_char(name, value)?;
1474                Ok(self)
1475            }
1476        }
1477    }
1478
1479    /// Adds a CHAR column if `value` is `Some`. QWP-only.
1480    pub fn column_char_opt<'a, N>(
1481        &mut self,
1482        name: N,
1483        value: Option<u16>,
1484    ) -> crate::Result<&mut Self>
1485    where
1486        N: AsRef<str> + TryInto<ColumnName<'a>>,
1487        Error: From<N::Error>,
1488    {
1489        if let Some(value) = value {
1490            self.column_char(name, value)
1491        } else {
1492            Ok(self)
1493        }
1494    }
1495
1496    /// Adds a BINARY column (opaque byte sequence). QWP-only.
1497    ///
1498    /// BINARY (`0x17`) is part of the QWP v1 spec.
1499    pub fn column_binary<'a, N>(&mut self, name: N, value: &[u8]) -> crate::Result<&mut Self>
1500    where
1501        N: AsRef<str> + TryInto<ColumnName<'a>>,
1502        Error: From<N::Error>,
1503    {
1504        let _ = (&name, value);
1505        match &mut self.inner {
1506            BufferInner::Ilp(_) => Err(error::fmt!(
1507                InvalidApiCall,
1508                "column_binary requires a QWP transport (ws:: or udp::)"
1509            )),
1510            #[cfg(any(feature = "_sender-qwp-udp", feature = "_sender-qwp-ws"))]
1511            BufferInner::Qwp(inner) => {
1512                inner.column_binary(name, value)?;
1513                Ok(self)
1514            }
1515            #[cfg(feature = "_sender-qwp-ws")]
1516            BufferInner::QwpWs(inner) => {
1517                inner.column_binary(name, value)?;
1518                Ok(self)
1519            }
1520        }
1521    }
1522
1523    /// Adds a BINARY column if `value` is `Some`. QWP-only.
1524    pub fn column_binary_opt<'a, N>(
1525        &mut self,
1526        name: N,
1527        value: Option<&[u8]>,
1528    ) -> crate::Result<&mut Self>
1529    where
1530        N: AsRef<str> + TryInto<ColumnName<'a>>,
1531        Error: From<N::Error>,
1532    {
1533        if let Some(value) = value {
1534            self.column_binary(name, value)
1535        } else {
1536            Ok(self)
1537        }
1538    }
1539
1540    /// Adds a GEOHASH column. `precision_bits` must be in `1..=60` and is
1541    /// pinned per column (subsequent rows must match). QWP-only.
1542    pub fn column_geohash<'a, N>(
1543        &mut self,
1544        name: N,
1545        bits: u64,
1546        precision_bits: u8,
1547    ) -> crate::Result<&mut Self>
1548    where
1549        N: AsRef<str> + TryInto<ColumnName<'a>>,
1550        Error: From<N::Error>,
1551    {
1552        let _ = (&name, &bits, &precision_bits);
1553        match &mut self.inner {
1554            BufferInner::Ilp(_) => Err(error::fmt!(
1555                InvalidApiCall,
1556                "column_geohash requires a QWP transport (ws:: or udp::)"
1557            )),
1558            #[cfg(any(feature = "_sender-qwp-udp", feature = "_sender-qwp-ws"))]
1559            BufferInner::Qwp(inner) => {
1560                inner.column_geohash(name, bits, precision_bits)?;
1561                Ok(self)
1562            }
1563            #[cfg(feature = "_sender-qwp-ws")]
1564            BufferInner::QwpWs(inner) => {
1565                inner.column_geohash(name, bits, precision_bits)?;
1566                Ok(self)
1567            }
1568        }
1569    }
1570
1571    /// Adds a GEOHASH column if `value` is `Some`. QWP-only.
1572    pub fn column_geohash_opt<'a, N>(
1573        &mut self,
1574        name: N,
1575        value: Option<(u64, u8)>,
1576    ) -> crate::Result<&mut Self>
1577    where
1578        N: AsRef<str> + TryInto<ColumnName<'a>>,
1579        Error: From<N::Error>,
1580    {
1581        if let Some((bits, precision)) = value {
1582            self.column_geohash(name, bits, precision)
1583        } else {
1584            Ok(self)
1585        }
1586    }
1587
1588    #[allow(private_bounds)]
1589    /// Adds an array column to the current row.
1590    ///
1591    /// Arrays require ILP protocol version 2 or later. QWP supports `f64`
1592    /// (DOUBLE_ARRAY, `0x11`) and `i64` (LONG_ARRAY, `0x12`) element types.
1593    /// LONG_ARRAY is part of the QWP v1 spec. Server-side ingest does not
1594    /// currently implement this wire type; batches using it will be rejected
1595    /// with a descriptive error. This may change in future server releases.
1596    pub fn column_arr<'a, N, T, D>(&mut self, name: N, view: &T) -> crate::Result<&mut Self>
1597    where
1598        N: AsRef<str> + TryInto<ColumnName<'a>>,
1599        T: NdArrayView<D>,
1600        D: ArrayElement + ArrayElementSealed,
1601        Error: From<N::Error>,
1602    {
1603        match &mut self.inner {
1604            BufferInner::Ilp(inner) => {
1605                if D::type_tag() != 10 {
1606                    return Err(error::fmt!(
1607                        InvalidApiCall,
1608                        "column_arr with non-f64 element type requires a QWP transport (ws:: or udp::)"
1609                    ));
1610                }
1611                inner.column_arr(name, view)?;
1612            }
1613            #[cfg(any(feature = "_sender-qwp-udp", feature = "_sender-qwp-ws"))]
1614            BufferInner::Qwp(inner) => {
1615                inner.column_arr(name, view)?;
1616            }
1617            #[cfg(feature = "_sender-qwp-ws")]
1618            BufferInner::QwpWs(inner) => {
1619                inner.column_arr(name, view)?;
1620            }
1621        }
1622        Ok(self)
1623    }
1624
1625    /// Adds an array column if `value` is `Some`; otherwise leaves the row unchanged.
1626    #[allow(private_bounds)]
1627    pub fn column_arr_opt<'a, N, T, D>(
1628        &mut self,
1629        name: N,
1630        value: Option<&T>,
1631    ) -> crate::Result<&mut Self>
1632    where
1633        N: AsRef<str> + TryInto<ColumnName<'a>>,
1634        T: NdArrayView<D>,
1635        D: ArrayElement + ArrayElementSealed,
1636        Error: From<N::Error>,
1637    {
1638        if let Some(value) = value {
1639            self.column_arr(name, value)
1640        } else {
1641            Ok(self)
1642        }
1643    }
1644
1645    /// Adds a timestamp column to the current row.
1646    ///
1647    /// Accepts either microsecond or nanosecond timestamps.
1648    #[inline(always)]
1649    pub fn column_ts<'a, N, T>(&mut self, name: N, value: T) -> crate::Result<&mut Self>
1650    where
1651        N: AsRef<str> + TryInto<ColumnName<'a>>,
1652        T: TryInto<Timestamp>,
1653        Error: From<N::Error>,
1654        Error: From<T::Error>,
1655    {
1656        match &mut self.inner {
1657            BufferInner::Ilp(inner) => {
1658                inner.column_ts(name, value)?;
1659            }
1660            #[cfg(any(feature = "_sender-qwp-udp", feature = "_sender-qwp-ws"))]
1661            BufferInner::Qwp(inner) => {
1662                inner.column_ts(name, value)?;
1663            }
1664            #[cfg(feature = "_sender-qwp-ws")]
1665            BufferInner::QwpWs(inner) => {
1666                inner.column_ts(name, value)?;
1667            }
1668        }
1669        Ok(self)
1670    }
1671
1672    /// Adds a timestamp column if `value` is `Some`; otherwise leaves the row unchanged.
1673    pub fn column_ts_opt<'a, N, T>(&mut self, name: N, value: Option<T>) -> crate::Result<&mut Self>
1674    where
1675        N: AsRef<str> + TryInto<ColumnName<'a>>,
1676        T: TryInto<Timestamp>,
1677        Error: From<N::Error>,
1678        Error: From<T::Error>,
1679    {
1680        if let Some(value) = value {
1681            self.column_ts(name, value)
1682        } else {
1683            Ok(self)
1684        }
1685    }
1686
1687    /// Completes the current row with a designated timestamp.
1688    ///
1689    /// After this call you may begin the next row with [`Buffer::table`] or
1690    /// flush the buffer. Accepts either microsecond or nanosecond timestamps.
1691    #[inline(always)]
1692    pub fn at<T>(&mut self, timestamp: T) -> crate::Result<()>
1693    where
1694        T: TryInto<Timestamp>,
1695        Error: From<T::Error>,
1696    {
1697        match &mut self.inner {
1698            BufferInner::Ilp(inner) => inner.at(timestamp),
1699            #[cfg(any(feature = "_sender-qwp-udp", feature = "_sender-qwp-ws"))]
1700            BufferInner::Qwp(inner) => inner.at(timestamp),
1701            #[cfg(feature = "_sender-qwp-ws")]
1702            BufferInner::QwpWs(inner) => inner.at(timestamp),
1703        }
1704    }
1705
1706    /// Completes the current row without a designated timestamp so the server
1707    /// assigns one.
1708    ///
1709    /// This is not equivalent to calling [`Buffer::at`] with the current client
1710    /// time.
1711    #[inline(always)]
1712    pub fn at_now(&mut self) -> crate::Result<()> {
1713        match &mut self.inner {
1714            BufferInner::Ilp(inner) => inner.at_now(),
1715            #[cfg(any(feature = "_sender-qwp-udp", feature = "_sender-qwp-ws"))]
1716            BufferInner::Qwp(inner) => inner.at_now(),
1717            #[cfg(feature = "_sender-qwp-ws")]
1718            BufferInner::QwpWs(inner) => inner.at_now(),
1719        }
1720    }
1721}
1722
1723#[cfg(test)]
1724mod tests {
1725    use super::{Bookmark, Buffer, ColumnName, StoredBookmark, TableName};
1726    use crate::ErrorCode;
1727    use crate::ingress::ProtocolVersion;
1728
1729    #[test]
1730    fn stored_bookmark_reports_missing_bookmark_when_never_captured() {
1731        let bookmark = Bookmark::from_raw(7, 1);
1732        let stored = StoredBookmark::<u8>::new();
1733        let err = stored.restore(7, bookmark).unwrap_err();
1734        assert_eq!(err.code(), ErrorCode::InvalidApiCall);
1735        assert_eq!(err.msg(), "Can't rewind to the bookmark: No bookmark set.");
1736    }
1737
1738    #[test]
1739    fn stored_bookmark_ignores_invalid_zero_origin_clear() {
1740        let mut stored = StoredBookmark::<u8>::new();
1741        let bookmark = stored.capture(7, 42);
1742
1743        stored.clear_if_matches(7, Bookmark::from_raw(0, 0));
1744
1745        assert_eq!(stored.restore(7, bookmark).unwrap(), 42);
1746    }
1747
1748    #[test]
1749    fn buffer_column_i8_rejects_ilp_buffer() {
1750        let mut buf = Buffer::new(ProtocolVersion::V2);
1751        buf.table("trades").unwrap();
1752        let err = buf.column_i8("v", 1).unwrap_err();
1753        assert_eq!(err.code(), ErrorCode::InvalidApiCall);
1754        assert!(
1755            err.msg().contains("column_i8"),
1756            "error message should name column_i8: {}",
1757            err.msg()
1758        );
1759    }
1760
1761    #[test]
1762    fn buffer_column_i16_rejects_ilp_buffer() {
1763        let mut buf = Buffer::new(ProtocolVersion::V2);
1764        buf.table("trades").unwrap();
1765        let err = buf.column_i16("v", 1).unwrap_err();
1766        assert_eq!(err.code(), ErrorCode::InvalidApiCall);
1767        assert!(err.msg().contains("column_i16"), "{}", err.msg());
1768    }
1769
1770    #[test]
1771    fn buffer_column_i32_rejects_ilp_buffer() {
1772        let mut buf = Buffer::new(ProtocolVersion::V2);
1773        buf.table("trades").unwrap();
1774        let err = buf.column_i32("v", 1).unwrap_err();
1775        assert_eq!(err.code(), ErrorCode::InvalidApiCall);
1776        assert!(err.msg().contains("column_i32"), "{}", err.msg());
1777    }
1778
1779    #[test]
1780    fn buffer_column_dec64_rejects_ilp_buffer() {
1781        let mut buf = Buffer::new(ProtocolVersion::V3);
1782        buf.table("trades").unwrap();
1783        let err = buf.column_dec64("v", "1.25").unwrap_err();
1784        assert_eq!(err.code(), ErrorCode::InvalidApiCall);
1785        assert!(err.msg().contains("column_dec64"), "{}", err.msg());
1786    }
1787
1788    #[test]
1789    fn buffer_column_dec128_rejects_ilp_buffer() {
1790        let mut buf = Buffer::new(ProtocolVersion::V3);
1791        buf.table("trades").unwrap();
1792        let err = buf.column_dec128("v", "1.25").unwrap_err();
1793        assert_eq!(err.code(), ErrorCode::InvalidApiCall);
1794        assert!(err.msg().contains("column_dec128"), "{}", err.msg());
1795    }
1796
1797    #[test]
1798    fn buffer_column_uuid_rejects_ilp_buffer() {
1799        let mut buf = Buffer::new(ProtocolVersion::V2);
1800        buf.table("trades").unwrap();
1801        let err = buf.column_uuid("v", 1, 2).unwrap_err();
1802        assert_eq!(err.code(), ErrorCode::InvalidApiCall);
1803        assert!(err.msg().contains("column_uuid"), "{}", err.msg());
1804    }
1805
1806    #[test]
1807    fn buffer_column_long256_rejects_ilp_buffer() {
1808        let mut buf = Buffer::new(ProtocolVersion::V2);
1809        buf.table("trades").unwrap();
1810        let err = buf.column_long256("v", &[0u8; 32]).unwrap_err();
1811        assert_eq!(err.code(), ErrorCode::InvalidApiCall);
1812        assert!(err.msg().contains("column_long256"), "{}", err.msg());
1813    }
1814
1815    #[test]
1816    fn buffer_column_ipv4_rejects_ilp_buffer() {
1817        let mut buf = Buffer::new(ProtocolVersion::V2);
1818        buf.table("trades").unwrap();
1819        let err = buf
1820            .column_ipv4("v", std::net::Ipv4Addr::new(127, 0, 0, 1))
1821            .unwrap_err();
1822        assert_eq!(err.code(), ErrorCode::InvalidApiCall);
1823        assert!(err.msg().contains("column_ipv4"), "{}", err.msg());
1824    }
1825
1826    #[test]
1827    fn buffer_column_date_rejects_ilp_buffer() {
1828        let mut buf = Buffer::new(ProtocolVersion::V2);
1829        buf.table("t").unwrap();
1830        let err = buf.column_date("v", 42).unwrap_err();
1831        assert_eq!(err.code(), ErrorCode::InvalidApiCall);
1832        assert!(err.msg().contains("column_date"), "{}", err.msg());
1833    }
1834
1835    #[test]
1836    fn buffer_column_char_rejects_ilp_buffer() {
1837        let mut buf = Buffer::new(ProtocolVersion::V2);
1838        buf.table("t").unwrap();
1839        let err = buf.column_char("v", 0x0041).unwrap_err();
1840        assert_eq!(err.code(), ErrorCode::InvalidApiCall);
1841        assert!(err.msg().contains("column_char"), "{}", err.msg());
1842    }
1843
1844    #[test]
1845    fn buffer_column_binary_rejects_ilp_buffer() {
1846        let mut buf = Buffer::new(ProtocolVersion::V2);
1847        buf.table("t").unwrap();
1848        let err = buf.column_binary("v", b"abc").unwrap_err();
1849        assert_eq!(err.code(), ErrorCode::InvalidApiCall);
1850        assert!(err.msg().contains("column_binary"), "{}", err.msg());
1851    }
1852
1853    #[test]
1854    fn buffer_column_geohash_rejects_ilp_buffer() {
1855        let mut buf = Buffer::new(ProtocolVersion::V2);
1856        buf.table("t").unwrap();
1857        let err = buf.column_geohash("v", 0xABCD, 16).unwrap_err();
1858        assert_eq!(err.code(), ErrorCode::InvalidApiCall);
1859        assert!(err.msg().contains("column_geohash"), "{}", err.msg());
1860    }
1861
1862    #[test]
1863    fn buffer_column_f32_rejects_ilp_buffer() {
1864        let mut buf = Buffer::new(ProtocolVersion::V2);
1865        buf.table("t").unwrap();
1866        let err = buf.column_f32("v", 1.5_f32).unwrap_err();
1867        assert_eq!(err.code(), ErrorCode::InvalidApiCall);
1868        assert!(err.msg().contains("column_f32"), "{}", err.msg());
1869    }
1870
1871    #[test]
1872    fn name_validation_table_name_uses_byte_offset_for_invalid_char() {
1873        let err = match TableName::new("é?") {
1874            Ok(_) => panic!("expected invalid table name"),
1875            Err(err) => err,
1876        };
1877        assert_eq!(err.code(), ErrorCode::InvalidName);
1878        assert_eq!(
1879            err.msg(),
1880            r#"Bad string "é?": Table names can't contain a '?' character, which was found at byte position 2."#
1881        );
1882    }
1883
1884    #[test]
1885    fn name_validation_table_name_rejects_trailing_dot_at_byte_offset() {
1886        let err = match TableName::new("é.") {
1887            Ok(_) => panic!("expected invalid table name"),
1888            Err(err) => err,
1889        };
1890        assert_eq!(err.code(), ErrorCode::InvalidName);
1891        assert_eq!(
1892            err.msg(),
1893            r#"Bad string "é.": Found invalid dot `.` at position 2."#
1894        );
1895    }
1896
1897    #[test]
1898    fn name_validation_column_name_uses_byte_offset_for_invalid_char() {
1899        let err = match ColumnName::new("é?") {
1900            Ok(_) => panic!("expected invalid column name"),
1901            Err(err) => err,
1902        };
1903        assert_eq!(err.code(), ErrorCode::InvalidName);
1904        assert_eq!(
1905            err.msg(),
1906            r#"Bad string "é?": Column names can't contain a '?' character, which was found at byte position 2."#
1907        );
1908    }
1909}
questdb/ingress/buffer.rs

questdb/ingress/
buffer.rs
