feat: enforce FOREIGN KEY constraints

Accept inline REFERENCES parent(col) and table-level FOREIGN KEY (a, b)
REFERENCES parent(x, y) in CREATE TABLE, named or unnamed. At DDL the
parent table must already exist and each referenced column set must be the
parent's PRIMARY KEY or exactly the columns of a UNIQUE index; otherwise
the statement fails with 0A000 and a "foreign key mismatch" explanation
instead of storing a constraint the engine cannot enforce. Self-references
resolve against the table being created. ON DELETE/ON UPDATE CASCADE,
SET NULL, SET DEFAULT, and DEFERRABLE are rejected with 0A000 (NO ACTION
and RESTRICT are accepted and enforced immediately).

Enforcement lives in storage so every write path is covered:

- Point writes (INSERT, INSERT ... SELECT, UPDATE of the foreign-key
  columns, parent DELETE/UPDATE) validate before the durable write and
  report a typed ForeignKeyViolationError (SQLSTATE 23503).
- Scan-based UPDATE/DELETE and multi-row statements stage through the
  session and are validated at COMMIT against the final row image of the
  foreign-key-connected component, so a staged child insert plus a staged
  parent delete cannot commit into a dangling reference, while deleting a
  parent and its children in one transaction succeeds in either order.
- MATCH SIMPLE semantics: a partially NULL child key is not enforced.
  Self-referential rows may reference themselves or a sibling row of the
  same statement.
- A foreign-key write gate serializes conflicting writes, and the closure
  is registered as scan reads so a concurrent writer fails the commit with
  40001 rather than letting a dangling reference through.

The constraints persist in the schema, survive reload, and appear in the
emulated sqlite_master/sqlite_schema CREATE TABLE SQL. 23503 is wired
through the PostgreSQL simple/extended protocol and the HTTP endpoints;
pg_constraint and information_schema foreign keys are still not emulated.
PRAGMA foreign_keys now reports 1 (always enforced) and rejects OFF.

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; ON DELETE/ON UPDATE CASCADE, SET NULL, SET DEFAULT, and DEFERRABLE are rejected with 0A000; reported in sqlite_master/sqlite_schema table SQL only (pg_constraint and information_schema foreign keys are not emulated)"},
 			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"},
 		},
@@ -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,28 @@ 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", "self", "CASCADE", "DEFERRABLE", "0A000", "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

@@ -3781,19 +3781,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 +3838,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 +3848,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 +3888,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
@@ -4283,8 +4290,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,
 			)
 		}
 	}
@@ -4795,19 +4802,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, fmt.Errorf("feature not supported: foreign keys are always enforced; PRAGMA foreign_keys = OFF is not supported")
+			}
 			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, fmt.Errorf("feature not supported: foreign keys are always enforced; PRAGMA foreign_keys = OFF is not supported")
+			}
 			return result, nil
 		}
-		return nil, fmt.Errorf("PizzaSQL does not enforce foreign keys; only PRAGMA foreign_keys = OFF is supported")
+		return nil, fmt.Errorf("feature not supported: PRAGMA foreign_keys accepts ON or OFF")
 	default:
 		return nil, fmt.Errorf("unknown pragma: %s", stmt.Name)
 	}

pkg/executor/foreign_key_test.go

@@ -0,0 +1,638 @@
+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)
+
+	// A self-referencing row cannot be deleted while it points at itself.
+	_, err = execSQL(e, "DELETE FROM nodes WHERE id = 1")
+	requireForeignKeyViolation(t, err)
+
+	execMust(t, e, "INSERT INTO nodes (id, parent_id) VALUES (5, 5)")
+	execMust(t, e, "DELETE FROM nodes WHERE id = 5")
+
+	// 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)
+	}
+}
+
+// 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)
+	}
+}

pkg/executor/foreign_keys.go

@@ -0,0 +1,240 @@
+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,
+	}, 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
+}

pkg/executor/review_fixes_test.go

@@ -195,19 +195,19 @@ 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")
 	}
 }
 

pkg/executor/sqlite_catalog.go

@@ -339,6 +339,23 @@ 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, ", "))
+		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,69 @@
+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)
+	}
+}

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,460 @@
+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
+}
+
+// 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
+}
+
+// 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,29 @@ 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"`
+}
+
+// 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 +173,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 +499,17 @@ 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...),
+			}
+		}
+	}
 	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