Remove unnecessary FK resolution on schema parsing

core/incremental/compiler.rs

@@ -2245,6 +2245,7 @@ mod tests {
                 is_strict: false,
                 has_autoincrement: false,
                 unique_sets: vec![],
+                foreign_keys: vec![],
             };
             schema.add_btree_table(Arc::new(users_table));
 
@@ -2298,6 +2299,7 @@ mod tests {
                 is_strict: false,
                 has_autoincrement: false,
                 unique_sets: vec![],
+                foreign_keys: vec![],
             };
             schema.add_btree_table(Arc::new(products_table));
 
@@ -2363,6 +2365,7 @@ mod tests {
                 has_autoincrement: false,
                 is_strict: false,
                 unique_sets: vec![],
+                foreign_keys: vec![],
             };
             schema.add_btree_table(Arc::new(orders_table));
 
@@ -2401,6 +2404,7 @@ mod tests {
                 is_strict: false,
                 has_autoincrement: false,
                 unique_sets: vec![],
+                foreign_keys: vec![],
             };
             schema.add_btree_table(Arc::new(customers_table));
 
@@ -2463,6 +2467,7 @@ mod tests {
                 is_strict: false,
                 has_autoincrement: false,
                 unique_sets: vec![],
+                foreign_keys: vec![],
             };
             schema.add_btree_table(Arc::new(purchases_table));
 
@@ -2513,6 +2518,7 @@ mod tests {
                 is_strict: false,
                 has_autoincrement: false,
                 unique_sets: vec![],
+                foreign_keys: vec![],
             };
             schema.add_btree_table(Arc::new(vendors_table));
 
@@ -2550,6 +2556,7 @@ mod tests {
                 is_strict: false,
                 has_autoincrement: false,
                 unique_sets: vec![],
+                foreign_keys: vec![],
             };
             schema.add_btree_table(Arc::new(sales_table));
 

core/incremental/view.rs

@@ -1411,6 +1411,7 @@ mod tests {
             has_rowid: true,
             is_strict: false,
             unique_sets: vec![],
+            foreign_keys: vec![],
             has_autoincrement: false,
         };
 
@@ -1460,6 +1461,7 @@ mod tests {
             has_rowid: true,
             is_strict: false,
             has_autoincrement: false,
+            foreign_keys: vec![],
             unique_sets: vec![],
         };
 
@@ -1509,6 +1511,7 @@ mod tests {
             has_rowid: true,
             is_strict: false,
             has_autoincrement: false,
+            foreign_keys: vec![],
             unique_sets: vec![],
         };
 
@@ -1558,6 +1561,7 @@ mod tests {
             has_rowid: true, // Has implicit rowid but no alias
             is_strict: false,
             has_autoincrement: false,
+            foreign_keys: vec![],
             unique_sets: vec![],
         };
 

core/schema.rs

@@ -300,18 +300,9 @@ impl Schema {
         self.views.get(&name).cloned()
     }
 
-    pub fn add_btree_table(&mut self, mut table: Arc<BTreeTable>) -> Result<()> {
+    pub fn add_btree_table(&mut self, table: Arc<BTreeTable>) {
         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>) {
@@ -769,10 +760,7 @@ impl Schema {
                         }
                     }
 
-                    if let Some(mv_store) = mv_store {
+                    self.add_btree_table(Arc::new(table));
-                        mv_store.mark_table_as_loaded(root_page);
-                    }
-                    self.add_btree_table(Arc::new(table))?;
                 }
             }
             "index" => {
@@ -883,114 +871,6 @@ 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![];
@@ -2909,6 +2789,7 @@ mod tests {
                 hidden: false,
             }],
             unique_sets: vec![],
+            foreign_keys: vec![],
         };
 
         let result =

core/translate/insert.rs

@@ -83,11 +83,13 @@ pub fn translate_insert(
         );
     }
     let table_name = &tbl_name.name;
-    let has_child_fks = connection.foreign_keys_enabled()
+    let fk_enabled = connection.foreign_keys_enabled();
+    let has_child_fks = fk_enabled
         && !resolver
             .schema
             .get_foreign_keys_for_table(table_name.as_str())
             .is_empty();
+    let has_parent_fks = fk_enabled && resolver.schema.any_incoming_fk_to(table_name.as_str());
 
     // Check if this is a system table that should be protected from direct writes
     if crate::schema::is_system_table(table_name.as_str()) {
@@ -241,7 +243,7 @@ pub fn translate_insert(
         connection,
     )?;
 
-    if has_child_fks {
+    if has_child_fks || has_parent_fks {
         program.emit_insn(Insn::FkCounter {
             increment_value: 1,
             check_abort: false,
@@ -1042,7 +1044,7 @@ pub fn translate_insert(
             }
         }
     }
-    if has_child_fks {
+    if has_child_fks || has_parent_fks {
         emit_fk_checks_for_insert(&mut program, resolver, &insertion, table_name.as_str())?;
     }
 
@@ -1144,6 +1146,7 @@ pub fn translate_insert(
                 &mut result_columns,
                 cdc_table.as_ref().map(|c| c.0),
                 row_done_label,
+                connection,
             )?;
         } else {
             // UpsertDo::Nothing case

core/translate/logical.rs

@@ -2389,6 +2389,7 @@ mod tests {
             name: "users".to_string(),
             root_page: 2,
             primary_key_columns: vec![("id".to_string(), turso_parser::ast::SortOrder::Asc)],
+            foreign_keys: vec![],
             columns: vec![
                 SchemaColumn {
                     name: Some("id".to_string()),
@@ -2505,6 +2506,7 @@ mod tests {
             is_strict: false,
             has_autoincrement: false,
             unique_sets: vec![],
+            foreign_keys: vec![],
         };
         schema.add_btree_table(Arc::new(orders_table));
 
@@ -2567,6 +2569,7 @@ mod tests {
             is_strict: false,
             has_autoincrement: false,
             unique_sets: vec![],
+            foreign_keys: vec![],
         };
         schema.add_btree_table(Arc::new(products_table));
 

core/translate/optimizer/join.rs

@@ -1664,6 +1664,7 @@ mod tests {
             has_rowid: true,
             is_strict: false,
             unique_sets: vec![],
+            foreign_keys: vec![],
         })
     }
 

testing/all.test

@@ -47,3 +47,4 @@ source $testdir/vtab.test
 source $testdir/upsert.test
 source $testdir/window.test
 source $testdir/partial_idx.test
+source $testdir/foreign_keys.test