citadel_sql/
types.rs

1use std::cmp::Ordering;
2use std::fmt;
3use std::hash::{Hash, Hasher};
4
5pub use compact_str::CompactString;
6
7use crate::parser::Expr;
8
9/// SQL data types.
10#[derive(Debug, Clone, Copy, PartialEq, Eq)]
11pub enum DataType {
12    Null,
13    Integer,
14    Real,
15    Text,
16    Blob,
17    Boolean,
18    Time,
19    Date,
20    Timestamp,
21    Interval,
22}
23
24impl DataType {
25    pub fn type_tag(self) -> u8 {
26        match self {
27            DataType::Null => 0,
28            DataType::Blob => 1,
29            DataType::Text => 2,
30            DataType::Boolean => 3,
31            DataType::Integer => 4,
32            DataType::Real => 5,
33            DataType::Time => 6,
34            DataType::Date => 7,
35            DataType::Timestamp => 8,
36            DataType::Interval => 9,
37        }
38    }
39
40    pub fn from_tag(tag: u8) -> Option<Self> {
41        match tag {
42            0 => Some(DataType::Null),
43            1 => Some(DataType::Blob),
44            2 => Some(DataType::Text),
45            3 => Some(DataType::Boolean),
46            4 => Some(DataType::Integer),
47            5 => Some(DataType::Real),
48            6 => Some(DataType::Time),
49            7 => Some(DataType::Date),
50            8 => Some(DataType::Timestamp),
51            9 => Some(DataType::Interval),
52            _ => None,
53        }
54    }
55}
56
57impl fmt::Display for DataType {
58    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
59        match self {
60            DataType::Null => write!(f, "NULL"),
61            DataType::Integer => write!(f, "INTEGER"),
62            DataType::Real => write!(f, "REAL"),
63            DataType::Text => write!(f, "TEXT"),
64            DataType::Blob => write!(f, "BLOB"),
65            DataType::Boolean => write!(f, "BOOLEAN"),
66            DataType::Time => write!(f, "TIME"),
67            DataType::Date => write!(f, "DATE"),
68            DataType::Timestamp => write!(f, "TIMESTAMP"),
69            DataType::Interval => write!(f, "INTERVAL"),
70        }
71    }
72}
73
74/// SQL value. Temporal epochs: days/µs since 1970-01-01 UTC.
75/// `Date`/`Timestamp` reserve `i{32,64}::{MAX,MIN}` as `±infinity` sentinels.
76#[derive(Debug, Clone, Default)]
77pub enum Value {
78    #[default]
79    Null,
80    Integer(i64),
81    Real(f64),
82    Text(CompactString),
83    Blob(Vec<u8>),
84    Boolean(bool),
85    Time(i64),
86    Date(i32),
87    Timestamp(i64),
88    Interval {
89        months: i32,
90        days: i32,
91        micros: i64,
92    },
93}
94
95impl Value {
96    pub fn data_type(&self) -> DataType {
97        match self {
98            Value::Null => DataType::Null,
99            Value::Integer(_) => DataType::Integer,
100            Value::Real(_) => DataType::Real,
101            Value::Text(_) => DataType::Text,
102            Value::Blob(_) => DataType::Blob,
103            Value::Boolean(_) => DataType::Boolean,
104            Value::Time(_) => DataType::Time,
105            Value::Date(_) => DataType::Date,
106            Value::Timestamp(_) => DataType::Timestamp,
107            Value::Interval { .. } => DataType::Interval,
108        }
109    }
110
111    pub fn is_null(&self) -> bool {
112        matches!(self, Value::Null)
113    }
114
115    /// Returns true for `±infinity` sentinel values on DATE / TIMESTAMP; false otherwise.
116    pub fn is_finite_temporal(&self) -> bool {
117        match self {
118            Value::Date(d) => *d != i32::MAX && *d != i32::MIN,
119            Value::Timestamp(t) => *t != i64::MAX && *t != i64::MIN,
120            _ => true,
121        }
122    }
123
124    /// Attempt to coerce this value to the target type.
125    pub fn coerce_to(&self, target: DataType) -> Option<Value> {
126        match (self, target) {
127            (_, DataType::Null) => Some(Value::Null),
128            (Value::Null, _) => Some(Value::Null),
129            (Value::Integer(i), DataType::Integer) => Some(Value::Integer(*i)),
130            (Value::Integer(i), DataType::Real) => Some(Value::Real(*i as f64)),
131            (Value::Real(r), DataType::Real) => Some(Value::Real(*r)),
132            (Value::Real(r), DataType::Integer) => Some(Value::Integer(*r as i64)),
133            (Value::Text(s), DataType::Text) => Some(Value::Text(s.clone())),
134            (Value::Blob(b), DataType::Blob) => Some(Value::Blob(b.clone())),
135            (Value::Boolean(b), DataType::Boolean) => Some(Value::Boolean(*b)),
136            (Value::Boolean(b), DataType::Integer) => Some(Value::Integer(if *b { 1 } else { 0 })),
137            (Value::Integer(i), DataType::Boolean) => Some(Value::Boolean(*i != 0)),
138            (Value::Time(t), DataType::Time) => Some(Value::Time(*t)),
139            (Value::Date(d), DataType::Date) => Some(Value::Date(*d)),
140            (Value::Timestamp(t), DataType::Timestamp) => Some(Value::Timestamp(*t)),
141            (
142                Value::Interval {
143                    months,
144                    days,
145                    micros,
146                },
147                DataType::Interval,
148            ) => Some(Value::Interval {
149                months: *months,
150                days: *days,
151                micros: *micros,
152            }),
153            _ => None,
154        }
155    }
156
157    pub fn coerce_into(self, target: DataType) -> Option<Value> {
158        if self.is_null() || target == DataType::Null {
159            return Some(Value::Null);
160        }
161        if self.data_type() == target {
162            return Some(self);
163        }
164        match (self, target) {
165            (Value::Integer(i), DataType::Real) => Some(Value::Real(i as f64)),
166            (Value::Real(r), DataType::Integer) => Some(Value::Integer(r as i64)),
167            (Value::Boolean(b), DataType::Integer) => Some(Value::Integer(if b { 1 } else { 0 })),
168            (Value::Integer(i), DataType::Boolean) => Some(Value::Boolean(i != 0)),
169            (Value::Text(s), DataType::Date) => {
170                crate::datetime::parse_date(&s).ok().map(Value::Date)
171            }
172            (Value::Text(s), DataType::Time) => {
173                crate::datetime::parse_time(&s).ok().map(Value::Time)
174            }
175            (Value::Text(s), DataType::Timestamp) => crate::datetime::parse_timestamp(&s)
176                .ok()
177                .map(Value::Timestamp),
178            (Value::Text(s), DataType::Interval) => {
179                crate::datetime::parse_interval(&s)
180                    .ok()
181                    .map(|(m, d, u)| Value::Interval {
182                        months: m,
183                        days: d,
184                        micros: u,
185                    })
186            }
187            // INTEGER → TIMESTAMP: Unix epoch seconds.
188            (Value::Integer(n), DataType::Timestamp) => {
189                n.checked_mul(1_000_000).map(Value::Timestamp)
190            }
191            (Value::Integer(n), DataType::Date) => {
192                if n >= i32::MIN as i64 && n <= i32::MAX as i64 {
193                    Some(Value::Date(n as i32))
194                } else {
195                    None
196                }
197            }
198            (Value::Integer(n), DataType::Time) => {
199                if (0..=86_400_000_000).contains(&n) {
200                    Some(Value::Time(n))
201                } else {
202                    None
203                }
204            }
205            (Value::Integer(n), DataType::Interval) => {
206                if n >= i32::MIN as i64 && n <= i32::MAX as i64 {
207                    Some(Value::Interval {
208                        months: 0,
209                        days: n as i32,
210                        micros: 0,
211                    })
212                } else {
213                    None
214                }
215            }
216            (Value::Timestamp(t), DataType::Integer) => Some(Value::Integer(t / 1_000_000)),
217            (Value::Date(d), DataType::Integer) => Some(Value::Integer(d as i64)),
218            (Value::Time(t), DataType::Integer) => Some(Value::Integer(t)),
219            (Value::Date(d), DataType::Timestamp) => {
220                (d as i64).checked_mul(86_400_000_000).map(Value::Timestamp)
221            }
222            (Value::Timestamp(t), DataType::Date) => {
223                // div_euclid floors correctly for negative µs (pre-1970).
224                let days = t.div_euclid(86_400_000_000);
225                if days >= i32::MIN as i64 && days <= i32::MAX as i64 {
226                    Some(Value::Date(days as i32))
227                } else {
228                    None
229                }
230            }
231            (v, DataType::Text)
232                if matches!(
233                    v.data_type(),
234                    DataType::Date | DataType::Time | DataType::Timestamp | DataType::Interval
235                ) =>
236            {
237                Some(Value::Text(v.to_string().into()))
238            }
239            _ => None,
240        }
241    }
242
243    /// Numeric ordering for Integer and Real values (promotes to f64 for mixed).
244    fn numeric_cmp(&self, other: &Value) -> Option<Ordering> {
245        match (self, other) {
246            (Value::Integer(a), Value::Integer(b)) => Some(a.cmp(b)),
247            (Value::Real(a), Value::Real(b)) => a.partial_cmp(b),
248            (Value::Integer(a), Value::Real(b)) => (*a as f64).partial_cmp(b),
249            (Value::Real(a), Value::Integer(b)) => a.partial_cmp(&(*b as f64)),
250            _ => None,
251        }
252    }
253}
254
255impl PartialEq for Value {
256    // Field-wise for Eq/Hash/Ord transitivity. SQL-level `=` on INTERVAL
257    // normalizes separately (see eval.rs).
258    fn eq(&self, other: &Self) -> bool {
259        match (self, other) {
260            (Value::Null, Value::Null) => true,
261            (Value::Integer(a), Value::Integer(b)) => a == b,
262            (Value::Real(a), Value::Real(b)) => a == b,
263            (Value::Integer(a), Value::Real(b)) => (*a as f64) == *b,
264            (Value::Real(a), Value::Integer(b)) => *a == (*b as f64),
265            (Value::Text(a), Value::Text(b)) => a == b,
266            (Value::Blob(a), Value::Blob(b)) => a == b,
267            (Value::Boolean(a), Value::Boolean(b)) => a == b,
268            (Value::Time(a), Value::Time(b)) => a == b,
269            (Value::Date(a), Value::Date(b)) => a == b,
270            (Value::Timestamp(a), Value::Timestamp(b)) => a == b,
271            (
272                Value::Interval {
273                    months: am,
274                    days: ad,
275                    micros: au,
276                },
277                Value::Interval {
278                    months: bm,
279                    days: bd,
280                    micros: bu,
281                },
282            ) => am == bm && ad == bd && au == bu,
283            _ => false,
284        }
285    }
286}
287
288impl Eq for Value {}
289
290impl Hash for Value {
291    fn hash<H: Hasher>(&self, state: &mut H) {
292        match self {
293            Value::Null => 0u8.hash(state),
294            Value::Integer(i) => {
295                // Hash via f64 bits so Integer(n) and Real(n.0) produce the same hash,
296                // matching the cross-type PartialEq contract.
297                1u8.hash(state);
298                (*i as f64).to_bits().hash(state);
299            }
300            Value::Real(r) => {
301                1u8.hash(state);
302                r.to_bits().hash(state);
303            }
304            Value::Text(s) => {
305                2u8.hash(state);
306                s.hash(state);
307            }
308            Value::Blob(b) => {
309                3u8.hash(state);
310                b.hash(state);
311            }
312            Value::Boolean(b) => {
313                4u8.hash(state);
314                b.hash(state);
315            }
316            Value::Time(t) => {
317                5u8.hash(state);
318                t.hash(state);
319            }
320            Value::Date(d) => {
321                6u8.hash(state);
322                d.hash(state);
323            }
324            Value::Timestamp(t) => {
325                7u8.hash(state);
326                t.hash(state);
327            }
328            Value::Interval {
329                months,
330                days,
331                micros,
332            } => {
333                8u8.hash(state);
334                months.hash(state);
335                days.hash(state);
336                micros.hash(state);
337            }
338        }
339    }
340}
341
342impl PartialOrd for Value {
343    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
344        Some(self.cmp(other))
345    }
346}
347
348impl Ord for Value {
349    // Order: NULL < BOOLEAN < numeric < TIME < DATE < TIMESTAMP < INTERVAL < TEXT < BLOB.
350    // INTERVAL compares field-wise for trait-invariant safety; SQL-level ops normalize.
351    fn cmp(&self, other: &Self) -> Ordering {
352        match (self, other) {
353            (Value::Null, Value::Null) => Ordering::Equal,
354            (Value::Null, _) => Ordering::Less,
355            (_, Value::Null) => Ordering::Greater,
356
357            (Value::Boolean(a), Value::Boolean(b)) => a.cmp(b),
358            (Value::Boolean(_), _) => Ordering::Less,
359            (_, Value::Boolean(_)) => Ordering::Greater,
360
361            (Value::Integer(_) | Value::Real(_), Value::Integer(_) | Value::Real(_)) => {
362                self.numeric_cmp(other).unwrap_or(Ordering::Equal)
363            }
364            (Value::Integer(_) | Value::Real(_), _) => Ordering::Less,
365            (_, Value::Integer(_) | Value::Real(_)) => Ordering::Greater,
366
367            (Value::Time(a), Value::Time(b)) => a.cmp(b),
368            (Value::Time(_), _) => Ordering::Less,
369            (_, Value::Time(_)) => Ordering::Greater,
370
371            (Value::Date(a), Value::Date(b)) => a.cmp(b),
372            (Value::Date(_), _) => Ordering::Less,
373            (_, Value::Date(_)) => Ordering::Greater,
374
375            (Value::Timestamp(a), Value::Timestamp(b)) => a.cmp(b),
376            (Value::Timestamp(_), _) => Ordering::Less,
377            (_, Value::Timestamp(_)) => Ordering::Greater,
378
379            (
380                Value::Interval {
381                    months: am,
382                    days: ad,
383                    micros: au,
384                },
385                Value::Interval {
386                    months: bm,
387                    days: bd,
388                    micros: bu,
389                },
390            ) => am.cmp(bm).then(ad.cmp(bd)).then(au.cmp(bu)),
391            (Value::Interval { .. }, _) => Ordering::Less,
392            (_, Value::Interval { .. }) => Ordering::Greater,
393
394            (Value::Text(a), Value::Text(b)) => a.cmp(b),
395            (Value::Text(_), _) => Ordering::Less,
396            (_, Value::Text(_)) => Ordering::Greater,
397
398            (Value::Blob(a), Value::Blob(b)) => a.cmp(b),
399        }
400    }
401}
402
403impl fmt::Display for Value {
404    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
405        match self {
406            Value::Null => write!(f, "NULL"),
407            Value::Integer(i) => write!(f, "{i}"),
408            Value::Real(r) => {
409                if r.fract() == 0.0 && r.is_finite() {
410                    write!(f, "{r:.1}")
411                } else {
412                    write!(f, "{r}")
413                }
414            }
415            Value::Text(s) => write!(f, "{s}"),
416            Value::Blob(b) => write!(f, "X'{}'", hex_encode(b)),
417            Value::Boolean(b) => write!(f, "{}", if *b { "TRUE" } else { "FALSE" }),
418            Value::Time(t) => write!(f, "{}", crate::datetime::format_time(*t)),
419            Value::Date(d) => write!(f, "{}", crate::datetime::format_date(*d)),
420            Value::Timestamp(t) => write!(f, "{}", crate::datetime::format_timestamp(*t)),
421            Value::Interval {
422                months,
423                days,
424                micros,
425            } => {
426                write!(
427                    f,
428                    "{}",
429                    crate::datetime::format_interval(*months, *days, *micros)
430                )
431            }
432        }
433    }
434}
435
436fn hex_encode(data: &[u8]) -> String {
437    let mut s = String::with_capacity(data.len() * 2);
438    for byte in data {
439        s.push_str(&format!("{byte:02X}"));
440    }
441    s
442}
443
444/// Column definition.
445#[derive(Debug, Clone)]
446pub struct ColumnDef {
447    pub name: String,
448    pub data_type: DataType,
449    pub nullable: bool,
450    pub position: u16,
451    pub default_expr: Option<Expr>,
452    pub default_sql: Option<String>,
453    pub check_expr: Option<Expr>,
454    pub check_sql: Option<String>,
455    pub check_name: Option<String>,
456    /// Display-only flag for `TIMESTAMPTZ` / `TIMETZ`; storage is i64 µs UTC.
457    pub is_with_timezone: bool,
458    pub generated_expr: Option<Expr>,
459    pub generated_sql: Option<String>,
460    pub generated_kind: Option<crate::parser::GeneratedKind>,
461}
462
463/// Index definition stored as part of the table schema.
464#[derive(Debug, Clone)]
465pub struct IndexDef {
466    pub name: String,
467    pub columns: Vec<u16>,
468    pub unique: bool,
469}
470
471/// View definition stored in the _views metadata table.
472#[derive(Debug, Clone)]
473pub struct ViewDef {
474    pub name: String,
475    pub sql: String,
476    pub column_aliases: Vec<String>,
477}
478
479const VIEW_DEF_VERSION: u8 = 1;
480
481impl ViewDef {
482    pub fn serialize(&self) -> Vec<u8> {
483        let mut buf = Vec::new();
484        buf.push(VIEW_DEF_VERSION);
485
486        let name_bytes = self.name.as_bytes();
487        buf.extend_from_slice(&(name_bytes.len() as u16).to_le_bytes());
488        buf.extend_from_slice(name_bytes);
489
490        let sql_bytes = self.sql.as_bytes();
491        buf.extend_from_slice(&(sql_bytes.len() as u32).to_le_bytes());
492        buf.extend_from_slice(sql_bytes);
493
494        buf.extend_from_slice(&(self.column_aliases.len() as u16).to_le_bytes());
495        for alias in &self.column_aliases {
496            let alias_bytes = alias.as_bytes();
497            buf.extend_from_slice(&(alias_bytes.len() as u16).to_le_bytes());
498            buf.extend_from_slice(alias_bytes);
499        }
500
501        buf
502    }
503
504    pub fn deserialize(data: &[u8]) -> crate::error::Result<Self> {
505        if data.is_empty() || data[0] != VIEW_DEF_VERSION {
506            return Err(crate::error::SqlError::InvalidValue(
507                "invalid view definition version".into(),
508            ));
509        }
510        let mut pos = 1;
511
512        let name_len = u16::from_le_bytes([data[pos], data[pos + 1]]) as usize;
513        pos += 2;
514        let name = String::from_utf8_lossy(&data[pos..pos + name_len]).into_owned();
515        pos += name_len;
516
517        let sql_len =
518            u32::from_le_bytes([data[pos], data[pos + 1], data[pos + 2], data[pos + 3]]) as usize;
519        pos += 4;
520        let sql = String::from_utf8_lossy(&data[pos..pos + sql_len]).into_owned();
521        pos += sql_len;
522
523        let alias_count = u16::from_le_bytes([data[pos], data[pos + 1]]) as usize;
524        pos += 2;
525        let mut column_aliases = Vec::with_capacity(alias_count);
526        for _ in 0..alias_count {
527            let alias_len = u16::from_le_bytes([data[pos], data[pos + 1]]) as usize;
528            pos += 2;
529            let alias = String::from_utf8_lossy(&data[pos..pos + alias_len]).into_owned();
530            pos += alias_len;
531            column_aliases.push(alias);
532        }
533
534        Ok(Self {
535            name,
536            sql,
537            column_aliases,
538        })
539    }
540}
541
542/// Table-level CHECK constraint stored in schema.
543#[derive(Debug, Clone)]
544pub struct TableCheckDef {
545    pub name: Option<String>,
546    pub expr: Expr,
547    pub sql: String,
548}
549
550/// Foreign key definition stored in schema.
551#[derive(Debug, Clone)]
552pub struct ForeignKeySchemaEntry {
553    pub name: Option<String>,
554    pub columns: Vec<u16>,
555    pub foreign_table: String,
556    pub referred_columns: Vec<String>,
557}
558
559/// Table schema stored in the _schema table.
560#[derive(Debug, Clone)]
561pub struct TableSchema {
562    pub name: String,
563    pub columns: Vec<ColumnDef>,
564    pub primary_key_columns: Vec<u16>,
565    pub indices: Vec<IndexDef>,
566    pub check_constraints: Vec<TableCheckDef>,
567    pub foreign_keys: Vec<ForeignKeySchemaEntry>,
568    pk_idx_cache: Vec<usize>,
569    non_pk_idx_cache: Vec<usize>,
570    /// Physical encoding slots dropped via DROP COLUMN. Sorted.
571    /// Old rows have data at these positions (skipped on decode);
572    /// new rows encode NULL to maintain position consistency.
573    dropped_non_pk_slots: Vec<u16>,
574    /// Physical encoding position -> logical column index.
575    /// `usize::MAX` for dropped slots.
576    decode_mapping_cache: Vec<usize>,
577    /// Logical non-PK order -> physical encoding position.
578    /// `encoding_positions_cache[i]` is the physical slot for `non_pk_idx_cache[i]`.
579    encoding_positions_cache: Vec<u16>,
580    has_virtual_columns_cache: bool,
581}
582
583impl TableSchema {
584    pub fn new(
585        name: String,
586        columns: Vec<ColumnDef>,
587        primary_key_columns: Vec<u16>,
588        indices: Vec<IndexDef>,
589        check_constraints: Vec<TableCheckDef>,
590        foreign_keys: Vec<ForeignKeySchemaEntry>,
591    ) -> Self {
592        Self::with_drops(
593            name,
594            columns,
595            primary_key_columns,
596            indices,
597            check_constraints,
598            foreign_keys,
599            vec![],
600        )
601    }
602
603    pub fn with_drops(
604        name: String,
605        columns: Vec<ColumnDef>,
606        primary_key_columns: Vec<u16>,
607        indices: Vec<IndexDef>,
608        check_constraints: Vec<TableCheckDef>,
609        foreign_keys: Vec<ForeignKeySchemaEntry>,
610        dropped_non_pk_slots: Vec<u16>,
611    ) -> Self {
612        let pk_idx_cache: Vec<usize> = primary_key_columns.iter().map(|&i| i as usize).collect();
613        let non_pk_idx_cache: Vec<usize> = (0..columns.len())
614            .filter(|i| !primary_key_columns.contains(&(*i as u16)))
615            .collect();
616
617        let physical_count = non_pk_idx_cache.len() + dropped_non_pk_slots.len();
618        let mut decode_mapping_cache = vec![usize::MAX; physical_count];
619        let mut encoding_positions_cache = Vec::with_capacity(non_pk_idx_cache.len());
620
621        let mut drop_idx = 0;
622        let mut live_idx = 0;
623        for (phys_pos, slot) in decode_mapping_cache.iter_mut().enumerate() {
624            if drop_idx < dropped_non_pk_slots.len()
625                && dropped_non_pk_slots[drop_idx] as usize == phys_pos
626            {
627                drop_idx += 1;
628            } else {
629                *slot = non_pk_idx_cache[live_idx];
630                encoding_positions_cache.push(phys_pos as u16);
631                live_idx += 1;
632            }
633        }
634
635        let has_virtual_columns_cache = columns.iter().any(|c| {
636            matches!(
637                c.generated_kind,
638                Some(crate::parser::GeneratedKind::Virtual)
639            )
640        });
641
642        Self {
643            name,
644            columns,
645            primary_key_columns,
646            indices,
647            check_constraints,
648            foreign_keys,
649            pk_idx_cache,
650            non_pk_idx_cache,
651            dropped_non_pk_slots,
652            decode_mapping_cache,
653            encoding_positions_cache,
654            has_virtual_columns_cache,
655        }
656    }
657
658    pub fn has_virtual_columns(&self) -> bool {
659        self.has_virtual_columns_cache
660    }
661
662    /// Rebuild caches (preserving dropped slots). Use after mutating fields in place.
663    pub fn rebuild(self) -> Self {
664        let drops = self.dropped_non_pk_slots;
665        Self::with_drops(
666            self.name,
667            self.columns,
668            self.primary_key_columns,
669            self.indices,
670            self.check_constraints,
671            self.foreign_keys,
672            drops,
673        )
674    }
675
676    /// Returns true if any column-level or table-level CHECK constraints exist.
677    pub fn has_checks(&self) -> bool {
678        !self.check_constraints.is_empty() || self.columns.iter().any(|c| c.check_expr.is_some())
679    }
680
681    /// Physical encoding position -> logical column index mapping.
682    /// Length = physical_non_pk_count. `usize::MAX` for dropped slots.
683    pub fn decode_col_mapping(&self) -> &[usize] {
684        &self.decode_mapping_cache
685    }
686
687    /// Logical non-PK order -> physical encoding position.
688    /// `encoding_positions()[i]` is the physical slot for `non_pk_indices()[i]`.
689    pub fn encoding_positions(&self) -> &[u16] {
690        &self.encoding_positions_cache
691    }
692
693    /// Total physical non-PK column count (live + dropped slots).
694    pub fn physical_non_pk_count(&self) -> usize {
695        self.non_pk_idx_cache.len() + self.dropped_non_pk_slots.len()
696    }
697
698    /// Physical encoding slots that have been dropped via DROP COLUMN.
699    pub fn dropped_non_pk_slots(&self) -> &[u16] {
700        &self.dropped_non_pk_slots
701    }
702
703    /// Return a new schema with the column at `drop_pos` marked as dropped.
704    /// Row data is not rewritten; decode skips the slot. Logical positions
705    /// above `drop_pos` shift down; table-level CHECKs referencing it are
706    /// filtered out.
707    pub fn without_column(&self, drop_pos: usize) -> Self {
708        let non_pk_order = self
709            .non_pk_idx_cache
710            .iter()
711            .position(|&i| i == drop_pos)
712            .expect("cannot drop PK column via without_column");
713        let physical_slot = self.encoding_positions_cache[non_pk_order];
714
715        let mut new_dropped = self.dropped_non_pk_slots.clone();
716        new_dropped.push(physical_slot);
717        new_dropped.sort();
718
719        let dropped_name = &self.columns[drop_pos].name;
720        let drop_pos_u16 = drop_pos as u16;
721
722        let mut columns: Vec<ColumnDef> = self
723            .columns
724            .iter()
725            .enumerate()
726            .filter(|(i, _)| *i != drop_pos)
727            .map(|(_, c)| {
728                let mut col = c.clone();
729                if col.position > drop_pos_u16 {
730                    col.position -= 1;
731                }
732                col
733            })
734            .collect();
735        for (i, col) in columns.iter_mut().enumerate() {
736            col.position = i as u16;
737        }
738
739        let primary_key_columns: Vec<u16> = self
740            .primary_key_columns
741            .iter()
742            .map(|&p| if p > drop_pos_u16 { p - 1 } else { p })
743            .collect();
744
745        let indices: Vec<IndexDef> = self
746            .indices
747            .iter()
748            .map(|idx| IndexDef {
749                name: idx.name.clone(),
750                columns: idx
751                    .columns
752                    .iter()
753                    .map(|&c| if c > drop_pos_u16 { c - 1 } else { c })
754                    .collect(),
755                unique: idx.unique,
756            })
757            .collect();
758
759        let foreign_keys: Vec<ForeignKeySchemaEntry> = self
760            .foreign_keys
761            .iter()
762            .map(|fk| ForeignKeySchemaEntry {
763                name: fk.name.clone(),
764                columns: fk
765                    .columns
766                    .iter()
767                    .map(|&c| if c > drop_pos_u16 { c - 1 } else { c })
768                    .collect(),
769                foreign_table: fk.foreign_table.clone(),
770                referred_columns: fk.referred_columns.clone(),
771            })
772            .collect();
773
774        // Filter out table-level CHECKs that reference the dropped column
775        let dropped_lower = dropped_name.to_ascii_lowercase();
776        let check_constraints: Vec<TableCheckDef> = self
777            .check_constraints
778            .iter()
779            .filter(|c| !c.sql.to_ascii_lowercase().contains(&dropped_lower))
780            .cloned()
781            .collect();
782
783        Self::with_drops(
784            self.name.clone(),
785            columns,
786            primary_key_columns,
787            indices,
788            check_constraints,
789            foreign_keys,
790            new_dropped,
791        )
792    }
793}
794
795const SCHEMA_VERSION: u8 = 5;
796
797fn write_opt_string(buf: &mut Vec<u8>, s: &Option<String>) {
798    match s {
799        Some(s) => {
800            let bytes = s.as_bytes();
801            buf.extend_from_slice(&(bytes.len() as u16).to_le_bytes());
802            buf.extend_from_slice(bytes);
803        }
804        None => buf.extend_from_slice(&0u16.to_le_bytes()),
805    }
806}
807
808fn read_opt_string(data: &[u8], pos: &mut usize) -> Option<String> {
809    let len = u16::from_le_bytes([data[*pos], data[*pos + 1]]) as usize;
810    *pos += 2;
811    if len == 0 {
812        None
813    } else {
814        let s = String::from_utf8_lossy(&data[*pos..*pos + len]).into_owned();
815        *pos += len;
816        Some(s)
817    }
818}
819
820fn read_string(data: &[u8], pos: &mut usize) -> String {
821    let len = u16::from_le_bytes([data[*pos], data[*pos + 1]]) as usize;
822    *pos += 2;
823    let s = String::from_utf8_lossy(&data[*pos..*pos + len]).into_owned();
824    *pos += len;
825    s
826}
827
828impl TableSchema {
829    pub fn serialize(&self) -> Vec<u8> {
830        let mut buf = Vec::new();
831        buf.push(SCHEMA_VERSION);
832
833        let name_bytes = self.name.as_bytes();
834        buf.extend_from_slice(&(name_bytes.len() as u16).to_le_bytes());
835        buf.extend_from_slice(name_bytes);
836
837        buf.extend_from_slice(&(self.columns.len() as u16).to_le_bytes());
838
839        for col in &self.columns {
840            let col_name = col.name.as_bytes();
841            buf.extend_from_slice(&(col_name.len() as u16).to_le_bytes());
842            buf.extend_from_slice(col_name);
843            buf.push(col.data_type.type_tag());
844            buf.push(if col.nullable { 1 } else { 0 });
845            buf.extend_from_slice(&col.position.to_le_bytes());
846        }
847
848        buf.extend_from_slice(&(self.primary_key_columns.len() as u16).to_le_bytes());
849        for &pk_idx in &self.primary_key_columns {
850            buf.extend_from_slice(&pk_idx.to_le_bytes());
851        }
852
853        buf.extend_from_slice(&(self.indices.len() as u16).to_le_bytes());
854        for idx in &self.indices {
855            let idx_name = idx.name.as_bytes();
856            buf.extend_from_slice(&(idx_name.len() as u16).to_le_bytes());
857            buf.extend_from_slice(idx_name);
858            buf.extend_from_slice(&(idx.columns.len() as u16).to_le_bytes());
859            for &col_idx in &idx.columns {
860                buf.extend_from_slice(&col_idx.to_le_bytes());
861            }
862            buf.push(if idx.unique { 1 } else { 0 });
863        }
864
865        for col in &self.columns {
866            let mut flags: u8 = 0;
867            if col.default_sql.is_some() {
868                flags |= 1;
869            }
870            if col.check_sql.is_some() {
871                flags |= 2;
872            }
873            buf.push(flags);
874            if let Some(ref sql) = col.default_sql {
875                let bytes = sql.as_bytes();
876                buf.extend_from_slice(&(bytes.len() as u16).to_le_bytes());
877                buf.extend_from_slice(bytes);
878            }
879            if let Some(ref sql) = col.check_sql {
880                let bytes = sql.as_bytes();
881                buf.extend_from_slice(&(bytes.len() as u16).to_le_bytes());
882                buf.extend_from_slice(bytes);
883                write_opt_string(&mut buf, &col.check_name);
884            }
885        }
886
887        buf.extend_from_slice(&(self.check_constraints.len() as u16).to_le_bytes());
888        for chk in &self.check_constraints {
889            write_opt_string(&mut buf, &chk.name);
890            let sql_bytes = chk.sql.as_bytes();
891            buf.extend_from_slice(&(sql_bytes.len() as u16).to_le_bytes());
892            buf.extend_from_slice(sql_bytes);
893        }
894
895        buf.extend_from_slice(&(self.foreign_keys.len() as u16).to_le_bytes());
896        for fk in &self.foreign_keys {
897            write_opt_string(&mut buf, &fk.name);
898            buf.extend_from_slice(&(fk.columns.len() as u16).to_le_bytes());
899            for &col_idx in &fk.columns {
900                buf.extend_from_slice(&col_idx.to_le_bytes());
901            }
902            let ft_bytes = fk.foreign_table.as_bytes();
903            buf.extend_from_slice(&(ft_bytes.len() as u16).to_le_bytes());
904            buf.extend_from_slice(ft_bytes);
905            buf.extend_from_slice(&(fk.referred_columns.len() as u16).to_le_bytes());
906            for rc in &fk.referred_columns {
907                let rc_bytes = rc.as_bytes();
908                buf.extend_from_slice(&(rc_bytes.len() as u16).to_le_bytes());
909                buf.extend_from_slice(rc_bytes);
910            }
911        }
912
913        buf.extend_from_slice(&(self.dropped_non_pk_slots.len() as u16).to_le_bytes());
914        for &slot in &self.dropped_non_pk_slots {
915            buf.extend_from_slice(&slot.to_le_bytes());
916        }
917
918        for col in &self.columns {
919            let kind_tag: u8 = match col.generated_kind {
920                None => 0,
921                Some(crate::parser::GeneratedKind::Stored) => 1,
922                Some(crate::parser::GeneratedKind::Virtual) => 2,
923            };
924            buf.push(kind_tag);
925            if kind_tag != 0 {
926                let sql = col.generated_sql.as_deref().unwrap_or("");
927                let bytes = sql.as_bytes();
928                buf.extend_from_slice(&(bytes.len() as u32).to_le_bytes());
929                buf.extend_from_slice(bytes);
930            }
931        }
932
933        buf
934    }
935
936    pub fn deserialize(data: &[u8]) -> crate::error::Result<Self> {
937        let mut pos = 0;
938
939        if data.is_empty() || !matches!(data[0], 1 | 2 | 3 | 4 | SCHEMA_VERSION) {
940            return Err(crate::error::SqlError::InvalidValue(
941                "invalid schema version".into(),
942            ));
943        }
944        let version = data[0];
945        pos += 1;
946
947        let name_len = u16::from_le_bytes([data[pos], data[pos + 1]]) as usize;
948        pos += 2;
949        let name = String::from_utf8_lossy(&data[pos..pos + name_len]).into_owned();
950        pos += name_len;
951
952        let col_count = u16::from_le_bytes([data[pos], data[pos + 1]]) as usize;
953        pos += 2;
954
955        let mut columns = Vec::with_capacity(col_count);
956        for _ in 0..col_count {
957            let col_name_len = u16::from_le_bytes([data[pos], data[pos + 1]]) as usize;
958            pos += 2;
959            let col_name = String::from_utf8_lossy(&data[pos..pos + col_name_len]).into_owned();
960            pos += col_name_len;
961            let data_type = DataType::from_tag(data[pos]).ok_or_else(|| {
962                crate::error::SqlError::InvalidValue("unknown data type tag".into())
963            })?;
964            pos += 1;
965            let nullable = data[pos] != 0;
966            pos += 1;
967            let position = u16::from_le_bytes([data[pos], data[pos + 1]]);
968            pos += 2;
969            columns.push(ColumnDef {
970                name: col_name,
971                data_type,
972                nullable,
973                position,
974                default_expr: None,
975                default_sql: None,
976                check_expr: None,
977                check_sql: None,
978                check_name: None,
979                is_with_timezone: false,
980                generated_expr: None,
981                generated_sql: None,
982                generated_kind: None,
983            });
984        }
985
986        let pk_count = u16::from_le_bytes([data[pos], data[pos + 1]]) as usize;
987        pos += 2;
988        let mut primary_key_columns = Vec::with_capacity(pk_count);
989        for _ in 0..pk_count {
990            let pk_idx = u16::from_le_bytes([data[pos], data[pos + 1]]);
991            pos += 2;
992            primary_key_columns.push(pk_idx);
993        }
994
995        let indices = if version >= 2 && pos + 2 <= data.len() {
996            let idx_count = u16::from_le_bytes([data[pos], data[pos + 1]]) as usize;
997            pos += 2;
998            let mut idxs = Vec::with_capacity(idx_count);
999            for _ in 0..idx_count {
1000                let idx_name_len = u16::from_le_bytes([data[pos], data[pos + 1]]) as usize;
1001                pos += 2;
1002                let idx_name = String::from_utf8_lossy(&data[pos..pos + idx_name_len]).into_owned();
1003                pos += idx_name_len;
1004                let col_count = u16::from_le_bytes([data[pos], data[pos + 1]]) as usize;
1005                pos += 2;
1006                let mut cols = Vec::with_capacity(col_count);
1007                for _ in 0..col_count {
1008                    let col_idx = u16::from_le_bytes([data[pos], data[pos + 1]]);
1009                    pos += 2;
1010                    cols.push(col_idx);
1011                }
1012                let unique = data[pos] != 0;
1013                pos += 1;
1014                idxs.push(IndexDef {
1015                    name: idx_name,
1016                    columns: cols,
1017                    unique,
1018                });
1019            }
1020            idxs
1021        } else {
1022            vec![]
1023        };
1024
1025        let mut check_constraints = Vec::new();
1026        let mut foreign_keys = Vec::new();
1027
1028        if version >= 3 && pos < data.len() {
1029            for col in &mut columns {
1030                let flags = data[pos];
1031                pos += 1;
1032                if flags & 1 != 0 {
1033                    let sql = read_string(data, &mut pos);
1034                    col.default_expr = Some(crate::parser::parse_sql_expr(&sql).map_err(|_| {
1035                        crate::error::SqlError::InvalidValue(format!(
1036                            "cannot parse DEFAULT expression: {sql}"
1037                        ))
1038                    })?);
1039                    col.default_sql = Some(sql);
1040                }
1041                if flags & 2 != 0 {
1042                    let sql = read_string(data, &mut pos);
1043                    col.check_expr = Some(crate::parser::parse_sql_expr(&sql).map_err(|_| {
1044                        crate::error::SqlError::InvalidValue(format!(
1045                            "cannot parse CHECK expression: {sql}"
1046                        ))
1047                    })?);
1048                    col.check_sql = Some(sql);
1049                    col.check_name = read_opt_string(data, &mut pos);
1050                }
1051            }
1052
1053            let chk_count = u16::from_le_bytes([data[pos], data[pos + 1]]) as usize;
1054            pos += 2;
1055            for _ in 0..chk_count {
1056                let name = read_opt_string(data, &mut pos);
1057                let sql = read_string(data, &mut pos);
1058                let expr = crate::parser::parse_sql_expr(&sql).map_err(|_| {
1059                    crate::error::SqlError::InvalidValue(format!(
1060                        "cannot parse CHECK expression: {sql}"
1061                    ))
1062                })?;
1063                check_constraints.push(TableCheckDef { name, expr, sql });
1064            }
1065
1066            let fk_count = u16::from_le_bytes([data[pos], data[pos + 1]]) as usize;
1067            pos += 2;
1068            for _ in 0..fk_count {
1069                let name = read_opt_string(data, &mut pos);
1070                let col_count = u16::from_le_bytes([data[pos], data[pos + 1]]) as usize;
1071                pos += 2;
1072                let mut cols = Vec::with_capacity(col_count);
1073                for _ in 0..col_count {
1074                    let col_idx = u16::from_le_bytes([data[pos], data[pos + 1]]);
1075                    pos += 2;
1076                    cols.push(col_idx);
1077                }
1078                let foreign_table = read_string(data, &mut pos);
1079                let ref_count = u16::from_le_bytes([data[pos], data[pos + 1]]) as usize;
1080                pos += 2;
1081                let mut referred_columns = Vec::with_capacity(ref_count);
1082                for _ in 0..ref_count {
1083                    referred_columns.push(read_string(data, &mut pos));
1084                }
1085                foreign_keys.push(ForeignKeySchemaEntry {
1086                    name,
1087                    columns: cols,
1088                    foreign_table,
1089                    referred_columns,
1090                });
1091            }
1092        }
1093        let mut dropped_non_pk_slots = Vec::new();
1094        if version >= 4 && pos + 2 <= data.len() {
1095            let slot_count = u16::from_le_bytes([data[pos], data[pos + 1]]) as usize;
1096            pos += 2;
1097            for _ in 0..slot_count {
1098                let slot = u16::from_le_bytes([data[pos], data[pos + 1]]);
1099                pos += 2;
1100                dropped_non_pk_slots.push(slot);
1101            }
1102        }
1103        if version >= 5 && pos < data.len() {
1104            for col in &mut columns {
1105                let kind_tag = data[pos];
1106                pos += 1;
1107                if kind_tag != 0 {
1108                    let len = u32::from_le_bytes([
1109                        data[pos],
1110                        data[pos + 1],
1111                        data[pos + 2],
1112                        data[pos + 3],
1113                    ]) as usize;
1114                    pos += 4;
1115                    let sql = String::from_utf8_lossy(&data[pos..pos + len]).into_owned();
1116                    pos += len;
1117                    let expr = crate::parser::parse_sql_expr(&sql).map_err(|_| {
1118                        crate::error::SqlError::InvalidValue(format!(
1119                            "cannot parse GENERATED expression: {sql}"
1120                        ))
1121                    })?;
1122                    col.generated_sql = Some(sql);
1123                    col.generated_expr = Some(expr);
1124                    col.generated_kind = Some(match kind_tag {
1125                        1 => crate::parser::GeneratedKind::Stored,
1126                        2 => crate::parser::GeneratedKind::Virtual,
1127                        _ => {
1128                            return Err(crate::error::SqlError::InvalidValue(
1129                                "unknown GENERATED kind tag".into(),
1130                            ));
1131                        }
1132                    });
1133                }
1134            }
1135        }
1136        let _ = pos;
1137
1138        Ok(Self::with_drops(
1139            name,
1140            columns,
1141            primary_key_columns,
1142            indices,
1143            check_constraints,
1144            foreign_keys,
1145            dropped_non_pk_slots,
1146        ))
1147    }
1148
1149    /// Get column index by name (case-insensitive).
1150    pub fn column_index(&self, name: &str) -> Option<usize> {
1151        self.columns
1152            .iter()
1153            .position(|c| c.name.eq_ignore_ascii_case(name))
1154    }
1155
1156    /// Get indices of non-PK columns (columns stored in the B+ tree value).
1157    pub fn non_pk_indices(&self) -> &[usize] {
1158        &self.non_pk_idx_cache
1159    }
1160
1161    /// Get the PK column indices as usize.
1162    pub fn pk_indices(&self) -> &[usize] {
1163        &self.pk_idx_cache
1164    }
1165
1166    /// Get index definition by name (case-insensitive).
1167    pub fn index_by_name(&self, name: &str) -> Option<&IndexDef> {
1168        let lower = name.to_ascii_lowercase();
1169        self.indices.iter().find(|i| i.name == lower)
1170    }
1171
1172    /// Get the KV table name for an index.
1173    pub fn index_table_name(table_name: &str, index_name: &str) -> Vec<u8> {
1174        format!("__idx_{table_name}_{index_name}").into_bytes()
1175    }
1176}
1177
1178/// Result of executing a SQL statement.
1179#[derive(Debug)]
1180pub enum ExecutionResult {
1181    RowsAffected(u64),
1182    Query(QueryResult),
1183    Ok,
1184}
1185
1186/// Result of a SELECT query.
1187#[derive(Debug, Clone)]
1188pub struct QueryResult {
1189    pub columns: Vec<String>,
1190    pub rows: Vec<Vec<Value>>,
1191}
1192
1193#[cfg(test)]
1194#[path = "types_tests.rs"]
1195mod tests;
citadel_sql/types.rs

citadel_sql/
types.rs
