Merge pull request #35 from database-pizza/engine/b-foreign-key

feat: enforce FOREIGN KEY constraints (immediate, no cascades)

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: false, Status: "rejected", SQLState: "0A000", Notes: "inline and table-level FOREIGN KEY are rejected; enforce referential integrity in the application, cascades are not implemented"},
+			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,
@@ -224,6 +224,7 @@ func Current() *Document {
 		ErrorCodes: map[string]string{
 			"23502": "not_null_violation",
 			"23505": "unique_violation",
+			"23503": "foreign_key_violation",
 			"23514": "check_violation",
 			"40001": "serialization_failure",
 			"0A000": "feature_not_supported",

pkg/capabilities/capabilities_test.go

@@ -147,3 +147,32 @@ func TestCompositePrimaryKeyIsAdvertised(t *testing.T) {
 		t.Fatalf("document does not advertise composite_primary_key as supported: %s", body)
 	}
 }
+
+// Foreign keys are advertised as enforced, with the SQLSTATE callers should
+// expect for a dangling reference and the explicit exclusions for cascades and
+// deferred enforcement.
+func TestForeignKeyIsAdvertised(t *testing.T) {
+	doc := Current()
+	fk := doc.Constraints.ForeignKey
+	if !fk.Supported {
+		t.Fatalf("foreign_key = %+v, want supported", fk)
+	}
+	if fk.Status != "enforced" {
+		t.Fatalf("foreign_key status = %q, want enforced", fk.Status)
+	}
+	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", "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)
+		}
+	}
+	if got := doc.ErrorCodes[pgerror.CodeForeignKeyViolation]; got != "foreign_key_violation" {
+		t.Fatalf("23503 mapping = %q, want foreign_key_violation", got)
+	}
+}

pkg/executor/check_fk_test.go

@@ -1,29 +0,0 @@
-package executor
-
-import (
-	"errors"
-	"testing"
-
-	"github.com/danfragoso/pizzasql-next/pkg/pgerror"
-)
-
-func TestForeignKeyConstraintsRejectedAtDDL(t *testing.T) {
-	_, schema, table := newTestDB(t)
-	e := newExec(schema, table)
-	execMust(t, e, "CREATE TABLE parent (id INTEGER PRIMARY KEY)")
-
-	for _, sql := range []string{
-		"CREATE TABLE child1 (id INTEGER PRIMARY KEY, parent_id INTEGER REFERENCES parent(id))",
-		"CREATE TABLE child2 (id INTEGER PRIMARY KEY, parent_id INTEGER, FOREIGN KEY (parent_id) REFERENCES parent(id))",
-		"CREATE TABLE child3 (id INTEGER PRIMARY KEY, parent_id INTEGER, CONSTRAINT child_parent FOREIGN KEY (parent_id) REFERENCES parent(id) ON DELETE CASCADE)",
-	} {
-		if _, err := execSQL(e, sql); !errors.Is(err, pgerror.ErrFeatureNotSupported) {
-			t.Fatalf("%s: error = %v, want ErrFeatureNotSupported", sql, err)
-		}
-	}
-
-	execMust(t, e, "CREATE TABLE t (id INTEGER PRIMARY KEY)")
-	if _, err := execSQL(e, "ALTER TABLE t ADD COLUMN parent_id INTEGER REFERENCES parent(id)"); !errors.Is(err, pgerror.ErrFeatureNotSupported) {
-		t.Fatalf("ALTER FK error = %v, want ErrFeatureNotSupported", err)
-	}
-}

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,
@@ -3781,19 +3789,17 @@ func (e *Executor) executeCreateTable(stmt *parser.CreateTableStmt) (*Result, er
 				// Collected after every column is known, because a table-level
 				// predicate may reference a later column.
 			case parser.ConstraintForeignKey:
-				return nil, pgerror.Unsupported(
-					"FOREIGN KEY constraints are not supported: column %q declares REFERENCES %s; enforce referential integrity in the application (the engine never validated foreign keys)",
-					col.Name, constraint.RefTable,
-				)
+				// Resolved once every column and the primary key are known, so a
+				// self-reference sees this table's final key.
 			}
 		}
 
 		schema.Columns = append(schema.Columns, col)
 	}
 
-	// Handle table-level constraints. CHECK predicates are persisted and
-	// enforced; FOREIGN KEY is still rejected because the engine never validates
-	// referential integrity.
+	// Handle table-level constraints. CHECK predicates and FOREIGN KEY
+	// declarations are persisted and enforced; foreign keys are resolved after
+	// every column and the primary key are known.
 	var tablePrimaryKeyColumns []string
 	for _, constraint := range stmt.Constraints {
 		switch constraint.Type {
@@ -3840,9 +3846,7 @@ func (e *Executor) executeCreateTable(stmt *parser.CreateTableStmt) (*Result, er
 		case parser.ConstraintCheck:
 			// Handled with the column-level predicates once every column is known.
 		case parser.ConstraintForeignKey:
-			return nil, pgerror.Unsupported(
-				"FOREIGN KEY constraints are not supported: enforce referential integrity in the application instead (the engine never validated foreign keys)",
-			)
+			// Resolved with the inline REFERENCES constraints below.
 		}
 	}
 
@@ -3852,6 +3856,11 @@ func (e *Executor) executeCreateTable(stmt *parser.CreateTableStmt) (*Result, er
 			stmt.Table.Name, strings.Join(inlinePrimaryKeyColumns, ", "),
 		)
 	}
+	if len(inlinePrimaryKeyColumns) == 1 {
+		// Mirror the catalog's normalization so foreign-key validation below can
+		// resolve a self-reference against this table's primary key.
+		schema.PrimaryKey = inlinePrimaryKeyColumns[0]
+	}
 	if len(tablePrimaryKeyColumns) > 0 {
 		schema.PrimaryKeyColumns = tablePrimaryKeyColumns
 		for i := range schema.Columns {
@@ -3887,6 +3896,12 @@ func (e *Executor) executeCreateTable(stmt *parser.CreateTableStmt) (*Result, er
 	}
 	schema.Checks = checks
 
+	foreignKeys, err := e.createTableForeignKeys(stmt, schema, uniqIndexes)
+	if err != nil {
+		return nil, err
+	}
+	schema.ForeignKeys = foreignKeys
+
 	if err := e.schema.CreateTable(schema); err != nil {
 		if stmt.IfNotExists && strings.Contains(err.Error(), "table already exists") {
 			return NewResult("CREATE TABLE"), nil
@@ -3981,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 {
@@ -3988,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)
@@ -4283,8 +4309,8 @@ func (e *Executor) executeAlterTableAddColumn(table string, action *parser.AddCo
 			checkConstraints = append(checkConstraints, constraint)
 		case parser.ConstraintForeignKey:
 			return nil, pgerror.Unsupported(
-				"FOREIGN KEY constraints are not supported: column %q declares REFERENCES %s; enforce referential integrity in the application instead",
-				col.Name, constraint.RefTable,
+				"ALTER TABLE ADD COLUMN cannot declare REFERENCES %s on column %q; declare the foreign key in CREATE TABLE",
+				constraint.RefTable, col.Name,
 			)
 		}
 	}
@@ -4360,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
 	}
@@ -4373,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
 	}
@@ -4394,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
 	}
