mongreldb_core/
constraint.rs

1//! Engine-side declarative constraints (unique, foreign key, check) enforced
2//! authoritatively inside the core transaction path. Opt-in per-table via
3//! [`crate::schema::Schema::constraints`]; tables with an empty constraint set
4//! (the default — including every legacy table and every Kit-managed table)
5//! behave exactly as before. This subsystem is independent of the Kit's own
6//! guard-table mechanism: the Kit continues to enforce its constraints exactly
7//! as before, and these engine constraints only fire for tables whose schema
8//! carries a non-empty [`TableConstraints`].
9//!
10//! Enforcement is performed in [`crate::database::Database::commit_transaction`]
11//! as a pre-sequencer validation pass (under the transaction's read snapshot,
12//! outside the WAL mutex) plus `WriteKey::Unique` registration so that two
13//! concurrent transactions inserting the same key cannot both commit.
14
15use crate::error::{MongrelError, Result};
16use crate::memtable::Value;
17use serde::{Deserialize, Serialize};
18use std::cmp::Ordering;
19use std::collections::HashMap;
20
21/// A declarative constraint set attached to a table's schema. Empty by default.
22#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq)]
23pub struct TableConstraints {
24    #[serde(default)]
25    pub uniques: Vec<UniqueConstraint>,
26    #[serde(default)]
27    pub foreign_keys: Vec<ForeignKey>,
28    #[serde(default)]
29    pub checks: Vec<CheckConstraint>,
30}
31
32impl TableConstraints {
33    pub fn is_empty(&self) -> bool {
34        self.uniques.is_empty() && self.foreign_keys.is_empty() && self.checks.is_empty()
35    }
36}
37
38/// A multi-column uniqueness constraint (beyond the single-column `PRIMARY_KEY`
39/// flag). Enforced via an existence scan against the read snapshot plus
40/// `WriteKey::Unique` registration at commit (first-committer-wins).
41#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
42pub struct UniqueConstraint {
43    pub id: u16,
44    pub name: String,
45    pub columns: Vec<u16>,
46}
47
48/// ON DELETE action for a [`ForeignKey`].
49#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Default)]
50pub enum FkAction {
51    /// Reject the parent delete if any child row references it (default).
52    #[default]
53    Restrict,
54    /// Cascade the delete to child rows.
55    Cascade,
56    /// Set the referencing columns to `NULL` in child rows.
57    SetNull,
58}
59
60/// A foreign key: the listed `columns` must reference an existing row in
61/// `ref_table` whose `ref_columns` match. The parent table is named by string so
62/// the link survives the parent's numeric table-id assignment across reopens.
63#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
64pub struct ForeignKey {
65    pub id: u16,
66    pub name: String,
67    pub columns: Vec<u16>,
68    pub ref_table: String,
69    pub ref_columns: Vec<u16>,
70    #[serde(default)]
71    pub on_delete: FkAction,
72}
73
74/// A CHECK constraint: the row is rejected when `expr` evaluates to `False`
75/// (SQL three-valued logic: `Unknown`/`True` both pass).
76#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
77pub struct CheckConstraint {
78    pub id: u16,
79    pub name: String,
80    pub expr: CheckExpr,
81}
82
83/// A minimal boolean expression IR for CHECK constraints, evaluated against a
84/// row's cells. Terms ([`CheckExpr::Col`] / [`CheckExpr::Lit`]) resolve to
85/// [`Value`]; comparisons and logical ops are boolean. Any comparison involving
86/// `Value::Null` yields three-valued `Unknown`.
87#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
88pub enum CheckExpr {
89    /// Constant true (the trivially-satisfied CHECK).
90    True,
91    /// A bare column reference used as a boolean (truthiness coercion).
92    Col(u16),
93    /// A literal value.
94    Lit(Value),
95    IsNull(u16),
96    IsNotNull(u16),
97    Eq(Box<CheckExpr>, Box<CheckExpr>),
98    Ne(Box<CheckExpr>, Box<CheckExpr>),
99    Lt(Box<CheckExpr>, Box<CheckExpr>),
100    Le(Box<CheckExpr>, Box<CheckExpr>),
101    Gt(Box<CheckExpr>, Box<CheckExpr>),
102    Ge(Box<CheckExpr>, Box<CheckExpr>),
103    And(Box<CheckExpr>, Box<CheckExpr>),
104    Or(Box<CheckExpr>, Box<CheckExpr>),
105    Not(Box<CheckExpr>),
106}
107
108/// Three-valued logic result for CHECK evaluation.
109#[derive(Debug, Clone, Copy, PartialEq, Eq)]
110enum Tri {
111    True,
112    False,
113    Unknown,
114}
115
116impl CheckExpr {
117    /// Evaluate the expression against a row's cells. A CHECK constraint is
118    /// satisfied unless this returns [`Tri::False`].
119    pub fn satisfied(&self, cells: &HashMap<u16, Value>) -> bool {
120        !matches!(self.eval(cells), Tri::False)
121    }
122
123    fn eval(&self, cells: &HashMap<u16, Value>) -> Tri {
124        match self {
125            CheckExpr::True => Tri::True,
126            CheckExpr::Col(id) => match cells.get(id) {
127                None | Some(Value::Null) => Tri::Unknown,
128                Some(v) => Tri::from_truthy(v),
129            },
130            CheckExpr::Lit(v) => Tri::from_truthy(v),
131            CheckExpr::IsNull(id) => match cells.get(id) {
132                None | Some(Value::Null) => Tri::True,
133                Some(_) => Tri::False,
134            },
135            CheckExpr::IsNotNull(id) => match cells.get(id) {
136                None | Some(Value::Null) => Tri::False,
137                Some(_) => Tri::True,
138            },
139            CheckExpr::Eq(a, b) => compare(a.eval_term(cells), b.eval_term(cells), |o| {
140                o == Ordering::Equal
141            }),
142            CheckExpr::Ne(a, b) => compare(a.eval_term(cells), b.eval_term(cells), |o| {
143                o != Ordering::Equal
144            }),
145            CheckExpr::Lt(a, b) => compare(a.eval_term(cells), b.eval_term(cells), |o| {
146                o == Ordering::Less
147            }),
148            CheckExpr::Le(a, b) => compare(a.eval_term(cells), b.eval_term(cells), |o| {
149                o != Ordering::Greater
150            }),
151            CheckExpr::Gt(a, b) => compare(a.eval_term(cells), b.eval_term(cells), |o| {
152                o == Ordering::Greater
153            }),
154            CheckExpr::Ge(a, b) => compare(a.eval_term(cells), b.eval_term(cells), |o| {
155                o != Ordering::Less
156            }),
157            CheckExpr::And(a, b) => and3(a.eval(cells), b.eval(cells)),
158            CheckExpr::Or(a, b) => or3(a.eval(cells), b.eval(cells)),
159            CheckExpr::Not(a) => not3(a.eval(cells)),
160        }
161    }
162
163    /// Evaluate a term (Col/Lit) to a [`Value`]; boolean expressions coerce via
164    /// truthiness (True→1, False/Unknown→Null) so they can still be compared.
165    fn eval_term(&self, cells: &HashMap<u16, Value>) -> Value {
166        match self {
167            CheckExpr::Col(id) => cells.get(id).cloned().unwrap_or(Value::Null),
168            CheckExpr::Lit(v) => v.clone(),
169            other => match other.eval(cells) {
170                Tri::True => Value::Int64(1),
171                Tri::False | Tri::Unknown => Value::Null,
172            },
173        }
174    }
175}
176
177impl Tri {
178    fn from_truthy(v: &Value) -> Tri {
179        match v {
180            Value::Null => Tri::Unknown,
181            Value::Bool(b) => {
182                if *b {
183                    Tri::True
184                } else {
185                    Tri::False
186                }
187            }
188            Value::Int64(n) => {
189                if *n != 0 {
190                    Tri::True
191                } else {
192                    Tri::False
193                }
194            }
195            Value::Float64(f) => {
196                if *f != 0.0 {
197                    Tri::True
198                } else {
199                    Tri::False
200                }
201            }
202            Value::Bytes(b) => {
203                if b.is_empty() {
204                    Tri::False
205                } else {
206                    Tri::True
207                }
208            }
209            Value::Embedding(v) => {
210                if v.is_empty() {
211                    Tri::False
212                } else {
213                    Tri::True
214                }
215            }
216        }
217    }
218}
219
220fn and3(a: Tri, b: Tri) -> Tri {
221    match (a, b) {
222        (Tri::False, _) | (_, Tri::False) => Tri::False,
223        (Tri::Unknown, _) | (_, Tri::Unknown) => Tri::Unknown,
224        _ => Tri::True,
225    }
226}
227
228fn or3(a: Tri, b: Tri) -> Tri {
229    match (a, b) {
230        (Tri::True, _) | (_, Tri::True) => Tri::True,
231        (Tri::Unknown, _) | (_, Tri::Unknown) => Tri::Unknown,
232        _ => Tri::False,
233    }
234}
235
236fn not3(a: Tri) -> Tri {
237    match a {
238        Tri::True => Tri::False,
239        Tri::False => Tri::True,
240        Tri::Unknown => Tri::Unknown,
241    }
242}
243
244/// Run a comparison from two values. If either term is `Null` the comparison is
245/// `Unknown`; otherwise apply `pred` to the concrete ordering (incomparable
246/// types also yield `Unknown`).
247fn compare(a: Value, b: Value, pred: impl Fn(Ordering) -> bool) -> Tri {
248    if matches!(a, Value::Null) || matches!(b, Value::Null) {
249        return Tri::Unknown;
250    }
251    match value_cmp(&a, &b) {
252        Some(o) => {
253            if pred(o) {
254                Tri::True
255            } else {
256                Tri::False
257            }
258        }
259        None => Tri::Unknown,
260    }
261}
262
263/// Cross-type comparison for values. Same-shape values compare naturally;
264/// Int64/Float64 cross-compare numerically; Bytes lexicographically; Bool as
265/// 0/1; other distinct-type pairs are `None` (incomparable).
266pub(crate) fn value_cmp(a: &Value, b: &Value) -> Option<Ordering> {
267    match (a, b) {
268        (Value::Null, Value::Null) => Some(Ordering::Equal),
269        (Value::Bool(x), Value::Bool(y)) => Some((*x as u8).cmp(&(*y as u8))),
270        (Value::Int64(x), Value::Int64(y)) => Some(x.cmp(y)),
271        (Value::Float64(x), Value::Float64(y)) => x.partial_cmp(y),
272        (Value::Int64(x), Value::Float64(y)) => (*x as f64).partial_cmp(y),
273        (Value::Float64(x), Value::Int64(y)) => x.partial_cmp(&(*y as f64)),
274        (Value::Bytes(x), Value::Bytes(y)) => Some(x.cmp(y)),
275        (Value::Embedding(x), Value::Embedding(y)) => {
276            for (a, b) in x.iter().zip(y.iter()) {
277                match a.partial_cmp(b)? {
278                    Ordering::Equal => continue,
279                    non_eq => return Some(non_eq),
280                }
281            }
282            Some(x.len().cmp(&y.len()))
283        }
284        _ => None,
285    }
286}
287
288/// Encode a set of column values into a stable, unambiguous composite key for
289/// uniqueness / FK matching. Returns `None` if any referenced column is missing
290/// or `Null` — SQL semantics: a UNIQUE constraint ignores rows where any
291/// constrained column is NULL, and an FK with a NULL component is not checked.
292pub(crate) fn encode_composite_key(
293    columns: &[u16],
294    cells: &HashMap<u16, Value>,
295) -> Option<Vec<u8>> {
296    let mut out = Vec::new();
297    for cid in columns {
298        let v = cells.get(cid)?;
299        if matches!(v, Value::Null) {
300            return None;
301        }
302        let k = v.encode_key();
303        // Length-prefix so concatenated multi-column keys stay unambiguous.
304        out.extend_from_slice(&(k.len() as u32).to_be_bytes());
305        out.extend_from_slice(&k);
306    }
307    Some(out)
308}
309
310/// Validate CHECK constraints for a single staged row.
311pub(crate) fn validate_checks(
312    checks: &[CheckConstraint],
313    cells: &HashMap<u16, Value>,
314) -> Result<()> {
315    for c in checks {
316        if !c.expr.satisfied(cells) {
317            return Err(MongrelError::InvalidArgument(format!(
318                "CHECK constraint '{}' failed",
319                c.name
320            )));
321        }
322    }
323    Ok(())
324}
325
326#[cfg(test)]
327mod tests {
328    use super::*;
329
330    fn m(cols: &[(u16, Value)]) -> HashMap<u16, Value> {
331        cols.iter().cloned().collect()
332    }
333
334    #[test]
335    fn check_eq_literal() {
336        // col 1 == 5
337        let e = CheckExpr::Eq(
338            Box::new(CheckExpr::Col(1)),
339            Box::new(CheckExpr::Lit(Value::Int64(5))),
340        );
341        assert!(e.satisfied(&m(&[(1, Value::Int64(5))])));
342        assert!(!e.satisfied(&m(&[(1, Value::Int64(6))])));
343        // null comparison → unknown → satisfied (CHECK passes on unknown)
344        assert!(e.satisfied(&m(&[(1, Value::Null)])));
345    }
346
347    #[test]
348    fn check_range_and() {
349        // 0 <= col1 <= 100
350        let e = CheckExpr::And(
351            Box::new(CheckExpr::Ge(
352                Box::new(CheckExpr::Col(1)),
353                Box::new(CheckExpr::Lit(Value::Int64(0))),
354            )),
355            Box::new(CheckExpr::Le(
356                Box::new(CheckExpr::Col(1)),
357                Box::new(CheckExpr::Lit(Value::Int64(100))),
358            )),
359        );
360        assert!(e.satisfied(&m(&[(1, Value::Int64(50))])));
361        assert!(!e.satisfied(&m(&[(1, Value::Int64(101))])));
362        assert!(!e.satisfied(&m(&[(1, Value::Int64(-1))])));
363    }
364
365    #[test]
366    fn check_numeric_cross_type() {
367        let e = CheckExpr::Lt(
368            Box::new(CheckExpr::Col(1)),
369            Box::new(CheckExpr::Lit(Value::Float64(10.0))),
370        );
371        assert!(e.satisfied(&m(&[(1, Value::Int64(5))])));
372        assert!(!e.satisfied(&m(&[(1, Value::Int64(20))])));
373    }
374
375    #[test]
376    fn check_not_incomparable_is_unknown_passes() {
377        // comparing int to bytes is incomparable → unknown → satisfied
378        let e = CheckExpr::Eq(
379            Box::new(CheckExpr::Col(1)),
380            Box::new(CheckExpr::Lit(Value::Bytes(b"x".to_vec()))),
381        );
382        assert!(e.satisfied(&m(&[(1, Value::Int64(5))])));
383    }
384
385    #[test]
386    fn encode_composite_key_skips_null() {
387        let k = encode_composite_key(&[1, 2], &m(&[(1, Value::Int64(5)), (2, Value::Null)]));
388        assert!(k.is_none());
389        let k = encode_composite_key(&[1, 2], &m(&[(1, Value::Int64(5)), (2, Value::Int64(7))]));
390        assert!(k.is_some());
391        // distinct values → distinct keys
392        let k2 = encode_composite_key(&[1, 2], &m(&[(1, Value::Int64(6)), (2, Value::Int64(7))]));
393        assert_ne!(k.unwrap(), k2.unwrap());
394    }
395}
mongreldb_core/constraint.rs

mongreldb_core/
constraint.rs
