diff --git a/core/schema.rs b/core/schema.rs index b79b37017..6619aaaa2 100644 --- a/core/schema.rs +++ b/core/schema.rs @@ -395,31 +395,6 @@ impl Schema { self.indexes_enabled } - pub fn get_foreign_keys_for_table(&self, table_name: &str) -> Vec> { - 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)> { - 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, @@ -871,11 +846,9 @@ impl Schema { Ok(()) } - pub fn incoming_fks_to(&self, table_name: &str) -> Vec { + pub fn incoming_fks_to(&self, table_name: &str) -> Vec { 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"); @@ -995,7 +968,7 @@ impl Schema { find_parent_unique(&parent_cols) }; - out.push(IncomingFkRef { + out.push(ResolvedFkRef { child_table: Arc::clone(&child), fk: Arc::clone(fk), parent_cols, @@ -1010,6 +983,117 @@ impl Schema { out } + pub fn outgoing_fks_of(&self, child_table: &str) -> Vec { + let child_name = normalize_ident(child_table); + let Some(child) = self.get_btree_table(&child_name) else { + return vec![]; + }; + + // Helper to find the UNIQUE/index on the parent that matches the resolved parent cols + let find_parent_unique = + |parent_tbl: &BTreeTable, cols: &Vec| -> Option> { + 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, _), 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() + }; + + let mut out = Vec::new(); + for fk in &child.foreign_keys { + let parent_name = normalize_ident(&fk.parent_table); + let Some(parent_tbl) = self.get_btree_table(&parent_name) else { + continue; + }; + + // Normalize columns (same rules you used in validation) + let child_cols: Vec = fk + .child_columns + .iter() + .map(|s| normalize_ident(s)) + .collect(); + let parent_cols: Vec = if fk.parent_columns.is_empty() { + if !parent_tbl.primary_key_columns.is_empty() { + parent_tbl + .primary_key_columns + .iter() + .map(|(n, _)| normalize_ident(n)) + .collect() + } else { + vec!["rowid".to_string()] + } + } else { + fk.parent_columns + .iter() + .map(|s| normalize_ident(s)) + .collect() + }; + + // Positions + let child_pos: Vec = child_cols + .iter() + .map(|c| child.get_column(c).expect("child col missing").0) + .collect(); + let parent_pos: Vec = parent_cols + .iter() + .map(|c| { + parent_tbl + .get_column(c) + .map(|(i, _)| i) + .or_else(|| c.eq_ignore_ascii_case("rowid").then_some(0)) + .expect("parent col missing") + }) + .collect(); + + // Parent uses rowid? + 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_tbl, &parent_cols) + }; + + out.push(ResolvedFkRef { + 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 { @@ -1020,6 +1104,37 @@ impl Schema { .any(|fk| fk.parent_table == table_name) }) } + + /// Returns if this table declares any outgoing FKs (is a child of some parent) + pub fn has_child_fks(&self, table_name: &str) -> bool { + self.get_table(table_name) + .and_then(|t| t.btree()) + .is_some_and(|t| !t.foreign_keys.is_empty()) + } + + /// Return the *declared* (unresolved) FKs for a table. Callers that need + /// positions/rowid/unique info should use `incoming_fks_to` instead. + pub fn get_fks_for_table(&self, table_name: &str) -> Vec> { + self.get_table(table_name) + .and_then(|t| t.btree()) + .map(|t| t.foreign_keys.clone()) + .unwrap_or_default() + } + + /// Return pairs of (child_table_name, FK) for FKs that reference `parent_table` + pub fn get_referencing_fks(&self, parent_table: &str) -> Vec<(String, Arc)> { + 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 + } } impl Clone for Schema { @@ -1752,11 +1867,11 @@ pub fn _build_pseudo_table(columns: &[ResultColumn]) -> PseudoCursorType { #[derive(Debug, Clone)] pub struct ForeignKey { - /// Columns in this table + /// Columns in this table (child side) pub child_columns: Vec, - /// Referenced table + /// Referenced (parent) table pub parent_table: String, - /// Referenced columns + /// Parent-side referenced columns pub parent_columns: Vec, pub on_delete: RefAct, pub on_update: RefAct, @@ -1765,9 +1880,9 @@ pub struct ForeignKey { pub deferred: bool, } -/// A single foreign key where `parent_table == target`. +/// A single resolved foreign key where `parent_table == target`. #[derive(Clone, Debug)] -pub struct IncomingFkRef { +pub struct ResolvedFkRef { /// Child table that owns the FK. pub child_table: Arc, /// The FK as declared on the child table. @@ -1788,7 +1903,7 @@ pub struct IncomingFkRef { pub parent_unique_index: Option>, } -impl IncomingFkRef { +impl ResolvedFkRef { /// Returns if any referenced parent column can change when these column positions are updated. pub fn parent_key_may_change( &self, diff --git a/core/translate/emitter.rs b/core/translate/emitter.rs index 5e60617e6..f569743be 100644 --- a/core/translate/emitter.rs +++ b/core/translate/emitter.rs @@ -1,6 +1,7 @@ // This module contains code for emitting bytecode instructions for SQL query execution. // It handles translating high-level SQL operations into low-level bytecode that can be executed by the virtual machine. +use std::collections::HashSet; use std::num::NonZeroUsize; use std::sync::Arc; @@ -23,7 +24,7 @@ use super::select::emit_simple_count; use super::subquery::emit_subqueries; use crate::error::SQLITE_CONSTRAINT_PRIMARYKEY; use crate::function::Func; -use crate::schema::{BTreeTable, Column, Schema, Table, ROWID_SENTINEL}; +use crate::schema::{BTreeTable, Column, ResolvedFkRef, Schema, Table, ROWID_SENTINEL}; use crate::translate::compound_select::emit_program_for_compound_select; use crate::translate::expr::{ emit_returning_results, translate_expr_no_constant_opt, walk_expr_mut, NoConstantOptReason, @@ -431,6 +432,25 @@ fn emit_program_for_delete( }); } + let has_parent_fks = connection.foreign_keys_enabled() && { + let table_name = plan + .table_references + .joined_tables() + .first() + .unwrap() + .table + .get_name(); + resolver.schema.any_incoming_fk_to(table_name) + }; + // Open FK scope for the whole statement + if has_parent_fks { + program.emit_insn(Insn::FkCounter { + increment_value: 1, + check_abort: false, + is_scope: true, + }); + } + // Initialize cursors and other resources needed for query execution init_loop( program, @@ -469,7 +489,13 @@ fn emit_program_for_delete( None, )?; program.preassign_label_to_next_insn(after_main_loop_label); - + if has_parent_fks { + program.emit_insn(Insn::FkCounter { + increment_value: -1, + check_abort: true, + is_scope: true, + }); + } // Finalize program program.result_columns = plan.result_columns; program.table_references.extend(plan.table_references); @@ -514,6 +540,19 @@ fn emit_delete_insns( dest: key_reg, }); + if connection.foreign_keys_enabled() + && unsafe { &*table_reference }.btree().is_some() + && t_ctx.resolver.schema.any_incoming_fk_to(table_name) + { + emit_fk_parent_existence_checks( + program, + &t_ctx.resolver, + table_name, + main_table_cursor_id, + key_reg, + )?; + } + if unsafe { &*table_reference }.virtual_table().is_some() { let conflict_action = 0u16; let start_reg = key_reg; @@ -692,6 +731,518 @@ fn emit_delete_insns( Ok(()) } +/// Emit parent-side FK counter maintenance for UPDATE on a table with a composite PK. +/// +/// For every child FK that targets `parent_table_name`: +/// 1. Pass 1: If any child row currently references the OLD parent key, +/// increment the global FK counter (deferred violation potential). +/// We try an index probe on child(child_cols...) if available, else do a table scan. +/// 2. Pass 2: If any child row references the NEW parent key, decrement the counter +/// (because the reference would be “retargeted” by the update). +pub fn emit_fk_parent_pk_change_counters( + program: &mut ProgramBuilder, + incoming: &[ResolvedFkRef], + resolver: &Resolver, + old_pk_start: usize, + new_pk_start: usize, + n_cols: usize, +) -> crate::Result<()> { + if incoming.is_empty() { + return Ok(()); + } + for fk_ref in incoming.iter() { + let child_tbl = &fk_ref.child_table; + let child_cols = &fk_ref.fk.child_columns; + // Prefer exact-prefix index on child + let idx = resolver.schema.get_indices(&child_tbl.name).find(|ix| { + ix.columns.len() == child_cols.len() + && ix + .columns + .iter() + .zip(child_cols.iter()) + .all(|(ic, cc)| ic.name.eq_ignore_ascii_case(cc)) + }); + + if let Some(ix) = idx { + let icur = program.alloc_cursor_id(CursorType::BTreeIndex(ix.clone())); + program.emit_insn(Insn::OpenRead { + cursor_id: icur, + root_page: ix.root_page, + db: 0, + }); + + // Build child-probe key from OLD parent PK (1:1 map ensured by the column-name equality above) + // We just copy the OLD PK registers, apply index affinities before the probe. + let probe_start = old_pk_start; + + // Apply affinities for composite comparison + let aff: String = ix + .columns + .iter() + .map(|ic| { + let (_, col) = child_tbl + .get_column(&ic.name) + .expect("indexed child column not found"); + col.affinity().aff_mask() + }) + .collect(); + if let Some(count) = NonZeroUsize::new(n_cols) { + program.emit_insn(Insn::Affinity { + start_reg: probe_start, + count, + affinities: aff, + }); + } + + let found = program.allocate_label(); + program.emit_insn(Insn::Found { + cursor_id: icur, + target_pc: found, + record_reg: probe_start, + num_regs: n_cols, + }); + + // Not found => no increment + program.emit_insn(Insn::Close { cursor_id: icur }); + let skip = program.allocate_label(); + program.emit_insn(Insn::Goto { target_pc: skip }); + + // Found => increment + program.preassign_label_to_next_insn(found); + program.emit_insn(Insn::Close { cursor_id: icur }); + program.emit_insn(Insn::FkCounter { + increment_value: 1, + check_abort: false, + is_scope: false, + }); + program.preassign_label_to_next_insn(skip); + } else { + // Table-scan fallback with per-column checks (jump-if-NULL semantics) + let ccur = program.alloc_cursor_id(CursorType::BTreeTable(child_tbl.clone())); + program.emit_insn(Insn::OpenRead { + cursor_id: ccur, + root_page: child_tbl.root_page, + db: 0, + }); + + let done = program.allocate_label(); + program.emit_insn(Insn::Rewind { + cursor_id: ccur, + pc_if_empty: done, + }); + + let loop_top = program.allocate_label(); + let next_row = program.allocate_label(); + program.preassign_label_to_next_insn(loop_top); + + for (i, child_name) in child_cols.iter().enumerate() { + let (pos, _) = child_tbl.get_column(child_name).ok_or_else(|| { + crate::LimboError::InternalError(format!("child col {child_name} missing")) + })?; + let tmp = program.alloc_register(); + program.emit_insn(Insn::Column { + cursor_id: ccur, + column: pos, + dest: tmp, + default: None, + }); + + // Treat NULL as non-match: jump away immediately + program.emit_insn(Insn::IsNull { + reg: tmp, + target_pc: next_row, + }); + + // Eq(tmp, old_pk[i]) with Binary collation, jump-if-NULL enabled + let cont = program.allocate_label(); + program.emit_insn(Insn::Eq { + lhs: tmp, + rhs: old_pk_start + i, + target_pc: cont, + flags: CmpInsFlags::default().jump_if_null(), + collation: Some(super::collate::CollationSeq::Binary), + }); + program.emit_insn(Insn::Goto { + target_pc: next_row, + }); + program.preassign_label_to_next_insn(cont); + } + + // All columns matched OLD -> increment + program.emit_insn(Insn::FkCounter { + increment_value: 1, + check_abort: false, + is_scope: false, + }); + + program.preassign_label_to_next_insn(next_row); + program.emit_insn(Insn::Next { + cursor_id: ccur, + pc_if_next: loop_top, + }); + program.preassign_label_to_next_insn(done); + program.emit_insn(Insn::Close { cursor_id: ccur }); + } + } + + // PASS 2: count children of NEW key + for fk_ref in incoming.iter() { + let child_tbl = &fk_ref.child_table; + let child_cols = &fk_ref.fk.child_columns; + + let idx = resolver.schema.get_indices(&child_tbl.name).find(|ix| { + ix.columns.len() == child_cols.len() + && ix + .columns + .iter() + .zip(child_cols.iter()) + .all(|(ic, cc)| ic.name.eq_ignore_ascii_case(cc)) + }); + + if let Some(ix) = idx { + let icur = program.alloc_cursor_id(CursorType::BTreeIndex(ix.clone())); + program.emit_insn(Insn::OpenRead { + cursor_id: icur, + root_page: ix.root_page, + db: 0, + }); + + // Build probe from NEW PK registers; apply affinities + let probe_start = new_pk_start; + let aff: String = ix + .columns + .iter() + .map(|ic| { + let (_, col) = child_tbl + .get_column(&ic.name) + .expect("indexed child column not found"); + col.affinity().aff_mask() + }) + .collect(); + if let Some(count) = NonZeroUsize::new(n_cols) { + program.emit_insn(Insn::Affinity { + start_reg: probe_start, + count, + affinities: aff, + }); + } + + let found = program.allocate_label(); + program.emit_insn(Insn::Found { + cursor_id: icur, + target_pc: found, + record_reg: probe_start, + num_regs: n_cols, + }); + + // Not found => no decrement + program.emit_insn(Insn::Close { cursor_id: icur }); + let skip = program.allocate_label(); + program.emit_insn(Insn::Goto { target_pc: skip }); + + // Found => decrement + program.preassign_label_to_next_insn(found); + program.emit_insn(Insn::Close { cursor_id: icur }); + program.emit_insn(Insn::FkCounter { + increment_value: -1, + check_abort: false, + is_scope: false, + }); + program.preassign_label_to_next_insn(skip); + } else { + // Table-scan fallback on NEW key + let ccur = program.alloc_cursor_id(CursorType::BTreeTable(child_tbl.clone())); + program.emit_insn(Insn::OpenRead { + cursor_id: ccur, + root_page: child_tbl.root_page, + db: 0, + }); + + let done = program.allocate_label(); + program.emit_insn(Insn::Rewind { + cursor_id: ccur, + pc_if_empty: done, + }); + + let loop_top = program.allocate_label(); + let next_row = program.allocate_label(); + program.preassign_label_to_next_insn(loop_top); + + for (i, child_name) in child_cols.iter().enumerate() { + let (pos, _) = child_tbl.get_column(child_name).ok_or_else(|| { + crate::LimboError::InternalError(format!("child col {child_name} missing")) + })?; + let tmp = program.alloc_register(); + program.emit_insn(Insn::Column { + cursor_id: ccur, + column: pos, + dest: tmp, + default: None, + }); + + program.emit_insn(Insn::IsNull { + reg: tmp, + target_pc: next_row, + }); + + let cont = program.allocate_label(); + program.emit_insn(Insn::Eq { + lhs: tmp, + rhs: new_pk_start + i, + target_pc: cont, + flags: CmpInsFlags::default().jump_if_null(), + collation: Some(super::collate::CollationSeq::Binary), + }); + program.emit_insn(Insn::Goto { + target_pc: next_row, + }); + program.preassign_label_to_next_insn(cont); + } + + // All columns matched NEW: decrement + program.emit_insn(Insn::FkCounter { + increment_value: -1, + check_abort: false, + is_scope: false, + }); + + program.preassign_label_to_next_insn(next_row); + program.emit_insn(Insn::Next { + cursor_id: ccur, + pc_if_next: loop_top, + }); + program.preassign_label_to_next_insn(done); + program.emit_insn(Insn::Close { cursor_id: ccur }); + } + } + Ok(()) +} + +/// Emit checks that prevent updating/deleting a parent row that is still referenced by a child. +/// +/// If the global deferred-FK counter is zero, we skip all checks (fast path for no outstanding refs). +/// For each incoming FK: +/// Build the parent key (in FK parent-column order) from the current row. +/// Probe the child table for any row whose FK columns equal that key. +/// - If an exact child index exists on the FK columns, use `NotFound` against that index. +/// - Otherwise, scan the child table and compare each FK column (NULL short-circuits to “no match”). +/// If a referencing child is found: +/// - Deferred FK: increment counter (violation will be raised at COMMIT). +/// - Immediate FK: raise `SQLITE_CONSTRAINT_FOREIGNKEY` now. +pub fn emit_fk_parent_existence_checks( + program: &mut ProgramBuilder, + resolver: &Resolver, + parent_table_name: &str, + parent_cursor_id: usize, + parent_rowid_reg: usize, +) -> Result<()> { + let after_all = program.allocate_label(); + program.emit_insn(Insn::FkIfZero { + target_pc: after_all, + if_zero: true, + }); + + let parent_bt = resolver + .schema + .get_btree_table(parent_table_name) + .ok_or_else(|| crate::LimboError::InternalError("parent not btree".into()))?; + + for fk_ref in resolver.schema.incoming_fks_to(parent_table_name) { + // Resolve parent key columns + let parent_cols: Vec = if fk_ref.fk.parent_columns.is_empty() { + parent_bt + .primary_key_columns + .iter() + .map(|(n, _)| n.clone()) + .collect() + } else { + fk_ref.fk.parent_columns.clone() + }; + + // Load parent key values for THIS row into regs, in parent_cols order + let parent_cols_len = parent_cols.len(); + let parent_key_start = program.alloc_registers(parent_cols_len); + for (i, pcol) in parent_cols.iter().enumerate() { + let src = if pcol.eq_ignore_ascii_case("rowid") { + parent_rowid_reg + } else { + let (pos, col) = parent_bt + .get_column(&normalize_ident(pcol)) + .ok_or_else(|| { + crate::LimboError::InternalError(format!("col {pcol} missing")) + })?; + if col.is_rowid_alias { + parent_rowid_reg + } else { + // read current cell's column value + program.emit_insn(Insn::Column { + cursor_id: parent_cursor_id, + column: pos, + dest: parent_key_start + i, + default: None, + }); + continue; + } + }; + program.emit_insn(Insn::Copy { + src_reg: src, + dst_reg: parent_key_start + i, + extra_amount: 0, + }); + } + + // Build child-side probe key in child_columns order, from parent_key_start + // + // Map parent_col to child_col position 1:1 + let child_cols = &fk_ref.fk.child_columns; + // Try to find an index on child(child_cols...) to do an existance check + let child_idx = resolver + .schema + .get_indices(&fk_ref.child_table.name) + .find(|idx| { + idx.columns.len() == child_cols.len() + && idx + .columns + .iter() + .zip(child_cols.iter()) + .all(|(ic, cc)| ic.name.eq_ignore_ascii_case(cc)) + }); + + if let Some(idx) = child_idx { + // Index existence probe: Found -> violation + let icur = program.alloc_cursor_id(CursorType::BTreeIndex(idx.clone())); + program.emit_insn(Insn::OpenRead { + cursor_id: icur, + root_page: idx.root_page, + db: 0, + }); + + // Pack the child key regs from the parent key regs in fk order. + // Same order because we matched columns 1:1 above + let probe_start = program.alloc_registers(parent_cols_len); + for i in 0..parent_cols_len { + program.emit_insn(Insn::Copy { + src_reg: parent_key_start + i, + dst_reg: probe_start + i, + extra_amount: 0, + }); + } + + let ok = program.allocate_label(); + program.emit_insn(Insn::NotFound { + cursor_id: icur, + target_pc: ok, + record_reg: probe_start, + num_regs: parent_cols_len, + }); + + // found referencing child row = violation path + program.emit_insn(Insn::Close { cursor_id: icur }); + if fk_ref.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.preassign_label_to_next_insn(ok); + program.emit_insn(Insn::Close { cursor_id: icur }); + } else { + // Fallback: table-scan the child table + let ccur = program.alloc_cursor_id(CursorType::BTreeTable(fk_ref.child_table.clone())); + program.emit_insn(Insn::OpenRead { + cursor_id: ccur, + root_page: fk_ref.child_table.root_page, + db: 0, + }); + + let done = program.allocate_label(); + program.emit_insn(Insn::Rewind { + cursor_id: ccur, + pc_if_empty: done, + }); + + // Loop labels local to this scan + let loop_top = program.allocate_label(); + let next_row = program.allocate_label(); + + program.preassign_label_to_next_insn(loop_top); + + // For each FK column: require a match, if NULL or mismatch -> next_row + for (i, child_col) in child_cols.iter().enumerate() { + let (pos, _) = fk_ref + .child_table + .get_column(&normalize_ident(child_col)) + .ok_or_else(|| { + crate::LimboError::InternalError(format!("child col {child_col} missing")) + })?; + + let tmp = program.alloc_register(); + program.emit_insn(Insn::Column { + cursor_id: ccur, + column: pos, + dest: tmp, + default: None, + }); + + // NULL FK value => this child row cannot reference the parent, skip row + program.emit_insn(Insn::IsNull { + reg: tmp, + target_pc: next_row, + }); + + // Equal? continue to check next column; else jump to next_row + let cont_i = program.allocate_label(); + program.emit_insn(Insn::Eq { + lhs: tmp, + rhs: parent_key_start + i, + target_pc: cont_i, + flags: CmpInsFlags::default(), + collation: program.curr_collation(), + }); + // Not equal -> skip this child row + program.emit_insn(Insn::Goto { + target_pc: next_row, + }); + + // Equal path resumes here, then we check the next column + program.preassign_label_to_next_insn(cont_i); + } + + // If we reached here, all FK columns matched, violation + if fk_ref.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(), + }); + } + + // Advance to next child row and loop + program.preassign_label_to_next_insn(next_row); + program.emit_insn(Insn::Next { + cursor_id: ccur, + pc_if_next: loop_top, + }); + + program.preassign_label_to_next_insn(done); + program.emit_insn(Insn::Close { cursor_id: ccur }); + } + } + program.resolve_label(after_all, program.offset()); + Ok(()) +} + #[instrument(skip_all, level = Level::DEBUG)] fn emit_program_for_update( connection: &Arc, @@ -736,6 +1287,25 @@ fn emit_program_for_update( program.decr_nesting(); } + let fk_enabled = connection.foreign_keys_enabled(); + let table_name = plan + .table_references + .joined_tables() + .first() + .unwrap() + .table + .get_name(); + let has_child_fks = fk_enabled && !resolver.schema.get_fks_for_table(table_name).is_empty(); + let has_parent_fks = fk_enabled && resolver.schema.any_incoming_fk_to(table_name); + // statement-level FK scope open + if has_child_fks || has_parent_fks { + program.emit_insn(Insn::FkCounter { + increment_value: 1, + check_abort: false, + is_scope: true, + }); + } + // Initialize the main loop init_loop( program, @@ -803,6 +1373,13 @@ fn emit_program_for_update( program.preassign_label_to_next_insn(after_main_loop_label); + if has_child_fks || has_parent_fks { + program.emit_insn(Insn::FkCounter { + increment_value: -1, + check_abort: true, + is_scope: true, + }); + } after(program); program.result_columns = plan.returning.unwrap_or_default(); @@ -1067,6 +1644,234 @@ fn emit_update_insns( } } + if connection.foreign_keys_enabled() { + let rowid_new_reg = rowid_set_clause_reg.unwrap_or(beg); + if let Some(table_btree) = unsafe { &*table_ref }.btree() { + //first, stablize the image of the NEW row in the registers + if !table_btree.primary_key_columns.is_empty() { + let set_cols: std::collections::HashSet = plan + .set_clauses + .iter() + .filter_map(|(i, _)| if *i == ROWID_SENTINEL { None } else { Some(*i) }) + .collect(); + for (pk_name, _) in &table_btree.primary_key_columns { + let (pos, col) = table_btree.get_column(pk_name).unwrap(); + if !set_cols.contains(&pos) { + if col.is_rowid_alias { + program.emit_insn(Insn::Copy { + src_reg: rowid_new_reg, + dst_reg: start + pos, + extra_amount: 0, + }); + } else { + program.emit_insn(Insn::Column { + cursor_id, + column: pos, + dest: start + pos, + default: None, + }); + } + } + } + } + if t_ctx.resolver.schema.has_child_fks(table_name) { + // Child-side checks: + // this ensures updated row still satisfies child FKs that point OUT from this table + emit_fk_child_existence_checks( + program, + &t_ctx.resolver, + &table_btree, + table_name, + start, + rowid_new_reg, + &plan + .set_clauses + .iter() + .map(|(i, _)| *i) + .collect::>(), + )?; + } + // Parent-side checks: + // We only need to do work if the referenced key (the parent key) might change. + // we detect that by comparing OLD vs NEW primary key representation + // then run parent FK checks only when it actually changes. + if t_ctx.resolver.schema.any_incoming_fk_to(table_name) { + let updated_parent_positions: HashSet = + plan.set_clauses.iter().map(|(i, _)| *i).collect(); + + // If no incoming FK’s parent key can be affected by these updates, skip the whole parent-FK block. + let incoming = t_ctx.resolver.schema.incoming_fks_to(table_name); + let parent_tbl = &table_btree; + let maybe_affects_parent_key = incoming + .iter() + .any(|r| r.parent_key_may_change(&updated_parent_positions, parent_tbl)); + if maybe_affects_parent_key { + let pk_len = table_btree.primary_key_columns.len(); + match pk_len { + 0 => { + // Rowid table: the implicit PK is rowid. + // If rowid is unchanged then we skip, else check that no child row still references the OLD key. + let skip_parent_fk = program.allocate_label(); + let old_rowid_reg = beg; + let new_rowid_reg = rowid_set_clause_reg.unwrap_or(beg); + + program.emit_insn(Insn::Eq { + lhs: new_rowid_reg, + rhs: old_rowid_reg, + target_pc: skip_parent_fk, + flags: CmpInsFlags::default(), + collation: program.curr_collation(), + }); + // Rowid changed: check incoming FKs (children) that reference this parent row + emit_fk_parent_existence_checks( + program, + &t_ctx.resolver, + table_name, + cursor_id, + old_rowid_reg, + )?; + program.preassign_label_to_next_insn(skip_parent_fk); + } + 1 => { + // Single-column declared PK, may be a rowid alias or a real column. + // If PK value unchanged then skip, else verify no child still references OLD key. + let (pk_name, _) = &table_btree.primary_key_columns[0]; + let (pos, col) = table_btree.get_column(pk_name).unwrap(); + + let old_reg = program.alloc_register(); + if col.is_rowid_alias { + program.emit_insn(Insn::RowId { + cursor_id, + dest: old_reg, + }); + } else { + program.emit_insn(Insn::Column { + cursor_id, + column: pos, + dest: old_reg, + default: None, + }); + } + let new_reg = if col.is_rowid_alias { + rowid_new_reg + } else { + start + pos + }; + + let skip_parent_fk = program.allocate_label(); + program.emit_insn(Insn::Eq { + lhs: old_reg, + rhs: new_reg, + target_pc: skip_parent_fk, + flags: CmpInsFlags::default(), + collation: program.curr_collation(), + }); + emit_fk_parent_existence_checks( + program, + &t_ctx.resolver, + table_name, + cursor_id, + beg, + )?; + program.preassign_label_to_next_insn(skip_parent_fk); + } + _ => { + // Composite PK: + // 1. Materialize OLD PK vector from current row. + // 2. Materialize NEW PK vector from updated registers. + // 3. If any component differs, the PK changes -> run composite parent-FK update flow. + let old_pk_start = program.alloc_registers(pk_len); + for (i, (pk_name, _)) in + table_btree.primary_key_columns.iter().enumerate() + { + let (pos, col) = table_btree.get_column(pk_name).unwrap(); + if col.is_rowid_alias { + program.emit_insn(Insn::Copy { + src_reg: beg, + dst_reg: old_pk_start + i, + extra_amount: 0, + }); + } else { + program.emit_insn(Insn::Column { + cursor_id, + column: pos, + dest: old_pk_start + i, + default: None, + }); + } + } + + // Build NEW PK values from the updated registers + let new_pk_start = program.alloc_registers(pk_len); + for (i, (pk_name, _)) in + table_btree.primary_key_columns.iter().enumerate() + { + let (pos, col) = table_btree.get_column(pk_name).unwrap(); + let src = if col.is_rowid_alias { + rowid_new_reg + } else { + start + pos // Updated value from SET clause + }; + program.emit_insn(Insn::Copy { + src_reg: src, + dst_reg: new_pk_start + i, + extra_amount: 0, + }); + } + + // Compare OLD vs NEW to see if PK is changing + let skip_parent_fk = program.allocate_label(); + let pk_changed = program.allocate_label(); + + for i in 0..pk_len { + if i == pk_len - 1 { + // Last comparison, if equal, all are equal + program.emit_insn(Insn::Eq { + lhs: old_pk_start + i, + rhs: new_pk_start + i, + target_pc: skip_parent_fk, + flags: CmpInsFlags::default(), + collation: program.curr_collation(), + }); + // Not equal - PK is changing + program.emit_insn(Insn::Goto { + target_pc: pk_changed, + }); + } else { + // Not last comparison + let next_check = program.allocate_label(); + program.emit_insn(Insn::Eq { + lhs: old_pk_start + i, + rhs: new_pk_start + i, + target_pc: next_check, // Equal, check next component + flags: CmpInsFlags::default(), + collation: program.curr_collation(), + }); + // Not equal - PK is changing + program.emit_insn(Insn::Goto { + target_pc: pk_changed, + }); + program.preassign_label_to_next_insn(next_check); + } + } + program.preassign_label_to_next_insn(pk_changed); + // PK changed: maintain the deferred FK counter in two passes + emit_fk_parent_pk_change_counters( + program, + &incoming, + &t_ctx.resolver, + old_pk_start, + new_pk_start, + pk_len, + )?; + program.preassign_label_to_next_insn(skip_parent_fk); + } + } + } + } + } + } + for (index, (idx_cursor_id, record_reg)) in plan.indexes_to_update.iter().zip(&index_cursors) { // We need to know whether or not the OLD values satisfied the predicate on the // partial index, so we can know whether or not to delete the old index entry, @@ -1518,6 +2323,152 @@ fn emit_update_insns( Ok(()) } +pub fn emit_fk_child_existence_checks( + program: &mut ProgramBuilder, + resolver: &Resolver, + table: &BTreeTable, + table_name: &str, + start_reg: usize, + rowid_reg: usize, + updated_cols: &HashSet, +) -> Result<()> { + let after_all = program.allocate_label(); + program.emit_insn(Insn::FkIfZero { + target_pc: after_all, + if_zero: true, + }); + + for fk_ref in resolver.schema.outgoing_fks_of(table_name) { + // Skip when the child key is untouched (including rowid-alias special case) + if !fk_ref.child_key_changed(updated_cols, table) { + continue; + } + + let fk_ok = program.allocate_label(); + + // look for NULLs in any child FK column + for child_name in &fk_ref.child_cols { + let (i, col) = table.get_column(child_name).unwrap(); + let src = if col.is_rowid_alias { + rowid_reg + } else { + start_reg + i + }; + program.emit_insn(Insn::IsNull { + reg: src, + target_pc: fk_ok, + }); + } + + if fk_ref.parent_uses_rowid { + // Fast rowid probe on the parent table + let parent_tbl = resolver + .schema + .get_btree_table(&fk_ref.fk.parent_table) + .expect("Parent must be btree"); + + let pcur = program.alloc_cursor_id(CursorType::BTreeTable(parent_tbl.clone())); + program.emit_insn(Insn::OpenRead { + cursor_id: pcur, + root_page: parent_tbl.root_page, + db: 0, + }); + + let (i_child, col_child) = table.get_column(&fk_ref.child_cols[0]).unwrap(); + let val_reg = if col_child.is_rowid_alias { + rowid_reg + } else { + start_reg + i_child + }; + + let violation = program.allocate_label(); + program.emit_insn(Insn::NotExists { + cursor: pcur, + rowid_reg: val_reg, + target_pc: violation, + }); + program.emit_insn(Insn::Close { cursor_id: pcur }); + program.emit_insn(Insn::Goto { target_pc: fk_ok }); + + program.preassign_label_to_next_insn(violation); + program.emit_insn(Insn::Close { cursor_id: pcur }); + if fk_ref.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 { + // Unique-index probe on the parent (already resolved) + let parent_idx = fk_ref + .parent_unique_index + .as_ref() + .expect("parent unique index required"); + 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 probe key from NEW child values in fk order + let n = fk_ref.child_cols.len(); + let probe_start = program.alloc_registers(n); + for (k, child_name) in fk_ref.child_cols.iter().enumerate() { + let (i, col) = table.get_column(child_name).unwrap(); + program.emit_insn(Insn::Copy { + src_reg: if col.is_rowid_alias { + rowid_reg + } else { + start_reg + i + }, + dst_reg: probe_start + k, + extra_amount: 0, + }); + } + + let found = program.allocate_label(); + program.emit_insn(Insn::Found { + cursor_id: icur, + target_pc: found, + record_reg: probe_start, + num_regs: n, + }); + + // Not found => violation + program.emit_insn(Insn::Close { cursor_id: icur }); + if fk_ref.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 }); + } + + program.preassign_label_to_next_insn(fk_ok); + } + + program.resolve_label(after_all, program.offset()); + Ok(()) +} + pub fn prepare_cdc_if_necessary( program: &mut ProgramBuilder, schema: &Schema, diff --git a/core/translate/insert.rs b/core/translate/insert.rs index 8f4b8158f..01d1355a1 100644 --- a/core/translate/insert.rs +++ b/core/translate/insert.rs @@ -87,7 +87,7 @@ pub fn translate_insert( let has_child_fks = fk_enabled && !resolver .schema - .get_foreign_keys_for_table(table_name.as_str()) + .get_fks_for_table(table_name.as_str()) .is_empty(); let has_parent_fks = fk_enabled && resolver.schema.any_incoming_fk_to(table_name.as_str()); @@ -1146,7 +1146,6 @@ pub fn translate_insert( &mut result_columns, cdc_table.as_ref().map(|c| c.0), row_done_label, - connection, )?; } else { // UpsertDo::Nothing case @@ -1910,7 +1909,7 @@ fn emit_fk_checks_for_insert( }); // Iterate child FKs declared on this table - for fk in resolver.schema.get_foreign_keys_for_table(table_name) { + for fk in resolver.schema.get_fks_for_table(table_name) { let fk_ok = program.allocate_label(); // If any child column is NULL, skip this FK diff --git a/core/translate/pragma.rs b/core/translate/pragma.rs index 57a658212..0a527a68c 100644 --- a/core/translate/pragma.rs +++ b/core/translate/pragma.rs @@ -4,7 +4,7 @@ use chrono::Datelike; use std::sync::Arc; use turso_macros::match_ignore_ascii_case; -use turso_parser::ast::{self, ColumnDefinition, Expr, Literal, Name}; +use turso_parser::ast::{self, ColumnDefinition, Expr, Literal}; use turso_parser::ast::{PragmaName, QualifiedName}; use super::integrity_check::translate_integrity_check; @@ -388,10 +388,10 @@ fn update_pragma( Ok((program, TransactionMode::None)) } PragmaName::ForeignKeys => { - let enabled = match &value { - Expr::Id(name) | Expr::Name(name) => { - let name_str = name.as_str().as_bytes(); - match_ignore_ascii_case!(match name_str { + let enabled = match value { + Expr::Name(name) | Expr::Id(name) => { + let name_bytes = name.as_str().as_bytes(); + match_ignore_ascii_case!(match name_bytes { b"ON" | b"TRUE" | b"YES" | b"1" => true, _ => false, }) diff --git a/core/translate/upsert.rs b/core/translate/upsert.rs index ffcff23e5..f9bfd5af9 100644 --- a/core/translate/upsert.rs +++ b/core/translate/upsert.rs @@ -5,10 +5,15 @@ use std::{collections::HashMap, sync::Arc}; use turso_parser::ast::{self, Upsert}; use crate::error::SQLITE_CONSTRAINT_PRIMARYKEY; +use crate::translate::emitter::{ + emit_fk_child_existence_checks, emit_fk_parent_existence_checks, + emit_fk_parent_pk_change_counters, +}; use crate::translate::expr::{walk_expr, WalkControl}; use crate::translate::insert::format_unique_violation_desc; use crate::translate::planner::ROWID_STRS; use crate::vdbe::insn::CmpInsFlags; +use crate::Connection; use crate::{ bail_parse_error, error::SQLITE_CONSTRAINT_NOTNULL, @@ -346,6 +351,7 @@ pub fn emit_upsert( returning: &mut [ResultSetColumn], cdc_cursor_id: Option, row_done_label: BranchOffset, + connection: &Arc, ) -> crate::Result<()> { // Seek & snapshot CURRENT program.emit_insn(Insn::SeekRowid { @@ -464,10 +470,179 @@ pub fn emit_upsert( } } + let (changed_cols, rowid_changed) = collect_changed_cols(table, set_pairs); + + if let Some(bt) = table.btree() { + if connection.foreign_keys_enabled() { + let rowid_new_reg = new_rowid_reg.unwrap_or(conflict_rowid_reg); + + // Child-side checks + if resolver.schema.has_child_fks(bt.name.as_str()) { + emit_fk_child_existence_checks( + program, + resolver, + &bt, + table.get_name(), + new_start, + rowid_new_reg, + &changed_cols, + )?; + } + + // Parent-side checks only if any incoming FK could care + if resolver.schema.any_incoming_fk_to(table.get_name()) { + // if parent key can't change, skip + let updated_parent_positions: HashSet = + set_pairs.iter().map(|(i, _)| *i).collect(); + let incoming = resolver.schema.incoming_fks_to(table.get_name()); + let parent_key_may_change = incoming + .iter() + .any(|r| r.parent_key_may_change(&updated_parent_positions, &bt)); + + if parent_key_may_change { + let skip_parent_fk = program.allocate_label(); + let pk_len = bt.primary_key_columns.len(); + + match pk_len { + 0 => { + // implicit rowid + program.emit_insn(Insn::Eq { + lhs: rowid_new_reg, + rhs: conflict_rowid_reg, + target_pc: skip_parent_fk, + flags: CmpInsFlags::default(), + collation: program.curr_collation(), + }); + emit_fk_parent_existence_checks( + program, + resolver, + table.get_name(), + tbl_cursor_id, + conflict_rowid_reg, + )?; + program.preassign_label_to_next_insn(skip_parent_fk); + } + 1 => { + // single-col declared PK + let (pk_name, _) = &bt.primary_key_columns[0]; + let (pos, col) = bt.get_column(pk_name).unwrap(); + + let old_reg = program.alloc_register(); + if col.is_rowid_alias { + program.emit_insn(Insn::RowId { + cursor_id: tbl_cursor_id, + dest: old_reg, + }); + } else { + program.emit_insn(Insn::Column { + cursor_id: tbl_cursor_id, + column: pos, + dest: old_reg, + default: None, + }); + } + let new_reg = new_start + pos; + + let skip = program.allocate_label(); + program.emit_insn(Insn::Eq { + lhs: old_reg, + rhs: new_reg, + target_pc: skip, + flags: CmpInsFlags::default(), + collation: program.curr_collation(), + }); + emit_fk_parent_existence_checks( + program, + resolver, + table.get_name(), + tbl_cursor_id, + conflict_rowid_reg, + )?; + program.preassign_label_to_next_insn(skip); + } + _ => { + // composite PK: build OLD/NEW vectors and do the 2-pass counter logic + let old_pk_start = program.alloc_registers(pk_len); + for (i, (pk_name, _)) in bt.primary_key_columns.iter().enumerate() { + let (pos, col) = bt.get_column(pk_name).unwrap(); + if col.is_rowid_alias { + // old rowid (UPSERT target) == conflict_rowid_reg + program.emit_insn(Insn::Copy { + src_reg: conflict_rowid_reg, + dst_reg: old_pk_start + i, + extra_amount: 0, + }); + } else { + program.emit_insn(Insn::Column { + cursor_id: tbl_cursor_id, + column: pos, + dest: old_pk_start + i, + default: None, + }); + } + } + + let new_pk_start = program.alloc_registers(pk_len); + for (i, (pk_name, _)) in bt.primary_key_columns.iter().enumerate() { + let (pos, col) = bt.get_column(pk_name).unwrap(); + let src = if col.is_rowid_alias { + rowid_new_reg + } else { + new_start + pos + }; + program.emit_insn(Insn::Copy { + src_reg: src, + dst_reg: new_pk_start + i, + extra_amount: 0, + }); + } + + // Compare OLD vs NEW, if all equal then skip + let skip = program.allocate_label(); + let changed = program.allocate_label(); + for i in 0..pk_len { + if i == pk_len - 1 { + program.emit_insn(Insn::Eq { + lhs: old_pk_start + i, + rhs: new_pk_start + i, + target_pc: skip, + flags: CmpInsFlags::default(), + collation: program.curr_collation(), + }); + program.emit_insn(Insn::Goto { target_pc: changed }); + } else { + let next = program.allocate_label(); + program.emit_insn(Insn::Eq { + lhs: old_pk_start + i, + rhs: new_pk_start + i, + target_pc: next, + flags: CmpInsFlags::default(), + collation: program.curr_collation(), + }); + program.emit_insn(Insn::Goto { target_pc: changed }); + program.preassign_label_to_next_insn(next); + } + } + + program.preassign_label_to_next_insn(changed); + emit_fk_parent_pk_change_counters( + program, + &incoming, + resolver, + old_pk_start, + new_pk_start, + pk_len, + )?; + program.preassign_label_to_next_insn(skip); + } + } + } + } + } + } + // Index rebuild (DELETE old, INSERT new), honoring partial-index WHEREs if let Some(before) = before_start { - let (changed_cols, rowid_changed) = collect_changed_cols(table, set_pairs); - for (idx_name, _root, idx_cid) in idx_cursors { let idx_meta = resolver .schema diff --git a/tests/integration/fuzz/mod.rs b/tests/integration/fuzz/mod.rs index eb3f8a1af..42b7660f4 100644 --- a/tests/integration/fuzz/mod.rs +++ b/tests/integration/fuzz/mod.rs @@ -3,12 +3,11 @@ pub mod grammar_generator; #[cfg(test)] mod tests { use rand::seq::{IndexedRandom, SliceRandom}; - use std::collections::HashSet; - use turso_core::DatabaseOpts; - use rand::{Rng, SeedableRng}; use rand_chacha::ChaCha8Rng; use rusqlite::{params, types::Value}; + use std::{collections::HashSet, io::Write}; + use turso_core::DatabaseOpts; use crate::{ common::{ @@ -646,6 +645,422 @@ mod tests { "Different results! limbo: {:?}, sqlite: {:?}, seed: {}, query: {}, table def: {}", limbo_rows, sqlite_rows, seed, query, table_defs[i] ); + + } + } + } + } + + pub fn fk_single_pk_mutation_fuzz() { + let _ = env_logger::try_init(); + let (mut rng, seed) = rng_from_time(); + println!("fk_single_pk_mutation_fuzz seed: {seed}"); + + const OUTER_ITERS: usize = 50; + const INNER_ITERS: usize = 200; + + for outer in 0..OUTER_ITERS { + println!("fk_single_pk_mutation_fuzz {}/{}", outer + 1, OUTER_ITERS); + + let limbo_db = TempDatabase::new_empty(true); + let sqlite_db = TempDatabase::new_empty(true); + let limbo = limbo_db.connect_limbo(); + let sqlite = rusqlite::Connection::open(sqlite_db.path.clone()).unwrap(); + + // Statement log for this iteration + let mut stmts: Vec = Vec::new(); + let mut log_and_exec = |sql: &str| { + stmts.push(sql.to_string()); + sql.to_string() + }; + + // Enable FKs in both engines + let s = log_and_exec("PRAGMA foreign_keys=ON"); + limbo_exec_rows(&limbo_db, &limbo, &s); + sqlite.execute(&s, params![]).unwrap(); + + // DDL + let s = log_and_exec("CREATE TABLE p(id INTEGER PRIMARY KEY, a INT, b INT)"); + limbo_exec_rows(&limbo_db, &limbo, &s); + sqlite.execute(&s, params![]).unwrap(); + + let s = log_and_exec( + "CREATE TABLE c(id INTEGER PRIMARY KEY, x INT, y INT, FOREIGN KEY(x) REFERENCES p(id))", + ); + limbo_exec_rows(&limbo_db, &limbo, &s); + sqlite.execute(&s, params![]).unwrap(); + + // Seed parent + let n_par = rng.random_range(5..=40); + let mut used_ids = std::collections::HashSet::new(); + for _ in 0..n_par { + let mut id; + loop { + id = rng.random_range(1..=200) as i64; + if used_ids.insert(id) { + break; + } + } + let a = rng.random_range(-5..=25); + let b = rng.random_range(-5..=25); + let stmt = log_and_exec(&format!("INSERT INTO p VALUES ({id}, {a}, {b})")); + limbo_exec_rows(&limbo_db, &limbo, &stmt); + sqlite.execute(&stmt, params![]).unwrap(); + } + + // Seed child + let n_child = rng.random_range(5..=80); + for i in 0..n_child { + let id = 1000 + i as i64; + let x = if rng.random_bool(0.8) { + *used_ids.iter().choose(&mut rng).unwrap() + } else { + rng.random_range(1..=220) as i64 + }; + let y = rng.random_range(-10..=10); + let stmt = log_and_exec(&format!("INSERT INTO c VALUES ({id}, {x}, {y})")); + match ( + sqlite.execute(&stmt, params![]), + limbo_exec_rows_fallible(&limbo_db, &limbo, &stmt), + ) { + (Ok(_), Ok(_)) => {} + (Err(_), Err(_)) => {} + (x, y) => { + eprintln!("\n=== FK fuzz failure (seeding mismatch) ==="); + eprintln!("seed: {seed}, outer: {}", outer + 1); + eprintln!("sqlite: {x:?}, limbo: {y:?}"); + eprintln!("last stmt: {stmt}"); + eprintln!("--- replay statements ({}) ---", stmts.len()); + for (i, s) in stmts.iter().enumerate() { + eprintln!("{:04}: {};", i + 1, s); + } + panic!("Seeding child insert mismatch"); + } + } + } + + // Mutations + for _ in 0..INNER_ITERS { + let action = rng.random_range(0..6); + let stmt = match action { + // Parent INSERT + 0 => { + let mut id; + let mut tries = 0; + loop { + id = rng.random_range(1..=250) as i64; + if !used_ids.contains(&id) || tries > 10 { + break; + } + tries += 1; + } + let a = rng.random_range(-5..=25); + let b = rng.random_range(-5..=25); + format!("INSERT INTO p VALUES({id}, {a}, {b})") + } + // Parent UPDATE + 1 => { + if rng.random_bool(0.5) { + let old = rng.random_range(1..=250); + let new_id = rng.random_range(1..=260); + format!("UPDATE p SET id={new_id} WHERE id={old}") + } else { + let a = rng.random_range(-5..=25); + let b = rng.random_range(-5..=25); + let tgt = rng.random_range(1..=260); + format!("UPDATE p SET a={a}, b={b} WHERE id={tgt}") + } + } + // Parent DELETE + 2 => { + let del_id = rng.random_range(1..=260); + format!("DELETE FROM p WHERE id={del_id}") + } + // Child INSERT + 3 => { + let id = rng.random_range(1000..=2000); + let x = if rng.random_bool(0.7) { + if let Some(p) = used_ids.iter().choose(&mut rng) { + *p + } else { + rng.random_range(1..=260) as i64 + } + } else { + rng.random_range(1..=260) as i64 + }; + let y = rng.random_range(-10..=10); + format!("INSERT INTO c VALUES({id}, {x}, {y})") + } + // Child UPDATE + 4 => { + let pick = rng.random_range(1000..=2000); + if rng.random_bool(0.6) { + let new_x = if rng.random_bool(0.7) { + if let Some(p) = used_ids.iter().choose(&mut rng) { + *p + } else { + rng.random_range(1..=260) as i64 + } + } else { + rng.random_range(1..=260) as i64 + }; + format!("UPDATE c SET x={new_x} WHERE id={pick}") + } else { + let new_y = rng.random_range(-10..=10); + format!("UPDATE c SET y={new_y} WHERE id={pick}") + } + } + // Child DELETE + _ => { + let pick = rng.random_range(1000..=2000); + format!("DELETE FROM c WHERE id={pick}") + } + }; + + let stmt = log_and_exec(&stmt); + + let sres = sqlite.execute(&stmt, params![]); + let lres = limbo_exec_rows_fallible(&limbo_db, &limbo, &stmt); + + match (sres, lres) { + (Ok(_), Ok(_)) => { + if stmt.starts_with("INSERT INTO p VALUES(") { + if let Some(tok) = stmt.split_whitespace().nth(4) { + if let Some(idtok) = tok.split(['(', ',']).nth(1) { + if let Ok(idnum) = idtok.parse::() { + used_ids.insert(idnum); + } + } + } + } + let sp = sqlite_exec_rows(&sqlite, "SELECT id,a,b FROM p ORDER BY id"); + let sc = sqlite_exec_rows(&sqlite, "SELECT id,x,y FROM c ORDER BY id"); + let lp = + limbo_exec_rows(&limbo_db, &limbo, "SELECT id,a,b FROM p ORDER BY id"); + let lc = + limbo_exec_rows(&limbo_db, &limbo, "SELECT id,x,y FROM c ORDER BY id"); + + if sp != lp || sc != lc { + eprintln!("\n=== FK fuzz failure (state mismatch) ==="); + eprintln!("seed: {seed}, outer: {}", outer + 1); + eprintln!("last stmt: {stmt}"); + eprintln!("sqlite p: {sp:?}\nsqlite c: {sc:?}"); + eprintln!("limbo p: {lp:?}\nlimbo c: {lc:?}"); + eprintln!("--- replay statements ({}) ---", stmts.len()); + for (i, s) in stmts.iter().enumerate() { + eprintln!("{:04}: {};", i + 1, s); + } + panic!("State mismatch"); + } + } + (Err(_), Err(_)) => { /* parity OK */ } + (ok_sqlite, ok_limbo) => { + eprintln!("\n=== FK fuzz failure (outcome mismatch) ==="); + eprintln!("seed: {seed}, outer: {}", outer + 1); + eprintln!("sqlite: {ok_sqlite:?}, limbo: {ok_limbo:?}"); + eprintln!("last stmt: {stmt}"); + // dump final states to help decide who is right + let sp = sqlite_exec_rows(&sqlite, "SELECT id,a,b FROM p ORDER BY id"); + let sc = sqlite_exec_rows(&sqlite, "SELECT id,x,y FROM c ORDER BY id"); + let lp = + limbo_exec_rows(&limbo_db, &limbo, "SELECT id,a,b FROM p ORDER BY id"); + let lc = + limbo_exec_rows(&limbo_db, &limbo, "SELECT id,x,y FROM c ORDER BY id"); + eprintln!("sqlite p: {sp:?}\nsqlite c: {sc:?}"); + eprintln!("turso p: {lp:?}\nturso c: {lc:?}"); + eprintln!( + "--- writing ({}) statements to fk_fuzz_statements.sql ---", + stmts.len() + ); + let mut file = std::fs::File::create("fk_fuzz_statements.sql").unwrap(); + for s in stmts.iter() { + let _ = file.write_fmt(format_args!("{s};\n")); + } + file.flush().unwrap(); + panic!("DML outcome mismatch, statements written to tests/fk_fuzz_statements.sql"); + } + } + } + } + } + + #[test] + #[ignore] // TODO: un-ignore when UNIQUE constraints are fixed + pub fn fk_composite_pk_mutation_fuzz() { + let _ = env_logger::try_init(); + let (mut rng, seed) = rng_from_time(); + println!("fk_composite_pk_mutation_fuzz seed: {seed}"); + + const OUTER_ITERS: usize = 30; + const INNER_ITERS: usize = 200; + + for outer in 0..OUTER_ITERS { + println!( + "fk_composite_pk_mutation_fuzz {}/{}", + outer + 1, + OUTER_ITERS + ); + + let limbo_db = TempDatabase::new_empty(true); + let sqlite_db = TempDatabase::new_empty(true); + let limbo = limbo_db.connect_limbo(); + let sqlite = rusqlite::Connection::open(sqlite_db.path.clone()).unwrap(); + + let mut stmts: Vec = Vec::new(); + let mut log_and_exec = |sql: &str| { + stmts.push(sql.to_string()); + sql.to_string() + }; + + // Enable FKs in both engines + let _ = log_and_exec("PRAGMA foreign_keys=ON"); + limbo_exec_rows(&limbo_db, &limbo, "PRAGMA foreign_keys=ON"); + sqlite.execute("PRAGMA foreign_keys=ON", params![]).unwrap(); + + // Parent PK is composite (a,b). Child references (x,y) -> (a,b). + let s = log_and_exec( + "CREATE TABLE p(a INT NOT NULL, b INT NOT NULL, v INT, PRIMARY KEY(a,b))", + ); + limbo_exec_rows(&limbo_db, &limbo, &s); + sqlite.execute(&s, params![]).unwrap(); + + let s = log_and_exec( + "CREATE TABLE c(id INTEGER PRIMARY KEY, x INT, y INT, w INT, \ + FOREIGN KEY(x,y) REFERENCES p(a,b))", + ); + limbo_exec_rows(&limbo_db, &limbo, &s); + sqlite.execute(&s, params![]).unwrap(); + + // Seed parent: small grid of (a,b) + let mut pairs: Vec<(i64, i64)> = Vec::new(); + for _ in 0..rng.random_range(5..=25) { + let a = rng.random_range(-3..=6); + let b = rng.random_range(-3..=6); + if !pairs.contains(&(a, b)) { + pairs.push((a, b)); + let v = rng.random_range(0..=20); + let stmt = log_and_exec(&format!("INSERT INTO p VALUES({a},{b},{v})")); + limbo_exec_rows(&limbo_db, &limbo, &stmt); + sqlite.execute(&stmt, params![]).unwrap(); + } + } + + // Seed child rows, 70% chance to reference existing (a,b) + for i in 0..rng.random_range(5..=60) { + let id = 5000 + i as i64; + let (x, y) = if rng.random_bool(0.7) { + *pairs.choose(&mut rng).unwrap_or(&(0, 0)) + } else { + (rng.random_range(-4..=7), rng.random_range(-4..=7)) + }; + let w = rng.random_range(-10..=10); + let stmt = log_and_exec(&format!("INSERT INTO c VALUES({id}, {x}, {y}, {w})")); + let _ = sqlite.execute(&stmt, params![]); + let _ = limbo_exec_rows_fallible(&limbo_db, &limbo, &stmt); + } + + for _ in 0..INNER_ITERS { + let op = rng.random_range(0..6); + let stmt = log_and_exec(&match op { + // INSERT parent + 0 => { + let a = rng.random_range(-4..=8); + let b = rng.random_range(-4..=8); + let v = rng.random_range(0..=20); + format!("INSERT INTO p VALUES({a},{b},{v})") + } + // UPDATE parent composite key (a,b) + 1 => { + let a_old = rng.random_range(-4..=8); + let b_old = rng.random_range(-4..=8); + let a_new = rng.random_range(-4..=8); + let b_new = rng.random_range(-4..=8); + format!("UPDATE p SET a={a_new}, b={b_new} WHERE a={a_old} AND b={b_old}") + } + // DELETE parent + 2 => { + let a = rng.random_range(-4..=8); + let b = rng.random_range(-4..=8); + format!("DELETE FROM p WHERE a={a} AND b={b}") + } + // INSERT child + 3 => { + let id = rng.random_range(5000..=7000); + let (x, y) = if rng.random_bool(0.7) { + *pairs.choose(&mut rng).unwrap_or(&(0, 0)) + } else { + (rng.random_range(-4..=8), rng.random_range(-4..=8)) + }; + let w = rng.random_range(-10..=10); + format!("INSERT INTO c VALUES({id},{x},{y},{w})") + } + // UPDATE child FK columns (x,y) + 4 => { + let id = rng.random_range(5000..=7000); + let (x, y) = if rng.random_bool(0.7) { + *pairs.choose(&mut rng).unwrap_or(&(0, 0)) + } else { + (rng.random_range(-4..=8), rng.random_range(-4..=8)) + }; + format!("UPDATE c SET x={x}, y={y} WHERE id={id}") + } + // DELETE child + _ => { + let id = rng.random_range(5000..=7000); + format!("DELETE FROM c WHERE id={id}") + } + }); + + let sres = sqlite.execute(&stmt, params![]); + let lres = limbo_exec_rows_fallible(&limbo_db, &limbo, &stmt); + + match (sres, lres) { + (Ok(_), Ok(_)) => { + // Compare canonical states + let sp = sqlite_exec_rows(&sqlite, "SELECT a,b,v FROM p ORDER BY a,b,v"); + let sc = sqlite_exec_rows(&sqlite, "SELECT id,x,y,w FROM c ORDER BY id"); + let lp = limbo_exec_rows( + &limbo_db, + &limbo, + "SELECT a,b,v FROM p ORDER BY a,b,v", + ); + let lc = limbo_exec_rows( + &limbo_db, + &limbo, + "SELECT id,x,y,w FROM c ORDER BY id", + ); + assert_eq!(sp, lp, "seed {seed}, stmt {stmt}"); + assert_eq!(sc, lc, "seed {seed}, stmt {stmt}"); + } + (Err(_), Err(_)) => { /* both errored -> parity OK */ } + (ok_s, ok_l) => { + eprintln!( + "Mismatch sqlite={ok_s:?}, limbo={ok_l:?}, stmt={stmt}, seed={seed}" + ); + let sp = sqlite_exec_rows(&sqlite, "SELECT a,b,v FROM p ORDER BY a,b,v"); + let sc = sqlite_exec_rows(&sqlite, "SELECT id,x,y,w FROM c ORDER BY id"); + let lp = limbo_exec_rows( + &limbo_db, + &limbo, + "SELECT a,b,v FROM p ORDER BY a,b,v", + ); + let lc = limbo_exec_rows( + &limbo_db, + &limbo, + "SELECT id,x,y,w FROM c ORDER BY id", + ); + eprintln!( + "sqlite p={sp:?}\nsqlite c={sc:?}\nlimbo p={lp:?}\nlimbo c={lc:?}" + ); + let mut file = + std::fs::File::create("fk_composite_fuzz_statements.sql").unwrap(); + for s in stmts.iter() { + let _ = writeln!(&file, "{s};"); + } + file.flush().unwrap(); + panic!("DML outcome mismatch, sql file written to tests/fk_composite_fuzz_statements.sql"); + } + } } } }