fix: close FOREIGN KEY DDL and catalog review gaps

ALTER TABLE could leave Schema.ForeignKeys stale: DROP/RENAME COLUMN and
RENAME TABLE rewrote only the column/catalog metadata, so a constraint kept
naming a missing column (silently disabling enforcement or failing every
write) and a renamed parent table was no longer checked. Reject those DDLs
with 0A000 when a persisted FOREIGN KEY depends on the column or names the
table as its parent, mirroring the CHECK guard. This covers both sides of a
constraint, including self-references.

Also close the remaining review findings:

- Reject DROP TABLE of a referenced parent (including a self-reference) with
  0A000 instead of leaving the child constraints dangling.
- Take the foreign-key write gate around CREATE TABLE and DROP TABLE so the
  graph cannot change underneath a concurrent insert/delete validation.
- Persist declared ON DELETE/ON UPDATE actions (NO ACTION/RESTRICT) and render
  them in the emulated sqlite_master/sqlite_schema table SQL.
- Return pgerror.Unsupported for PRAGMA foreign_keys = OFF so both the HTTP and
  PostgreSQL surfaces report 0A000 instead of XX000.
- Document Schema.ForeignKeys in PKBFI_STORAGE_MIGRATION.md: a pre-FK engine
  drops the field on the next schema write, so a rollback silently stops
  enforcing referential integrity.
- Capability notes now state always-on enforcement (PRAGMA foreign_keys reports
  1, OFF rejected, SQLite defaults OFF), the database-wide foreign-key write
  gate, that PRAGMA foreign_key_list is not emulated, and that MATCH is not
  parsed (MATCH SIMPLE only).

Tests: ALTER DROP/RENAME COLUMN on both sides, RENAME TABLE and DROP TABLE of a
referenced parent, self-reference DDL rejections, declared action round-trip in
sqlite_master, PRAGMA OFF over PostgreSQL, parent+child insert in one
transaction, sibling references in one multi-row insert, INSERT OR REPLACE on a
referenced parent, reordered UNIQUE(x, y) target for REFERENCES (y, x), and a
100-round concurrent child-insert vs parent-delete race asserting no dangling
row and never two commits.

PKBFI_STORAGE_MIGRATION.md

@@ -39,6 +39,13 @@ The migration command exits after writing and verifying the destination. It refu
   pre-CHECK engine ignores it on read and drops it on the next schema write, so
   rolling back to such an engine silently stops enforcing CHECK constraints;
   restore the newer engine (or re-create the constraints) after a rollback.
+- `Schema.ForeignKeys` (`foreign_keys`) is an additive, optional JSON array of
+  foreign-key declarations (child columns, referenced table and columns, and any
+  declared `ON DELETE`/`ON UPDATE` action text). A pre-foreign-key engine ignores
+  the field on read and drops it on the next schema write, so rolling back to
+  such an engine silently stops enforcing referential integrity: dangling child
+  rows become insertable and deleting a referenced parent no longer fails.
+  Restore the newer engine (or re-create the constraints) after a rollback.
 - `Schema.PrimaryKeyColumns` (`primary_key_columns`) is an additive, optional
   JSON field for composite primary keys. A single-column catalog still carries
   only `primary_key`; a composite catalog persists `primary_key` as an empty

pkg/capabilities/capabilities.go

