diff --git a/testing/trigger.test b/testing/trigger.test
new file mode 100755
index 000000000..714310849
--- /dev/null
+++ b/testing/trigger.test
@@ -0,0 +1,747 @@
+#!/usr/bin/env tclsh
+
+set testdir [file dirname $argv0]
+source $testdir/tester.tcl
+
+# Basic CREATE TRIGGER
+do_execsql_test_on_specific_db {:memory:} trigger-create-basic {
+  CREATE TABLE test1 (x INTEGER, y TEXT);
+  CREATE TRIGGER t1 BEFORE INSERT ON test1 BEGIN INSERT INTO test1 VALUES (100, 'triggered'); END;
+  INSERT INTO test1 VALUES (1, 'hello');
+  SELECT * FROM test1 ORDER BY rowid;
+} {100|triggered
+1|hello}
+
+# CREATE TRIGGER IF NOT EXISTS
+do_execsql_test_on_specific_db {:memory:} trigger-create-if-not-exists {
+  CREATE TABLE test2 (x INTEGER PRIMARY KEY);
+  CREATE TRIGGER IF NOT EXISTS t2 BEFORE INSERT ON test2 BEGIN SELECT 1; END;
+  CREATE TRIGGER IF NOT EXISTS t2 BEFORE INSERT ON test2 BEGIN SELECT 1; END;
+  SELECT name FROM sqlite_schema WHERE type='trigger' AND name='t2';
+} {t2}
+
+# DROP TRIGGER
+do_execsql_test_on_specific_db {:memory:} trigger-drop-basic {
+  CREATE TABLE test3 (x INTEGER PRIMARY KEY);
+  CREATE TRIGGER t3 BEFORE INSERT ON test3 BEGIN SELECT 1; END;
+  DROP TRIGGER t3;
+  SELECT name FROM sqlite_schema WHERE type='trigger' AND name='t3';
+} {}
+
+# DROP TRIGGER IF EXISTS
+do_execsql_test_on_specific_db {:memory:} trigger-drop-if-exists {
+  CREATE TABLE test4 (x INTEGER PRIMARY KEY);
+  DROP TRIGGER IF EXISTS nonexistent;
+  CREATE TRIGGER t4 BEFORE INSERT ON test4 BEGIN SELECT 1; END;
+  DROP TRIGGER IF EXISTS t4;
+  SELECT name FROM sqlite_schema WHERE type='trigger' AND name='t4';
+} {}
+
+# BEFORE INSERT trigger
+do_execsql_test_on_specific_db {:memory:} trigger-before-insert {
+  CREATE TABLE test5 (x INTEGER PRIMARY KEY, y TEXT);
+  CREATE TRIGGER t5 BEFORE INSERT ON test5 BEGIN UPDATE test5 SET y = 'before_' || y WHERE x = NEW.x; END;
+  INSERT INTO test5 VALUES (1, 'hello');
+  SELECT * FROM test5;
+} {1|hello}
+
+# AFTER INSERT trigger
+do_execsql_test_on_specific_db {:memory:} trigger-after-insert {
+  CREATE TABLE test6 (x INTEGER PRIMARY KEY, y TEXT);
+  CREATE TABLE log6 (x INTEGER, y TEXT);
+  CREATE TRIGGER t6 AFTER INSERT ON test6 BEGIN INSERT INTO log6 VALUES (NEW.x, NEW.y); END;
+  INSERT INTO test6 VALUES (1, 'hello');
+  SELECT * FROM log6;
+} {1|hello}
+
+# BEFORE UPDATE trigger
+do_execsql_test_on_specific_db {:memory:} trigger-before-update {
+  CREATE TABLE test7 (x INTEGER PRIMARY KEY, y TEXT);
+  INSERT INTO test7 VALUES (1, 'hello');
+  CREATE TRIGGER t7 BEFORE UPDATE ON test7 BEGIN UPDATE test7 SET y = 'before_' || NEW.y WHERE x = OLD.x; END;
+  UPDATE test7 SET y = 'world' WHERE x = 1;
+  SELECT * FROM test7;
+} {1|world}
+
+# AFTER UPDATE trigger
+do_execsql_test_on_specific_db {:memory:} trigger-after-update {
+  CREATE TABLE test8 (x INTEGER PRIMARY KEY, y TEXT);
+  CREATE TABLE log8 (old_x INTEGER, old_y TEXT, new_x INTEGER, new_y TEXT);
+  INSERT INTO test8 VALUES (1, 'hello');
+  CREATE TRIGGER t8 AFTER UPDATE ON test8 BEGIN INSERT INTO log8 VALUES (OLD.x, OLD.y, NEW.x, NEW.y); END;
+  UPDATE test8 SET y = 'world' WHERE x = 1;
+  SELECT * FROM log8;
+} {1|hello|1|world}
+
+# BEFORE DELETE trigger
+do_execsql_test_on_specific_db {:memory:} trigger-before-delete {
+  CREATE TABLE test9 (x INTEGER PRIMARY KEY, y TEXT);
+  CREATE TABLE log9 (x INTEGER, y TEXT);
+  INSERT INTO test9 VALUES (1, 'hello');
+  CREATE TRIGGER t9 BEFORE DELETE ON test9 BEGIN INSERT INTO log9 VALUES (OLD.x, OLD.y); END;
+  DELETE FROM test9 WHERE x = 1;
+  SELECT * FROM log9;
+} {1|hello}
+
+# AFTER DELETE trigger
+do_execsql_test_on_specific_db {:memory:} trigger-after-delete {
+  CREATE TABLE test10 (x INTEGER PRIMARY KEY, y TEXT);
+  CREATE TABLE log10 (x INTEGER, y TEXT);
+  INSERT INTO test10 VALUES (1, 'hello');
+  CREATE TRIGGER t10 AFTER DELETE ON test10 BEGIN INSERT INTO log10 VALUES (OLD.x, OLD.y); END;
+  DELETE FROM test10 WHERE x = 1;
+  SELECT * FROM log10;
+} {1|hello}
+
+# Trigger with WHEN clause
+do_execsql_test_on_specific_db {:memory:} trigger-when-clause {
+  CREATE TABLE test11 (x INTEGER PRIMARY KEY, y INTEGER);
+  CREATE TABLE log11 (x INTEGER);
+  CREATE TRIGGER t11 AFTER INSERT ON test11 WHEN NEW.y > 10 BEGIN INSERT INTO log11 VALUES (NEW.x); END;
+  INSERT INTO test11 VALUES (1, 5);
+  INSERT INTO test11 VALUES (2, 15);
+  SELECT * FROM log11;
+} {2}
+
+# Multiple triggers on same event
+do_execsql_test_on_specific_db {:memory:} trigger-multiple-same-event {
+  CREATE TABLE test12 (x INTEGER PRIMARY KEY);
+  CREATE TABLE log12 (msg TEXT);
+  CREATE TRIGGER t12a BEFORE INSERT ON test12 BEGIN INSERT INTO log12 VALUES ('trigger1'); END;
+  CREATE TRIGGER t12b BEFORE INSERT ON test12 BEGIN INSERT INTO log12 VALUES ('trigger2'); END;
+  INSERT INTO test12 VALUES (1);
+  SELECT * FROM log12 ORDER BY msg;
+} {trigger1
+trigger2}
+
+# Triggers dropped when table is dropped
+do_execsql_test_on_specific_db {:memory:} trigger-drop-table-drops-triggers {
+  CREATE TABLE test13 (x INTEGER PRIMARY KEY);
+  CREATE TRIGGER t13 BEFORE INSERT ON test13 BEGIN SELECT 1; END;
+  DROP TABLE test13;
+  SELECT name FROM sqlite_schema WHERE type='trigger' AND name='t13';
+} {}
+
+# NEW and OLD references
+do_execsql_test_on_specific_db {:memory:} trigger-new-old-references {
+  CREATE TABLE test14 (x INTEGER PRIMARY KEY, y TEXT);
+  CREATE TABLE log14 (msg TEXT);
+  INSERT INTO test14 VALUES (1, 'hello');
+  CREATE TRIGGER t14 AFTER UPDATE ON test14 BEGIN INSERT INTO log14 VALUES ('old=' || OLD.y || ' new=' || NEW.y); END;
+  UPDATE test14 SET y = 'world' WHERE x = 1;
+  SELECT * FROM log14;
+} {"old=hello new=world"}
+
+# Trigger with UPDATE OF clause
+do_execsql_test_on_specific_db {:memory:} trigger-update-of {
+  CREATE TABLE test15 (x INTEGER PRIMARY KEY, y TEXT, z TEXT);
+  CREATE TABLE log15 (msg TEXT);
+  INSERT INTO test15 VALUES (1, 'hello', 'world');
+  CREATE TRIGGER t15 AFTER UPDATE OF y ON test15 BEGIN INSERT INTO log15 VALUES ('y changed'); END;
+  UPDATE test15 SET z = 'foo' WHERE x = 1;
+  SELECT * FROM log15;
+  UPDATE test15 SET y = 'bar' WHERE x = 1;
+  SELECT * FROM log15;
+} {"y changed"}
+
+# Recursive trigger - AFTER INSERT inserting into same table
+do_execsql_test_on_specific_db {:memory:} trigger-recursive-after-insert {
+  CREATE TABLE test16 (x INTEGER PRIMARY KEY);
+  CREATE TRIGGER t16 AFTER INSERT ON test16 BEGIN INSERT INTO test16 VALUES (NEW.x + 1); END;
+  INSERT INTO test16 VALUES (1);
+  SELECT * FROM test16 ORDER BY x;
+} {1
+2}
+
+# Multiple UPDATE OF columns
+do_execsql_test_on_specific_db {:memory:} trigger-update-of-multiple {
+  CREATE TABLE test17 (x INTEGER PRIMARY KEY, y TEXT, z TEXT);
+  CREATE TABLE log17 (msg TEXT);
+  INSERT INTO test17 VALUES (1, 'a', 'b');
+  CREATE TRIGGER t17 AFTER UPDATE OF y, z ON test17 BEGIN INSERT INTO log17 VALUES ('updated'); END;
+  UPDATE test17 SET y = 'c' WHERE x = 1;
+  SELECT COUNT(*) FROM log17;
+  UPDATE test17 SET x = 2 WHERE x = 1;
+  SELECT COUNT(*) FROM log17;
+} {1
+1}
+
+# Complex WHEN clause with AND
+do_execsql_test_on_specific_db {:memory:} trigger-when-complex {
+  CREATE TABLE test18 (x INTEGER PRIMARY KEY, y INTEGER);
+  CREATE TABLE log18 (x INTEGER);
+  CREATE TRIGGER t18 BEFORE INSERT ON test18 WHEN NEW.y > 5 AND NEW.y < 10 BEGIN INSERT INTO log18 VALUES (NEW.x); END;
+  INSERT INTO test18 VALUES (1, 3);
+  INSERT INTO test18 VALUES (2, 7);
+  INSERT INTO test18 VALUES (3, 12);
+  SELECT * FROM log18;
+} {2}
+
+# Nested triggers - trigger firing another trigger
+do_execsql_test_on_specific_db {:memory:} trigger-nested {
+  CREATE TABLE test19 (x INTEGER PRIMARY KEY);
+  CREATE TABLE test20 (x INTEGER PRIMARY KEY);
+  CREATE TABLE log19 (msg TEXT);
+  CREATE TRIGGER t19 AFTER INSERT ON test19 BEGIN INSERT INTO test20 VALUES (NEW.x); END;
+  CREATE TRIGGER t20 AFTER INSERT ON test20 BEGIN INSERT INTO log19 VALUES ('t20 inserted: ' || NEW.x); END;
+  INSERT INTO test19 VALUES (1);
+  SELECT * FROM log19;
+} {"t20 inserted: 1"}
+
+# BEFORE INSERT inserting into same table (recursive)
+do_execsql_test_on_specific_db {:memory:} trigger-recursive-before-insert {
+  CREATE TABLE test21 (x INTEGER PRIMARY KEY);
+  CREATE TRIGGER t21 BEFORE INSERT ON test21 BEGIN INSERT INTO test21 VALUES (NEW.x + 100); END;
+  INSERT INTO test21 VALUES (1);
+  SELECT * FROM test21 ORDER BY x;
+} {1
+101}
+
+# AFTER UPDATE with WHEN clause and UPDATE OF
+do_execsql_test_on_specific_db {:memory:} trigger-update-of-when {
+  CREATE TABLE test22 (x INTEGER PRIMARY KEY, y TEXT);
+  CREATE TRIGGER t22 AFTER UPDATE OF y ON test22 WHEN OLD.y != NEW.y BEGIN UPDATE test22 SET y = UPPER(NEW.y) WHERE x = NEW.x; END;
+  INSERT INTO test22 VALUES (1, 'hello');
+  UPDATE test22 SET y = 'world' WHERE x = 1;
+  SELECT * FROM test22;
+} {1|WORLD}
+
+# Multiple statements in BEFORE INSERT trigger
+do_execsql_test_on_specific_db {:memory:} trigger-multiple-statements-before-insert {
+  CREATE TABLE test23 (x INTEGER PRIMARY KEY, y TEXT);
+  CREATE TABLE log23a (msg TEXT);
+  CREATE TABLE log23b (msg TEXT);
+  CREATE TRIGGER t23 BEFORE INSERT ON test23 BEGIN INSERT INTO log23a VALUES ('before1'); INSERT INTO log23b VALUES ('before2'); END;
+  INSERT INTO test23 VALUES (1, 'hello');
+  SELECT * FROM log23a;
+  SELECT * FROM log23b;
+} {before1
+before2}
+
+# Multiple statements in AFTER INSERT trigger
+do_execsql_test_on_specific_db {:memory:} trigger-multiple-statements-after-insert {
+  CREATE TABLE test24 (x INTEGER PRIMARY KEY, y TEXT);
+  CREATE TABLE log24a (msg TEXT);
+  CREATE TABLE log24b (msg TEXT);
+  CREATE TRIGGER t24 AFTER INSERT ON test24 BEGIN INSERT INTO log24a VALUES ('after1'); INSERT INTO log24b VALUES ('after2'); END;
+  INSERT INTO test24 VALUES (1, 'hello');
+  SELECT * FROM log24a;
+  SELECT * FROM log24b;
+} {after1
+after2}
+
+# Multiple statements in BEFORE UPDATE trigger
+do_execsql_test_on_specific_db {:memory:} trigger-multiple-statements-before-update {
+  CREATE TABLE test25 (x INTEGER PRIMARY KEY, y TEXT);
+  CREATE TABLE log25a (msg TEXT);
+  CREATE TABLE log25b (msg TEXT);
+  INSERT INTO test25 VALUES (1, 'hello');
+  CREATE TRIGGER t25 BEFORE UPDATE ON test25 BEGIN INSERT INTO log25a VALUES ('before_update1'); INSERT INTO log25b VALUES ('before_update2'); END;
+  UPDATE test25 SET y = 'world' WHERE x = 1;
+  SELECT * FROM log25a;
+  SELECT * FROM log25b;
+} {before_update1
+before_update2}
+
+# Multiple statements in AFTER UPDATE trigger
+do_execsql_test_on_specific_db {:memory:} trigger-multiple-statements-after-update {
+  CREATE TABLE test26 (x INTEGER PRIMARY KEY, y TEXT);
+  CREATE TABLE log26a (msg TEXT);
+  CREATE TABLE log26b (msg TEXT);
+  INSERT INTO test26 VALUES (1, 'hello');
+  CREATE TRIGGER t26 AFTER UPDATE ON test26 BEGIN INSERT INTO log26a VALUES ('after_update1'); INSERT INTO log26b VALUES ('after_update2'); END;
+  UPDATE test26 SET y = 'world' WHERE x = 1;
+  SELECT * FROM log26a;
+  SELECT * FROM log26b;
+} {after_update1
+after_update2}
+
+# Multiple statements in BEFORE DELETE trigger
+do_execsql_test_on_specific_db {:memory:} trigger-multiple-statements-before-delete {
+  CREATE TABLE test27 (x INTEGER PRIMARY KEY, y TEXT);
+  CREATE TABLE log27a (msg TEXT);
+  CREATE TABLE log27b (msg TEXT);
+  INSERT INTO test27 VALUES (1, 'hello');
+  CREATE TRIGGER t27 BEFORE DELETE ON test27 BEGIN INSERT INTO log27a VALUES ('before_delete1'); INSERT INTO log27b VALUES ('before_delete2'); END;
+  DELETE FROM test27 WHERE x = 1;
+  SELECT * FROM log27a;
+  SELECT * FROM log27b;
+} {before_delete1
+before_delete2}
+
+# Multiple statements in AFTER DELETE trigger
+do_execsql_test_on_specific_db {:memory:} trigger-multiple-statements-after-delete {
+  CREATE TABLE test28 (x INTEGER PRIMARY KEY, y TEXT);
+  CREATE TABLE log28a (msg TEXT);
+  CREATE TABLE log28b (msg TEXT);
+  INSERT INTO test28 VALUES (1, 'hello');
+  CREATE TRIGGER t28 AFTER DELETE ON test28 BEGIN INSERT INTO log28a VALUES ('after_delete1'); INSERT INTO log28b VALUES ('after_delete2'); END;
+  DELETE FROM test28 WHERE x = 1;
+  SELECT * FROM log28a;
+  SELECT * FROM log28b;
+} {after_delete1
+after_delete2}
+
+# Multiple statements with mixed operations in BEFORE INSERT trigger
+do_execsql_test_on_specific_db {:memory:} trigger-multiple-statements-mixed-before-insert {
+  CREATE TABLE test29 (x INTEGER PRIMARY KEY, y TEXT);
+  CREATE TABLE log29 (x INTEGER, y TEXT);
+  CREATE TABLE counter29 (cnt INTEGER);
+  INSERT INTO counter29 VALUES (0);
+  CREATE TRIGGER t29 BEFORE INSERT ON test29 BEGIN INSERT INTO log29 VALUES (NEW.x, NEW.y); UPDATE counter29 SET cnt = cnt + 1; END;
+  INSERT INTO test29 VALUES (1, 'hello');
+  INSERT INTO test29 VALUES (2, 'world');
+  SELECT * FROM log29 ORDER BY x;
+  SELECT * FROM counter29;
+} {1|hello
+2|world
+2}
+
+# Multiple statements with conditional logic in trigger
+do_execsql_test_on_specific_db {:memory:} trigger-multiple-statements-conditional {
+  CREATE TABLE test30 (x INTEGER PRIMARY KEY, y INTEGER);
+  CREATE TABLE log30 (msg TEXT);
+  CREATE TRIGGER t30 AFTER INSERT ON test30 BEGIN INSERT INTO log30 VALUES ('inserted: ' || NEW.x); INSERT INTO log30 VALUES ('value: ' || NEW.y); INSERT INTO log30 VALUES ('done'); END;
+  INSERT INTO test30 VALUES (1, 10);
+  INSERT INTO test30 VALUES (2, 20);
+  SELECT * FROM log30 ORDER BY msg;
+} {done
+done
+"inserted: 1"
+"inserted: 2"
+"value: 10"
+"value: 20"}
+
+# Trigger that updates the same row being updated (recursive update)
+do_execsql_test_on_specific_db {:memory:} trigger-update-same-row {
+  CREATE TABLE test31 (x INTEGER PRIMARY KEY, y INTEGER, z INTEGER);
+  INSERT INTO test31 VALUES (1, 10, 0);
+  CREATE TRIGGER t31 AFTER UPDATE OF y ON test31 BEGIN UPDATE test31 SET z = z + 1 WHERE x = NEW.x; END;
+  UPDATE test31 SET y = 20 WHERE x = 1;
+  SELECT * FROM test31;
+} {1|20|1}
+
+# Trigger that deletes the row being inserted
+do_execsql_test_on_specific_db {:memory:} trigger-delete-inserted-row {
+  CREATE TABLE test32 (x INTEGER PRIMARY KEY, y INTEGER);
+  CREATE TABLE log32 (msg TEXT);
+  CREATE TRIGGER t32 AFTER INSERT ON test32 BEGIN DELETE FROM test32 WHERE x = NEW.x AND NEW.y < 0; INSERT INTO log32 VALUES ('processed: ' || NEW.x); END;
+  INSERT INTO test32 VALUES (1, 10);
+  INSERT INTO test32 VALUES (2, -5);
+  SELECT * FROM test32;
+  SELECT * FROM log32 ORDER BY msg;
+} {1|10
+"processed: 1"
+"processed: 2"}
+
+# Trigger chain: INSERT triggers UPDATE which triggers DELETE
+do_execsql_test_on_specific_db {:memory:} trigger-chain-insert-update-delete {
+  CREATE TABLE test34 (x INTEGER PRIMARY KEY, y INTEGER);
+  CREATE TABLE log34 (msg TEXT);
+  CREATE TRIGGER t34a AFTER INSERT ON test34 BEGIN UPDATE test34 SET y = y + 1 WHERE x = NEW.x; END;
+  CREATE TRIGGER t34b AFTER UPDATE ON test34 BEGIN DELETE FROM test34 WHERE y > 100; INSERT INTO log34 VALUES ('updated: ' || NEW.x); END;
+  INSERT INTO test34 VALUES (1, 50);
+  INSERT INTO test34 VALUES (2, 100);
+  SELECT * FROM test34 ORDER BY x;
+  SELECT * FROM log34 ORDER BY msg;
+} {1|51
+"updated: 1"
+"updated: 2"}
+
+# Trigger that inserts into the same table (recursive insert)
+do_execsql_test_on_specific_db {:memory:} trigger-recursive-insert {
+  CREATE TABLE test35 (x INTEGER PRIMARY KEY, y INTEGER, depth INTEGER);
+  CREATE TRIGGER t35 AFTER INSERT ON test35 WHEN NEW.depth < 3 BEGIN INSERT INTO test35 VALUES (NEW.x * 10, NEW.y + 1, NEW.depth + 1); END;
+  INSERT INTO test35 VALUES (1, 0, 0);
+  SELECT * FROM test35 ORDER BY x;
+} {1|0|0
+10|1|1}
+
+# Trigger that updates OLD values in BEFORE UPDATE
+do_execsql_test_on_specific_db {:memory:} trigger-update-old-reference {
+  CREATE TABLE test36 (x INTEGER PRIMARY KEY, y INTEGER, z INTEGER);
+  CREATE TABLE log36 (old_y INTEGER, new_y INTEGER);
+  INSERT INTO test36 VALUES (1, 10, 0);
+  CREATE TRIGGER t36 BEFORE UPDATE OF y ON test36 BEGIN INSERT INTO log36 VALUES (OLD.y, NEW.y); UPDATE test36 SET z = OLD.y + NEW.y WHERE x = NEW.x; END;
+  UPDATE test36 SET y = 20 WHERE x = 1;
+  SELECT * FROM test36;
+  SELECT * FROM log36;
+} {1|20|30
+10|20}
+
+# Trigger that causes constraint violation in another table
+do_execsql_test_on_specific_db {:memory:} trigger-constraint-violation {
+  CREATE TABLE test37a (x INTEGER PRIMARY KEY);
+  CREATE TABLE test37b (x INTEGER PRIMARY KEY, y INTEGER);
+  INSERT INTO test37b VALUES (1, 100);
+  CREATE TRIGGER t37 AFTER INSERT ON test37a BEGIN INSERT OR IGNORE INTO test37b VALUES (NEW.x, NEW.x * 10); END;
+  INSERT INTO test37a VALUES (1);
+  SELECT * FROM test37b ORDER BY x;
+} {1|100}
+
+# Multiple triggers on same event with interdependencies
+do_execsql_test_on_specific_db {:memory:} trigger-multiple-interdependent {
+  CREATE TABLE test38 (x INTEGER PRIMARY KEY, y INTEGER, z INTEGER);
+  CREATE TRIGGER t38a AFTER INSERT ON test38 BEGIN UPDATE test38 SET y = 100 WHERE x = NEW.x; END;
+  CREATE TRIGGER t38b AFTER INSERT ON test38 BEGIN UPDATE test38 SET z = 200 WHERE x = NEW.x; END;
+  INSERT INTO test38 VALUES (1, 10, 20);
+  SELECT * FROM test38;
+} {1|100|200}
+
+# Trigger that references both OLD and NEW in UPDATE
+do_execsql_test_on_specific_db {:memory:} trigger-old-new-arithmetic {
+  CREATE TABLE test39 (x INTEGER PRIMARY KEY, y INTEGER, delta INTEGER);
+  INSERT INTO test39 VALUES (1, 10, 0);
+  CREATE TRIGGER t39 BEFORE UPDATE OF y ON test39 BEGIN UPDATE test39 SET delta = NEW.y - OLD.y WHERE x = NEW.x; END;
+  UPDATE test39 SET y = 25 WHERE x = 1;
+  SELECT * FROM test39;
+} {1|25|15}
+
+# Trigger that performs DELETE based on NEW values
+do_execsql_test_on_specific_db {:memory:} trigger-delete-on-insert-condition {
+  CREATE TABLE test40 (x INTEGER PRIMARY KEY, y INTEGER);
+  CREATE TABLE archive40 (x INTEGER, y INTEGER);
+  INSERT INTO test40 VALUES (1, 10);
+  INSERT INTO test40 VALUES (2, 20);
+  CREATE TRIGGER t40 AFTER INSERT ON test40 BEGIN DELETE FROM test40 WHERE y < NEW.y; INSERT INTO archive40 SELECT * FROM test40 WHERE x = NEW.x; END;
+  INSERT INTO test40 VALUES (3, 15);
+  SELECT * FROM test40 ORDER BY x;
+  SELECT * FROM archive40 ORDER BY x;
+} {2|20
+3|15
+3|15}
+
+# Trigger with WHEN clause that modifies the condition column
+do_execsql_test_on_specific_db {:memory:} trigger-when-clause-modification {
+  CREATE TABLE test41 (x INTEGER PRIMARY KEY, y INTEGER, processed INTEGER DEFAULT 0);
+  CREATE TRIGGER t41 AFTER INSERT ON test41 WHEN NEW.y > 50 BEGIN UPDATE test41 SET processed = 1 WHERE x = NEW.x; END;
+  INSERT INTO test41 (x, y) VALUES (1, 30);
+  INSERT INTO test41 (x, y) VALUES (2, 60);
+  INSERT INTO test41 (x, y) VALUES (3, 70);
+  SELECT * FROM test41 ORDER BY x;
+} {1|30|0
+2|60|1
+3|70|1}
+
+# Trigger that creates circular dependency between two tables
+do_execsql_test_on_specific_db {:memory:} trigger-circular-dependency {
+  CREATE TABLE test42a (x INTEGER PRIMARY KEY, y INTEGER);
+  CREATE TABLE test42b (x INTEGER PRIMARY KEY, y INTEGER);
+  CREATE TRIGGER t42a AFTER INSERT ON test42a BEGIN INSERT OR IGNORE INTO test42b VALUES (NEW.x, NEW.y + 1); END;
+  CREATE TRIGGER t42b AFTER INSERT ON test42b BEGIN INSERT OR IGNORE INTO test42a VALUES (NEW.x + 100, NEW.y + 1); END;
+  INSERT INTO test42a VALUES (1, 10);
+  SELECT * FROM test42a ORDER BY x;
+  SELECT * FROM test42b ORDER BY x;
+} {1|10
+101|12
+1|11}
+
+# BEFORE trigger modifying primary key value
+do_execsql_test_on_specific_db {:memory:} trigger-before-modify-primary-key {
+  CREATE TABLE test43 (x INTEGER PRIMARY KEY, y INTEGER);
+  CREATE TRIGGER t43 BEFORE INSERT ON test43 BEGIN UPDATE test43 SET x = NEW.x + 1000 WHERE x = NEW.x; END;
+  INSERT INTO test43 VALUES (1, 10);
+  SELECT * FROM test43;
+} {1|10}
+
+# UPDATE OF trigger modifying the same column it watches
+do_execsql_test_on_specific_db {:memory:} trigger-update-of-modify-same-column {
+  CREATE TABLE test44 (x INTEGER PRIMARY KEY, y INTEGER);
+  CREATE TABLE log44 (msg TEXT);
+  INSERT INTO test44 VALUES (1, 10);
+  CREATE TRIGGER t44 AFTER UPDATE OF y ON test44 BEGIN UPDATE test44 SET y = y * 2 WHERE x = NEW.x; INSERT INTO log44 VALUES ('triggered'); END;
+  UPDATE test44 SET y = 5 WHERE x = 1;
+  SELECT * FROM test44;
+  SELECT COUNT(*) FROM log44;
+} {1|10
+1}
+
+# BEFORE trigger modifying column used in UPDATE OF clause
+do_execsql_test_on_specific_db {:memory:} trigger-before-modify-update-of-column {
+  CREATE TABLE test45 (x INTEGER PRIMARY KEY, y INTEGER, z INTEGER);
+  CREATE TABLE log45 (msg TEXT);
+  INSERT INTO test45 VALUES (1, 10, 0);
+  CREATE TRIGGER t45a BEFORE UPDATE OF y ON test45 BEGIN UPDATE test45 SET y = y + 100 WHERE x = NEW.x; END;
+  CREATE TRIGGER t45b AFTER UPDATE OF y ON test45 BEGIN INSERT INTO log45 VALUES ('y=' || NEW.y); END;
+  UPDATE test45 SET y = 20 WHERE x = 1;
+  SELECT * FROM test45;
+  SELECT * FROM log45;
+} {1|20|0
+y=110
+y=20}
+
+# Trigger modifying column used in WHEN clause
+do_execsql_test_on_specific_db {:memory:} trigger-modify-when-clause-column {
+  CREATE TABLE test47 (x INTEGER PRIMARY KEY, y INTEGER, z INTEGER);
+  CREATE TABLE log47 (msg TEXT);
+  INSERT INTO test47 VALUES (1, 10, 0);
+  CREATE TRIGGER t47 BEFORE UPDATE ON test47 WHEN NEW.y > 5 BEGIN UPDATE test47 SET y = y - 10 WHERE x = NEW.x; INSERT INTO log47 VALUES ('modified'); END;
+  UPDATE test47 SET y = 15 WHERE x = 1;
+  SELECT * FROM test47;
+  SELECT COUNT(*) FROM log47;
+} {1|15|0
+1}
+
+# Trigger causing unique constraint violation with INSERT OR IGNORE
+do_execsql_test_on_specific_db {:memory:} trigger-unique-violation-ignore {
+  CREATE TABLE test48 (x INTEGER PRIMARY KEY, y INTEGER UNIQUE);
+  CREATE TABLE log48 (x INTEGER);
+  INSERT INTO test48 VALUES (1, 100);
+  CREATE TRIGGER t48 AFTER INSERT ON test48 BEGIN INSERT OR IGNORE INTO test48 VALUES (NEW.x + 1, NEW.y); INSERT INTO log48 VALUES (NEW.x); END;
+  INSERT INTO test48 VALUES (2, 200);
+  SELECT * FROM test48 ORDER BY x;
+  SELECT * FROM log48 ORDER BY x;
+} {1|100
+2|200
+2}
+
+# Multiple triggers modifying same column in different ways
+do_execsql_test_on_specific_db {:memory:} trigger-multiple-modify-same-column {
+  CREATE TABLE test49 (x INTEGER PRIMARY KEY, y INTEGER);
+  CREATE TRIGGER t49a BEFORE INSERT ON test49 BEGIN UPDATE test49 SET y = y + 1 WHERE x = NEW.x; END;
+  CREATE TRIGGER t49b BEFORE INSERT ON test49 BEGIN UPDATE test49 SET y = y * 2 WHERE x = NEW.x; END;
+  CREATE TRIGGER t49c BEFORE INSERT ON test49 BEGIN UPDATE test49 SET y = y + 10 WHERE x = NEW.x; END;
+  INSERT INTO test49 VALUES (1, 5);
+  SELECT * FROM test49;
+} {1|5}
+
+# BEFORE trigger modifying NEW values used in WHEN clause
+do_execsql_test_on_specific_db {:memory:} trigger-before-modify-new-in-when {
+  CREATE TABLE test51 (x INTEGER PRIMARY KEY, y INTEGER);
+  CREATE TABLE log51 (msg TEXT);
+  INSERT INTO test51 VALUES (1, 5);
+  CREATE TRIGGER t51a BEFORE UPDATE ON test51 BEGIN UPDATE test51 SET y = y + 20 WHERE x = NEW.x; END;
+  CREATE TRIGGER t51b AFTER UPDATE ON test51 WHEN NEW.y > 10 BEGIN INSERT INTO log51 VALUES ('y=' || NEW.y); END;
+  UPDATE test51 SET y = 8 WHERE x = 1;
+  SELECT * FROM test51;
+  SELECT * FROM log51;
+} {1|8
+y=25}
+
+# UPDATE OF on non-existent column (should be silently ignored)
+do_execsql_test_on_specific_db {:memory:} trigger-update-of-nonexistent-column {
+  CREATE TABLE test52 (x INTEGER PRIMARY KEY, y INTEGER);
+  CREATE TABLE log52 (msg TEXT);
+  INSERT INTO test52 VALUES (1, 10);
+  CREATE TRIGGER t52 AFTER UPDATE OF nonexistent ON test52 BEGIN INSERT INTO log52 VALUES ('triggered'); END;
+  UPDATE test52 SET y = 20 WHERE x = 1;
+  SELECT COUNT(*) FROM log52;
+} {0}
+
+# Trigger modifying same column multiple times in one trigger body
+do_execsql_test_on_specific_db {:memory:} trigger-modify-column-multiple-times {
+  CREATE TABLE test53 (x INTEGER PRIMARY KEY, y INTEGER);
+  INSERT INTO test53 VALUES (1, 10);
+  CREATE TRIGGER t53 AFTER UPDATE OF y ON test53 BEGIN UPDATE test53 SET y = y + 1 WHERE x = NEW.x; UPDATE test53 SET y = y * 2 WHERE x = NEW.x; UPDATE test53 SET y = y + 5 WHERE x = NEW.x; END;
+  UPDATE test53 SET y = 5 WHERE x = 1;
+  SELECT * FROM test53;
+} {1|17}
+
+# Trigger that modifies rowid indirectly through updates
+do_execsql_test_on_specific_db {:memory:} trigger-modify-rowid-indirect {
+  CREATE TABLE test55 (x INTEGER PRIMARY KEY, y INTEGER);
+  CREATE TABLE log55 (old_rowid INTEGER, new_rowid INTEGER);
+  INSERT INTO test55 VALUES (1, 10);
+  INSERT INTO test55 VALUES (2, 20);
+  CREATE TRIGGER t55 AFTER UPDATE ON test55 BEGIN INSERT INTO log55 VALUES (OLD.x, NEW.x); END;
+  UPDATE test55 SET x = 3 WHERE x = 1;
+  SELECT * FROM test55 ORDER BY x;
+  SELECT * FROM log55;
+} {2|20
+3|10
+1|3}
+
+# Trigger that references OLD.rowid and NEW.rowid with rowid alias column
+do_execsql_test_on_specific_db {:memory:} trigger-rowid-alias-old-new {
+  CREATE TABLE test55b (id INTEGER PRIMARY KEY, y INTEGER);
+  CREATE TABLE log55b (old_id INTEGER, new_id INTEGER, old_rowid INTEGER, new_rowid INTEGER);
+  INSERT INTO test55b VALUES (1, 10);
+  INSERT INTO test55b VALUES (2, 20);
+  CREATE TRIGGER t55b AFTER UPDATE ON test55b BEGIN INSERT INTO log55b VALUES (OLD.id, NEW.id, OLD.rowid, NEW.rowid); END;
+  UPDATE test55b SET id = 3 WHERE id = 1;
+  SELECT * FROM test55b ORDER BY id;
+  SELECT * FROM log55b;
+} {2|20
+3|10
+1|3|1|3}
+
+# Trigger with BEFORE UPDATE that modifies rowid alias in WHEN clause
+do_execsql_test_on_specific_db {:memory:} trigger-before-update-rowid-alias-when {
+  CREATE TABLE test55c (id INTEGER PRIMARY KEY, y INTEGER);
+  CREATE TABLE log55c (msg TEXT);
+  INSERT INTO test55c VALUES (1, 10);
+  CREATE TRIGGER t55c BEFORE UPDATE ON test55c WHEN NEW.id > 5 BEGIN UPDATE test55c SET y = y + 100 WHERE id = NEW.id; END;
+  UPDATE test55c SET id = 10, y = 20 WHERE id = 1;
+  SELECT * FROM test55c;
+  SELECT COUNT(*) FROM log55c;
+} {10|20
+0}
+
+# Trigger referencing OLD.rowid in DELETE with rowid alias column
+do_execsql_test_on_specific_db {:memory:} trigger-delete-old-rowid-alias {
+  CREATE TABLE test55d (pk INTEGER PRIMARY KEY, y INTEGER);
+  CREATE TABLE log55d (deleted_pk INTEGER, deleted_rowid INTEGER);
+  INSERT INTO test55d VALUES (1, 10);
+  INSERT INTO test55d VALUES (2, 20);
+  CREATE TRIGGER t55d AFTER DELETE ON test55d BEGIN INSERT INTO log55d VALUES (OLD.pk, OLD.rowid); END;
+  DELETE FROM test55d WHERE pk = 1;
+  SELECT * FROM test55d;
+  SELECT * FROM log55d;
+} {2|20
+1|1}
+
+# Trigger with NEW.rowid in INSERT with rowid alias
+do_execsql_test_on_specific_db {:memory:} trigger-insert-new-rowid-alias {
+  CREATE TABLE test55e (myid INTEGER PRIMARY KEY, y INTEGER);
+  CREATE TABLE log55e (new_myid INTEGER, new_rowid INTEGER);
+  CREATE TRIGGER t55e AFTER INSERT ON test55e BEGIN INSERT INTO log55e VALUES (NEW.myid, NEW.rowid); END;
+  INSERT INTO test55e VALUES (5, 50);
+  INSERT INTO test55e VALUES (10, 100);
+  SELECT * FROM test55e ORDER BY myid;
+  SELECT * FROM log55e ORDER BY new_myid;
+} {5|50
+10|100
+5|5
+10|10}
+
+# Trigger modifying rowid through UPDATE with complex WHERE using rowid alias
+do_execsql_test_on_specific_db {:memory:} trigger-update-rowid-complex-where {
+  CREATE TABLE test55f (rid INTEGER PRIMARY KEY, y INTEGER);
+  CREATE TABLE log55f (old_rid INTEGER, new_rid INTEGER);
+  INSERT INTO test55f VALUES (1, 10);
+  INSERT INTO test55f VALUES (2, 20);
+  INSERT INTO test55f VALUES (3, 30);
+  CREATE TRIGGER t55f AFTER UPDATE ON test55f BEGIN INSERT INTO log55f VALUES (OLD.rid, NEW.rid); UPDATE test55f SET y = y + 1 WHERE rid = NEW.rid; END;
+  UPDATE test55f SET rid = 100 WHERE rid = 2;
+  SELECT * FROM test55f ORDER BY rid;
+  SELECT * FROM log55f;
+} {1|10
+3|30
+100|21
+2|100}
+
+# Trigger using both rowid and rowid alias in same expression
+do_execsql_test_on_specific_db {:memory:} trigger-rowid-and-alias-mixed {
+  CREATE TABLE test55g (id INTEGER PRIMARY KEY, y INTEGER);
+  CREATE TABLE log55g (id_val INTEGER, rowid_val INTEGER, match INTEGER);
+  INSERT INTO test55g VALUES (1, 10);
+  CREATE TRIGGER t55g AFTER INSERT ON test55g BEGIN INSERT INTO log55g VALUES (NEW.id, NEW.rowid, NEW.id = NEW.rowid); END;
+  INSERT INTO test55g VALUES (5, 50);
+  SELECT * FROM log55g;
+} {5|5|1}
+
+# Trigger that causes cascading updates through multiple tables
+do_execsql_test_on_specific_db {:memory:} trigger-cascading-updates {
+  CREATE TABLE test57a (x INTEGER PRIMARY KEY, y INTEGER);
+  CREATE TABLE test57b (x INTEGER PRIMARY KEY, y INTEGER);
+  CREATE TABLE log57 (msg TEXT);
+  INSERT INTO test57a VALUES (1, 10);
+  INSERT INTO test57b VALUES (1, 100);
+  CREATE TRIGGER t57a AFTER UPDATE ON test57a BEGIN UPDATE test57b SET y = NEW.y * 10 WHERE x = NEW.x; END;
+  CREATE TRIGGER t57b AFTER UPDATE ON test57b BEGIN INSERT INTO log57 VALUES ('b updated: ' || NEW.y); END;
+  UPDATE test57a SET y = 20 WHERE x = 1;
+  SELECT * FROM test57a;
+  SELECT * FROM test57b;
+  SELECT * FROM log57;
+} {1|20
+1|200
+"b updated: 200"}
+
+# Trigger modifying columns used in unique constraint
+do_execsql_test_on_specific_db {:memory:} trigger-modify-unique-column {
+  CREATE TABLE test58 (x INTEGER PRIMARY KEY, y INTEGER UNIQUE);
+  CREATE TABLE log58 (msg TEXT);
+  INSERT INTO test58 VALUES (1, 10);
+  INSERT INTO test58 VALUES (2, 20);
+  CREATE TRIGGER t58 AFTER UPDATE ON test58 BEGIN UPDATE test58 SET y = y + 1 WHERE x != NEW.x; INSERT INTO log58 VALUES ('updated'); END;
+  UPDATE test58 SET y = 15 WHERE x = 1;
+  SELECT * FROM test58 ORDER BY x;
+  SELECT COUNT(*) FROM log58;
+} {1|15
+2|21
+1}
+
+# BEFORE INSERT trigger that modifies NEW.x before insertion
+do_execsql_test_on_specific_db {:memory:} trigger-before-insert-modify-new {
+  CREATE TABLE test59 (x INTEGER PRIMARY KEY, y INTEGER);
+  CREATE TRIGGER t59 BEFORE INSERT ON test59 BEGIN UPDATE test59 SET x = NEW.x + 100 WHERE x = NEW.x; END;
+  INSERT INTO test59 VALUES (1, 10);
+  SELECT * FROM test59;
+} {1|10}
+
+# Trigger with UPDATE OF multiple columns, modifying one of them
+do_execsql_test_on_specific_db {:memory:} trigger-update-of-multiple-modify-one {
+  CREATE TABLE test60 (x INTEGER PRIMARY KEY, y INTEGER, z INTEGER);
+  CREATE TABLE log60 (msg TEXT);
+  INSERT INTO test60 VALUES (1, 10, 100);
+  CREATE TRIGGER t60 AFTER UPDATE OF y, z ON test60 BEGIN UPDATE test60 SET y = y + 50 WHERE x = NEW.x; INSERT INTO log60 VALUES ('triggered'); END;
+  UPDATE test60 SET y = 20 WHERE x = 1;
+  SELECT * FROM test60;
+  SELECT COUNT(*) FROM log60;
+  UPDATE test60 SET z = 200 WHERE x = 1;
+  SELECT * FROM test60;
+  SELECT COUNT(*) FROM log60;
+} {1|70|100
+1
+1|120|200
+2}
+
+# Trigger modifying column that's part of composite unique constraint
+do_execsql_test_on_specific_db {:memory:} trigger-modify-composite-unique {
+  CREATE TABLE test62 (x INTEGER, y INTEGER, UNIQUE(x, y));
+  CREATE TABLE log62 (msg TEXT);
+  INSERT INTO test62 VALUES (1, 10);
+  INSERT INTO test62 VALUES (2, 20);
+  CREATE TRIGGER t62 AFTER UPDATE ON test62 BEGIN UPDATE test62 SET y = y + 1 WHERE x = NEW.x + 1; INSERT INTO log62 VALUES ('updated'); END;
+  UPDATE test62 SET y = 15 WHERE x = 1;
+  SELECT * FROM test62 ORDER BY x;
+  SELECT COUNT(*) FROM log62;
+} {1|15
+2|21
+1}
+
+# Trigger with WHEN clause that becomes false after BEFORE trigger modification
+do_execsql_test_on_specific_db {:memory:} trigger-when-becomes-false-after-before {
+  CREATE TABLE test63 (x INTEGER PRIMARY KEY, y INTEGER);
+  CREATE TABLE log63 (msg TEXT);
+  INSERT INTO test63 VALUES (1, 10);
+  CREATE TRIGGER t63a BEFORE UPDATE ON test63 BEGIN UPDATE test63 SET y = y - 15 WHERE x = NEW.x; END;
+  CREATE TRIGGER t63b AFTER UPDATE ON test63 WHEN NEW.y > 0 BEGIN INSERT INTO log63 VALUES ('y=' || NEW.y); END;
+  UPDATE test63 SET y = 20 WHERE x = 1;
+  SELECT * FROM test63;
+  SELECT COUNT(*) FROM log63;
+} {1|20
+1}
+
+# Trigger that inserts into same table with different rowid
+do_execsql_test_on_specific_db {:memory:} trigger-recursive-different-rowid {
+  CREATE TABLE test64 (x INTEGER PRIMARY KEY, y INTEGER);
+  CREATE TRIGGER t64 AFTER INSERT ON test64 WHEN NEW.y < 100 BEGIN INSERT INTO test64 VALUES (NEW.x + 1000, NEW.y + 10); END;
+  INSERT INTO test64 VALUES (1, 5);
+  INSERT INTO test64 VALUES (2, 50);
+  SELECT * FROM test64 ORDER BY x;
+} {1|5
+2|50
+1001|15
+1002|60}
+
+# Trigger modifying primary key through UPDATE in BEFORE trigger
+do_execsql_test_on_specific_db {:memory:} trigger-before-update-primary-key {
+  CREATE TABLE test65 (x INTEGER PRIMARY KEY, y INTEGER);
+  INSERT INTO test65 VALUES (1, 10);
+  CREATE TRIGGER t65 BEFORE UPDATE ON test65 BEGIN UPDATE test65 SET x = NEW.x + 100 WHERE x = OLD.x; END;
+  UPDATE test65 SET y = 20 WHERE x = 1;
+  SELECT * FROM test65 ORDER BY x;
+} {101|10}
+
+# Trigger modifying row during BEFORE UPDATE - parent expression behavior
+# The parent UPDATE's SET clause expressions (c0 = c1+1) use OLD values for columns
+# referenced in the expression (c1), but the final row gets the modified values from
+# the BEFORE trigger for columns not in the parent's SET clause (c1, c2).
+do_execsql_test_on_specific_db {:memory:} trigger-before-update-parent-expression-old-values {
+  CREATE TABLE test66 (c0 INTEGER, c1 INTEGER, c2 INTEGER);
+  CREATE TRIGGER tu66 BEFORE UPDATE ON test66 BEGIN UPDATE test66 SET c1=666, c2=666; END;
+  INSERT INTO test66 VALUES (1,1,1);
+  UPDATE test66 SET c0 = c1+1;
+  SELECT * FROM test66;
+} {2|666|666}
diff --git a/tests/fuzz/mod.rs b/tests/fuzz/mod.rs
index d116c054c..95668f00c 100644
--- a/tests/fuzz/mod.rs
+++ b/tests/fuzz/mod.rs
@@ -651,16 +651,21 @@ mod fuzz_tests {
 
     #[test]
     #[allow(unused_assignments)]
-    pub fn fk_deferred_constraints_fuzz() {
+    pub fn fk_deferred_constraints_and_triggers_fuzz() {
+        let _ = tracing_subscriber::fmt::try_init();
         let _ = env_logger::try_init();
         let (mut rng, seed) = rng_from_time_or_env();
-        println!("fk_deferred_constraints_fuzz seed: {seed}");
+        println!("fk_deferred_constraints_and_triggers_fuzz seed: {seed}");
 
         const OUTER_ITERS: usize = 10;
         const INNER_ITERS: usize = 100;
 
         for outer in 0..OUTER_ITERS {
-            println!("fk_deferred_constraints_fuzz {}/{}", outer + 1, OUTER_ITERS);
+            println!(
+                "fk_deferred_constraints_and_triggers_fuzz {}/{}",
+                outer + 1,
+                OUTER_ITERS
+            );
 
             let limbo_db = TempDatabase::new_empty();
             let sqlite_db = TempDatabase::new_empty();
@@ -757,6 +762,99 @@ mod fuzz_tests {
                 }
             }
 
+            // Add triggers on every outer iteration (max 2 triggers)
+            // Create a log table for trigger operations
+            let s = log_and_exec(
+                "CREATE TABLE trigger_log(action TEXT, table_name TEXT, id_val INT, extra_val INT)",
+            );
+            limbo_exec_rows(&limbo_db, &limbo, &s);
+            sqlite.execute(&s, params![]).unwrap();
+
+            // Create a stats table for tracking operations
+            let s = log_and_exec("CREATE TABLE trigger_stats(op_type TEXT PRIMARY KEY, count INT)");
+            limbo_exec_rows(&limbo_db, &limbo, &s);
+            sqlite.execute(&s, params![]).unwrap();
+
+            // Define all available trigger types
+            let trigger_definitions: Vec<&str> = vec![
+                // BEFORE INSERT trigger on parent - logs and potentially creates a child
+                "CREATE TRIGGER trig_parent_before_insert BEFORE INSERT ON parent BEGIN
+                 INSERT INTO trigger_log VALUES ('BEFORE_INSERT', 'parent', NEW.id, NEW.a);
+                 INSERT INTO trigger_stats VALUES ('parent_insert', 1) ON CONFLICT(op_type) DO UPDATE SET count=count+1;
+                 -- Sometimes create a deferred child referencing this parent
+                 INSERT INTO child_deferred VALUES (NEW.id + 10000, NEW.id, NEW.a);
+                END",
+                // AFTER INSERT trigger on child_deferred - logs and updates parent
+                "CREATE TRIGGER trig_child_deferred_after_insert AFTER INSERT ON child_deferred BEGIN
+                 INSERT INTO trigger_log VALUES ('AFTER_INSERT', 'child_deferred', NEW.id, NEW.pid);
+                 INSERT INTO trigger_stats VALUES ('child_deferred_insert', 1) ON CONFLICT(op_type) DO UPDATE SET count=count+1;
+                 -- Update parent's 'a' column if parent exists
+                 UPDATE parent SET a = a + 1 WHERE id = NEW.pid;
+                END",
+                // BEFORE UPDATE OF 'a' on parent - logs and modifies the update
+                "CREATE TRIGGER trig_parent_before_update_a BEFORE UPDATE OF a ON parent BEGIN
+                 INSERT INTO trigger_log VALUES ('BEFORE_UPDATE_A', 'parent', OLD.id, OLD.a);
+                 INSERT INTO trigger_stats VALUES ('parent_update_a', 1) ON CONFLICT(op_type) DO UPDATE SET count=count+1;
+                 -- Also update 'b' column when 'a' is updated
+                 UPDATE parent SET b = NEW.a * 2 WHERE id = NEW.id;
+                END",
+                // AFTER UPDATE OF 'pid' on child_deferred - logs and creates/updates related records
+                "CREATE TRIGGER trig_child_deferred_after_update_pid AFTER UPDATE OF pid ON child_deferred BEGIN
+                 INSERT INTO trigger_log VALUES ('AFTER_UPDATE_PID', 'child_deferred', NEW.id, NEW.pid);
+                 INSERT INTO trigger_stats VALUES ('child_deferred_update_pid', 1) ON CONFLICT(op_type) DO UPDATE SET count=count+1;
+                 -- Create a child_immediate referencing the new parent
+                 INSERT INTO child_immediate VALUES (NEW.id + 20000, NEW.pid, NEW.x);
+                 -- Update parent's 'b' column
+                 UPDATE parent SET b = b + 1 WHERE id = NEW.pid;
+                END",
+                // BEFORE DELETE on parent - logs and cascades to children
+                "CREATE TRIGGER trig_parent_before_delete BEFORE DELETE ON parent BEGIN
+                 INSERT INTO trigger_log VALUES ('BEFORE_DELETE', 'parent', OLD.id, OLD.a);
+                 INSERT INTO trigger_stats VALUES ('parent_delete', 1) ON CONFLICT(op_type) DO UPDATE SET count=count+1;
+                 -- Delete all children that reference the deleted parent
+                 DELETE FROM child_deferred WHERE pid = OLD.id;
+                END",
+                // AFTER DELETE on child_deferred - logs and updates parent stats
+                "CREATE TRIGGER trig_child_deferred_after_delete AFTER DELETE ON child_deferred BEGIN
+                 INSERT INTO trigger_log VALUES ('AFTER_DELETE', 'child_deferred', OLD.id, OLD.pid);
+                 INSERT INTO trigger_stats VALUES ('child_deferred_delete', 1) ON CONFLICT(op_type) DO UPDATE SET count=count+1;
+                 -- Update parent's 'a' column
+                 UPDATE parent SET a = a - 1 WHERE id = OLD.pid;
+                END",
+                // BEFORE INSERT on child_immediate - logs, creates parent if needed, updates stats
+                "CREATE TRIGGER trig_child_immediate_before_insert BEFORE INSERT ON child_immediate BEGIN
+                 INSERT INTO trigger_log VALUES ('BEFORE_INSERT', 'child_immediate', NEW.id, NEW.pid);
+                 INSERT INTO trigger_stats VALUES ('child_immediate_insert', 1) ON CONFLICT(op_type) DO UPDATE SET count=count+1;
+                 -- Create parent if it doesn't exist (with a default value)
+                 INSERT OR IGNORE INTO parent VALUES (NEW.pid, NEW.y, NEW.y * 2);
+                 -- Update parent's 'a' column
+                 UPDATE parent SET a = a + NEW.y WHERE id = NEW.pid;
+                END",
+                // AFTER UPDATE OF 'y' on child_immediate - logs and cascades updates
+                "CREATE TRIGGER trig_child_immediate_after_update_y AFTER UPDATE OF y ON child_immediate BEGIN
+                 INSERT INTO trigger_log VALUES ('AFTER_UPDATE_Y', 'child_immediate', NEW.id, NEW.y);
+                 INSERT INTO trigger_stats VALUES ('child_immediate_update_y', 1) ON CONFLICT(op_type) DO UPDATE SET count=count+1;
+                 -- Update parent's 'a' based on the change
+                 UPDATE parent SET a = a + (NEW.y - OLD.y) WHERE id = NEW.pid;
+                 -- Also create a deferred child referencing the same parent
+                 INSERT INTO child_deferred VALUES (NEW.id + 30000, NEW.pid, NEW.y);
+                END",
+            ];
+
+            // Randomly select up to 2 triggers from the list
+            let num_triggers = rng.random_range(1..=2);
+            let mut selected_indices = std::collections::HashSet::new();
+            while selected_indices.len() < num_triggers {
+                selected_indices.insert(rng.random_range(0..trigger_definitions.len()));
+            }
+
+            // Create the selected triggers
+            for &idx in selected_indices.iter() {
+                let s = log_and_exec(trigger_definitions[idx]);
+                limbo_exec_rows(&limbo_db, &limbo, &s);
+                sqlite.execute(&s, params![]).unwrap();
+            }
+
             // Transaction-based mutations with mix of deferred and immediate operations
             let mut in_tx = false;
             for tx_num in 0..INNER_ITERS {
@@ -1856,16 +1954,128 @@ mod fuzz_tests {
             }
         }
     }
-
     #[test]
     /// Create a table with a random number of columns and indexes, and then randomly update or delete rows from the table.
     /// Verify that the results are the same for SQLite and Turso.
     pub fn table_index_mutation_fuzz() {
+        /// Format a nice diff between two result sets for better error messages
+        #[allow(clippy::too_many_arguments)]
+        fn format_rows_diff(
+            sqlite_rows: &[Vec<Value>],
+            limbo_rows: &[Vec<Value>],
+            seed: u64,
+            query: &str,
+            table_def: &str,
+            indexes: &[String],
+            trigger: Option<&String>,
+            dml_statements: &[String],
+        ) -> String {
+            let mut diff = String::new();
+            let sqlite_rows_len = sqlite_rows.len();
+            let limbo_rows_len = limbo_rows.len();
+            diff.push_str(&format!(
+            "\n\n=== Row Count Difference ===\nSQLite: {sqlite_rows_len} rows, Limbo: {limbo_rows_len} rows\n",
+        ));
+
+            // Find rows that differ at the same index
+            let max_len = sqlite_rows.len().max(limbo_rows.len());
+            let mut diff_indices = Vec::new();
+            for i in 0..max_len {
+                let sqlite_row = sqlite_rows.get(i);
+                let limbo_row = limbo_rows.get(i);
+                if sqlite_row != limbo_row {
+                    diff_indices.push(i);
+                }
+            }
+
+            if !diff_indices.is_empty() {
+                diff.push_str("\n=== Rows Differing at Same Index (showing first 10) ===\n");
+                for &idx in diff_indices.iter().take(10) {
+                    diff.push_str(&format!("\nIndex {idx}:\n"));
+                    if let Some(sqlite_row) = sqlite_rows.get(idx) {
+                        diff.push_str(&format!("  SQLite: {sqlite_row:?}\n"));
+                    } else {
+                        diff.push_str("  SQLite: <missing>\n");
+                    }
+                    if let Some(limbo_row) = limbo_rows.get(idx) {
+                        diff.push_str(&format!("  Limbo:  {limbo_row:?}\n"));
+                    } else {
+                        diff.push_str("  Limbo:  <missing>\n");
+                    }
+                }
+                if diff_indices.len() > 10 {
+                    diff.push_str(&format!(
+                        "\n... and {} more differences\n",
+                        diff_indices.len() - 10
+                    ));
+                }
+            }
+
+            // Find rows that are in one but not the other (using linear search since Value doesn't implement Hash)
+            let mut only_in_sqlite = Vec::new();
+            for sqlite_row in sqlite_rows.iter() {
+                if !limbo_rows.iter().any(|limbo_row| limbo_row == sqlite_row) {
+                    only_in_sqlite.push(sqlite_row);
+                }
+            }
+
+            let mut only_in_limbo = Vec::new();
+            for limbo_row in limbo_rows.iter() {
+                if !sqlite_rows.iter().any(|sqlite_row| sqlite_row == limbo_row) {
+                    only_in_limbo.push(limbo_row);
+                }
+            }
+
+            if !only_in_sqlite.is_empty() {
+                diff.push_str("\n=== Rows Only in SQLite (showing first 10) ===\n");
+                for row in only_in_sqlite.iter().take(10) {
+                    diff.push_str(&format!("  {row:?}\n"));
+                }
+                if only_in_sqlite.len() > 10 {
+                    diff.push_str(&format!(
+                        "\n... and {} more rows\n",
+                        only_in_sqlite.len() - 10
+                    ));
+                }
+            }
+
+            if !only_in_limbo.is_empty() {
+                diff.push_str("\n=== Rows Only in Limbo (showing first 10) ===\n");
+                for row in only_in_limbo.iter().take(10) {
+                    diff.push_str(&format!("  {row:?}\n"));
+                }
+                if only_in_limbo.len() > 10 {
+                    diff.push_str(&format!(
+                        "\n... and {} more rows\n",
+                        only_in_limbo.len() - 10
+                    ));
+                }
+            }
+
+            diff.push_str(&format!(
+                "\n=== Context ===\nSeed: {seed}\nQuery: {query}\n",
+            ));
+
+            diff.push_str("\n=== DDL/DML to Reproduce ===\n");
+            diff.push_str(&format!("{table_def};\n"));
+            for idx in indexes.iter() {
+                diff.push_str(&format!("{idx};\n"));
+            }
+            if let Some(trigger) = trigger {
+                diff.push_str(&format!("{trigger};\n"));
+            }
+            for dml in dml_statements.iter() {
+                diff.push_str(&format!("{dml};\n"));
+            }
+
+            diff
+        }
+
         let _ = env_logger::try_init();
         let (mut rng, seed) = rng_from_time_or_env();
         println!("table_index_mutation_fuzz seed: {seed}");
 
-        const OUTER_ITERATIONS: usize = 100;
+        const OUTER_ITERATIONS: usize = 30;
         for i in 0..OUTER_ITERATIONS {
             println!(
                 "table_index_mutation_fuzz iteration {}/{}",
@@ -1926,8 +2136,16 @@ mod fuzz_tests {
                 sqlite_conn.execute(t, params![]).unwrap();
             }
 
+            let use_trigger = rng.random_bool(1.0);
+
             // Generate initial data
-            let num_inserts = rng.random_range(10..=1000);
+            // Triggers can cause quadratic complexity to the tested operations so limit total row count
+            // whenever we have one to make the test runtime reasonable.
+            let num_inserts = if use_trigger {
+                rng.random_range(10..=100)
+            } else {
+                rng.random_range(10..=1000)
+            };
             let mut tuples = HashSet::new();
             while tuples.len() < num_inserts {
                 tuples.insert(
@@ -1953,13 +2171,15 @@ mod fuzz_tests {
                 .map(|i| format!("c{i}"))
                 .collect::<Vec<_>>()
                 .join(", ");
+            let insert_type = match rng.random_range(0..3) {
+                0 => "",
+                1 => "OR REPLACE",
+                2 => "OR IGNORE",
+                _ => unreachable!(),
+            };
             let insert = format!(
                 "INSERT {} INTO t ({}) VALUES {}",
-                if rng.random_bool(0.4) {
-                    "OR IGNORE"
-                } else {
-                    ""
-                },
+                insert_type,
                 col_names,
                 insert_values.join(", ")
             );
@@ -1969,6 +2189,145 @@ mod fuzz_tests {
             sqlite_conn.execute(&insert, params![]).unwrap();
             limbo_exec_rows(&limbo_db, &limbo_conn, &insert);
 
+            // Self-affecting triggers (e.g CREATE TRIGGER t BEFORE DELETE ON t BEGIN UPDATE t ... END) are
+            // an easy source of bugs, so create one some of the time.
+            let trigger = if use_trigger {
+                // Create a random trigger
+                let trigger_time = if rng.random_bool(0.5) {
+                    "BEFORE"
+                } else {
+                    "AFTER"
+                };
+                let trigger_event = match rng.random_range(0..3) {
+                    0 => "INSERT".to_string(),
+                    1 => {
+                        // Optionally specify columns for UPDATE trigger
+                        if rng.random_bool(0.5) {
+                            let update_col = rng.random_range(0..num_cols);
+                            format!("UPDATE OF c{update_col}")
+                        } else {
+                            "UPDATE".to_string()
+                        }
+                    }
+                    2 => "DELETE".to_string(),
+                    _ => unreachable!(),
+                };
+
+                // Determine if OLD/NEW references are available based on trigger event
+                let has_old =
+                    trigger_event.starts_with("UPDATE") || trigger_event.starts_with("DELETE");
+                let has_new =
+                    trigger_event.starts_with("UPDATE") || trigger_event.starts_with("INSERT");
+
+                // Generate trigger action (INSERT, UPDATE, or DELETE)
+                let trigger_action = match rng.random_range(0..3) {
+                    0 => {
+                        // INSERT action
+                        let values = (0..num_cols)
+                            .map(|i| {
+                                // Randomly use OLD/NEW values if available
+                                if has_old && rng.random_bool(0.3) {
+                                    format!("OLD.c{i}")
+                                } else if has_new && rng.random_bool(0.3) {
+                                    format!("NEW.c{i}")
+                                } else {
+                                    rng.random_range(0..1000).to_string()
+                                }
+                            })
+                            .collect::<Vec<_>>()
+                            .join(", ");
+                        let insert_conflict_action = match rng.random_range(0..3) {
+                            0 => "",
+                            1 => " OR REPLACE",
+                            2 => " OR IGNORE",
+                            _ => unreachable!(),
+                        };
+                        format!(
+                            "INSERT{insert_conflict_action} INTO t ({col_names}) VALUES ({values})"
+                        )
+                    }
+                    1 => {
+                        // UPDATE action
+                        let update_col = rng.random_range(0..num_cols);
+                        let new_value = if has_old && rng.random_bool(0.3) {
+                            let ref_col = rng.random_range(0..num_cols);
+                            // Sometimes make it a function of the OLD column
+                            if rng.random_bool(0.5) {
+                                let operator = *["+", "-", "*"].choose(&mut rng).unwrap();
+                                let amount = rng.random_range(1..100);
+                                format!("OLD.c{ref_col} {operator} {amount}")
+                            } else {
+                                format!("OLD.c{ref_col}")
+                            }
+                        } else if has_new && rng.random_bool(0.3) {
+                            let ref_col = rng.random_range(0..num_cols);
+                            // Sometimes make it a function of the NEW column
+                            if rng.random_bool(0.5) {
+                                let operator = *["+", "-", "*"].choose(&mut rng).unwrap();
+                                let amount = rng.random_range(1..100);
+                                format!("NEW.c{ref_col} {operator} {amount}")
+                            } else {
+                                format!("NEW.c{ref_col}")
+                            }
+                        } else {
+                            rng.random_range(0..1000).to_string()
+                        };
+                        let op = match rng.random_range(0..=3) {
+                            0 => "<",
+                            1 => "<=",
+                            2 => ">",
+                            3 => ">=",
+                            _ => unreachable!(),
+                        };
+                        let threshold = if has_old && rng.random_bool(0.3) {
+                            let ref_col = rng.random_range(0..num_cols);
+                            format!("OLD.c{ref_col}")
+                        } else if has_new && rng.random_bool(0.3) {
+                            let ref_col = rng.random_range(0..num_cols);
+                            format!("NEW.c{ref_col}")
+                        } else {
+                            rng.random_range(0..1000).to_string()
+                        };
+                        format!("UPDATE t SET c{update_col} = {new_value} WHERE c{update_col} {op} {threshold}")
+                    }
+                    2 => {
+                        // DELETE action
+                        let delete_col = rng.random_range(0..num_cols);
+                        let op = match rng.random_range(0..=3) {
+                            0 => "<",
+                            1 => "<=",
+                            2 => ">",
+                            3 => ">=",
+                            _ => unreachable!(),
+                        };
+                        let threshold = if has_old && rng.random_bool(0.3) {
+                            let ref_col = rng.random_range(0..num_cols);
+                            format!("OLD.c{ref_col}")
+                        } else if has_new && rng.random_bool(0.3) {
+                            let ref_col = rng.random_range(0..num_cols);
+                            format!("NEW.c{ref_col}")
+                        } else {
+                            rng.random_range(0..1000).to_string()
+                        };
+                        format!("DELETE FROM t WHERE c{delete_col} {op} {threshold}")
+                    }
+                    _ => unreachable!(),
+                };
+
+                let create_trigger = format!(
+                    "CREATE TRIGGER test_trigger {trigger_time} {trigger_event} ON t BEGIN {trigger_action}; END;",
+                );
+
+                sqlite_conn.execute(&create_trigger, params![]).unwrap();
+                limbo_exec_rows(&limbo_db, &limbo_conn, &create_trigger);
+                Some(create_trigger)
+            } else {
+                None
+            };
+            if let Some(ref trigger) = trigger {
+                println!("{trigger};");
+            }
+
             const COMPARISONS: [&str; 3] = ["=", "<", ">"];
             const INNER_ITERATIONS: usize = 20;
 
@@ -1976,7 +2335,6 @@ mod fuzz_tests {
                 let do_update = rng.random_range(0..2) == 0;
 
                 let comparison = COMPARISONS[rng.random_range(0..COMPARISONS.len())];
-                let affected_col = rng.random_range(0..num_cols);
                 let predicate_col = rng.random_range(0..num_cols);
                 let predicate_value = rng.random_range(0..1000);
 
@@ -2000,6 +2358,7 @@ mod fuzz_tests {
                 };
 
                 let query = if do_update {
+                    let affected_col = rng.random_range(0..num_cols);
                     let num_updates = rng.random_range(1..=num_cols);
                     let mut values = Vec::new();
                     for _ in 0..num_updates {
@@ -2048,10 +2407,19 @@ mod fuzz_tests {
                 let sqlite_rows = sqlite_exec_rows(&sqlite_conn, &verify_query);
                 let limbo_rows = limbo_exec_rows(&limbo_db, &limbo_conn, &verify_query);
 
-                assert_eq!(
-                    sqlite_rows, limbo_rows,
-                    "Different results after mutation! limbo: {limbo_rows:?}, sqlite: {sqlite_rows:?}, seed: {seed}, query: {query}",
-                );
+                if sqlite_rows != limbo_rows {
+                    let diff_msg = format_rows_diff(
+                        &sqlite_rows,
+                        &limbo_rows,
+                        seed,
+                        &query,
+                        &table_def,
+                        &indexes,
+                        trigger.as_ref(),
+                        &dml_statements,
+                    );
+                    panic!("Different results after mutation!{diff_msg}");
+                }
 
                 // Run integrity check on limbo db using rusqlite
                 if let Err(e) = rusqlite_integrity_check(&limbo_db.path) {
@@ -2059,6 +2427,9 @@ mod fuzz_tests {
                     for t in indexes.iter() {
                         println!("{t};");
                     }
+                    if let Some(trigger) = trigger {
+                        println!("{trigger};");
+                    }
                     for t in dml_statements.iter() {
                         println!("{t};");
                     }
diff --git a/tests/integration/mod.rs b/tests/integration/mod.rs
index dcb8e3b0f..b03df00be 100644
--- a/tests/integration/mod.rs
+++ b/tests/integration/mod.rs
@@ -5,6 +5,7 @@ mod index_method;
 mod pragma;
 mod query_processing;
 mod storage;
+mod trigger;
 mod wal;
 
 #[cfg(test)]
diff --git a/tests/integration/trigger.rs b/tests/integration/trigger.rs
new file mode 100644
index 000000000..22122e0b7
--- /dev/null
+++ b/tests/integration/trigger.rs
@@ -0,0 +1,888 @@
+use crate::common::TempDatabase;
+
+#[test]
+fn test_create_trigger() {
+    let _ = tracing_subscriber::fmt()
+        .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
+        .try_init();
+
+    let db = TempDatabase::new_empty();
+    let conn = db.connect_limbo();
+
+    conn.execute("CREATE TABLE test (x, y TEXT)").unwrap();
+
+    conn.execute(
+        "CREATE TRIGGER t1 BEFORE INSERT ON test BEGIN
+         INSERT INTO test VALUES (100, 'triggered');
+        END",
+    )
+    .unwrap();
+
+    conn.execute("INSERT INTO test VALUES (1, 'hello')")
+        .unwrap();
+
+    let mut stmt = conn.prepare("SELECT * FROM test ORDER BY rowid").unwrap();
+    let mut results = Vec::new();
+    loop {
+        match stmt.step().unwrap() {
+            turso_core::StepResult::Row => {
+                let row = stmt.row().unwrap();
+                results.push((
+                    row.get_value(0).as_int().unwrap(),
+                    row.get_value(1).cast_text().unwrap().to_string(),
+                ));
+            }
+            turso_core::StepResult::Done => break,
+            turso_core::StepResult::IO => {
+                stmt.run_once().unwrap();
+            }
+            _ => panic!("Unexpected step result"),
+        }
+    }
+
+    // Row inserted by trigger goes first
+    assert_eq!(results[0], (100, "triggered".to_string()));
+    assert_eq!(results[1], (1, "hello".to_string()));
+}
+
+#[test]
+fn test_drop_trigger() {
+    let db = TempDatabase::new_empty();
+    let conn = db.connect_limbo();
+
+    conn.execute("CREATE TABLE test (x INTEGER PRIMARY KEY)")
+        .unwrap();
+
+    conn.execute("CREATE TRIGGER t1 BEFORE INSERT ON test BEGIN SELECT 1; END")
+        .unwrap();
+
+    // Verify trigger exists
+    let mut stmt = conn
+        .prepare("SELECT name FROM sqlite_schema WHERE type='trigger' AND name='t1'")
+        .unwrap();
+    let mut results = Vec::new();
+    loop {
+        match stmt.step().unwrap() {
+            turso_core::StepResult::Row => {
+                let row = stmt.row().unwrap();
+                results.push(row.get_value(0).cast_text().unwrap().to_string());
+            }
+            turso_core::StepResult::Done => break,
+            turso_core::StepResult::IO => {
+                stmt.run_once().unwrap();
+            }
+            _ => panic!("Unexpected step result"),
+        }
+    }
+    assert_eq!(results.len(), 1);
+
+    conn.execute("DROP TRIGGER t1").unwrap();
+
+    // Verify trigger is gone
+    let mut stmt = conn
+        .prepare("SELECT name FROM sqlite_schema WHERE type='trigger' AND name='t1'")
+        .unwrap();
+    let mut results = Vec::new();
+    loop {
+        match stmt.step().unwrap() {
+            turso_core::StepResult::Row => {
+                let row = stmt.row().unwrap();
+                results.push(row.get_value(0).cast_text().unwrap().to_string());
+            }
+            turso_core::StepResult::Done => break,
+            turso_core::StepResult::IO => {
+                stmt.run_once().unwrap();
+            }
+            _ => panic!("Unexpected step result"),
+        }
+    }
+    assert_eq!(results.len(), 0);
+}
+
+#[test]
+fn test_trigger_after_insert() {
+    let db = TempDatabase::new_empty();
+    let conn = db.connect_limbo();
+
+    conn.execute("CREATE TABLE test (x INTEGER PRIMARY KEY, y TEXT)")
+        .unwrap();
+    conn.execute("CREATE TABLE log (x INTEGER, y TEXT)")
+        .unwrap();
+
+    conn.execute(
+        "CREATE TRIGGER t1 AFTER INSERT ON test BEGIN
+         INSERT INTO log VALUES (NEW.x, NEW.y);
+        END",
+    )
+    .unwrap();
+
+    conn.execute("INSERT INTO test VALUES (1, 'hello')")
+        .unwrap();
+
+    let mut stmt = conn.prepare("SELECT * FROM log").unwrap();
+    let mut results = Vec::new();
+    loop {
+        match stmt.step().unwrap() {
+            turso_core::StepResult::Row => {
+                let row = stmt.row().unwrap();
+                results.push((
+                    row.get_value(0).as_int().unwrap(),
+                    row.get_value(1).cast_text().unwrap().to_string(),
+                ));
+            }
+            turso_core::StepResult::Done => break,
+            turso_core::StepResult::IO => {
+                stmt.run_once().unwrap();
+            }
+            _ => panic!("Unexpected step result"),
+        }
+    }
+
+    assert_eq!(results.len(), 1);
+    assert_eq!(results[0], (1, "hello".to_string()));
+}
+
+#[test]
+fn test_trigger_when_clause() {
+    let db = TempDatabase::new_empty();
+    let conn = db.connect_limbo();
+
+    conn.execute("CREATE TABLE test (x INTEGER PRIMARY KEY, y INTEGER)")
+        .unwrap();
+    conn.execute("CREATE TABLE log (x INTEGER)").unwrap();
+
+    conn.execute(
+        "CREATE TRIGGER t1 AFTER INSERT ON test WHEN NEW.y > 10 BEGIN
+         INSERT INTO log VALUES (NEW.x);
+        END",
+    )
+    .unwrap();
+
+    conn.execute("INSERT INTO test VALUES (1, 5)").unwrap();
+    conn.execute("INSERT INTO test VALUES (2, 15)").unwrap();
+
+    let mut stmt = conn.prepare("SELECT * FROM log").unwrap();
+    let mut results = Vec::new();
+    loop {
+        match stmt.step().unwrap() {
+            turso_core::StepResult::Row => {
+                let row = stmt.row().unwrap();
+                results.push(row.get_value(0).as_int().unwrap());
+            }
+            turso_core::StepResult::Done => break,
+            turso_core::StepResult::IO => {
+                stmt.run_once().unwrap();
+            }
+            _ => panic!("Unexpected step result"),
+        }
+    }
+
+    assert_eq!(results.len(), 1);
+    assert_eq!(results[0], 2);
+}
+
+#[test]
+fn test_trigger_drop_table_drops_triggers() {
+    let db = TempDatabase::new_empty();
+    let conn = db.connect_limbo();
+
+    conn.execute("CREATE TABLE test (x INTEGER PRIMARY KEY)")
+        .unwrap();
+    conn.execute("CREATE TRIGGER t1 BEFORE INSERT ON test BEGIN SELECT 1; END")
+        .unwrap();
+
+    // Verify trigger exists
+    let mut stmt = conn
+        .prepare("SELECT name FROM sqlite_schema WHERE type='trigger' AND name='t1'")
+        .unwrap();
+    let mut results = Vec::new();
+    loop {
+        match stmt.step().unwrap() {
+            turso_core::StepResult::Row => {
+                let row = stmt.row().unwrap();
+                results.push(row.get_value(0).cast_text().unwrap().to_string());
+            }
+            turso_core::StepResult::Done => break,
+            turso_core::StepResult::IO => {
+                stmt.run_once().unwrap();
+            }
+            _ => panic!("Unexpected step result"),
+        }
+    }
+    assert_eq!(results.len(), 1);
+
+    conn.execute("DROP TABLE test").unwrap();
+
+    // Verify trigger is gone
+    let mut stmt = conn
+        .prepare("SELECT name FROM sqlite_schema WHERE type='trigger' AND name='t1'")
+        .unwrap();
+    let mut results = Vec::new();
+    loop {
+        match stmt.step().unwrap() {
+            turso_core::StepResult::Row => {
+                let row = stmt.row().unwrap();
+                results.push(row.get_value(0).cast_text().unwrap().to_string());
+            }
+            turso_core::StepResult::Done => break,
+            turso_core::StepResult::IO => {
+                stmt.run_once().unwrap();
+            }
+            _ => panic!("Unexpected step result"),
+        }
+    }
+    assert_eq!(results.len(), 0);
+}
+
+#[test]
+fn test_trigger_new_old_references() {
+    let db = TempDatabase::new_empty();
+    let conn = db.connect_limbo();
+
+    conn.execute("CREATE TABLE test (x INTEGER PRIMARY KEY, y TEXT)")
+        .unwrap();
+    conn.execute("CREATE TABLE log (msg TEXT)").unwrap();
+
+    conn.execute("INSERT INTO test VALUES (1, 'hello')")
+        .unwrap();
+
+    conn.execute(
+        "CREATE TRIGGER t1 AFTER UPDATE ON test BEGIN
+         INSERT INTO log VALUES ('old=' || OLD.y || ' new=' || NEW.y);
+        END",
+    )
+    .unwrap();
+
+    conn.execute("UPDATE test SET y = 'world' WHERE x = 1")
+        .unwrap();
+
+    let mut stmt = conn.prepare("SELECT * FROM log").unwrap();
+    let mut results = Vec::new();
+    loop {
+        match stmt.step().unwrap() {
+            turso_core::StepResult::Row => {
+                let row = stmt.row().unwrap();
+                results.push(row.get_value(0).cast_text().unwrap().to_string());
+            }
+            turso_core::StepResult::Done => break,
+            turso_core::StepResult::IO => {
+                stmt.run_once().unwrap();
+            }
+            _ => panic!("Unexpected step result"),
+        }
+    }
+
+    assert_eq!(results.len(), 1);
+    assert_eq!(results[0], "old=hello new=world");
+}
+
+#[test]
+fn test_multiple_triggers_same_event() {
+    let db = TempDatabase::new_empty();
+    let conn = db.connect_limbo();
+
+    conn.execute("CREATE TABLE test (x INTEGER PRIMARY KEY)")
+        .unwrap();
+    conn.execute("CREATE TABLE log (msg TEXT)").unwrap();
+
+    conn.execute(
+        "CREATE TRIGGER t1 BEFORE INSERT ON test BEGIN
+         INSERT INTO log VALUES ('trigger1');
+        END",
+    )
+    .unwrap();
+
+    conn.execute(
+        "CREATE TRIGGER t2 BEFORE INSERT ON test BEGIN
+         INSERT INTO log VALUES ('trigger2');
+        END",
+    )
+    .unwrap();
+
+    conn.execute("INSERT INTO test VALUES (1)").unwrap();
+
+    let mut stmt = conn.prepare("SELECT * FROM log ORDER BY msg").unwrap();
+    let mut results = Vec::new();
+    loop {
+        match stmt.step().unwrap() {
+            turso_core::StepResult::Row => {
+                let row = stmt.row().unwrap();
+                results.push(row.get_value(0).cast_text().unwrap().to_string());
+            }
+            turso_core::StepResult::Done => break,
+            turso_core::StepResult::IO => {
+                stmt.run_once().unwrap();
+            }
+            _ => panic!("Unexpected step result"),
+        }
+    }
+
+    assert_eq!(results.len(), 2);
+    assert_eq!(results[0], "trigger1");
+    assert_eq!(results[1], "trigger2");
+}
+
+#[test]
+fn test_two_triggers_on_same_table() {
+    let db = TempDatabase::new_empty();
+    let conn = db.connect_limbo();
+
+    conn.execute("CREATE TABLE test (x, msg TEXT)").unwrap();
+    conn.execute("CREATE TABLE log (msg TEXT)").unwrap();
+
+    // Trigger A: fires on INSERT to test, inserts into log and test (which would trigger B)
+    conn.execute(
+        "CREATE TRIGGER trigger_a AFTER INSERT ON test BEGIN
+         INSERT INTO log VALUES ('trigger_a fired for x=' || NEW.x);
+         INSERT INTO test VALUES (NEW.x + 100, 'from_a');
+        END",
+    )
+    .unwrap();
+
+    // Trigger B: fires on INSERT to test, inserts into log and test (which would trigger A)
+    conn.execute(
+        "CREATE TRIGGER trigger_b AFTER INSERT ON test BEGIN
+         INSERT INTO log VALUES ('trigger_b fired for x=' || NEW.x);
+         INSERT INTO test VALUES (NEW.x + 200, 'from_b');
+        END",
+    )
+    .unwrap();
+
+    // Insert initial row - this should trigger A, which triggers B, which tries to trigger A again (prevented)
+    conn.execute("INSERT INTO test VALUES (1, 'initial')")
+        .unwrap();
+
+    // Check log entries to verify recursion was prevented
+    let mut stmt = conn.prepare("SELECT * FROM log ORDER BY rowid").unwrap();
+    let mut results = Vec::new();
+    loop {
+        match stmt.step().unwrap() {
+            turso_core::StepResult::Row => {
+                let row = stmt.row().unwrap();
+                results.push(row.get_value(0).cast_text().unwrap().to_string());
+            }
+            turso_core::StepResult::Done => break,
+            turso_core::StepResult::IO => {
+                stmt.run_once().unwrap();
+            }
+            _ => panic!("Unexpected step result"),
+        }
+    }
+
+    // At minimum, we should see both triggers fire and not infinite loop
+    assert!(
+        results.len() >= 2,
+        "Expected at least 2 log entries, got {}",
+        results.len()
+    );
+    assert!(
+        results.iter().any(|s| s.contains("trigger_a")),
+        "trigger_a should have fired"
+    );
+    assert!(
+        results.iter().any(|s| s.contains("trigger_b")),
+        "trigger_b should have fired"
+    );
+}
+
+#[test]
+fn test_trigger_mutual_recursion() {
+    let db = TempDatabase::new_empty();
+    let conn = db.connect_limbo();
+
+    conn.execute("CREATE TABLE t (id INTEGER, msg TEXT)")
+        .unwrap();
+    conn.execute("CREATE TABLE u (id INTEGER, msg TEXT)")
+        .unwrap();
+
+    // Trigger on T: fires on INSERT to t, inserts into u
+    conn.execute(
+        "CREATE TRIGGER trigger_on_t AFTER INSERT ON t BEGIN
+         INSERT INTO u VALUES (NEW.id + 1000, 'from_t');
+        END",
+    )
+    .unwrap();
+
+    // Trigger on U: fires on INSERT to u, inserts into t
+    conn.execute(
+        "CREATE TRIGGER trigger_on_u AFTER INSERT ON u BEGIN
+         INSERT INTO t VALUES (NEW.id + 2000, 'from_u');
+        END",
+    )
+    .unwrap();
+
+    // Insert initial row into t - this should trigger the chain
+    conn.execute("INSERT INTO t VALUES (1, 'initial')").unwrap();
+
+    // Check that both tables have entries
+    let mut stmt = conn.prepare("SELECT * FROM t ORDER BY rowid").unwrap();
+    let mut t_results = Vec::new();
+    loop {
+        match stmt.step().unwrap() {
+            turso_core::StepResult::Row => {
+                let row = stmt.row().unwrap();
+                t_results.push((
+                    row.get_value(0).as_int().unwrap(),
+                    row.get_value(1).cast_text().unwrap().to_string(),
+                ));
+            }
+            turso_core::StepResult::Done => break,
+            turso_core::StepResult::IO => {
+                stmt.run_once().unwrap();
+            }
+            _ => panic!("Unexpected step result"),
+        }
+    }
+
+    let mut stmt = conn.prepare("SELECT * FROM u ORDER BY rowid").unwrap();
+    let mut u_results = Vec::new();
+    loop {
+        match stmt.step().unwrap() {
+            turso_core::StepResult::Row => {
+                let row = stmt.row().unwrap();
+                u_results.push((
+                    row.get_value(0).as_int().unwrap(),
+                    row.get_value(1).cast_text().unwrap().to_string(),
+                ));
+            }
+            turso_core::StepResult::Done => break,
+            turso_core::StepResult::IO => {
+                stmt.run_once().unwrap();
+            }
+            _ => panic!("Unexpected step result"),
+        }
+    }
+
+    // Verify the chain executed without infinite recursion
+    assert!(!t_results.is_empty(), "Expected at least 1 entry in t");
+    assert!(!u_results.is_empty(), "Expected at least 1 entry in u");
+
+    // Verify initial insert
+    assert_eq!(t_results[0], (1, "initial".to_string()));
+
+    // Verify trigger on t fired (inserted into u)
+    assert_eq!(u_results[0], (1001, "from_t".to_string()));
+}
+
+#[test]
+fn test_after_insert_trigger() {
+    let db = TempDatabase::new_empty();
+    let conn = db.connect_limbo();
+
+    // Create table and log table
+    conn.execute("CREATE TABLE items (id INTEGER PRIMARY KEY, name TEXT)")
+        .unwrap();
+    conn.execute("CREATE TABLE audit_log (action TEXT, item_id INTEGER, item_name TEXT)")
+        .unwrap();
+
+    // Create AFTER INSERT trigger
+    conn.execute(
+        "CREATE TRIGGER after_insert_items
+         AFTER INSERT ON items
+         BEGIN
+             INSERT INTO audit_log VALUES ('INSERT', NEW.id, NEW.name);
+         END",
+    )
+    .unwrap();
+
+    // Insert data
+    conn.execute("INSERT INTO items VALUES (1, 'apple')")
+        .unwrap();
+    conn.execute("INSERT INTO items VALUES (2, 'banana')")
+        .unwrap();
+
+    // Verify audit log
+    let mut stmt = conn
+        .prepare("SELECT * FROM audit_log ORDER BY rowid")
+        .unwrap();
+    let mut results = Vec::new();
+    loop {
+        match stmt.step().unwrap() {
+            turso_core::StepResult::Row => {
+                let row = stmt.row().unwrap();
+                results.push((
+                    row.get_value(0).cast_text().unwrap().to_string(),
+                    row.get_value(1).as_int().unwrap(),
+                    row.get_value(2).cast_text().unwrap().to_string(),
+                ));
+            }
+            turso_core::StepResult::Done => break,
+            turso_core::StepResult::IO => {
+                stmt.run_once().unwrap();
+            }
+            _ => panic!("Unexpected step result"),
+        }
+    }
+
+    assert_eq!(results.len(), 2);
+    assert_eq!(results[0], ("INSERT".to_string(), 1, "apple".to_string()));
+    assert_eq!(results[1], ("INSERT".to_string(), 2, "banana".to_string()));
+}
+
+#[test]
+fn test_before_update_of_trigger() {
+    let db = TempDatabase::new_empty();
+    let conn = db.connect_limbo();
+
+    // Create table with multiple columns
+    conn.execute("CREATE TABLE products (id INTEGER PRIMARY KEY, name TEXT, price INTEGER)")
+        .unwrap();
+    conn.execute(
+        "CREATE TABLE price_history (product_id INTEGER, old_price INTEGER, new_price INTEGER)",
+    )
+    .unwrap();
+
+    // Create BEFORE UPDATE OF trigger - only fires when price column is updated
+    conn.execute(
+        "CREATE TRIGGER before_update_price
+         BEFORE UPDATE OF price ON products
+         BEGIN
+             INSERT INTO price_history VALUES (OLD.id, OLD.price, NEW.price);
+         END",
+    )
+    .unwrap();
+
+    // Insert initial data
+    conn.execute("INSERT INTO products VALUES (1, 'widget', 100)")
+        .unwrap();
+    conn.execute("INSERT INTO products VALUES (2, 'gadget', 200)")
+        .unwrap();
+
+    // Update price - should fire trigger
+    conn.execute("UPDATE products SET price = 150 WHERE id = 1")
+        .unwrap();
+
+    // Update name only - should NOT fire trigger
+    conn.execute("UPDATE products SET name = 'super widget' WHERE id = 1")
+        .unwrap();
+
+    // Update both name and price - should fire trigger
+    conn.execute("UPDATE products SET name = 'mega gadget', price = 250 WHERE id = 2")
+        .unwrap();
+
+    // Verify price history
+    let mut stmt = conn
+        .prepare("SELECT * FROM price_history ORDER BY rowid")
+        .unwrap();
+    let mut results = Vec::new();
+    loop {
+        match stmt.step().unwrap() {
+            turso_core::StepResult::Row => {
+                let row = stmt.row().unwrap();
+                results.push((
+                    row.get_value(0).as_int().unwrap(),
+                    row.get_value(1).as_int().unwrap(),
+                    row.get_value(2).as_int().unwrap(),
+                ));
+            }
+            turso_core::StepResult::Done => break,
+            turso_core::StepResult::IO => {
+                stmt.run_once().unwrap();
+            }
+            _ => panic!("Unexpected step result"),
+        }
+    }
+
+    // Should have 2 entries (not 3, because name-only update didn't fire)
+    assert_eq!(results.len(), 2);
+    assert_eq!(results[0], (1, 100, 150));
+    assert_eq!(results[1], (2, 200, 250));
+}
+
+#[test]
+fn test_after_update_of_trigger() {
+    let db = TempDatabase::new_empty();
+    let conn = db.connect_limbo();
+
+    // Create table
+    conn.execute("CREATE TABLE employees (id INTEGER PRIMARY KEY, name TEXT, salary INTEGER)")
+        .unwrap();
+    conn.execute("CREATE TABLE salary_changes (emp_id INTEGER, old_salary INTEGER, new_salary INTEGER, change_amount INTEGER)")
+        .unwrap();
+
+    // Create AFTER UPDATE OF trigger with multiple statements
+    conn.execute(
+        "CREATE TRIGGER after_update_salary
+         AFTER UPDATE OF salary ON employees
+         BEGIN
+             INSERT INTO salary_changes VALUES (NEW.id, OLD.salary, NEW.salary, NEW.salary - OLD.salary);
+         END",
+    )
+    .unwrap();
+
+    // Insert initial data
+    conn.execute("INSERT INTO employees VALUES (1, 'Alice', 50000)")
+        .unwrap();
+    conn.execute("INSERT INTO employees VALUES (2, 'Bob', 60000)")
+        .unwrap();
+
+    // Update salary
+    conn.execute("UPDATE employees SET salary = 55000 WHERE id = 1")
+        .unwrap();
+    conn.execute("UPDATE employees SET salary = 65000 WHERE id = 2")
+        .unwrap();
+
+    // Verify salary changes
+    let mut stmt = conn
+        .prepare("SELECT * FROM salary_changes ORDER BY rowid")
+        .unwrap();
+    let mut results = Vec::new();
+    loop {
+        match stmt.step().unwrap() {
+            turso_core::StepResult::Row => {
+                let row = stmt.row().unwrap();
+                results.push((
+                    row.get_value(0).as_int().unwrap(),
+                    row.get_value(1).as_int().unwrap(),
+                    row.get_value(2).as_int().unwrap(),
+                    row.get_value(3).as_int().unwrap(),
+                ));
+            }
+            turso_core::StepResult::Done => break,
+            turso_core::StepResult::IO => {
+                stmt.run_once().unwrap();
+            }
+            _ => panic!("Unexpected step result"),
+        }
+    }
+
+    assert_eq!(results.len(), 2);
+    assert_eq!(results[0], (1, 50000, 55000, 5000));
+    assert_eq!(results[1], (2, 60000, 65000, 5000));
+}
+
+fn log(s: &str) -> &str {
+    tracing::info!("{}", s);
+    s
+}
+
+#[test]
+fn test_before_delete_trigger() {
+    let db = TempDatabase::new_empty();
+    let conn = db.connect_limbo();
+
+    // Create tables
+    conn.execute(log(
+        "CREATE TABLE users (id INTEGER PRIMARY KEY, username TEXT)",
+    ))
+    .unwrap();
+    conn.execute(log(
+        "CREATE TABLE deleted_users (id INTEGER, username TEXT, deleted_at INTEGER)",
+    ))
+    .unwrap();
+
+    // Create BEFORE DELETE trigger
+    conn.execute(log("CREATE TRIGGER before_delete_users
+         BEFORE DELETE ON users
+         BEGIN
+             INSERT INTO deleted_users VALUES (OLD.id, OLD.username, 12345);
+         END"))
+        .unwrap();
+
+    // Insert data
+    conn.execute(log("INSERT INTO users VALUES (1, 'alice')"))
+        .unwrap();
+    conn.execute(log("INSERT INTO users VALUES (2, 'bob')"))
+        .unwrap();
+    conn.execute(log("INSERT INTO users VALUES (3, 'charlie')"))
+        .unwrap();
+
+    // Delete some users
+    conn.execute(log("DELETE FROM users WHERE id = 2")).unwrap();
+    conn.execute(log("DELETE FROM users WHERE id = 3")).unwrap();
+
+    // Verify deleted_users table
+    let mut stmt = conn
+        .prepare(log("SELECT * FROM deleted_users ORDER BY id"))
+        .unwrap();
+    let mut results = Vec::new();
+    loop {
+        match stmt.step().unwrap() {
+            turso_core::StepResult::Row => {
+                let row = stmt.row().unwrap();
+                results.push((
+                    row.get_value(0).as_int().unwrap(),
+                    row.get_value(1).cast_text().unwrap().to_string(),
+                    row.get_value(2).as_int().unwrap(),
+                ));
+            }
+            turso_core::StepResult::Done => break,
+            turso_core::StepResult::IO => {
+                stmt.run_once().unwrap();
+            }
+            _ => panic!("Unexpected step result"),
+        }
+    }
+
+    assert_eq!(results.len(), 2);
+    assert_eq!(results[0], (2, "bob".to_string(), 12345));
+    assert_eq!(results[1], (3, "charlie".to_string(), 12345));
+
+    // Verify remaining users
+    let mut stmt = conn.prepare(log("SELECT COUNT(*) FROM users")).unwrap();
+    loop {
+        match stmt.step().unwrap() {
+            turso_core::StepResult::Row => {
+                let row = stmt.row().unwrap();
+                assert_eq!(row.get_value(0).as_int().unwrap(), 1);
+                break;
+            }
+            turso_core::StepResult::Done => break,
+            turso_core::StepResult::IO => {
+                stmt.run_once().unwrap();
+            }
+            _ => panic!("Unexpected step result"),
+        }
+    }
+}
+
+#[test]
+fn test_after_delete_trigger() {
+    let db = TempDatabase::new_empty();
+    let conn = db.connect_limbo();
+
+    // Create tables
+    conn.execute(
+        "CREATE TABLE orders (id INTEGER PRIMARY KEY, customer_id INTEGER, amount INTEGER)",
+    )
+    .unwrap();
+    conn.execute(
+        "CREATE TABLE order_archive (order_id INTEGER, customer_id INTEGER, amount INTEGER)",
+    )
+    .unwrap();
+
+    // Create AFTER DELETE trigger
+    conn.execute(
+        "CREATE TRIGGER after_delete_orders
+         AFTER DELETE ON orders
+         BEGIN
+             INSERT INTO order_archive VALUES (OLD.id, OLD.customer_id, OLD.amount);
+         END",
+    )
+    .unwrap();
+
+    // Insert data
+    conn.execute("INSERT INTO orders VALUES (1, 100, 50)")
+        .unwrap();
+    conn.execute("INSERT INTO orders VALUES (2, 101, 75)")
+        .unwrap();
+    conn.execute("INSERT INTO orders VALUES (3, 100, 100)")
+        .unwrap();
+
+    // Delete orders
+    conn.execute("DELETE FROM orders WHERE customer_id = 100")
+        .unwrap();
+
+    // Verify archive
+    let mut stmt = conn
+        .prepare("SELECT * FROM order_archive ORDER BY order_id")
+        .unwrap();
+    let mut results = Vec::new();
+    loop {
+        match stmt.step().unwrap() {
+            turso_core::StepResult::Row => {
+                let row = stmt.row().unwrap();
+                results.push((
+                    row.get_value(0).as_int().unwrap(),
+                    row.get_value(1).as_int().unwrap(),
+                    row.get_value(2).as_int().unwrap(),
+                ));
+            }
+            turso_core::StepResult::Done => break,
+            turso_core::StepResult::IO => {
+                stmt.run_once().unwrap();
+            }
+            _ => panic!("Unexpected step result"),
+        }
+    }
+
+    assert_eq!(results.len(), 2);
+    assert_eq!(results[0], (1, 100, 50));
+    assert_eq!(results[1], (3, 100, 100));
+}
+
+#[test]
+fn test_trigger_with_multiple_statements() {
+    let db = TempDatabase::new_empty();
+    let conn = db.connect_limbo();
+
+    // Create tables
+    conn.execute("CREATE TABLE accounts (id INTEGER PRIMARY KEY, balance INTEGER)")
+        .unwrap();
+    conn.execute(
+        "CREATE TABLE transactions (account_id INTEGER, old_balance INTEGER, new_balance INTEGER)",
+    )
+    .unwrap();
+    conn.execute("CREATE TABLE audit (message TEXT)").unwrap();
+
+    // Create trigger with multiple statements
+    conn.execute(
+        "CREATE TRIGGER track_balance_changes
+         AFTER UPDATE OF balance ON accounts
+         BEGIN
+             INSERT INTO transactions VALUES (NEW.id, OLD.balance, NEW.balance);
+             INSERT INTO audit VALUES ('Balance changed for account ' || NEW.id);
+         END",
+    )
+    .unwrap();
+
+    // Insert initial data
+    conn.execute("INSERT INTO accounts VALUES (1, 1000)")
+        .unwrap();
+    conn.execute("INSERT INTO accounts VALUES (2, 2000)")
+        .unwrap();
+
+    // Update balances
+    conn.execute("UPDATE accounts SET balance = 1500 WHERE id = 1")
+        .unwrap();
+    conn.execute("UPDATE accounts SET balance = 2500 WHERE id = 2")
+        .unwrap();
+
+    // Verify transactions table
+    let mut stmt = conn
+        .prepare("SELECT * FROM transactions ORDER BY rowid")
+        .unwrap();
+    let mut trans_results = Vec::new();
+    loop {
+        match stmt.step().unwrap() {
+            turso_core::StepResult::Row => {
+                let row = stmt.row().unwrap();
+                trans_results.push((
+                    row.get_value(0).as_int().unwrap(),
+                    row.get_value(1).as_int().unwrap(),
+                    row.get_value(2).as_int().unwrap(),
+                ));
+            }
+            turso_core::StepResult::Done => break,
+            turso_core::StepResult::IO => {
+                stmt.run_once().unwrap();
+            }
+            _ => panic!("Unexpected step result"),
+        }
+    }
+
+    assert_eq!(trans_results.len(), 2);
+    assert_eq!(trans_results[0], (1, 1000, 1500));
+    assert_eq!(trans_results[1], (2, 2000, 2500));
+
+    // Verify audit table
+    let mut stmt = conn.prepare("SELECT * FROM audit ORDER BY rowid").unwrap();
+    let mut audit_results = Vec::new();
+    loop {
+        match stmt.step().unwrap() {
+            turso_core::StepResult::Row => {
+                let row = stmt.row().unwrap();
+                audit_results.push(row.get_value(0).cast_text().unwrap().to_string());
+            }
+            turso_core::StepResult::Done => break,
+            turso_core::StepResult::IO => {
+                stmt.run_once().unwrap();
+            }
+            _ => panic!("Unexpected step result"),
+        }
+    }
+
+    assert_eq!(audit_results.len(), 2);
+    assert_eq!(audit_results[0], "Balance changed for account 1");
+    assert_eq!(audit_results[1], "Balance changed for account 2");
+}