Create ForeignKey, ResolvedFkRef types and FK resolution

core/lib.rs

@@ -63,17 +63,16 @@ pub use io::{
 };
 use parking_lot::RwLock;
 use schema::Schema;
-use std::cell::Cell;
 use std::{
     borrow::Cow,
-    cell::RefCell,
+    cell::{Cell, RefCell},
     collections::HashMap,
     fmt::{self, Display},
     num::NonZero,
     ops::Deref,
     rc::Rc,
     sync::{
-        atomic::{AtomicBool, AtomicI32, AtomicI64, AtomicU16, AtomicUsize, Ordering},
+        atomic::{AtomicBool, AtomicI32, AtomicI64, AtomicIsize, AtomicU16, AtomicUsize, Ordering},
         Arc, LazyLock, Mutex, Weak,
     },
     time::Duration,
@@ -583,6 +582,7 @@ impl Database {
             data_sync_retry: AtomicBool::new(false),
             busy_timeout: RwLock::new(Duration::new(0, 0)),
             is_mvcc_bootstrap_connection: AtomicBool::new(is_mvcc_bootstrap_connection),
+            fk_pragma: AtomicBool::new(false),
         });
         self.n_connections
             .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
@@ -1100,6 +1100,7 @@ pub struct Connection {
     busy_timeout: RwLock<std::time::Duration>,
     /// Whether this is an internal connection used for MVCC bootstrap
     is_mvcc_bootstrap_connection: AtomicBool,
+    fk_pragma: AtomicBool,
 }
 
 impl Drop for Connection {
@@ -1532,6 +1533,21 @@ impl Connection {
         Ok(db)
     }
 
+    pub fn set_foreign_keys_enabled(&self, enable: bool) {
+        self.fk_pragma.store(enable, Ordering::Release);
+    }
+    pub fn foreign_keys_enabled(&self) -> bool {
+        self.fk_pragma.load(Ordering::Acquire)
+    }
+
+    pub(crate) fn clear_deferred_foreign_key_violations(&self) -> isize {
+        self.fk_deferred_violations.swap(0, Ordering::Release)
+    }
+
+    pub(crate) fn get_deferred_foreign_key_violations(&self) -> isize {
+        self.fk_deferred_violations.load(Ordering::Acquire)
+    }
+
     pub fn maybe_update_schema(&self) {
         let current_schema_version = self.schema.read().schema_version;
         let schema = self.db.schema.lock().unwrap();

core/schema.rs

@@ -89,7 +89,9 @@ use std::ops::Deref;
 use std::sync::Arc;
 use std::sync::Mutex;
 use tracing::trace;
-use turso_parser::ast::{self, ColumnDefinition, Expr, Literal, SortOrder, TableOptions};
+use turso_parser::ast::{
+    self, ColumnDefinition, Expr, InitDeferredPred, Literal, RefAct, SortOrder, TableOptions,
+};
 use turso_parser::{
     ast::{Cmd, CreateTableBody, ResultColumn, Stmt},
     parser::Parser,
@@ -298,9 +300,18 @@ impl Schema {
         self.views.get(&name).cloned()
     }
 
-    pub fn add_btree_table(&mut self, table: Arc<BTreeTable>) {
+    pub fn add_btree_table(&mut self, mut table: Arc<BTreeTable>) -> Result<()> {
         let name = normalize_ident(&table.name);
+        let mut resolved_fks: Vec<Arc<ForeignKey>> = Vec::with_capacity(table.foreign_keys.len());
+        // when we built the BTreeTable from SQL, we didn't have access to the Schema to validate
+        // any FK relationships, so we do that now
+        self.validate_and_normalize_btree_foreign_keys(&table, &mut resolved_fks)?;
+
+        // there should only be 1 reference to the table so Arc::make_mut shouldnt copy
+        let t = Arc::make_mut(&mut table);
+        t.foreign_keys = resolved_fks;
         self.tables.insert(name, Table::BTree(table).into());
+        Ok(())
     }
 
     pub fn add_virtual_table(&mut self, table: Arc<VirtualTable>) {
@@ -393,6 +404,31 @@ impl Schema {
         self.indexes_enabled
     }
 
+    pub fn get_foreign_keys_for_table(&self, table_name: &str) -> Vec<Arc<ForeignKey>> {
+        self.get_table(table_name)
+            .and_then(|t| t.btree())
+            .map(|t| t.foreign_keys.clone())
+            .unwrap_or_default()
+    }
+
+    /// Get foreign keys where this table is the parent (referenced by other tables)
+    pub fn get_referencing_foreign_keys(
+        &self,
+        parent_table: &str,
+    ) -> Vec<(String, Arc<ForeignKey>)> {
+        let mut refs = Vec::new();
+        for table in self.tables.values() {
+            if let Table::BTree(btree) = table.deref() {
+                for fk in &btree.foreign_keys {
+                    if fk.parent_table == parent_table {
+                        refs.push((btree.name.as_str().to_string(), fk.clone()));
+                    }
+                }
+            }
+        }
+        refs
+    }
+
     /// Update [Schema] by scanning the first root page (sqlite_schema)
     pub fn make_from_btree(
         &mut self,
@@ -646,6 +682,7 @@ impl Schema {
                 has_rowid: true,
                 is_strict: false,
                 has_autoincrement: false,
+                foreign_keys: vec![],
 
                 unique_sets: vec![],
             })));
@@ -732,7 +769,10 @@ impl Schema {
                         }
                     }
 
-                    self.add_btree_table(Arc::new(table));
+                    if let Some(mv_store) = mv_store {
+                        mv_store.mark_table_as_loaded(root_page);
+                    }
+                    self.add_btree_table(Arc::new(table))?;
                 }
             }
             "index" => {
@@ -842,6 +882,264 @@ impl Schema {
 
         Ok(())
     }
+
+    fn validate_and_normalize_btree_foreign_keys(
+        &self,
+        table: &Arc<BTreeTable>,
+        resolved_fks: &mut Vec<Arc<ForeignKey>>,
+    ) -> Result<()> {
+        for key in &table.foreign_keys {
+            let Some(parent) = self.get_btree_table(&key.parent_table) else {
+                return Err(LimboError::ParseError(format!(
+                    "Foreign key references missing table {}",
+                    key.parent_table
+                )));
+            };
+
+            let child_cols: Vec<String> = key
+                .child_columns
+                .iter()
+                .map(|c| normalize_ident(c))
+                .collect();
+            for c in &child_cols {
+                if table.get_column(c).is_none() && !c.eq_ignore_ascii_case("rowid") {
+                    return Err(LimboError::ParseError(format!(
+                        "Foreign key child column not found: {}.{}",
+                        table.name, c
+                    )));
+                }
+            }
+
+            // Resolve parent cols:
+            // if explicitly listed, we normalize them
+            // else, we default to parent's PRIMARY KEY columns.
+            // if parent has no declared PK, SQLite defaults to single "rowid"
+            let parent_cols: Vec<String> = if key.parent_columns.is_empty() {
+                if !parent.primary_key_columns.is_empty() {
+                    parent
+                        .primary_key_columns
+                        .iter()
+                        .map(|(n, _)| normalize_ident(n))
+                        .collect()
+                } else {
+                    vec!["rowid".to_string()]
+                }
+            } else {
+                key.parent_columns
+                    .iter()
+                    .map(|c| normalize_ident(c))
+                    .collect()
+            };
+
+            if parent_cols.len() != child_cols.len() {
+                return Err(LimboError::ParseError(format!(
+                    "Foreign key column count mismatch: child {child_cols:?} vs parent {parent_cols:?}",
+                )));
+            }
+
+            // Ensure each parent col exists
+            for col in &parent_cols {
+                if !col.eq_ignore_ascii_case("rowid") && parent.get_column(col).is_none() {
+                    return Err(LimboError::ParseError(format!(
+                        "Foreign key references missing column {}.{col}",
+                        key.parent_table
+                    )));
+                }
+            }
+
+            // Parent side must be UNIQUE/PK, rowid counts as unique
+            let parent_is_pk = !parent.primary_key_columns.is_empty()
+                && parent_cols.len() == parent.primary_key_columns.len()
+                && parent_cols
+                    .iter()
+                    .zip(&parent.primary_key_columns)
+                    .all(|(a, (b, _))| a.eq_ignore_ascii_case(b));
+
+            let parent_is_rowid =
+                parent_cols.len() == 1 && parent_cols[0].eq_ignore_ascii_case("rowid");
+
+            let parent_is_unique = parent_is_pk
+                || parent_is_rowid
+                || self.get_indices(&parent.name).any(|idx| {
+                    idx.unique
+                        && idx.columns.len() == parent_cols.len()
+                        && idx
+                            .columns
+                            .iter()
+                            .zip(&parent_cols)
+                            .all(|(ic, pc)| ic.name.eq_ignore_ascii_case(pc))
+                });
+
+            if !parent_is_unique {
+                return Err(LimboError::ParseError(format!(
+                    "Foreign key references {}({:?}) which is not UNIQUE or PRIMARY KEY",
+                    key.parent_table, parent_cols
+                )));
+            }
+
+            let resolved = ForeignKey {
+                parent_table: normalize_ident(&key.parent_table),
+                parent_columns: parent_cols,
+                child_columns: child_cols,
+                on_delete: key.on_delete,
+                on_update: key.on_update,
+                on_insert: key.on_insert,
+                deferred: key.deferred,
+            };
+            resolved_fks.push(Arc::new(resolved));
+        }
+        Ok(())
+    }
+
+    pub fn incoming_fks_to(&self, table_name: &str) -> Vec<IncomingFkRef> {
+        let target = normalize_ident(table_name);
+        let mut out = vec![];
+
+        // Resolve the parent table once
+        let parent_tbl = self
+            .get_btree_table(&target)
+            .expect("incoming_fks_to: parent table must exist");
+
+        // Precompute helper to find parent unique index, if it's not the rowid
+        let find_parent_unique = |cols: &Vec<String>| -> Option<Arc<Index>> {
+            // If matches PK exactly, we don't need a secondary index probe
+            let matches_pk = !parent_tbl.primary_key_columns.is_empty()
+                && parent_tbl.primary_key_columns.len() == cols.len()
+                && parent_tbl
+                    .primary_key_columns
+                    .iter()
+                    .zip(cols.iter())
+                    .all(|((n, _ord), c)| n.eq_ignore_ascii_case(c));
+
+            if matches_pk {
+                return None;
+            }
+
+            self.get_indices(&parent_tbl.name)
+                .find(|idx| {
+                    idx.unique
+                        && idx.columns.len() == cols.len()
+                        && idx
+                            .columns
+                            .iter()
+                            .zip(cols.iter())
+                            .all(|(ic, pc)| ic.name.eq_ignore_ascii_case(pc))
+                })
+                .cloned()
+        };
+
+        for t in self.tables.values() {
+            let Some(child) = t.btree() else {
+                continue;
+            };
+
+            for fk in &child.foreign_keys {
+                if normalize_ident(&fk.parent_table) != target {
+                    continue;
+                }
+
+                // Resolve + normalize columns
+                let child_cols: Vec<String> = fk
+                    .child_columns
+                    .iter()
+                    .map(|c| normalize_ident(c))
+                    .collect();
+
+                // If no explicit parent columns were given, they were validated in add_btree_table()
+                // to match the parent's PK. We resolve them the same way here.
+                let parent_cols: Vec<String> = if fk.parent_columns.is_empty() {
+                    parent_tbl
+                        .primary_key_columns
+                        .iter()
+                        .map(|(n, _)| normalize_ident(n))
+                        .collect()
+                } else {
+                    fk.parent_columns
+                        .iter()
+                        .map(|c| normalize_ident(c))
+                        .collect()
+                };
+
+                // Child positions
+                let child_pos: Vec<usize> = child_cols
+                    .iter()
+                    .map(|cname| {
+                        child.get_column(cname).map(|(i, _)| i).unwrap_or_else(|| {
+                            panic!(
+                                "incoming_fks_to: child col {}.{} missing",
+                                child.name, cname
+                            )
+                        })
+                    })
+                    .collect();
+
+                let parent_pos: Vec<usize> = parent_cols
+                    .iter()
+                    .map(|cname| {
+                        // Allow "rowid" sentinel; return 0 but it won't be used when parent_uses_rowid == true
+                        parent_tbl
+                            .get_column(cname)
+                            .map(|(i, _)| i)
+                            .or_else(|| {
+                                if cname.eq_ignore_ascii_case("rowid") {
+                                    Some(0)
+                                } else {
+                                    None
+                                }
+                            })
+                            .unwrap_or_else(|| {
+                                panic!(
+                                    "incoming_fks_to: parent col {}.{cname} missing",
+                                    parent_tbl.name
+                                )
+                            })
+                    })
+                    .collect();
+
+                // Detect parent rowid usage (single-column and rowid/alias)
+                let parent_uses_rowid = parent_cols.len() == 1 && {
+                    let c = parent_cols[0].as_str();
+                    c.eq_ignore_ascii_case("rowid")
+                        || parent_tbl.columns.iter().any(|col| {
+                            col.is_rowid_alias
+                                && col
+                                    .name
+                                    .as_deref()
+                                    .is_some_and(|n| n.eq_ignore_ascii_case(c))
+                        })
+                };
+
+                let parent_unique_index = if parent_uses_rowid {
+                    None
+                } else {
+                    find_parent_unique(&parent_cols)
+                };
+
+                out.push(IncomingFkRef {
+                    child_table: Arc::clone(&child),
+                    fk: Arc::clone(fk),
+                    parent_cols,
+                    child_cols,
+                    child_pos,
+                    parent_pos,
+                    parent_uses_rowid,
+                    parent_unique_index,
+                });
+            }
+        }
+        out
+    }
+
+    pub fn any_incoming_fk_to(&self, table_name: &str) -> bool {
+        self.tables.values().any(|t| {
+            let Some(bt) = t.btree() else {
+                return false;
+            };
+            bt.foreign_keys
+                .iter()
+                .any(|fk| fk.parent_table == table_name)
+        })
+    }
 }
 
 impl Clone for Schema {
@@ -1016,6 +1314,7 @@ pub struct BTreeTable {
     pub is_strict: bool,
     pub has_autoincrement: bool,
     pub unique_sets: Vec<UniqueSet>,
+    pub foreign_keys: Vec<Arc<ForeignKey>>,
 }
 
 impl BTreeTable {
@@ -1146,6 +1445,7 @@ pub fn create_table(tbl_name: &str, body: &CreateTableBody, root_page: i64) -> R
     let mut has_rowid = true;
     let mut has_autoincrement = false;
     let mut primary_key_columns = vec![];
+    let mut foreign_keys = vec![];
     let mut cols = vec![];
     let is_strict: bool;
     let mut unique_sets: Vec<UniqueSet> = vec![];
@@ -1219,6 +1519,85 @@ pub fn create_table(tbl_name: &str, body: &CreateTableBody, root_page: i64) -> R
                         is_primary_key: false,
                     };
                     unique_sets.push(unique_set);
+                } else if let ast::TableConstraint::ForeignKey {
+                    columns,
+                    clause,
+                    defer_clause,
+                } = &c.constraint
+                {
+                    let child_columns: Vec<String> = columns
+                        .iter()
+                        .map(|ic| normalize_ident(ic.col_name.as_str()))
+                        .collect();
+
+                    // derive parent columns: explicit or default to parent PK
+                    let parent_table = normalize_ident(clause.tbl_name.as_str());
+                    let parent_columns: Vec<String> = clause
+                        .columns
+                        .iter()
+                        .map(|ic| normalize_ident(ic.col_name.as_str()))
+                        .collect();
+
+                    // arity check
+                    if child_columns.len() != parent_columns.len() {
+                        crate::bail_parse_error!(
+                            "foreign key on \"{}\" has {} child column(s) but {} parent column(s)",
+                            tbl_name,
+                            child_columns.len(),
+                            parent_columns.len()
+                        );
+                    }
+                    // deferrable semantics
+                    let deferred = match defer_clause {
+                        Some(d) => {
+                            d.deferrable
+                                && matches!(
+                                    d.init_deferred,
+                                    Some(InitDeferredPred::InitiallyDeferred)
+                                )
+                        }
+                        None => false, // NOT DEFERRABLE INITIALLY IMMEDIATE by default
+                    };
+                    let fk = ForeignKey {
+                        parent_table,
+                        parent_columns,
+                        child_columns,
+                        on_delete: clause
+                            .args
+                            .iter()
+                            .find_map(|a| {
+                                if let ast::RefArg::OnDelete(x) = a {
+                                    Some(*x)
+                                } else {
+                                    None
+                                }
+                            })
+                            .unwrap_or(RefAct::NoAction),
+                        on_insert: clause
+                            .args
+                            .iter()
+                            .find_map(|a| {
+                                if let ast::RefArg::OnInsert(x) = a {
+                                    Some(*x)
+                                } else {
+                                    None
+                                }
+                            })
+                            .unwrap_or(RefAct::NoAction),
+                        on_update: clause
+                            .args
+                            .iter()
+                            .find_map(|a| {
+                                if let ast::RefArg::OnUpdate(x) = a {
+                                    Some(*x)
+                                } else {
+                                    None
+                                }
+                            })
+                            .unwrap_or(RefAct::NoAction),
+                        deferred,
+                    };
+                    foreign_keys.push(Arc::new(fk));
                 }
             }
             for ast::ColumnDefinition {
@@ -1259,7 +1638,7 @@ pub fn create_table(tbl_name: &str, body: &CreateTableBody, root_page: i64) -> R
                 let mut unique = false;
                 let mut collation = None;
                 for c_def in constraints {
-                    match c_def.constraint {
+                    match &c_def.constraint {
                         ast::ColumnConstraint::PrimaryKey {
                             order: o,
                             auto_increment,
@@ -1272,11 +1651,11 @@ pub fn create_table(tbl_name: &str, body: &CreateTableBody, root_page: i64) -> R
                                 );
                             }
                             primary_key = true;
-                            if auto_increment {
+                            if *auto_increment {
                                 has_autoincrement = true;
                             }
                             if let Some(o) = o {
-                                order = o;
+                                order = *o;
                             }
                             unique_sets.push(UniqueSet {
                                 columns: vec![(name.clone(), order)],
@@ -1305,6 +1684,55 @@ pub fn create_table(tbl_name: &str, body: &CreateTableBody, root_page: i64) -> R
                         ast::ColumnConstraint::Collate { ref collation_name } => {
                             collation = Some(CollationSeq::new(collation_name.as_str())?);
                         }
+                        ast::ColumnConstraint::ForeignKey {
+                            clause,
+                            defer_clause,
+                        } => {
+                            let fk = ForeignKey {
+                                parent_table: clause.tbl_name.to_string(),
+                                parent_columns: clause
+                                    .columns
+                                    .iter()
+                                    .map(|c| c.col_name.as_str().to_string())
+                                    .collect(),
+                                on_delete: clause
+                                    .args
+                                    .iter()
+                                    .find_map(|arg| {
+                                        if let ast::RefArg::OnDelete(act) = arg {
+                                            Some(*act)
+                                        } else {
+                                            None
+                                        }
+                                    })
+                                    .unwrap_or(RefAct::NoAction),
+                                on_insert: clause
+                                    .args
+                                    .iter()
+                                    .find_map(|arg| {
+                                        if let ast::RefArg::OnInsert(act) = arg {
+                                            Some(*act)
+                                        } else {
+                                            None
+                                        }
+                                    })
+                                    .unwrap_or(RefAct::NoAction),
+                                on_update: clause
+                                    .args
+                                    .iter()
+                                    .find_map(|arg| {
+                                        if let ast::RefArg::OnUpdate(act) = arg {
+                                            Some(*act)
+                                        } else {
+                                            None
+                                        }
+                                    })
+                                    .unwrap_or(RefAct::NoAction),
+                                child_columns: vec![name.clone()],
+                                deferred: defer_clause.is_some(),
+                            };
+                            foreign_keys.push(Arc::new(fk));
+                        }
                         _ => {}
                     }
                 }
@@ -1384,6 +1812,7 @@ pub fn create_table(tbl_name: &str, body: &CreateTableBody, root_page: i64) -> R
         has_autoincrement,
         columns: cols,
         is_strict,
+        foreign_keys,
         unique_sets: {
             // If there are any unique sets that have identical column names in the same order (even if they are PRIMARY KEY and UNIQUE and have different sort orders), remove the duplicates.
             // Examples:
@@ -1441,6 +1870,93 @@ pub fn _build_pseudo_table(columns: &[ResultColumn]) -> PseudoCursorType {
     table
 }
 
+#[derive(Debug, Clone)]
+pub struct ForeignKey {
+    /// Columns in this table
+    pub child_columns: Vec<String>,
+    /// Referenced table
+    pub parent_table: String,
+    /// Referenced columns
+    pub parent_columns: Vec<String>,
+    pub on_delete: RefAct,
+    pub on_update: RefAct,
+    pub on_insert: RefAct,
+    /// DEFERRABLE INITIALLY DEFERRED
+    pub deferred: bool,
+}
+
+/// A single foreign key where `parent_table == target`.
+#[derive(Clone, Debug)]
+pub struct IncomingFkRef {
+    /// Child table that owns the FK.
+    pub child_table: Arc<BTreeTable>,
+    /// The FK as declared on the child table.
+    pub fk: Arc<ForeignKey>,
+
+    /// Resolved, normalized column names.
+    pub parent_cols: Vec<String>,
+    pub child_cols: Vec<String>,
+
+    /// Column positions in the child/parent tables (pos_in_table)
+    pub child_pos: Vec<usize>,
+    pub parent_pos: Vec<usize>,
+
+    /// If the parent key is rowid or a rowid-alias (single-column only)
+    pub parent_uses_rowid: bool,
+    /// For non-rowid parents: the UNIQUE index that enforces the parent key.
+    /// (None when `parent_uses_rowid == true`.)
+    pub parent_unique_index: Option<Arc<Index>>,
+}
+
+impl IncomingFkRef {
+    /// Returns if any referenced parent column can change when these column positions are updated.
+    pub fn parent_key_may_change(
+        &self,
+        updated_parent_positions: &HashSet<usize>,
+        parent_tbl: &BTreeTable,
+    ) -> bool {
+        if self.parent_uses_rowid {
+            // parent rowid changes if the parent's rowid or alias is updated
+            if let Some((idx, _)) = parent_tbl
+                .columns
+                .iter()
+                .enumerate()
+                .find(|(_, c)| c.is_rowid_alias)
+            {
+                return updated_parent_positions.contains(&idx);
+            }
+            // Without a rowid alias, a direct rowid update is represented separately with ROWID_SENTINEL
+            return true;
+        }
+        self.parent_pos
+            .iter()
+            .any(|p| updated_parent_positions.contains(p))
+    }
+
+    /// Returns if any child column of this FK is in `updated_child_positions`
+    pub fn child_key_changed(
+        &self,
+        updated_child_positions: &HashSet<usize>,
+        child_tbl: &BTreeTable,
+    ) -> bool {
+        if self
+            .child_pos
+            .iter()
+            .any(|p| updated_child_positions.contains(p))
+        {
+            return true;
+        }
+        // special case: if FK uses a rowid alias on child, and rowid changed
+        if self.child_cols.len() == 1 {
+            let (i, col) = child_tbl.get_column(&self.child_cols[0]).unwrap();
+            if col.is_rowid_alias && updated_child_positions.contains(&i) {
+                return true;
+            }
+        }
+        false
+    }
+}
+
 #[derive(Debug, Clone)]
 pub struct Column {
     pub name: Option<String>,
@@ -1782,6 +2298,7 @@ pub fn sqlite_schema_table() -> BTreeTable {
                 hidden: false,
             },
         ],
+        foreign_keys: vec![],
         unique_sets: vec![],
     }
 }

core/translate/insert.rs

@@ -83,6 +83,11 @@ pub fn translate_insert(
         );
     }
     let table_name = &tbl_name.name;
+    let has_child_fks = connection.foreign_keys_enabled()
+        && !resolver
+            .schema
+            .get_foreign_keys_for_table(table_name.as_str())
+            .is_empty();
 
     // Check if this is a system table that should be protected from direct writes
     if crate::schema::is_system_table(table_name.as_str()) {
@@ -222,6 +227,8 @@ pub fn translate_insert(
     let halt_label = program.allocate_label();
     let loop_start_label = program.allocate_label();
     let row_done_label = program.allocate_label();
+    let stmt_epilogue = program.allocate_label();
+    let mut select_exhausted_label: Option<BranchOffset> = None;
 
     let cdc_table = prepare_cdc_if_necessary(&mut program, resolver.schema, table.get_name())?;
 
@@ -234,6 +241,14 @@ pub fn translate_insert(
         connection,
     )?;
 
+    if has_child_fks {
+        program.emit_insn(Insn::FkCounter {
+            increment_value: 1,
+            check_abort: false,
+            is_scope: true,
+        });
+    }
+
     let mut yield_reg_opt = None;
     let mut temp_table_ctx = None;
     let (num_values, cursor_id) = match body {
@@ -254,11 +269,11 @@ pub fn translate_insert(
                     jump_on_definition: jump_on_definition_label,
                     start_offset: start_offset_label,
                 });
-
                 program.preassign_label_to_next_insn(start_offset_label);
 
                 let query_destination = QueryDestination::CoroutineYield {
                     yield_reg,
+                    // keep implementation_start as halt_label (producer internals)
                     coroutine_implementation_start: halt_label,
                 };
                 program.incr_nesting();
@@ -298,18 +313,14 @@ pub fn translate_insert(
                     });
 
                     // Main loop
-                    // FIXME: rollback is not implemented. E.g. if you insert 2 rows and one fails to unique constraint violation,
-                    // the other row will still be inserted.
                     program.preassign_label_to_next_insn(loop_start_label);
-
                     let yield_label = program.allocate_label();
-
                     program.emit_insn(Insn::Yield {
                         yield_reg,
-                        end_offset: yield_label,
+                        end_offset: yield_label, // stays local, we’ll route at loop end
                     });
-                    let record_reg = program.alloc_register();
 
+                    let record_reg = program.alloc_register();
                     let affinity_str = if columns.is_empty() {
                         btree_table
                             .columns
@@ -352,7 +363,6 @@ pub fn translate_insert(
                         rowid_reg,
                         prev_largest_reg: 0,
                     });
-
                     program.emit_insn(Insn::Insert {
                         cursor: temp_cursor_id,
                         key_reg: rowid_reg,
@@ -361,12 +371,10 @@ pub fn translate_insert(
                         flag: InsertFlags::new().require_seek(),
                         table_name: "".to_string(),
                     });
-
                     // loop back
                     program.emit_insn(Insn::Goto {
                         target_pc: loop_start_label,
                     });
-
                     program.preassign_label_to_next_insn(yield_label);
 
                     program.emit_insn(Insn::OpenWrite {
@@ -381,13 +389,14 @@ pub fn translate_insert(
                         db: 0,
                     });
 
-                    // Main loop
-                    // FIXME: rollback is not implemented. E.g. if you insert 2 rows and one fails to unique constraint violation,
-                    // the other row will still be inserted.
                     program.preassign_label_to_next_insn(loop_start_label);
+
+                    // on EOF, jump to select_exhausted to check FK constraints
+                    let select_exhausted = program.allocate_label();
+                    select_exhausted_label = Some(select_exhausted);
                     program.emit_insn(Insn::Yield {
                         yield_reg,
-                        end_offset: halt_label,
+                        end_offset: select_exhausted,
                     });
                 }
 
@@ -1033,6 +1042,9 @@ pub fn translate_insert(
             }
         }
     }
+    if has_child_fks {
+        emit_fk_checks_for_insert(&mut program, resolver, &insertion, table_name.as_str())?;
+    }
 
     program.emit_insn(Insn::Insert {
         cursor: cursor_id,
@@ -1154,15 +1166,38 @@ pub fn translate_insert(
             program.emit_insn(Insn::Close {
                 cursor_id: temp_table_ctx.cursor_id,
             });
+            program.emit_insn(Insn::Goto {
+                target_pc: stmt_epilogue,
+            });
         } else {
             // For multiple rows which not require a temp table, loop back
             program.resolve_label(row_done_label, program.offset());
             program.emit_insn(Insn::Goto {
                 target_pc: loop_start_label,
             });
+            if let Some(sel_eof) = select_exhausted_label {
+                program.preassign_label_to_next_insn(sel_eof);
+                program.emit_insn(Insn::Goto {
+                    target_pc: stmt_epilogue,
+                });
+            }
         }
     } else {
         program.resolve_label(row_done_label, program.offset());
+        // single-row falls through to epilogue
+        program.emit_insn(Insn::Goto {
+            target_pc: stmt_epilogue,
+        });
+    }
+
+    program.preassign_label_to_next_insn(stmt_epilogue);
+    if has_child_fks {
+        // close FK scope and surface deferred violations
+        program.emit_insn(Insn::FkCounter {
+            increment_value: -1,
+            check_abort: true,
+            is_scope: true,
+        });
     }
 
     program.resolve_label(halt_label, program.offset());
@@ -1857,3 +1892,196 @@ fn emit_update_sqlite_sequence(
 
     Ok(())
 }
+
+/// Emit child->parent foreign key checks for an INSERT, for the current row
+fn emit_fk_checks_for_insert(
+    program: &mut ProgramBuilder,
+    resolver: &Resolver,
+    insertion: &Insertion,
+    table_name: &str,
+) -> Result<()> {
+    let after_all = program.allocate_label();
+    program.emit_insn(Insn::FkIfZero {
+        target_pc: after_all,
+        if_zero: true,
+    });
+
+    // Iterate child FKs declared on this table
+    for fk in resolver.schema.get_foreign_keys_for_table(table_name) {
+        let fk_ok = program.allocate_label();
+
+        // If any child column is NULL, skip this FK
+        for child_col in &fk.child_columns {
+            let mapping = insertion
+                .get_col_mapping_by_name(child_col)
+                .ok_or_else(|| {
+                    crate::LimboError::InternalError(format!("FK column {child_col} not found"))
+                })?;
+            let src = if mapping.column.is_rowid_alias {
+                insertion.key_register()
+            } else {
+                mapping.register
+            };
+            program.emit_insn(Insn::IsNull {
+                reg: src,
+                target_pc: fk_ok,
+            });
+        }
+
+        // Parent lookup: rowid path or unique-index path
+        let parent_tbl = resolver.schema.get_table(&fk.parent_table).ok_or_else(|| {
+            crate::LimboError::InternalError(format!("Parent table {} not found", fk.parent_table))
+        })?;
+
+        let uses_rowid = {
+            // If single parent column equals rowid or aliases rowid
+            fk.parent_columns.len() == 1 && {
+                let parent_col = fk.parent_columns[0].as_str();
+                parent_col.eq_ignore_ascii_case("rowid")
+                    || parent_tbl.columns().iter().any(|c| {
+                        c.is_rowid_alias
+                            && c.name
+                                .as_ref()
+                                .is_some_and(|n| n.eq_ignore_ascii_case(parent_col))
+                    })
+            }
+        };
+
+        if uses_rowid {
+            // Simple rowid probe on parent table
+            let parent_bt = parent_tbl.btree().ok_or_else(|| {
+                crate::LimboError::InternalError("Parent table is not a BTree".into())
+            })?;
+            let pcur = program.alloc_cursor_id(CursorType::BTreeTable(parent_bt.clone()));
+            program.emit_insn(Insn::OpenRead {
+                cursor_id: pcur,
+                root_page: parent_bt.root_page,
+                db: 0,
+            });
+
+            // Child value register
+            let cm = insertion
+                .get_col_mapping_by_name(&fk.child_columns[0])
+                .ok_or_else(|| {
+                    crate::LimboError::InternalError("FK child column not found".into())
+                })?;
+            let val_reg = if cm.column.is_rowid_alias {
+                insertion.key_register()
+            } else {
+                cm.register
+            };
+
+            let violation = program.allocate_label();
+            // NotExists: jump to violation if missing in parent
+            program.emit_insn(Insn::NotExists {
+                cursor: pcur,
+                rowid_reg: val_reg,
+                target_pc: violation,
+            });
+            // OK
+            program.emit_insn(Insn::Close { cursor_id: pcur });
+            program.emit_insn(Insn::Goto { target_pc: fk_ok });
+
+            // Violation
+            program.preassign_label_to_next_insn(violation);
+            program.emit_insn(Insn::Close { cursor_id: pcur });
+
+            // Deferred vs immediate
+            if fk.deferred {
+                program.emit_insn(Insn::FkCounter {
+                    increment_value: 1,
+                    check_abort: false,
+                    is_scope: false,
+                });
+            } else {
+                program.emit_insn(Insn::Halt {
+                    err_code: crate::error::SQLITE_CONSTRAINT_FOREIGNKEY,
+                    description: "FOREIGN KEY constraint failed".to_string(),
+                });
+            }
+        } else {
+            // Multi-column (or non-rowid) parent, we have to match a UNIQUE index with
+            // the exact column set and order
+            let parent_idx = resolver
+                .schema
+                .get_indices(&fk.parent_table)
+                .find(|idx| {
+                    idx.unique
+                        && idx.columns.len() == fk.parent_columns.len()
+                        && idx
+                            .columns
+                            .iter()
+                            .zip(fk.parent_columns.iter())
+                            .all(|(ic, pc)| ic.name.eq_ignore_ascii_case(pc))
+                })
+                .ok_or_else(|| {
+                    crate::LimboError::InternalError(format!(
+                        "No UNIQUE index on parent {}({:?}) for FK",
+                        fk.parent_table, fk.parent_columns
+                    ))
+                })?;
+
+            let icur = program.alloc_cursor_id(CursorType::BTreeIndex(parent_idx.clone()));
+            program.emit_insn(Insn::OpenRead {
+                cursor_id: icur,
+                root_page: parent_idx.root_page,
+                db: 0,
+            });
+
+            // Build packed search key registers from the *child* values
+            let n = fk.child_columns.len();
+            let start = program.alloc_registers(n);
+            for (i, child_col) in fk.child_columns.iter().enumerate() {
+                let cm = insertion
+                    .get_col_mapping_by_name(child_col)
+                    .ok_or_else(|| {
+                        crate::LimboError::InternalError(format!("Column {child_col} not found"))
+                    })?;
+                let src = if cm.column.is_rowid_alias {
+                    insertion.key_register()
+                } else {
+                    cm.register
+                };
+                program.emit_insn(Insn::Copy {
+                    src_reg: src,
+                    dst_reg: start + i,
+                    extra_amount: 0,
+                });
+            }
+
+            let found = program.allocate_label();
+            program.emit_insn(Insn::Found {
+                cursor_id: icur,
+                target_pc: found,
+                record_reg: start,
+                num_regs: n,
+            });
+
+            // Violation path
+            program.emit_insn(Insn::Close { cursor_id: icur });
+            if fk.deferred {
+                program.emit_insn(Insn::FkCounter {
+                    increment_value: 1,
+                    check_abort: false,
+                    is_scope: false,
+                });
+            } else {
+                program.emit_insn(Insn::Halt {
+                    err_code: crate::error::SQLITE_CONSTRAINT_FOREIGNKEY,
+                    description: "FOREIGN KEY constraint failed".to_string(),
+                });
+            }
+            program.emit_insn(Insn::Goto { target_pc: fk_ok });
+
+            // Found OK
+            program.preassign_label_to_next_insn(found);
+            program.emit_insn(Insn::Close { cursor_id: icur });
+        }
+
+        // Done with this FK
+        program.preassign_label_to_next_insn(fk_ok);
+    }
+
+    program.resolve_label(after_all, program.offset());
+    Ok(())
+}

core/translate/planner.rs

@@ -478,6 +478,7 @@ fn parse_table(
             has_autoincrement: false,
 
             unique_sets: vec![],
+            foreign_keys: vec![],
         });
         drop(view_guard);
 

core/translate/pragma.rs

@@ -389,7 +389,14 @@ fn update_pragma(
         }
         PragmaName::ForeignKeys => {
             let enabled = match &value {
-                Expr::Literal(Literal::Keyword(name)) | Expr::Id(name) => {
+                Expr::Id(name) | Expr::Name(name) => {
+                    let name_str = name.as_str().as_bytes();
+                    match_ignore_ascii_case!(match name_str {
+                        b"ON" | b"TRUE" | b"YES" | b"1" => true,
+                        _ => false,
+                    })
+                }
+                Expr::Literal(Literal::Keyword(name) | Literal::String(name)) => {
                     let name_bytes = name.as_bytes();
                     match_ignore_ascii_case!(match name_bytes {
                         b"ON" | b"TRUE" | b"YES" | b"1" => true,
@@ -399,7 +406,7 @@ fn update_pragma(
                 Expr::Literal(Literal::Numeric(n)) => !matches!(n.as_str(), "0"),
                 _ => false,
             };
-            connection.set_foreign_keys(enabled);
+            connection.set_foreign_keys_enabled(enabled);
             Ok((program, TransactionMode::None))
         }
     }

core/translate/schema.rs

@@ -812,6 +812,7 @@ pub fn translate_drop_table(
             }],
             is_strict: false,
             unique_sets: vec![],
+            foreign_keys: vec![],
         });
         // cursor id 2
         let ephemeral_cursor_id = program.alloc_cursor_id(CursorType::BTreeTable(simple_table_rc));

core/translate/update.rs

@@ -353,6 +353,7 @@ pub fn prepare_update_plan(
             }],
             is_strict: false,
             unique_sets: vec![],
+            foreign_keys: vec![],
         });
 
         let temp_cursor_id = program.alloc_cursor_id(CursorType::BTreeTable(table.clone()));