@@ -169,7 +169,7 @@ func Current() *Document {
 			Unique:        Feature{Supported: true, Status: "enforced", SQLState: "23505", Notes: "unique indexes, INLINE UNIQUE, and ON CONFLICT are enforced atomically"},
 			PrimaryKey:    Feature{Supported: true, Status: "enforced", SQLState: "23505", Notes: "inline and table-level single-column keys; INTEGER PRIMARY KEY remains the rowid alias"},
 			CompositePK:   Feature{Supported: true, Status: "enforced", SQLState: "23505", Notes: "table-level PRIMARY KEY (a, b) with ordered columns; key columns are implicitly NOT NULL (23502) and a composite key is never a rowid alias; two inline PRIMARY KEY declarations or an inline/table-level mix are rejected with 0A000; the key is durable under a length-prefixed encoding and reported by PRAGMA table_info pk ordinals and sqlite_master table SQL, but no implicit index is created for it; a conflict target that names the key columns matches them in any order, and ON CONFLICT DO UPDATE cannot assign a key column (0A000)"},
-			ForeignKey:    Feature{Supported: true, Status: "enforced", SQLState: "23503", Notes: "inline REFERENCES and table-level FOREIGN KEY (named or unnamed, single or multi-column) are accepted in CREATE TABLE; the parent table must already exist and the referenced columns must be its PRIMARY KEY or exactly the column set of a UNIQUE index, otherwise DDL fails with 0A000 (foreign key mismatch); a dangling reference is 23503 on INSERT, multi-row INSERT, UPDATE, and parent DELETE/UPDATE, and COMMIT re-validates the transaction's final state so a staged child insert and a staged parent delete cannot both land; MATCH SIMPLE (a NULL component skips the check) and self-references are supported; ON DELETE/ON UPDATE CASCADE, SET NULL, SET DEFAULT, and DEFERRABLE are rejected with 0A000; reported in sqlite_master/sqlite_schema table SQL only (pg_constraint and information_schema foreign keys are not emulated)"},
+			ForeignKey:    Feature{Supported: true, Status: "enforced", SQLState: "23503", Notes: "inline REFERENCES and table-level FOREIGN KEY (named or unnamed, single or multi-column) are accepted in CREATE TABLE; the parent table must already exist and the referenced columns must be its PRIMARY KEY or exactly the column set of a UNIQUE index, otherwise DDL fails with 0A000 (foreign key mismatch); a dangling reference is 23503 on INSERT, multi-row INSERT, UPDATE, and parent DELETE/UPDATE, and COMMIT re-validates the transaction's final state so a staged child insert and a staged parent delete cannot both land; MATCH SIMPLE (a NULL component skips the check) and self-references are supported; the MATCH keyword is not parsed, so MATCH SIMPLE is the only behavior; ON DELETE/ON UPDATE CASCADE, SET NULL, SET DEFAULT, and DEFERRABLE are rejected with 0A000 (NO ACTION and RESTRICT are accepted and enforced immediately); foreign keys are always enforced, so PRAGMA foreign_keys reports 1 and rejects OFF (SQLite defaults it to OFF) and PRAGMA foreign_key_list is not emulated; ALTER TABLE DROP/RENAME COLUMN and RENAME TABLE are rejected with 0A000 when a constraint depends on the column or names the table as its parent, and DROP TABLE of a referenced parent (including a self-reference) is rejected with 0A000; writes touching a participating table serialize on one database-wide foreign-key write gate; the clause is reported in sqlite_master/sqlite_schema table SQL, including declared ON DELETE/ON UPDATE actions, but pg_constraint and information_schema foreign keys are not emulated"},
 			Check:         Feature{Supported: true, Status: "enforced", SQLState: "23514", Notes: "inline and table-level CHECK are evaluated on INSERT, multi-row INSERT, UPDATE, and ON CONFLICT DO UPDATE; NULL/unknown results pass; predicates are restricted to deterministic, subquery-free expressions over unqualified columns; exposed in sqlite_master/sqlite_schema table SQL only (pg_constraint and information_schema.check_constraints are not emulated)"},
 			DefaultValues: Feature{Supported: true, Status: "evaluated_at_write_time", SQLState: "0A000", Notes: "constant defaults are stored as their evaluated value; CURRENT_TIMESTAMP, CURRENT_DATE, CURRENT_TIME, datetime('now'), date('now'), and time('now') are stored as SQL text and evaluated once per statement on write (once per ALTER statement for an ADD COLUMN backfill), so they are never frozen at DDL time; any other non-constant default is rejected with 0A000"},
 		},
