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        table_database: None,
218    })
219}
220
221/// Generates the child's check: a row whose key is complete must have a parent.
222///
223/// A key with a NULL in it is not checked at all. That is `MATCH SIMPLE`, which
224/// is the only match mode SQLite implements whatever the clause says, and it is
225/// why the guard is a conjunction of `IS NOT NULL` rather than a single test.
226fn child_check(
227    child: &TableInfo,
228    parent: &TableInfo,
229    key: &ForeignKeyInfo,
230    event: ForeignKeyEvent,
231    database: &[u8],
232    near: &[Vec<u8>],
233    far: &[Vec<u8>],
234) -> String {
235    let mut guards: Vec<String> = near
236        .iter()
237        .map(|column| format!("NEW.{} IS NOT NULL", quote(column)))
238        .collect();
239    let lookup = children_of(far, near, "NEW");
240    guards.push(format!(
241        "NOT EXISTS (SELECT 1 FROM {} WHERE {lookup})",
242        qualified(database, &parent.name)
243    ));
244    let fires = match event {
245        ForeignKeyEvent::ChildUpdate => format!("BEFORE UPDATE OF {} ON", column_list(near)),
246        _ => "BEFORE INSERT ON".to_string(),
247    };
248    format!(
249        "CREATE TRIGGER {} {fires} {} BEGIN SELECT RAISE(ABORT, '{VIOLATION_MESSAGE}') WHERE {}; END",
250        quote(&trigger_name(child, key, event)),
251        quote(&child.name),
252        conjunction(&guards)
253    )
254}
255
256/// Generates what happens to the children when a parent row goes or changes.
257fn parent_action(
258    child: &TableInfo,
259    parent: &TableInfo,
260    key: &ForeignKeyInfo,
261    event: ForeignKeyEvent,
262    database: &[u8],
263    near: &[Vec<u8>],
264    far: &[Vec<u8>],
265    deferred: bool,
266) -> Option<String> {
267    let action = match event {
268        ForeignKeyEvent::ParentDelete => key.on_delete,
269        _ => key.on_update,
270    };
271    let matching = children_of(near, far, "OLD");
272    let target = qualified(database, &child.name);
273    let body = match action {
274        ReferentialAction::NoAction | ReferentialAction::Restrict => {
275            // RESTRICT is not deferrable: it refuses the write where it
276            // happens, whatever the constraint's timing says. NO ACTION with a
277            // deferred constraint is checked when the transaction commits, so
278            // there is no trigger for it here.
279            if deferred && action == ReferentialAction::NoAction {
280                return None;
281            }
282            format!(
283                "SELECT RAISE(ABORT, '{VIOLATION_MESSAGE}') WHERE EXISTS (SELECT 1 FROM {target} WHERE {matching});"
284            )
285        }
286        ReferentialAction::Cascade => match event {
287            ForeignKeyEvent::ParentDelete => {
288                format!("DELETE FROM {target} WHERE {matching};")
289            }
290            _ => {
291                let sets: Vec<String> = near
292                    .iter()
293                    .zip(far.iter())
294                    .map(|(child_column, parent_column)| {
295                        format!("{} = NEW.{}", quote(child_column), quote(parent_column))
296                    })
297                    .collect();
298                format!("UPDATE {target} SET {} WHERE {matching};", sets.join(", "))
299            }
300        },
301        ReferentialAction::SetNull => {
302            let sets: Vec<String> = near
303                .iter()
304                .map(|column| format!("{} = NULL", quote(column)))
305                .collect();
306            format!("UPDATE {target} SET {} WHERE {matching};", sets.join(", "))
307        }
308        ReferentialAction::SetDefault => {
309            let mut sets = Vec::with_capacity(near.len());
310            for (position, column) in key.columns.iter().zip(near.iter()) {
311                let default = child
312                    .columns
313                    .get(usize::from(*position))
314                    .and_then(|info| info.default_sql.clone())
315                    .unwrap_or_else(|| b"NULL".to_vec());
316                sets.push(format!(
317                    "{} = ({})",
318                    quote(column),
319                    String::from_utf8_lossy(&default)
320                ));
321            }
322            format!("UPDATE {target} SET {} WHERE {matching};", sets.join(", "))
323        }
324    };
325    // RESTRICT fires before the parent row is written, the rest afterwards.
326    // The difference is visible: a `BEFORE DELETE` trigger that removes the
327    // children itself satisfies NO ACTION and does not satisfy RESTRICT.
328    let time = if action == ReferentialAction::Restrict {
329        "BEFORE"
330    } else {
331        "AFTER"
332    };
333    let fires = match event {
334        ForeignKeyEvent::ParentDelete => format!("{time} DELETE ON"),
335        _ => format!("{time} UPDATE OF {} ON", column_list(far)),
336    };
337    // An update that leaves the key alone is not a change to the key, and
338    // firing for it would cascade a row onto itself.
339    let guard = match event {
340        ForeignKeyEvent::ParentUpdate => {
341            let changed: Vec<String> = far
342                .iter()
343                .map(|column| {
344                    let name = quote(column);
345                    format!("OLD.{name} IS NOT NEW.{name}")
346                })
347                .collect();
348            format!(" WHEN {}", changed.join(" OR "))
349        }
350        _ => String::new(),
351    };
352    Some(format!(
353        "CREATE TRIGGER {} {fires} {}{guard} BEGIN {body} END",
354        quote(&trigger_name(child, key, event)),
355        quote(&parent.name)
356    ))
357}
358
359/// Renders a comma-separated list of quoted column names.
360fn column_list(columns: &[Vec<u8>]) -> String {
361    columns
362        .iter()
363        .map(|column| quote(column))
364        .collect::<Vec<_>>()
365        .join(", ")
366}
367
368/// Builds the triggers every table's writes fire because of a foreign key.
369///
370/// It runs once per schema, over every table at once, because that is the only
371/// point at which both sides of a key are visible: a child records the key and
372/// nothing records the reverse direction, so the parent's side is found by
373/// asking every table what it points at.
374///
375/// A key that cannot be enforced - a parent that is not there, or parent
376/// columns that are not a key of the parent - produces an entry with no trigger
377/// and the message to report. That is SQLite's timing: the schema loads, and
378/// the first write that needs the constraint is what fails.
379pub fn plan_schema(tables: &mut [TableInfo], database: &[u8], limits: &Limits) {
380    mark_cycles(tables);
381    let snapshot: Vec<TableInfo> = tables.to_vec();
382    for table in tables.iter_mut() {
383        if table.kind != TableKind::Table {
384            continue;
385        }
386        table.foreign_key_triggers = plan_table(table, &snapshot, database, limits);
387    }
388}
389
390/// Marks every key whose parent can lead back to its own child table.
391///
392/// The graph is small - one node per table, one edge per key - so the search is
393/// a plain walk from each key's parent looking for its child. What it answers
394/// is whether applying this key's action can fire the same key again.
395fn mark_cycles(tables: &mut [TableInfo]) {
396    let edges: Vec<(Vec<u8>, Vec<u8>)> = tables
397        .iter()
398        .flat_map(|table| {
399            table
400                .foreign_keys
401                .iter()
402                .map(|key| (table.folded.clone(), key.parent_folded.clone()))
403        })
404        .collect();
405    for table in tables.iter_mut() {
406        for key in &mut table.foreign_keys {
407            key.cyclic = reaches(&edges, &key.parent_folded, &table.folded);
408        }
409    }
410}
411
412/// Reports whether `from` can reach `wanted` by following child-to-parent
413/// edges backwards, which is the direction an action travels.
414fn reaches(edges: &[(Vec<u8>, Vec<u8>)], from: &[u8], wanted: &[u8]) -> bool {
415    let mut seen: Vec<Vec<u8>> = Vec::new();
416    let mut pending: Vec<Vec<u8>> = vec![from.to_vec()];
417    while let Some(table) = pending.pop() {
418        if table == wanted {
419            return true;
420        }
421        if seen.contains(&table) {
422            continue;
423        }
424        seen.push(table.clone());
425        for (child, parent) in edges {
426            if *child == table {
427                pending.push(parent.clone());
428            }
429        }
430    }
431    false
432}
433
434/// Returns the statement that repairs one cyclic key, or `None` when the key
435/// has nothing to repair.
436///
437/// This is the other half of a cyclic action. The trigger takes the first
438/// level - the rows that pointed directly at the row that went - and this
439/// takes what that leaves: every row whose key now has no parent. Repeating it
440/// until nothing changes reaches the leaves, however deep they are, and it
441/// terminates because every pass either changes a row or stops.
442///
443/// `NO ACTION` and `RESTRICT` are absent on purpose: they refuse rather than
444/// repair, and the trigger has already refused.
445pub fn sweep_statement(
446    child: &TableInfo,
447    parent: &TableInfo,
448    key: &ForeignKeyInfo,
449    database: &[u8],
450) -> Option<String> {
451    let near = child_columns(key, child)?;
452    let far = parent_columns(key, parent)?;
453    if near.len() != far.len() || near.is_empty() {
454        return None;
455    }
456    let outer = quote(&child.name);
457    let mut guards: Vec<String> = near
458        .iter()
459        .map(|column| format!("{outer}.{} IS NOT NULL", quote(column)))
460        .collect();
461    let lookup: Vec<String> = far
462        .iter()
463        .zip(near.iter())
464        .map(|(parent_column, child_column)| {
465            format!(
466                "p.{} = {outer}.{}",
467                quote(parent_column),
468                quote(child_column)
469            )
470        })
471        .collect();
472    guards.push(format!(
473        "NOT EXISTS (SELECT 1 FROM {} AS p WHERE {})",
474        qualified(database, &parent.name),
475        conjunction(&lookup)
476    ));
477    let target = qualified(database, &child.name);
478    let where_clause = conjunction(&guards);
479    match key.on_delete {
480        ReferentialAction::Cascade => Some(format!("DELETE FROM {target} WHERE {where_clause}")),
481        ReferentialAction::SetNull => {
482            let sets: Vec<String> = near
483                .iter()
484                .map(|column| format!("{} = NULL", quote(column)))
485                .collect();
486            Some(format!(
487                "UPDATE {target} SET {} WHERE {where_clause}",
488                sets.join(", ")
489            ))
490        }
491        ReferentialAction::SetDefault => {
492            let mut sets = Vec::with_capacity(near.len());
493            for (position, column) in key.columns.iter().zip(near.iter()) {
494                let default = child
495                    .columns
496                    .get(usize::from(*position))
497                    .and_then(|info| info.default_sql.clone())
498                    .unwrap_or_else(|| b"NULL".to_vec());
499                sets.push(format!(
500                    "{} = ({})",
501                    quote(column),
502                    String::from_utf8_lossy(&default)
503                ));
504            }
505            Some(format!(
506                "UPDATE {target} SET {} WHERE {where_clause}",
507                sets.join(", ")
508            ))
509        }
510        ReferentialAction::NoAction | ReferentialAction::Restrict => None,
511    }
512}
513
514/// Builds the entries for one table, both directions.
515fn plan_table(
516    table: &TableInfo,
517    tables: &[TableInfo],
518    database: &[u8],
519    limits: &Limits,
520) -> Vec<ForeignKeyTrigger> {
521    let mut planned = Vec::new();
522    for key in &table.foreign_keys {
523        let parent = tables
524            .iter()
525            .find(|candidate| candidate.folded == key.parent_folded);
526        let Some(parent) = parent else {
527            planned.push(unusable(
528                key,
529                format!(
530                    "no such table: {}.{}",
531                    String::from_utf8_lossy(database),
532                    String::from_utf8_lossy(&key.parent)
533                ),
534                true,
535                key.parent_folded == table.folded,
536            ));
537            continue;
538        };
539        if !parent_key_is_unique(parent, key) {
540            planned.push(unusable(
541                key,
542                mismatch(table, parent),
543                true,
544                parent.folded == table.folded,
545            ));
546            continue;
547        }
548        for event in [ForeignKeyEvent::ChildInsert, ForeignKeyEvent::ChildUpdate] {
549            if let Some(trigger) = trigger_for(table, parent, key, event, database, false, limits) {
550                planned.push(ForeignKeyTrigger {
551                    is_check: true,
552                    deferred: key.is_deferred(),
553                    trigger: Some(trigger),
554                    fault: Vec::new(),
555                    self_referencing: parent.folded == table.folded,
556                });
557            }
558        }
559    }
560    for child in tables {
561        if child.kind != TableKind::Table {
562            continue;
563        }
564        for key in &child.foreign_keys {
565            if key.parent_folded != table.folded {
566                continue;
567            }
568            if !parent_key_is_unique(table, key) {
569                planned.push(unusable(
570                    key,
571                    mismatch(child, table),
572                    false,
573                    child.folded == table.folded,
574                ));
575                continue;
576            }
577            for event in [ForeignKeyEvent::ParentDelete, ForeignKeyEvent::ParentUpdate] {
578                let Some(trigger) = trigger_for(child, table, key, event, database, false, limits)
579                else {
580                    continue;
581                };
582                let action = match event {
583                    ForeignKeyEvent::ParentDelete => key.on_delete,
584                    _ => key.on_update,
585                };
586                planned.push(ForeignKeyTrigger {
587                    // RESTRICT refuses, and is never deferred; NO ACTION
588                    // refuses and is deferred with its key; the three that
589                    // repair are not checks at all.
590                    is_check: action == ReferentialAction::NoAction,
591                    deferred: key.is_deferred(),
592                    trigger: Some(trigger),
593                    fault: Vec::new(),
594                    self_referencing: child.folded == table.folded,
595                });
596            }
597        }
598    }
599    planned
600}
601
602/// Returns the message SQLite reports for a key whose parent does not match.
603fn mismatch(child: &TableInfo, parent: &TableInfo) -> String {
604    format!(
605        "foreign key mismatch - \"{}\" referencing \"{}\"",
606        String::from_utf8_lossy(&child.name),
607        String::from_utf8_lossy(&parent.name)
608    )
609}
610
611/// Returns an entry that reports a fault instead of enforcing anything.
612///
613/// @param key - the key that cannot be enforced
614/// @param message - what to report when something writes
615/// @param is_check - whether it would have refused rather than repaired
616/// @param self_referencing - whether the key's child and parent are one table
617fn unusable(
618    key: &ForeignKeyInfo,
619    message: String,
620    is_check: bool,
621    self_referencing: bool,
622) -> ForeignKeyTrigger {
623    ForeignKeyTrigger {
624        is_check,
625        deferred: key.is_deferred(),
626        trigger: None,
627        fault: message.into_bytes(),
628        self_referencing,
629    }
630}
631
632/// Reports whether a key's parent columns are a key of the parent.
633///
634/// SQLite requires it: the parent columns must be the primary key or carry a
635/// UNIQUE index, because a key that could match two parent rows would make
636/// `ON DELETE CASCADE` ambiguous. A parent that does not satisfy it is a
637/// `foreign key mismatch`, reported when something writes.
638pub fn parent_key_is_unique(parent: &TableInfo, key: &ForeignKeyInfo) -> bool {
639    let Some(wanted) = parent_columns(key, parent) else {
640        return false;
641    };
642    let folded: Vec<Vec<u8>> = wanted
643        .iter()
644        .map(|name| name.to_ascii_lowercase())
645        .collect();
646    // A single column that is the rowid alias is the table's own key.
647    if folded.len() == 1 {
648        if let Some(alias) = parent.rowid_alias {
649            if let Some(column) = parent.columns.get(usize::from(alias)) {
650                if folded.first() == Some(&column.folded) {
651                    return true;
652                }
653            }
654        }
655    }
656    let primary = parent.primary_key();
657    if !primary.is_empty() && primary.len() == folded.len() {
658        let names: Vec<Vec<u8>> = primary
659            .iter()
660            .filter_map(|position| parent.columns.get(usize::from(*position)))
661            .map(|column| column.folded.clone())
662            .collect();
663        if same_set(&names, &folded) {
664            return true;
665        }
666    }
667    parent.indexes.iter().any(|index| {
668        index.unique && index.columns.len() == folded.len() && {
669            let names: Vec<Vec<u8>> = index
670                .columns
671                .iter()
672                .filter_map(|key| key.column)
673                .filter_map(|position| parent.columns.get(usize::from(position)))
674                .map(|column| column.folded.clone())
675                .collect();
676            same_set(&names, &folded)
677        }
678    })
679}
680
681/// Reports whether two column lists name the same columns, in any order.
682///
683/// Order does not matter to a key: `REFERENCES p(a, b)` is satisfied by a
684/// unique index on `(b, a)`, because either one makes the pair unique.
685fn same_set(left: &[Vec<u8>], right: &[Vec<u8>]) -> bool {
686    left.len() == right.len() && right.iter().all(|name| left.contains(name))
687}
688
689/// Returns the `SELECT` that finds every row of a child table whose key has no
690/// parent, which is what `PRAGMA foreign_key_check` reports and what a deferred
691/// constraint is tested with at commit.
692///
693/// It is a query rather than a scan written by hand, so it uses the planner and
694/// the indexes an ordinary query would - a check over a million-row child with
695/// an index on its key is an index lookup per row, not a second scan.
696pub fn violation_query(
697    child: &TableInfo,
698    parent: &TableInfo,
699    key: &ForeignKeyInfo,
700    database: &[u8],
701) -> Option<String> {
702    let near = child_columns(key, child)?;
703    let far = parent_columns(key, parent)?;
704    if near.len() != far.len() || near.is_empty() {
705        return None;
706    }
707    let mut guards: Vec<String> = near
708        .iter()
709        .map(|column| format!("c.{} IS NOT NULL", quote(column)))
710        .collect();
711    let lookup: Vec<String> = far
712        .iter()
713        .zip(near.iter())
714        .map(|(parent_column, child_column)| {
715            format!("p.{} = c.{}", quote(parent_column), quote(child_column))
716        })
717        .collect();
718    guards.push(format!(
719        "NOT EXISTS (SELECT 1 FROM {} AS p WHERE {})",
720        qualified(database, &parent.name),
721        conjunction(&lookup)
722    ));
723    // A WITHOUT ROWID table has no rowid to report, and SQLite prints NULL
724    // for it rather than refusing to check the table.
725    let identity = if child.without_rowid {
726        "NULL"
727    } else {
728        "c.rowid"
729    };
730    Some(format!(
731        "SELECT {identity} FROM {} AS c WHERE {}",
732        qualified(database, &child.name),
733        conjunction(&guards)
734    ))
735}
736
737#[cfg(test)]
738mod tests {
739    use super::*;
740    use crate::catalog_view::{ColumnInfo, TableKind};
741    use inillucent_value::Affinity;
742
743    /// Builds a table with the named columns, for the generator tests.
744    fn table(name: &[u8], columns: &[&[u8]]) -> TableInfo {
745        TableInfo {
746            name: name.to_vec(),
747            folded: name.to_ascii_lowercase(),
748            database: 0,
749            root: 2,
750            columns: columns
751                .iter()
752                .map(|column| ColumnInfo {
753                    name: column.to_vec(),
754                    folded: column.to_ascii_lowercase(),
755                    declared_type: Vec::new(),
756                    affinity: Affinity::Blob,
757                    collation: b"binary".to_vec(),
758                    not_null: false,
759                    not_null_conflict: None,
760                    primary_key_conflict: None,
761                    default_sql: None,
762                    primary_key_position: None,
763                    hidden: false,
764                    generated: false,
765                    stored: false,
766                    generated_sql: None,
767                })
768                .collect(),
769            rowid_alias: None,
770            without_rowid: false,
771            strict: false,
772            autoincrement: false,
773            kind: TableKind::Table,
774            create_sql: Vec::new(),
775            indexes: Vec::new(),
776            view: None,
777            triggers: Vec::new(),
778            analysed_rows: None,
779            checks: Vec::new(),
780            foreign_keys: Vec::new(),
781            foreign_key_triggers: Vec::new(),
782            module: None,
783        }
784    }
785
786    /// Builds a key over one child column pointing at one parent column.
787    fn key(on_delete: ReferentialAction, on_update: ReferentialAction) -> ForeignKeyInfo {
788        ForeignKeyInfo {
789            id: 0,
790            columns: vec![1],
791            parent: b"p".to_vec(),
792            parent_folded: b"p".to_vec(),
793            parent_columns: vec![b"id".to_vec()],
794            on_delete,
795            on_update,
796            match_clause: Vec::new(),
797            deferrable: false,
798            initially_deferred: false,
799            cyclic: false,
800        }
801    }
802
803    /// Every generated trigger has to parse. A generator that produced text the
804    /// parser refuses would enforce nothing at all, silently.
805    #[test]
806    fn every_generated_trigger_parses() {
807        let child = table(b"c", &[b"id", b"pid"]);
808        let parent = table(b"p", &[b"id"]);
809        let limits = Limits::default();
810        let actions = [
811            ReferentialAction::NoAction,
812            ReferentialAction::Restrict,
813            ReferentialAction::Cascade,
814            ReferentialAction::SetNull,
815            ReferentialAction::SetDefault,
816        ];
817        let events = [
818            ForeignKeyEvent::ChildInsert,
819            ForeignKeyEvent::ChildUpdate,
820            ForeignKeyEvent::ParentDelete,
821            ForeignKeyEvent::ParentUpdate,
822        ];
823        for action in actions {
824            let key = key(action, action);
825            for event in events {
826                let built = trigger_for(&child, &parent, &key, event, b"main", false, &limits);
827                assert!(
828                    built.is_some(),
829                    "{action:?} on {event:?} produced no trigger"
830                );
831            }
832        }
833    }
834
835    /// The child's check fires before the write, tests every key column for
836    /// NULL, and looks the parent up by the columns the clause named.
837    #[test]
838    fn the_child_check_reads_as_it_should() {
839        let child = table(b"c", &[b"id", b"pid"]);
840        let parent = table(b"p", &[b"id"]);
841        let key = key(ReferentialAction::NoAction, ReferentialAction::NoAction);
842        let sql = child_check(
843            &child,
844            &parent,
845            &key,
846            ForeignKeyEvent::ChildInsert,
847            b"main",
848            &[b"pid".to_vec()],
849            &[b"id".to_vec()],
850        );
851        assert!(sql.contains("BEFORE INSERT ON \"c\""), "{sql}");
852        assert!(sql.contains("NEW.\"pid\" IS NOT NULL"), "{sql}");
853        assert!(sql.contains("NOT EXISTS"), "{sql}");
854        assert!(sql.contains("FOREIGN KEY constraint failed"), "{sql}");
855    }
856
857    /// RESTRICT fires before the parent write and NO ACTION after it, which is
858    /// the one place the two differ.
859    #[test]
860    fn restrict_fires_before_and_no_action_after() {
861        let child = table(b"c", &[b"id", b"pid"]);
862        let parent = table(b"p", &[b"id"]);
863        let limits = Limits::default();
864        for (action, expected) in [
865            (ReferentialAction::Restrict, TriggerTime::Before),
866            (ReferentialAction::NoAction, TriggerTime::After),
867        ] {
868            let key = key(action, action);
869            let built = trigger_for(
870                &child,
871                &parent,
872                &key,
873                ForeignKeyEvent::ParentDelete,
874                b"main",
875                false,
876                &limits,
877            )
878            .expect("the trigger is generated");
879            assert_eq!(built.time, expected, "{action:?}");
880        }
881    }
882
883    /// A deferred constraint generates no check on the child and no NO ACTION
884    /// on the parent - both wait for the commit - but RESTRICT and the cascades
885    /// still fire where they are.
886    #[test]
887    fn a_deferred_key_defers_only_its_checks() {
888        let child = table(b"c", &[b"id", b"pid"]);
889        let parent = table(b"p", &[b"id"]);
890        let limits = Limits::default();
891        let deferred = key(ReferentialAction::NoAction, ReferentialAction::NoAction);
892        assert!(trigger_for(
893            &child,
894            &parent,
895            &deferred,
896            ForeignKeyEvent::ChildInsert,
897            b"main",
898            true,
899            &limits
900        )
901        .is_none());
902        assert!(trigger_for(
903            &child,
904            &parent,
905            &deferred,
906            ForeignKeyEvent::ParentDelete,
907            b"main",
908            true,
909            &limits
910        )
911        .is_none());
912        let restrict = key(ReferentialAction::Restrict, ReferentialAction::Restrict);
913        assert!(trigger_for(
914            &child,
915            &parent,
916            &restrict,
917            ForeignKeyEvent::ParentDelete,
918            b"main",
919            true,
920            &limits
921        )
922        .is_some());
923        let cascade = key(ReferentialAction::Cascade, ReferentialAction::Cascade);
924        assert!(trigger_for(
925            &child,
926            &parent,
927            &cascade,
928            ForeignKeyEvent::ParentDelete,
929            b"main",
930            true,
931            &limits
932        )
933        .is_some());
934    }
935
936    /// A parent update fires only when the key actually changed, and cascades
937    /// the new key onto the rows that carried the old one.
938    #[test]
939    fn a_parent_update_guards_on_the_key_changing() {
940        let child = table(b"c", &[b"id", b"pid"]);
941        let parent = table(b"p", &[b"id"]);
942        let key = key(ReferentialAction::Cascade, ReferentialAction::Cascade);
943        let sql = parent_action(
944            &child,
945            &parent,
946            &key,
947            ForeignKeyEvent::ParentUpdate,
948            b"main",
949            &[b"pid".to_vec()],
950            &[b"id".to_vec()],
951            false,
952        )
953        .expect("the trigger is generated");
954        assert!(sql.contains("AFTER UPDATE OF \"id\""), "{sql}");
955        assert!(sql.contains("WHEN OLD.\"id\" IS NOT NEW.\"id\""), "{sql}");
956        assert!(sql.contains("SET \"pid\" = NEW.\"id\""), "{sql}");
957        assert!(sql.contains("WHERE \"pid\" = OLD.\"id\""), "{sql}");
958    }
959
960    /// An identifier with a quote in it survives the round trip, because the
961    /// generated text is parsed again rather than merely printed.
962    #[test]
963    fn an_awkward_identifier_is_quoted() {
964        assert_eq!(quote(b"we\"ird"), "\"we\"\"ird\"");
965        let child = table(b"we\"ird", &[b"id", b"pid"]);
966        let parent = table(b"p", &[b"id"]);
967        let key = key(ReferentialAction::Cascade, ReferentialAction::Cascade);
968        let limits = Limits::default();
969        assert!(trigger_for(
970            &child,
971            &parent,
972            &key,
973            ForeignKeyEvent::ChildInsert,
974            b"main",
975            false,
976            &limits
977        )
978        .is_some());
979    }
980
981    /// A composite key compares every column, in the order the clause wrote.
982    #[test]
983    fn a_composite_key_compares_every_column() {
984        let matching = children_of(
985            &[b"a".to_vec(), b"b".to_vec()],
986            &[b"x".to_vec(), b"y".to_vec()],
987            "OLD",
988        );
989        assert_eq!(matching, "\"a\" = OLD.\"x\" AND \"b\" = OLD.\"y\"");
990    }
991}