Skip to main content

inillucent_sql/
foreign_key.rs

1//! Foreign keys, as the triggers they are.
2//!
3//! Invariant: a foreign key is enforced by exactly the machinery a written
4//! trigger is enforced by. The clause is turned into `CREATE TRIGGER` text,
5//! parsed by the same parser, bound by the same binder and inlined by the same
6//! compiler - so `ON DELETE CASCADE` and the `DELETE` somebody wrote by hand
7//! cannot disagree about what a conflict clause does, what `OLD` means, or what
8//! order things happen in. SQLite makes the same choice for the same reason.
9//!
10//! Generating text rather than building bound structures is deliberate. The
11//! text is printable, so a diagnostic can show what a constraint actually does,
12//! and it is the same shape a person would have written - which means every
13//! test that covers written triggers covers this too.
14//!
15//! Four kinds of trigger come out of one clause:
16//!
17//! - the child's check, on `INSERT` and on `UPDATE OF` its own key columns,
18//!   which refuses a row whose parent is not there;
19//! - the parent's check, on `DELETE` and on `UPDATE OF` its key, which refuses
20//!   to strand a child - this is `NO ACTION` and `RESTRICT`;
21//! - the parent's `CASCADE`, which deletes or updates the children with it;
22//! - the parent's `SET NULL` and `SET DEFAULT`, which keep the children and
23//!   let go of the key.
24//!
25//! Reference: <https://sqlite.org/foreignkeys.html>.
26
27use inillucent_base::limits::Limits;
28
29use crate::ast::{ReferentialAction, TriggerTime};
30use crate::catalog_view::{
31    ForeignKeyInfo, ForeignKeyTrigger, TableInfo, TableKind, TriggerEventInfo, TriggerInfo,
32};
33use crate::parser::parse_next_statement;
34
35/// The message SQLite reports for every foreign-key violation.
36pub const VIOLATION_MESSAGE: &str = "FOREIGN KEY constraint failed";
37
38/// Which write a synthesised trigger is generated for.
39#[derive(Clone, Copy, Debug, Eq, PartialEq)]
40pub enum ForeignKeyEvent {
41    /// A row is being added to the child table.
42    ChildInsert,
43    /// A row of the child table is being changed.
44    ChildUpdate,
45    /// A row is being taken out of the parent table.
46    ParentDelete,
47    /// A row of the parent table is being changed.
48    ParentUpdate,
49}
50
51impl ForeignKeyEvent {}
52
53/// The parent columns a key refers to.
54///
55/// A clause that named none refers to the parent's primary key, and that is
56/// resolved here rather than in the catalog because the catalog reads one table
57/// at a time and the parent may not have been read yet.
58pub fn parent_columns(key: &ForeignKeyInfo, parent: &TableInfo) -> Option<Vec<Vec<u8>>> {
59    if !key.parent_columns.is_empty() {
60        return Some(key.parent_columns.clone());
61    }
62    let primary = parent.primary_key();
63    if primary.is_empty() {
64        return None;
65    }
66    let mut names = Vec::with_capacity(primary.len());
67    for position in primary {
68        names.push(parent.columns.get(usize::from(position))?.name.clone());
69    }
70    Some(names)
71}
72
73/// Returns the child column names of a key, in the order they were written.
74fn child_columns(key: &ForeignKeyInfo, child: &TableInfo) -> Option<Vec<Vec<u8>>> {
75    let mut names = Vec::with_capacity(key.columns.len());
76    for position in &key.columns {
77        names.push(child.columns.get(usize::from(*position))?.name.clone());
78    }
79    Some(names)
80}
81
82/// Writes an identifier the way it can be read back.
83fn quoted(name: &[u8], out: &mut String) {
84    out.push('"');
85    for byte in name {
86        if *byte == b'"' {
87            out.push('"');
88        }
89        out.push(char::from(*byte));
90    }
91    out.push('"');
92}
93
94/// Returns an identifier as a quoted string.
95fn quote(name: &[u8]) -> String {
96    let mut out = String::new();
97    quoted(name, &mut out);
98    out
99}
100
101/// Returns `db."table"`, so a body cannot be captured by a `temp` table of the
102/// same name.
103fn qualified(database: &[u8], table: &[u8]) -> String {
104    let mut out = quote(database);
105    out.push('.');
106    quoted(table, &mut out);
107    out
108}
109
110/// Joins the parts of a key comparison with `AND`.
111fn conjunction(parts: &[String]) -> String {
112    if parts.is_empty() {
113        return "1".to_string();
114    }
115    parts.join(" AND ")
116}
117
118/// Returns `"c1" = OLD."p1" AND ...`, which finds the children of one parent.
119fn children_of(child: &[Vec<u8>], parent: &[Vec<u8>], row: &str) -> String {
120    let mut parts = Vec::with_capacity(child.len());
121    for (near, far) in child.iter().zip(parent.iter()) {
122        parts.push(format!("{} = {row}.{}", quote(near), quote(far)));
123    }
124    conjunction(&parts)
125}
126
127/// Returns the name a synthesised trigger is known by.
128///
129/// It has to be unique and it has to be stable: the binder's recursion guard is
130/// a list of names, so two different constraints on the same table must not
131/// collide, and the same constraint must be recognisable when the cascade
132/// reaches it again.
133fn trigger_name(child: &TableInfo, key: &ForeignKeyInfo, event: ForeignKeyEvent) -> Vec<u8> {
134    let suffix = match event {
135        ForeignKeyEvent::ChildInsert => "ci",
136        ForeignKeyEvent::ChildUpdate => "cu",
137        ForeignKeyEvent::ParentDelete => "pd",
138        ForeignKeyEvent::ParentUpdate => "pu",
139    };
140    let mut name = b"sqlite_fk_".to_vec();
141    name.extend_from_slice(&child.folded);
142    name.push(b'_');
143    name.extend_from_slice(key.id.to_string().as_bytes());
144    name.push(b'_');
145    name.extend_from_slice(suffix.as_bytes());
146    name
147}
148
149/// Builds the trigger that enforces one key for one event, if there is one.
150///
151/// `None` means the event needs no trigger - a `NO ACTION` parent key whose
152/// checks are deferred to the commit, for instance, or a child key whose
153/// columns this update does not touch.
154pub fn trigger_for(
155    child: &TableInfo,
156    parent: &TableInfo,
157    key: &ForeignKeyInfo,
158    event: ForeignKeyEvent,
159    database: &[u8],
160    deferred: bool,
161    limits: &Limits,
162) -> Option<TriggerInfo> {
163    let near = child_columns(key, child)?;
164    let far = parent_columns(key, parent)?;
165    if near.len() != far.len() || near.is_empty() {
166        return None;
167    }
168    let sql = match event {
169        ForeignKeyEvent::ChildInsert | ForeignKeyEvent::ChildUpdate => {
170            if deferred {
171                return None;
172            }
173            child_check(child, parent, key, event, database, &near, &far)
174        }
175        ForeignKeyEvent::ParentDelete | ForeignKeyEvent::ParentUpdate => {
176            parent_action(child, parent, key, event, database, &near, &far, deferred)?
177        }
178    };
179    build(&sql, trigger_name(child, key, event), limits)
180}
181
182/// Parses generated trigger text into the form the binder consumes.
183///
184/// A generator that produced text the parser refuses would be a defect this
185/// function cannot repair, so it returns `None` and the caller enforces
186/// nothing - which is caught by the tests rather than by a user.
187fn build(sql: &str, name: Vec<u8>, limits: &Limits) -> Option<TriggerInfo> {
188    let parsed = parse_next_statement(sql.as_bytes(), 0, limits).ok()?;
189    let crate::ast::Statement::CreateTrigger {
190        time,
191        event,
192        when,
193        body,
194        ..
195    } = &parsed.statement
196    else {
197        return None;
198    };
199    let event = match event {
200        crate::ast::TriggerEvent::Insert => TriggerEventInfo::Insert,
201        crate::ast::TriggerEvent::Delete => TriggerEventInfo::Delete,
202        crate::ast::TriggerEvent::Update(columns) => TriggerEventInfo::Update(
203            columns
204                .iter()
205                .map(|column| parsed.ast.folded(*column).to_vec())
206                .collect(),
207        ),
208    };
209    Some(TriggerInfo {
210        folded: name.to_ascii_lowercase(),
211        name,
212        time: time.unwrap_or(TriggerTime::Before),
213        event,
214        when: *when,
215        body: body.clone(),
216        ast: parsed.ast,
217    })
218}
219
220/// Generates the child's check: a row whose key is complete must have a parent.
221///
222/// A key with a NULL in it is not checked at all. That is `MATCH SIMPLE`, which
223/// is the only match mode SQLite implements whatever the clause says, and it is
224/// why the guard is a conjunction of `IS NOT NULL` rather than a single test.
225fn child_check(
226    child: &TableInfo,
227    parent: &TableInfo,
228    key: &ForeignKeyInfo,
229    event: ForeignKeyEvent,
230    database: &[u8],
231    near: &[Vec<u8>],
232    far: &[Vec<u8>],
233) -> String {
234    let mut guards: Vec<String> = near
235        .iter()
236        .map(|column| format!("NEW.{} IS NOT NULL", quote(column)))
237        .collect();
238    let lookup = children_of(far, near, "NEW");
239    guards.push(format!(
240        "NOT EXISTS (SELECT 1 FROM {} WHERE {lookup})",
241        qualified(database, &parent.name)
242    ));
243    let fires = match event {
244        ForeignKeyEvent::ChildUpdate => format!("BEFORE UPDATE OF {} ON", column_list(near)),
245        _ => "BEFORE INSERT ON".to_string(),
246    };
247    format!(
248        "CREATE TRIGGER {} {fires} {} BEGIN SELECT RAISE(ABORT, '{VIOLATION_MESSAGE}') WHERE {}; END",
249        quote(&trigger_name(child, key, event)),
250        quote(&child.name),
251        conjunction(&guards)
252    )
253}
254
255/// Generates what happens to the children when a parent row goes or changes.
256fn parent_action(
257    child: &TableInfo,
258    parent: &TableInfo,
259    key: &ForeignKeyInfo,
260    event: ForeignKeyEvent,
261    database: &[u8],
262    near: &[Vec<u8>],
263    far: &[Vec<u8>],
264    deferred: bool,
265) -> Option<String> {
266    let action = match event {
267        ForeignKeyEvent::ParentDelete => key.on_delete,
268        _ => key.on_update,
269    };
270    let matching = children_of(near, far, "OLD");
271    let target = qualified(database, &child.name);
272    let body = match action {
273        ReferentialAction::NoAction | ReferentialAction::Restrict => {
274            // RESTRICT is not deferrable: it refuses the write where it
275            // happens, whatever the constraint's timing says. NO ACTION with a
276            // deferred constraint is checked when the transaction commits, so
277            // there is no trigger for it here.
278            if deferred && action == ReferentialAction::NoAction {
279                return None;
280            }
281            format!(
282                "SELECT RAISE(ABORT, '{VIOLATION_MESSAGE}') WHERE EXISTS (SELECT 1 FROM {target} WHERE {matching});"
283            )
284        }
285        ReferentialAction::Cascade => match event {
286            ForeignKeyEvent::ParentDelete => {
287                format!("DELETE FROM {target} WHERE {matching};")
288            }
289            _ => {
290                let sets: Vec<String> = near
291                    .iter()
292                    .zip(far.iter())
293                    .map(|(child_column, parent_column)| {
294                        format!("{} = NEW.{}", quote(child_column), quote(parent_column))
295                    })
296                    .collect();
297                format!("UPDATE {target} SET {} WHERE {matching};", sets.join(", "))
298            }
299        },
300        ReferentialAction::SetNull => {
301            let sets: Vec<String> = near
302                .iter()
303                .map(|column| format!("{} = NULL", quote(column)))
304                .collect();
305            format!("UPDATE {target} SET {} WHERE {matching};", sets.join(", "))
306        }
307        ReferentialAction::SetDefault => {
308            let mut sets = Vec::with_capacity(near.len());
309            for (position, column) in key.columns.iter().zip(near.iter()) {
310                let default = child
311                    .columns
312                    .get(usize::from(*position))
313                    .and_then(|info| info.default_sql.clone())
314                    .unwrap_or_else(|| b"NULL".to_vec());
315                sets.push(format!(
316                    "{} = ({})",
317                    quote(column),
318                    String::from_utf8_lossy(&default)
319                ));
320            }
321            format!("UPDATE {target} SET {} WHERE {matching};", sets.join(", "))
322        }
323    };
324    // RESTRICT fires before the parent row is written, the rest afterwards.
325    // The difference is visible: a `BEFORE DELETE` trigger that removes the
326    // children itself satisfies NO ACTION and does not satisfy RESTRICT.
327    let time = if action == ReferentialAction::Restrict {
328        "BEFORE"
329    } else {
330        "AFTER"
331    };
332    let fires = match event {
333        ForeignKeyEvent::ParentDelete => format!("{time} DELETE ON"),
334        _ => format!("{time} UPDATE OF {} ON", column_list(far)),
335    };
336    // An update that leaves the key alone is not a change to the key, and
337    // firing for it would cascade a row onto itself.
338    let guard = match event {
339        ForeignKeyEvent::ParentUpdate => {
340            let changed: Vec<String> = far
341                .iter()
342                .map(|column| {
343                    let name = quote(column);
344                    format!("OLD.{name} IS NOT NEW.{name}")
345                })
346                .collect();
347            format!(" WHEN {}", changed.join(" OR "))
348        }
349        _ => String::new(),
350    };
351    Some(format!(
352        "CREATE TRIGGER {} {fires} {}{guard} BEGIN {body} END",
353        quote(&trigger_name(child, key, event)),
354        quote(&parent.name)
355    ))
356}
357
358/// Renders a comma-separated list of quoted column names.
359fn column_list(columns: &[Vec<u8>]) -> String {
360    columns
361        .iter()
362        .map(|column| quote(column))
363        .collect::<Vec<_>>()
364        .join(", ")
365}
366
367/// Builds the triggers every table's writes fire because of a foreign key.
368///
369/// It runs once per schema, over every table at once, because that is the only
370/// point at which both sides of a key are visible: a child records the key and
371/// nothing records the reverse direction, so the parent's side is found by
372/// asking every table what it points at.
373///
374/// A key that cannot be enforced - a parent that is not there, or parent
375/// columns that are not a key of the parent - produces an entry with no trigger
376/// and the message to report. That is SQLite's timing: the schema loads, and
377/// the first write that needs the constraint is what fails.
378pub fn plan_schema(tables: &mut [TableInfo], database: &[u8], limits: &Limits) {
379    mark_cycles(tables);
380    let snapshot: Vec<TableInfo> = tables.to_vec();
381    for table in tables.iter_mut() {
382        if table.kind != TableKind::Table {
383            continue;
384        }
385        table.foreign_key_triggers = plan_table(table, &snapshot, database, limits);
386    }
387}
388
389/// Marks every key whose parent can lead back to its own child table.
390///
391/// The graph is small - one node per table, one edge per key - so the search is
392/// a plain walk from each key's parent looking for its child. What it answers
393/// is whether applying this key's action can fire the same key again.
394fn mark_cycles(tables: &mut [TableInfo]) {
395    let edges: Vec<(Vec<u8>, Vec<u8>)> = tables
396        .iter()
397        .flat_map(|table| {
398            table
399                .foreign_keys
400                .iter()
401                .map(|key| (table.folded.clone(), key.parent_folded.clone()))
402        })
403        .collect();
404    for table in tables.iter_mut() {
405        for key in &mut table.foreign_keys {
406            key.cyclic = reaches(&edges, &key.parent_folded, &table.folded);
407        }
408    }
409}
410
411/// Reports whether `from` can reach `wanted` by following child-to-parent
412/// edges backwards, which is the direction an action travels.
413fn reaches(edges: &[(Vec<u8>, Vec<u8>)], from: &[u8], wanted: &[u8]) -> bool {
414    let mut seen: Vec<Vec<u8>> = Vec::new();
415    let mut pending: Vec<Vec<u8>> = vec![from.to_vec()];
416    while let Some(table) = pending.pop() {
417        if table == wanted {
418            return true;
419        }
420        if seen.contains(&table) {
421            continue;
422        }
423        seen.push(table.clone());
424        for (child, parent) in edges {
425            if *child == table {
426                pending.push(parent.clone());
427            }
428        }
429    }
430    false
431}
432
433/// Returns the statement that repairs one cyclic key, or `None` when the key
434/// has nothing to repair.
435///
436/// This is the other half of a cyclic action. The trigger takes the first
437/// level - the rows that pointed directly at the row that went - and this
438/// takes what that leaves: every row whose key now has no parent. Repeating it
439/// until nothing changes reaches the leaves, however deep they are, and it
440/// terminates because every pass either changes a row or stops.
441///
442/// `NO ACTION` and `RESTRICT` are absent on purpose: they refuse rather than
443/// repair, and the trigger has already refused.
444pub fn sweep_statement(
445    child: &TableInfo,
446    parent: &TableInfo,
447    key: &ForeignKeyInfo,
448    database: &[u8],
449) -> Option<String> {
450    let near = child_columns(key, child)?;
451    let far = parent_columns(key, parent)?;
452    if near.len() != far.len() || near.is_empty() {
453        return None;
454    }
455    let outer = quote(&child.name);
456    let mut guards: Vec<String> = near
457        .iter()
458        .map(|column| format!("{outer}.{} IS NOT NULL", quote(column)))
459        .collect();
460    let lookup: Vec<String> = far
461        .iter()
462        .zip(near.iter())
463        .map(|(parent_column, child_column)| {
464            format!(
465                "p.{} = {outer}.{}",
466                quote(parent_column),
467                quote(child_column)
468            )
469        })
470        .collect();
471    guards.push(format!(
472        "NOT EXISTS (SELECT 1 FROM {} AS p WHERE {})",
473        qualified(database, &parent.name),
474        conjunction(&lookup)
475    ));
476    let target = qualified(database, &child.name);
477    let where_clause = conjunction(&guards);
478    match key.on_delete {
479        ReferentialAction::Cascade => Some(format!("DELETE FROM {target} WHERE {where_clause}")),
480        ReferentialAction::SetNull => {
481            let sets: Vec<String> = near
482                .iter()
483                .map(|column| format!("{} = NULL", quote(column)))
484                .collect();
485            Some(format!(
486                "UPDATE {target} SET {} WHERE {where_clause}",
487                sets.join(", ")
488            ))
489        }
490        ReferentialAction::SetDefault => {
491            let mut sets = Vec::with_capacity(near.len());
492            for (position, column) in key.columns.iter().zip(near.iter()) {
493                let default = child
494                    .columns
495                    .get(usize::from(*position))
496                    .and_then(|info| info.default_sql.clone())
497                    .unwrap_or_else(|| b"NULL".to_vec());
498                sets.push(format!(
499                    "{} = ({})",
500                    quote(column),
501                    String::from_utf8_lossy(&default)
502                ));
503            }
504            Some(format!(
505                "UPDATE {target} SET {} WHERE {where_clause}",
506                sets.join(", ")
507            ))
508        }
509        ReferentialAction::NoAction | ReferentialAction::Restrict => None,
510    }
511}
512
513/// Builds the entries for one table, both directions.
514fn plan_table(
515    table: &TableInfo,
516    tables: &[TableInfo],
517    database: &[u8],
518    limits: &Limits,
519) -> Vec<ForeignKeyTrigger> {
520    let mut planned = Vec::new();
521    for key in &table.foreign_keys {
522        let parent = tables
523            .iter()
524            .find(|candidate| candidate.folded == key.parent_folded);
525        let Some(parent) = parent else {
526            planned.push(unusable(
527                key,
528                format!(
529                    "no such table: {}.{}",
530                    String::from_utf8_lossy(database),
531                    String::from_utf8_lossy(&key.parent)
532                ),
533                true,
534                key.parent_folded == table.folded,
535            ));
536            continue;
537        };
538        if !parent_key_is_unique(parent, key) {
539            planned.push(unusable(
540                key,
541                mismatch(table, parent),
542                true,
543                parent.folded == table.folded,
544            ));
545            continue;
546        }
547        for event in [ForeignKeyEvent::ChildInsert, ForeignKeyEvent::ChildUpdate] {
548            if let Some(trigger) = trigger_for(table, parent, key, event, database, false, limits) {
549                planned.push(ForeignKeyTrigger {
550                    is_check: true,
551                    deferred: key.is_deferred(),
552                    trigger: Some(trigger),
553                    fault: Vec::new(),
554                    self_referencing: parent.folded == table.folded,
555                });
556            }
557        }
558    }
559    for child in tables {
560        if child.kind != TableKind::Table {
561            continue;
562        }
563        for key in &child.foreign_keys {
564            if key.parent_folded != table.folded {
565                continue;
566            }
567            if !parent_key_is_unique(table, key) {
568                planned.push(unusable(
569                    key,
570                    mismatch(child, table),
571                    false,
572                    child.folded == table.folded,
573                ));
574                continue;
575            }
576            for event in [ForeignKeyEvent::ParentDelete, ForeignKeyEvent::ParentUpdate] {
577                let Some(trigger) = trigger_for(child, table, key, event, database, false, limits)
578                else {
579                    continue;
580                };
581                let action = match event {
582                    ForeignKeyEvent::ParentDelete => key.on_delete,
583                    _ => key.on_update,
584                };
585                planned.push(ForeignKeyTrigger {
586                    // RESTRICT refuses, and is never deferred; NO ACTION
587                    // refuses and is deferred with its key; the three that
588                    // repair are not checks at all.
589                    is_check: action == ReferentialAction::NoAction,
590                    deferred: key.is_deferred(),
591                    trigger: Some(trigger),
592                    fault: Vec::new(),
593                    self_referencing: child.folded == table.folded,
594                });
595            }
596        }
597    }
598    planned
599}
600
601/// Returns the message SQLite reports for a key whose parent does not match.
602fn mismatch(child: &TableInfo, parent: &TableInfo) -> String {
603    format!(
604        "foreign key mismatch - \"{}\" referencing \"{}\"",
605        String::from_utf8_lossy(&child.name),
606        String::from_utf8_lossy(&parent.name)
607    )
608}
609
610/// Returns an entry that reports a fault instead of enforcing anything.
611///
612/// @param key - the key that cannot be enforced
613/// @param message - what to report when something writes
614/// @param is_check - whether it would have refused rather than repaired
615/// @param self_referencing - whether the key's child and parent are one table
616fn unusable(
617    key: &ForeignKeyInfo,
618    message: String,
619    is_check: bool,
620    self_referencing: bool,
621) -> ForeignKeyTrigger {
622    ForeignKeyTrigger {
623        is_check,
624        deferred: key.is_deferred(),
625        trigger: None,
626        fault: message.into_bytes(),
627        self_referencing,
628    }
629}
630
631/// Reports whether a key's parent columns are a key of the parent.
632///
633/// SQLite requires it: the parent columns must be the primary key or carry a
634/// UNIQUE index, because a key that could match two parent rows would make
635/// `ON DELETE CASCADE` ambiguous. A parent that does not satisfy it is a
636/// `foreign key mismatch`, reported when something writes.
637pub fn parent_key_is_unique(parent: &TableInfo, key: &ForeignKeyInfo) -> bool {
638    let Some(wanted) = parent_columns(key, parent) else {
639        return false;
640    };
641    let folded: Vec<Vec<u8>> = wanted
642        .iter()
643        .map(|name| name.to_ascii_lowercase())
644        .collect();
645    // A single column that is the rowid alias is the table's own key.
646    if folded.len() == 1 {
647        if let Some(alias) = parent.rowid_alias {
648            if let Some(column) = parent.columns.get(usize::from(alias)) {
649                if folded.first() == Some(&column.folded) {
650                    return true;
651                }
652            }
653        }
654    }
655    let primary = parent.primary_key();
656    if !primary.is_empty() && primary.len() == folded.len() {
657        let names: Vec<Vec<u8>> = primary
658            .iter()
659            .filter_map(|position| parent.columns.get(usize::from(*position)))
660            .map(|column| column.folded.clone())
661            .collect();
662        if same_set(&names, &folded) {
663            return true;
664        }
665    }
666    parent.indexes.iter().any(|index| {
667        index.unique && index.columns.len() == folded.len() && {
668            let names: Vec<Vec<u8>> = index
669                .columns
670                .iter()
671                .filter_map(|key| key.column)
672                .filter_map(|position| parent.columns.get(usize::from(position)))
673                .map(|column| column.folded.clone())
674                .collect();
675            same_set(&names, &folded)
676        }
677    })
678}
679
680/// Reports whether two column lists name the same columns, in any order.
681///
682/// Order does not matter to a key: `REFERENCES p(a, b)` is satisfied by a
683/// unique index on `(b, a)`, because either one makes the pair unique.
684fn same_set(left: &[Vec<u8>], right: &[Vec<u8>]) -> bool {
685    left.len() == right.len() && right.iter().all(|name| left.contains(name))
686}
687
688/// Returns the `SELECT` that finds every row of a child table whose key has no
689/// parent, which is what `PRAGMA foreign_key_check` reports and what a deferred
690/// constraint is tested with at commit.
691///
692/// It is a query rather than a scan written by hand, so it uses the planner and
693/// the indexes an ordinary query would - a check over a million-row child with
694/// an index on its key is an index lookup per row, not a second scan.
695pub fn violation_query(
696    child: &TableInfo,
697    parent: &TableInfo,
698    key: &ForeignKeyInfo,
699    database: &[u8],
700) -> Option<String> {
701    let near = child_columns(key, child)?;
702    let far = parent_columns(key, parent)?;
703    if near.len() != far.len() || near.is_empty() {
704        return None;
705    }
706    let mut guards: Vec<String> = near
707        .iter()
708        .map(|column| format!("c.{} IS NOT NULL", quote(column)))
709        .collect();
710    let lookup: Vec<String> = far
711        .iter()
712        .zip(near.iter())
713        .map(|(parent_column, child_column)| {
714            format!("p.{} = c.{}", quote(parent_column), quote(child_column))
715        })
716        .collect();
717    guards.push(format!(
718        "NOT EXISTS (SELECT 1 FROM {} AS p WHERE {})",
719        qualified(database, &parent.name),
720        conjunction(&lookup)
721    ));
722    // A WITHOUT ROWID table has no rowid to report, and SQLite prints NULL
723    // for it rather than refusing to check the table.
724    let identity = if child.without_rowid {
725        "NULL"
726    } else {
727        "c.rowid"
728    };
729    Some(format!(
730        "SELECT {identity} FROM {} AS c WHERE {}",
731        qualified(database, &child.name),
732        conjunction(&guards)
733    ))
734}
735
736#[cfg(test)]
737mod tests {
738    use super::*;
739    use crate::catalog_view::{ColumnInfo, TableKind};
740    use inillucent_value::Affinity;
741
742    /// Builds a table with the named columns, for the generator tests.
743    fn table(name: &[u8], columns: &[&[u8]]) -> TableInfo {
744        TableInfo {
745            name: name.to_vec(),
746            folded: name.to_ascii_lowercase(),
747            database: 0,
748            root: 2,
749            columns: columns
750                .iter()
751                .map(|column| ColumnInfo {
752                    name: column.to_vec(),
753                    folded: column.to_ascii_lowercase(),
754                    declared_type: Vec::new(),
755                    affinity: Affinity::Blob,
756                    collation: b"binary".to_vec(),
757                    not_null: false,
758                    not_null_conflict: None,
759                    primary_key_conflict: None,
760                    default_sql: None,
761                    primary_key_position: None,
762                    hidden: false,
763                    generated: false,
764                    stored: false,
765                    generated_sql: None,
766                })
767                .collect(),
768            rowid_alias: None,
769            without_rowid: false,
770            strict: false,
771            autoincrement: false,
772            kind: TableKind::Table,
773            create_sql: Vec::new(),
774            indexes: Vec::new(),
775            view: None,
776            triggers: Vec::new(),
777            analysed_rows: None,
778            checks: Vec::new(),
779            foreign_keys: Vec::new(),
780            foreign_key_triggers: Vec::new(),
781            module: None,
782        }
783    }
784
785    /// Builds a key over one child column pointing at one parent column.
786    fn key(on_delete: ReferentialAction, on_update: ReferentialAction) -> ForeignKeyInfo {
787        ForeignKeyInfo {
788            id: 0,
789            columns: vec![1],
790            parent: b"p".to_vec(),
791            parent_folded: b"p".to_vec(),
792            parent_columns: vec![b"id".to_vec()],
793            on_delete,
794            on_update,
795            match_clause: Vec::new(),
796            deferrable: false,
797            initially_deferred: false,
798            cyclic: false,
799        }
800    }
801
802    /// Every generated trigger has to parse. A generator that produced text the
803    /// parser refuses would enforce nothing at all, silently.
804    #[test]
805    fn every_generated_trigger_parses() {
806        let child = table(b"c", &[b"id", b"pid"]);
807        let parent = table(b"p", &[b"id"]);
808        let limits = Limits::default();
809        let actions = [
810            ReferentialAction::NoAction,
811            ReferentialAction::Restrict,
812            ReferentialAction::Cascade,
813            ReferentialAction::SetNull,
814            ReferentialAction::SetDefault,
815        ];
816        let events = [
817            ForeignKeyEvent::ChildInsert,
818            ForeignKeyEvent::ChildUpdate,
819            ForeignKeyEvent::ParentDelete,
820            ForeignKeyEvent::ParentUpdate,
821        ];
822        for action in actions {
823            let key = key(action, action);
824            for event in events {
825                let built = trigger_for(&child, &parent, &key, event, b"main", false, &limits);
826                assert!(
827                    built.is_some(),
828                    "{action:?} on {event:?} produced no trigger"
829                );
830            }
831        }
832    }
833
834    /// The child's check fires before the write, tests every key column for
835    /// NULL, and looks the parent up by the columns the clause named.
836    #[test]
837    fn the_child_check_reads_as_it_should() {
838        let child = table(b"c", &[b"id", b"pid"]);
839        let parent = table(b"p", &[b"id"]);
840        let key = key(ReferentialAction::NoAction, ReferentialAction::NoAction);
841        let sql = child_check(
842            &child,
843            &parent,
844            &key,
845            ForeignKeyEvent::ChildInsert,
846            b"main",
847            &[b"pid".to_vec()],
848            &[b"id".to_vec()],
849        );
850        assert!(sql.contains("BEFORE INSERT ON \"c\""), "{sql}");
851        assert!(sql.contains("NEW.\"pid\" IS NOT NULL"), "{sql}");
852        assert!(sql.contains("NOT EXISTS"), "{sql}");
853        assert!(sql.contains("FOREIGN KEY constraint failed"), "{sql}");
854    }
855
856    /// RESTRICT fires before the parent write and NO ACTION after it, which is
857    /// the one place the two differ.
858    #[test]
859    fn restrict_fires_before_and_no_action_after() {
860        let child = table(b"c", &[b"id", b"pid"]);
861        let parent = table(b"p", &[b"id"]);
862        let limits = Limits::default();
863        for (action, expected) in [
864            (ReferentialAction::Restrict, TriggerTime::Before),
865            (ReferentialAction::NoAction, TriggerTime::After),
866        ] {
867            let key = key(action, action);
868            let built = trigger_for(
869                &child,
870                &parent,
871                &key,
872                ForeignKeyEvent::ParentDelete,
873                b"main",
874                false,
875                &limits,
876            )
877            .expect("the trigger is generated");
878            assert_eq!(built.time, expected, "{action:?}");
879        }
880    }
881
882    /// A deferred constraint generates no check on the child and no NO ACTION
883    /// on the parent - both wait for the commit - but RESTRICT and the cascades
884    /// still fire where they are.
885    #[test]
886    fn a_deferred_key_defers_only_its_checks() {
887        let child = table(b"c", &[b"id", b"pid"]);
888        let parent = table(b"p", &[b"id"]);
889        let limits = Limits::default();
890        let deferred = key(ReferentialAction::NoAction, ReferentialAction::NoAction);
891        assert!(trigger_for(
892            &child,
893            &parent,
894            &deferred,
895            ForeignKeyEvent::ChildInsert,
896            b"main",
897            true,
898            &limits
899        )
900        .is_none());
901        assert!(trigger_for(
902            &child,
903            &parent,
904            &deferred,
905            ForeignKeyEvent::ParentDelete,
906            b"main",
907            true,
908            &limits
909        )
910        .is_none());
911        let restrict = key(ReferentialAction::Restrict, ReferentialAction::Restrict);
912        assert!(trigger_for(
913            &child,
914            &parent,
915            &restrict,
916            ForeignKeyEvent::ParentDelete,
917            b"main",
918            true,
919            &limits
920        )
921        .is_some());
922        let cascade = key(ReferentialAction::Cascade, ReferentialAction::Cascade);
923        assert!(trigger_for(
924            &child,
925            &parent,
926            &cascade,
927            ForeignKeyEvent::ParentDelete,
928            b"main",
929            true,
930            &limits
931        )
932        .is_some());
933    }
934
935    /// A parent update fires only when the key actually changed, and cascades
936    /// the new key onto the rows that carried the old one.
937    #[test]
938    fn a_parent_update_guards_on_the_key_changing() {
939        let child = table(b"c", &[b"id", b"pid"]);
940        let parent = table(b"p", &[b"id"]);
941        let key = key(ReferentialAction::Cascade, ReferentialAction::Cascade);
942        let sql = parent_action(
943            &child,
944            &parent,
945            &key,
946            ForeignKeyEvent::ParentUpdate,
947            b"main",
948            &[b"pid".to_vec()],
949            &[b"id".to_vec()],
950            false,
951        )
952        .expect("the trigger is generated");
953        assert!(sql.contains("AFTER UPDATE OF \"id\""), "{sql}");
954        assert!(sql.contains("WHEN OLD.\"id\" IS NOT NEW.\"id\""), "{sql}");
955        assert!(sql.contains("SET \"pid\" = NEW.\"id\""), "{sql}");
956        assert!(sql.contains("WHERE \"pid\" = OLD.\"id\""), "{sql}");
957    }
958
959    /// An identifier with a quote in it survives the round trip, because the
960    /// generated text is parsed again rather than merely printed.
961    #[test]
962    fn an_awkward_identifier_is_quoted() {
963        assert_eq!(quote(b"we\"ird"), "\"we\"\"ird\"");
964        let child = table(b"we\"ird", &[b"id", b"pid"]);
965        let parent = table(b"p", &[b"id"]);
966        let key = key(ReferentialAction::Cascade, ReferentialAction::Cascade);
967        let limits = Limits::default();
968        assert!(trigger_for(
969            &child,
970            &parent,
971            &key,
972            ForeignKeyEvent::ChildInsert,
973            b"main",
974            false,
975            &limits
976        )
977        .is_some());
978    }
979
980    /// A composite key compares every column, in the order the clause wrote.
981    #[test]
982    fn a_composite_key_compares_every_column() {
983        let matching = children_of(
984            &[b"a".to_vec(), b"b".to_vec()],
985            &[b"x".to_vec(), b"y".to_vec()],
986            "OLD",
987        );
988        assert_eq!(matching, "\"a\" = OLD.\"x\" AND \"b\" = OLD.\"y\"");
989    }
990}