@@ -177,8 +177,8 @@ func Current() *Document {
 		DDL: DDL{
 			AddColumn:              Feature{Supported: true, Status: "materializes_defaults", Notes: "adds one nullable column; a constant DEFAULT is written into every existing row, and a whitelisted clock default is evaluated once per ALTER statement, so the rewrite cost is O(rows) and holds the table's exclusive gate"},
 			AddColumnBackfill:      Feature{Supported: true, Status: "rewrites_rows", Notes: "a DEFAULT is materialized into existing rows during the rewrite: constants as their stored value, whitelisted clock defaults evaluated at one instant per ALTER statement; ADD COLUMN ... NOT NULL without a DEFAULT fails with 23502 on a non-empty table, ADD COLUMN ... UNIQUE conflicts with 23505 when backfilling a non-NULL default into more than one row (a constant default is predicted before the rewrite, a clock default is validated against the values the rewrite writes), ADD COLUMN ... CHECK fails with 23514 when an existing row or the backfilled default violates the predicate, and the column is published together with its CHECK constraints and UNIQUE index in one atomic schema write"},
-			DropColumn:             Feature{Supported: true, Status: "rewrites_rows", Notes: "rejected with 0A000 when a persisted CHECK constraint references the column; the stored predicate text cannot be rewritten"},
-			RenameColumn:           Feature{Supported: true, Status: "rewrites_rows", Notes: "rejected with 0A000 when a persisted CHECK constraint or expression index references the column; the stored expression text cannot be rewritten"},
+			DropColumn:             Feature{Supported: true, Status: "rewrites_rows", Notes: "rejected with 0A000 when a persisted CHECK constraint references the column or a FOREIGN KEY depends on it (child column or referenced parent column); the stored constraint text cannot be rewritten"},
+			RenameColumn:           Feature{Supported: true, Status: "rewrites_rows", Notes: "rejected with 0A000 when a persisted CHECK constraint, expression index, or FOREIGN KEY depends on the column (child column or referenced parent column); the stored constraint text cannot be rewritten"},
 			RenameColumnIndexes:    Feature{Supported: true, Status: "rewrites_index_metadata", Notes: "column references in index definitions are rewritten in place, including composite and unique indexes; index names are unchanged; RENAME COLUMN is rejected when an expression index references the column (drop that index, rename, then recreate it)"},
 			CreateTableIfNotExists: true,
 			AddColumnIfNotExists:   true,

pkg/capabilities/capabilities_test.go

@@ -163,7 +163,11 @@ func TestForeignKeyIsAdvertised(t *testing.T) {
 	if fk.SQLState != pgerror.CodeForeignKeyViolation {
 		t.Fatalf("foreign_key SQLSTATE = %q, want %q", fk.SQLState, pgerror.CodeForeignKeyViolation)
 	}
-	for _, want := range []string{"UNIQUE", "COMMIT", "MATCH SIMPLE", "self", "CASCADE", "DEFERRABLE", "0A000", "pg_constraint"} {
+	for _, want := range []string{
+		"UNIQUE", "COMMIT", "MATCH SIMPLE", "MATCH keyword", "self", "CASCADE", "DEFERRABLE", "0A000",
+		"PRAGMA foreign_keys", "rejects OFF", "foreign_key_list", "database-wide foreign-key write gate",
+		"DROP/RENAME COLUMN", "DROP TABLE", "pg_constraint",
+	} {
 		if !strings.Contains(fk.Notes, want) {
 			t.Fatalf("foreign_key notes = %q, want it to mention %q", fk.Notes, want)
 		}

pkg/executor/executor.go

@@ -3724,6 +3724,14 @@ func (e *Executor) executeCreateTable(stmt *parser.CreateTableStmt) (*Result, er
 		return nil, fmt.Errorf("table already exists: %s", stmt.Table.Name)
 	}
 
+	// Creating a table can add a FOREIGN KEY to the graph. Serialize the DDL
+	// with point writes so a concurrent child insert or parent delete cannot
+	// validate against a half-published constraint.
+	if e.table != nil {
+		unlockFK := e.table.LockForeignKeyWriteGate()
+		defer unlockFK()
+	}
+
 	// Build schema
 	schema := &storage.Schema{
 		Name: stmt.Table.Name,
@@ -3988,6 +3996,14 @@ func conflictActionName(action parser.ConflictAction) string {
 
 // executeDropTable executes a DROP TABLE statement.
 func (e *Executor) executeDropTable(stmt *parser.DropTableStmt) (*Result, error) {
+	// Dropping a table can remove the parent or child side of a FOREIGN KEY.
+	// Serialize the DDL with point writes so a concurrent insert or delete cannot
+	// validate against a constraint that is being removed.
+	if e.table != nil {
+		unlockFK := e.table.LockForeignKeyWriteGate()
+		defer unlockFK()
+	}
+
 	for _, tableRef := range stmt.Tables {
 		if !e.schema.TableExists(tableRef.Name) {
 			if stmt.IfExists {
@@ -3995,6 +4011,9 @@ func (e *Executor) executeDropTable(stmt *parser.DropTableStmt) (*Result, error)
 			}
 			return nil, fmt.Errorf("table not found: %s", tableRef.Name)
 		}
+		if err := e.rejectForeignKeyDependentTableDrop(tableRef.Name); err != nil {
+			return nil, err
+		}
 
 		// First, drop all indexes associated with this table
 		indexes, _ := e.schema.ListTableIndexes(tableRef.Name)
@@ -4367,6 +4386,9 @@ func (e *Executor) executeAlterTableDropColumn(table string, action *parser.Drop
 	if err := e.rejectCheckColumnChange(table, action.Column, "drop"); err != nil {
 		return nil, err
 	}
+	if err := e.rejectForeignKeyColumnChange(table, action.Column, "drop"); err != nil {
+		return nil, err
+	}
 	if err := e.schema.DropColumn(table, action.Column); err != nil {
 		return nil, err
 	}
@@ -4380,6 +4402,9 @@ func (e *Executor) executeAlterTableDropColumn(table string, action *parser.Drop
 
 // executeAlterTableRename renames a table.
 func (e *Executor) executeAlterTableRename(table string, action *parser.RenameTableAction) (*Result, error) {
+	if err := e.rejectForeignKeyTableRename(table); err != nil {
+		return nil, err
+	}
 	if err := e.schema.RenameTable(table, action.NewName); err != nil {
 		return nil, err
 	}
@@ -4401,6 +4426,9 @@ func (e *Executor) executeAlterTableRenameColumn(table string, action *parser.Re
 	if err := e.rejectCheckColumnChange(table, action.OldName, "rename"); err != nil {
 		return nil, err
 	}
+	if err := e.rejectForeignKeyColumnChange(table, action.OldName, "rename"); err != nil {
+		return nil, err
+	}
 	if err := e.schema.RenameColumn(table, action.OldName, action.NewName); err != nil {
 		return nil, err
 	}
@@ -4813,17 +4841,17 @@ func (e *Executor) executePragma(stmt *parser.PragmaStmt) (*Result, error) {
 		}
 		if ref, ok := stmt.Value.(*parser.ColumnRef); ok && (strings.EqualFold(ref.Column, "on") || strings.EqualFold(ref.Column, "off")) {
 			if strings.EqualFold(ref.Column, "off") {
-				return nil, fmt.Errorf("feature not supported: foreign keys are always enforced; PRAGMA foreign_keys = OFF is not supported")
+				return nil, pgerror.Unsupported("foreign keys are always enforced; PRAGMA foreign_keys = OFF is not supported (SQLite defaults it to OFF)")
 			}
 			return result, nil
 		}
 		if literal, ok := stmt.Value.(*parser.LiteralExpr); ok && (literal.Value == "1" || literal.Value == "0") {
 			if literal.Value == "0" {
-				return nil, fmt.Errorf("feature not supported: foreign keys are always enforced; PRAGMA foreign_keys = OFF is not supported")
+				return nil, pgerror.Unsupported("foreign keys are always enforced; PRAGMA foreign_keys = OFF is not supported (SQLite defaults it to OFF)")
 			}
 			return result, nil
 		}
-		return nil, fmt.Errorf("feature not supported: PRAGMA foreign_keys accepts ON or OFF")
+		return nil, pgerror.Unsupported("PRAGMA foreign_keys accepts ON or OFF")
 	default:
 		return nil, fmt.Errorf("unknown pragma: %s", stmt.Name)
 	}

pkg/executor/foreign_key_concurrency_test.go

@@ -0,0 +1,66 @@
+package executor
+
+import (
+	"fmt"
+	"sync"
+	"testing"
+)
+
+// A child insert and a parent delete that run concurrently must never both
+// commit, and the table set must never contain a dangling child reference.
+// Both statements take the foreign-key write gate, so exactly one wins; if that
+// ever regresses, the loser is not a foreign-key violation or the invariant scan
+// finds an orphan.
+func TestForeignKeyConcurrentChildInsertAndParentDelete(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	inserter := newExec(schema, table)
+	deleter := newExec(schema, table)
+	execMust(t, inserter, "CREATE TABLE parent (id INTEGER PRIMARY KEY)")
+	execMust(t, inserter, "CREATE TABLE child (id INTEGER PRIMARY KEY, parent_id INTEGER REFERENCES parent(id))")
+
+	const rounds = 100
+	for round := 1; round <= rounds; round++ {
+		execMust(t, inserter, fmt.Sprintf("INSERT INTO parent (id) VALUES (%d)", round))
+
+		start := make(chan struct{})
+		insertDone := make(chan error, 1)
+		deleteDone := make(chan error, 1)
+		var wg sync.WaitGroup
+		wg.Add(2)
+		go func() {
+			defer wg.Done()
+			<-start
+			_, err := execSQL(inserter, fmt.Sprintf("INSERT INTO child (id, parent_id) VALUES (%d, %d)", round, round))
+			insertDone <- err
+		}()
+		go func() {
+			defer wg.Done()
+			<-start
+			_, err := execSQL(deleter, fmt.Sprintf("DELETE FROM parent WHERE id = %d", round))
+			deleteDone <- err
+		}()
+		close(start)
+		wg.Wait()
+		insertErr := <-insertDone
+		deleteErr := <-deleteDone
+
+		switch {
+		case insertErr == nil && deleteErr == nil:
+			t.Fatalf("round %d: child insert and parent delete both committed", round)
+		case insertErr == nil:
+			requireForeignKeyViolation(t, deleteErr)
+		case deleteErr == nil:
+			requireForeignKeyViolation(t, insertErr)
+		default:
+			t.Fatalf("round %d: both statements failed: insert=%v delete=%v", round, insertErr, deleteErr)
+		}
+
+		dangling := execMust(t, inserter, "SELECT count(*) FROM child c WHERE NOT EXISTS (SELECT 1 FROM parent p WHERE p.id = c.parent_id)")
+		if dangling.Rows[0][0] != int64(0) {
+			t.Fatalf("round %d: %v dangling child rows", round, dangling.Rows[0][0])
+		}
+
+		execMust(t, inserter, "DELETE FROM child")
+		execMust(t, inserter, fmt.Sprintf("DELETE FROM parent WHERE id = %d", round))
+	}
+}

pkg/executor/foreign_key_test.go

@@ -273,13 +273,21 @@ func TestForeignKeySelfReference(t *testing.T) {
 	_, err := execSQL(e, "INSERT INTO nodes (id, parent_id) VALUES (4, 99)")
 	requireForeignKeyViolation(t, err)
 
-	// A self-referencing row cannot be deleted while it points at itself.
+	// Row 1 exists and references itself, but row 2 also references it, so the
+	// delete is rejected because another row still points at it.
 	_, err = execSQL(e, "DELETE FROM nodes WHERE id = 1")
 	requireForeignKeyViolation(t, err)
 
+	// A row that only references itself can be deleted: the row and the
+	// reference leave together.
 	execMust(t, e, "INSERT INTO nodes (id, parent_id) VALUES (5, 5)")
 	execMust(t, e, "DELETE FROM nodes WHERE id = 5")
 
+	// Once the other child is gone, node 1's self-reference no longer blocks.
+	execMust(t, e, "DELETE FROM nodes WHERE id = 2")
+	execMust(t, e, "DELETE FROM nodes WHERE id = 1")
+	execMust(t, e, "INSERT INTO nodes (id, parent_id) VALUES (1, 1)")
+
 	// A composite self-reference resolves against this table's table-level key.
 	execMust(t, e, `CREATE TABLE tree (
 		a TEXT NOT NULL,
@@ -526,6 +534,19 @@ func TestForeignKeyInSQLiteCatalog(t *testing.T) {
 	if res.RowCount != 1 || !strings.Contains(res.Rows[0][0].(string), "FOREIGN KEY") {
 		t.Fatalf("sqlite_schema SQL = %v, want the foreign key", res.Rows)
 	}
+
+	// Declared referential actions round-trip through the catalog text.
+	execMust(t, e, `CREATE TABLE child_actions (
+		id INTEGER PRIMARY KEY,
+		parent_id INTEGER,
+		CONSTRAINT child_parent_actions FOREIGN KEY (parent_id) REFERENCES parent(id) ON DELETE RESTRICT ON UPDATE NO ACTION
+	)`)
+	res = execMust(t, e, "SELECT sql FROM sqlite_master WHERE name = 'child_actions'")
+	sql, _ = res.Rows[0][0].(string)
+	want := "CONSTRAINT `child_parent_actions` FOREIGN KEY (`parent_id`) REFERENCES `parent` (`id`) ON DELETE RESTRICT ON UPDATE NO ACTION"
+	if !strings.Contains(sql, want) {
+		t.Fatalf("catalog SQL %q does not contain %q", sql, want)
+	}
 }
 
 // Foreign keys coexist with the composite primary key, CHECK, and default
@@ -636,3 +657,148 @@ func TestForeignKeyInsertSelect(t *testing.T) {
 		t.Fatalf("child rows after valid INSERT ... SELECT = %v, want 1", res.Rows)
 	}
 }
+
+// Referential integrity is a final-state property: a parent and its child can be
+// inserted in one transaction, and the commit validates both.
+func TestForeignKeyTransactionInsertParentAndChild(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE parent (id INTEGER PRIMARY KEY)")
+	execMust(t, e, "CREATE TABLE child (id INTEGER PRIMARY KEY, parent_id INTEGER REFERENCES parent(id))")
+
+	execMust(t, e, "BEGIN")
+	execMust(t, e, "INSERT INTO parent (id) VALUES (1)")
+	execMust(t, e, "INSERT INTO child (id, parent_id) VALUES (10, 1)")
+	execMust(t, e, "COMMIT")
+
+	if res := execMust(t, e, "SELECT count(*) FROM child"); res.Rows[0][0] != int64(1) {
+		t.Fatalf("child rows = %v, want 1", res.Rows)
+	}
+}
+
+// Rows of one multi-row self-referential INSERT may reference each other: the
+// statement is validated against its own final row set.
+func TestForeignKeySiblingReferencesInOneInsert(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE nodes (id INTEGER PRIMARY KEY, parent_id INTEGER REFERENCES nodes(id))")
+
+	execMust(t, e, "INSERT INTO nodes (id, parent_id) VALUES (1, 2), (2, 1)")
+	if res := execMust(t, e, "SELECT count(*) FROM nodes"); res.Rows[0][0] != int64(2) {
+		t.Fatalf("nodes = %v, want 2", res.Rows)
+	}
+
+	_, err := execSQL(e, "INSERT INTO nodes (id, parent_id) VALUES (3, 4), (4, 99)")
+	requireForeignKeyViolation(t, err)
+	if res := execMust(t, e, "SELECT count(*) FROM nodes"); res.Rows[0][0] != int64(2) {
+		t.Fatalf("rejected sibling insert left rows behind: %v", res.Rows)
+	}
+}
+
+// INSERT OR REPLACE rebuilds the conflicting row inside one implicit
+// transaction, so replacing a referenced parent row with the same key is valid
+// while a replacement that moves the key orphans children and fails at commit.
+func TestForeignKeyInsertOrReplaceOnReferencedParent(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE parent (id INTEGER PRIMARY KEY, code TEXT UNIQUE)")
+	execMust(t, e, "CREATE TABLE child (id INTEGER PRIMARY KEY, parent_id INTEGER REFERENCES parent(id))")
+	execMust(t, e, "INSERT INTO parent (id, code) VALUES (1, 'a')")
+	execMust(t, e, "INSERT INTO child (id, parent_id) VALUES (10, 1)")
+
+	// Same key: the child keeps a valid parent, so the replacement commits.
+	execMust(t, e, "INSERT OR REPLACE INTO parent (id, code) VALUES (1, 'b')")
+	if res := execMust(t, e, "SELECT count(*) FROM child WHERE parent_id = 1"); res.Rows[0][0] != int64(1) {
+		t.Fatalf("child rows after in-place replace = %v, want 1", res.Rows)
+	}
+
+	// A unique-code conflict replaces row 1 with row 2, orphaning the child.
+	_, err := execSQL(e, "INSERT OR REPLACE INTO parent (id, code) VALUES (2, 'b')")
+	requireForeignKeyViolation(t, err)
+	if res := execMust(t, e, "SELECT id, code FROM parent"); res.RowCount != 1 || res.Rows[0][0] != int64(1) || res.Rows[0][1] != "b" {
+		t.Fatalf("parent after rejected replace = %v, want the original row", res.Rows)
+	}
+	if res := execMust(t, e, "SELECT count(*) FROM child"); res.Rows[0][0] != int64(1) {
+		t.Fatalf("child rows after rejected replace = %v, want 1", res.Rows)
+	}
+}
+
+// A UNIQUE parent index is matched as a column set, and the values are matched in
+// declaration order: FOREIGN KEY (a, b) REFERENCES parent(y, x) compares a to y
+// and b to x.
+func TestForeignKeyReorderedUniqueTarget(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE parent (x INTEGER, y INTEGER, UNIQUE (x, y))")
+	execMust(t, e, "CREATE TABLE child (a INTEGER, b INTEGER, FOREIGN KEY (a, b) REFERENCES parent (y, x))")
+	execMust(t, e, "INSERT INTO parent (x, y) VALUES (1, 2)")
+
+	execMust(t, e, "INSERT INTO child (a, b) VALUES (2, 1)")
+	_, err := execSQL(e, "INSERT INTO child (a, b) VALUES (1, 2)")
+	requireForeignKeyViolation(t, err)
+	_, err = execSQL(e, "DELETE FROM parent")
+	requireForeignKeyViolation(t, err)
+}
+
+// ALTER TABLE rejects changes that would strand a persisted foreign key. The
+// metadata stores table and column names verbatim, so it cannot be rewritten.
+func TestForeignKeyAlterTableRejections(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE u (code TEXT UNIQUE, note TEXT)")
+	execMust(t, e, "CREATE TABLE child (id INTEGER PRIMARY KEY, code TEXT REFERENCES u(code), memo TEXT)")
+
+	rejections := []struct {
+		name string
+		sql  string
+		want string
+	}{
+		{"drop child column", "ALTER TABLE child DROP COLUMN code", "cannot drop column code"},
+		{"rename child column", "ALTER TABLE child RENAME COLUMN code TO parent_code", "cannot rename column code"},
+		{"drop parent column", "ALTER TABLE u DROP COLUMN code", "cannot drop column code"},
+		{"rename parent column", "ALTER TABLE u RENAME COLUMN code TO code2", "cannot rename column code"},
+		{"rename referenced table", "ALTER TABLE u RENAME TO u2", "cannot rename table u"},
+		{"drop referenced table", "DROP TABLE u", "cannot drop table u"},
+	}
+	for _, tt := range rejections {
+		t.Run(tt.name, func(t *testing.T) {
+			requireUnsupported(t, execErr(t, e, tt.sql), tt.want)
+		})
+	}
+
+	// Unrelated columns and the child table itself are not constrained.
+	execMust(t, e, "ALTER TABLE child DROP COLUMN memo")
+	execMust(t, e, "ALTER TABLE u RENAME COLUMN note TO comment")
+	execMust(t, e, "ALTER TABLE u ADD COLUMN extra TEXT")
+	execMust(t, e, "ALTER TABLE child RENAME TO child2")
+
+	execMust(t, e, "INSERT INTO u (code, comment) VALUES ('a', 'x')")
+	execMust(t, e, "INSERT INTO child2 (id, code) VALUES (1, 'a')")
+	_, err := execSQL(e, "INSERT INTO child2 (id, code) VALUES (2, 'missing')")
+	requireForeignKeyViolation(t, err)
+}
+
+// A self-referential table is both sides of its own constraint, so renaming or
+// dropping it is rejected for the same reason.
+func TestForeignKeySelfReferenceAlterRejections(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE nodes (id INTEGER PRIMARY KEY, parent_id INTEGER REFERENCES nodes(id), memo TEXT)")
+
+	requireUnsupported(t, execErr(t, e, "ALTER TABLE nodes DROP COLUMN parent_id"), "cannot drop column parent_id")
+	requireUnsupported(t, execErr(t, e, "ALTER TABLE nodes RENAME COLUMN parent_id TO parent"), "cannot rename column parent_id")
+	requireUnsupported(t, execErr(t, e, "ALTER TABLE nodes RENAME TO n2"), "cannot rename table nodes")
+	requireUnsupported(t, execErr(t, e, "DROP TABLE nodes"), "cannot drop table nodes")
+
+	execMust(t, e, "ALTER TABLE nodes DROP COLUMN memo")
+	execMust(t, e, "INSERT INTO nodes (id, parent_id) VALUES (1, 1)")
+	_, err := execSQL(e, "INSERT INTO nodes (id, parent_id) VALUES (2, 9)")
+	requireForeignKeyViolation(t, err)
+}
+
+// execErr parses and executes SQL, returning the error (nil on success).
+func execErr(t *testing.T, e *Executor, sql string) error {
+	t.Helper()
+	_, err := execSQL(e, sql)
+	return err
+}

pkg/executor/foreign_keys.go

@@ -179,6 +179,8 @@ func (e *Executor) resolveForeignKey(table string, schema *storage.Schema, uniqu
 		Columns:    childColumns,
 		RefTable:   parent.Name,
 		RefColumns: canonical,
+		OnDelete:   declaration.onDelete.String(),
+		OnUpdate:   declaration.onUpdate.String(),
 	}, nil
 }
 
@@ -238,3 +240,57 @@ func sameColumnSet(a, b []string) bool {
 	}
 	return true
 }
+
+// rejectForeignKeyColumnChange refuses to drop or rename a column that a
+// persisted foreign key depends on, on the child side or as a referenced parent
+// column. Constraint metadata stores column names verbatim and cannot be
+// rewritten, so accepting the change would either silently stop enforcing the
+// constraint or leave the catalog naming a missing column.
+func (e *Executor) rejectForeignKeyColumnChange(table, column, action string) error {
+	labels, err := e.schema.ForeignKeyColumnDependency(table, column)
+	if err != nil {
+		return err
+	}
+	if len(labels) == 0 {
+		return nil
+	}
+	return pgerror.Unsupported(
+		"cannot %s column %s: foreign key %s depends on it; drop the constrained table (or recreate it without that constraint) first",
+		action, column, strings.Join(labels, ", "),
+	)
+}
+
+// rejectForeignKeyTableRename refuses to rename a table another constraint (or
+// its own constraint) names as the parent. The referenced table name is
+// persisted verbatim, so a rename would leave parent-side enforcement pointing
+// at a table that no longer exists.
+func (e *Executor) rejectForeignKeyTableRename(table string) error {
+	labels, err := e.schema.ForeignKeyTableDependency(table)
+	if err != nil {
+		return err
+	}
+	if len(labels) == 0 {
+		return nil
+	}
+	return pgerror.Unsupported(
+		"cannot rename table %s: foreign key %s references it; drop the constrained table (or recreate it without that constraint) first",
+		table, strings.Join(labels, ", "),
+	)
+}
+
+// rejectForeignKeyDependentTableDrop refuses to drop a table that a persisted
+// foreign key names as its parent, because the constraint would be left dangling
+// on the child table.
+func (e *Executor) rejectForeignKeyDependentTableDrop(table string) error {
+	labels, err := e.schema.ForeignKeyTableDependency(table)
+	if err != nil {
+		return err
+	}
+	if len(labels) == 0 {
+		return nil
+	}
+	return pgerror.Unsupported(
+		"cannot drop table %s: foreign key %s references it; drop or recreate the table that declares that constraint first",
+		table, strings.Join(labels, ", "),
+	)
+}

pkg/executor/review_fixes_test.go

@@ -1,6 +1,7 @@
 package executor
 
 import (
+	"errors"
 	"fmt"
 	"sync"
 	"testing"
@@ -8,6 +9,7 @@ import (
 
 	"github.com/danfragoso/pizzasql-next/pkg/lexer"
 	"github.com/danfragoso/pizzasql-next/pkg/parser"
+	"github.com/danfragoso/pizzasql-next/pkg/pgerror"
 	"github.com/danfragoso/pizzasql-next/pkg/storage"
 )
 
@@ -208,6 +210,8 @@ func TestForeignKeysPragmaAlwaysOn(t *testing.T) {
 	}
 	if _, err := execSQL(e, "PRAGMA foreign_keys = OFF"); err == nil {
 		t.Fatal("expected disabling always-on foreign keys to fail")
+	} else if !errors.Is(err, pgerror.ErrFeatureNotSupported) {
+		t.Fatalf("PRAGMA foreign_keys = OFF error = %v, want ErrFeatureNotSupported", err)
 	}
 }
 

pkg/executor/sqlite_catalog.go

@@ -351,6 +351,14 @@ func recreateCreateTableSQL(s *storage.Schema) string {
 			parent[i] = quoteIdent(name)
 		}
 		clause := fmt.Sprintf("FOREIGN KEY (%s) REFERENCES %s (%s)", strings.Join(child, ", "), quoteIdent(fk.RefTable), strings.Join(parent, ", "))
+		for _, action := range []struct {
+			clause string
+			value  string
+		}{{"ON DELETE", fk.OnDelete}, {"ON UPDATE", fk.OnUpdate}} {
+			if action.value != "" {
+				clause += " " + action.clause + " " + action.value
+			}
+		}
 		if fk.Name != "" {
 			clause = fmt.Sprintf("CONSTRAINT %s %s", quoteIdent(fk.Name), clause)
 		}

pkg/pgserver/foreign_key_test.go

@@ -67,3 +67,13 @@ func TestForeignKeyParentDeleteSQLState(t *testing.T) {
 		t.Fatalf("rejected delete removed the parent: %v", values)
 	}
 }
+
+// Foreign keys are always enforced, so disabling them is rejected as an
+// unsupported feature (0A000) rather than a generic execution error.
+func TestForeignKeyPragmaOffReturnsFeatureNotSupported(t *testing.T) {
+	c, client := newDataConnection(t)
+	messages := runQuery(t, c, client, simpleQueryMessage("PRAGMA foreign_keys = OFF"))
+	if code := errorCode(messages[0]); code != ErrCodeFeatureNotSupported {
+		t.Fatalf("PRAGMA foreign_keys = OFF SQLSTATE = %q, want %q", code, ErrCodeFeatureNotSupported)
+	}
+}

pkg/storage/foreign_key.go

@@ -261,6 +261,79 @@ func (m *SchemaManager) referencingForeignKeys(table string) ([]referencingForei
 	return catalog.referencing[strings.ToLower(table)], nil
 }
 
+// foreignKeyReferenceName formats a constraint for a DDL error message. The
+// child table is included because an unnamed constraint's label is only unique
+// within its table.
+func foreignKeyReferenceName(child string, fk ForeignKeyConstraint) string {
+	return fmt.Sprintf("%s on %s", fk.Label(), child)
+}
+
+// ForeignKeyColumnDependency returns the constraints that keep a column in
+// place: a FOREIGN KEY this table declares on it, or a FOREIGN KEY (from this
+// table or another one) that references it as a parent column. DROP/RENAME
+// COLUMN cannot rewrite persisted constraint text, so a non-empty result must
+// reject the change instead of silently disabling the constraint.
+func (m *SchemaManager) ForeignKeyColumnDependency(table, column string) ([]string, error) {
+	catalog, err := m.foreignKeyCatalogSnapshot()
+	if err != nil {
+		return nil, err
+	}
+	schema, err := m.GetSchema(table)
+	if err != nil {
+		return nil, err
+	}
+	seen := make(map[string]bool)
+	var labels []string
+	add := func(child string, fk ForeignKeyConstraint) {
+		label := foreignKeyReferenceName(child, fk)
+		if seen[label] {
+			return
+		}
+		seen[label] = true
+		labels = append(labels, label)
+	}
+	for _, fk := range schema.ForeignKeys {
+		for _, name := range fk.Columns {
+			if strings.EqualFold(name, column) {
+				add(schema.Name, fk)
+				break
+			}
+		}
+	}
+	for _, ref := range catalog.referencing[strings.ToLower(table)] {
+		for _, name := range ref.FK.RefColumns {
+			if strings.EqualFold(name, column) {
+				add(ref.Table, ref.FK)
+				break
+			}
+		}
+	}
+	sort.Strings(labels)
+	return labels, nil
+}
+
+// ForeignKeyTableDependency returns the constraints that name table as their
+// referenced parent, including a self-reference. RENAME TABLE cannot rewrite
+// them, and DROP TABLE would leave them dangling.
+func (m *SchemaManager) ForeignKeyTableDependency(table string) ([]string, error) {
+	catalog, err := m.foreignKeyCatalogSnapshot()
+	if err != nil {
+		return nil, err
+	}
+	seen := make(map[string]bool)
+	var labels []string
+	for _, ref := range catalog.referencing[strings.ToLower(table)] {
+		label := foreignKeyReferenceName(ref.Table, ref.FK)
+		if seen[label] {
+			continue
+		}
+		seen[label] = true
+		labels = append(labels, label)
+	}
+	sort.Strings(labels)
+	return labels, nil
+}
+
 // ForeignKeyParticipants reports whether a write to table can affect referential
 // integrity. A table participates when it declares a FOREIGN KEY or when another
 // table references it.
@@ -328,6 +401,16 @@ func (m *TableManager) lockForeignKeyWrites(tables ...string) (func(), error) {
 	return func() {}, nil
 }
 
+// LockForeignKeyWriteGate takes the foreign-key write gate unconditionally.
+// Schema DDL uses it around CREATE TABLE and DROP TABLE: the table that gains or
+// loses a constraint is not part of the foreign-key graph yet (or has just been
+// removed from it), so the participant check would skip the gate exactly when
+// the graph is changing. Callers must take it before any table gate.
+func (m *TableManager) LockForeignKeyWriteGate() func() {
+	m.foreignKeyMu.Lock()
+	return m.foreignKeyMu.Unlock
+}
+
 // foreignKeyStateValidator validates the final row state of a set of tables
 // against every FOREIGN KEY declaration among them.
 type foreignKeyStateValidator struct {

pkg/storage/schema.go

@@ -52,6 +52,11 @@ type ForeignKeyConstraint struct {
 	Columns    []string `json:"columns"`
 	RefTable   string   `json:"ref_table"`
 	RefColumns []string `json:"ref_columns"`
+	// OnDelete and OnUpdate hold the declared referential action text (only
+	// NO ACTION and RESTRICT are accepted). They are empty when the clause was
+	// omitted and are kept so sqlite_master can reproduce the declaration.
+	OnDelete string `json:"on_delete,omitempty"`
+	OnUpdate string `json:"on_update,omitempty"`
 }
 
 // Label identifies the constraint in error messages: the declared name when
@@ -507,6 +512,8 @@ func cloneSchema(schema *Schema) *Schema {
 				Columns:    append([]string(nil), fk.Columns...),
 				RefTable:   fk.RefTable,
 				RefColumns: append([]string(nil), fk.RefColumns...),
+				OnDelete:   fk.OnDelete,
+				OnUpdate:   fk.OnUpdate,
 			}
 		}
 	}