Skip to main content

fsqlite_types/
value.rs

1use std::cell::RefCell;
2use std::cmp::Ordering;
3use std::fmt;
4use std::hash::{Hash, Hasher};
5use std::sync::{Arc, OnceLock};
6
7use memchr::{memchr, memchr2, memmem};
8
9use crate::{StorageClass, StrictColumnType, StrictTypeError, TypeAffinity};
10
11// ============================================================================
12// Thread-Local Value Slab Allocator
13// ============================================================================
14
15/// Maximum number of values to keep in the thread-local pool.
16///
17/// 256 is chosen as a balance: large enough to cover typical row batch sizes,
18/// small enough to avoid unbounded memory retention per thread.
19const VALUE_POOL_CAP: usize = 256;
20
21thread_local! {
22    /// Thread-local pool of reusable `SqliteValue` objects.
23    ///
24    /// During hot-path execution (MakeRecord, Column decode), values are
25    /// acquired from this pool instead of allocating fresh, then returned
26    /// when the register is overwritten or the row changes.
27    static VALUE_POOL: RefCell<Vec<SqliteValue>> = const { RefCell::new(Vec::new()) };
28}
29
30#[cfg(test)]
31#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
32struct ValuePoolStats {
33    slab_alloc_count: usize,
34    slab_return_count: usize,
35    global_alloc_fallback_count: usize,
36    slab_high_water_mark: usize,
37}
38
39#[cfg(test)]
40impl ValuePoolStats {
41    const fn new() -> Self {
42        Self {
43            slab_alloc_count: 0,
44            slab_return_count: 0,
45            global_alloc_fallback_count: 0,
46            slab_high_water_mark: 0,
47        }
48    }
49}
50
51#[cfg(test)]
52thread_local! {
53    static VALUE_POOL_TEST_STATS: RefCell<ValuePoolStats> =
54        const { RefCell::new(ValuePoolStats::new()) };
55}
56
57#[cfg(test)]
58fn reset_value_pool_test_stats() {
59    VALUE_POOL_TEST_STATS.with(|stats| *stats.borrow_mut() = ValuePoolStats::new());
60}
61
62#[cfg(test)]
63fn value_pool_test_stats_snapshot() -> ValuePoolStats {
64    VALUE_POOL_TEST_STATS.with(|stats| *stats.borrow())
65}
66
67#[cfg(test)]
68fn record_value_pool_acquire(hit: bool) {
69    VALUE_POOL_TEST_STATS.with(|stats| {
70        let mut stats = stats.borrow_mut();
71        if hit {
72            stats.slab_alloc_count += 1;
73        } else {
74            stats.global_alloc_fallback_count += 1;
75        }
76    });
77}
78
79#[cfg(test)]
80fn record_value_pool_return(pool_len: usize) {
81    VALUE_POOL_TEST_STATS.with(|stats| {
82        let mut stats = stats.borrow_mut();
83        stats.slab_return_count += 1;
84        stats.slab_high_water_mark = stats.slab_high_water_mark.max(pool_len);
85    });
86}
87
88/// Acquire a value from the thread-local pool, if available.
89///
90/// Returns `Some(value)` if a pooled value was available, `None` otherwise.
91/// The returned value may contain stale data and should be overwritten
92/// with the desired variant before use.
93///
94/// # Example
95/// ```ignore
96/// let value = pool_acquire().unwrap_or(SqliteValue::Null);
97/// // Overwrite with actual data
98/// value = SqliteValue::Integer(42);
99/// ```
100#[inline]
101pub fn pool_acquire() -> Option<SqliteValue> {
102    let value = VALUE_POOL.with(|pool| pool.borrow_mut().pop());
103    #[cfg(test)]
104    record_value_pool_acquire(value.is_some());
105    value
106}
107
108/// Return a value to the thread-local pool for reuse.
109///
110/// Values are only pooled if the pool has capacity (max 256 entries).
111/// Excess values are dropped normally, preventing unbounded memory growth.
112///
113/// For best effect, return values just before they would be dropped,
114/// allowing future `pool_acquire` calls to skip allocation.
115#[inline]
116pub fn pool_return(value: SqliteValue) {
117    VALUE_POOL.with(|pool| {
118        let mut pool = pool.borrow_mut();
119        if pool.len() < VALUE_POOL_CAP {
120            pool.push(value);
121            #[cfg(test)]
122            record_value_pool_return(pool.len());
123        }
124        // If pool is full, value is dropped normally
125    });
126}
127
128/// Return a heap-carrying value to the thread-local pool when preserving its
129/// backing allocation is likely to pay off on the next decode/write.
130#[inline]
131pub fn pool_return_reusable(value: SqliteValue) {
132    if value_preserves_reusable_heap_storage(&value) {
133        pool_return(value);
134    }
135}
136
137/// Clear the thread-local value pool.
138///
139/// Use this to release memory when a thread's workload is complete,
140/// or in test teardown to ensure deterministic behavior.
141#[inline]
142pub fn pool_clear() {
143    VALUE_POOL.with(|pool| pool.borrow_mut().clear());
144}
145
146/// Returns the current number of values in the thread-local pool.
147///
148/// Useful for testing and diagnostics.
149#[inline]
150pub fn pool_len() -> usize {
151    VALUE_POOL.with(|pool| pool.borrow().len())
152}
153
154#[inline]
155fn value_preserves_reusable_heap_storage(value: &SqliteValue) -> bool {
156    match value {
157        SqliteValue::Text(text) => matches!(&text.repr, SmallTextRepr::HeapOwned { .. }),
158        SqliteValue::Blob(bytes) => Arc::strong_count(bytes) == 1,
159        _ => false,
160    }
161}
162
163/// Maximum inline string length for `SmallText`.
164///
165/// Strings up to this length are stored inline (on the stack/in the struct)
166/// without heap allocation. Longer strings fall back to `Arc<str>`.
167///
168/// 23 bytes inline + 1 byte for length/tag = 24 bytes total, which aligns
169/// with common cache line fractions and matches Arc<str>'s pointer size.
170const SMALL_TEXT_INLINE_CAP: usize = 23;
171
172/// A small-string-optimized text value.
173///
174/// Stores strings ≤ 23 bytes inline without heap allocation. Longer strings
175/// stay owned until the first clone, then lazily promote to `Arc<str>` for
176/// shared O(1) cloning. This eliminates both malloc and refcount traffic for
177/// the common single-owner path.
178pub struct SmallText {
179    /// Representation: either inline bytes or a lazily shared heap string.
180    repr: SmallTextRepr,
181}
182
183/// Internal representation for SmallText.
184enum SmallTextRepr {
185    /// Inline storage: length followed by up to 23 UTF-8 bytes.
186    Inline {
187        len: u8,
188        buf: [u8; SMALL_TEXT_INLINE_CAP],
189    },
190    /// Heap storage before the first clone.
191    ///
192    /// The `Arc<str>` is materialized lazily on demand so a single-owner value
193    /// pays no refcount cost until it is actually shared.
194    HeapOwned {
195        text: String,
196        shared: OnceLock<Arc<str>>,
197    },
198    /// Heap storage after the text has been shared.
199    HeapShared(Arc<str>),
200}
201
202impl Clone for SmallText {
203    fn clone(&self) -> Self {
204        Self {
205            repr: self.repr.clone(),
206        }
207    }
208}
209
210impl Clone for SmallTextRepr {
211    fn clone(&self) -> Self {
212        match self {
213            Self::Inline { len, buf } => Self::Inline {
214                len: *len,
215                buf: *buf,
216            },
217            Self::HeapOwned { text, shared } => {
218                let shared = Arc::clone(shared.get_or_init(|| Arc::from(text.as_str())));
219                Self::HeapShared(shared)
220            }
221            Self::HeapShared(text) => Self::HeapShared(Arc::clone(text)),
222        }
223    }
224}
225
226impl SmallText {
227    /// Create a new SmallText from a string slice.
228    #[inline]
229    pub fn new(s: &str) -> Self {
230        if s.len() <= SMALL_TEXT_INLINE_CAP {
231            let mut buf = [0u8; SMALL_TEXT_INLINE_CAP];
232            buf[..s.len()].copy_from_slice(s.as_bytes());
233            Self {
234                repr: SmallTextRepr::Inline {
235                    len: s.len() as u8,
236                    buf,
237                },
238            }
239        } else {
240            Self {
241                repr: SmallTextRepr::HeapOwned {
242                    text: s.to_owned(),
243                    shared: OnceLock::new(),
244                },
245            }
246        }
247    }
248
249    /// Create from an owned String, potentially reusing its allocation.
250    #[inline]
251    pub fn from_string<S>(s: S) -> Self
252    where
253        S: Into<String> + AsRef<str>,
254    {
255        if s.as_ref().len() <= SMALL_TEXT_INLINE_CAP {
256            Self::new(s.as_ref())
257        } else {
258            Self {
259                repr: SmallTextRepr::HeapOwned {
260                    text: s.into(),
261                    shared: OnceLock::new(),
262                },
263            }
264        }
265    }
266
267    /// Create from an Arc<str>, avoiding re-allocation if already heap.
268    #[inline]
269    pub fn from_arc(arc: Arc<str>) -> Self {
270        if arc.len() <= SMALL_TEXT_INLINE_CAP {
271            Self::new(&arc)
272        } else {
273            Self {
274                repr: SmallTextRepr::HeapShared(arc),
275            }
276        }
277    }
278
279    /// Overwrite this string, reusing the existing heap allocation when the
280    /// value is still single-owner.
281    #[inline]
282    pub fn overwrite(&mut self, s: &str) {
283        if s.len() <= SMALL_TEXT_INLINE_CAP {
284            let mut buf = [0u8; SMALL_TEXT_INLINE_CAP];
285            buf[..s.len()].copy_from_slice(s.as_bytes());
286            self.repr = SmallTextRepr::Inline {
287                len: s.len() as u8,
288                buf,
289            };
290            return;
291        }
292
293        match &mut self.repr {
294            SmallTextRepr::HeapOwned { text, shared } => {
295                text.clear();
296                text.push_str(s);
297                if shared.get().is_some() {
298                    *shared = OnceLock::new();
299                }
300            }
301            _ => {
302                self.repr = SmallTextRepr::HeapOwned {
303                    text: s.to_owned(),
304                    shared: OnceLock::new(),
305                };
306            }
307        }
308    }
309
310    /// Get the string as a slice.
311    ///
312    /// OPT-UTF8: the inline buffer is always valid UTF-8 by construction (see
313    /// the constructors and [`Self::overwrite`]), but because
314    /// `forbid(unsafe_code)` prevents `from_utf8_unchecked` we must run a
315    /// validator. `simdutf8::basic::from_utf8` is a drop-in for
316    /// `std::str::from_utf8` that uses runtime-dispatched SIMD and is
317    /// ~3-10x faster on the ASCII-dominant TEXT payloads that make up the
318    /// majority of real SQL workloads.
319    #[inline]
320    pub fn as_str(&self) -> &str {
321        match &self.repr {
322            SmallTextRepr::Inline { len, buf } => simdutf8::basic::from_utf8(&buf[..*len as usize])
323                .expect("SmallText inline representation must always contain valid UTF-8"),
324            SmallTextRepr::HeapOwned { text, .. } => text.as_str(),
325            SmallTextRepr::HeapShared(text) => text,
326        }
327    }
328
329    /// Get the raw bytes of this text value without going through `&str`.
330    ///
331    /// Unlike [`Self::as_str`] (which revalidates the inline buffer via
332    /// `from_utf8`), this directly returns the stored bytes. The returned
333    /// slice is guaranteed to be valid UTF-8 by construction: every code path
334    /// that writes to a `SmallText` (`new`, `from_string`, `from_arc`,
335    /// `overwrite`) sources its bytes from a `&str` or `Arc<str>`.
336    ///
337    /// This is useful on hot paths where a byte-wise equality check is the
338    /// only operation performed — for example, the record-decode fast path
339    /// that reuses an existing slot when incoming bytes match what is already
340    /// there. Skipping the internal `from_utf8` of `as_str` measurably
341    /// reduces per-column decode cost on INSERT/SELECT workloads.
342    #[inline]
343    #[must_use]
344    pub fn as_bytes_direct(&self) -> &[u8] {
345        match &self.repr {
346            SmallTextRepr::Inline { len, buf } => &buf[..*len as usize],
347            SmallTextRepr::HeapOwned { text, .. } => text.as_bytes(),
348            SmallTextRepr::HeapShared(text) => text.as_bytes(),
349        }
350    }
351
352    /// Get the length in bytes.
353    #[inline]
354    pub fn len(&self) -> usize {
355        match &self.repr {
356            SmallTextRepr::Inline { len, .. } => *len as usize,
357            SmallTextRepr::HeapOwned { text, .. } => text.len(),
358            SmallTextRepr::HeapShared(text) => text.len(),
359        }
360    }
361
362    /// Check if empty.
363    #[inline]
364    pub fn is_empty(&self) -> bool {
365        self.len() == 0
366    }
367
368    /// Check if stored inline (no heap allocation).
369    #[inline]
370    pub fn is_inline(&self) -> bool {
371        matches!(&self.repr, SmallTextRepr::Inline { .. })
372    }
373
374    /// Convert to Arc<str>, potentially allocating if currently inline.
375    #[inline]
376    pub fn into_arc(self) -> Arc<str> {
377        match self.repr {
378            SmallTextRepr::Inline { len, buf } => {
379                // See `as_str` for the simdutf8 rationale.
380                let s = simdutf8::basic::from_utf8(&buf[..len as usize])
381                    .expect("SmallText inline representation must always contain valid UTF-8");
382                Arc::from(s)
383            }
384            SmallTextRepr::HeapOwned { text, shared } => shared
385                .into_inner()
386                .unwrap_or_else(|| Arc::<str>::from(text)),
387            SmallTextRepr::HeapShared(text) => text,
388        }
389    }
390}
391
392impl Default for SmallText {
393    #[inline]
394    fn default() -> Self {
395        Self::new("")
396    }
397}
398
399impl fmt::Debug for SmallText {
400    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
401        fmt::Debug::fmt(self.as_str(), f)
402    }
403}
404
405impl fmt::Display for SmallText {
406    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
407        fmt::Display::fmt(self.as_str(), f)
408    }
409}
410
411impl PartialEq for SmallText {
412    #[inline]
413    fn eq(&self, other: &Self) -> bool {
414        self.as_str() == other.as_str()
415    }
416}
417
418impl Eq for SmallText {}
419
420impl PartialOrd for SmallText {
421    #[inline]
422    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
423        Some(self.cmp(other))
424    }
425}
426
427impl Ord for SmallText {
428    #[inline]
429    fn cmp(&self, other: &Self) -> Ordering {
430        self.as_str().cmp(other.as_str())
431    }
432}
433
434impl Hash for SmallText {
435    #[inline]
436    fn hash<H: Hasher>(&self, state: &mut H) {
437        self.as_str().hash(state);
438    }
439}
440
441impl From<&str> for SmallText {
442    #[inline]
443    fn from(s: &str) -> Self {
444        Self::new(s)
445    }
446}
447
448impl From<String> for SmallText {
449    #[inline]
450    fn from(s: String) -> Self {
451        Self::from_string(s)
452    }
453}
454
455impl From<Arc<str>> for SmallText {
456    #[inline]
457    fn from(arc: Arc<str>) -> Self {
458        Self::from_arc(arc)
459    }
460}
461
462impl AsRef<str> for SmallText {
463    #[inline]
464    fn as_ref(&self) -> &str {
465        self.as_str()
466    }
467}
468
469impl std::ops::Deref for SmallText {
470    type Target = str;
471
472    #[inline]
473    fn deref(&self) -> &Self::Target {
474        self.as_str()
475    }
476}
477
478impl std::borrow::Borrow<str> for SmallText {
479    #[inline]
480    fn borrow(&self) -> &str {
481        self.as_str()
482    }
483}
484
485// Serde implementations for SmallText
486impl serde::Serialize for SmallText {
487    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
488    where
489        S: serde::Serializer,
490    {
491        serializer.serialize_str(self.as_str())
492    }
493}
494
495impl<'de> serde::Deserialize<'de> for SmallText {
496    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
497    where
498        D: serde::Deserializer<'de>,
499    {
500        let s = String::deserialize(deserializer)?;
501        Ok(Self::from_string(s))
502    }
503}
504
505/// Scan the longest SQLite numeric prefix from a byte slice.
506///
507/// Recognises `[+-]? [0-9]* ('.' [0-9]*)? ([eE] [+-]? [0-9]+)?`.
508/// Returns the byte offset where the prefix ends, or 0 if no numeric prefix
509/// is present.
510fn scan_numeric_prefix(bytes: &[u8]) -> usize {
511    if bytes.is_empty() {
512        return 0;
513    }
514
515    let mut i = 0usize;
516    if bytes[i] == b'+' || bytes[i] == b'-' {
517        i += 1;
518    }
519
520    let mut has_digit = false;
521    while i < bytes.len() && bytes[i].is_ascii_digit() {
522        has_digit = true;
523        i += 1;
524    }
525
526    if i < bytes.len() && bytes[i] == b'.' {
527        i += 1;
528        while i < bytes.len() && bytes[i].is_ascii_digit() {
529            has_digit = true;
530            i += 1;
531        }
532    }
533
534    if !has_digit {
535        return 0;
536    }
537
538    if i < bytes.len() && (bytes[i] == b'e' || bytes[i] == b'E') {
539        let exp_start = i;
540        i += 1;
541        if i < bytes.len() && (bytes[i] == b'+' || bytes[i] == b'-') {
542            i += 1;
543        }
544        if i < bytes.len() && bytes[i].is_ascii_digit() {
545            while i < bytes.len() && bytes[i].is_ascii_digit() {
546                i += 1;
547            }
548        } else {
549            i = exp_start;
550        }
551    }
552
553    i
554}
555
556/// Parse the longest numeric prefix of `b` as an integer.
557#[allow(clippy::cast_possible_truncation)]
558fn parse_integer_prefix_bytes(b: &[u8]) -> i64 {
559    let mut start = 0;
560    while start < b.len() && b[start].is_ascii_whitespace() {
561        start += 1;
562    }
563    let trimmed = &b[start..];
564    let end = scan_numeric_prefix(trimmed);
565    if end == 0 {
566        return 0;
567    }
568    // SAFETY: scan_numeric_prefix only advances over ASCII bytes (digits, +, -, ., e, E),
569    // so the slice is always valid UTF-8.
570    let s = std::str::from_utf8(&trimmed[..end]).unwrap_or("");
571    let f = s.parse::<f64>().unwrap_or(0.0);
572    #[allow(clippy::manual_clamp)]
573    if f >= i64::MAX as f64 {
574        i64::MAX
575    } else if f <= i64::MIN as f64 {
576        i64::MIN
577    } else {
578        f as i64
579    }
580}
581
582/// Parse the longest numeric prefix of `s` as an integer.
583#[allow(clippy::cast_possible_truncation)]
584fn parse_integer_prefix(s: &str) -> i64 {
585    parse_integer_prefix_bytes(s.as_bytes())
586}
587
588/// Parse the longest numeric prefix of `b` as a float.
589fn parse_float_prefix_bytes(b: &[u8]) -> f64 {
590    let mut start = 0;
591    while start < b.len() && b[start].is_ascii_whitespace() {
592        start += 1;
593    }
594    let trimmed = &b[start..];
595    let end = scan_numeric_prefix(trimmed);
596    if end == 0 {
597        return 0.0;
598    }
599    // SAFETY: scan_numeric_prefix only advances over ASCII bytes (digits, +, -, ., e, E),
600    // so the slice is always valid UTF-8.
601    let s = std::str::from_utf8(&trimmed[..end]).unwrap_or("");
602    s.parse::<f64>().unwrap_or(0.0)
603}
604
605/// Parse the longest numeric prefix of `s` as a float.
606fn parse_float_prefix(s: &str) -> f64 {
607    parse_float_prefix_bytes(s.as_bytes())
608}
609
610fn trim_sqlite_ascii_whitespace(s: &str) -> &str {
611    s.trim_matches(|ch: char| ch.is_ascii_whitespace())
612}
613
614fn cast_text_prefix_to_numeric(s: &str) -> SqliteValue {
615    let trimmed = trim_sqlite_ascii_whitespace(s);
616    let end = scan_numeric_prefix(trimmed.as_bytes());
617    if end == 0 {
618        return SqliteValue::Integer(0);
619    }
620
621    let prefix = &trimmed[..end];
622    let is_integer_syntax = !prefix
623        .as_bytes()
624        .iter()
625        .any(|byte| matches!(*byte, b'.' | b'e' | b'E'));
626
627    if is_integer_syntax && let Ok(value) = prefix.parse::<i64>() {
628        return SqliteValue::Integer(value);
629    }
630
631    if let Ok(value) = prefix.parse::<f64>() {
632        if value.is_finite()
633            && (-9_223_372_036_854_775_808.0..9_223_372_036_854_775_808.0).contains(&value)
634        {
635            #[allow(clippy::cast_possible_truncation, clippy::cast_precision_loss)]
636            let truncated = value as i64;
637            #[allow(clippy::float_cmp, clippy::cast_precision_loss)]
638            if truncated as f64 == value {
639                return SqliteValue::Integer(truncated);
640            }
641        }
642        return SqliteValue::Float(value);
643    }
644
645    SqliteValue::Integer(0)
646}
647
648/// A dynamically-typed SQLite value.
649///
650/// Corresponds to C SQLite's `sqlite3_value` / `Mem` type. SQLite has five
651/// fundamental storage classes: NULL, INTEGER, REAL, TEXT, and BLOB.
652#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
653pub enum SqliteValue {
654    /// A NULL value.
655    Null,
656    /// A signed 64-bit integer.
657    Integer(i64),
658    /// A 64-bit IEEE floating point number.
659    Float(f64),
660    /// A UTF-8 text string.
661    ///
662    /// Uses `SmallText` for small-string optimization: strings ≤ 23 bytes
663    /// are stored inline without heap allocation. Longer strings use
664    /// `Arc<str>` internally for O(1) cloning.
665    Text(SmallText),
666    /// A binary large object.
667    ///
668    /// Uses `Arc<[u8]>` for the same O(1)-clone benefit as `Text`.
669    Blob(Arc<[u8]>),
670}
671
672impl SqliteValue {
673    /// Returns the type affinity that best describes this value.
674    pub const fn affinity(&self) -> TypeAffinity {
675        match self {
676            Self::Null | Self::Blob(_) => TypeAffinity::Blob,
677            Self::Integer(_) => TypeAffinity::Integer,
678            Self::Float(_) => TypeAffinity::Real,
679            Self::Text(_) => TypeAffinity::Text,
680        }
681    }
682
683    /// Returns the storage class of this value.
684    pub const fn storage_class(&self) -> StorageClass {
685        match self {
686            Self::Null => StorageClass::Null,
687            Self::Integer(_) => StorageClass::Integer,
688            Self::Float(_) => StorageClass::Real,
689            Self::Text(_) => StorageClass::Text,
690            Self::Blob(_) => StorageClass::Blob,
691        }
692    }
693
694    /// Apply column type affinity coercion (advisory mode).
695    ///
696    /// In non-STRICT tables, affinity is advisory: values are coerced when
697    /// possible but never rejected. Follows SQLite §3.4 rules from
698    /// <https://www.sqlite.org/datatype3.html#type_affinity_of_a_column>.
699    ///
700    /// - TEXT affinity: numeric values converted to text before storing.
701    /// - NUMERIC affinity: text parsed as integer/real if well-formed; exact-integer reals become integer.
702    /// - INTEGER affinity: identical to NUMERIC for storage/comparison coercion (differ only in CAST).
703    /// - REAL affinity: like NUMERIC, plus integers forced to float.
704    /// - BLOB affinity: no conversion.
705    #[must_use]
706    #[allow(
707        clippy::cast_possible_truncation,
708        clippy::cast_precision_loss,
709        clippy::float_cmp
710    )]
711    pub fn apply_affinity(self, affinity: TypeAffinity) -> Self {
712        match affinity {
713            TypeAffinity::Blob => self,
714            TypeAffinity::Text => match self {
715                Self::Null | Self::Text(_) | Self::Blob(_) => self,
716                Self::Integer(_) | Self::Float(_) => {
717                    let t = self.to_text();
718                    Self::Text(SmallText::from_string(t))
719                }
720            },
721            TypeAffinity::Numeric | TypeAffinity::Integer => match &self {
722                Self::Text(s) => try_coerce_text_to_numeric(s.as_str()).unwrap_or(self),
723                Self::Float(f) => {
724                    if *f >= -9_223_372_036_854_775_808.0 && *f < 9_223_372_036_854_775_808.0 {
725                        let i = *f as i64;
726                        if (i as f64) == *f {
727                            return Self::Integer(i);
728                        }
729                    }
730                    self
731                }
732                _ => self,
733            },
734            TypeAffinity::Real => match &self {
735                Self::Text(s) => try_coerce_text_to_numeric(s.as_str())
736                    .map(|v| match v {
737                        Self::Integer(i) => Self::Float(i as f64),
738                        other => other,
739                    })
740                    .unwrap_or(self),
741                Self::Integer(i) => Self::Float(*i as f64),
742                _ => self,
743            },
744        }
745    }
746
747    /// Validate a value against a STRICT table column type.
748    ///
749    /// NULL is always accepted (nullability is enforced separately via NOT NULL).
750    /// Returns `Ok(value)` with possible implicit coercion (REAL columns accept
751    /// integers, converting them to float), or `Err` if the storage class is
752    /// incompatible.
753    #[allow(clippy::cast_precision_loss)]
754    pub fn validate_strict(self, col_type: StrictColumnType) -> Result<Self, StrictTypeError> {
755        if matches!(self, Self::Null) {
756            return Ok(self);
757        }
758        match col_type {
759            StrictColumnType::Any => Ok(self),
760            StrictColumnType::Integer => match self {
761                Self::Integer(_) => Ok(self),
762                other => Err(StrictTypeError {
763                    expected: col_type,
764                    actual: other.storage_class(),
765                }),
766            },
767            StrictColumnType::Real => match self {
768                Self::Float(_) => Ok(self),
769                Self::Integer(i) => Ok(Self::Float(i as f64)),
770                other => Err(StrictTypeError {
771                    expected: col_type,
772                    actual: other.storage_class(),
773                }),
774            },
775            StrictColumnType::Text => match self {
776                Self::Text(_) => Ok(self),
777                other => Err(StrictTypeError {
778                    expected: col_type,
779                    actual: other.storage_class(),
780                }),
781            },
782            StrictColumnType::Blob => match self {
783                Self::Blob(_) => Ok(self),
784                other => Err(StrictTypeError {
785                    expected: col_type,
786                    actual: other.storage_class(),
787                }),
788            },
789        }
790    }
791
792    /// Returns true if this is a NULL value.
793    #[inline(always)]
794    #[allow(clippy::inline_always)]
795    pub const fn is_null(&self) -> bool {
796        matches!(self, Self::Null)
797    }
798
799    /// Try to extract an integer value.
800    #[inline]
801    pub const fn as_integer(&self) -> Option<i64> {
802        match self {
803            Self::Integer(i) => Some(*i),
804            _ => None,
805        }
806    }
807
808    /// Try to extract a float value.
809    #[inline]
810    pub fn as_float(&self) -> Option<f64> {
811        match self {
812            Self::Float(f) => Some(*f),
813            _ => None,
814        }
815    }
816
817    /// Try to extract a text reference.
818    #[inline]
819    pub fn as_text(&self) -> Option<&str> {
820        match self {
821            Self::Text(s) => Some(s),
822            _ => None,
823        }
824    }
825
826    /// Try to extract a blob reference.
827    #[inline]
828    pub fn as_blob(&self) -> Option<&[u8]> {
829        match self {
830            Self::Blob(b) => Some(b),
831            _ => None,
832        }
833    }
834
835    /// Convert to an integer following SQLite's type coercion rules.
836    ///
837    /// - NULL -> 0
838    /// - Integer -> itself
839    /// - Float -> truncated to i64
840    /// - Text -> attempt to parse, 0 on failure
841    /// - Blob -> parse bytes as numeric string, 0 on failure
842    #[inline(always)]
843    #[allow(clippy::inline_always)]
844    #[allow(clippy::cast_possible_truncation)]
845    pub fn to_integer(&self) -> i64 {
846        match self {
847            Self::Null => 0,
848            Self::Integer(i) => *i,
849            Self::Float(f) => *f as i64,
850            Self::Text(s) => parse_integer_prefix(s),
851            Self::Blob(b) => parse_integer_prefix_bytes(b),
852        }
853    }
854
855    /// Convert to a float following SQLite's type coercion rules.
856    ///
857    /// - NULL -> 0.0
858    /// - Integer -> as f64
859    /// - Float -> itself
860    /// - Text -> attempt to parse, 0.0 on failure
861    /// - Blob -> parse bytes as numeric string, 0.0 on failure
862    #[inline(always)]
863    #[allow(clippy::inline_always)]
864    #[allow(clippy::cast_precision_loss)]
865    pub fn to_float(&self) -> f64 {
866        match self {
867            Self::Null => 0.0,
868            Self::Integer(i) => *i as f64,
869            Self::Float(f) => *f,
870            Self::Text(s) => parse_float_prefix(s),
871            Self::Blob(b) => parse_float_prefix_bytes(b),
872        }
873    }
874
875    /// Coerce a value for SQLite `sum()` accumulation.
876    ///
877    /// `sum()` keeps integer accumulation only for INTEGER values and text that
878    /// is entirely a signed 64-bit integer literal after trimming SQLite ASCII
879    /// whitespace. Other text and all blobs participate through the REAL
880    /// accumulator, even when their numeric prefix is integer-looking.
881    #[must_use]
882    pub fn to_sum_numeric_value(&self) -> Self {
883        match self {
884            Self::Null => Self::Null,
885            Self::Integer(i) => Self::Integer(*i),
886            Self::Float(f) => Self::Float(*f),
887            Self::Text(s) => {
888                let trimmed = trim_sqlite_ascii_whitespace(s.as_str());
889                if let Ok(integer) = trimmed.parse::<i64>() {
890                    Self::Integer(integer)
891                } else {
892                    Self::Float(parse_float_prefix(s))
893                }
894            }
895            Self::Blob(b) => Self::Float(parse_float_prefix_bytes(b)),
896        }
897    }
898
899    /// Borrow the inner text string without allocating.
900    ///
901    /// Returns `Some(&str)` for `Text` values, `None` otherwise.
902    /// Use this in comparisons, LIKE patterns, and WHERE clause
903    /// evaluation to avoid the clone that `to_text()` incurs.
904    #[inline]
905    #[must_use]
906    pub fn as_text_str(&self) -> Option<&str> {
907        match self {
908            Self::Text(s) => Some(s),
909            _ => None,
910        }
911    }
912
913    /// Borrow the inner blob bytes without allocating.
914    #[inline]
915    #[must_use]
916    pub fn as_blob_bytes(&self) -> Option<&[u8]> {
917        match self {
918            Self::Blob(b) => Some(b),
919            _ => None,
920        }
921    }
922
923    /// Convert to text following SQLite's CAST(x AS TEXT) coercion rules.
924    ///
925    /// For blobs, this interprets the raw bytes as UTF-8 (with lossy
926    /// replacement for invalid sequences), matching C SQLite behavior.
927    /// For the SQL-literal hex format (`X'...'`), use the `Display` impl.
928    pub fn to_text(&self) -> String {
929        match self {
930            Self::Null => String::new(),
931            Self::Integer(i) => i.to_string(),
932            Self::Float(f) => format_sqlite_float(*f),
933            Self::Text(s) => s.to_string(),
934            Self::Blob(b) => String::from_utf8_lossy(b).into_owned(),
935        }
936    }
937
938    /// Convert to NUMERIC using SQLite CAST semantics rather than affinity.
939    ///
940    /// Unlike NUMERIC affinity, CAST always produces a numeric storage class for
941    /// text/blob input, using the longest leading numeric prefix or `0` when no
942    /// numeric prefix exists.
943    #[must_use]
944    pub fn cast_to_numeric(&self) -> Self {
945        match self {
946            Self::Null => Self::Null,
947            Self::Integer(i) => Self::Integer(*i),
948            Self::Float(f) => Self::Float(*f),
949            Self::Text(s) => cast_text_prefix_to_numeric(s),
950            Self::Blob(b) => cast_text_prefix_to_numeric(&String::from_utf8_lossy(b)),
951        }
952    }
953
954    /// Returns the SQLite `typeof()` string for this value.
955    ///
956    /// Matches C sqlite3: "null", "integer", "real", "text", or "blob".
957    pub const fn typeof_str(&self) -> &'static str {
958        match self {
959            Self::Null => "null",
960            Self::Integer(_) => "integer",
961            Self::Float(_) => "real",
962            Self::Text(_) => "text",
963            Self::Blob(_) => "blob",
964        }
965    }
966
967    /// Returns the SQLite `length()` result for this value.
968    ///
969    /// - NULL → NULL (represented as None)
970    /// - TEXT → character count
971    /// - BLOB → byte count
972    /// - INTEGER/REAL → character count of text representation
973    pub fn sql_length(&self) -> Option<i64> {
974        match self {
975            Self::Null => None,
976            Self::Text(s) => Some(i64::try_from(s.chars().count()).unwrap_or(i64::MAX)),
977            Self::Blob(b) => Some(i64::try_from(b.len()).unwrap_or(i64::MAX)),
978            Self::Integer(_) | Self::Float(_) => {
979                let t = self.to_text();
980                Some(i64::try_from(t.chars().count()).unwrap_or(i64::MAX))
981            }
982        }
983    }
984
985    /// Check equality for UNIQUE constraint purposes.
986    ///
987    /// In SQLite, NULL != NULL for uniqueness: if either value is NULL, the
988    /// result is `false` (they are never considered duplicates). Non-NULL values
989    /// compare by storage class ordering (same as `PartialEq`).
990    pub fn unique_eq(&self, other: &Self) -> bool {
991        if self.is_null() || other.is_null() {
992            return false;
993        }
994        matches!(self.partial_cmp(other), Some(Ordering::Equal))
995    }
996
997    /// Convert a floating-point arithmetic result into a SQLite value.
998    ///
999    /// SQLite does not surface NaN; NaN is normalized to NULL while ±Inf remain REAL.
1000    fn float_result_or_null(result: f64) -> Self {
1001        if result.is_nan() {
1002            Self::Null
1003        } else {
1004            Self::Float(result)
1005        }
1006    }
1007
1008    /// Mirrors C SQLite's `numericType()` (SQLite VDBE:496): returns true if this
1009    /// value should be treated as an integer for arithmetic purposes.
1010    ///
1011    /// Integer values are obviously integer-typed. Text/Blob values that parse
1012    /// as i64 are also integer-typed. Float and Null are not.
1013    #[inline]
1014    pub fn is_integer_numeric_type(&self) -> bool {
1015        fn text_is_integer_numeric_type(s: &str) -> bool {
1016            let trimmed = s.trim_start();
1017            let end = scan_numeric_prefix(trimmed.as_bytes());
1018            end > 0
1019                && !trimmed.as_bytes()[..end]
1020                    .iter()
1021                    .any(|byte| matches!(*byte, b'.' | b'e' | b'E'))
1022        }
1023
1024        match self {
1025            Self::Integer(_) => true,
1026            Self::Float(_) | Self::Null => false,
1027            Self::Text(s) => text_is_integer_numeric_type(s),
1028            Self::Blob(b) => text_is_integer_numeric_type(&String::from_utf8_lossy(b)),
1029        }
1030    }
1031
1032    /// Returns true if this value should be treated as a float for arithmetic.
1033    /// A value is "float numeric type" only if it has a numeric prefix
1034    /// containing '.', 'e', or 'E'. Non-numeric text/blob is NOT float
1035    /// (it coerces to integer 0 in C SQLite's OP_Add/Sub/Mul).
1036    #[inline]
1037    fn is_float_numeric_type(&self) -> bool {
1038        fn text_is_float(s: &str) -> bool {
1039            let trimmed = s.trim_start();
1040            let end = scan_numeric_prefix(trimmed.as_bytes());
1041            end > 0
1042                && trimmed.as_bytes()[..end]
1043                    .iter()
1044                    .any(|byte| matches!(*byte, b'.' | b'e' | b'E'))
1045        }
1046        match self {
1047            Self::Float(_) => true,
1048            Self::Integer(_) | Self::Null => false,
1049            Self::Text(s) => text_is_float(s),
1050            Self::Blob(b) => text_is_float(&String::from_utf8_lossy(b)),
1051        }
1052    }
1053
1054    /// Add two values following SQLite's overflow semantics.
1055    ///
1056    /// - Integer + Integer: checked add; overflows promote to REAL.
1057    /// - Any REAL operand: float addition.
1058    /// - NULL propagates (NULL + x = NULL).
1059    /// - Text/Blob coerced via `numericType()`: if both parse as integer,
1060    ///   integer math is used (SQLite VDBE:1932-1934).
1061    #[inline(always)]
1062    #[allow(clippy::inline_always)]
1063    #[must_use]
1064    #[allow(clippy::cast_precision_loss)]
1065    pub fn sql_add(&self, other: &Self) -> Self {
1066        match (self, other) {
1067            (Self::Null, _) | (_, Self::Null) => Self::Null,
1068            (Self::Integer(a), Self::Integer(b)) => match a.checked_add(*b) {
1069                Some(result) => Self::Integer(result),
1070                None => Self::float_result_or_null(*a as f64 + *b as f64),
1071            },
1072            // If neither operand is a float-type (i.e. both are integer,
1073            // integer-text, or non-numeric text/blob), use integer arithmetic.
1074            // Non-numeric text like "hello" coerces to integer 0, not float 0.0.
1075            _ if !self.is_float_numeric_type() && !other.is_float_numeric_type() => {
1076                let a = self.to_integer();
1077                let b = other.to_integer();
1078                match a.checked_add(b) {
1079                    Some(result) => Self::Integer(result),
1080                    None => Self::float_result_or_null(a as f64 + b as f64),
1081                }
1082            }
1083            _ => Self::float_result_or_null(self.to_float() + other.to_float()),
1084        }
1085    }
1086
1087    /// Subtract two values following SQLite's overflow semantics.
1088    ///
1089    /// Integer - Integer with overflow promotes to REAL.
1090    #[inline(always)]
1091    #[allow(clippy::inline_always)]
1092    #[must_use]
1093    #[allow(clippy::cast_precision_loss)]
1094    pub fn sql_sub(&self, other: &Self) -> Self {
1095        match (self, other) {
1096            (Self::Null, _) | (_, Self::Null) => Self::Null,
1097            (Self::Integer(a), Self::Integer(b)) => match a.checked_sub(*b) {
1098                Some(result) => Self::Integer(result),
1099                None => Self::float_result_or_null(*a as f64 - *b as f64),
1100            },
1101            _ if !self.is_float_numeric_type() && !other.is_float_numeric_type() => {
1102                let a = self.to_integer();
1103                let b = other.to_integer();
1104                match a.checked_sub(b) {
1105                    Some(result) => Self::Integer(result),
1106                    None => Self::float_result_or_null(a as f64 - b as f64),
1107                }
1108            }
1109            _ => Self::float_result_or_null(self.to_float() - other.to_float()),
1110        }
1111    }
1112
1113    /// Multiply two values following SQLite's overflow semantics.
1114    ///
1115    /// Integer * Integer with overflow promotes to REAL.
1116    #[inline(always)]
1117    #[allow(clippy::inline_always)]
1118    #[must_use]
1119    #[allow(clippy::cast_precision_loss)]
1120    pub fn sql_mul(&self, other: &Self) -> Self {
1121        match (self, other) {
1122            (Self::Null, _) | (_, Self::Null) => Self::Null,
1123            (Self::Integer(a), Self::Integer(b)) => match a.checked_mul(*b) {
1124                Some(result) => Self::Integer(result),
1125                None => Self::float_result_or_null(*a as f64 * *b as f64),
1126            },
1127            (Self::Integer(a), Self::Float(b)) => Self::float_result_or_null(*a as f64 * *b),
1128            (Self::Float(a), Self::Integer(b)) => Self::float_result_or_null(*a * *b as f64),
1129            (Self::Float(a), Self::Float(b)) => Self::float_result_or_null(*a * *b),
1130            _ if !self.is_float_numeric_type() && !other.is_float_numeric_type() => {
1131                let a = self.to_integer();
1132                let b = other.to_integer();
1133                match a.checked_mul(b) {
1134                    Some(result) => Self::Integer(result),
1135                    None => Self::float_result_or_null(a as f64 * b as f64),
1136                }
1137            }
1138            _ => Self::float_result_or_null(self.to_float() * other.to_float()),
1139        }
1140    }
1141
1142    /// The sort order key for NULL values (SQLite sorts NULLs first).
1143    const fn sort_class(&self) -> u8 {
1144        match self {
1145            Self::Null => 0,
1146            Self::Integer(_) | Self::Float(_) => 1,
1147            Self::Text(_) => 2,
1148            Self::Blob(_) => 3,
1149        }
1150    }
1151}
1152
1153/// Check if two composite UNIQUE keys are duplicates (SQLite NULL semantics).
1154///
1155/// Returns `true` only if ALL corresponding components are non-NULL and equal.
1156/// If ANY component in either key is NULL, the keys are NOT duplicates (per
1157/// SQLite's NULL != NULL rule for UNIQUE constraints).
1158///
1159/// Both slices must have the same length (panics otherwise).
1160pub fn unique_key_duplicates(a: &[SqliteValue], b: &[SqliteValue]) -> bool {
1161    assert_eq!(a.len(), b.len(), "UNIQUE key columns must match");
1162    a.iter().zip(b.iter()).all(|(va, vb)| va.unique_eq(vb))
1163}
1164
1165/// Match a string against a SQL LIKE pattern with SQLite semantics.
1166///
1167/// - `%` matches zero or more characters.
1168/// - `_` matches exactly one character.
1169/// - Case-insensitive for ASCII A-Z only (no Unicode case folding without ICU).
1170/// - `escape` optionally specifies the escape character for literal `%`/`_`.
1171pub fn sql_like(pattern: &str, text: &str, escape: Option<char>) -> bool {
1172    sql_like_cased(pattern, text, escape, false)
1173}
1174
1175/// Match a string against a SQL LIKE pattern, honoring the connection's
1176/// `PRAGMA case_sensitive_like` setting.
1177///
1178/// When `case_sensitive` is `false` (the default) this is identical to
1179/// [`sql_like`]: ASCII case is folded. When `case_sensitive` is `true`
1180/// (`PRAGMA case_sensitive_like = ON`) the literal portions of the pattern are
1181/// matched byte-exact — wildcards (`%`, `_`) still behave the same.
1182#[must_use]
1183pub fn sql_like_cased(
1184    pattern: &str,
1185    text: &str,
1186    escape: Option<char>,
1187    case_sensitive: bool,
1188) -> bool {
1189    if let Some((kind, literal)) = classify_sql_like_fast_path(pattern, escape) {
1190        return sql_like_fast_path_matches_cased(kind, literal, text, case_sensitive);
1191    }
1192
1193    sql_like_inner(
1194        &pattern.chars().collect::<Vec<_>>(),
1195        &text.chars().collect::<Vec<_>>(),
1196        escape,
1197        0,
1198        0,
1199        case_sensitive,
1200    )
1201}
1202
1203#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1204pub enum SqlLikeFastPathKind {
1205    MatchAll,
1206    Exact,
1207    Prefix,
1208    Suffix,
1209    Contains,
1210}
1211
1212impl SqlLikeFastPathKind {
1213    #[must_use]
1214    pub const fn opcode_tag(self) -> i32 {
1215        match self {
1216            Self::MatchAll => 0,
1217            Self::Exact => 1,
1218            Self::Prefix => 2,
1219            Self::Suffix => 3,
1220            Self::Contains => 4,
1221        }
1222    }
1223
1224    #[must_use]
1225    pub const fn from_opcode_tag(tag: i32) -> Option<Self> {
1226        match tag {
1227            0 => Some(Self::MatchAll),
1228            1 => Some(Self::Exact),
1229            2 => Some(Self::Prefix),
1230            3 => Some(Self::Suffix),
1231            4 => Some(Self::Contains),
1232            _ => None,
1233        }
1234    }
1235}
1236
1237#[must_use]
1238pub fn sql_like_fast_path_matches(kind: SqlLikeFastPathKind, literal: &str, text: &str) -> bool {
1239    sql_like_fast_path_matches_cased(kind, literal, text, false)
1240}
1241
1242/// Like [`sql_like_fast_path_matches`] but honoring `case_sensitive`.
1243///
1244/// When `case_sensitive` is `true` the literal is compared byte-exact
1245/// (`PRAGMA case_sensitive_like = ON`); otherwise ASCII case is folded.
1246#[must_use]
1247pub fn sql_like_fast_path_matches_cased(
1248    kind: SqlLikeFastPathKind,
1249    literal: &str,
1250    text: &str,
1251    case_sensitive: bool,
1252) -> bool {
1253    match kind {
1254        SqlLikeFastPathKind::MatchAll => true,
1255        SqlLikeFastPathKind::Exact => {
1256            if case_sensitive {
1257                literal.as_bytes() == text.as_bytes()
1258            } else {
1259                ascii_ci_eq_bytes(literal.as_bytes(), text.as_bytes())
1260            }
1261        }
1262        SqlLikeFastPathKind::Prefix => {
1263            if case_sensitive {
1264                text.as_bytes().starts_with(literal.as_bytes())
1265            } else {
1266                ascii_ci_starts_with(text, literal)
1267            }
1268        }
1269        SqlLikeFastPathKind::Suffix => {
1270            if case_sensitive {
1271                text.as_bytes().ends_with(literal.as_bytes())
1272            } else {
1273                ascii_ci_ends_with(text, literal)
1274            }
1275        }
1276        SqlLikeFastPathKind::Contains => {
1277            if case_sensitive {
1278                literal.is_empty() || memmem::find(text.as_bytes(), literal.as_bytes()).is_some()
1279            } else {
1280                ascii_ci_contains(text, literal)
1281            }
1282        }
1283    }
1284}
1285
1286/// Reusable matcher for simple LIKE patterns classified by `classify_sql_like_fast_path`.
1287pub struct SqlLikeFastPathMatcher<'a> {
1288    kind: SqlLikeFastPathKind,
1289    literal: &'a str,
1290    contains_finder: Option<memmem::Finder<'a>>,
1291    case_sensitive: bool,
1292}
1293
1294impl<'a> SqlLikeFastPathMatcher<'a> {
1295    #[must_use]
1296    pub fn new(kind: SqlLikeFastPathKind, literal: &'a str) -> Self {
1297        Self::new_cased(kind, literal, false)
1298    }
1299
1300    /// Build a matcher honoring `case_sensitive` (`PRAGMA case_sensitive_like`).
1301    #[must_use]
1302    pub fn new_cased(kind: SqlLikeFastPathKind, literal: &'a str, case_sensitive: bool) -> Self {
1303        let contains_finder = (kind == SqlLikeFastPathKind::Contains && !literal.is_empty())
1304            .then(|| memmem::Finder::new(literal.as_bytes()));
1305        Self {
1306            kind,
1307            literal,
1308            contains_finder,
1309            case_sensitive,
1310        }
1311    }
1312
1313    #[must_use]
1314    pub fn matches(&self, text: &str) -> bool {
1315        if let (SqlLikeFastPathKind::Contains, Some(finder)) = (self.kind, &self.contains_finder) {
1316            let text_bytes = text.as_bytes();
1317            let needle_bytes = self.literal.as_bytes();
1318            if needle_bytes.len() > text_bytes.len() {
1319                return false;
1320            }
1321            if finder.find(text_bytes).is_some() {
1322                return true;
1323            }
1324            // A byte-exact substring (the finder above) is the *only* match when
1325            // the connection's LIKE is case-sensitive; otherwise fall back to the
1326            // ASCII-case-folded scan.
1327            if self.case_sensitive {
1328                return false;
1329            }
1330            return ascii_ci_contains_folded_scan(text_bytes, needle_bytes);
1331        }
1332        sql_like_fast_path_matches_cased(self.kind, self.literal, text, self.case_sensitive)
1333    }
1334}
1335
1336#[must_use]
1337pub fn classify_sql_like_fast_path(
1338    pattern: &str,
1339    escape: Option<char>,
1340) -> Option<(SqlLikeFastPathKind, &str)> {
1341    if escape.is_some() || pattern.contains('_') {
1342        return None;
1343    }
1344    if !pattern.contains('%') {
1345        return Some((SqlLikeFastPathKind::Exact, pattern));
1346    }
1347    if pattern.chars().all(|ch| ch == '%') {
1348        return Some((SqlLikeFastPathKind::MatchAll, ""));
1349    }
1350
1351    let trimmed_start = pattern.trim_start_matches('%');
1352    let trimmed_end = pattern.trim_end_matches('%');
1353    if pattern.starts_with('%') && pattern.ends_with('%') {
1354        let core = trimmed_start.trim_end_matches('%');
1355        if core.is_empty() {
1356            return Some((SqlLikeFastPathKind::MatchAll, ""));
1357        }
1358        if !core.contains('%') {
1359            return Some((SqlLikeFastPathKind::Contains, core));
1360        }
1361    }
1362    if !pattern.starts_with('%') && trimmed_end.len() < pattern.len() && !trimmed_end.contains('%')
1363    {
1364        return Some((SqlLikeFastPathKind::Prefix, trimmed_end));
1365    }
1366    if !pattern.ends_with('%')
1367        && trimmed_start.len() < pattern.len()
1368        && !trimmed_start.contains('%')
1369    {
1370        return Some((SqlLikeFastPathKind::Suffix, trimmed_start));
1371    }
1372    None
1373}
1374
1375fn sql_like_inner(
1376    pattern: &[char],
1377    text: &[char],
1378    escape: Option<char>,
1379    pi: usize,
1380    ti: usize,
1381    case_sensitive: bool,
1382) -> bool {
1383    let mut pi = pi;
1384    let mut ti = ti;
1385
1386    while pi < pattern.len() {
1387        let pc = pattern[pi];
1388
1389        // Handle escape character.
1390        if Some(pc) == escape {
1391            pi += 1;
1392            if pi >= pattern.len() {
1393                return false; // Trailing escape is malformed.
1394            }
1395            // Match the escaped character literally.
1396            if ti >= text.len() || !chars_eq(pattern[pi], text[ti], case_sensitive) {
1397                return false;
1398            }
1399            pi += 1;
1400            ti += 1;
1401            continue;
1402        }
1403
1404        match pc {
1405            '%' => {
1406                // Skip consecutive % wildcards.
1407                while pi < pattern.len() && pattern[pi] == '%' {
1408                    pi += 1;
1409                }
1410                // If % is at end of pattern, matches everything.
1411                if pi >= pattern.len() {
1412                    return true;
1413                }
1414                // Try matching rest of pattern at each position.
1415                for start in ti..=text.len() {
1416                    if sql_like_inner(pattern, text, escape, pi, start, case_sensitive) {
1417                        return true;
1418                    }
1419                }
1420                return false;
1421            }
1422            '_' => {
1423                if ti >= text.len() {
1424                    return false;
1425                }
1426                pi += 1;
1427                ti += 1;
1428            }
1429            _ => {
1430                if ti >= text.len() || !chars_eq(pc, text[ti], case_sensitive) {
1431                    return false;
1432                }
1433                pi += 1;
1434                ti += 1;
1435            }
1436        }
1437    }
1438    ti >= text.len()
1439}
1440
1441/// LIKE literal-character comparison: byte-exact when `case_sensitive`
1442/// (`PRAGMA case_sensitive_like = ON`), otherwise ASCII case-folded (default).
1443#[inline]
1444fn chars_eq(a: char, b: char, case_sensitive: bool) -> bool {
1445    if case_sensitive {
1446        a == b
1447    } else {
1448        ascii_ci_eq(a, b)
1449    }
1450}
1451
1452/// ASCII-only case-insensitive character comparison (SQLite LIKE semantics).
1453fn ascii_ci_eq(a: char, b: char) -> bool {
1454    if a == b {
1455        return true;
1456    }
1457    // Only fold ASCII A-Z / a-z.
1458    a.is_ascii() && b.is_ascii() && a.eq_ignore_ascii_case(&b)
1459}
1460
1461#[inline]
1462fn ascii_fold_byte(byte: u8) -> u8 {
1463    byte.to_ascii_lowercase()
1464}
1465
1466#[inline]
1467fn ascii_ci_eq_byte(left: u8, right: u8) -> bool {
1468    left == right || ((left ^ right) == 0x20 && left.is_ascii_alphabetic())
1469}
1470
1471fn ascii_ci_eq_bytes(left: &[u8], right: &[u8]) -> bool {
1472    if left.len() != right.len() {
1473        return false;
1474    }
1475    let mut idx = 0;
1476    while idx < left.len() {
1477        if !ascii_ci_eq_byte(left[idx], right[idx]) {
1478            return false;
1479        }
1480        idx += 1;
1481    }
1482    true
1483}
1484
1485fn ascii_ci_starts_with(text: &str, prefix: &str) -> bool {
1486    let text = text.as_bytes();
1487    let prefix = prefix.as_bytes();
1488    text.len() >= prefix.len() && ascii_ci_eq_bytes(&text[..prefix.len()], prefix)
1489}
1490
1491fn ascii_ci_ends_with(text: &str, suffix: &str) -> bool {
1492    let text = text.as_bytes();
1493    let suffix = suffix.as_bytes();
1494    text.len() >= suffix.len() && ascii_ci_eq_bytes(&text[text.len() - suffix.len()..], suffix)
1495}
1496
1497fn ascii_ci_contains(text: &str, needle: &str) -> bool {
1498    let text = text.as_bytes();
1499    let needle = needle.as_bytes();
1500    if needle.is_empty() {
1501        return true;
1502    }
1503    if needle.len() > text.len() {
1504        return false;
1505    }
1506    if memmem::find(text, needle).is_some() {
1507        return true;
1508    }
1509
1510    ascii_ci_contains_folded_scan(text, needle)
1511}
1512
1513fn ascii_ci_contains_folded_scan(text: &[u8], needle: &[u8]) -> bool {
1514    if needle.is_empty() {
1515        return true;
1516    }
1517    if needle.len() > text.len() {
1518        return false;
1519    }
1520    let max_start = text.len() - needle.len();
1521    let first = needle[0];
1522    let first_folded = ascii_fold_byte(first);
1523    let first_alt = if first.is_ascii_alphabetic() {
1524        first_folded.to_ascii_uppercase()
1525    } else {
1526        first_folded
1527    };
1528    let mut start = 0;
1529    while start <= max_start {
1530        let rel = if first_folded == first_alt {
1531            memchr(first_folded, &text[start..=max_start])
1532        } else {
1533            memchr2(first_folded, first_alt, &text[start..=max_start])
1534        };
1535        let Some(rel) = rel else {
1536            break;
1537        };
1538        start += rel;
1539        if ascii_ci_eq_bytes(&text[start + 1..start + needle.len()], &needle[1..]) {
1540            return true;
1541        }
1542        start += 1;
1543    }
1544    false
1545}
1546
1547/// Accumulator for SQL `sum()` aggregate with SQLite overflow semantics.
1548///
1549/// Unlike expression arithmetic (which promotes to REAL on overflow), `sum()`
1550/// raises an error on integer overflow only if all non-NULL inputs remain in
1551/// the integer accumulator. A later REAL input suppresses the overflow error
1552/// and returns the approximate REAL sum, matching C sqlite3 behavior.
1553#[derive(Debug, Clone)]
1554pub struct SumAccumulator {
1555    /// Running integer sum (if still in integer mode).
1556    int_sum: i64,
1557    /// Running float sum retained in parallel so a later REAL input can fall
1558    /// back without losing an integer that overflowed the exact accumulator.
1559    float_sum: f64,
1560    /// KBN compensation error term.
1561    float_err: f64,
1562    /// Whether we've seen any non-NULL value.
1563    has_value: bool,
1564    /// Whether we're in float mode (any REAL-like input).
1565    is_float: bool,
1566    /// Whether an integer overflow occurred (error condition).
1567    overflow: bool,
1568}
1569
1570impl Default for SumAccumulator {
1571    fn default() -> Self {
1572        Self::new()
1573    }
1574}
1575
1576/// Kahan-Babuska-Neumaier compensated summation step matching upstream
1577/// aggregate precision behavior.
1578#[inline]
1579fn kbn_step(sum: &mut f64, err: &mut f64, value: f64) {
1580    let s = *sum;
1581    let t = s + value;
1582    if s.abs() > value.abs() {
1583        *err += (s - t) + value;
1584    } else {
1585        *err += (value - t) + s;
1586    }
1587    *sum = t;
1588}
1589
1590impl SumAccumulator {
1591    /// Create a new accumulator.
1592    pub const fn new() -> Self {
1593        Self {
1594            int_sum: 0,
1595            float_sum: 0.0,
1596            float_err: 0.0,
1597            has_value: false,
1598            is_float: false,
1599            overflow: false,
1600        }
1601    }
1602
1603    /// Add a value to the running sum.
1604    #[allow(clippy::cast_precision_loss)]
1605    pub fn accumulate(&mut self, val: &SqliteValue) {
1606        match val.to_sum_numeric_value() {
1607            SqliteValue::Null | SqliteValue::Text(_) | SqliteValue::Blob(_) => {}
1608            SqliteValue::Integer(i) => {
1609                self.has_value = true;
1610                if !self.is_float && !self.overflow {
1611                    match self.int_sum.checked_add(i) {
1612                        Some(result) => self.int_sum = result,
1613                        None => self.overflow = true,
1614                    }
1615                }
1616                kbn_step(&mut self.float_sum, &mut self.float_err, i as f64);
1617            }
1618            SqliteValue::Float(f) => {
1619                self.has_value = true;
1620                self.is_float = true;
1621                kbn_step(&mut self.float_sum, &mut self.float_err, f);
1622            }
1623        }
1624    }
1625
1626    /// Finalize the sum. Returns `Err` if integer overflow occurred while the
1627    /// accumulator stayed integer, `Ok(NULL)` if no non-NULL values were seen,
1628    /// or the sum value.
1629    pub fn finish(&self) -> Result<SqliteValue, SumOverflowError> {
1630        if !self.is_float && self.overflow {
1631            return Err(SumOverflowError);
1632        }
1633        if !self.has_value {
1634            return Ok(SqliteValue::Null);
1635        }
1636        if self.is_float {
1637            Ok(SqliteValue::Float(self.float_sum + self.float_err))
1638        } else {
1639            Ok(SqliteValue::Integer(self.int_sum))
1640        }
1641    }
1642}
1643
1644/// Error returned when `sum()` encounters integer overflow.
1645#[derive(Debug, Clone, PartialEq, Eq)]
1646pub struct SumOverflowError;
1647
1648impl fmt::Display for SumOverflowError {
1649    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1650        f.write_str("integer overflow in sum()")
1651    }
1652}
1653
1654impl fmt::Display for SqliteValue {
1655    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1656        match self {
1657            Self::Null => f.write_str("NULL"),
1658            Self::Integer(i) => write!(f, "{i}"),
1659            Self::Float(v) => f.write_str(&format_sqlite_float(*v)),
1660            Self::Text(s) => write!(f, "'{s}'"),
1661            Self::Blob(b) => {
1662                f.write_str("X'")?;
1663                for byte in b.iter() {
1664                    write!(f, "{byte:02X}")?;
1665                }
1666                f.write_str("'")
1667            }
1668        }
1669    }
1670}
1671
1672impl PartialEq for SqliteValue {
1673    fn eq(&self, other: &Self) -> bool {
1674        matches!(self.partial_cmp(other), Some(Ordering::Equal))
1675    }
1676}
1677
1678impl Eq for SqliteValue {}
1679
1680impl PartialOrd for SqliteValue {
1681    #[inline]
1682    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
1683        Some(self.cmp(other))
1684    }
1685}
1686
1687impl Ord for SqliteValue {
1688    #[inline]
1689    fn cmp(&self, other: &Self) -> Ordering {
1690        // SQLite sort order: NULL < numeric < text < blob
1691        let class_a = self.sort_class();
1692        let class_b = other.sort_class();
1693
1694        if class_a != class_b {
1695            return class_a.cmp(&class_b);
1696        }
1697
1698        match (self, other) {
1699            (Self::Null, Self::Null) => Ordering::Equal,
1700            (Self::Integer(a), Self::Integer(b)) => a.cmp(b),
1701            (Self::Float(a), Self::Float(b)) => a.partial_cmp(b).unwrap_or_else(|| a.total_cmp(b)),
1702            (Self::Integer(a), Self::Float(b)) => int_float_cmp(*a, *b),
1703            (Self::Float(a), Self::Integer(b)) => int_float_cmp(*b, *a).reverse(),
1704            (Self::Text(a), Self::Text(b)) => a.cmp(b),
1705            (Self::Blob(a), Self::Blob(b)) => a.cmp(b),
1706            _ => unreachable!(),
1707        }
1708    }
1709}
1710
1711impl From<i64> for SqliteValue {
1712    fn from(i: i64) -> Self {
1713        Self::Integer(i)
1714    }
1715}
1716
1717impl From<i32> for SqliteValue {
1718    fn from(i: i32) -> Self {
1719        Self::Integer(i64::from(i))
1720    }
1721}
1722
1723impl From<f64> for SqliteValue {
1724    fn from(f: f64) -> Self {
1725        Self::float_result_or_null(f)
1726    }
1727}
1728
1729impl From<String> for SqliteValue {
1730    fn from(s: String) -> Self {
1731        // SmallText stores strings ≤ 23 bytes inline without heap allocation.
1732        // Longer strings use Arc<str> internally.
1733        Self::Text(SmallText::from_string(s))
1734    }
1735}
1736
1737impl From<&str> for SqliteValue {
1738    fn from(s: &str) -> Self {
1739        Self::Text(SmallText::new(s))
1740    }
1741}
1742
1743impl From<Arc<str>> for SqliteValue {
1744    fn from(s: Arc<str>) -> Self {
1745        Self::Text(SmallText::from_arc(s))
1746    }
1747}
1748
1749impl From<Vec<u8>> for SqliteValue {
1750    fn from(b: Vec<u8>) -> Self {
1751        // Arc::from(Vec<u8>) reuses the Vec's heap buffer via
1752        // Vec → Box<[u8]> → Arc<[u8]>, avoiding a redundant copy.
1753        Self::Blob(Arc::from(b))
1754    }
1755}
1756
1757impl From<&[u8]> for SqliteValue {
1758    fn from(b: &[u8]) -> Self {
1759        Self::Blob(Arc::from(b))
1760    }
1761}
1762
1763impl From<Arc<[u8]>> for SqliteValue {
1764    fn from(b: Arc<[u8]>) -> Self {
1765        Self::Blob(b)
1766    }
1767}
1768
1769impl<T: Into<Self>> From<Option<T>> for SqliteValue {
1770    fn from(opt: Option<T>) -> Self {
1771        match opt {
1772            Some(v) => v.into(),
1773            None => Self::Null,
1774        }
1775    }
1776}
1777
1778/// Try to coerce a text string to INTEGER or REAL following SQLite NUMERIC
1779/// affinity rules. Returns `None` if the text is not a well-formed numeric
1780/// literal.
1781#[allow(
1782    clippy::cast_possible_truncation,
1783    clippy::cast_precision_loss,
1784    clippy::float_cmp
1785)]
1786fn try_coerce_text_to_numeric(s: &str) -> Option<SqliteValue> {
1787    let trimmed = trim_sqlite_ascii_whitespace(s);
1788    if trimmed.is_empty() {
1789        return None;
1790    }
1791    // Try integer first (preferred for NUMERIC affinity).
1792    if let Ok(i) = trimmed.parse::<i64>() {
1793        return Some(SqliteValue::Integer(i));
1794    }
1795    // Try float. Reject non-finite results (NaN, Infinity) since SQLite
1796    // does not recognise "nan", "inf", or "infinity" as numeric literals.
1797    // However, it does recognize literals like "1e999" which evaluate to Inf.
1798    if let Ok(f) = trimmed.parse::<f64>() {
1799        if !f.is_finite() {
1800            let lower = trimmed.to_ascii_lowercase();
1801            if lower.contains("inf") || lower.contains("nan") {
1802                return None;
1803            }
1804        }
1805        // If the float is an exact integer value within bounds, store as integer.
1806        // Checking bounds prevents incorrect saturation for values >= 2^63.
1807        if (-9_223_372_036_854_775_808.0..9_223_372_036_854_775_808.0).contains(&f) {
1808            #[allow(clippy::cast_possible_truncation)]
1809            let i = f as i64;
1810            #[allow(clippy::cast_precision_loss)]
1811            if (i as f64) == f {
1812                return Some(SqliteValue::Integer(i));
1813            }
1814        }
1815        return Some(SqliteValue::Float(f));
1816    }
1817    None
1818}
1819
1820/// Compare an integer with a float, preserving precision for large i64 values.
1821///
1822/// Matches C SQLite's `sqlite3IntFloatCompare` algorithm. The naive
1823/// `(i as f64).partial_cmp(&r)` loses precision for |i| > 2^53.
1824#[allow(clippy::cast_precision_loss, clippy::cast_possible_truncation)]
1825pub fn int_float_cmp(i: i64, r: f64) -> Ordering {
1826    if r.is_nan() {
1827        // SQLite treats NaN as NULL, and all integers are greater than NULL.
1828        return Ordering::Greater;
1829    }
1830    // If r is out of i64 range, the answer is obvious.
1831    if r < -9_223_372_036_854_775_808.0 {
1832        return Ordering::Greater;
1833    }
1834    if r >= 9_223_372_036_854_775_808.0 {
1835        return Ordering::Less;
1836    }
1837    // Truncate float to integer and compare integer parts.
1838    let y = r as i64;
1839    match i.cmp(&y) {
1840        Ordering::Less => Ordering::Less,
1841        Ordering::Greater => Ordering::Greater,
1842        // Integer parts equal — use float comparison as tiebreaker.
1843        Ordering::Equal => {
1844            let s = i as f64;
1845            s.partial_cmp(&r).unwrap_or(Ordering::Equal)
1846        }
1847    }
1848}
1849
1850/// Format a floating-point value as text matching SQLite's `%!.*g` behavior.
1851///
1852/// SQLite 3.52 and newer default REAL-to-TEXT conversion to 17 significant
1853/// digits through `sqlite3VdbeMemStringify()`. The `!` flag keeps a decimal
1854/// point so REAL text stays distinct from INTEGER text, for example `120.0`
1855/// rather than `120`.
1856#[must_use]
1857pub fn format_sqlite_float(f: f64) -> String {
1858    if f.is_nan() {
1859        return "NaN".to_owned();
1860    }
1861    if f.is_infinite() {
1862        return if f.is_sign_positive() {
1863            "Inf".to_owned()
1864        } else {
1865            "-Inf".to_owned()
1866        };
1867    }
1868    render_sqlite_float_decode(&sqlite_float_decode(f))
1869}
1870
1871const SQLITE_FLOAT_SIGNIFICANT_DIGITS: usize = 17;
1872const SQLITE_FLOAT_MAX_ROUND_DIGITS: usize = 20;
1873const SQLITE_FLOAT_GENERIC_PRECISION: i32 = 16;
1874const SQLITE_POWERS_OF_TEN_FIRST: i32 = -348;
1875const SQLITE_POWERS_OF_TEN_LAST: i32 = 347;
1876
1877#[derive(Debug)]
1878struct SqliteFloatDecode {
1879    digits: Vec<u8>,
1880    decimal_point: i32,
1881    negative: bool,
1882}
1883
1884fn render_sqlite_float_decode(decoded: &SqliteFloatDecode) -> String {
1885    let exponent = decoded.decimal_point - 1;
1886    if !(-4..=SQLITE_FLOAT_GENERIC_PRECISION).contains(&exponent) {
1887        return render_sqlite_float_exponential(decoded, exponent);
1888    }
1889    render_sqlite_float_fixed(decoded, exponent)
1890}
1891
1892fn render_sqlite_float_fixed(decoded: &SqliteFloatDecode, exponent: i32) -> String {
1893    let mut out = String::with_capacity(decoded.digits.len() + 8);
1894    if decoded.negative {
1895        out.push('-');
1896    }
1897
1898    let mut precision = SQLITE_FLOAT_GENERIC_PRECISION - exponent;
1899    let mut digit_idx = 0usize;
1900    let mut e2 = decoded.decimal_point - 1;
1901
1902    if e2 < 0 {
1903        out.push('0');
1904    } else {
1905        while e2 >= 0 {
1906            if let Some(&digit) = decoded.digits.get(digit_idx) {
1907                out.push(char::from(digit));
1908                digit_idx += 1;
1909            } else {
1910                out.push('0');
1911            }
1912            e2 -= 1;
1913        }
1914    }
1915
1916    out.push('.');
1917
1918    if e2 < -1 && precision > 0 {
1919        let zero_count = (-1 - e2).min(precision);
1920        for _ in 0..zero_count {
1921            out.push('0');
1922        }
1923        precision -= zero_count;
1924    }
1925
1926    if precision > 0 {
1927        let digits_after_decimal =
1928            (decoded.digits.len().saturating_sub(digit_idx)).min(precision as usize);
1929        for &digit in &decoded.digits[digit_idx..digit_idx + digits_after_decimal] {
1930            out.push(char::from(digit));
1931        }
1932    }
1933
1934    trim_sqlite_float_tail(&mut out);
1935    out
1936}
1937
1938fn render_sqlite_float_exponential(decoded: &SqliteFloatDecode, exponent: i32) -> String {
1939    let mut out = String::with_capacity(decoded.digits.len() + 8);
1940    if decoded.negative {
1941        out.push('-');
1942    }
1943
1944    let first = decoded.digits.first().copied().unwrap_or(b'0');
1945    out.push(char::from(first));
1946    out.push('.');
1947    let digits_after_decimal =
1948        (decoded.digits.len().saturating_sub(1)).min(SQLITE_FLOAT_GENERIC_PRECISION as usize);
1949    for &digit in decoded.digits.iter().skip(1).take(digits_after_decimal) {
1950        out.push(char::from(digit));
1951    }
1952    trim_sqlite_float_tail(&mut out);
1953
1954    out.push('e');
1955    let mut abs_exp = exponent;
1956    if abs_exp < 0 {
1957        out.push('-');
1958        abs_exp = -abs_exp;
1959    } else {
1960        out.push('+');
1961    }
1962    if abs_exp >= 100 {
1963        out.push(char::from(b'0' + (abs_exp / 100) as u8));
1964        abs_exp %= 100;
1965    }
1966    out.push(char::from(b'0' + (abs_exp / 10) as u8));
1967    out.push(char::from(b'0' + (abs_exp % 10) as u8));
1968    out
1969}
1970
1971fn trim_sqlite_float_tail(out: &mut String) {
1972    while out.ends_with('0') {
1973        out.pop();
1974    }
1975    if out.ends_with('.') {
1976        out.push('0');
1977    }
1978}
1979
1980fn sqlite_float_decode(f: f64) -> SqliteFloatDecode {
1981    let negative = f < 0.0;
1982    let r = if negative { -f } else { f };
1983    if r == 0.0 {
1984        return SqliteFloatDecode {
1985            digits: vec![b'0'],
1986            decimal_point: 1,
1987            negative: false,
1988        };
1989    }
1990
1991    let bits = r.to_bits();
1992    let raw_exponent = ((bits >> 52) & 0x7ff) as i32;
1993    let mut mantissa = bits & 0x000f_ffff_ffff_ffff;
1994    let binary_exponent = if raw_exponent == 0 {
1995        let leading = mantissa.leading_zeros();
1996        mantissa <<= leading;
1997        -1074 - leading as i32
1998    } else {
1999        mantissa = (mantissa << 11) | (1_u64 << 63);
2000        raw_exponent - 1086
2001    };
2002
2003    let (decimal, decimal_exponent) = sqlite_fp2_convert10(mantissa, binary_exponent, 18);
2004    let mut digits = decimal.to_string().into_bytes();
2005    let mut digit_count = digits.len();
2006    let mut decimal_point = digit_count as i32 + decimal_exponent;
2007    let mut round_at = SQLITE_FLOAT_SIGNIFICANT_DIGITS;
2008
2009    if round_at < digit_count || digit_count > SQLITE_FLOAT_MAX_ROUND_DIGITS {
2010        if round_at == SQLITE_FLOAT_SIGNIFICANT_DIGITS {
2011            round_at = sqlite_adjust_17_digit_rounding(
2012                r,
2013                &digits,
2014                decimal_exponent,
2015                digit_count,
2016                decimal_point,
2017                round_at,
2018            );
2019        }
2020        if digits.get(round_at).copied().unwrap_or(b'0') >= b'5' {
2021            let mut idx = round_at - 1;
2022            loop {
2023                digits[idx] += 1;
2024                if digits[idx] <= b'9' {
2025                    break;
2026                }
2027                digits[idx] = b'0';
2028                if idx == 0 {
2029                    digits.insert(0, b'1');
2030                    round_at += 1;
2031                    decimal_point += 1;
2032                    break;
2033                }
2034                idx -= 1;
2035            }
2036        }
2037        digit_count = round_at;
2038        digits.truncate(digit_count);
2039    }
2040
2041    while digit_count > 1 && digits[digit_count - 1] == b'0' {
2042        digit_count -= 1;
2043    }
2044    digits.truncate(digit_count);
2045
2046    SqliteFloatDecode {
2047        digits,
2048        decimal_point,
2049        negative,
2050    }
2051}
2052
2053fn sqlite_adjust_17_digit_rounding(
2054    r: f64,
2055    digits: &[u8],
2056    decimal_exponent: i32,
2057    digit_count: usize,
2058    decimal_point: i32,
2059    round_at: usize,
2060) -> usize {
2061    if digits.len() <= SQLITE_FLOAT_SIGNIFICANT_DIGITS {
2062        return round_at;
2063    }
2064
2065    if digits[15] == b'9' && digits[14] == b'9' {
2066        let mut keep = 14usize;
2067        while keep > 0 && digits[keep - 1] == b'9' {
2068            keep -= 1;
2069        }
2070        let candidate = if keep == 0 {
2071            1
2072        } else {
2073            decimal_digits_to_u64(&digits[..keep]) + 1
2074        };
2075        if r == sqlite_fp10_convert2(
2076            candidate,
2077            decimal_exponent + digit_count as i32 - keep as i32,
2078        ) {
2079            return keep + 1;
2080        }
2081    } else if decimal_point >= digit_count as i32
2082        || (digits[15] == b'0' && digits[14] == b'0' && digits[13] == b'0')
2083    {
2084        let mut keep = 13usize;
2085        while keep > 0 && digits[keep - 1] == b'0' {
2086            keep -= 1;
2087        }
2088        if keep > 0 {
2089            let candidate = decimal_digits_to_u64(&digits[..keep]);
2090            if r == sqlite_fp10_convert2(
2091                candidate,
2092                decimal_exponent + digit_count as i32 - keep as i32,
2093            ) {
2094                return keep + 1;
2095            }
2096        }
2097    }
2098
2099    round_at
2100}
2101
2102fn decimal_digits_to_u64(digits: &[u8]) -> u64 {
2103    digits
2104        .iter()
2105        .fold(0_u64, |acc, digit| acc * 10 + u64::from(*digit - b'0'))
2106}
2107
2108fn sqlite_fp2_convert10(mantissa: u64, binary_exponent: i32, digits: i32) -> (u64, i32) {
2109    let power = digits - 1 - pwr2_to_10(binary_exponent + 63);
2110    let (power_hi, power_lo) = power_of_ten(power);
2111    let (mut high, _) = sqlite_multiply_128(mantissa, power_hi);
2112    let _ = power_lo;
2113    if digits == 18 {
2114        high >>= -(binary_exponent + pwr10_to_2(power) + 2) as u32;
2115        (high.wrapping_add((high << 1) & 2) >> 1, -power)
2116    } else {
2117        high >>= -(binary_exponent + pwr10_to_2(power) + 1) as u32;
2118        (high, -power)
2119    }
2120}
2121
2122fn sqlite_fp10_convert2(decimal: u64, power: i32) -> f64 {
2123    if power < SQLITE_POWERS_OF_TEN_FIRST {
2124        return 0.0;
2125    }
2126    if power > SQLITE_POWERS_OF_TEN_LAST {
2127        return f64::INFINITY;
2128    }
2129
2130    let bit_width = 64 - decimal.leading_zeros() as i32;
2131    let binary_power = pwr10_to_2(power);
2132    let mut exponent = 53 - bit_width - binary_power;
2133    if exponent > 1074 {
2134        if exponent >= 1130 {
2135            return 0.0;
2136        }
2137        exponent = 1074;
2138    }
2139
2140    let shift = -(exponent - (64 - bit_width) + binary_power + 3);
2141    let shift = shift.clamp(0, 63) as u32;
2142    let (mut power_hi, mut power_lo) = power_of_ten(power);
2143    if power_lo != 0 {
2144        power_hi = power_hi.wrapping_add(1);
2145        power_lo = !power_lo;
2146    }
2147
2148    let shifted_decimal = decimal << (64 - bit_width);
2149    let (mut high, low) = sqlite_multiply_128(shifted_decimal, power_hi);
2150    let mid1 = (low >> 32) as u32;
2151    let mut sticky = 1_u64;
2152    if (high & low_mask(shift)) == 0 {
2153        let (mid2_high, _) = sqlite_multiply_128(shifted_decimal, u64::from(power_lo) << 32);
2154        let mid2 = (mid2_high >> 32) as u32;
2155        sticky = u64::from(mid1.wrapping_sub(mid2) > 1);
2156        high = high.wrapping_sub(u64::from(mid1 < mid2));
2157    }
2158
2159    let mut rounded = (high >> shift) | sticky;
2160    let adjust = u32::from(rounded >= (1_u64 << 55) - 2);
2161    if adjust != 0 {
2162        rounded = (rounded >> adjust) | (rounded & 1);
2163        exponent -= adjust as i32;
2164    }
2165
2166    let mut bits = (rounded + 1 + ((rounded >> 2) & 1)) >> 2;
2167    if exponent <= -972 {
2168        return f64::INFINITY;
2169    }
2170    if (bits & (1_u64 << 52)) != 0 {
2171        bits = (bits & !(1_u64 << 52)) | ((1075 - exponent) as u64) << 52;
2172    }
2173    f64::from_bits(bits)
2174}
2175
2176fn low_mask(bits: u32) -> u64 {
2177    if bits == 0 { 0 } else { (1_u64 << bits) - 1 }
2178}
2179
2180fn sqlite_multiply_128(left: u64, right: u64) -> (u64, u64) {
2181    let product = u128::from(left) * u128::from(right);
2182    ((product >> 64) as u64, product as u64)
2183}
2184
2185fn sqlite_multiply_160(high: u64, low: u32, right: u64) -> (u64, u32) {
2186    let product =
2187        u128::from(high) * u128::from(right) + ((u128::from(low) * u128::from(right)) >> 32);
2188    (
2189        (product >> 64) as u64,
2190        ((product >> 32) & u128::from(u32::MAX)) as u32,
2191    )
2192}
2193
2194fn pwr10_to_2(power: i32) -> i32 {
2195    (power * 108_853) >> 15
2196}
2197
2198fn pwr2_to_10(power: i32) -> i32 {
2199    (power * 78_913) >> 18
2200}
2201
2202fn power_of_ten(power: i32) -> (u64, u32) {
2203    const BASE: [u64; 27] = [
2204        0x8000_0000_0000_0000,
2205        0xa000_0000_0000_0000,
2206        0xc800_0000_0000_0000,
2207        0xfa00_0000_0000_0000,
2208        0x9c40_0000_0000_0000,
2209        0xc350_0000_0000_0000,
2210        0xf424_0000_0000_0000,
2211        0x9896_8000_0000_0000,
2212        0xbebc_2000_0000_0000,
2213        0xee6b_2800_0000_0000,
2214        0x9502_f900_0000_0000,
2215        0xba43_b740_0000_0000,
2216        0xe8d4_a510_0000_0000,
2217        0x9184_e72a_0000_0000,
2218        0xb5e6_20f4_8000_0000,
2219        0xe35f_a931_a000_0000,
2220        0x8e1b_c9bf_0400_0000,
2221        0xb1a2_bc2e_c500_0000,
2222        0xde0b_6b3a_7640_0000,
2223        0x8ac7_2304_89e8_0000,
2224        0xad78_ebc5_ac62_0000,
2225        0xd8d7_26b7_177a_8000,
2226        0x8786_7832_6eac_9000,
2227        0xa968_163f_0a57_b400,
2228        0xd3c2_1bce_cced_a100,
2229        0x8459_5161_4014_84a0,
2230        0xa56f_a5b9_9019_a5c8,
2231    ];
2232    const SCALE: [u64; 26] = [
2233        0x8049_a4ac_0c58_11ae,
2234        0xcf42_894a_5dce_35ea,
2235        0xa76c_5823_38ed_2621,
2236        0x873e_4f75_e222_4e68,
2237        0xda7f_5bf5_9096_6848,
2238        0xb080_392c_c434_9dec,
2239        0x8e93_8662_882a_f53e,
2240        0xe658_29b3_046b_0afa,
2241        0xba12_1a46_50e4_ddeb,
2242        0x964e_858c_91ba_2655,
2243        0xf2d5_6790_ab41_c2a2,
2244        0xc428_d05a_a475_1e4c,
2245        0x9e74_d1b7_91e0_7e48,
2246        0xcccc_cccc_cccc_cccc,
2247        0xcecb_8f27_f420_0f3a,
2248        0xa70c_3c40_a64e_6c51,
2249        0x86f0_ac99_b4e8_dafd,
2250        0xda01_ee64_1a70_8de9,
2251        0xb01a_e745_b101_e9e4,
2252        0x8e41_ade9_fbeb_c27d,
2253        0xe5d3_ef28_2a24_2e81,
2254        0xb9a7_4a06_37ce_2ee1,
2255        0x95f8_3d0a_1fb6_9cd9,
2256        0xf24a_01a7_3cf2_dccf,
2257        0xc3b8_3581_09e8_4f07,
2258        0x9e19_db92_b4e3_1ba9,
2259    ];
2260    const SCALE_LO: [u32; 26] = [
2261        0x205b_896d,
2262        0x5206_4cad,
2263        0xaf2a_f2b8,
2264        0x5a77_44a7,
2265        0xaf39_a475,
2266        0xbd8d_794e,
2267        0x547e_b47b,
2268        0x0cb4_a5a3,
2269        0x92f3_4d62,
2270        0x3a6a_07f9,
2271        0xfae2_7299,
2272        0xaa97_e14c,
2273        0x775e_a265,
2274        0xcccc_cccc,
2275        0x0000_0000,
2276        0x9990_90b6,
2277        0x69a0_28bb,
2278        0xe80e_6f48,
2279        0x5ec0_5dd0,
2280        0x1458_8f14,
2281        0x8f16_68c9,
2282        0x6d95_3e2c,
2283        0x4abd_af10,
2284        0xbc63_3b39,
2285        0x0a86_2f81,
2286        0x6c07_a2c2,
2287    ];
2288
2289    debug_assert!((SQLITE_POWERS_OF_TEN_FIRST..=SQLITE_POWERS_OF_TEN_LAST).contains(&power));
2290
2291    let (group, offset) = if power < 0 {
2292        if power == -1 {
2293            return (SCALE[13], SCALE_LO[13]);
2294        }
2295        let mut group = power / 27;
2296        let mut offset = power % 27;
2297        if offset != 0 {
2298            group -= 1;
2299            offset += 27;
2300        }
2301        (group, offset)
2302    } else if power < 27 {
2303        return (BASE[power as usize], 0);
2304    } else {
2305        (power / 27, power % 27)
2306    };
2307
2308    let scale_idx = (group + 13) as usize;
2309    let mut high = SCALE[scale_idx];
2310    if offset == 0 {
2311        return (high, SCALE_LO[scale_idx]);
2312    }
2313
2314    let (scaled, mut low) = sqlite_multiply_160(high, SCALE_LO[scale_idx], BASE[offset as usize]);
2315    high = scaled;
2316    if (high & (1_u64 << 63)) == 0 {
2317        high = (high << 1) | u64::from(low >> 31);
2318        low = (low << 1) | 1;
2319    }
2320    (high, low)
2321}
2322
2323#[cfg(test)]
2324#[allow(clippy::float_cmp, clippy::approx_constant)]
2325mod tests {
2326    use super::*;
2327
2328    struct ValuePoolTestGuard;
2329
2330    impl ValuePoolTestGuard {
2331        fn new() -> Self {
2332            pool_clear();
2333            reset_value_pool_test_stats();
2334            Self
2335        }
2336    }
2337
2338    impl Drop for ValuePoolTestGuard {
2339        fn drop(&mut self) {
2340            pool_clear();
2341            reset_value_pool_test_stats();
2342        }
2343    }
2344
2345    fn log_value_pool_test_stats(test_name: &str) -> ValuePoolStats {
2346        let stats = value_pool_test_stats_snapshot();
2347        eprintln!(
2348            "bead_id=bd-nsvud test={test_name} slab_alloc_count={} slab_return_count={} global_alloc_fallback_count={} slab_high_water_mark={} pool_len={}",
2349            stats.slab_alloc_count,
2350            stats.slab_return_count,
2351            stats.global_alloc_fallback_count,
2352            stats.slab_high_water_mark,
2353            pool_len(),
2354        );
2355        stats
2356    }
2357
2358    #[test]
2359    fn test_slab_basic_alloc_dealloc() {
2360        let _guard = ValuePoolTestGuard::new();
2361        const ROUND_TRIP_COUNT: usize = 100;
2362
2363        assert_eq!(pool_len(), 0);
2364        assert_eq!(pool_acquire(), None);
2365        assert_eq!(
2366            value_pool_test_stats_snapshot(),
2367            ValuePoolStats {
2368                slab_alloc_count: 0,
2369                slab_return_count: 0,
2370                global_alloc_fallback_count: 1,
2371                slab_high_water_mark: 0,
2372            }
2373        );
2374
2375        reset_value_pool_test_stats();
2376        for value in 0..ROUND_TRIP_COUNT {
2377            pool_return(SqliteValue::Integer(value as i64));
2378        }
2379        assert_eq!(pool_len(), ROUND_TRIP_COUNT);
2380        assert_eq!(
2381            value_pool_test_stats_snapshot(),
2382            ValuePoolStats {
2383                slab_alloc_count: 0,
2384                slab_return_count: ROUND_TRIP_COUNT,
2385                global_alloc_fallback_count: 0,
2386                slab_high_water_mark: ROUND_TRIP_COUNT,
2387            }
2388        );
2389
2390        reset_value_pool_test_stats();
2391        for expected in (0..ROUND_TRIP_COUNT).rev() {
2392            assert_eq!(pool_acquire(), Some(SqliteValue::Integer(expected as i64)));
2393        }
2394        assert_eq!(pool_len(), 0);
2395        assert_eq!(
2396            log_value_pool_test_stats("test_slab_basic_alloc_dealloc"),
2397            ValuePoolStats {
2398                slab_alloc_count: ROUND_TRIP_COUNT,
2399                slab_return_count: 0,
2400                global_alloc_fallback_count: 0,
2401                slab_high_water_mark: 0,
2402            }
2403        );
2404    }
2405
2406    #[test]
2407    fn test_slab_exhaustion_fallback() {
2408        let _guard = ValuePoolTestGuard::new();
2409
2410        for value in 0..=VALUE_POOL_CAP {
2411            pool_return(SqliteValue::Integer(value as i64));
2412        }
2413        assert_eq!(pool_len(), VALUE_POOL_CAP);
2414        assert_eq!(
2415            value_pool_test_stats_snapshot(),
2416            ValuePoolStats {
2417                slab_alloc_count: 0,
2418                slab_return_count: VALUE_POOL_CAP,
2419                global_alloc_fallback_count: 0,
2420                slab_high_water_mark: VALUE_POOL_CAP,
2421            }
2422        );
2423
2424        reset_value_pool_test_stats();
2425        for _ in 0..VALUE_POOL_CAP {
2426            assert!(pool_acquire().is_some());
2427        }
2428        assert_eq!(pool_acquire(), None);
2429        assert_eq!(pool_len(), 0);
2430        assert_eq!(
2431            log_value_pool_test_stats("test_slab_exhaustion_fallback"),
2432            ValuePoolStats {
2433                slab_alloc_count: VALUE_POOL_CAP,
2434                slab_return_count: 0,
2435                global_alloc_fallback_count: 1,
2436                slab_high_water_mark: 0,
2437            }
2438        );
2439    }
2440
2441    #[test]
2442    fn test_slab_no_leak() {
2443        let _guard = ValuePoolTestGuard::new();
2444        const ITERATIONS: usize = 10_000;
2445
2446        let (weak_tx, weak_rx) = std::sync::mpsc::channel();
2447        let (release_tx, release_rx) = std::sync::mpsc::channel();
2448
2449        let worker = std::thread::spawn(move || {
2450            pool_clear();
2451            reset_value_pool_test_stats();
2452
2453            let mut pooled_weak = None;
2454            let mut overflow_weak = None;
2455            for value in 0..ITERATIONS {
2456                let payload: Arc<[u8]> =
2457                    Arc::from(vec![(value % 251) as u8; 64].into_boxed_slice());
2458                if value == 0 {
2459                    pooled_weak = Some(Arc::downgrade(&payload));
2460                } else if value == ITERATIONS - 1 {
2461                    overflow_weak = Some(Arc::downgrade(&payload));
2462                }
2463                pool_return(SqliteValue::Blob(payload));
2464            }
2465
2466            assert_eq!(
2467                pool_len(),
2468                VALUE_POOL_CAP,
2469                "the slab must retain at most VALUE_POOL_CAP entries",
2470            );
2471            weak_tx
2472                .send((
2473                    pooled_weak.expect("capture pooled weak handle"),
2474                    overflow_weak.expect("capture overflow weak handle"),
2475                    log_value_pool_test_stats("test_slab_no_leak"),
2476                ))
2477                .expect("send slab leak stats");
2478            release_rx.recv().expect("wait for release");
2479        });
2480
2481        let (pooled_weak, overflow_weak, stats) =
2482            weak_rx.recv().expect("receive weak blob handles");
2483        assert!(
2484            pooled_weak.upgrade().is_some(),
2485            "pooled blob should remain alive while the owning thread is running"
2486        );
2487        assert!(
2488            overflow_weak.upgrade().is_none(),
2489            "values beyond VALUE_POOL_CAP should fall back to normal drop instead of staying pooled"
2490        );
2491        assert_eq!(
2492            stats,
2493            ValuePoolStats {
2494                slab_alloc_count: 0,
2495                slab_return_count: VALUE_POOL_CAP,
2496                global_alloc_fallback_count: 0,
2497                slab_high_water_mark: VALUE_POOL_CAP,
2498            }
2499        );
2500
2501        release_tx.send(()).expect("release worker thread");
2502        worker.join().expect("join worker");
2503
2504        assert!(
2505            pooled_weak.upgrade().is_none(),
2506            "thread-local slab contents must be dropped when the thread exits"
2507        );
2508    }
2509
2510    #[test]
2511    fn test_slab_thread_local_isolation() {
2512        let _guard = ValuePoolTestGuard::new();
2513
2514        pool_return(SqliteValue::Integer(11));
2515        assert_eq!(pool_len(), 1);
2516
2517        let worker = std::thread::spawn(|| {
2518            pool_clear();
2519            reset_value_pool_test_stats();
2520
2521            assert_eq!(pool_len(), 0, "worker thread must start with an empty slab");
2522            pool_return(SqliteValue::Integer(22));
2523            assert_eq!(pool_len(), 1);
2524            assert_eq!(
2525                value_pool_test_stats_snapshot(),
2526                ValuePoolStats {
2527                    slab_alloc_count: 0,
2528                    slab_return_count: 1,
2529                    global_alloc_fallback_count: 0,
2530                    slab_high_water_mark: 1,
2531                }
2532            );
2533            assert_eq!(pool_acquire(), Some(SqliteValue::Integer(22)));
2534            assert_eq!(pool_len(), 0);
2535        });
2536        worker.join().expect("join worker");
2537
2538        assert_eq!(
2539            pool_len(),
2540            1,
2541            "worker thread slab operations must not affect the caller thread"
2542        );
2543        assert_eq!(pool_acquire(), Some(SqliteValue::Integer(11)));
2544        assert_eq!(pool_len(), 0);
2545        let stats = log_value_pool_test_stats("test_slab_thread_local_isolation");
2546        assert_eq!(
2547            stats,
2548            ValuePoolStats {
2549                slab_alloc_count: 1,
2550                slab_return_count: 1,
2551                global_alloc_fallback_count: 0,
2552                slab_high_water_mark: 1,
2553            }
2554        );
2555    }
2556
2557    #[test]
2558    fn test_slab_zero_malloc_steady_state() {
2559        let _guard = ValuePoolTestGuard::new();
2560        const WARM_POOL_DEPTH: usize = VALUE_POOL_CAP;
2561        const ITERATIONS: usize = 1_000;
2562        const INITIAL_TEXT: &str =
2563            "steady-state pooled string backing store for bd-nsvud warmup payload";
2564        const REUSED_TEXT: &str = "steady-state pooled overwrite stays in-buffer";
2565
2566        assert!(
2567            REUSED_TEXT.len() <= INITIAL_TEXT.len(),
2568            "steady-state overwrite must fit within the warmed heap allocation"
2569        );
2570
2571        for _ in 0..WARM_POOL_DEPTH {
2572            pool_return(SqliteValue::Text(SmallText::new(INITIAL_TEXT)));
2573        }
2574        assert_eq!(pool_len(), WARM_POOL_DEPTH);
2575
2576        reset_value_pool_test_stats();
2577        for _ in 0..ITERATIONS {
2578            let mut reused = pool_acquire().unwrap_or(SqliteValue::Null);
2579            let SqliteValue::Text(existing) = &mut reused else {
2580                panic!("warmed slab entry should remain a text value");
2581            };
2582            let original_ptr = existing.as_str().as_ptr();
2583            existing.overwrite(REUSED_TEXT);
2584            assert_eq!(
2585                existing.as_str().as_ptr(),
2586                original_ptr,
2587                "steady-state overwrite should reuse the warmed heap buffer",
2588            );
2589            assert_eq!(existing.as_str(), REUSED_TEXT);
2590            pool_return(reused);
2591        }
2592
2593        assert_eq!(pool_len(), WARM_POOL_DEPTH);
2594        assert_eq!(
2595            log_value_pool_test_stats("test_slab_zero_malloc_steady_state"),
2596            ValuePoolStats {
2597                slab_alloc_count: ITERATIONS,
2598                slab_return_count: ITERATIONS,
2599                global_alloc_fallback_count: 0,
2600                slab_high_water_mark: WARM_POOL_DEPTH,
2601            }
2602        );
2603    }
2604
2605    #[test]
2606    fn test_small_text_heap_clone_lazily_promotes_to_shared_arc() {
2607        let text = SmallText::new("this string is definitely longer than twenty three bytes");
2608        let SmallTextRepr::HeapOwned { shared, .. } = &text.repr else {
2609            panic!("long text should start in heap-owned mode");
2610        };
2611        assert!(
2612            shared.get().is_none(),
2613            "long text should not allocate Arc eagerly before cloning"
2614        );
2615
2616        let cloned = text.clone();
2617
2618        let SmallTextRepr::HeapOwned { shared, .. } = &text.repr else {
2619            panic!("original text should remain heap-owned after clone");
2620        };
2621        assert!(
2622            shared.get().is_some(),
2623            "first clone should materialize a shared Arc lazily"
2624        );
2625        assert!(
2626            matches!(cloned.repr, SmallTextRepr::HeapShared(_)),
2627            "cloned text should use the shared Arc representation"
2628        );
2629        assert_eq!(text.as_str(), cloned.as_str());
2630    }
2631
2632    #[test]
2633    fn test_small_text_overwrite_reuses_unique_heap_buffer() {
2634        let mut text = SmallText::new("this string is definitely longer than twenty three bytes");
2635        let (original_ptr, original_capacity) = match &text.repr {
2636            SmallTextRepr::HeapOwned { text, shared } => {
2637                assert!(shared.get().is_none(), "fresh heap text should be unshared");
2638                (text.as_ptr(), text.capacity())
2639            }
2640            _ => panic!("long text should start in heap-owned mode"),
2641        };
2642
2643        text.overwrite("another long string that still fits the same allocation");
2644
2645        match &text.repr {
2646            SmallTextRepr::HeapOwned { text, shared } => {
2647                assert!(
2648                    shared.get().is_none(),
2649                    "overwrite should keep text single-owner"
2650                );
2651                assert_eq!(text.as_ptr(), original_ptr);
2652                assert_eq!(text.capacity(), original_capacity);
2653                assert_eq!(
2654                    text.as_str(),
2655                    "another long string that still fits the same allocation"
2656                );
2657            }
2658            _ => panic!("overwrite should keep long text in heap-owned mode"),
2659        }
2660    }
2661
2662    #[test]
2663    fn test_small_text_overwrite_detaches_from_shared_arc() {
2664        let original = "this string is definitely longer than twenty three bytes";
2665        let mut text = SmallText::new(original);
2666        let (original_ptr, original_capacity) = match &text.repr {
2667            SmallTextRepr::HeapOwned { text, .. } => (text.as_ptr(), text.capacity()),
2668            _ => panic!("long text should start in heap-owned mode"),
2669        };
2670        let replacement = "replacement text that must not mutate the shared clone";
2671        assert!(
2672            replacement.len() <= original_capacity,
2673            "replacement should fit the original heap allocation for this regression",
2674        );
2675        let clone = text.clone();
2676
2677        text.overwrite(replacement);
2678
2679        assert_eq!(
2680            clone.as_str(),
2681            original,
2682            "existing shared clones must keep the original contents"
2683        );
2684        assert_eq!(text.as_str(), replacement);
2685        match &text.repr {
2686            SmallTextRepr::HeapOwned { text, shared } => {
2687                assert_eq!(
2688                    text.as_ptr(),
2689                    original_ptr,
2690                    "overwriting a cloned long string should keep the owned buffer",
2691                );
2692                assert_eq!(
2693                    text.capacity(),
2694                    original_capacity,
2695                    "detaching from the shared cache should preserve capacity",
2696                );
2697                assert!(
2698                    shared.get().is_none(),
2699                    "overwrite should reset the lazy shared cache after detaching"
2700                );
2701            }
2702            _ => panic!("overwrite should restore heap-owned mode"),
2703        }
2704    }
2705
2706    #[test]
2707    fn test_pool_return_reusable_keeps_only_reusable_heap_storage() {
2708        let _guard = ValuePoolTestGuard::new();
2709
2710        pool_return_reusable(SqliteValue::Text(SmallText::new("tiny")));
2711        assert_eq!(
2712            pool_len(),
2713            0,
2714            "inline text should not occupy reusable slab slots",
2715        );
2716
2717        let owned_text = SmallText::new("this string is definitely longer than twenty three bytes");
2718        let _clone = owned_text.clone();
2719        pool_return_reusable(SqliteValue::Text(owned_text));
2720        assert_eq!(
2721            pool_len(),
2722            1,
2723            "heap-owned text should stay reusable even after serving shared clones",
2724        );
2725        assert!(matches!(pool_acquire(), Some(SqliteValue::Text(_))));
2726        assert_eq!(pool_len(), 0);
2727
2728        let shared_text =
2729            Arc::<str>::from("this string is definitely longer than twenty three bytes");
2730        pool_return_reusable(SqliteValue::Text(SmallText::from_arc(Arc::clone(
2731            &shared_text,
2732        ))));
2733        assert_eq!(
2734            pool_len(),
2735            0,
2736            "arc-backed shared text should not enter the reusable slab",
2737        );
2738
2739        let shared_blob = Arc::<[u8]>::from([0xCA_u8, 0xFE, 0xBA, 0xBE].as_slice());
2740        pool_return_reusable(SqliteValue::Blob(Arc::clone(&shared_blob)));
2741        assert_eq!(
2742            pool_len(),
2743            0,
2744            "shared blob allocations should not displace reusable slab entries",
2745        );
2746
2747        let unique_blob = Arc::<[u8]>::from([1_u8, 2, 3, 4].as_slice());
2748        pool_return_reusable(SqliteValue::Blob(unique_blob));
2749        assert_eq!(
2750            pool_len(),
2751            1,
2752            "unique blob allocations should remain eligible for slab reuse",
2753        );
2754    }
2755
2756    #[test]
2757    fn test_small_text_concurrent_clone_promotion_keeps_contents_stable() {
2758        let text = Arc::new(SmallText::new(
2759            "this string is definitely longer than twenty three bytes",
2760        ));
2761        let expected = text.as_str().to_owned();
2762        let SmallTextRepr::HeapOwned { shared, .. } = &text.repr else {
2763            panic!("long text should start in heap-owned mode");
2764        };
2765        assert!(
2766            shared.get().is_none(),
2767            "shared Arc should still be lazy before concurrent clones"
2768        );
2769
2770        let barrier = Arc::new(std::sync::Barrier::new(5));
2771        let mut workers = Vec::new();
2772        for _ in 0..4 {
2773            let text = Arc::clone(&text);
2774            let barrier = Arc::clone(&barrier);
2775            let expected = expected.clone();
2776            workers.push(std::thread::spawn(move || {
2777                barrier.wait();
2778                for _ in 0..64 {
2779                    let cloned = (*text).clone();
2780                    assert_eq!(cloned.as_str(), expected);
2781                    assert!(
2782                        matches!(cloned.repr, SmallTextRepr::HeapShared(_)),
2783                        "concurrent clone should reuse the shared Arc representation"
2784                    );
2785                }
2786            }));
2787        }
2788
2789        barrier.wait();
2790        for worker in workers {
2791            worker
2792                .join()
2793                .expect("join concurrent small-text clone worker");
2794        }
2795
2796        let SmallTextRepr::HeapOwned { shared, .. } = &text.repr else {
2797            panic!("original text should remain heap-owned after clone promotion");
2798        };
2799        let shared = shared
2800            .get()
2801            .expect("concurrent clones should promote the lazy shared Arc");
2802        assert_eq!(shared.as_ref(), expected);
2803        assert_eq!(text.as_str(), expected);
2804    }
2805
2806    #[test]
2807    fn null_properties() {
2808        let v = SqliteValue::Null;
2809        assert!(v.is_null());
2810        assert_eq!(v.to_integer(), 0);
2811        assert_eq!(v.to_float(), 0.0);
2812        assert_eq!(v.to_text(), "");
2813        assert_eq!(v.to_string(), "NULL");
2814    }
2815
2816    #[test]
2817    fn integer_properties() {
2818        let v = SqliteValue::Integer(42);
2819        assert!(!v.is_null());
2820        assert_eq!(v.as_integer(), Some(42));
2821        assert_eq!(v.to_integer(), 42);
2822        assert_eq!(v.to_float(), 42.0);
2823        assert_eq!(v.to_text(), "42");
2824    }
2825
2826    #[test]
2827    fn float_properties() {
2828        let v = SqliteValue::Float(3.14);
2829        assert_eq!(v.as_float(), Some(3.14));
2830        assert_eq!(v.to_integer(), 3);
2831        assert_eq!(v.to_text(), "3.14");
2832    }
2833
2834    #[test]
2835    fn text_properties() {
2836        let v = SqliteValue::Text(SmallText::new("hello"));
2837        assert_eq!(v.as_text(), Some("hello"));
2838        assert_eq!(v.to_integer(), 0);
2839        assert_eq!(v.to_float(), 0.0);
2840    }
2841
2842    #[test]
2843    fn text_numeric_coercion() {
2844        let v = SqliteValue::Text(SmallText::new("123"));
2845        assert_eq!(v.to_integer(), 123);
2846        assert_eq!(v.to_float(), 123.0);
2847
2848        let v = SqliteValue::Text(SmallText::new("3.14"));
2849        assert_eq!(v.to_integer(), 3);
2850        assert_eq!(v.to_float(), 3.14);
2851    }
2852
2853    #[test]
2854    fn text_numeric_coercion_ignores_hex_text_prefixes() {
2855        let v = SqliteValue::Text(SmallText::new("0x10"));
2856        assert_eq!(v.to_integer(), 0);
2857        assert_eq!(v.to_float(), 0.0);
2858
2859        let v = SqliteValue::Blob(Arc::from(b"0x10".as_slice()));
2860        assert_eq!(v.to_integer(), 0);
2861        assert_eq!(v.to_float(), 0.0);
2862    }
2863
2864    #[test]
2865    fn sum_numeric_value_preserves_sqlite_integer_text_boundary() {
2866        assert_eq!(
2867            SqliteValue::Text(SmallText::new(" +123 ")).to_sum_numeric_value(),
2868            SqliteValue::Integer(123)
2869        );
2870        assert_eq!(
2871            SqliteValue::Text(SmallText::new("\u{00a0}123")).to_sum_numeric_value(),
2872            SqliteValue::Float(0.0)
2873        );
2874        assert_eq!(
2875            SqliteValue::Text(SmallText::new("123\u{00a0}")).to_sum_numeric_value(),
2876            SqliteValue::Float(123.0)
2877        );
2878        assert_eq!(
2879            SqliteValue::Text(SmallText::new("1.0")).to_sum_numeric_value(),
2880            SqliteValue::Float(1.0)
2881        );
2882        assert_eq!(
2883            SqliteValue::Text(SmallText::new("123abc")).to_sum_numeric_value(),
2884            SqliteValue::Float(123.0)
2885        );
2886        assert_eq!(
2887            SqliteValue::Text(SmallText::new("")).to_sum_numeric_value(),
2888            SqliteValue::Float(0.0)
2889        );
2890        assert_eq!(
2891            SqliteValue::Blob(Arc::from(b"123".as_slice())).to_sum_numeric_value(),
2892            SqliteValue::Float(123.0)
2893        );
2894    }
2895
2896    #[test]
2897    fn test_integer_numeric_type_uses_sqlite_prefix_rules() {
2898        assert!(SqliteValue::Text(SmallText::new("123abc")).is_integer_numeric_type());
2899        assert!(SqliteValue::Blob(Arc::from(b"123a".as_slice())).is_integer_numeric_type());
2900        assert!(!SqliteValue::Text(SmallText::new("1.5e2abc")).is_integer_numeric_type());
2901        assert!(!SqliteValue::Text(SmallText::new("abc")).is_integer_numeric_type());
2902    }
2903
2904    #[test]
2905    fn test_sqlite_value_integer_real_comparison_equal() {
2906        let int_value = SqliteValue::Integer(3);
2907        let real_value = SqliteValue::Float(3.0);
2908        assert_eq!(int_value.partial_cmp(&real_value), Some(Ordering::Equal));
2909        assert_eq!(real_value.partial_cmp(&int_value), Some(Ordering::Equal));
2910    }
2911
2912    #[test]
2913    fn test_sqlite_value_text_to_integer_coercion() {
2914        let text_value = SqliteValue::Text(SmallText::new("123"));
2915        let coerced = text_value.apply_affinity(TypeAffinity::Integer);
2916        assert_eq!(coerced, SqliteValue::Integer(123));
2917    }
2918
2919    #[test]
2920    fn blob_properties() {
2921        let v = SqliteValue::Blob(Arc::from([0xDE, 0xAD].as_slice()));
2922        assert_eq!(v.as_blob(), Some(&[0xDE, 0xAD][..]));
2923        assert_eq!(v.to_integer(), 0);
2924        assert_eq!(v.to_float(), 0.0);
2925        // to_text() interprets blob bytes as UTF-8 (matching CAST(blob AS TEXT)).
2926        // 0xDE 0xAD is valid UTF-8 encoding of U+07AD.
2927        assert_eq!(v.to_text(), "\u{07AD}");
2928    }
2929
2930    #[test]
2931    fn display_formatting() {
2932        assert_eq!(SqliteValue::Null.to_string(), "NULL");
2933        assert_eq!(SqliteValue::Integer(42).to_string(), "42");
2934        assert_eq!(SqliteValue::Integer(-1).to_string(), "-1");
2935        assert_eq!(SqliteValue::Float(1.5).to_string(), "1.5");
2936        assert_eq!(SqliteValue::Text(SmallText::new("hi")).to_string(), "'hi'");
2937        assert_eq!(
2938            SqliteValue::Blob(Arc::from([0xCA, 0xFE].as_slice())).to_string(),
2939            "X'CAFE'"
2940        );
2941    }
2942
2943    #[test]
2944    fn sort_order_null_first() {
2945        let null = SqliteValue::Null;
2946        let int = SqliteValue::Integer(0);
2947        let text = SqliteValue::Text(SmallText::new(""));
2948        let blob = SqliteValue::Blob(Arc::from(&[] as &[u8]));
2949
2950        assert!(null < int);
2951        assert!(int < text);
2952        assert!(text < blob);
2953    }
2954
2955    #[test]
2956    fn sort_order_integers() {
2957        let a = SqliteValue::Integer(1);
2958        let b = SqliteValue::Integer(2);
2959        assert!(a < b);
2960        assert_eq!(a.partial_cmp(&a), Some(Ordering::Equal));
2961    }
2962
2963    #[test]
2964    fn sort_order_mixed_numeric() {
2965        let int = SqliteValue::Integer(1);
2966        let float = SqliteValue::Float(1.5);
2967        assert!(int < float);
2968
2969        let int = SqliteValue::Integer(2);
2970        assert!(int > float);
2971    }
2972
2973    #[test]
2974    fn test_int_float_precision_at_i64_boundary() {
2975        // i64::MAX cast to f64 rounds UP to 9223372036854775808.0.
2976        // The naive (i as f64) comparison would say Equal, but C SQLite
2977        // correctly reports i64::MAX < 9223372036854775808.0.
2978        let imax = SqliteValue::Integer(i64::MAX);
2979        let fmax = SqliteValue::Float(9_223_372_036_854_775_808.0);
2980        assert_eq!(
2981            imax.partial_cmp(&fmax),
2982            Some(Ordering::Less),
2983            "i64::MAX must be Less than 9223372036854775808.0"
2984        );
2985
2986        // Two distinct large integers that map to the same f64.
2987        let a = SqliteValue::Integer(i64::MAX);
2988        let b = SqliteValue::Integer(i64::MAX - 1);
2989        let f = SqliteValue::Float(i64::MAX as f64);
2990        // a > b, but both should compare consistently vs the float.
2991        assert_eq!(a.partial_cmp(&b), Some(Ordering::Greater));
2992        // Both are less than the rounded-up float.
2993        assert_eq!(a.partial_cmp(&f), Some(Ordering::Less));
2994        assert_eq!(b.partial_cmp(&f), Some(Ordering::Less));
2995    }
2996
2997    #[test]
2998    fn test_int_float_precision_symmetric() {
2999        // Float-vs-Integer should be the reverse of Integer-vs-Float.
3000        let i = SqliteValue::Integer(i64::MAX);
3001        let f = SqliteValue::Float(9_223_372_036_854_775_808.0);
3002        assert_eq!(f.partial_cmp(&i), Some(Ordering::Greater));
3003    }
3004
3005    #[test]
3006    fn test_int_float_exact_representation() {
3007        // For exactly representable values, equality still works.
3008        let i = SqliteValue::Integer(42);
3009        let f = SqliteValue::Float(42.0);
3010        assert_eq!(i.partial_cmp(&f), Some(Ordering::Equal));
3011        assert_eq!(f.partial_cmp(&i), Some(Ordering::Equal));
3012
3013        // Integer 3 vs Float 3.5 — Integer is less.
3014        let i = SqliteValue::Integer(3);
3015        let f = SqliteValue::Float(3.5);
3016        assert_eq!(i.partial_cmp(&f), Some(Ordering::Less));
3017        assert_eq!(f.partial_cmp(&i), Some(Ordering::Greater));
3018    }
3019
3020    #[test]
3021    fn from_conversions() {
3022        assert_eq!(SqliteValue::from(42i64).as_integer(), Some(42));
3023        assert_eq!(SqliteValue::from(42i32).as_integer(), Some(42));
3024        assert_eq!(SqliteValue::from(1.5f64).as_float(), Some(1.5));
3025        assert_eq!(SqliteValue::from("hello").as_text(), Some("hello"));
3026        assert_eq!(
3027            SqliteValue::from(String::from("world")).as_text(),
3028            Some("world")
3029        );
3030        assert_eq!(SqliteValue::from(vec![1u8, 2]).as_blob(), Some(&[1, 2][..]));
3031        assert!(SqliteValue::from(None::<i64>).is_null());
3032        assert_eq!(SqliteValue::from(Some(42i64)).as_integer(), Some(42));
3033    }
3034
3035    #[test]
3036    fn affinity() {
3037        assert_eq!(SqliteValue::Null.affinity(), TypeAffinity::Blob);
3038        assert_eq!(SqliteValue::Integer(0).affinity(), TypeAffinity::Integer);
3039        assert_eq!(SqliteValue::Float(0.0).affinity(), TypeAffinity::Real);
3040        assert_eq!(
3041            SqliteValue::Text(SmallText::new("")).affinity(),
3042            TypeAffinity::Text
3043        );
3044        assert_eq!(
3045            SqliteValue::Blob(Arc::from(&[] as &[u8])).affinity(),
3046            TypeAffinity::Blob
3047        );
3048    }
3049
3050    #[test]
3051    fn null_equality() {
3052        // In SQLite, NULL == NULL is false, but for sorting they are equal
3053        let a = SqliteValue::Null;
3054        let b = SqliteValue::Null;
3055        assert_eq!(a.partial_cmp(&b), Some(Ordering::Equal));
3056    }
3057
3058    // ── bd-13r.1: Type Affinity Advisory + STRICT Enforcement ──
3059
3060    #[test]
3061    fn test_storage_class_variants() {
3062        assert_eq!(SqliteValue::Null.storage_class(), StorageClass::Null);
3063        assert_eq!(
3064            SqliteValue::Integer(42).storage_class(),
3065            StorageClass::Integer
3066        );
3067        assert_eq!(SqliteValue::Float(3.14).storage_class(), StorageClass::Real);
3068        assert_eq!(
3069            SqliteValue::Text("hi".into()).storage_class(),
3070            StorageClass::Text
3071        );
3072        assert_eq!(
3073            SqliteValue::Blob(Arc::from([1u8].as_slice())).storage_class(),
3074            StorageClass::Blob
3075        );
3076    }
3077
3078    #[test]
3079    fn test_type_affinity_advisory_text_into_integer_ok() {
3080        // INSERT TEXT "hello" into INTEGER-affinity column: text stays as text
3081        // (not a well-formed numeric literal).
3082        let val = SqliteValue::Text("hello".into());
3083        let coerced = val.apply_affinity(TypeAffinity::Integer);
3084        assert!(coerced.as_text().is_some());
3085        assert_eq!(coerced.as_text().unwrap(), "hello");
3086
3087        // INSERT TEXT "42" into INTEGER-affinity column: coerced to integer.
3088        let val = SqliteValue::Text("42".into());
3089        let coerced = val.apply_affinity(TypeAffinity::Integer);
3090        assert_eq!(coerced.as_integer(), Some(42));
3091    }
3092
3093    #[test]
3094    fn test_type_affinity_advisory_integer_into_text_ok() {
3095        // INSERT INTEGER 42 into TEXT-affinity column: coerced to text "42".
3096        let val = SqliteValue::Integer(42);
3097        let coerced = val.apply_affinity(TypeAffinity::Text);
3098        assert_eq!(coerced.as_text(), Some("42"));
3099    }
3100
3101    #[test]
3102    fn test_type_affinity_comparison_coercion_matches_oracle() {
3103        // NUMERIC affinity coerces text "123" to integer.
3104        let val = SqliteValue::Text("123".into());
3105        let coerced = val.apply_affinity(TypeAffinity::Numeric);
3106        assert_eq!(coerced.as_integer(), Some(123));
3107
3108        // NUMERIC affinity coerces text "3.14" to real.
3109        let val = SqliteValue::Text("3.14".into());
3110        let coerced = val.apply_affinity(TypeAffinity::Numeric);
3111        assert_eq!(coerced.as_float(), Some(3.14));
3112
3113        // NUMERIC affinity leaves text "hello" as text.
3114        let val = SqliteValue::Text("hello".into());
3115        let coerced = val.apply_affinity(TypeAffinity::Numeric);
3116        assert!(coerced.as_text().is_some());
3117
3118        // BLOB affinity never converts anything.
3119        let val = SqliteValue::Integer(42);
3120        let coerced = val.apply_affinity(TypeAffinity::Blob);
3121        assert_eq!(coerced.as_integer(), Some(42));
3122
3123        // INTEGER affinity converts exact-integer floats to integer.
3124        let val = SqliteValue::Float(5.0);
3125        let coerced = val.apply_affinity(TypeAffinity::Integer);
3126        assert_eq!(coerced.as_integer(), Some(5));
3127
3128        // INTEGER affinity keeps non-exact floats as float.
3129        let val = SqliteValue::Float(5.5);
3130        let coerced = val.apply_affinity(TypeAffinity::Integer);
3131        assert_eq!(coerced.as_float(), Some(5.5));
3132
3133        // REAL affinity forces integers to float.
3134        let val = SqliteValue::Integer(7);
3135        let coerced = val.apply_affinity(TypeAffinity::Real);
3136        assert_eq!(coerced.as_float(), Some(7.0));
3137
3138        // REAL affinity coerces text "9" to float 9.0.
3139        let val = SqliteValue::Text("9".into());
3140        let coerced = val.apply_affinity(TypeAffinity::Real);
3141        assert_eq!(coerced.as_float(), Some(9.0));
3142    }
3143
3144    #[test]
3145    fn test_cast_to_numeric_uses_sqlite_cast_rules() {
3146        assert_eq!(
3147            SqliteValue::Text(SmallText::new("123abc")).cast_to_numeric(),
3148            SqliteValue::Integer(123)
3149        );
3150        assert_eq!(
3151            SqliteValue::Text(SmallText::new("1.5e2abc")).cast_to_numeric(),
3152            SqliteValue::Integer(150)
3153        );
3154        assert_eq!(
3155            SqliteValue::Text(SmallText::new("abc")).cast_to_numeric(),
3156            SqliteValue::Integer(0)
3157        );
3158        assert_eq!(
3159            SqliteValue::Blob(Arc::from(b"123a".as_slice())).cast_to_numeric(),
3160            SqliteValue::Integer(123)
3161        );
3162
3163        match SqliteValue::Text(SmallText::new("1e999")).cast_to_numeric() {
3164            SqliteValue::Float(value) => assert!(value.is_infinite() && value.is_sign_positive()),
3165            other => panic!("expected +inf REAL from NUMERIC cast, got {other:?}"),
3166        }
3167    }
3168
3169    #[test]
3170    fn test_strict_table_rejects_text_into_integer() {
3171        let val = SqliteValue::Text("hello".into());
3172        let result = val.validate_strict(StrictColumnType::Integer);
3173        assert!(result.is_err());
3174        let err = result.unwrap_err();
3175        assert_eq!(err.expected, StrictColumnType::Integer);
3176        assert_eq!(err.actual, StorageClass::Text);
3177    }
3178
3179    #[test]
3180    fn test_strict_table_allows_exact_type() {
3181        // INTEGER into INTEGER column: ok.
3182        let val = SqliteValue::Integer(42);
3183        assert!(val.validate_strict(StrictColumnType::Integer).is_ok());
3184
3185        // REAL into REAL column: ok.
3186        let val = SqliteValue::Float(3.14);
3187        assert!(val.validate_strict(StrictColumnType::Real).is_ok());
3188
3189        // TEXT into TEXT column: ok.
3190        let val = SqliteValue::Text("hello".into());
3191        assert!(val.validate_strict(StrictColumnType::Text).is_ok());
3192
3193        // BLOB into BLOB column: ok.
3194        let val = SqliteValue::Blob(Arc::from([1u8, 2, 3].as_slice()));
3195        assert!(val.validate_strict(StrictColumnType::Blob).is_ok());
3196
3197        // NULL into any STRICT column: ok (nullability enforced separately).
3198        assert!(
3199            SqliteValue::Null
3200                .validate_strict(StrictColumnType::Integer)
3201                .is_ok()
3202        );
3203        assert!(
3204            SqliteValue::Null
3205                .validate_strict(StrictColumnType::Text)
3206                .is_ok()
3207        );
3208
3209        // ANY accepts everything.
3210        let val = SqliteValue::Integer(42);
3211        assert!(val.validate_strict(StrictColumnType::Any).is_ok());
3212        let val = SqliteValue::Text("hi".into());
3213        assert!(val.validate_strict(StrictColumnType::Any).is_ok());
3214    }
3215
3216    #[test]
3217    fn test_strict_real_accepts_integer_with_coercion() {
3218        // STRICT REAL column accepts INTEGER and coerces to float.
3219        let val = SqliteValue::Integer(42);
3220        let result = val.validate_strict(StrictColumnType::Real).unwrap();
3221        assert_eq!(result.as_float(), Some(42.0));
3222    }
3223
3224    #[test]
3225    fn test_strict_rejects_wrong_storage_classes() {
3226        // REAL into INTEGER column: rejected.
3227        assert!(
3228            SqliteValue::Float(3.14)
3229                .validate_strict(StrictColumnType::Integer)
3230                .is_err()
3231        );
3232
3233        // BLOB into TEXT column: rejected.
3234        assert!(
3235            SqliteValue::Blob(Arc::from([1u8].as_slice()))
3236                .validate_strict(StrictColumnType::Text)
3237                .is_err()
3238        );
3239
3240        // INTEGER into TEXT column: rejected.
3241        assert!(
3242            SqliteValue::Integer(1)
3243                .validate_strict(StrictColumnType::Text)
3244                .is_err()
3245        );
3246
3247        // TEXT into BLOB column: rejected.
3248        assert!(
3249            SqliteValue::Text("x".into())
3250                .validate_strict(StrictColumnType::Blob)
3251                .is_err()
3252        );
3253    }
3254
3255    #[test]
3256    fn test_strict_column_type_parsing() {
3257        assert_eq!(
3258            StrictColumnType::from_type_name("INT"),
3259            Some(StrictColumnType::Integer)
3260        );
3261        assert_eq!(
3262            StrictColumnType::from_type_name("INTEGER"),
3263            Some(StrictColumnType::Integer)
3264        );
3265        assert_eq!(
3266            StrictColumnType::from_type_name("REAL"),
3267            Some(StrictColumnType::Real)
3268        );
3269        assert_eq!(
3270            StrictColumnType::from_type_name("TEXT"),
3271            Some(StrictColumnType::Text)
3272        );
3273        assert_eq!(
3274            StrictColumnType::from_type_name("BLOB"),
3275            Some(StrictColumnType::Blob)
3276        );
3277        assert_eq!(
3278            StrictColumnType::from_type_name("ANY"),
3279            Some(StrictColumnType::Any)
3280        );
3281        // Invalid type name in STRICT mode.
3282        assert_eq!(StrictColumnType::from_type_name("VARCHAR(255)"), None);
3283        assert_eq!(StrictColumnType::from_type_name("NUMERIC"), None);
3284    }
3285
3286    #[test]
3287    fn test_affinity_advisory_never_rejects() {
3288        // Advisory affinity NEVER rejects a value. All combinations must succeed.
3289        let values = vec![
3290            SqliteValue::Null,
3291            SqliteValue::Integer(42),
3292            SqliteValue::Float(3.14),
3293            SqliteValue::Text("hello".into()),
3294            SqliteValue::Blob(Arc::from([0xDE, 0xAD].as_slice())),
3295        ];
3296        let affinities = [
3297            TypeAffinity::Integer,
3298            TypeAffinity::Text,
3299            TypeAffinity::Blob,
3300            TypeAffinity::Real,
3301            TypeAffinity::Numeric,
3302        ];
3303        for val in &values {
3304            for aff in &affinities {
3305                // apply_affinity is infallible - it always returns a value.
3306                let _ = val.clone().apply_affinity(*aff);
3307            }
3308        }
3309    }
3310
3311    // ── bd-13r.2: UNIQUE NULL Semantics (NULL != NULL) ──
3312
3313    #[test]
3314    fn test_unique_allows_multiple_nulls_single_column() {
3315        // In UNIQUE columns, NULL != NULL: two NULLs are never duplicates.
3316        let a = SqliteValue::Null;
3317        let b = SqliteValue::Null;
3318        assert!(!a.unique_eq(&b));
3319    }
3320
3321    #[test]
3322    fn test_unique_allows_multiple_nulls_multi_column_partial_null() {
3323        // UNIQUE(a,b): (NULL,1) and (NULL,1) are NOT duplicates because
3324        // any NULL component makes the whole key non-duplicate.
3325        let row_a = [SqliteValue::Null, SqliteValue::Integer(1)];
3326        let row_b = [SqliteValue::Null, SqliteValue::Integer(1)];
3327        assert!(!unique_key_duplicates(&row_a, &row_b));
3328
3329        // UNIQUE(a,b): (1,NULL) and (1,NULL) are NOT duplicates.
3330        let row_a = [SqliteValue::Integer(1), SqliteValue::Null];
3331        let row_b = [SqliteValue::Integer(1), SqliteValue::Null];
3332        assert!(!unique_key_duplicates(&row_a, &row_b));
3333
3334        // UNIQUE(a,b): (NULL,NULL) and (NULL,NULL) are NOT duplicates.
3335        let row_a = [SqliteValue::Null, SqliteValue::Null];
3336        let row_b = [SqliteValue::Null, SqliteValue::Null];
3337        assert!(!unique_key_duplicates(&row_a, &row_b));
3338    }
3339
3340    #[test]
3341    fn test_unique_rejects_duplicate_non_null() {
3342        // Two identical non-NULL values ARE duplicates.
3343        let a = SqliteValue::Integer(42);
3344        let b = SqliteValue::Integer(42);
3345        assert!(a.unique_eq(&b));
3346
3347        // Composite: (1, "hello") and (1, "hello") ARE duplicates.
3348        let row_a = [SqliteValue::Integer(1), SqliteValue::Text("hello".into())];
3349        let row_b = [SqliteValue::Integer(1), SqliteValue::Text("hello".into())];
3350        assert!(unique_key_duplicates(&row_a, &row_b));
3351
3352        // Different values are NOT duplicates.
3353        let row_a = [SqliteValue::Integer(1), SqliteValue::Text("hello".into())];
3354        let row_b = [SqliteValue::Integer(1), SqliteValue::Text("world".into())];
3355        assert!(!unique_key_duplicates(&row_a, &row_b));
3356    }
3357
3358    #[test]
3359    fn test_unique_null_vs_non_null_distinct() {
3360        // NULL and a non-NULL value are never duplicates.
3361        let a = SqliteValue::Null;
3362        let b = SqliteValue::Integer(1);
3363        assert!(!a.unique_eq(&b));
3364        assert!(!b.unique_eq(&a));
3365
3366        // Composite: (NULL, 1) and (2, 1) are not duplicates (different first element).
3367        let row_a = [SqliteValue::Null, SqliteValue::Integer(1)];
3368        let row_b = [SqliteValue::Integer(2), SqliteValue::Integer(1)];
3369        assert!(!unique_key_duplicates(&row_a, &row_b));
3370    }
3371
3372    // ── bd-13r.4: Integer Overflow Semantics (Expr vs sum()) ──
3373
3374    #[test]
3375    #[allow(clippy::cast_precision_loss)]
3376    fn test_integer_overflow_promotes_real_expr_add() {
3377        let max = SqliteValue::Integer(i64::MAX);
3378        let one = SqliteValue::Integer(1);
3379        let result = max.sql_add(&one);
3380        // Overflow promotes to REAL (not integer).
3381        assert!(result.as_integer().is_none());
3382        assert!(result.as_float().is_some());
3383        // The float value is approximately i64::MAX + 1.
3384        assert!(result.as_float().unwrap() >= i64::MAX as f64);
3385    }
3386
3387    #[test]
3388    fn test_integer_overflow_promotes_real_expr_mul() {
3389        let max = SqliteValue::Integer(i64::MAX);
3390        let two = SqliteValue::Integer(2);
3391        let result = max.sql_mul(&two);
3392        // Overflow promotes to REAL.
3393        assert!(result.as_float().is_some());
3394    }
3395
3396    #[test]
3397    fn test_integer_overflow_promotes_real_expr_sub() {
3398        let min = SqliteValue::Integer(i64::MIN);
3399        let one = SqliteValue::Integer(1);
3400        let result = min.sql_sub(&one);
3401        // Underflow promotes to REAL.
3402        assert!(result.as_float().is_some());
3403    }
3404
3405    #[test]
3406    fn test_sum_overflow_errors() {
3407        let mut acc = SumAccumulator::new();
3408        acc.accumulate(&SqliteValue::Integer(i64::MAX));
3409        acc.accumulate(&SqliteValue::Integer(1));
3410        let result = acc.finish();
3411        assert!(result.is_err());
3412    }
3413
3414    #[test]
3415    fn test_sum_overflow_then_float_returns_real() {
3416        let mut acc = SumAccumulator::new();
3417        acc.accumulate(&SqliteValue::Integer(i64::MAX));
3418        acc.accumulate(&SqliteValue::Integer(1));
3419        acc.accumulate(&SqliteValue::Float(0.5));
3420        let result = acc.finish().unwrap();
3421        assert!(matches!(result, SqliteValue::Float(_)));
3422    }
3423
3424    #[test]
3425    fn test_sum_text_integer_literals_stay_integer() {
3426        let mut acc = SumAccumulator::new();
3427        acc.accumulate(&SqliteValue::Text(SmallText::new("1")));
3428        acc.accumulate(&SqliteValue::Text(SmallText::new("2")));
3429        let result = acc.finish().unwrap();
3430        assert_eq!(result.as_integer(), Some(3));
3431    }
3432
3433    #[test]
3434    fn test_sum_non_numeric_text_returns_real_zero() {
3435        let mut acc = SumAccumulator::new();
3436        acc.accumulate(&SqliteValue::Text(SmallText::new("abc")));
3437        let result = acc.finish().unwrap();
3438        assert_eq!(result.as_float(), Some(0.0));
3439    }
3440
3441    #[test]
3442    fn test_no_overflow_stays_integer() {
3443        // Non-overflow addition stays INTEGER.
3444        let a = SqliteValue::Integer(100);
3445        let b = SqliteValue::Integer(200);
3446        let result = a.sql_add(&b);
3447        assert_eq!(result.as_integer(), Some(300));
3448
3449        // Non-overflow multiplication stays INTEGER.
3450        let result = SqliteValue::Integer(7).sql_mul(&SqliteValue::Integer(6));
3451        assert_eq!(result.as_integer(), Some(42));
3452
3453        // Non-overflow subtraction stays INTEGER.
3454        let result = SqliteValue::Integer(50).sql_sub(&SqliteValue::Integer(8));
3455        assert_eq!(result.as_integer(), Some(42));
3456    }
3457
3458    #[test]
3459    fn test_sum_null_only_returns_null() {
3460        let mut acc = SumAccumulator::new();
3461        acc.accumulate(&SqliteValue::Null);
3462        acc.accumulate(&SqliteValue::Null);
3463        let result = acc.finish().unwrap();
3464        assert!(result.is_null());
3465    }
3466
3467    #[test]
3468    fn test_sum_mixed_int_float() {
3469        let mut acc = SumAccumulator::new();
3470        acc.accumulate(&SqliteValue::Integer(10));
3471        acc.accumulate(&SqliteValue::Float(2.5));
3472        acc.accumulate(&SqliteValue::Integer(3));
3473        let result = acc.finish().unwrap();
3474        // Once float is seen, result is float.
3475        assert_eq!(result.as_float(), Some(15.5));
3476    }
3477
3478    #[test]
3479    fn test_sum_integer_only() {
3480        let mut acc = SumAccumulator::new();
3481        acc.accumulate(&SqliteValue::Integer(10));
3482        acc.accumulate(&SqliteValue::Integer(20));
3483        acc.accumulate(&SqliteValue::Integer(30));
3484        let result = acc.finish().unwrap();
3485        assert_eq!(result.as_integer(), Some(60));
3486    }
3487
3488    #[test]
3489    fn test_sql_arithmetic_null_propagation() {
3490        let n = SqliteValue::Null;
3491        let i = SqliteValue::Integer(42);
3492        assert!(n.sql_add(&i).is_null());
3493        assert!(i.sql_add(&n).is_null());
3494        assert!(n.sql_sub(&i).is_null());
3495        assert!(n.sql_mul(&i).is_null());
3496    }
3497
3498    #[test]
3499    fn test_sql_inf_arithmetic_nan_normalized_to_null() {
3500        // +Inf + (-Inf) is NaN in IEEE-754 and must be normalized to NULL.
3501        let pos_inf = SqliteValue::Float(f64::INFINITY);
3502        let neg_inf = SqliteValue::Float(f64::NEG_INFINITY);
3503        assert!(pos_inf.sql_add(&neg_inf).is_null());
3504
3505        // +Inf - +Inf is also NaN and must normalize to NULL.
3506        assert!(pos_inf.sql_sub(&pos_inf).is_null());
3507    }
3508
3509    #[test]
3510    fn test_sql_mul_zero_times_inf_normalized_to_null() {
3511        // 0 * +Inf is NaN in IEEE-754 and must be normalized to NULL.
3512        let zero = SqliteValue::Float(0.0);
3513        let pos_inf = SqliteValue::Float(f64::INFINITY);
3514        assert!(zero.sql_mul(&pos_inf).is_null());
3515        assert!(
3516            SqliteValue::Integer(0).sql_mul(&pos_inf).is_null(),
3517            "mixed INTEGER/REAL multiplication should preserve NaN-to-NULL semantics"
3518        );
3519    }
3520
3521    #[test]
3522    fn test_sql_mul_mixed_int_float_stays_real() {
3523        let left = SqliteValue::Integer(10);
3524        let right = SqliteValue::Float(0.25);
3525        assert_eq!(left.sql_mul(&right).as_float(), Some(2.5));
3526        assert_eq!(right.sql_mul(&left).as_float(), Some(2.5));
3527    }
3528
3529    #[test]
3530    fn test_sql_inf_propagates_when_not_nan() {
3531        let pos_inf = SqliteValue::Float(f64::INFINITY);
3532        let one = SqliteValue::Integer(1);
3533        let add_result = pos_inf.sql_add(&one);
3534        assert!(
3535            matches!(add_result, SqliteValue::Float(v) if v.is_infinite() && v.is_sign_positive()),
3536            "expected +Inf propagation, got {add_result:?}"
3537        );
3538
3539        let neg_inf = SqliteValue::Float(f64::NEG_INFINITY);
3540        let sub_result = neg_inf.sql_sub(&one);
3541        assert!(
3542            matches!(sub_result, SqliteValue::Float(v) if v.is_infinite() && v.is_sign_negative()),
3543            "expected -Inf propagation, got {sub_result:?}"
3544        );
3545    }
3546
3547    #[test]
3548    fn test_from_f64_nan_normalizes_to_null() {
3549        let value = SqliteValue::from(f64::NAN);
3550        assert!(value.is_null());
3551    }
3552
3553    #[test]
3554    fn test_inf_comparisons_against_finite_values() {
3555        let pos_inf = SqliteValue::Float(f64::INFINITY);
3556        let neg_inf = SqliteValue::Float(f64::NEG_INFINITY);
3557        let finite_hi = SqliteValue::Float(1.0e308);
3558        let finite_lo = SqliteValue::Float(-1.0e308);
3559
3560        assert_eq!(pos_inf.partial_cmp(&finite_hi), Some(Ordering::Greater));
3561        assert_eq!(neg_inf.partial_cmp(&finite_lo), Some(Ordering::Less));
3562    }
3563
3564    // ── bd-13r.7: Empty String vs NULL Semantics ──
3565
3566    #[test]
3567    fn test_empty_string_is_not_null() {
3568        let empty = SqliteValue::Text(SmallText::new(""));
3569        // '' IS NULL → false.
3570        assert!(!empty.is_null());
3571        // '' IS NOT NULL → true (expressed as !is_null).
3572        assert!(!empty.is_null());
3573        // NULL IS NULL → true.
3574        assert!(SqliteValue::Null.is_null());
3575    }
3576
3577    #[test]
3578    fn test_length_empty_string_zero() {
3579        let empty = SqliteValue::Text(SmallText::new(""));
3580        assert_eq!(empty.sql_length(), Some(0));
3581    }
3582
3583    #[test]
3584    fn test_typeof_empty_string_text() {
3585        let empty = SqliteValue::Text(SmallText::new(""));
3586        assert_eq!(empty.typeof_str(), "text");
3587        // NULL has typeof "null".
3588        assert_eq!(SqliteValue::Null.typeof_str(), "null");
3589    }
3590
3591    #[test]
3592    fn test_empty_string_comparisons() {
3593        let empty1 = SqliteValue::Text(SmallText::new(""));
3594        let empty2 = SqliteValue::Text(SmallText::new(""));
3595        // '' = '' → true.
3596        assert_eq!(empty1.partial_cmp(&empty2), Some(std::cmp::Ordering::Equal));
3597
3598        // '' = NULL → NULL (comparison with NULL yields None/unknown).
3599        // In our PartialOrd, NULL and TEXT are different sort classes,
3600        // so NULL < TEXT (they are not equal).
3601        let null = SqliteValue::Null;
3602        assert_ne!(empty1.partial_cmp(&null), Some(std::cmp::Ordering::Equal));
3603    }
3604
3605    #[test]
3606    fn test_typeof_all_variants() {
3607        assert_eq!(SqliteValue::Null.typeof_str(), "null");
3608        assert_eq!(SqliteValue::Integer(0).typeof_str(), "integer");
3609        assert_eq!(SqliteValue::Float(0.0).typeof_str(), "real");
3610        assert_eq!(SqliteValue::Text("x".into()).typeof_str(), "text");
3611        assert_eq!(
3612            SqliteValue::Blob(Arc::from(&[] as &[u8])).typeof_str(),
3613            "blob"
3614        );
3615    }
3616
3617    #[test]
3618    fn test_sql_length_all_types() {
3619        // NULL → NULL (None).
3620        assert_eq!(SqliteValue::Null.sql_length(), None);
3621        // TEXT → character count.
3622        assert_eq!(SqliteValue::Text("hello".into()).sql_length(), Some(5));
3623        assert_eq!(SqliteValue::Text(SmallText::new("")).sql_length(), Some(0));
3624        // BLOB → byte count.
3625        assert_eq!(
3626            SqliteValue::Blob(Arc::from([1u8, 2, 3].as_slice())).sql_length(),
3627            Some(3)
3628        );
3629        // INTEGER → length of text representation.
3630        assert_eq!(SqliteValue::Integer(42).sql_length(), Some(2));
3631        // REAL → length of text representation.
3632        assert_eq!(SqliteValue::Float(3.14).sql_length(), Some(4)); // "3.14"
3633    }
3634
3635    // ── bd-13r.6: LIKE Semantics (ASCII-only case folding) ──
3636
3637    #[test]
3638    fn test_like_ascii_case_insensitive() {
3639        assert!(sql_like("A", "a", None));
3640        assert!(sql_like("a", "A", None));
3641        assert!(sql_like("hello", "HELLO", None));
3642        assert!(sql_like("HELLO", "hello", None));
3643        assert!(sql_like("HeLLo", "hEllO", None));
3644    }
3645
3646    #[test]
3647    fn test_like_unicode_case_sensitive_without_icu() {
3648        // Without ICU, Unicode case folding does NOT occur.
3649        assert!(!sql_like("ä", "Ä", None));
3650        assert!(!sql_like("Ä", "ä", None));
3651        // But exact match works.
3652        assert!(sql_like("ä", "ä", None));
3653    }
3654
3655    #[test]
3656    fn test_like_fast_path_does_not_fold_ascii_punctuation() {
3657        assert!(!sql_like("[", "{", None));
3658        assert!(!sql_like("@", "`", None));
3659    }
3660
3661    #[test]
3662    fn test_like_escape_handling() {
3663        // Escape literal % with backslash.
3664        assert!(sql_like("100\\%", "100%", Some('\\')));
3665        assert!(!sql_like("100\\%", "100x", Some('\\')));
3666
3667        // Escape literal _.
3668        assert!(sql_like("a\\_b", "a_b", Some('\\')));
3669        assert!(!sql_like("a\\_b", "axb", Some('\\')));
3670    }
3671
3672    #[test]
3673    fn test_like_wildcards_basic() {
3674        // % matches zero or more characters.
3675        assert!(sql_like("%", "", None));
3676        assert!(sql_like("%", "anything", None));
3677        assert!(sql_like("a%", "abc", None));
3678        assert!(sql_like("%c", "abc", None));
3679        assert!(sql_like("a%c", "abc", None));
3680        assert!(sql_like("a%c", "aXYZc", None));
3681        assert!(!sql_like("a%c", "abd", None));
3682
3683        // _ matches exactly one character.
3684        assert!(sql_like("_", "x", None));
3685        assert!(!sql_like("_", "", None));
3686        assert!(!sql_like("_", "xy", None));
3687        assert!(sql_like("a_c", "abc", None));
3688        assert!(!sql_like("a_c", "abbc", None));
3689    }
3690
3691    #[test]
3692    fn test_like_combined_wildcards() {
3693        assert!(sql_like("%_", "a", None));
3694        assert!(!sql_like("%_", "", None));
3695        assert!(sql_like("_%_", "ab", None));
3696        assert!(!sql_like("_%_", "a", None));
3697        assert!(sql_like("%a%b%", "xaybz", None));
3698        assert!(!sql_like("%a%b%", "xyz", None));
3699    }
3700
3701    #[test]
3702    fn test_like_exact_match() {
3703        assert!(sql_like("hello", "hello", None));
3704        assert!(!sql_like("hello", "world", None));
3705        assert!(sql_like("", "", None));
3706        assert!(!sql_like("a", "", None));
3707        assert!(!sql_like("", "a", None));
3708    }
3709
3710    #[test]
3711    fn test_like_fast_path_repeated_percent_shapes() {
3712        assert!(sql_like("ab%%", "ABcd", None));
3713        assert!(sql_like("%%cd", "abCD", None));
3714        assert!(sql_like("%%bc%%", "xxBCyy", None));
3715        assert!(sql_like("%%%%", "anything", None));
3716    }
3717
3718    #[test]
3719    fn test_like_fast_path_preserves_mixed_unicode_and_ascii_semantics() {
3720        assert!(sql_like("%éL%", "héllo", None));
3721        assert!(!sql_like("%Él%", "héllo", None));
3722        assert!(sql_like("Stra%", "straße", None));
3723    }
3724
3725    #[test]
3726    fn test_like_contains_fast_path_handles_overlapping_matches() {
3727        assert!(sql_like("%ana%", "bananas", None));
3728        assert!(sql_like("%NAN%", "baNanas", None));
3729        assert!(!sql_like("%ananasx%", "bananas", None));
3730    }
3731
3732    #[test]
3733    fn test_like_contains_fast_path_preserves_non_ascii_byte_matching() {
3734        assert!(sql_like("%ß%", "straße", None));
3735        assert!(!sql_like("%SS%", "straße", None));
3736    }
3737
3738    // ── format_sqlite_float ────────────────────────────────────────────
3739
3740    #[test]
3741    fn test_format_sqlite_float_whole_number() {
3742        assert_eq!(format_sqlite_float(120.0), "120.0");
3743        assert_eq!(format_sqlite_float(0.0), "0.0");
3744        assert_eq!(format_sqlite_float(-42.0), "-42.0");
3745        assert_eq!(format_sqlite_float(1.0), "1.0");
3746    }
3747
3748    #[test]
3749    fn test_format_sqlite_float_fractional() {
3750        assert_eq!(format_sqlite_float(3.14), "3.14");
3751        assert_eq!(format_sqlite_float(0.5), "0.5");
3752        assert_eq!(format_sqlite_float(-0.001), "-0.001");
3753    }
3754
3755    #[test]
3756    fn test_format_sqlite_float_special() {
3757        assert_eq!(format_sqlite_float(f64::NAN), "NaN");
3758        assert_eq!(format_sqlite_float(f64::INFINITY), "Inf");
3759        assert_eq!(format_sqlite_float(f64::NEG_INFINITY), "-Inf");
3760    }
3761
3762    #[test]
3763    fn test_format_sqlite_float_negative_zero() {
3764        // SQLite CAST(... AS TEXT) normalizes both zero signs to "0.0".
3765        assert_eq!(format_sqlite_float(-0.0), "0.0");
3766        assert_eq!(format_sqlite_float(0.0), "0.0");
3767    }
3768
3769    #[test]
3770    fn test_format_sqlite_float_matches_sqlite_17_digit_text_contract() {
3771        assert_eq!(format_sqlite_float(0.1 + 0.2), "0.30000000000000004");
3772        assert_eq!(format_sqlite_float(1.0 / 3.0), "0.33333333333333332");
3773        assert_eq!(format_sqlite_float(2.0 / 3.0), "0.66666666666666663");
3774        assert_eq!(format_sqlite_float(1.5e16), "15000000000000000.0");
3775        assert_eq!(
3776            format_sqlite_float(123_456_789_012_345.6),
3777            "123456789012345.59"
3778        );
3779        assert_eq!(format_sqlite_float(1.0e308), "1.0e+308");
3780        assert_eq!(format_sqlite_float(1.0e-308), "1.0e-308");
3781        assert_eq!(
3782            format_sqlite_float(9.223_372_036_854_776e18),
3783            "9.2233720368547758e+18"
3784        );
3785    }
3786
3787    #[test]
3788    fn test_float_to_text_includes_decimal_point() {
3789        let v = SqliteValue::Float(100.0);
3790        assert_eq!(v.to_text(), "100.0");
3791        let v = SqliteValue::Float(3.14);
3792        assert_eq!(v.to_text(), "3.14");
3793    }
3794
3795    // ── scan_numeric_prefix ──────────────────────────────────────────
3796
3797    #[test]
3798    fn test_scan_numeric_prefix_bare_dot() {
3799        // A bare "." has no digits — not a numeric prefix.
3800        assert_eq!(scan_numeric_prefix(b"."), 0);
3801        assert_eq!(scan_numeric_prefix(b"-."), 0);
3802        assert_eq!(scan_numeric_prefix(b"+."), 0);
3803        assert_eq!(scan_numeric_prefix(b"..1"), 0);
3804    }
3805
3806    #[test]
3807    fn test_scan_numeric_prefix_valid() {
3808        assert_eq!(scan_numeric_prefix(b"123"), 3);
3809        assert_eq!(scan_numeric_prefix(b"3.14"), 4);
3810        assert_eq!(scan_numeric_prefix(b".5"), 2);
3811        assert_eq!(scan_numeric_prefix(b"1e10"), 4);
3812        assert_eq!(scan_numeric_prefix(b"-42abc"), 3);
3813        assert_eq!(scan_numeric_prefix(b"+.5x"), 3);
3814        assert_eq!(scan_numeric_prefix(b"0.0"), 3);
3815    }
3816
3817    #[test]
3818    fn test_scan_numeric_prefix_empty_and_non_numeric() {
3819        assert_eq!(scan_numeric_prefix(b""), 0);
3820        assert_eq!(scan_numeric_prefix(b"abc"), 0);
3821        assert_eq!(scan_numeric_prefix(b"+"), 0);
3822        assert_eq!(scan_numeric_prefix(b"-"), 0);
3823    }
3824}