core/translate/view.rs

@@ -80,6 +80,7 @@ pub fn translate_create_materialized_view(
         has_autoincrement: false,
 
         unique_sets: vec![],
+        foreign_keys: vec![],
     });
 
     // Allocate a cursor for writing to the view's btree during population

core/translate/window.rs

@@ -505,6 +505,7 @@ pub fn init_window<'a>(
         is_strict: false,
         unique_sets: vec![],
         has_autoincrement: false,
+        foreign_keys: vec![],
     });
     let cursor_buffer_read = program.alloc_cursor_id(CursorType::BTreeTable(buffer_table.clone()));
     let cursor_buffer_write = program.alloc_cursor_id(CursorType::BTreeTable(buffer_table.clone()));

core/vdbe/execute.rs

@@ -1,5 +1,5 @@
 #![allow(unused_variables)]
-use crate::error::SQLITE_CONSTRAINT_UNIQUE;
+use crate::error::{SQLITE_CONSTRAINT_FOREIGNKEY, SQLITE_CONSTRAINT_UNIQUE};
 use crate::function::AlterTableFunc;
 use crate::mvcc::database::CheckpointStateMachine;
 use crate::numeric::{NullableInteger, Numeric};
@@ -2156,6 +2156,9 @@ pub fn halt(
                 "UNIQUE constraint failed: {description} (19)"
             )));
         }