@@ -4795,19 +4830,28 @@ func (e *Executor) executePragma(stmt *parser.PragmaStmt) (*Result, error) {
 	case "version":
 		return e.pragmaVersion()
 	case "foreign_keys":
+		// Foreign keys are always enforced, so the pragma reports them as on.
+		// Enabling is a no-op; disabling would require silently dropping
+		// constraint enforcement, which the engine does not offer.
 		result := NewResult("PRAGMA")
 		if stmt.Value == nil {
 			result.AddColumn("foreign_keys")
-			result.AddRow(int64(0))
+			result.AddRow(int64(1))
 			return result, nil
 		}
-		if ref, ok := stmt.Value.(*parser.ColumnRef); ok && strings.EqualFold(ref.Column, "off") {
+		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, 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 == "0" {
+		if literal, ok := stmt.Value.(*parser.LiteralExpr); ok && (literal.Value == "1" || literal.Value == "0") {
+			if literal.Value == "0" {
+				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("PizzaSQL does not enforce foreign keys; only PRAGMA foreign_keys = OFF is supported")
+		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

@@ -0,0 +1,804 @@
+package executor
+
+import (
+	"errors"
+	"strings"
+	"testing"
+
+	"github.com/danfragoso/pizzasql-next/pkg/pgerror"
+	"github.com/danfragoso/pizzasql-next/pkg/storage"
+)
+
+// requireForeignKeyViolation asserts err is a foreign-key violation and returns
+// its label so a test can check which constraint fired.
+func requireForeignKeyViolation(t *testing.T, err error) *storage.ForeignKeyViolationError {
+	t.Helper()
+	var violation *storage.ForeignKeyViolationError
+	if !errors.As(err, &violation) {
+		t.Fatalf("error = %v, want ForeignKeyViolationError", err)
+	}
+	if !strings.HasPrefix(err.Error(), "FOREIGN KEY constraint failed") {
+		t.Fatalf("message = %q, want the FOREIGN KEY failure prefix", err.Error())
+	}
+	return violation
+}
+
+func requireUnsupported(t *testing.T, err error, want string) {
+	t.Helper()
+	if err == nil {
+		t.Fatalf("expected an error containing %q, got nil", want)
+	}
+	if !errors.Is(err, pgerror.ErrFeatureNotSupported) {
+		t.Fatalf("error = %v, want ErrFeatureNotSupported (0A000)", err)
+	}
+	if !strings.Contains(err.Error(), want) {
+		t.Fatalf("error = %q, want it to contain %q", err, want)
+	}
+}
+
+// Inline REFERENCES, table-level FOREIGN KEY, and the named forms all persist
+// and enforce the same constraint.
+func TestForeignKeyDeclarationForms(t *testing.T) {
+	tests := []struct {
+		name       string
+		ddl        string
+		columns    string
+		refColumns string
+	}{
+		{
+			name:       "inline",
+			ddl:        "CREATE TABLE child (id INTEGER PRIMARY KEY, parent_id INTEGER REFERENCES parent(id))",
+			columns:    "parent_id",
+			refColumns: "id",
+		},
+		{
+			name:       "inline without column list uses the parent primary key",
+			ddl:        "CREATE TABLE child (id INTEGER PRIMARY KEY, parent_id INTEGER REFERENCES parent)",
+			columns:    "parent_id",
+			refColumns: "id",
+		},
+		{
+			name:       "table level",
+			ddl:        "CREATE TABLE child (id INTEGER PRIMARY KEY, parent_id INTEGER, FOREIGN KEY (parent_id) REFERENCES parent(id))",
+			columns:    "parent_id",
+			refColumns: "id",
+		},
+		{
+			name:       "named table level",
+			ddl:        "CREATE TABLE child (id INTEGER PRIMARY KEY, parent_id INTEGER, CONSTRAINT child_parent FOREIGN KEY (parent_id) REFERENCES parent(id) ON DELETE RESTRICT ON UPDATE NO ACTION)",
+			columns:    "parent_id",
+			refColumns: "id",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			_, schema, table := newTestDB(t)
+			e := newExec(schema, table)
+			execMust(t, e, "CREATE TABLE parent (id INTEGER PRIMARY KEY)")
+			execMust(t, e, tt.ddl)
+
+			stored, err := schema.GetSchema("child")
+			if err != nil {
+				t.Fatalf("get schema: %v", err)
+			}
+			if len(stored.ForeignKeys) != 1 {
+				t.Fatalf("foreign keys = %+v, want exactly one", stored.ForeignKeys)
+			}
+			fk := stored.ForeignKeys[0]
+			if strings.Join(fk.Columns, ",") != tt.columns {
+				t.Fatalf("columns = %v, want %s", fk.Columns, tt.columns)
+			}
+			if fk.RefTable != "parent" {
+				t.Fatalf("ref table = %q, want parent", fk.RefTable)
+			}
+			if strings.Join(fk.RefColumns, ",") != tt.refColumns {
+				t.Fatalf("ref columns = %v, want %s", fk.RefColumns, tt.refColumns)
+			}
+
+			execMust(t, e, "INSERT INTO parent (id) VALUES (1)")
+			execMust(t, e, "INSERT INTO child (id, parent_id) VALUES (10, 1)")
+			execMust(t, e, "INSERT INTO child (id, parent_id) VALUES (11, NULL)")
+			_, err = execSQL(e, "INSERT INTO child (id, parent_id) VALUES (12, 2)")
+			requireForeignKeyViolation(t, err)
+		})
+	}
+}
+
+// A composite foreign key matches the parent key component by component.
+func TestForeignKeyComposite(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE parent (a TEXT, b INTEGER, PRIMARY KEY (a, b))")
+	execMust(t, e, "CREATE TABLE child (id INTEGER PRIMARY KEY, pa TEXT, pb INTEGER, FOREIGN KEY (pa, pb) REFERENCES parent(a, b))")
+
+	stored, err := schema.GetSchema("child")
+	if err != nil {
+		t.Fatalf("get schema: %v", err)
+	}
+	if len(stored.ForeignKeys) != 1 || strings.Join(stored.ForeignKeys[0].RefColumns, ",") != "a,b" {
+		t.Fatalf("foreign keys = %+v, want (pa,pb) -> parent(a,b)", stored.ForeignKeys)
+	}
+
+	execMust(t, e, "INSERT INTO parent (a, b) VALUES ('x', 1)")
+	execMust(t, e, "INSERT INTO child (id, pa, pb) VALUES (1, 'x', 1)")
+
+	_, err = execSQL(e, "INSERT INTO child (id, pa, pb) VALUES (2, 'x', 2)")
+	requireForeignKeyViolation(t, err)
+	_, err = execSQL(e, "INSERT INTO child (id, pa, pb) VALUES (3, 'x', 1.5)")
+	requireForeignKeyViolation(t, err)
+
+	// Any NULL component disables the constraint (MATCH SIMPLE).
+	execMust(t, e, "INSERT INTO child (id, pa, pb) VALUES (4, NULL, 1)")
+	execMust(t, e, "INSERT INTO child (id, pa, pb) VALUES (5, 'x', NULL)")
+
+	if res := execMust(t, e, "SELECT count(*) FROM child"); res.Rows[0][0] != int64(3) {
+		t.Fatalf("child rows = %v, want 3", res.Rows)
+	}
+}
+
+// A child INSERT with a missing parent is rejected before anything is written.
+func TestForeignKeyInsertViolation(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))")
+
+	_, err := execSQL(e, "INSERT INTO child (id, parent_id) VALUES (1, 42)")
+	requireForeignKeyViolation(t, err)
+	if res := execMust(t, e, "SELECT count(*) FROM child"); res.Rows[0][0] != int64(0) {
+		t.Fatalf("child rows after rejected insert = %v, want 0", res.Rows)
+	}
+
+	execMust(t, e, "INSERT INTO parent (id) VALUES (42)")
+	execMust(t, e, "INSERT INTO child (id, parent_id) VALUES (1, 42)")
+}
+
+// A multi-row INSERT is atomic: one dangling row rejects the whole statement.
+func TestForeignKeyMultiRowInsertIsAtomic(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, "INSERT INTO parent (id) VALUES (1)")
+
+	_, err := execSQL(e, "INSERT INTO child (id, parent_id) VALUES (10, 1), (11, 2), (12, 1)")
+	requireForeignKeyViolation(t, err)
+	if res := execMust(t, e, "SELECT count(*) FROM child"); res.Rows[0][0] != int64(0) {
+		t.Fatalf("child rows after rejected multi-row insert = %v, want 0", res.Rows)
+	}
+
+	execMust(t, e, "INSERT INTO child (id, parent_id) VALUES (10, 1), (12, 1)")
+	if res := execMust(t, e, "SELECT count(*) FROM child"); res.Rows[0][0] != int64(2) {
+		t.Fatalf("child rows = %v, want 2", res.Rows)
+	}
+}
+
+// Updating the child foreign-key column to a missing parent is rejected, and a
+// valid update is accepted.
+func TestForeignKeyChildUpdate(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, "INSERT INTO parent (id) VALUES (1), (2)")
+	execMust(t, e, "INSERT INTO child (id, parent_id) VALUES (10, 1)")
+
+	_, err := execSQL(e, "UPDATE child SET parent_id = 99 WHERE id = 10")
+	requireForeignKeyViolation(t, err)
+	res := execMust(t, e, "SELECT parent_id FROM child WHERE id = 10")
+	if res.Rows[0][0] != int64(1) {
+		t.Fatalf("parent_id after rejected update = %v, want 1", res.Rows[0][0])
+	}
+
+	execMust(t, e, "UPDATE child SET parent_id = 2 WHERE id = 10")
+	res = execMust(t, e, "SELECT parent_id FROM child WHERE id = 10")
+	if res.Rows[0][0] != int64(2) {
+		t.Fatalf("parent_id after valid update = %v, want 2", res.Rows[0][0])
+	}
+
+	// The same check applies to the update half of an upsert.
+	_, err = execSQL(e, "INSERT INTO child (id, parent_id) VALUES (10, 99) ON CONFLICT (id) DO UPDATE SET parent_id = excluded.parent_id")
+	requireForeignKeyViolation(t, err)
+	execMust(t, e, "INSERT INTO child (id, parent_id) VALUES (10, 1) ON CONFLICT (id) DO UPDATE SET parent_id = excluded.parent_id")
+	res = execMust(t, e, "SELECT parent_id FROM child WHERE id = 10")
+	if res.Rows[0][0] != int64(1) {
+		t.Fatalf("parent_id after upsert = %v, want 1", res.Rows[0][0])
+	}
+}
+
+// Deleting a referenced parent is rejected; deleting the child first makes it
+// possible, and a parent with no children can always be deleted.
+func TestForeignKeyParentDelete(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, "INSERT INTO parent (id) VALUES (1), (2)")
+	execMust(t, e, "INSERT INTO child (id, parent_id) VALUES (10, 1)")
+
+	_, err := execSQL(e, "DELETE FROM parent WHERE id = 1")
+	requireForeignKeyViolation(t, err)
+	if res := execMust(t, e, "SELECT count(*) FROM parent"); res.Rows[0][0] != int64(2) {
+		t.Fatalf("parent rows after rejected delete = %v, want 2", res.Rows)
+	}
+
+	// A scan-based delete takes the same path through the transaction commit.
+	_, err = execSQL(e, "DELETE FROM parent")
+	requireForeignKeyViolation(t, err)
+
+	execMust(t, e, "DELETE FROM child WHERE id = 10")
+	execMust(t, e, "DELETE FROM parent WHERE id = 1")
+	execMust(t, e, "DELETE FROM parent WHERE id = 2")
+}
+
+// Updating a referenced parent key is rejected while a child points at it and
+// accepted once nothing references the old value.
+func TestForeignKeyParentKeyUpdate(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE parent (id INTEGER PRIMARY KEY, v TEXT)")
+	execMust(t, e, "CREATE TABLE child (id INTEGER PRIMARY KEY, parent_id INTEGER REFERENCES parent(id))")
+	execMust(t, e, "INSERT INTO parent (id, v) VALUES (1, 'one'), (2, 'two')")
+	execMust(t, e, "INSERT INTO child (id, parent_id) VALUES (10, 1)")
+
+	// A non-referenced column update is always allowed.
+	execMust(t, e, "UPDATE parent SET v = 'uno' WHERE id = 1")
+
+	_, err := execSQL(e, "UPDATE parent SET id = 5 WHERE id = 1")
+	requireForeignKeyViolation(t, err)
+	res := execMust(t, e, "SELECT id FROM parent WHERE v = 'uno'")
+	if res.Rows[0][0] != int64(1) {
+		t.Fatalf("parent id after rejected update = %v, want 1", res.Rows[0][0])
+	}
+
+	execMust(t, e, "DELETE FROM child WHERE id = 10")
+	execMust(t, e, "UPDATE parent SET id = 5 WHERE id = 1")
+	if res := execMust(t, e, "SELECT id FROM parent WHERE v = 'uno'"); res.Rows[0][0] != int64(5) {
+		t.Fatalf("parent id after valid update = %v, want 5", res.Rows[0][0])
+	}
+}
+
+// A row may reference itself and another row of the same table; a dangling
+// self-reference is rejected.
+func TestForeignKeySelfReference(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, 1)")
+	execMust(t, e, "INSERT INTO nodes (id, parent_id) VALUES (2, 1)")
+	execMust(t, e, "INSERT INTO nodes (id, parent_id) VALUES (3, NULL)")
+
+	_, err := execSQL(e, "INSERT INTO nodes (id, parent_id) VALUES (4, 99)")
+	requireForeignKeyViolation(t, err)
+
+	// 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,
+		b TEXT NOT NULL,
+		pa TEXT,
+		pb TEXT,
+		PRIMARY KEY (a, b),
+		FOREIGN KEY (pa, pb) REFERENCES tree(a, b)
+	)`)
+	execMust(t, e, "INSERT INTO tree (a, b, pa, pb) VALUES ('r', '1', 'r', '1')")
+	execMust(t, e, "INSERT INTO tree (a, b, pa, pb) VALUES ('c', '1', 'r', '1')")
+	_, err = execSQL(e, "INSERT INTO tree (a, b, pa, pb) VALUES ('x', '1', 'missing', '1')")
+	requireForeignKeyViolation(t, err)
+}
+
+// A transaction is validated at COMMIT against its final state, so a staged
+// child insert and a staged parent delete cannot commit into a dangling
+// reference in either order.
+func TestForeignKeyTransactionCommitEnforcement(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, "INSERT INTO parent (id) VALUES (1)")
+
+	t.Run("staged child insert then parent delete", func(t *testing.T) {
+		execMust(t, e, "BEGIN")
+		execMust(t, e, "INSERT INTO child (id, parent_id) VALUES (10, 1)")
+		execMust(t, e, "DELETE FROM parent WHERE id = 1")
+		_, err := execSQL(e, "COMMIT")
+		requireForeignKeyViolation(t, err)
+
+		if res := execMust(t, e, "SELECT count(*) FROM child"); res.Rows[0][0] != int64(0) {
+			t.Fatalf("child rows after rejected commit = %v, want 0", res.Rows)
+		}
+		if res := execMust(t, e, "SELECT count(*) FROM parent"); res.Rows[0][0] != int64(1) {
+			t.Fatalf("parent rows after rejected commit = %v, want 1", res.Rows)
+		}
+	})
+
+	t.Run("parent delete then staged child insert", func(t *testing.T) {
+		execMust(t, e, "BEGIN")
+		execMust(t, e, "DELETE FROM parent WHERE id = 1")
+		execMust(t, e, "INSERT INTO child (id, parent_id) VALUES (10, 1)")
+		_, err := execSQL(e, "COMMIT")
+		requireForeignKeyViolation(t, err)
+		execMust(t, e, "SELECT count(*) FROM parent")
+	})
+
+	t.Run("staged child insert with missing parent", func(t *testing.T) {
+		execMust(t, e, "BEGIN")
+		execMust(t, e, "INSERT INTO child (id, parent_id) VALUES (11, 77)")
+		_, err := execSQL(e, "COMMIT")
+		requireForeignKeyViolation(t, err)
+		if res := execMust(t, e, "SELECT count(*) FROM child"); res.Rows[0][0] != int64(0) {
+			t.Fatalf("child rows after rejected commit = %v, want 0", res.Rows)
+		}
+	})
+}
+
+// Deleting a parent and its children in one transaction succeeds in either
+// order because the commit sees the final, consistent state.
+func TestForeignKeyTransactionDeleteParentAndChild(t *testing.T) {
+	for _, childrenFirst := range []bool{false, true} {
+		name := "parent first"
+		if childrenFirst {
+			name = "children first"
+		}
+		t.Run(name, func(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, "INSERT INTO parent (id) VALUES (1), (2)")
+			execMust(t, e, "INSERT INTO child (id, parent_id) VALUES (10, 1), (11, 1)")
+
+			execMust(t, e, "BEGIN")
+			if childrenFirst {
+				execMust(t, e, "DELETE FROM child WHERE parent_id = 1")
+				execMust(t, e, "DELETE FROM parent WHERE id = 1")
+			} else {
+				execMust(t, e, "DELETE FROM parent WHERE id = 1")
+				execMust(t, e, "DELETE FROM child WHERE parent_id = 1")
+			}
+			execMust(t, e, "COMMIT")
+
+			if res := execMust(t, e, "SELECT count(*) FROM parent"); res.Rows[0][0] != int64(1) {
+				t.Fatalf("parent rows = %v, want 1", res.Rows)
+			}
+			if res := execMust(t, e, "SELECT count(*) FROM child"); res.Rows[0][0] != int64(0) {
+				t.Fatalf("child rows = %v, want 0", res.Rows)
+			}
+		})
+	}
+}
+
+// A parent column is only a valid target when it is the primary key or covered
+// by a UNIQUE index, and forward references fail at CREATE time.
+func TestForeignKeyDDLRejections(t *testing.T) {
+	tests := []struct {
+		name  string
+		setup []string
+		ddl   string
+		want  string
+	}{
+		{
+			name: "parent does not exist",
+			ddl:  "CREATE TABLE child (id INTEGER PRIMARY KEY, parent_id INTEGER REFERENCES parent(id))",
+			want: "does not exist",
+		},
+		{
+			name:  "parent columns are not unique",
+			setup: []string{"CREATE TABLE parent (id INTEGER PRIMARY KEY, code TEXT)"},
+			ddl:   "CREATE TABLE child (id INTEGER PRIMARY KEY, code TEXT REFERENCES parent(code))",
+			want:  "foreign key mismatch",
+		},
+		{
+			name:  "composite parent columns are not a unique set",
+			setup: []string{"CREATE TABLE parent (a TEXT UNIQUE, b INTEGER)"},
+			ddl:   "CREATE TABLE child (id INTEGER PRIMARY KEY, a TEXT, b INTEGER, FOREIGN KEY (a, b) REFERENCES parent(a, b))",
+			want:  "foreign key mismatch",
+		},
+		{
+			name:  "unknown child column",
+			setup: []string{"CREATE TABLE parent (id INTEGER PRIMARY KEY)"},
+			ddl:   "CREATE TABLE child (id INTEGER PRIMARY KEY, FOREIGN KEY (missing) REFERENCES parent(id))",
+			want:  `column "missing"`,
+		},
+		{
+			name:  "unknown parent column",
+			setup: []string{"CREATE TABLE parent (id INTEGER PRIMARY KEY)"},
+			ddl:   "CREATE TABLE child (id INTEGER PRIMARY KEY, parent_id INTEGER REFERENCES parent(missing))",
+			want:  "does not exist in table parent",
+		},
+		{
+			name:  "column count mismatch",
+			setup: []string{"CREATE TABLE parent (a TEXT, b TEXT, PRIMARY KEY (a, b))"},
+			ddl:   "CREATE TABLE child (id INTEGER PRIMARY KEY, a TEXT, FOREIGN KEY (a) REFERENCES parent(a, b))",
+			want:  "same number of columns",
+		},
+		{
+			name:  "on delete cascade",
+			setup: []string{"CREATE TABLE parent (id INTEGER PRIMARY KEY)"},
+			ddl:   "CREATE TABLE child (id INTEGER PRIMARY KEY, parent_id INTEGER REFERENCES parent(id) ON DELETE CASCADE)",
+			want:  "cascades are not implemented",
+		},
+		{
+			name:  "on update cascade",
+			setup: []string{"CREATE TABLE parent (id INTEGER PRIMARY KEY)"},
+			ddl:   "CREATE TABLE child (id INTEGER PRIMARY KEY, parent_id INTEGER, FOREIGN KEY (parent_id) REFERENCES parent(id) ON UPDATE CASCADE)",
+			want:  "cascades are not implemented",
+		},
+		{
+			name:  "on delete set null",
+			setup: []string{"CREATE TABLE parent (id INTEGER PRIMARY KEY)"},
+			ddl:   "CREATE TABLE child (id INTEGER PRIMARY KEY, parent_id INTEGER, FOREIGN KEY (parent_id) REFERENCES parent(id) ON DELETE SET NULL)",
+			want:  "cascades are not implemented",
+		},
+		{
+			name:  "on delete set default",
+			setup: []string{"CREATE TABLE parent (id INTEGER PRIMARY KEY)"},
+			ddl:   "CREATE TABLE child (id INTEGER PRIMARY KEY, parent_id INTEGER, FOREIGN KEY (parent_id) REFERENCES parent(id) ON DELETE SET DEFAULT)",
+			want:  "cascades are not implemented",
+		},
+		{
+			name:  "deferrable",
+			setup: []string{"CREATE TABLE parent (id INTEGER PRIMARY KEY)"},
+			ddl:   "CREATE TABLE child (id INTEGER PRIMARY KEY, parent_id INTEGER REFERENCES parent(id) DEFERRABLE INITIALLY DEFERRED)",
+			want:  "DEFERRABLE",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			_, schema, table := newTestDB(t)
+			e := newExec(schema, table)
+			for _, sql := range tt.setup {
+				execMust(t, e, sql)
+			}
+			_, err := execSQL(e, tt.ddl)
+			requireUnsupported(t, err, tt.want)
+			if schema.TableExists("child") {
+				t.Fatal("rejected CREATE TABLE left a schema behind")
+			}
+		})
+	}
+
+	// ALTER TABLE ADD COLUMN still cannot introduce a foreign key.
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE parent (id INTEGER PRIMARY KEY)")
+	execMust(t, e, "CREATE TABLE t (id INTEGER PRIMARY KEY)")
+	_, err := execSQL(e, "ALTER TABLE t ADD COLUMN parent_id INTEGER REFERENCES parent(id)")
+	requireUnsupported(t, err, "CREATE TABLE")
+}
+
+// The persisted constraint keeps being enforced after the schema is re-read
+// from storage.
+func TestForeignKeyDurableReload(t *testing.T) {
+	pool, 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, CONSTRAINT child_parent FOREIGN KEY (parent_id) REFERENCES parent(id))")
+	execMust(t, e, "INSERT INTO parent (id) VALUES (1)")
+	execMust(t, e, "INSERT INTO child (id, parent_id) VALUES (10, 1)")
+
+	reloaded := storage.NewSchemaManager(pool, "testdb")
+	reopened := newExec(reloaded, storage.NewTableManager(pool, reloaded, "testdb"))
+
+	stored, err := reloaded.GetSchema("child")
+	if err != nil {
+		t.Fatalf("reload schema: %v", err)
+	}
+	if len(stored.ForeignKeys) != 1 || stored.ForeignKeys[0].Name != "child_parent" {
+		t.Fatalf("reloaded foreign keys = %+v, want child_parent", stored.ForeignKeys)
+	}
+
+	_, err = execSQL(reopened, "INSERT INTO child (id, parent_id) VALUES (11, 2)")
+	requireForeignKeyViolation(t, err)
+	_, err = execSQL(reopened, "DELETE FROM parent WHERE id = 1")
+	requireForeignKeyViolation(t, err)
+	execMust(t, reopened, "INSERT INTO child (id, parent_id) VALUES (12, 1)")
+}
+
+// The emulated SQLite catalog reports the foreign key in the table SQL.
+func TestForeignKeyInSQLiteCatalog(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, CONSTRAINT child_parent FOREIGN KEY (parent_id) REFERENCES parent(id))")
+
+	res := execMust(t, e, "SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'child'")
+	if res.RowCount != 1 {
+		t.Fatalf("sqlite_master rows = %d, want 1", res.RowCount)
+	}
+	sql, _ := res.Rows[0][0].(string)
+	for _, want := range []string{"CONSTRAINT `child_parent` FOREIGN KEY (`parent_id`) REFERENCES `parent` (`id`)"} {
+		if !strings.Contains(sql, want) {
+			t.Fatalf("catalog SQL %q does not contain %q", sql, want)
+		}
+	}
+
+	res = execMust(t, e, "SELECT sql FROM sqlite_schema WHERE name = 'child'")
+	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
+// support already in the engine.
+func TestForeignKeyRegressionCompositeKeyCheckDefaults(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, `CREATE TABLE parent (
+		a TEXT,
+		b INTEGER,
+		PRIMARY KEY (a, b)
+	)`)
+	execMust(t, e, `CREATE TABLE child (
+		id INTEGER PRIMARY KEY,
+		pa TEXT NOT NULL,
+		pb INTEGER NOT NULL,
+		state TEXT NOT NULL DEFAULT 'new' CHECK (state IN ('new', 'done')),
+		FOREIGN KEY (pa, pb) REFERENCES parent(a, b)
+	)`)
+
+	execMust(t, e, "INSERT INTO parent (a, b) VALUES ('x', 1)")
+	execMust(t, e, "INSERT INTO child (id, pa, pb) VALUES (1, 'x', 1)")
+	if res := execMust(t, e, "SELECT state FROM child WHERE id = 1"); res.Rows[0][0] != "new" {
+		t.Fatalf("default state = %v, want new", res.Rows[0][0])
+	}
+	execMust(t, e, "UPDATE child SET state = 'done' WHERE id = 1")
+
+	_, err := execSQL(e, "INSERT INTO child (id, pa, pb) VALUES (2, 'x', 2)")
+	requireForeignKeyViolation(t, err)
+	_, err = execSQL(e, "INSERT INTO child (id, pa, pb, state) VALUES (2, 'x', 1, 'bogus')")
+	requireCheckViolation(t, err, "")
+	_, err = execSQL(e, "INSERT INTO child (id, pa, pb) VALUES (1, 'x', 1)")
+	if err == nil {
+		t.Fatal("duplicate composite parent key or primary key was accepted")
+	}
+}
+
+// A parent column covered by a UNIQUE index (not the primary key) is a valid
+// foreign-key target, whether the uniqueness comes from a constraint or an
+// explicit CREATE UNIQUE INDEX.
+func TestForeignKeyReferencesUniqueIndex(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE by_constraint (id INTEGER PRIMARY KEY, code TEXT UNIQUE)")
+	execMust(t, e, "CREATE TABLE by_index (id INTEGER PRIMARY KEY, code TEXT)")
+	execMust(t, e, "CREATE UNIQUE INDEX idx_by_index_code ON by_index (code)")
+	execMust(t, e, "CREATE TABLE child_constraint (id INTEGER PRIMARY KEY, code TEXT REFERENCES by_constraint(code))")
+	execMust(t, e, "CREATE TABLE child_index (id INTEGER PRIMARY KEY, code TEXT REFERENCES by_index(code))")
+
+	execMust(t, e, "INSERT INTO by_constraint (id, code) VALUES (1, 'a')")
+	execMust(t, e, "INSERT INTO by_index (id, code) VALUES (1, 'b')")
+	execMust(t, e, "INSERT INTO child_constraint (id, code) VALUES (10, 'a')")
+	execMust(t, e, "INSERT INTO child_index (id, code) VALUES (10, 'b')")
+
+	_, err := execSQL(e, "INSERT INTO child_constraint (id, code) VALUES (11, 'missing')")
+	requireForeignKeyViolation(t, err)
+	_, err = execSQL(e, "INSERT INTO child_index (id, code) VALUES (11, 'missing')")
+	requireForeignKeyViolation(t, err)
+
+	_, err = execSQL(e, "DELETE FROM by_constraint WHERE code = 'a'")
+	requireForeignKeyViolation(t, err)
+	_, err = execSQL(e, "DELETE FROM by_index WHERE code = 'b'")
+	requireForeignKeyViolation(t, err)
+}
+
+// A scan-based parent UPDATE of a referenced column is validated at the
+// statement boundary just like a point update.
+func TestForeignKeyParentScanUpdate(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE parent (id INTEGER PRIMARY KEY, v INTEGER)")
+	execMust(t, e, "CREATE TABLE child (id INTEGER PRIMARY KEY, parent_id INTEGER REFERENCES parent(id))")
+	execMust(t, e, "INSERT INTO parent (id, v) VALUES (1, 1), (2, 1)")
+	execMust(t, e, "INSERT INTO child (id, parent_id) VALUES (10, 1)")
+
+	_, err := execSQL(e, "UPDATE parent SET id = id + 10 WHERE v = 1")
+	requireForeignKeyViolation(t, err)
+	if res := execMust(t, e, "SELECT count(*) FROM parent WHERE id > 10"); res.Rows[0][0] != int64(0) {
+		t.Fatalf("rejected scan update moved rows: %v", res.Rows)
+	}
+
+	execMust(t, e, "DELETE FROM child WHERE id = 10")
+	execMust(t, e, "UPDATE parent SET id = id + 10 WHERE v = 1")
+	if res := execMust(t, e, "SELECT count(*) FROM parent WHERE id > 10"); res.Rows[0][0] != int64(2) {
+		t.Fatalf("valid scan update rows = %v, want 2", res.Rows)
+	}
+}
+
+// INSERT ... SELECT goes through the same enforcement as a VALUES insert.
+func TestForeignKeyInsertSelect(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 staging (id INTEGER PRIMARY KEY, parent_id INTEGER)")
+	execMust(t, e, "CREATE TABLE child (id INTEGER PRIMARY KEY, parent_id INTEGER REFERENCES parent(id))")
+	execMust(t, e, "INSERT INTO parent (id) VALUES (1)")
+	execMust(t, e, "INSERT INTO staging (id, parent_id) VALUES (10, 1), (11, 2)")
+
+	_, err := execSQL(e, "INSERT INTO child (id, parent_id) SELECT id, parent_id FROM staging")
+	requireForeignKeyViolation(t, err)
+	if res := execMust(t, e, "SELECT count(*) FROM child"); res.Rows[0][0] != int64(0) {
+		t.Fatalf("child rows after rejected INSERT ... SELECT = %v, want 0", res.Rows)
+	}
+
+	execMust(t, e, "DELETE FROM staging WHERE id = 11")
+	execMust(t, e, "INSERT INTO child (id, parent_id) SELECT id, parent_id FROM staging")
+	if res := execMust(t, e, "SELECT count(*) FROM child"); res.Rows[0][0] != int64(1) {
+		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

@@ -0,0 +1,296 @@
+package executor
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/danfragoso/pizzasql-next/pkg/parser"
+	"github.com/danfragoso/pizzasql-next/pkg/pgerror"
+	"github.com/danfragoso/pizzasql-next/pkg/storage"
+)
+
+// foreignKeyDeclaration is one parsed FOREIGN KEY (or inline REFERENCES) before
+// it is resolved against the parent table.
+type foreignKeyDeclaration struct {
+	name       string
+	columns    []string
+	refTable   string
+	refColumns []string
+	onDelete   parser.ReferentialAction
+	onUpdate   parser.ReferentialAction
+	deferrable bool
+}
+
+// createTableForeignKeys resolves the FOREIGN KEY declarations of a CREATE TABLE
+// into durable constraints. The parent table must already exist with its
+// unique key in place, except for a self-reference to the table being created;
+// a forward reference fails here instead of being stored and silently ignored.
+func (e *Executor) createTableForeignKeys(stmt *parser.CreateTableStmt, schema *storage.Schema, uniqueIndexes []*storage.Index) ([]storage.ForeignKeyConstraint, error) {
+	var declarations []foreignKeyDeclaration
+	for _, colDef := range stmt.Columns {
+		for _, constraint := range colDef.Constraints {
+			if constraint.Type != parser.ConstraintForeignKey {
+				continue
+			}
+			declarations = append(declarations, foreignKeyDeclaration{
+				name:       constraint.Name,
+				columns:    []string{colDef.Name},
+				refTable:   constraint.RefTable,
+				refColumns: constraint.RefColumns,
+				onDelete:   constraint.OnDelete,
+				onUpdate:   constraint.OnUpdate,
+				deferrable: constraint.Deferrable,
+			})
+		}
+	}
+	for _, constraint := range stmt.Constraints {
+		if constraint.Type != parser.ConstraintForeignKey {
+			continue
+		}
+		declarations = append(declarations, foreignKeyDeclaration{
+			name:       constraint.Name,
+			columns:    constraint.Columns,
+			refTable:   constraint.RefTable,
+			refColumns: constraint.RefColumns,
+			onDelete:   constraint.OnDelete,
+			onUpdate:   constraint.OnUpdate,
+			deferrable: constraint.Deferrable,
+		})
+	}
+
+	foreignKeys := make([]storage.ForeignKeyConstraint, 0, len(declarations))
+	for _, declaration := range declarations {
+		foreignKey, err := e.resolveForeignKey(stmt.Table.Name, schema, uniqueIndexes, declaration)
+		if err != nil {
+			return nil, err
+		}
+		foreignKeys = append(foreignKeys, foreignKey)
+	}
+	return foreignKeys, nil
+}
+
+// resolveForeignKey validates one declaration and returns its durable form.
+func (e *Executor) resolveForeignKey(table string, schema *storage.Schema, uniqueIndexes []*storage.Index, declaration foreignKeyDeclaration) (storage.ForeignKeyConstraint, error) {
+	label := declaration.label()
+	if declaration.deferrable {
+		return storage.ForeignKeyConstraint{}, pgerror.Unsupported(
+			"foreign key %s on table %s is DEFERRABLE; foreign keys are always enforced immediately and deferred constraints are not implemented",
+			label, table,
+		)
+	}
+	for _, action := range []struct {
+		clause string
+		value  parser.ReferentialAction
+	}{{"ON DELETE", declaration.onDelete}, {"ON UPDATE", declaration.onUpdate}} {
+		switch action.value {
+		case parser.RefActionCascade, parser.RefActionSetNull, parser.RefActionSetDefault:
+			return storage.ForeignKeyConstraint{}, pgerror.Unsupported(
+				"foreign key %s on table %s uses %s %s; cascades are not implemented, use NO ACTION or RESTRICT",
+				label, table, action.clause, action.value,
+			)
+		}
+	}
+
+	childColumns := make([]string, 0, len(declaration.columns))
+	seen := make(map[string]bool, len(declaration.columns))
+	for _, name := range declaration.columns {
+		column, ok := schema.GetColumn(name)
+		if !ok {
+			return storage.ForeignKeyConstraint{}, pgerror.Unsupported(
+				"foreign key %s on table %s names column %q, which does not exist",
+				label, table, name,
+			)
+		}
+		if seen[strings.ToLower(column.Name)] {
+			return storage.ForeignKeyConstraint{}, pgerror.Unsupported(
+				"foreign key %s on table %s repeats column %q",
+				label, table, column.Name,
+			)
+		}
+		seen[strings.ToLower(column.Name)] = true
+		childColumns = append(childColumns, column.Name)
+	}
+	if len(childColumns) == 0 {
+		return storage.ForeignKeyConstraint{}, pgerror.Unsupported(
+			"foreign key %s on table %s declares no columns", label, table,
+		)
+	}
+
+	selfReference := strings.EqualFold(declaration.refTable, table)
+	var parent *storage.Schema
+	var parentIndexes []*storage.Index
+	if selfReference {
+		parent = schema
+		parentIndexes = uniqueIndexes
+	} else {
+		resolved, err := e.schema.GetSchema(declaration.refTable)
+		if err != nil {
+			return storage.ForeignKeyConstraint{}, pgerror.Unsupported(
+				"foreign key %s on table %s references table %q, which does not exist; create the parent table before the child table",
+				label, table, declaration.refTable,
+			)
+		}
+		parent = resolved
+		indexes, err := e.schema.ListTableIndexes(declaration.refTable)
+		if err != nil {
+			return storage.ForeignKeyConstraint{}, err
+		}
+		parentIndexes = indexes
+	}
+
+	refColumns := append([]string(nil), declaration.refColumns...)
+	if len(refColumns) == 0 {
+		if parent.UsesSyntheticRowID() {
+			return storage.ForeignKeyConstraint{}, pgerror.Unsupported(
+				"foreign key %s on table %s references table %s without a column list, but %s has no PRIMARY KEY; declare a PRIMARY KEY or list parent columns covered by a UNIQUE index",
+				label, table, parent.Name, parent.Name,
+			)
+		}
+		refColumns = parent.PrimaryKeyList()
+	}
+	if len(childColumns) != len(refColumns) {
+		return storage.ForeignKeyConstraint{}, pgerror.Unsupported(
+			"foreign key %s on table %s has %d child column(s) but references %d parent column(s); both sides of a foreign key need the same number of columns",
+			label, table, len(childColumns), len(refColumns),
+		)
+	}
+
+	canonical := make([]string, 0, len(refColumns))
+	for _, name := range refColumns {
+		column, ok := parent.GetColumn(name)
+		if !ok {
+			return storage.ForeignKeyConstraint{}, pgerror.Unsupported(
+				"foreign key %s on table %s references column %q, which does not exist in table %s",
+				label, table, name, parent.Name,
+			)
+		}
+		canonical = append(canonical, column.Name)
+	}
+
+	if !foreignKeyTargetIsUnique(parent, canonical, parentIndexes) {
+		return storage.ForeignKeyConstraint{}, pgerror.Unsupported(
+			"foreign key mismatch: table %s references %s(%s), but the referenced columns are neither the parent's PRIMARY KEY nor covered by a UNIQUE index; declare UNIQUE (%s) on %s first",
+			table, parent.Name, strings.Join(canonical, ", "), strings.Join(canonical, ", "), parent.Name,
+		)
+	}
+
+	return storage.ForeignKeyConstraint{
+		Name:       declaration.name,
+		Columns:    childColumns,
+		RefTable:   parent.Name,
+		RefColumns: canonical,
+		OnDelete:   declaration.onDelete.String(),
+		OnUpdate:   declaration.onUpdate.String(),
+	}, nil
+}
+
+// label names the declaration in a DDL error before it has been resolved.
+func (d foreignKeyDeclaration) label() string {
+	if d.name != "" {
+		return d.name
+	}
+	if len(d.columns) == 0 {
+		return "FOREIGN KEY"
+	}
+	return fmt.Sprintf("FOREIGN KEY (%s)", strings.Join(d.columns, ", "))
+}
+
+// foreignKeyTargetIsUnique reports whether columns is exactly the parent's
+// primary key or exactly the column set of a UNIQUE index on the parent.
+func foreignKeyTargetIsUnique(parent *storage.Schema, columns []string, indexes []*storage.Index) bool {
+	if !parent.UsesSyntheticRowID() && parent.PrimaryKeyTargetMatches(columns) {
+		return true
+	}
+	for _, index := range indexes {
+		if !index.Unique || len(index.Columns) != len(columns) {
+			continue
+		}
+		plain := true
+		names := make([]string, len(index.Columns))
+		for i, column := range index.Columns {
+			if column.Expression != "" {
+				plain = false
+				break
+			}
+			names[i] = column.Name
+		}
+		if plain && sameColumnSet(names, columns) {
+			return true
+		}
+	}
+	return false
+}
+
+// sameColumnSet reports whether both lists name the same columns, in any order.
+func sameColumnSet(a, b []string) bool {
+	if len(a) != len(b) {
+		return false
+	}
+	for _, name := range a {
+		found := false
+		for _, other := range b {
+			if strings.EqualFold(name, other) {
+				found = true
+				break
+			}
+		}
+		if !found {
+			return false
+		}
+	}
+	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"
 )
 
@@ -195,19 +197,21 @@ func TestAnalyzeIsSafeNoOp(t *testing.T) {
 	}
 }
 
-func TestForeignKeysPragmaNoOp(t *testing.T) {
+func TestForeignKeysPragmaAlwaysOn(t *testing.T) {
 	_, schema, table := newTestDB(t)
 	e := newExec(schema, table)
 
-	if res := execMust(t, e, "PRAGMA foreign_keys = OFF"); res.CommandTag != "PRAGMA" {
+	if res := execMust(t, e, "PRAGMA foreign_keys = ON"); res.CommandTag != "PRAGMA" {
 		t.Fatalf("PRAGMA tag = %q", res.CommandTag)
 	}
 	res := execMust(t, e, "PRAGMA foreign_keys")
-	if res.RowCount != 1 || res.Rows[0][0] != int64(0) {
-		t.Fatalf("foreign_keys = %v, want 0", res.Rows)
+	if res.RowCount != 1 || res.Rows[0][0] != int64(1) {
+		t.Fatalf("foreign_keys = %v, want 1", res.Rows)
 	}
-	if _, err := execSQL(e, "PRAGMA foreign_keys = ON"); err == nil {
-		t.Fatal("expected enabling unsupported foreign keys to fail")
+	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

@@ -339,6 +339,31 @@ func recreateCreateTableSQL(s *storage.Schema) string {
 		}
 		cols = append(cols, fmt.Sprintf("CHECK (%s)", check.Expression))
 	}
+	// Foreign keys are stored as resolved table-level constraints, so an inline
+	// REFERENCES declaration is reported in its equivalent table-level form.
+	for _, fk := range s.ForeignKeys {
+		child := make([]string, len(fk.Columns))
+		for i, name := range fk.Columns {
+			child[i] = quoteIdent(name)
+		}
+		parent := make([]string, len(fk.RefColumns))
+		for i, name := range fk.RefColumns {
+			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)
+		}
+		cols = append(cols, clause)
+	}
 	return fmt.Sprintf("CREATE TABLE %s (%s)", quoteIdent(s.Name), strings.Join(cols, ", "))
 }
 

pkg/httpserver/handler.go

@@ -433,7 +433,15 @@ func constraintViolationHTTPError(err error) *HTTPError {
 	var notNullErr *storage.NotNullViolationError
 	var uniqueErr *storage.UniqueViolationError
 	var primaryKeyErr *storage.PrimaryKeyViolationError
+	var foreignKeyErr *storage.ForeignKeyViolationError
 	switch {
+	case errors.As(err, &foreignKeyErr):
+		return &HTTPError{
+			Status:  http.StatusConflict,
+			Code:    "23503",
+			Message: err.Error(),
+			Details: map[string]interface{}{"constraint": foreignKeyErr.Constraint},
+		}
 	case errors.As(err, &checkErr):
 		return &HTTPError{
 			Status:  http.StatusBadRequest,
@@ -494,6 +502,15 @@ func sqlStateHTTPError(err error) *HTTPError {
 			Message: message,
 		}
 	}
+	if strings.HasPrefix(message, "FOREIGN KEY constraint failed: ") {
+		constraint := strings.TrimPrefix(message, "FOREIGN KEY constraint failed: ")
+		return &HTTPError{
+			Status:  http.StatusConflict,
+			Code:    "23503",
+			Message: message,
+			Details: map[string]interface{}{"constraint": constraint},
+		}
+	}
 	if strings.HasPrefix(message, "CHECK constraint failed: ") {
 		constraint := strings.TrimPrefix(message, "CHECK constraint failed: ")
 		return &HTTPError{

pkg/httpserver/sqlstate_test.go

@@ -124,6 +124,8 @@ func TestSQLStateHTTPErrorMapping(t *testing.T) {
 		code   string
 	}{
 		{"unique", fmt.Errorf("UNIQUE constraint failed: uniq_users_email"), http.StatusConflict, "23505"},
+		{"foreign key", fmt.Errorf("FOREIGN KEY constraint failed: parent_id -> parent"), http.StatusConflict, "23503"},
+		{"typed foreign key", &storage.ForeignKeyViolationError{Table: "child", Constraint: "parent_id -> parent"}, http.StatusConflict, "23503"},
 		{"not null", fmt.Errorf("NOT NULL constraint failed: t.v"), http.StatusBadRequest, "23502"},
 		{"check", fmt.Errorf("CHECK constraint failed: positive_age"), http.StatusBadRequest, "23514"},
 		{"typed check", &storage.CheckViolationError{Table: "t", Constraint: "positive_age"}, http.StatusBadRequest, "23514"},
@@ -170,3 +172,30 @@ func TestHTTPExecuteCommitCheckViolationReturns23514(t *testing.T) {
 		t.Fatalf("code = %q, want 23514", resp.Error.Code)
 	}
 }
+
+// A dangling foreign key reports 23503 through the query endpoint, with the
+// constraint label in details.
+func TestHTTPForeignKeyViolationReturns23503(t *testing.T) {
+	server := setupKVTestServer(t)
+	queryHTTP(t, server, "CREATE TABLE parent (id INTEGER PRIMARY KEY)")
+	queryHTTP(t, server, "CREATE TABLE child (id INTEGER PRIMARY KEY, parent_id INTEGER, CONSTRAINT child_parent FOREIGN KEY (parent_id) REFERENCES parent(id))")
+
+	body, _ := json.Marshal(QueryRequest{SQL: "INSERT INTO child (id, parent_id) VALUES (1, 99)"})
+	r := httptest.NewRequest(http.MethodPost, "/query", strings.NewReader(string(body)))
+	w := httptest.NewRecorder()
+	server.handleQuery(w, r)
+
+	if w.Code != http.StatusConflict {
+		t.Fatalf("status = %d, want 409; body %s", w.Code, w.Body.String())
+	}
+	var resp httpErrorBody
+	if err := json.NewDecoder(w.Body).Decode(&resp); err != nil {
+		t.Fatalf("decode: %v", err)
+	}
+	if resp.Error.Code != "23503" {
+		t.Fatalf("code = %q, want 23503", resp.Error.Code)
+	}
+	if constraint, _ := resp.Error.Details["constraint"].(string); constraint != "child_parent" {
+		t.Fatalf("details.constraint = %q, want child_parent", constraint)
+	}
+}

pkg/parser/ast.go

@@ -214,12 +214,20 @@ type DataType struct {
 
 // ColumnConstraint represents a column-level constraint.
 type ColumnConstraint struct {
-	Type      ConstraintType
-	Name      string // optional constraint name
-	Default   Expr   // for DEFAULT
-	RefTable  string // for REFERENCES
-	RefColumn string // for REFERENCES
-	Check     Expr   // for CHECK
+	Type       ConstraintType
+	Name       string   // optional constraint name
+	Default    Expr     // for DEFAULT
+	RefTable   string   // for REFERENCES
+	RefColumns []string // for REFERENCES (empty means the parent's primary key)
+	Check      Expr     // for CHECK
+	// OnDelete and OnUpdate record the foreign-key referential actions. The
+	// engine enforces foreign keys immediately and implements no cascades, so
+	// only NO ACTION and RESTRICT are accepted; the rest are rejected at DDL.
+	OnDelete ReferentialAction
+	OnUpdate ReferentialAction
+	// Deferrable is set for a DEFERRABLE clause. Deferred enforcement is not
+	// implemented, so DDL rejects it instead of accepting a weaker guarantee.
+	Deferrable bool
 	// OnConflict is the conflict resolution algorithm declared with
 	// ON CONFLICT REPLACE/etc. (ConflictAbort is the zero value/default).
 	OnConflict    ConflictAction
@@ -239,6 +247,37 @@ const (
 	ConstraintAutoIncrement
 )
 
+// ReferentialAction is a parsed foreign-key ON DELETE/ON UPDATE action.
+type ReferentialAction int
+
+const (
+	// RefActionUnspecified is the default (no clause, equivalent to NO ACTION).
+	RefActionUnspecified ReferentialAction = iota
+	RefActionNoAction
+	RefActionRestrict
+	RefActionCascade
+	RefActionSetNull
+	RefActionSetDefault
+)
+
+// String returns the SQL spelling of the action.
+func (a ReferentialAction) String() string {
+	switch a {
+	case RefActionNoAction:
+		return "NO ACTION"
+	case RefActionRestrict:
+		return "RESTRICT"
+	case RefActionCascade:
+		return "CASCADE"
+	case RefActionSetNull:
+		return "SET NULL"
+	case RefActionSetDefault:
+		return "SET DEFAULT"
+	default:
+		return ""
+	}
+}
+
 // TableConstraint represents a table-level constraint.
 type TableConstraint struct {
 	Type       ConstraintType
@@ -247,6 +286,11 @@ type TableConstraint struct {
 	RefTable   string   // for FOREIGN KEY
 	RefColumns []string // for FOREIGN KEY
 	Check      Expr     // for CHECK
+	// OnDelete and OnUpdate record the foreign-key referential actions.
+	OnDelete ReferentialAction
+	OnUpdate ReferentialAction
+	// Deferrable is set for a DEFERRABLE clause.
+	Deferrable bool
 	// OnConflict is the conflict resolution declared with ON CONFLICT
 	// REPLACE/etc.; HasOnConflict distinguishes it from the default ABORT.
 	OnConflict    ConflictAction

pkg/parser/parser.go

@@ -1449,16 +1449,14 @@ func (p *Parser) parseColumnConstraint() (*ColumnConstraint, bool, error) {
 		constraint.RefTable = p.curToken.Literal
 		p.nextToken()
 		if p.curTokenIs(lexer.TokenLParen) {
-			p.nextToken()
-			if !p.curTokenIs(lexer.TokenIdent) {
-				return nil, false, p.curError("expected column name")
-			}
-			constraint.RefColumn = p.curToken.Literal
-			p.nextToken()
-			if !p.curTokenIs(lexer.TokenRParen) {
-				return nil, false, p.curError("expected )")
+			refCols, err := p.parseParenIdentList()
+			if err != nil {
+				return nil, false, err
 			}
-			p.nextToken()
+			constraint.RefColumns = refCols
+		}
+		if err := p.parseForeignKeyClauses(&constraint.OnDelete, &constraint.OnUpdate, &constraint.Deferrable); err != nil {
+			return nil, false, err
 		}
 
 	case lexer.TokenAUTOINCREMENT:
@@ -1536,23 +1534,88 @@ func (p *Parser) parseOnConflictAction() (ConflictAction, bool, error) {
 	}
 }
 
-// consumeReferentialAction consumes one foreign-key referential action
-// (CASCADE, RESTRICT, SET NULL, SET DEFAULT, NO ACTION). It is a no-op for the
-// engine, which does not enforce foreign keys.
-func (p *Parser) consumeReferentialAction() {
-	if p.curTokenIs(lexer.TokenSET) {
-		p.nextToken() // NULL or DEFAULT
+// parseForeignKeyClauses parses the optional ON DELETE/ON UPDATE referential
+// actions and the optional DEFERRABLE clause that follow a REFERENCES target.
+// The engine enforces foreign keys immediately, so the parsed actions are kept
+// for the DDL layer to accept (NO ACTION/RESTRICT) or reject (cascades), and a
+// DEFERRABLE clause is recorded so it can be rejected instead of silently
+// weakening the constraint.
+func (p *Parser) parseForeignKeyClauses(onDelete, onUpdate *ReferentialAction, deferrable *bool) error {
+	for p.curTokenIs(lexer.TokenON) {
+		p.nextToken()
+		var target *ReferentialAction
+		switch {
+		case p.curTokenIs(lexer.TokenDELETE):
+			target = onDelete
+		case p.curTokenIs(lexer.TokenUPDATE):
+			target = onUpdate
+		default:
+			return p.curError("expected DELETE or UPDATE after ON")
+		}
+		p.nextToken()
+		action, err := p.parseReferentialAction()
+		if err != nil {
+			return err
+		}
+		*target = action
+	}
+	if p.curTokenIs(lexer.TokenNOT) && p.peekToken.Type == lexer.TokenIdent && strings.EqualFold(p.peekToken.Literal, "DEFERRABLE") {
+		// NOT DEFERRABLE is the default; consume it so it parses. A following
+		// NOT NULL belongs to the column constraint list, so only consume NOT
+		// when DEFERRABLE actually follows.
+		p.nextToken()
 		p.nextToken()
-		return
+		return nil
+	}
+	if p.curIdentIs("DEFERRABLE") {
+		*deferrable = true
+		p.nextToken()
+		if p.curIdentIs("INITIALLY") {
+			p.nextToken()
+			switch {
+			case p.curIdentIs("DEFERRED"):
+				p.nextToken()
+			case p.curIdentIs("IMMEDIATE"):
+				p.nextToken()
+			default:
+				return p.curError("expected DEFERRED or IMMEDIATE after INITIALLY")
+			}
+		}
 	}
-	if p.curTokenIs(lexer.TokenIdent) && strings.EqualFold(p.curToken.Literal, "NO") {
+	return nil
+}
+
+// parseReferentialAction parses one foreign-key action.
+func (p *Parser) parseReferentialAction() (ReferentialAction, error) {
+	switch {
+	case p.curIdentIs("CASCADE"):
+		p.nextToken()
+		return RefActionCascade, nil
+	case p.curIdentIs("RESTRICT"):
+		p.nextToken()
+		return RefActionRestrict, nil
+	case p.curIdentIs("NO"):
+		p.nextToken()
+		if !p.curIdentIs("ACTION") {
+			return RefActionUnspecified, p.curError("expected ACTION after NO")
+		}
 		p.nextToken()
-		if p.curTokenIs(lexer.TokenIdent) {
-			p.nextToken() // ACTION
+		return RefActionNoAction, nil
+	case p.curTokenIs(lexer.TokenSET):
+		p.nextToken()
+		switch {
+		case p.curTokenIs(lexer.TokenNULL):
+			p.nextToken()
+			return RefActionSetNull, nil
+		case p.curTokenIs(lexer.TokenDEFAULT):
+			p.nextToken()
+			return RefActionSetDefault, nil
+		default:
+			return RefActionUnspecified, p.curError("expected NULL or DEFAULT after SET")
 		}
-		return
+	default:
+		return RefActionUnspecified, p.curError("expected CASCADE, RESTRICT, NO ACTION, SET NULL, or SET DEFAULT")
 	}
-	p.nextToken() // CASCADE, RESTRICT, ...
 }
 
 func (p *Parser) parseTableConstraint() (*TableConstraint, error) {
@@ -1631,15 +1694,8 @@ func (p *Parser) parseTableConstraint() (*TableConstraint, error) {
 		}
 		constraint.RefColumns = refCols
 
-		// Consume and ignore referential actions (ON DELETE/UPDATE ...). The
-		// engine does not enforce foreign keys, but the clause must parse.
-		for p.curTokenIs(lexer.TokenON) {
-			p.nextToken()
-			if !p.curTokenIs(lexer.TokenDELETE) && !p.curTokenIs(lexer.TokenUPDATE) {
-				return nil, p.curError("expected DELETE or UPDATE after ON")
-			}
-			p.nextToken()
-			p.consumeReferentialAction()
+		if err := p.parseForeignKeyClauses(&constraint.OnDelete, &constraint.OnUpdate, &constraint.Deferrable); err != nil {
+			return nil, err
 		}
 
 	case lexer.TokenCHECK:
@@ -2076,6 +2132,13 @@ func (p *Parser) parsePragma() (*PragmaStmt, error) {
 	// Check for value assignment: PRAGMA name = value
 	if p.curTokenIs(lexer.TokenEq) {
 		p.nextToken()
+		// ON is a keyword token, so it cannot start an expression. Accept it here
+		// as the value for boolean pragmas such as PRAGMA foreign_keys = ON.
+		if p.curTokenIs(lexer.TokenON) {
+			stmt.Value = &ColumnRef{Column: "on"}
+			p.nextToken()
+			return stmt, nil
+		}
 		val, err := p.parseExpr()
 		if err != nil {
 			return nil, err

pkg/parser/parser_test.go

@@ -1764,3 +1764,42 @@ func TestParseDetachOptional(t *testing.T) {
 		t.Errorf("expected alias 'other', got '%s'", detach.Alias)
 	}
 }
+
+// The foreign-key clause parser must not swallow a following column constraint:
+// NOT NULL after REFERENCES belongs to the column, and NOT DEFERRABLE is the
+// default that is accepted and recorded as immediate enforcement.
+func TestParseInlineForeignKeyClauseBoundaries(t *testing.T) {
+	stmt := parse(t, "CREATE TABLE t (id INTEGER PRIMARY KEY, pid INTEGER REFERENCES p(id) NOT NULL UNIQUE)")
+	create := stmt.(*CreateTableStmt)
+	column := create.Columns[1]
+	if len(column.Constraints) != 3 {
+		t.Fatalf("constraints = %+v, want REFERENCES, NOT NULL, and UNIQUE", column.Constraints)
+	}
+	if column.Constraints[0].Type != ConstraintForeignKey || column.Constraints[0].RefTable != "p" {
+		t.Fatalf("foreign key = %+v", column.Constraints[0])
+	}
+	if len(column.Constraints[0].RefColumns) != 1 || column.Constraints[0].RefColumns[0] != "id" {
+		t.Fatalf("ref columns = %v, want [id]", column.Constraints[0].RefColumns)
+	}
+	if column.Constraints[1].Type != ConstraintNotNull || column.Constraints[2].Type != ConstraintUnique {
+		t.Fatalf("following constraints = %+v, want NOT NULL then UNIQUE", column.Constraints[1:])
+	}
+
+	stmt = parse(t, "CREATE TABLE t2 (pid INTEGER REFERENCES p(id) NOT DEFERRABLE)")
+	column = stmt.(*CreateTableStmt).Columns[0]
+	if len(column.Constraints) != 1 || column.Constraints[0].Type != ConstraintForeignKey {
+		t.Fatalf("constraints = %+v, want one foreign key", column.Constraints)
+	}
+	if column.Constraints[0].Deferrable {
+		t.Fatal("NOT DEFERRABLE was recorded as deferrable")
+	}
+
+	stmt = parse(t, "CREATE TABLE t3 (pid INTEGER REFERENCES p(id) ON DELETE RESTRICT ON UPDATE NO ACTION DEFERRABLE INITIALLY DEFERRED)")
+	constraint := stmt.(*CreateTableStmt).Columns[0].Constraints[0]
+	if constraint.OnDelete != RefActionRestrict || constraint.OnUpdate != RefActionNoAction {
+		t.Fatalf("actions = %v/%v, want RESTRICT/NO ACTION", constraint.OnDelete, constraint.OnUpdate)
+	}
+	if !constraint.Deferrable {
+		t.Fatal("DEFERRABLE INITIALLY DEFERRED was not recorded")
+	}
+}

pkg/pgerror/pgerror.go

@@ -11,6 +11,7 @@ import (
 const (
 	CodeFeatureNotSupported  = "0A000"
 	CodeUniqueViolation      = "23505"
+	CodeForeignKeyViolation  = "23503"
 	CodeCheckViolation       = "23514"
 	CodeSerializationFailure = "40001"
 	CodeSyntaxError          = "42601"

pkg/pgserver/compat_suite_test.go

@@ -137,11 +137,13 @@ func TestCompatibilitySuite(t *testing.T) {
 	expectCode("locking.for_update_0A000", `SELECT * FROM accounts FOR UPDATE`, "0A000")
 	expectCode("arrays.any_0A000", `SELECT * FROM accounts WHERE id = ANY('{1,2}')`, "0A000")
 
-	// CHECK is accepted and enforced; FOREIGN KEY is still rejected at DDL
-	// instead of being silently discarded.
+	// CHECK and FOREIGN KEY are accepted and enforced.
 	expectOK("ddl.check_accepted", `CREATE TABLE checked (n INTEGER CHECK (n > 0))`)
 	expectCode("constraints.check_23514", `INSERT INTO checked (n) VALUES (-1)`, "23514")
-	expectCode("ddl.foreign_key_0A000", `CREATE TABLE child (id INTEGER PRIMARY KEY, parent_id INTEGER, FOREIGN KEY (parent_id) REFERENCES accounts(id))`, "0A000")
+	expectOK("ddl.foreign_key_accepted", `CREATE TABLE child (id INTEGER PRIMARY KEY, parent_id INTEGER, FOREIGN KEY (parent_id) REFERENCES accounts(id))`)
+	expectOK("dml.foreign_key_valid", `INSERT INTO child (id, parent_id) VALUES (1, 1)`)
+	expectCode("constraints.foreign_key_23503", `INSERT INTO child (id, parent_id) VALUES (2, 99)`, "23503")
+	expectCode("constraints.foreign_key_parent_delete_23503", `DELETE FROM accounts WHERE id = 1`, "23503")
 
 	expectCode("errors.syntax_42601", `SELEC 1`, "42601")
 

pkg/pgserver/connection.go

@@ -417,6 +417,7 @@ func (c *Connection) handleQuery(msg *Message) error {
 			var uniqueErr *storage.UniqueViolationError
 			var notNullErr *storage.NotNullViolationError
 			var checkErr *storage.CheckViolationError
+			var foreignKeyErr *storage.ForeignKeyViolationError
 			if errors.Is(err, storage.ErrSerialization) {
 				code = ErrCodeSerializationFailure
 				c.txStatus = TxStatusIdle
@@ -434,6 +435,13 @@ func (c *Connection) handleQuery(msg *Message) error {
 				} else if c.txStatus == TxStatusInBlock {
 					c.txStatus = TxStatusFailed
 				}
+			} else if errors.As(err, &foreignKeyErr) {
+				code = ErrCodeForeignKeyViolation
+				if _, commit := stmt.(*parser.CommitStmt); commit {
+					c.txStatus = TxStatusIdle
+				} else if c.txStatus == TxStatusInBlock {
+					c.txStatus = TxStatusFailed
+				}
 			} else if errors.As(err, &uniqueErr) || errors.As(err, &primaryKeyErr) {
 				code = ErrCodeUniqueViolation
 				if uniqueErr != nil {
@@ -673,6 +681,7 @@ func (c *Connection) handleExecute(msg *Message) error {
 		var uniqueErr *storage.UniqueViolationError
 		var notNullErr *storage.NotNullViolationError
 		var checkErr *storage.CheckViolationError
+		var foreignKeyErr *storage.ForeignKeyViolationError
 		if errors.Is(err, storage.ErrSerialization) {
 			code = ErrCodeSerializationFailure
 			c.txStatus = TxStatusIdle
@@ -690,6 +699,13 @@ func (c *Connection) handleExecute(msg *Message) error {
 			} else if c.txStatus == TxStatusInBlock {
 				c.txStatus = TxStatusFailed
 			}
+		} else if errors.As(err, &foreignKeyErr) {
+			code = ErrCodeForeignKeyViolation
+			if _, commit := stmt.(*parser.CommitStmt); commit {
+				c.txStatus = TxStatusIdle
+			} else if c.txStatus == TxStatusInBlock {
+				c.txStatus = TxStatusFailed
+			}
 		} else if errors.As(err, &uniqueErr) || errors.As(err, &primaryKeyErr) {
 			code = ErrCodeUniqueViolation
 			if _, commit := stmt.(*parser.CommitStmt); commit {

pkg/pgserver/foreign_key_test.go

@@ -0,0 +1,79 @@
+package pgserver
+
+import (
+	"testing"
+)
+
+// A dangling foreign key reports SQLSTATE 23503 on both the simple and the
+// extended protocol, and inside a transaction block it leaves the transaction
+// failed like the other integrity errors.
+func TestForeignKeySQLStateAcrossProtocolPaths(t *testing.T) {
+	for _, extended := range []bool{false, true} {
+		for _, transaction := range []bool{false, true} {
+			c, client := newDataConnection(t)
+			runQuery(t, c, client, simpleQueryMessage("CREATE TABLE parent (id INTEGER PRIMARY KEY)"))
+			runQuery(t, c, client, simpleQueryMessage("CREATE TABLE child (id INTEGER PRIMARY KEY, parent_id INTEGER REFERENCES parent(id))"))
+			if transaction {
+				runQuery(t, c, client, simpleQueryMessage("BEGIN"))
+			}
+
+			sql := "INSERT INTO child (id, parent_id) VALUES (1, 99)"
+			var messages []*Message
+			if extended {
+				messages = runExtendedExecution(t, c, client, sql)
+			} else {
+				messages = runQuery(t, c, client, simpleQueryMessage(sql))
+			}
+			if transaction {
+				// A staged write is validated at COMMIT, not by the INSERT.
+				if code := errorCode(messages[0]); code != "" {
+					t.Fatalf("extended=%v staged INSERT SQLSTATE = %q, want none", extended, code)
+				}
+				messages = runQuery(t, c, client, simpleQueryMessage("COMMIT"))
+			}
+			if code := errorCode(messages[0]); code != ErrCodeForeignKeyViolation {
+				t.Fatalf("extended=%v transaction=%v SQLSTATE = %q, want %q", extended, transaction, code, ErrCodeForeignKeyViolation)
+			}
+			if transaction {
+				if c.txStatus != TxStatusIdle {
+					t.Fatalf("tx status = %v, want idle after a failed COMMIT", c.txStatus)
+				}
+				runQuery(t, c, client, simpleQueryMessage("ROLLBACK"))
+			}
+
+			count := runQuery(t, c, client, simpleQueryMessage("SELECT COUNT(*) FROM child"))
+			if values := dataRowValues(t, count[1]); len(values) != 1 || values[0] != "0" {
+				t.Fatalf("rejected insert left rows behind: %v", values)
+			}
+		}
+	}
+}
+
+// Deleting a referenced parent reports 23503 over the wire and leaves the
+// parent row in place.
+func TestForeignKeyParentDeleteSQLState(t *testing.T) {
+	c, client := newDataConnection(t)
+	runQuery(t, c, client, simpleQueryMessage("CREATE TABLE parent (id INTEGER PRIMARY KEY)"))
+	runQuery(t, c, client, simpleQueryMessage("CREATE TABLE child (id INTEGER PRIMARY KEY, parent_id INTEGER REFERENCES parent(id))"))
+	runQuery(t, c, client, simpleQueryMessage("INSERT INTO parent (id) VALUES (1)"))
+	runQuery(t, c, client, simpleQueryMessage("INSERT INTO child (id, parent_id) VALUES (10, 1)"))
+
+	messages := runQuery(t, c, client, simpleQueryMessage("DELETE FROM parent WHERE id = 1"))
+	if code := errorCode(messages[0]); code != ErrCodeForeignKeyViolation {
+		t.Fatalf("DELETE SQLSTATE = %q, want %q", code, ErrCodeForeignKeyViolation)
+	}
+	count := runQuery(t, c, client, simpleQueryMessage("SELECT COUNT(*) FROM parent"))
+	if values := dataRowValues(t, count[1]); len(values) != 1 || values[0] != "1" {
+		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/pgserver/protocol.go

@@ -96,6 +96,7 @@ const (
 	ErrCodeTransactionAborted   = "25P02"
 	ErrCodeSerializationFailure = "40001"
 	ErrCodeUniqueViolation      = "23505"
+	ErrCodeForeignKeyViolation  = "23503"
 	ErrCodeCheckViolation       = "23514"
 )
 

pkg/pgserver/unsupported_constraints_test.go

@@ -1,21 +0,0 @@
-package pgserver
-
-import (
-	"testing"
-)
-
-// CHECK is enforced (see check_test.go); FOREIGN KEY is still rejected.
-func TestForeignKeyRejectedOverSimpleQuery(t *testing.T) {
-	c, client := newDataConnection(t)
-	runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte("CREATE TABLE parent (id INTEGER PRIMARY KEY)"), 0)})
-
-	for _, sql := range []string{
-		"CREATE TABLE t1 (id INTEGER PRIMARY KEY, parent_id INTEGER REFERENCES parent(id))",
-		"CREATE TABLE t2 (id INTEGER PRIMARY KEY, parent_id INTEGER, FOREIGN KEY (parent_id) REFERENCES parent(id))",
-	} {
-		msgs := runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte(sql), 0)})
-		if code := errorCode(msgs[0]); code != ErrCodeFeatureNotSupported {
-			t.Fatalf("%s: error code = %q, want %q", sql, code, ErrCodeFeatureNotSupported)
-		}
-	}
-}

pkg/storage/foreign_key.go

@@ -0,0 +1,543 @@
+package storage
+
+import (
+	"fmt"
+	"sort"
+	"strconv"
+	"strings"
+)
+
+// ForeignKeyViolationError reports a row that would leave a foreign key
+// dangling. It is a distinct type so the protocol layers can report SQLSTATE
+// 23503 instead of a generic execution error.
+type ForeignKeyViolationError struct {
+	// Table is the child table whose row cannot be written, or whose rows would
+	// be orphaned by the parent change.
+	Table string
+	// Constraint is the label of the violated FOREIGN KEY declaration.
+	Constraint string
+}
+
+func (e *ForeignKeyViolationError) Error() string {
+	if e.Constraint != "" {
+		return fmt.Sprintf("FOREIGN KEY constraint failed: %s", e.Constraint)
+	}
+	return "FOREIGN KEY constraint failed: " + e.Table
+}
+
+// foreignKeyValues returns the formatted values a row holds for the given
+// columns, and reports false when any component is NULL or missing. MATCH
+// SIMPLE semantics: a partially NULL key is never matched against the parent, so
+// the constraint does not apply to that row.
+func foreignKeyValues(row Row, columns []string) ([]string, bool) {
+	values := make([]string, len(columns))
+	for i, name := range columns {
+		value, ok := rowColumnValue(row, name)
+		if !ok || value == nil {
+			return nil, false
+		}
+		values[i] = formatIndexValue(value)
+	}
+	return values, true
+}
+
+// foreignKeyMatches reports whether row's columns equal want.
+func foreignKeyMatches(row Row, columns, want []string) bool {
+	if len(columns) != len(want) {
+		return false
+	}
+	got, ok := foreignKeyValues(row, columns)
+	if !ok {
+		return false
+	}
+	for i := range want {
+		if got[i] != want[i] {
+			return false
+		}
+	}
+	return true
+}
+
+func foreignKeyValuesEqual(a, b []string) bool {
+	if len(a) != len(b) {
+		return false
+	}
+	for i := range a {
+		if a[i] != b[i] {
+			return false
+		}
+	}
+	return true
+}
+
+// parentRowExists reports whether table holds a row whose columns match want.
+// extra supplies rows that are not durable yet (the rows of the current
+// statement), so a self-referential row can reference itself or another row of
+// the same multi-row insert. The scan deliberately takes no table gate: callers
+// hold the table's write gate already, and re-entering it would deadlock on a
+// self-referential constraint.
+func (m *TableManager) parentRowExists(table string, extra []Row, columns, want []string) (bool, error) {
+	for _, row := range extra {
+		if foreignKeyMatches(row, columns, want) {
+			return true, nil
+		}
+	}
+	found := false
+	err := m.scanRows(table, func(row Row) (bool, error) {
+		if foreignKeyMatches(row, columns, want) {
+			found = true
+			return true, nil
+		}
+		return false, nil
+	})
+	return found, err
+}
+
+// validateInsertForeignKeys enforces every FOREIGN KEY declared on schema for
+// rows that are about to be written. rows holds all rows of the statement so a
+// self-referential insert can reference a sibling row in the same batch.
+func (m *TableManager) validateInsertForeignKeys(schema *Schema, rows []Row) error {
+	if len(schema.ForeignKeys) == 0 {
+		return nil
+	}
+	for _, fk := range schema.ForeignKeys {
+		for _, row := range rows {
+			want, ok := foreignKeyValues(row, fk.Columns)
+			if !ok {
+				continue
+			}
+			var extra []Row
+			if strings.EqualFold(fk.RefTable, schema.Name) {
+				extra = rows
+			}
+			found, err := m.parentRowExists(fk.RefTable, extra, fk.RefColumns, want)
+			if err != nil {
+				return fmt.Errorf("validate foreign key %s: %w", fk.Label(), err)
+			}
+			if !found {
+				return &ForeignKeyViolationError{Table: schema.Name, Constraint: fk.Label()}
+			}
+		}
+	}
+	return nil
+}
+
+// validateParentKeyRemoval rejects a parent change that would orphan an existing
+// child row. oldRow is the durable parent image; newRow is the replacement image
+// for an UPDATE (nil for a DELETE). A change that leaves every referenced value
+// untouched cannot orphan anything and is not checked.
+//
+// A self-referential table needs no special exemption for a row that references
+// itself: the row being deleted (or re-keyed) is skipped in the child scan and
+// its replacement image is evaluated instead, so deleting a self-referencing row
+// works while re-keying it into a dangling reference still fails.
+func (m *TableManager) validateParentKeyRemoval(table string, oldRow, newRow Row) error {
+	referencing, err := m.schema.referencingForeignKeys(table)
+	if err != nil {
+		return err
+	}
+	selfPK := ""
+	var selfSchema *Schema
+	if len(referencing) > 0 {
+		schema, err := m.schema.GetSchema(table)
+		if err != nil {
+			return err
+		}
+		selfSchema = schema
+		if pk, err := schema.PrimaryKeyForRow(oldRow); err == nil {
+			selfPK = pk
+		}
+	}
+	for _, ref := range referencing {
+		oldKey, ok := foreignKeyValues(oldRow, ref.FK.RefColumns)
+		if !ok {
+			continue
+		}
+		if newRow != nil {
+			if newKey, ok := foreignKeyValues(newRow, ref.FK.RefColumns); ok && foreignKeyValuesEqual(oldKey, newKey) {
+				continue
+			}
+		}
+		selfTable := strings.EqualFold(ref.Table, table)
+		found := false
+		err := m.scanRows(ref.Table, func(row Row) (bool, error) {
+			if selfTable && selfPK != "" {
+				if pk, err := selfSchema.PrimaryKeyForRow(row); err == nil && pk == selfPK {
+					return false, nil
+				}
+			}
+			if foreignKeyMatches(row, ref.FK.Columns, oldKey) {
+				found = true
+				return true, nil
+			}
+			return false, nil
+		})
+		if err != nil {
+			return fmt.Errorf("validate foreign key %s: %w", ref.FK.Label(), err)
+		}
+		if !found && selfTable && newRow != nil && foreignKeyMatches(newRow, ref.FK.Columns, oldKey) {
+			found = true
+		}
+		if found {
+			return &ForeignKeyViolationError{Table: ref.Table, Constraint: ref.FK.Label()}
+		}
+	}
+	return nil
+}
+
+// referencingForeignKey is one FOREIGN KEY declaration that targets a table.
+type referencingForeignKey struct {
+	Table string
+	FK    ForeignKeyConstraint
+}
+
+// foreignKeyCatalog is the cached foreign-key metadata of the database. It is
+// keyed by the schema manager version so a DDL change rebuilds it.
+type foreignKeyCatalog struct {
+	version      uint64
+	participants map[string]bool
+	referencing  map[string][]referencingForeignKey
+	adjacency    map[string]map[string]bool
+}
+
+// foreignKeyCatalogSnapshot returns the current foreign-key metadata, rebuilding
+// it when the schema version changed.
+func (m *SchemaManager) foreignKeyCatalogSnapshot() (*foreignKeyCatalog, error) {
+	version := m.Version()
+	m.fkCatalogMu.Lock()
+	cached := m.fkCatalog
+	m.fkCatalogMu.Unlock()
+	if cached != nil && cached.version == version {
+		return cached, nil
+	}
+
+	tables, err := m.ListTables()
+	if err != nil {
+		return nil, err
+	}
+	catalog := &foreignKeyCatalog{
+		version:      version,
+		participants: make(map[string]bool),
+		referencing:  make(map[string][]referencingForeignKey),
+		adjacency:    make(map[string]map[string]bool),
+	}
+	link := func(a, b string) {
+		if catalog.adjacency[a] == nil {
+			catalog.adjacency[a] = make(map[string]bool)
+		}
+		catalog.adjacency[a][b] = true
+	}
+	for _, name := range tables {
+		schema, err := m.GetSchema(name)
+		if err != nil {
+			return nil, err
+		}
+		child := strings.ToLower(schema.Name)
+		for _, fk := range schema.ForeignKeys {
+			parent := strings.ToLower(fk.RefTable)
+			catalog.participants[child] = true
+			catalog.participants[parent] = true
+			catalog.referencing[parent] = append(catalog.referencing[parent], referencingForeignKey{Table: schema.Name, FK: fk})
+			link(child, parent)
+			link(parent, child)
+		}
+	}
+
+	m.fkCatalogMu.Lock()
+	if m.Version() == version {
+		m.fkCatalog = catalog
+	}
+	m.fkCatalogMu.Unlock()
+	return catalog, nil
+}
+
+// referencingForeignKeys returns every FOREIGN KEY declaration whose parent is
+// table.
+func (m *SchemaManager) referencingForeignKeys(table string) ([]referencingForeignKey, error) {
+	catalog, err := m.foreignKeyCatalogSnapshot()
+	if err != nil {
+		return nil, err
+	}
+	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.
+func (m *SchemaManager) ForeignKeyParticipants(table string) (bool, error) {
+	catalog, err := m.foreignKeyCatalogSnapshot()
+	if err != nil {
+		return false, err
+	}
+	return catalog.participants[strings.ToLower(table)], nil
+}
+
+// ForeignKeyClosure returns the tables whose rows must be re-validated when the
+// given tables change: the foreign-key-connected component of every changed
+// table that participates in a constraint. The result is empty when no changed
+// table participates.
+func (m *SchemaManager) ForeignKeyClosure(tables []string) (map[string]bool, error) {
+	catalog, err := m.foreignKeyCatalogSnapshot()
+	if err != nil {
+		return nil, err
+	}
+	closure := make(map[string]bool)
+	queue := make([]string, 0, len(tables))
+	for _, table := range tables {
+		if catalog.participants[strings.ToLower(table)] {
+			queue = append(queue, strings.ToLower(table))
+		}
+	}
+	for len(queue) > 0 {
+		table := queue[0]
+		queue = queue[1:]
+		if closure[table] {
+			continue
+		}
+		closure[table] = true
+		neighbors := make([]string, 0, len(catalog.adjacency[table]))
+		for neighbor := range catalog.adjacency[table] {
+			neighbors = append(neighbors, neighbor)
+		}
+		sort.Strings(neighbors)
+		for _, neighbor := range neighbors {
+			if !closure[neighbor] {
+				queue = append(queue, neighbor)
+			}
+		}
+	}
+	return closure, nil
+}
+
+// lockForeignKeyWrites serializes writes that can affect referential integrity,
+// so a validation scan cannot interleave with a concurrent insert, update, or
+// delete on the other side of a constraint. It must be acquired before any
+// table gate: commits validate several tables at once, and the canonical order
+// (foreign-key gate, then table gates) is what keeps that deadlock-free.
+func (m *TableManager) lockForeignKeyWrites(tables ...string) (func(), error) {
+	for _, table := range tables {
+		participates, err := m.schema.ForeignKeyParticipants(table)
+		if err != nil {
+			return nil, err
+		}
+		if participates {
+			m.foreignKeyMu.Lock()
+			return m.foreignKeyMu.Unlock, nil
+		}
+	}
+	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 {
+	tables  map[string][]Row
+	keySets map[string]map[string]bool
+	schemas map[string]*Schema
+}
+
+// validateForeignKeyState checks every child row of the given tables against its
+// parent rows. tables maps a lowercased table name to its final row set (durable
+// rows with staged deletes applied plus staged writes).
+func (v *foreignKeyStateValidator) validateForeignKeyState() error {
+	if v.keySets == nil {
+		v.keySets = make(map[string]map[string]bool)
+	}
+	for table, rows := range v.tables {
+		schema := v.schemas[table]
+		if schema == nil {
+			continue
+		}
+		for _, fk := range schema.ForeignKeys {
+			parent := strings.ToLower(fk.RefTable)
+			parentRows, ok := v.tables[parent]
+			if !ok {
+				return fmt.Errorf("foreign key %s: parent table %s is not available for validation", fk.Label(), fk.RefTable)
+			}
+			key := parent + "\x00" + joinForeignKeyValues(fk.RefColumns)
+			set, ok := v.keySets[key]
+			if !ok {
+				set = make(map[string]bool, len(parentRows))
+				for _, row := range parentRows {
+					values, ok := foreignKeyValues(row, fk.RefColumns)
+					if !ok {
+						continue
+					}
+					set[joinForeignKeyValues(values)] = true
+				}
+				v.keySets[key] = set
+			}
+			for _, row := range rows {
+				values, ok := foreignKeyValues(row, fk.Columns)
+				if !ok {
+					continue
+				}
+				if !set[joinForeignKeyValues(values)] {
+					return &ForeignKeyViolationError{Table: schema.Name, Constraint: fk.Label()}
+				}
+			}
+		}
+	}
+	return nil
+}
+
+func joinForeignKeyValues(values []string) string {
+	var b strings.Builder
+	for _, value := range values {
+		b.WriteString(strconv.Itoa(len(value)))
+		b.WriteByte(':')
+		b.WriteString(value)
+	}
+	return b.String()
+}
+
+// foreignKeyClosure returns the tables a commit must re-validate, or nil when no
+// staged table participates in a foreign key.
+func (s *Session) foreignKeyClosure() (map[string]bool, error) {
+	if len(s.overlay) == 0 {
+		return nil, nil
+	}
+	tables := make([]string, 0, len(s.overlay))
+	for table := range s.overlay {
+		tables = append(tables, table)
+	}
+	return s.schema.ForeignKeyClosure(tables)
+}
+
+// validateForeignKeyFinalState checks the committed row image of every table in
+// the constrained component. The caller holds the session lock and the table
+// gates of those tables.
+func (s *Session) validateForeignKeyFinalState(closure map[string]bool) error {
+	if len(closure) == 0 {
+		return nil
+	}
+	schemas := make(map[string]*Schema, len(closure))
+	tables := make(map[string][]Row, len(closure))
+	for table := range closure {
+		schema, err := s.schema.GetSchema(table)
+		if err != nil {
+			return err
+		}
+		rows, err := s.finalRows(table, schema)
+		if err != nil {
+			return err
+		}
+		schemas[table] = schema
+		tables[table] = rows
+	}
+	validator := foreignKeyStateValidator{tables: tables, schemas: schemas}
+	return validator.validateForeignKeyState()
+}
+
+// finalRows returns the table's durable rows with the transaction overlay
+// applied: deleted keys are removed, staged rows are added, and a key that was
+// both deleted and re-staged appears once with its staged image. It takes no
+// table gate because the caller holds the table's exclusive gate.
+func (s *Session) finalRows(table string, schema *Schema) ([]Row, error) {
+	overlay := s.overlay[strings.ToLower(table)]
+	rows := make([]Row, 0)
+	err := s.table.scanRows(table, func(row Row) (bool, error) {
+		pk, err := schema.PrimaryKeyForRow(row)
+		if err != nil {
+			return true, err
+		}
+		if _, staged := overlay[s.table.dataKey(table, pk)]; staged {
+			return false, nil
+		}
+		rows = append(rows, row)
+		return false, nil
+	})
+	if err != nil {
+		return nil, err
+	}
+	for _, entry := range overlay {
+		if entry.absent {
+			continue
+		}
+		rows = append(rows, cloneRow(entry.row))
+	}
+	return rows, nil
+}

pkg/storage/schema.go

@@ -38,6 +38,34 @@ type Schema struct {
 	// level alike. Inline and table-level declarations are equivalent once
 	// parsed, so both live in this one list.
 	Checks []CheckConstraint `json:"checks,omitempty"`
+	// ForeignKeys holds the immediate FOREIGN KEY declarations. Each constraint
+	// is enforced on every write and re-validated at transaction commit.
+	ForeignKeys []ForeignKeyConstraint `json:"foreign_keys,omitempty"`
+}
+
+// ForeignKeyConstraint is a persisted FOREIGN KEY declaration. Columns are the
+// child columns in declaration order; RefColumns are the parent columns they are
+// matched against. Both sides have the same length, and the parent side is
+// validated at DDL to be the parent's PRIMARY KEY or a UNIQUE index.
+type ForeignKeyConstraint struct {
+	Name       string   `json:"name,omitempty"`
+	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
+// there is one, otherwise the child column list.
+func (fk ForeignKeyConstraint) Label() string {
+	if fk.Name != "" {
+		return fk.Name
+	}
+	return strings.Join(fk.Columns, ", ") + " -> " + fk.RefTable
 }
 
 // CheckConstraint is a persisted CHECK predicate. The predicate is stored as
@@ -150,6 +178,11 @@ type SchemaManager struct {
 	defaultEvalMu sync.RWMutex
 	defaultEval   DefaultEvaluator
 	defaultClock  DefaultClock
+
+	// fkCatalogMu guards fkCatalog, the cached foreign-key metadata used to find
+	// referencing tables without re-reading every schema on every write.
+	fkCatalogMu sync.Mutex
+	fkCatalog   *foreignKeyCatalog
 }
 
 // DefaultEvaluator evaluates a persisted non-constant DEFAULT expression, such
@@ -471,6 +504,19 @@ func cloneSchema(schema *Schema) *Schema {
 	cloned.Columns = append([]Column(nil), schema.Columns...)
 	cloned.Checks = append([]CheckConstraint(nil), schema.Checks...)
 	cloned.PrimaryKeyColumns = append([]string(nil), schema.PrimaryKeyColumns...)
+	if schema.ForeignKeys != nil {
+		cloned.ForeignKeys = make([]ForeignKeyConstraint, len(schema.ForeignKeys))
+		for i, fk := range schema.ForeignKeys {
+			cloned.ForeignKeys[i] = ForeignKeyConstraint{
+				Name:       fk.Name,
+				Columns:    append([]string(nil), fk.Columns...),
+				RefTable:   fk.RefTable,
+				RefColumns: append([]string(nil), fk.RefColumns...),
+				OnDelete:   fk.OnDelete,
+				OnUpdate:   fk.OnUpdate,
+			}
+		}
+	}
 	return &cloned
 }
 

pkg/storage/table.go

@@ -55,6 +55,12 @@ type TableManager struct {
 	// concurrently. Index is derived from the full data key (database+table+pk).
 	stripes [64]sync.Mutex
 
+	// foreignKeyMu serializes writes that touch a table participating in a
+	// FOREIGN KEY, so a child-side or parent-side validation scan cannot
+	// interleave with a conflicting write. It is always acquired before any
+	// table gate.
+	foreignKeyMu sync.Mutex
+
 	// generations tracks full-table scans. predicateGenerations narrows indexed
 	// equality reads to one index value so unrelated writes do not conflict.
 	genMu                sync.Mutex
@@ -595,6 +601,16 @@ func (m *TableManager) InsertWithRowID(table string, row Row) (int64, error) {
 	rowid, _ := rowIDFromRow(nr)
 	wasInit := m.countInitialized(table)
 
+	// Foreign-key validation scans the parent (and, for a parent change, the
+	// child) rows. Take the foreign-key write gate before the table gate so a
+	// concurrent write on the other side of the constraint cannot interleave and
+	// so the lock order stays foreign-key gate -> table gate everywhere.
+	unlockFK, err := m.lockForeignKeyWrites(table)
+	if err != nil {
+		return 0, err
+	}
+	defer unlockFK()
+
 	// Point writers share this gate with each other. Transaction commits and
 	// scan-based writes take it exclusively, so generation validation and cache
 	// publication are ordered without serializing writes to different keys.
@@ -620,6 +636,9 @@ func (m *TableManager) InsertWithRowID(table string, row Row) (int64, error) {
 	if err := m.schema.validateChecks(schema, nr); err != nil {
 		return 0, err
 	}
+	if err := m.validateInsertForeignKeys(schema, []Row{nr}); err != nil {
+		return 0, err
+	}
 	data, err := encodeRow(nr)
 	if err != nil {
 		return 0, fmt.Errorf("failed to serialize row: %w", err)
@@ -710,6 +729,12 @@ func (m *TableManager) InsertBulkWithLastRowID(table string, rows []Row) (int, i
 	if len(rows) == 0 {
 		return 0, 0, nil
 	}
+	unlockFK, err := m.lockForeignKeyWrites(table)
+	if err != nil {
+		return 0, 0, err
+	}
+	defer unlockFK()
+
 	tl := m.tableLock(table)
 	tl.Lock()
 	defer tl.Unlock()
@@ -732,6 +757,12 @@ func (m *TableManager) InsertBulkWithLastRowID(table string, rows []Row) (int, i
 		normalized = append(normalized, nr)
 		dataKeys = append(dataKeys, key)
 	}
+	// Every row of the batch is visible to the others, so a self-referential
+	// multi-row insert may reference a sibling that is written in the same
+	// atomic batch.
+	if err := m.validateInsertForeignKeys(schema, normalized); err != nil {
+		return 0, 0, err
+	}
 
 	// Serialize all rows into batch operations. A parallel slice keeps the
 	// normalized Row for each op for index maintenance after the write.
@@ -1015,6 +1046,11 @@ func (m *TableManager) updateLocked(table string, updateFn func(Row) (Row, error
 // same row fails with a serialization error instead of being silently lost.
 func (m *TableManager) UpdateByPK(table, pk string, updateFn func(Row) (Row, error)) (Row, bool, error) {
 	key := m.dataKey(table, pk)
+	unlockFK, err := m.lockForeignKeyWrites(table)
+	if err != nil {
+		return nil, false, err
+	}
+	defer unlockFK()
 	unlock, uniq, err := m.lockForWrite(table)
 	if err != nil {
 		return nil, false, err
@@ -1061,6 +1097,12 @@ func (m *TableManager) UpdateByPK(table, pk string, updateFn func(Row) (Row, err
 	if err := m.schema.validateChecks(schema, row); err != nil {
 		return nil, false, err
 	}
+	if err := m.validateInsertForeignKeys(schema, []Row{row}); err != nil {
+		return nil, false, err
+	}
+	if err := m.validateParentKeyRemoval(table, oldRow, row); err != nil {
+		return nil, false, err
+	}
 	data, err := encodeRow(row)
 	if err != nil {
 		return nil, false, err
@@ -1123,6 +1165,12 @@ func (m *TableManager) DeleteByPK(table, pk string) (Row, bool, error) {
 		return nil, false, err
 	}
 
+	unlockFK, err := m.lockForeignKeyWrites(table)
+	if err != nil {
+		return nil, false, err
+	}
+	defer unlockFK()
+
 	key := m.dataKey(table, pk)
 	tl := m.tableLock(table)
 	tl.RLock()
@@ -1140,6 +1188,10 @@ func (m *TableManager) DeleteByPK(table, pk string) (Row, bool, error) {
 		return nil, false, err
 	}
 
+	if err := m.validateParentKeyRemoval(table, row, nil); err != nil {
+		return nil, false, err
+	}
+
 	committed, err := m.compareWritePoint(
 		[]CompareCheck{{Key: []byte(key), LSN: lsn}},
 		[]BatchOp{{Op: batchDelete, Key: []byte(key)}},

pkg/storage/tx.go

@@ -169,6 +169,29 @@ func (s *Session) CommitWithLSN() (uint64, error) {
 		return 0, fmt.Errorf("current transaction is aborted")
 	}
 
+	// Referential integrity spans tables: a commit must validate the final
+	// child rows against the final parent rows, including rows another table
+	// staged in this same transaction. Take the foreign-key write gate before
+	// any table gate (the canonical lock order) and register the constrained
+	// component as scan reads so acquireCommitLocks holds it exclusively while
+	// the validation scans run. A concurrent write on either side would bump one
+	// of those generations and fail the commit with 40001 instead of committing
+	// a dangling reference.
+	closure, err := s.foreignKeyClosure()
+	if err != nil {
+		s.resetLocked()
+		return 0, err
+	}
+	if len(closure) > 0 {
+		s.table.foreignKeyMu.Lock()
+		defer s.table.foreignKeyMu.Unlock()
+	}
+	for table := range closure {
+		if _, ok := s.scanGens[table]; !ok {
+			s.scanGens[table] = s.table.generation(table)
+		}
+	}
+
 	// Acquire the per-table gates, escalating any table whose UNIQUE index
 	// appeared after the initial unlocked probe. See acquireCommitLocks.
 	_, _, unlockAll, uniqueTables, err := s.acquireCommitLocks()
@@ -205,6 +228,16 @@ func (s *Session) CommitWithLSN() (uint64, error) {
 		}
 	}
 
+	// Validate the final overlay against every foreign key in the constrained
+	// component. The per-table gates acquired above exclude concurrent writers,
+	// so this sees a stable state; checking the final state (rather than each
+	// statement) is what lets a transaction delete a parent and its children in
+	// either order as long as the committed state stays consistent.
+	if err := s.validateForeignKeyFinalState(closure); err != nil {
+		s.resetLocked()
+		return 0, err
+	}
+
 	// Validate the final overlay of every unique-indexed written table against
 	// durable rows. This runs under the exclusive gate acquired above, so it
 	// cannot race a concurrent writer. Duplicate unique values fail the commit