+        SQLITE_CONSTRAINT_FOREIGNKEY => {
+            return Err(LimboError::Constraint(format!("{description} (19)")));
+        }
         _ => {
             return Err(LimboError::Constraint(format!(
                 "undocumented halt error code {description}"
@@ -8287,18 +8290,35 @@ pub fn op_fk_counter(
         FkCounter {
             increment_value,
             check_abort,
+            is_scope,
         },
         insn
     );
-    state.fk_constraint_counter = state.fk_constraint_counter.saturating_add(*increment_value);
+    if *is_scope {
+        // Adjust FK scope depth
+        state.fk_scope_counter = state.fk_scope_counter.saturating_add(*increment_value);
 
-    // If check_abort is true and counter is negative, abort with constraint error
+        // raise if there were deferred violations in this statement.
-    // This shouldn't happen in well-formed bytecode but acts as a safety check
+        if *check_abort {
-    if *check_abort && state.fk_constraint_counter < 0 {
+            if state.fk_scope_counter < 0 {
                 return Err(LimboError::Constraint(
                     "FOREIGN KEY constraint failed".into(),
                 ));
             }
+            if state.fk_scope_counter == 0 && state.fk_deferred_violations > 0 {
+                // Clear violations for safety, a new statement will re-open scope.
+                state.fk_deferred_violations = 0;
+                return Err(LimboError::Constraint(
+                    "FOREIGN KEY constraint failed".into(),
+                ));
+            }
+        }
+    } else {
+        // Adjust deferred violations counter
+        state.fk_deferred_violations = state
+            .fk_deferred_violations
+            .saturating_add(*increment_value);
+    }
 
     state.pc += 1;
     Ok(InsnFunctionStepResult::Step)
@@ -8318,12 +8338,14 @@ pub fn op_fk_if_zero(
     // Foreign keys are disabled globally
     // p1 is true AND deferred constraint counter is zero
     // p1 is false AND deferred constraint counter is non-zero
+    let scope_zero = state.fk_scope_counter == 0;
+
     let should_jump = if !fk_enabled {
         true
     } else if *if_zero {
-        state.fk_constraint_counter == 0
+        scope_zero
     } else {
-        state.fk_constraint_counter != 0
+        !scope_zero
     };
 
     if should_jump {

core/vdbe/explain.rs

@@ -1804,11 +1804,11 @@ pub fn insn_to_row(
             0,
             String::new(),
         ),
-        Insn::FkCounter{check_abort, increment_value} => (
+        Insn::FkCounter{check_abort, increment_value, is_scope } => (
         "FkCounter",
             *check_abort as i32,
             *increment_value as i32,
-            0,
+            *is_scope as i32,
             Value::build_text(""),
             0,
             String::new(),

core/vdbe/insn.rs

@@ -1175,6 +1175,7 @@ pub enum Insn {
     FkCounter {
         check_abort: bool,
         increment_value: isize,
+        is_scope: bool,
     },
     // This opcode tests if a foreign key constraint-counter is currently zero. If so, jump to instruction P2. Otherwise, fall through to the next instruction.
     // If P1 is non-zero, then the jump is taken if the database constraint-counter is zero (the one that counts deferred constraint violations).

core/vdbe/mod.rs

@@ -313,7 +313,7 @@ pub struct ProgramState {
     /// This is used when statement in auto-commit mode reseted after previous uncomplete execution - in which case we may need to rollback transaction started on previous attempt
     /// Note, that MVCC transactions are always explicit - so they do not update auto_txn_cleanup marker
     pub(crate) auto_txn_cleanup: TxnCleanup,
-    fk_constraint_counter: isize,
+    fk_scope_counter: isize,
 }
 
 impl ProgramState {
@@ -360,7 +360,7 @@ impl ProgramState {
             op_checkpoint_state: OpCheckpointState::StartCheckpoint,
             view_delta_state: ViewDeltaCommitState::NotStarted,
             auto_txn_cleanup: TxnCleanup::None,
-            fk_constraint_counter: 0,
+            fk_scope_counter: 0,
         }
     }