Merge pull request #32 from database-pizza/engine/b-check

engine: implement CHECK constraints

PKBFI_STORAGE_MIGRATION.md

@@ -30,6 +30,10 @@ The migration command exits after writing and verifying the destination. It refu
 - Migration copies the final live key/value state. Legacy journal history is not imported, and imported records begin at the migration baseline LSN.
 - Existing PizzaSQL row values remain readable as legacy JSON after migration. New and updated rows use the versioned binary tuple encoding, so an eager row rewrite is unnecessary.
 - Schema and catalog values remain JSON because they are cold metadata.
+- `Schema.Checks` (CHECK constraints) is an additive, optional JSON field. A
+  pre-CHECK engine ignores it on read and drops it on the next schema write, so
+  rolling back to such an engine silently stops enforcing CHECK constraints;
+  restore the newer engine (or re-create the constraints) after a rollback.
 - Successful PKBFI writes are acknowledged only after PizzaKV's durability sync completes.
 - A `.pkvdb` file must have one PizzaSQL process owner. Optimistic scan and
   predicate conflict detection uses process-local generations and is not safe

pkg/capabilities/capabilities.go

@@ -170,15 +170,15 @@ func Current() *Document {
 			PrimaryKey:    Feature{Supported: true, Status: "enforced", SQLState: "23505", Notes: "single-column only"},
 			CompositePK:   Feature{Supported: false, Status: "rejected", SQLState: "0A000", Notes: "declare a single-column primary key"},
 			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"},
-			Check:         Feature{Supported: false, Status: "rejected", SQLState: "0A000", Notes: "inline and table-level CHECK are rejected; enforce the rule in the application"},
+			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: "constant_only", SQLState: "0A000", Notes: "non-constant defaults (CURRENT_TIMESTAMP, function calls) are rejected instead of being frozen at DDL time"},
 		},
 
 		DDL: DDL{
 			AddColumn:              Feature{Supported: true, Status: "materializes_constant_default", Notes: "adds one nullable column; a constant DEFAULT is written into every existing row, so the rewrite cost is O(rows) and holds the table's exclusive gate"},
-			AddColumnBackfill:      Feature{Supported: true, Status: "rewrites_rows", Notes: "a constant DEFAULT is materialized into existing rows at DDL time; ADD COLUMN ... NOT NULL without a DEFAULT fails with 23502 on a non-empty table, ADD COLUMN ... UNIQUE conflicts with 23505 when backfilling a non-NULL default into more than one row, and the column is published together with its UNIQUE index in one atomic schema write"},
-			DropColumn:             Feature{Supported: true, Status: "rewrites_rows"},
-			RenameColumn:           Feature{Supported: true, Status: "rewrites_rows"},
+			AddColumnBackfill:      Feature{Supported: true, Status: "rewrites_rows", Notes: "a constant DEFAULT is materialized into existing rows at DDL time; ADD COLUMN ... NOT NULL without a DEFAULT fails with 23502 on a non-empty table, ADD COLUMN ... UNIQUE conflicts with 23505 when backfilling a non-NULL default into more than one row, ADD COLUMN ... CHECK fails with 23514 when an existing row or the backfilled default violates the predicate, and the column is published together with its CHECK constraints and UNIQUE index in one atomic schema write"},
+			DropColumn:             Feature{Supported: true, Status: "rewrites_rows", Notes: "rejected with 0A000 when a persisted CHECK constraint references the column; the stored predicate text cannot be rewritten"},
+			RenameColumn:           Feature{Supported: true, Status: "rewrites_rows", Notes: "rejected with 0A000 when a persisted CHECK constraint or expression index references the column; the stored expression text cannot be rewritten"},
 			RenameColumnIndexes:    Feature{Supported: true, Status: "rewrites_index_metadata", Notes: "column references in index definitions are rewritten in place, including composite and unique indexes; index names are unchanged; RENAME COLUMN is rejected when an expression index references the column (drop that index, rename, then recreate it)"},
 			CreateTableIfNotExists: true,
 			AddColumnIfNotExists:   true,
@@ -224,6 +224,7 @@ func Current() *Document {
 		ErrorCodes: map[string]string{
 			"23502": "not_null_violation",
 			"23505": "unique_violation",
+			"23514": "check_violation",
 			"40001": "serialization_failure",
 			"0A000": "feature_not_supported",
 			"42501": "insufficient_privilege (managed proxy)",

pkg/capabilities/capabilities_test.go

@@ -2,8 +2,10 @@ package capabilities
 
 import (
 	"encoding/json"
+	"strings"
 	"testing"
 
+	"github.com/danfragoso/pizzasql-next/pkg/pgerror"
 	"github.com/danfragoso/pizzasql-next/pkg/version"
 )
 
@@ -69,3 +71,36 @@ func TestAddColumnBackfillIsAdvertised(t *testing.T) {
 		t.Fatalf("add_column status = %q, but a constant DEFAULT rewrites rows", doc.DDL.AddColumn.Status)
 	}
 }
+
+func TestCheckConstraintIsAdvertised(t *testing.T) {
+	doc := Current()
+	check := doc.Constraints.Check
+	if !check.Supported {
+		t.Fatalf("check = %+v, want supported", check)
+	}
+	if check.Status != "enforced" {
+		t.Fatalf("check status = %q, want enforced", check.Status)
+	}
+	if check.SQLState != pgerror.CodeCheckViolation {
+		t.Fatalf("check SQLSTATE = %q, want %q", check.SQLState, pgerror.CodeCheckViolation)
+	}
+	if check.Notes == "" {
+		t.Fatal("check notes must describe the enforcement surface")
+	}
+	if got := doc.ErrorCodes[pgerror.CodeCheckViolation]; got != "check_violation" {
+		t.Fatalf("23514 mapping = %q, want check_violation", got)
+	}
+}
+
+func TestDDLNotesDescribeCheckDependencies(t *testing.T) {
+	doc := Current()
+	if !strings.Contains(doc.DDL.DropColumn.Notes, "CHECK") || !strings.Contains(doc.DDL.DropColumn.Notes, "0A000") {
+		t.Fatalf("drop_column notes = %q, want the persisted CHECK restriction", doc.DDL.DropColumn.Notes)
+	}
+	if !strings.Contains(doc.DDL.RenameColumn.Notes, "CHECK") || !strings.Contains(doc.DDL.RenameColumn.Notes, "0A000") {
+		t.Fatalf("rename_column notes = %q, want the persisted CHECK restriction", doc.DDL.RenameColumn.Notes)
+	}
+	if !strings.Contains(doc.DDL.AddColumnBackfill.Notes, "23514") {
+		t.Fatalf("add_column_backfill notes = %q, want the CHECK backfill failure", doc.DDL.AddColumnBackfill.Notes)
+	}
+}

pkg/executor/check_constraint_test.go

@@ -0,0 +1,525 @@
+package executor
+
+import (
+	"errors"
+	"strings"
+	"testing"
+
+	"github.com/danfragoso/pizzasql-next/pkg/lexer"
+	"github.com/danfragoso/pizzasql-next/pkg/parser"
+	"github.com/danfragoso/pizzasql-next/pkg/pgerror"
+	"github.com/danfragoso/pizzasql-next/pkg/storage"
+)
+
+// requireCheckViolation asserts that err is a CHECK violation for constraint.
+func requireCheckViolation(t *testing.T, err error, constraint string) {
+	t.Helper()
+	var violation *storage.CheckViolationError
+	if !errors.As(err, &violation) {
+		t.Fatalf("error = %v, want CheckViolationError", err)
+	}
+	if constraint != "" && violation.Constraint != constraint {
+		t.Fatalf("constraint = %q, want %q", violation.Constraint, constraint)
+	}
+	if !strings.HasPrefix(err.Error(), "CHECK constraint failed: ") {
+		t.Fatalf("message = %q, want the CHECK constraint failure prefix", err.Error())
+	}
+}
+
+// Inline, table-level, named, and unnamed CHECK declarations all persist and
+// enforce the same predicate.
+func TestCheckConstraintDeclarationForms(t *testing.T) {
+	tests := []struct {
+		name       string
+		ddl        string
+		constraint string // expected label on violation
+	}{
+		{
+			name: "inline unnamed",
+			ddl:  "CREATE TABLE t (id INTEGER PRIMARY KEY, age INTEGER CHECK (age > 0))",
+		},
+		{
+			name: "table level unnamed",
+			ddl:  "CREATE TABLE t (id INTEGER PRIMARY KEY, age INTEGER, CHECK (age > 0))",
+		},
+		{
+			name:       "column level named",
+			ddl:        "CREATE TABLE t (id INTEGER PRIMARY KEY, age INTEGER CONSTRAINT positive_age CHECK (age > 0))",
+			constraint: "positive_age",
+		},
+		{
+			name:       "table level named",
+			ddl:        "CREATE TABLE t (id INTEGER PRIMARY KEY, age INTEGER, CONSTRAINT positive_age CHECK (age > 0))",
+			constraint: "positive_age",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			_, schema, table := newTestDB(t)
+			e := newExec(schema, table)
+			execMust(t, e, tt.ddl)
+
+			stored, err := schema.GetSchema("t")
+			if err != nil {
+				t.Fatalf("get schema: %v", err)
+			}
+			if len(stored.Checks) != 1 {
+				t.Fatalf("checks = %+v, want exactly one persisted CHECK", stored.Checks)
+			}
+			if tt.constraint != "" && stored.Checks[0].Name != tt.constraint {
+				t.Fatalf("check name = %q, want %q", stored.Checks[0].Name, tt.constraint)
+			}
+
+			if _, err := execSQL(e, "INSERT INTO t (id, age) VALUES (1, -1)"); err == nil {
+				t.Fatal("insert of a violating row succeeded")
+			} else {
+				requireCheckViolation(t, err, tt.constraint)
+			}
+			execMust(t, e, "INSERT INTO t (id, age) VALUES (2, 1)")
+			execMust(t, e, "INSERT INTO t (id, age) VALUES (3, NULL)")
+		})
+	}
+}
+
+// A CHECK predicate that is NULL or unknown for a row passes, matching SQL
+// three-valued logic, on INSERT and on UPDATE.
+func TestCheckConstraintNullAndUnknownPass(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE t (id INTEGER PRIMARY KEY, age INTEGER, CHECK (age > 0 AND age < 200))")
+
+	execMust(t, e, "INSERT INTO t (id, age) VALUES (1, NULL)")
+	execMust(t, e, "INSERT INTO t (id) VALUES (2)")
+
+	execMust(t, e, "INSERT INTO t (id, age) VALUES (3, 30)")
+	execMust(t, e, "UPDATE t SET age = NULL WHERE id = 3")
+
+	res := execMust(t, e, "SELECT id, age FROM t ORDER BY id")
+	if res.RowCount != 3 {
+		t.Fatalf("rows = %d, want 3 nullable rows", res.RowCount)
+	}
+	for _, row := range res.Rows {
+		if row[1] != nil {
+			t.Fatalf("age = %v, want NULL", row[1])
+		}
+	}
+}
+
+// Multi-row INSERT is atomic: a violating row rejects the whole statement and
+// leaves no rows behind.
+func TestCheckConstraintMultiRowInsertIsAtomic(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE t (id INTEGER PRIMARY KEY, age INTEGER CHECK (age > 0))")
+
+	_, err := execSQL(e, "INSERT INTO t (id, age) VALUES (1, 10), (2, -5)")
+	requireCheckViolation(t, err, "")
+
+	if res := execMust(t, e, "SELECT id FROM t"); res.RowCount != 0 {
+		t.Fatalf("rows after rejected multi-row insert = %d, want 0", res.RowCount)
+	}
+
+	execMust(t, e, "INSERT INTO t (id, age) VALUES (1, 10), (2, 20)")
+	if res := execMust(t, e, "SELECT id FROM t ORDER BY id"); res.RowCount != 2 {
+		t.Fatalf("rows = %d, want 2", res.RowCount)
+	}
+}
+
+// UPDATE re-evaluates every CHECK predicate against the post-assignment row and
+// rolls the row back when a predicate fails.
+func TestCheckConstraintUpdateViolation(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE t (id INTEGER PRIMARY KEY, age INTEGER CHECK (age > 0))")
+	execMust(t, e, "INSERT INTO t (id, age) VALUES (1, 10)")
+
+	_, err := execSQL(e, "UPDATE t SET age = -1 WHERE id = 1")
+	requireCheckViolation(t, err, "")
+
+	res := execMust(t, e, "SELECT age FROM t WHERE id = 1")
+	if res.RowCount != 1 || res.Rows[0][0] != int64(10) {
+		t.Fatalf("age after rejected update = %v, want 10", res.Rows)
+	}
+
+	execMust(t, e, "UPDATE t SET age = age + 5 WHERE id = 1")
+	res = execMust(t, e, "SELECT age FROM t WHERE id = 1")
+	if res.Rows[0][0] != int64(15) {
+		t.Fatalf("age after valid update = %v, want 15", res.Rows[0][0])
+	}
+}
+
+// ON CONFLICT DO UPDATE validates the merged row against the CHECK predicates
+// before it is staged.
+func TestCheckConstraintOnConflictDoUpdateViolation(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE t (id INTEGER PRIMARY KEY, qty INTEGER CHECK (qty > 0))")
+	execMust(t, e, "INSERT INTO t (id, qty) VALUES (1, 5)")
+
+	_, err := execSQL(e, "INSERT INTO t (id, qty) VALUES (1, -2) ON CONFLICT (id) DO UPDATE SET qty = excluded.qty")
+	requireCheckViolation(t, err, "")
+
+	res := execMust(t, e, "SELECT qty FROM t WHERE id = 1")
+	if res.Rows[0][0] != int64(5) {
+		t.Fatalf("qty after rejected upsert = %v, want 5", res.Rows[0][0])
+	}
+
+	execMust(t, e, "INSERT INTO t (id, qty) VALUES (1, 7) ON CONFLICT (id) DO UPDATE SET qty = excluded.qty")
+	res = execMust(t, e, "SELECT qty FROM t WHERE id = 1")
+	if res.Rows[0][0] != int64(7) {
+		t.Fatalf("qty after valid upsert = %v, want 7", res.Rows[0][0])
+	}
+}
+
+// Every CHECK on a table is enforced, and the violation names the constraint
+// that rejected the row.
+func TestCheckConstraintMultipleChecks(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, `CREATE TABLE t (
+		id INTEGER PRIMARY KEY,
+		age INTEGER CHECK (age > 0),
+		score INTEGER,
+		CONSTRAINT score_range CHECK (score >= 0 AND score <= 100)
+	)`)
+
+	stored, err := schema.GetSchema("t")
+	if err != nil {
+		t.Fatalf("get schema: %v", err)
+	}
+	if len(stored.Checks) != 2 {
+		t.Fatalf("checks = %+v, want 2", stored.Checks)
+	}
+
+	_, err = execSQL(e, "INSERT INTO t (id, age, score) VALUES (1, -1, 50)")
+	requireCheckViolation(t, err, "")
+
+	_, err = execSQL(e, "INSERT INTO t (id, age, score) VALUES (2, 10, 101)")
+	requireCheckViolation(t, err, "score_range")
+
+	execMust(t, e, "INSERT INTO t (id, age, score) VALUES (3, 10, 100)")
+	if res := execMust(t, e, "SELECT id FROM t"); res.RowCount != 1 {
+		t.Fatalf("rows = %d, want 1", res.RowCount)
+	}
+}
+
+// ADD COLUMN ... CHECK validates the rows that already exist, using the value
+// the constant-default backfill will write (or NULL when there is no default).
+func TestAlterTableAddColumnCheckValidatesExistingRows(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE t (id INTEGER PRIMARY KEY)")
+	execMust(t, e, "INSERT INTO t (id) VALUES (1), (2)")
+
+	// A default satisfying the predicate is backfilled and the DDL succeeds.
+	execMust(t, e, "ALTER TABLE t ADD COLUMN qty INTEGER DEFAULT 5 CHECK (qty >= 0)")
+
+	stored, err := schema.GetSchema("t")
+	if err != nil {
+		t.Fatalf("get schema: %v", err)
+	}
+	if _, ok := stored.GetColumn("qty"); !ok {
+		t.Fatal("ADD COLUMN ... CHECK published no column")
+	}
+	if len(stored.Checks) != 1 {
+		t.Fatalf("checks = %+v, want 1", stored.Checks)
+	}
+	res := execMust(t, e, "SELECT qty FROM t ORDER BY id")
+	if len(res.Rows) != 2 || res.Rows[0][0] != int64(5) || res.Rows[1][0] != int64(5) {
+		t.Fatalf("backfilled qty = %v, want [5 5]", res.Rows)
+	}
+
+	// A violation is enforced for later writes as well.
+	if _, err := execSQL(e, "INSERT INTO t (id, qty) VALUES (3, -1)"); err == nil {
+		t.Fatal("insert violating the ADD COLUMN check succeeded")
+	} else {
+		requireCheckViolation(t, err, "")
+	}
+
+	// A default that violates the predicate rejects the DDL and publishes
+	// nothing: no column, no constraint, and no rewritten rows.
+	_, err = execSQL(e, "ALTER TABLE t ADD COLUMN bad INTEGER DEFAULT -1 CHECK (bad >= 0)")
+	requireCheckViolation(t, err, "")
+	stored, err = schema.GetSchema("t")
+	if err != nil {
+		t.Fatalf("get schema after rejected DDL: %v", err)
+	}
+	if _, ok := stored.GetColumn("bad"); ok {
+		t.Fatal("rejected ADD COLUMN ... CHECK left the column in the schema")
+	}
+	if len(stored.Checks) != 1 {
+		t.Fatalf("checks after rejected DDL = %+v, want the original one", stored.Checks)
+	}
+
+	// Without a default the existing rows read NULL, which passes.
+	execMust(t, e, "ALTER TABLE t ADD COLUMN note TEXT CHECK (LENGTH(note) > 0)")
+	if res := execMust(t, e, "SELECT id, note FROM t ORDER BY id"); res.RowCount != 2 || res.Rows[0][1] != nil {
+		t.Fatalf("rows after nullable ADD COLUMN = %v, want NULL notes", res.Rows)
+	}
+}
+
+// A transaction that stages a row before a concurrent ADD COLUMN ... CHECK is
+// validated at commit, so the backfilled default cannot bypass the predicate.
+func TestAlterTableAddColumnCheckRejectsRowStagedBeforeDDL(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	writer := newExec(schema, table)
+	ddl := newExec(schema, table)
+	execMust(t, writer, "CREATE TABLE t (id INTEGER PRIMARY KEY)")
+
+	execMust(t, writer, "BEGIN")
+	execMust(t, writer, "INSERT INTO t (id) VALUES (1)")
+
+	// The table is durably empty, so the ALTER accepts the constraint.
+	execMust(t, ddl, "ALTER TABLE t ADD COLUMN qty INTEGER DEFAULT -1 CHECK (qty >= 0)")
+
+	_, err := execSQL(writer, "COMMIT")
+	requireCheckViolation(t, err, "")
+
+	if got := execMust(t, ddl, "SELECT id FROM t").RowCount; got != 0 {
+		t.Fatalf("rows after rejected commit = %d, want 0", got)
+	}
+}
+
+// The persisted predicate is durable: a fresh SchemaManager loaded from the same
+// store keeps enforcing it.
+func TestCheckConstraintDurableReload(t *testing.T) {
+	pool, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE t (id INTEGER PRIMARY KEY, age INTEGER, CONSTRAINT positive_age CHECK (age > 0))")
+	execMust(t, e, "INSERT INTO t (id, age) VALUES (1, 10)")
+
+	reloaded := storage.NewSchemaManager(pool, "testdb")
+	reloadedTable := storage.NewTableManager(pool, reloaded, "testdb")
+	reopened := newExec(reloaded, reloadedTable)
+
+	stored, err := reloaded.GetSchema("t")
+	if err != nil {
+		t.Fatalf("reload schema: %v", err)
+	}
+	if len(stored.Checks) != 1 || stored.Checks[0].Name != "positive_age" || !strings.Contains(stored.Checks[0].Expression, "age") {
+		t.Fatalf("reloaded checks = %+v, want the persisted positive_age predicate", stored.Checks)
+	}
+
+	_, err = execSQL(reopened, "INSERT INTO t (id, age) VALUES (2, -3)")
+	requireCheckViolation(t, err, "positive_age")
+
+	execMust(t, reopened, "INSERT INTO t (id, age) VALUES (3, 30)")
+	execMust(t, reopened, "UPDATE t SET age = 40 WHERE id = 3")
+	if _, err := execSQL(reopened, "UPDATE t SET age = -40 WHERE id = 3"); err == nil {
+		t.Fatal("update violating the reloaded check succeeded")
+	}
+}
+
+// The emulated SQLite catalog exposes the constraint in the table SQL.
+func TestCheckConstraintInSQLiteCatalog(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE t (id INTEGER PRIMARY KEY, age INTEGER, CONSTRAINT positive_age CHECK (age > 0), CHECK (id > 0))")
+
+	res := execMust(t, e, "SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 't'")
+	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 `positive_age` CHECK", "(id > 0)"} {
+		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 = 't'")
+	if res.RowCount != 1 || !strings.Contains(res.Rows[0][0].(string), "positive_age") {
+		t.Fatalf("sqlite_schema SQL = %v, want the named CHECK", res.Rows)
+	}
+}
+
+// Non-deterministic and non-row predicates are rejected with feature-not-
+// supported instead of being accepted as constraints the engine cannot enforce.
+func TestCheckConstraintRejectsUnsupportedExpressions(t *testing.T) {
+	tests := []struct {
+		name  string
+		setup []string
+		ddl   string
+	}{
+		{
+			name: "subquery",
+			ddl:  "CREATE TABLE t (id INTEGER PRIMARY KEY, age INTEGER CHECK (age > (SELECT 1)))",
+		},
+		{
+			name: "aggregate",
+			ddl:  "CREATE TABLE t (id INTEGER PRIMARY KEY, age INTEGER CHECK (COUNT(age) > 0))",
+		},
+		{
+			name: "window function",
+			ddl:  "CREATE TABLE t (id INTEGER PRIMARY KEY, age INTEGER CHECK (ROW_NUMBER() OVER (ORDER BY age) > 0))",
+		},
+		{
+			name: "non-deterministic function",
+			ddl:  "CREATE TABLE t (id INTEGER PRIMARY KEY, age INTEGER CHECK (RANDOM() > 0))",
+		},
+		{
+			name: "qualified column",
+			ddl:  "CREATE TABLE t (id INTEGER PRIMARY KEY, age INTEGER CHECK (t.age > 0))",
+		},
+		{
+			name: "qualified column in parentheses",
+			ddl:  "CREATE TABLE t (id INTEGER PRIMARY KEY, age INTEGER CHECK ((t.age) > 0))",
+		},
+		{
+			name: "qualified column in cast",
+			ddl:  "CREATE TABLE t (id INTEGER PRIMARY KEY, age INTEGER CHECK (CAST(t.age AS INTEGER) > 0))",
+		},
+		{
+			name: "unknown column",
+			ddl:  "CREATE TABLE t (id INTEGER PRIMARY KEY, age INTEGER CHECK (missing > 0))",
+		},
+		{
+			name:  "alter add column subquery",
+			setup: []string{"CREATE TABLE t (id INTEGER PRIMARY KEY)"},
+			ddl:   "ALTER TABLE t ADD COLUMN age INTEGER CHECK (age > (SELECT 1))",
+		},
+	}
+
+	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)
+			}
+
+			if _, err := execSQL(e, tt.ddl); !errors.Is(err, pgerror.ErrFeatureNotSupported) {
+				t.Fatalf("%s: error = %v, want ErrFeatureNotSupported", tt.ddl, err)
+			}
+			// A rejected declaration publishes nothing: no table when the DDL
+			// was CREATE TABLE, and no added column or check for ALTER TABLE.
+			stored, err := schema.GetSchema("t")
+			if err != nil {
+				if len(tt.setup) == 0 {
+					return
+				}
+				t.Fatalf("get schema: %v", err)
+			}
+			if len(stored.Checks) != 0 {
+				t.Fatalf("checks after rejected DDL = %+v, want none", stored.Checks)
+			}
+			if _, ok := stored.GetColumn("age"); ok {
+				t.Fatalf("column age after rejected DDL = ok, want absent")
+			}
+		})
+	}
+}
+
+// A CHECK predicate is stored as SQL text, so DROP/RENAME COLUMN refuse to leave
+// it referencing a column that no longer exists.
+func TestCheckConstraintGuardsColumnChange(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE t (id INTEGER PRIMARY KEY, age INTEGER CHECK (age > 0), note TEXT)")
+
+	for _, sql := range []string{
+		"ALTER TABLE t RENAME COLUMN age TO years",
+		"ALTER TABLE t DROP COLUMN age",
+	} {
+		if _, err := execSQL(e, sql); !errors.Is(err, pgerror.ErrFeatureNotSupported) {
+			t.Fatalf("%s: error = %v, want ErrFeatureNotSupported", sql, err)
+		}
+	}
+
+	// Columns the predicate does not reference can still be renamed or dropped.
+	execMust(t, e, "ALTER TABLE t RENAME COLUMN note TO memo")
+	execMust(t, e, "INSERT INTO t (id, age, memo) VALUES (1, 5, 'ok')")
+	execMust(t, e, "ALTER TABLE t DROP COLUMN memo")
+}
+
+// The accepted expression surface survives the round trip through the durable
+// schema text: each predicate is re-parsed and still enforced after a reload.
+func TestCheckConstraintExpressionRoundTrip(t *testing.T) {
+	tests := []struct {
+		name     string
+		check    string
+		columns  string
+		valid    string // VALUES list for a row that satisfies the predicate
+		violated string // VALUES list for a row that violates it
+	}{
+		{"comparison and boolean", "age > 0 AND age < 200", "id, age", "(1, 10)", "(1, 500)"},
+		{"LIKE", "name LIKE 'a%'", "id, name", "(1, 'abc')", "(1, 'zzz')"},
+		{"CAST", "CAST(score AS INTEGER) >= 0", "id, score", "(1, 5)", "(1, -5)"},
+		{"BETWEEN", "LENGTH(code) BETWEEN 1 AND 8", "id, code", "(1, 'abcd')", "(1, 'toolongcode')"},
+		{"function", "LOWER(status) = 'ready'", "id, status", "(1, 'READY')", "(1, 'waiting')"},
+		{"is null", "note IS NOT NULL", "id, note", "(1, 'x')", "(1, NULL)"},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			pool, schema, table := newTestDB(t)
+			e := newExec(schema, table)
+			execMust(t, e, `CREATE TABLE t (
+				id INTEGER PRIMARY KEY,
+				age INTEGER,
+				name TEXT,
+				score INTEGER,
+				code TEXT,
+				status TEXT,
+				note TEXT,
+				CHECK (`+tt.check+`)
+			)`)
+			execMust(t, e, "INSERT INTO t ("+tt.columns+") VALUES "+tt.valid)
+
+			reloaded := storage.NewSchemaManager(pool, "testdb")
+			reopened := newExec(reloaded, storage.NewTableManager(pool, reloaded, "testdb"))
+
+			if _, err := execSQL(reopened, "INSERT INTO t ("+tt.columns+") VALUES "+tt.violated); err == nil {
+				t.Fatalf("reloaded table accepted a row violating %s", tt.check)
+			} else {
+				requireCheckViolation(t, err, "")
+			}
+		})
+	}
+}
+
+// The persisted predicate text must round-trip. A quoted keyword column is the
+// motivating case: FormatExpr quotes it, so CREATE TABLE, later writes, and a
+// durable reload all keep binding to the column instead of the keyword.
+func TestCheckConstraintQuotedKeywordColumn(t *testing.T) {
+	pool, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, `CREATE TABLE t (id INTEGER PRIMARY KEY, "order" INTEGER CHECK ("order" > 0))`)
+
+	stored, err := schema.GetSchema("t")
+	if err != nil {
+		t.Fatalf("get schema: %v", err)
+	}
+	if len(stored.Checks) != 1 {
+		t.Fatalf("checks = %+v, want 1", stored.Checks)
+	}
+	if !strings.Contains(stored.Checks[0].Expression, `"order"`) {
+		t.Fatalf("stored expression = %q, want the keyword column quoted", stored.Checks[0].Expression)
+	}
+
+	_, err = execSQL(e, `INSERT INTO t (id, "order") VALUES (1, -1)`)
+	requireCheckViolation(t, err, "")
+	execMust(t, e, `INSERT INTO t (id, "order") VALUES (1, 3)`)
+
+	// The durable text stays evaluable after reload.
+	reloaded := storage.NewSchemaManager(pool, "testdb")
+	reopened := newExec(reloaded, storage.NewTableManager(pool, reloaded, "testdb"))
+	_, err = execSQL(reopened, `INSERT INTO t (id, "order") VALUES (2, -2)`)
+	requireCheckViolation(t, err, "")
+	execMust(t, reopened, `INSERT INTO t (id, "order") VALUES (2, 4)`)
+}
+
+// A predicate whose formatted text cannot be re-parsed is rejected at DDL time
+// instead of being published as a constraint that fails every later write.
+func TestCheckConstraintRejectsNonRoundTrippingExpression(t *testing.T) {
+	schema := &storage.Schema{Name: "t", Columns: []storage.Column{{Name: "id", Type: "INTEGER"}}}
+	// FormatExpr renders a number literal verbatim; this one is not valid SQL.
+	_, err := checkConstraintFor("", &parser.LiteralExpr{Type: lexer.TokenNumber, Value: "1 +"}, schema)
+	if !errors.Is(err, pgerror.ErrFeatureNotSupported) {
+		t.Fatalf("error = %v, want ErrFeatureNotSupported", err)
+	}
+	if !strings.Contains(err.Error(), "cannot be stored") {
+		t.Fatalf("error = %q, want the round-trip explanation", err.Error())
+	}
+}

pkg/executor/check_fk_test.go

@@ -7,26 +7,6 @@ import (
 	"github.com/danfragoso/pizzasql-next/pkg/pgerror"
 )
 
-func TestCheckConstraintsRejectedAtDDL(t *testing.T) {
-	_, schema, table := newTestDB(t)
-	e := newExec(schema, table)
-
-	for _, sql := range []string{
-		"CREATE TABLE t1 (id INTEGER PRIMARY KEY, age INTEGER CHECK (age > 0))",
-		"CREATE TABLE t2 (id INTEGER PRIMARY KEY, age INTEGER, CHECK (age > 0))",
-		"CREATE TABLE t3 (id INTEGER PRIMARY KEY, age INTEGER, CONSTRAINT positive_age CHECK (age > 0))",
-	} {
-		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 age INTEGER CHECK (age > 0)"); !errors.Is(err, pgerror.ErrFeatureNotSupported) {
-		t.Fatalf("ALTER CHECK error = %v, want ErrFeatureNotSupported", err)
-	}
-}
-
 func TestForeignKeyConstraintsRejectedAtDDL(t *testing.T) {
 	_, schema, table := newTestDB(t)
 	e := newExec(schema, table)

pkg/executor/checks.go

@@ -0,0 +1,132 @@
+package executor
+
+import (
+	"fmt"
+
+	"github.com/danfragoso/pizzasql-next/pkg/parser"
+	"github.com/danfragoso/pizzasql-next/pkg/pgerror"
+	"github.com/danfragoso/pizzasql-next/pkg/storage"
+)
+
+// EvalCheckExpression evaluates a persisted CHECK predicate against a row. It is
+// the storage layer's CHECK evaluator, stateless and concurrency-safe like
+// EvalStoredExpression.
+//
+// The result follows SQL three-valued logic: NULL and any other value that is
+// not definitely false passes the constraint. The truth test is the engine's
+// WHERE predicate test (toBool), so a CHECK and an equivalent WHERE filter agree
+// on the same row.
+func EvalCheckExpression(expression string, row storage.Row) (bool, error) {
+	expr, err := parseStoredExpr(expression)
+	if err != nil {
+		return false, err
+	}
+	value, err := statelessEval.evalExpr(expr, row)
+	if err != nil {
+		return false, err
+	}
+	if value == nil {
+		return true, nil
+	}
+	return toBool(value), nil
+}
+
+// checkConstraintFor turns a parsed CHECK predicate into its durable form. Only
+// deterministic, subquery-free predicates over unqualified columns of the table
+// are accepted: the expression is persisted as SQL text and re-evaluated on
+// every write, so an aggregate, window function, subquery, or clock/random
+// function could not be enforced consistently.
+func checkConstraintFor(name string, expr parser.Expr, schema *storage.Schema) (storage.CheckConstraint, error) {
+	text := parser.FormatExpr(expr)
+	if err := ValidateDeterministicExpr(expr); err != nil {
+		return storage.CheckConstraint{}, pgerror.Unsupported(
+			"CHECK expression %s is not allowed: %v; a CHECK predicate must be deterministic and free of subqueries, aggregates, and window functions",
+			text, err,
+		)
+	}
+	for _, ref := range collectTableColumnRefs(expr) {
+		if ref.tbl != "" {
+			return storage.CheckConstraint{}, pgerror.Unsupported(
+				"CHECK expression %s uses the qualified column reference %s.%s; name the column without a table qualifier so the predicate stays bound to this table",
+				text, ref.tbl, ref.col,
+			)
+		}
+	}
+	if err := ValidateIndexColumns(expr, schema); err != nil {
+		return storage.CheckConstraint{}, pgerror.Unsupported(
+			"CHECK expression %s is invalid: %v", text, err,
+		)
+	}
+	// The predicate is persisted as this exact text and re-parsed by the
+	// evaluator on every later write. Reject a DDL whose formatted text does not
+	// round-trip instead of publishing a constraint that fails every write.
+	if _, err := parseStoredExpr(text); err != nil {
+		return storage.CheckConstraint{}, pgerror.Unsupported(
+			"CHECK expression %s cannot be stored for later evaluation: %v; the predicate must round-trip through the engine's SQL text form",
+			text, err,
+		)
+	}
+	return storage.CheckConstraint{Name: name, Expression: text}, nil
+}
+
+// tableChecks appends the parsed CHECK constraints of a column definition or a
+// table-level constraint list. The schema passed in must already contain every
+// column the predicate may reference, so callers validate after the full column
+// list is built.
+func (e *Executor) tableChecks(schema *storage.Schema, columnConstraints []parser.ColumnConstraint, tableConstraints []parser.TableConstraint) ([]storage.CheckConstraint, error) {
+	var checks []storage.CheckConstraint
+	for _, constraint := range columnConstraints {
+		if constraint.Type != parser.ConstraintCheck {
+			continue
+		}
+		check, err := checkConstraintFor(constraint.Name, constraint.Check, schema)
+		if err != nil {
+			return nil, err
+		}
+		checks = append(checks, check)
+	}
+	for _, constraint := range tableConstraints {
+		if constraint.Type != parser.ConstraintCheck {
+			continue
+		}
+		check, err := checkConstraintFor(constraint.Name, constraint.Check, schema)
+		if err != nil {
+			return nil, err
+		}
+		checks = append(checks, check)
+	}
+	return checks, nil
+}
+
+// schemaWithColumn returns the table schema plus one not-yet-published column,
+// so a CHECK predicate declared on that column can be validated against the
+// table it will apply to.
+func schemaWithColumn(schema *storage.Schema, column storage.Column) *storage.Schema {
+	extended := *schema
+	extended.Columns = append(append([]storage.Column(nil), schema.Columns...), column)
+	return &extended
+}
+
+// rejectCheckColumnChange refuses to rename or drop a column that a persisted
+// CHECK predicate references. Predicates are stored as SQL text, so the engine
+// cannot rewrite them; leaving the old identifier behind would make the
+// constraint evaluate to NULL and silently stop enforcing it.
+func (e *Executor) rejectCheckColumnChange(table, column, action string) error {
+	schema, err := e.schema.GetSchema(table)
+	if err != nil {
+		return err
+	}
+	for _, check := range schema.Checks {
+		references, err := storedExpressionReferencesColumn(check.Expression, column)
+		if err != nil {
+			return fmt.Errorf("CHECK constraint %s: %w", check.Label(), err)
+		}
+		if references {
+			return pgerror.Unsupported(
+				"cannot %s column %s: CHECK constraint %s depends on it; recreate the table without that constraint first",
+				action, column, check.Label(),
+			)
+		}
+	}
+	return nil
+}

pkg/executor/executor.go

@@ -23,6 +23,12 @@ import (
 // feature-not-supported SQLSTATE rather than a generic internal error.
 var ErrUnsupportedFeature = errors.New("unsupported feature")
 
+// ErrNoTransaction marks a transaction-control statement that found no
+// transaction to act on. Adapters may report it as a benign no-op (PostgreSQL
+// treats ROLLBACK outside a transaction as a warning) instead of an internal
+// error.
+var ErrNoTransaction = errors.New("no transaction in progress")
+
 // Executor executes SQL statements.
 type Executor struct {
 	schema   *storage.SchemaManager
@@ -113,11 +119,12 @@ func New(schema *storage.SchemaManager, table *storage.TableManager) *Executor {
 		listens:           make(map[string]bool),
 	}
 
-	// Expression indexes persist only SQL text, so the storage layer needs a
-	// stateless evaluator to compute their values. The evaluator is a package
-	// function with no per-connection state, so registering it here is
-	// idempotent and safe across concurrent executors.
+	// Expression indexes and CHECK predicates persist only SQL text, so the
+	// storage layer needs a stateless evaluator to compute their values. The
+	// evaluator is a package function with no per-connection state, so
+	// registering it here is idempotent and safe across concurrent executors.
 	table.SetExpressionEvaluator(EvalStoredExpression)
+	schema.SetCheckEvaluator(EvalCheckExpression)
 
 	// Register the main database
 	executor.attachedDatabases["main"] = &DatabaseConnection{
@@ -3711,10 +3718,8 @@ func (e *Executor) executeCreateTable(stmt *parser.CreateTableStmt) (*Result, er
 			case parser.ConstraintAutoIncrement:
 				schema.AutoIncrement = true
 			case parser.ConstraintCheck:
-				return nil, pgerror.Unsupported(
-					"CHECK constraints are not supported: column %q declares CHECK; enforce the rule in the application (the engine parsed but never evaluated CHECK, so accepting it would silently allow invalid rows)",
-					col.Name,
-				)
+				// 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)",
@@ -3726,9 +3731,9 @@ func (e *Executor) executeCreateTable(stmt *parser.CreateTableStmt) (*Result, er
 		schema.Columns = append(schema.Columns, col)
 	}
 
-	// Handle table-level constraints. CHECK and FOREIGN KEY are rejected: they
-	// were previously parsed and discarded, which let applications believe a
-	// constraint existed when nothing enforced it.
+	// Handle table-level constraints. CHECK predicates are persisted and
+	// enforced; FOREIGN KEY is still rejected because the engine never validates
+	// referential integrity.
 	for _, constraint := range stmt.Constraints {
 		switch constraint.Type {
 		case parser.ConstraintPrimaryKey:
@@ -3745,9 +3750,7 @@ func (e *Executor) executeCreateTable(stmt *parser.CreateTableStmt) (*Result, er
 				}
 			}
 		case parser.ConstraintCheck:
-			return nil, pgerror.Unsupported(
-				"CHECK constraints are not supported: enforce the rule in the application instead (the engine parsed but never evaluated CHECK)",
-			)
+			// 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)",
@@ -3760,8 +3763,8 @@ func (e *Executor) executeCreateTable(stmt *parser.CreateTableStmt) (*Result, er
 	// produces, so both enforcement and the catalog observe them.
 	uniqIndexes := e.uniqueConstraintIndexes(stmt)
 
-	// Column references in generated expressions must resolve against the new
-	// table, so validate them once every column is known.
+	// Column references in generated expressions and CHECK predicates must
+	// resolve against the new table, so validate them once every column is known.
 	for _, colDef := range stmt.Columns {
 		if colDef.GeneratedExpr == nil {
 			continue
@@ -3770,6 +3773,15 @@ func (e *Executor) executeCreateTable(stmt *parser.CreateTableStmt) (*Result, er
 			return nil, err
 		}
 	}
+	var columnConstraints []parser.ColumnConstraint
+	for _, colDef := range stmt.Columns {
+		columnConstraints = append(columnConstraints, colDef.Constraints...)
+	}
+	checks, err := e.tableChecks(schema, columnConstraints, stmt.Constraints)
+	if err != nil {
+		return nil, err
+	}
+	schema.Checks = checks
 
 	if err := e.schema.CreateTable(schema); err != nil {
 		if stmt.IfNotExists && strings.Contains(err.Error(), "table already exists") {
@@ -4135,6 +4147,7 @@ func (e *Executor) executeAlterTableAddColumn(table string, action *parser.AddCo
 		Nullable: true,
 	}
 	var uniqueConstraint *parser.ColumnConstraint
+	var checkConstraints []parser.ColumnConstraint
 
 	// Process column constraints
 	for _, constraint := range action.Column.Constraints {
@@ -4156,10 +4169,9 @@ func (e *Executor) executeAlterTableAddColumn(table string, action *parser.AddCo
 				col.Default = val
 			}
 		case parser.ConstraintCheck:
-			return nil, pgerror.Unsupported(
-				"CHECK constraints are not supported: column %q declares CHECK; enforce the rule in the application instead",
-				col.Name,
-			)
+			// Validated with the other CHECK predicates once the column shape is
+			// final, so the predicate can reference the column being added.
+			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",
@@ -4174,11 +4186,12 @@ func (e *Executor) executeAlterTableAddColumn(table string, action *parser.AddCo
 	}
 
 	// The schema manager validates the new column against durable rows, rewrites
-	// them, and publishes the column together with its UNIQUE index while holding
-	// the table's exclusive gate, so a writer cannot slip a row in between the
-	// validation and the publication. Rows this session staged in an open
-	// transaction are invisible to that scan, so reject them here as well; the
-	// commit re-validates them against the schema that is current by then.
+	// them, and publishes the column together with its CHECK constraints and
+	// UNIQUE index while holding the table's exclusive gate, so a writer cannot
+	// slip a row in between the validation and the publication. Rows this session
+	// staged in an open transaction are invisible to that scan, so reject them
+	// here as well; the commit re-validates them against the schema that is
+	// current by then.
 	if !col.Nullable && col.Default == nil {
 		stagedRows, err := e.session.CountFast(table)
 		if err != nil {
@@ -4189,7 +4202,12 @@ func (e *Executor) executeAlterTableAddColumn(table string, action *parser.AddCo
 		}
 	}
 
-	if err := e.schema.AddColumn(table, col, uniqueIndex); err != nil {
+	checks, err := e.tableChecks(schemaWithColumn(schema, col), checkConstraints, nil)
+	if err != nil {
+		return nil, err
+	}
+
+	if err := e.schema.AddColumn(table, col, checks, uniqueIndex); err != nil {
 		if action.IfNotExists && strings.Contains(err.Error(), "column already exists") {
 			return NewResult("ALTER TABLE"), nil
 		}
@@ -4230,6 +4248,9 @@ func (e *Executor) addColumnUniqueIndex(table, column string, constraint *parser
 
 // executeAlterTableDropColumn drops a column from a table.
 func (e *Executor) executeAlterTableDropColumn(table string, action *parser.DropColumnAction) (*Result, error) {
+	if err := e.rejectCheckColumnChange(table, action.Column, "drop"); err != nil {
+		return nil, err
+	}
 	if err := e.schema.DropColumn(table, action.Column); err != nil {
 		return nil, err
 	}
@@ -4261,6 +4282,9 @@ func (e *Executor) executeAlterTableRenameColumn(table string, action *parser.Re
 	if err := e.rejectExpressionIndexRename(table, action.OldName); err != nil {
 		return nil, err
 	}
+	if err := e.rejectCheckColumnChange(table, action.OldName, "rename"); err != nil {
+		return nil, err
+	}
 	if err := e.schema.RenameColumn(table, action.OldName, action.NewName); err != nil {
 		return nil, err
 	}
@@ -4325,7 +4349,7 @@ func (e *Executor) executeBegin(stmt *parser.BeginStmt) (*Result, error) {
 // executeCommit commits the current transaction.
 func (e *Executor) executeCommit(stmt *parser.CommitStmt) (*Result, error) {
 	if !e.inTransaction {
-		return nil, fmt.Errorf("cannot commit: no transaction in progress")
+		return nil, fmt.Errorf("cannot commit: %w", ErrNoTransaction)
 	}
 
 	lsn, err := e.session.CommitWithLSN()
@@ -4350,7 +4374,7 @@ func (e *Executor) executeCommit(stmt *parser.CommitStmt) (*Result, error) {
 // executeRollback rolls back the current transaction or to a savepoint.
 func (e *Executor) executeRollback(stmt *parser.RollbackStmt) (*Result, error) {
 	if !e.inTransaction {
-		return nil, fmt.Errorf("cannot rollback: no transaction in progress")
+		return nil, fmt.Errorf("cannot rollback: %w", ErrNoTransaction)
 	}
 
 	if stmt.Savepoint != "" {
@@ -4397,7 +4421,7 @@ func (e *Executor) executeSavepoint(stmt *parser.SavepointStmt) (*Result, error)
 // executeRelease releases a savepoint.
 func (e *Executor) executeRelease(stmt *parser.ReleaseStmt) (*Result, error) {
 	if !e.inTransaction {
-		return nil, fmt.Errorf("cannot release savepoint: no transaction in progress")
+		return nil, fmt.Errorf("cannot release savepoint: %w", ErrNoTransaction)
 	}
 
 	// Find and remove the savepoint
@@ -7535,6 +7559,10 @@ func collectTableColumnRefs(expr parser.Expr) []tableColRef {
 			for _, a := range n.Args {
 				walk(a)
 			}
+		case *parser.ParenExpr:
+			walk(n.Expr)
+		case *parser.CastExpr:
+			walk(n.Expr)
 		}
 	}
 	walk(expr)

pkg/executor/sqlite_catalog.go

@@ -312,7 +312,7 @@ func (e *Executor) pragmaTableXInfo(table string) (*Result, error) {
 // implicit _rowid_ (and its aliases) are stripped, and identifiers are quoted so
 // xorm's IsColumnExist / GORM's HasColumn LIKE patterns match.
 func recreateCreateTableSQL(s *storage.Schema) string {
-	cols := make([]string, 0, len(s.Columns))
+	cols := make([]string, 0, len(s.Columns)+len(s.Checks))
 	for _, col := range s.Columns {
 		// Strip only the engine-injected hidden rowid column, never user
 		// columns that happen to be named oid/rowid/_rowid_.
@@ -321,6 +321,16 @@ func recreateCreateTableSQL(s *storage.Schema) string {
 		}
 		cols = append(cols, recreateColumnDef(s, col))
 	}
+	// Inline and table-level CHECK declarations are stored uniformly, so they
+	// are all rendered as table-level constraints: `CREATE TABLE t (a INTEGER,
+	// CHECK (a > 0))` is equivalent SQL for either declaration.
+	for _, check := range s.Checks {
+		if check.Name != "" {
+			cols = append(cols, fmt.Sprintf("CONSTRAINT %s CHECK (%s)", quoteIdent(check.Name), check.Expression))
+			continue
+		}
+		cols = append(cols, fmt.Sprintf("CHECK (%s)", check.Expression))
+	}
 	return fmt.Sprintf("CREATE TABLE %s (%s)", quoteIdent(s.Name), strings.Join(cols, ", "))
 }
 

pkg/httpserver/handler.go

@@ -293,6 +293,8 @@ func (s *Server) handleExecute(w http.ResponseWriter, r *http.Request) {
 			if _, err := exec.Execute(stmt); err != nil {
 				if errors.Is(err, storage.ErrSerialization) {
 					writeError(w, http.StatusConflict, "SERIALIZATION_FAILURE", err.Error(), nil)
+				} else if constraintErr := constraintViolationHTTPError(err); constraintErr != nil {
+					writeError(w, constraintErr.Status, constraintErr.Code, constraintErr.Message, constraintErr.Details)
 				} else {
 					writeError(w, http.StatusBadRequest, "TRANSACTION_ERROR", err.Error(), nil)
 				}
@@ -421,14 +423,60 @@ func (s *Server) handleVersion(w http.ResponseWriter, r *http.Request) {
 	writeJSON(w, http.StatusOK, resp, pretty)
 }
 
+// constraintViolationHTTPError maps a row-constraint violation onto the same
+// SQLSTATE response the PostgreSQL endpoint reports, and returns nil for any
+// other error. Transaction endpoints use it for a deferred unique, not-null, or
+// check failure at COMMIT while keeping their generic TRANSACTION_ERROR
+// fallback for errors that are not constraint violations.
+func constraintViolationHTTPError(err error) *HTTPError {
+	var checkErr *storage.CheckViolationError
+	var notNullErr *storage.NotNullViolationError
+	var uniqueErr *storage.UniqueViolationError
+	var primaryKeyErr *storage.PrimaryKeyViolationError
+	switch {
+	case errors.As(err, &checkErr):
+		return &HTTPError{
+			Status:  http.StatusBadRequest,
+			Code:    "23514",
+			Message: err.Error(),
+			Details: map[string]interface{}{"constraint": checkErr.Constraint},
+		}
+	case errors.As(err, &notNullErr):
+		return &HTTPError{
+			Status:  http.StatusBadRequest,
+			Code:    "23502",
+			Message: err.Error(),
+		}
+	case errors.As(err, &uniqueErr):
+		return &HTTPError{
+			Status:  http.StatusConflict,
+			Code:    "23505",
+			Message: err.Error(),
+			Details: map[string]interface{}{"constraint": uniqueErr.Constraint},
+		}
+	case errors.As(err, &primaryKeyErr):
+		return &HTTPError{
+			Status:  http.StatusConflict,
+			Code:    "23505",
+			Message: err.Error(),
+		}
+	default:
+		return nil
+	}
+}
+
 // sqlStateHTTPError maps engine errors onto the same SQLSTATE codes the
-// PostgreSQL endpoint reports. The executor returns plain errors on released
-// builds, so the unique-violation, not-null, and unsupported-feature markers are
-// matched on their stable message prefixes.
+// PostgreSQL endpoint reports. Typed constraint violations map first; the
+// executor also returns plain errors on released builds, so the
+// unique-violation, not-null, and unsupported-feature markers are matched on
+// their stable message prefixes as a fallback.
 func sqlStateHTTPError(err error) *HTTPError {
 	if err == nil {
 		return nil
 	}
+	if typed := constraintViolationHTTPError(err); typed != nil {
+		return typed
+	}
 	message := err.Error()
 	if strings.HasPrefix(message, "UNIQUE constraint failed: ") {
 		constraint := strings.TrimPrefix(message, "UNIQUE constraint failed: ")
@@ -446,6 +494,15 @@ func sqlStateHTTPError(err error) *HTTPError {
 			Message: message,
 		}
 	}
+	if strings.HasPrefix(message, "CHECK constraint failed: ") {
+		constraint := strings.TrimPrefix(message, "CHECK constraint failed: ")
+		return &HTTPError{
+			Status:  http.StatusBadRequest,
+			Code:    "23514",
+			Message: message,
+			Details: map[string]interface{}{"constraint": constraint},
+		}
+	}
 	if errors.Is(err, storage.ErrSerialization) {
 		return &HTTPError{
 			Status:  http.StatusConflict,
@@ -559,6 +616,8 @@ func (s *Server) handleTransactionCommit(w http.ResponseWriter, r *http.Request)
 	if _, err := tx.exec.Execute(stmt); err != nil {
 		if errors.Is(err, storage.ErrSerialization) {
 			writeError(w, http.StatusConflict, "SERIALIZATION_FAILURE", err.Error(), nil)
+		} else if constraintErr := constraintViolationHTTPError(err); constraintErr != nil {
+			writeError(w, constraintErr.Status, constraintErr.Code, constraintErr.Message, constraintErr.Details)
 		} else {
 			writeError(w, http.StatusBadRequest, "TRANSACTION_ERROR", err.Error(), nil)
 		}

pkg/httpserver/sqlstate_test.go

@@ -46,6 +46,76 @@ func TestHTTPUniqueViolationReturns23505(t *testing.T) {
 	}
 }
 
+// A constraint the commit discovers must report its SQLSTATE instead of the
+// generic TRANSACTION_ERROR the transaction endpoints use for other failures.
+func TestHTTPCommitCheckViolationReturns23514(t *testing.T) {
+	s := newTestDBServer(t)
+	if w := postQuery(t, s, "CREATE TABLE t (id INTEGER PRIMARY KEY)"); w.Code != http.StatusOK {
+		t.Fatalf("create table: %d %s", w.Code, w.Body.String())
+	}
+	txID := beginTransaction(t, s, "")
+	if w := postTransactionQuery(t, s, txID, "INSERT INTO t (id) VALUES (1)"); w.Code != http.StatusOK {
+		t.Fatalf("staged insert: %d %s", w.Code, w.Body.String())
+	}
+	// A second session publishes the constraint the staged row will violate.
+	if w := postQuery(t, s, "ALTER TABLE t ADD COLUMN qty INTEGER DEFAULT -1 CHECK (qty >= 0)"); w.Code != http.StatusOK {
+		t.Fatalf("add column: %d %s", w.Code, w.Body.String())
+	}
+
+	body, _ := json.Marshal(TransactionRequest{TransactionID: txID})
+	r := httptest.NewRequest(http.MethodPost, "/transaction/commit", strings.NewReader(string(body)))
+	w := httptest.NewRecorder()
+	s.handleTransactionCommit(w, r)
+
+	if w.Code != http.StatusBadRequest {
+		t.Fatalf("commit status = %d, want 400; 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 != "23514" {
+		t.Fatalf("code = %q, want 23514", resp.Error.Code)
+	}
+	constraint, _ := resp.Error.Details["constraint"].(string)
+	if constraint == "" || !strings.Contains(resp.Error.Message, constraint) {
+		t.Fatalf("details.constraint = %q, message = %q", constraint, resp.Error.Message)
+	}
+}
+
+// The transaction endpoints only replace TRANSACTION_ERROR for constraint
+// violations; other commit failures keep the generic code.
+func TestConstraintViolationHTTPErrorIgnoresOtherErrors(t *testing.T) {
+	if got := constraintViolationHTTPError(errors.New("boom")); got != nil {
+		t.Fatalf("generic error mapped to %+v, want nil", got)
+	}
+}
+
+func TestHTTPCheckViolationReturns23514(t *testing.T) {
+	server := setupKVTestServer(t)
+	queryHTTP(t, server, "CREATE TABLE t (id INTEGER PRIMARY KEY, age INTEGER, CONSTRAINT positive_age CHECK (age > 0))")
+
+	body, _ := json.Marshal(QueryRequest{SQL: "INSERT INTO t (id, age) VALUES (1, -1)"})
+	r := httptest.NewRequest(http.MethodPost, "/query", strings.NewReader(string(body)))
+	w := httptest.NewRecorder()
+	server.handleQuery(w, r)
+
+	if w.Code != http.StatusBadRequest {
+		t.Fatalf("status = %d, want 400; 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 != "23514" {
+		t.Fatalf("code = %q, want 23514", resp.Error.Code)
+	}
+	constraint, _ := resp.Error.Details["constraint"].(string)
+	if constraint != "positive_age" {
+		t.Fatalf("details.constraint = %q, want positive_age", constraint)
+	}
+}
+
 func TestSQLStateHTTPErrorMapping(t *testing.T) {
 	tests := []struct {
 		name   string
@@ -55,6 +125,11 @@ func TestSQLStateHTTPErrorMapping(t *testing.T) {
 	}{
 		{"unique", fmt.Errorf("UNIQUE constraint failed: uniq_users_email"), http.StatusConflict, "23505"},
 		{"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"},
+		{"typed not null", &storage.NotNullViolationError{Table: "t", Column: "v"}, http.StatusBadRequest, "23502"},
+		{"typed unique", &storage.UniqueViolationError{Constraint: "uq_email"}, http.StatusConflict, "23505"},
+		{"typed primary key", &storage.PrimaryKeyViolationError{Table: "t", Key: "1"}, http.StatusConflict, "23505"},
 		{"serialization", fmt.Errorf("commit: %w", storage.ErrSerialization), http.StatusConflict, "40001"},
 		{"unsupported", fmt.Errorf("feature not supported: DEFAULT is not constant"), http.StatusBadRequest, "0A000"},
 		{"other", errors.New("boom"), http.StatusInternalServerError, "EXECUTION_ERROR"},
@@ -68,3 +143,30 @@ func TestSQLStateHTTPErrorMapping(t *testing.T) {
 		})
 	}
 }
+
+// The batch /execute endpoint commits its own transaction, so a constraint the
+// commit discovers must map to 23514 there too instead of TRANSACTION_ERROR.
+func TestHTTPExecuteCommitCheckViolationReturns23514(t *testing.T) {
+	s := newTestDBServer(t)
+	if w := postQuery(t, s, "CREATE TABLE t (id INTEGER PRIMARY KEY)"); w.Code != http.StatusOK {
+		t.Fatalf("create table: %d %s", w.Code, w.Body.String())
+	}
+
+	w := postExecute(t, s, ExecuteRequest{
+		Transaction: true,
+		Statements: []QueryRequest{
+			{SQL: "INSERT INTO t (id) VALUES (1)"},
+			{SQL: "ALTER TABLE t ADD COLUMN qty INTEGER DEFAULT -1 CHECK (qty >= 0)"},
+		},
+	})
+	if w.Code != http.StatusBadRequest {
+		t.Fatalf("execute status = %d, want 400; 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 != "23514" {
+		t.Fatalf("code = %q, want 23514", resp.Error.Code)
+	}
+}

pkg/parser/format.go

@@ -189,7 +189,10 @@ func quoteFormatIdent(name string) string {
 			break
 		}
 	}
-	if bare {
+	// Formatted text is persisted and re-parsed, so an identifier that lexes as
+	// a SQL keyword must be quoted even when it is otherwise bare: a column
+	// named "order" has to format as "order" to stay a column reference.
+	if bare && lexer.LookupKeyword(strings.ToUpper(name)) == lexer.TokenIdent {
 		return name
 	}
 	return `"` + strings.ReplaceAll(name, `"`, `""`) + `"`

pkg/parser/format_test.go

@@ -0,0 +1,40 @@
+package parser
+
+import (
+	"testing"
+
+	"github.com/danfragoso/pizzasql-next/pkg/lexer"
+)
+
+// FormatExpr output is persisted in the schema and re-parsed on every later
+// evaluation, so an identifier that lexes as a keyword has to stay quoted or the
+// stored text would no longer name a column.
+func TestFormatExprQuotesKeywordIdentifiers(t *testing.T) {
+	tests := []struct {
+		name string
+		expr Expr
+		want string
+	}{
+		{"keyword column", &ColumnRef{Column: "order"}, `"order"`},
+		{"plain column", &ColumnRef{Column: "age"}, "age"},
+		{"keyword in expression", &BinaryExpr{
+			Left:  &ColumnRef{Column: "order"},
+			Op:    lexer.TokenGt,
+			Right: &LiteralExpr{Type: lexer.TokenNumber, Value: "0"},
+		}, `("order" > 0)`},
+		{"qualified keyword column", &ColumnRef{Table: "select", Column: "from"}, `"select"."from"`},
+		{"underscored column", &ColumnRef{Column: "user_id"}, "user_id"},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			formatted := FormatExpr(tt.expr)
+			if formatted != tt.want {
+				t.Fatalf("FormatExpr = %q, want %q", formatted, tt.want)
+			}
+			if _, err := ParseExpr(formatted); err != nil {
+				t.Fatalf("formatted expression %q does not re-parse: %v", formatted, err)
+			}
+		})
+	}
+}

pkg/parser/parser.go

@@ -1371,6 +1371,19 @@ func (p *Parser) isDataTypeKeyword() bool {
 func (p *Parser) parseColumnConstraint() (*ColumnConstraint, bool, error) {
 	constraint := &ColumnConstraint{}
 
+	// A column-level constraint can be named: `col TYPE CONSTRAINT name CHECK
+	// (...)` is the same constraint as the unnamed form.
+	named := false
+	if p.curTokenIs(lexer.TokenCONSTRAINT) {
+		p.nextToken()
+		if !p.curTokenIs(lexer.TokenIdent) {
+			return nil, false, p.curError("expected constraint name")
+		}
+		constraint.Name = p.curToken.Literal
+		named = true
+		p.nextToken()
+	}
+
 	switch p.curToken.Type {
 	case lexer.TokenPRIMARY:
 		p.nextToken()
@@ -1464,10 +1477,16 @@ func (p *Parser) parseColumnConstraint() (*ColumnConstraint, bool, error) {
 		// explicit NULL declaration carries no constraint. Consume it so
 		// "col TYPE NULL" parses without inventing a fake nullable constraint.
 		// NOT NULL is applied whenever it appears, regardless of ordering.
+		if named {
+			return nil, false, p.curError("expected constraint type after constraint name")
+		}
 		p.nextToken()
 		return nil, true, nil
 
 	default:
+		if named {
+			return nil, false, p.curError("expected constraint type after constraint name")
+		}
 		return nil, false, nil
 	}
 

pkg/pgerror/pgerror.go

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

pkg/pgserver/check_commit_test.go

@@ -0,0 +1,79 @@
+package pgserver
+
+import (
+	"testing"
+
+	"github.com/danfragoso/pizzasql-next/pkg/storage"
+	"github.com/danfragoso/pizzasql-next/pkg/testkv"
+)
+
+func simpleQueryMessage(sql string) *Message {
+	return &Message{Type: MsgQuery, Data: append([]byte(sql), 0)}
+}
+
+// A row constraint that only the commit can detect (a column and its CHECK were
+// published after the row was staged) must report its SQLSTATE, leave the wire
+// session idle, and let the client's cleanup ROLLBACK complete.
+func TestConstraintViolationAtCommitClearsTransactionState(t *testing.T) {
+	tests := []struct {
+		name string
+		ddl  string
+		code string
+	}{
+		{
+			name: "check",
+			ddl:  "ALTER TABLE t ADD COLUMN qty INTEGER DEFAULT -1 CHECK (qty >= 0)",
+			code: ErrCodeCheckViolation,
+		},
+		{
+			name: "not null",
+			ddl:  "ALTER TABLE t ADD COLUMN note TEXT NOT NULL",
+			code: "23502",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			kv := testkv.New(t)
+			pool := kv.Pool(8)
+			t.Cleanup(func() { pool.Close() })
+			schema := storage.NewSchemaManager(pool, "pg_commit_constraints")
+			table := storage.NewTableManager(pool, schema, "pg_commit_constraints")
+
+			writer, writerClient := newDataConnectionFor(t, schema, table)
+			ddl, ddlClient := newDataConnectionFor(t, schema, table)
+
+			runQuery(t, writer, writerClient, simpleQueryMessage("CREATE TABLE t (id INTEGER PRIMARY KEY)"))
+			runQuery(t, writer, writerClient, simpleQueryMessage("BEGIN"))
+			runQuery(t, writer, writerClient, simpleQueryMessage("INSERT INTO t (id) VALUES (1)"))
+
+			// The row is only staged, so the new constraint sees an empty table
+			// and the commit is where the staged row violates it.
+			runQuery(t, ddl, ddlClient, simpleQueryMessage(tt.ddl))
+
+			msgs := runQuery(t, writer, writerClient, simpleQueryMessage("COMMIT"))
+			if code := errorCode(msgs[0]); code != tt.code {
+				t.Fatalf("COMMIT SQLSTATE = %q, want %q", code, tt.code)
+			}
+			if writer.txStatus != TxStatusIdle {
+				t.Fatalf("tx status = %v, want idle after a failed COMMIT", writer.txStatus)
+			}
+
+			// Storage already rolled the transaction back, so the client's
+			// cleanup ROLLBACK is a no-op rather than an error.
+			msgs = runQuery(t, writer, writerClient, simpleQueryMessage("ROLLBACK"))
+			if len(msgs) == 0 || msgs[0].Type == MsgErrorResponse {
+				t.Fatalf("ROLLBACK after failed COMMIT = %v, want a successful no-op", msgs)
+			}
+			if writer.txStatus != TxStatusIdle {
+				t.Fatalf("tx status = %v, want idle after the no-op ROLLBACK", writer.txStatus)
+			}
+
+			// The session keeps working.
+			msgs = runQuery(t, writer, writerClient, simpleQueryMessage("SELECT 1"))
+			if len(msgs) < 2 || msgs[0].Type != MsgRowDescription {
+				t.Fatalf("SELECT after failed COMMIT = %v, want a result", msgs)
+			}
+		})
+	}
+}

pkg/pgserver/check_test.go

@@ -0,0 +1,29 @@
+package pgserver
+
+import (
+	"testing"
+)
+
+// A CHECK violation reports SQLSTATE 23514 on both the simple and the extended
+// protocol and, inside a transaction block, leaves the transaction failed like
+// the other integrity errors.
+func TestCheckConstraintSQLStateAndTransactionState(t *testing.T) {
+	for _, extended := range []bool{false, true} {
+		c, client := newDataConnection(t)
+		runQuery(t, c, client, &Message{Type: MsgQuery, Data: []byte("CREATE TABLE t (id INTEGER PRIMARY KEY, age INTEGER CHECK (age > 0))\x00")})
+		runQuery(t, c, client, &Message{Type: MsgQuery, Data: []byte("BEGIN\x00")})
+		var messages []*Message
+		if extended {
+			messages = runExtendedExecution(t, c, client, "INSERT INTO t VALUES (1,-1)")
+		} else {
+			messages = runQuery(t, c, client, &Message{Type: MsgQuery, Data: []byte("INSERT INTO t VALUES (1,-1)\x00")})
+		}
+		if code := errorCode(messages[0]); code != ErrCodeCheckViolation {
+			t.Fatalf("extended=%v SQLSTATE=%q, want %q", extended, code, ErrCodeCheckViolation)
+		}
+		if c.txStatus != TxStatusFailed {
+			t.Fatalf("extended=%v transaction not failed: %c", extended, c.txStatus)
+		}
+		runQuery(t, c, client, &Message{Type: MsgQuery, Data: []byte("ROLLBACK\x00")})
+	}
+}

pkg/pgserver/compat_suite_test.go

@@ -127,9 +127,10 @@ 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 and FOREIGN KEY are rejected at DDL instead of being silently
-	// discarded; enforce these rules in the application.
-	expectCode("ddl.check_0A000", `CREATE TABLE checked (n INTEGER CHECK (n > 0))`, "0A000")
+	// CHECK is accepted and enforced; FOREIGN KEY is still rejected at DDL
+	// instead of being silently discarded.
+	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")
 
 	expectCode("errors.syntax_42601", `SELEC 1`, "42601")

pkg/pgserver/connection.go

@@ -407,18 +407,31 @@ func (c *Connection) handleQuery(msg *Message) error {
 			}
 		}
 		result, err := c.executor.Execute(stmt)
+		if err != nil && isIdleRollback(stmt, c.txStatus, err) {
+			result, err = executor.NewResult("ROLLBACK"), nil
+		}
 		if err != nil {
 			code := ErrCodeInternalError
 			message := fmt.Sprintf("Execution error: %v", err)
 			var primaryKeyErr *storage.PrimaryKeyViolationError
 			var uniqueErr *storage.UniqueViolationError
 			var notNullErr *storage.NotNullViolationError
+			var checkErr *storage.CheckViolationError
 			if errors.Is(err, storage.ErrSerialization) {
 				code = ErrCodeSerializationFailure
 				c.txStatus = TxStatusIdle
 			} else if errors.As(err, &notNullErr) {
 				code = "23502"
-				if c.txStatus == TxStatusInBlock {
+				if _, commit := stmt.(*parser.CommitStmt); commit {
+					c.txStatus = TxStatusIdle
+				} else if c.txStatus == TxStatusInBlock {
+					c.txStatus = TxStatusFailed
+				}
+			} else if errors.As(err, &checkErr) {
+				code = ErrCodeCheckViolation
+				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) {
@@ -651,17 +664,30 @@ func (c *Connection) handleExecute(msg *Message) error {
 		return c.failExtended(code, fmt.Errorf("syntax error: %w", err))
 	}
 	result, err := c.executor.Execute(stmt)
+	if err != nil && isIdleRollback(stmt, c.txStatus, err) {
+		result, err = executor.NewResult("ROLLBACK"), nil
+	}
 	if err != nil {
 		code := ErrCodeInternalError
 		var primaryKeyErr *storage.PrimaryKeyViolationError
 		var uniqueErr *storage.UniqueViolationError
 		var notNullErr *storage.NotNullViolationError
+		var checkErr *storage.CheckViolationError
 		if errors.Is(err, storage.ErrSerialization) {
 			code = ErrCodeSerializationFailure
 			c.txStatus = TxStatusIdle
 		} else if errors.As(err, &notNullErr) {
 			code = "23502"
-			if c.txStatus == TxStatusInBlock {
+			if _, commit := stmt.(*parser.CommitStmt); commit {
+				c.txStatus = TxStatusIdle
+			} else if c.txStatus == TxStatusInBlock {
+				c.txStatus = TxStatusFailed
+			}
+		} else if errors.As(err, &checkErr) {
+			code = ErrCodeCheckViolation
+			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) {
@@ -1264,6 +1290,18 @@ func (c *Connection) sendBackendKeyData(processID, secretKey int32) error {
 	return c.writeMessage(MsgBackendKeyData, mb.Bytes())
 }
 
+// isIdleRollback reports whether a bare ROLLBACK found no transaction to roll
+// back. A COMMIT rejected by storage has already ended the transaction, so the
+// client's cleanup ROLLBACK must complete (PostgreSQL reports a warning and
+// succeeds) instead of surfacing an internal error on an idle session.
+func isIdleRollback(stmt parser.Statement, status byte, err error) bool {
+	rollback, ok := stmt.(*parser.RollbackStmt)
+	if !ok || rollback.Savepoint != "" {
+		return false
+	}
+	return status == TxStatusIdle && errors.Is(err, executor.ErrNoTransaction)
+}
+
 // sendReadyForQuery sends ready for query message
 func (c *Connection) sendReadyForQuery() error {
 	mb := NewMessageBuilder()

pkg/pgserver/features_test.go

@@ -17,16 +17,23 @@ import (
 // simple-query behavior can be asserted end to end.
 func newDataConnection(t *testing.T) (*Connection, net.Conn) {
 	t.Helper()
-	server, client := net.Pipe()
-	t.Cleanup(func() {
-		server.Close()
-		client.Close()
-	})
 	kv := testkv.New(t)
 	pool := kv.Pool(4)
 	t.Cleanup(func() { pool.Close() })
 	schema := storage.NewSchemaManager(pool, "pg_features")
 	table := storage.NewTableManager(pool, schema, "pg_features")
+	return newDataConnectionFor(t, schema, table)
+}
+
+// newDataConnectionFor wires a wire connection to an existing storage engine, so
+// a test can drive two independent sessions over one shared store.
+func newDataConnectionFor(t *testing.T, schema *storage.SchemaManager, table *storage.TableManager) (*Connection, net.Conn) {
+	t.Helper()
+	server, client := net.Pipe()
+	t.Cleanup(func() {
+		server.Close()
+		client.Close()
+	})
 	exec := executor.New(schema, table)
 	exec.SyncCatalog()
 

pkg/pgserver/protocol.go

@@ -96,6 +96,7 @@ const (
 	ErrCodeTransactionAborted   = "25P02"
 	ErrCodeSerializationFailure = "40001"
 	ErrCodeUniqueViolation      = "23505"
+	ErrCodeCheckViolation       = "23514"
 )
 
 // Message represents a PostgreSQL protocol message

pkg/pgserver/unsupported_constraints_test.go

@@ -4,14 +4,14 @@ import (
 	"testing"
 )
 
-func TestCheckAndForeignKeyRejectedOverSimpleQuery(t *testing.T) {
+// 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, age INTEGER CHECK (age > 0))",
-		"CREATE TABLE t2 (id INTEGER PRIMARY KEY, parent_id INTEGER REFERENCES parent(id))",
-		"CREATE TABLE t3 (id INTEGER PRIMARY KEY, parent_id INTEGER, FOREIGN KEY (parent_id) REFERENCES parent(id))",
+		"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 {

pkg/storage/constraints.go

@@ -35,6 +35,51 @@ func validateNotNull(schema *Schema, row Row) error {
 	return nil
 }
 
+// CheckViolationError identifies the CHECK predicate a write violated. It is a
+// distinct type so the protocol layers can report SQLSTATE 23514.
+type CheckViolationError struct {
+	Table      string
+	Constraint string
+}
+
+func (e *CheckViolationError) Error() string {
+	return "CHECK constraint failed: " + e.Constraint
+}
+
+// validateChecks evaluates every persisted CHECK predicate against a complete
+// row image. The evaluator reports NULL and other unknown results as passing, so
+// SQL three-valued logic is preserved: a constraint only rejects a definite
+// false.
+func (m *SchemaManager) validateChecks(schema *Schema, row Row) error {
+	if len(schema.Checks) == 0 {
+		return nil
+	}
+	eval := m.checkEvaluator()
+	if eval == nil {
+		return fmt.Errorf("cannot validate CHECK constraints on %s: no CHECK evaluator installed", schema.Name)
+	}
+	for _, check := range schema.Checks {
+		holds, err := eval(check.Expression, row)
+		if err != nil {
+			return fmt.Errorf("evaluate CHECK constraint %s on %s: %w", check.Label(), schema.Name, err)
+		}
+		if !holds {
+			return &CheckViolationError{Table: schema.Name, Constraint: check.Label()}
+		}
+	}
+	return nil
+}
+
+// validateRowConstraints enforces the NOT NULL and CHECK constraints a schema
+// declares against a complete row image. Every write path calls it after
+// defaults are applied and before the row becomes visible.
+func (m *SchemaManager) validateRowConstraints(schema *Schema, row Row) error {
+	if err := validateNotNull(schema, row); err != nil {
+		return err
+	}
+	return m.validateChecks(schema, row)
+}
+
 // applySchemaDefaults completes a row image that was built against an older
 // schema: columns the row predates receive their catalog default, and the row is
 // then validated against the required columns of the current schema.
@@ -46,7 +91,9 @@ func validateNotNull(schema *Schema, row Row) error {
 // absence. Re-applying the current schema keeps such a row from becoming a
 // durable NOT NULL violation, and gives it the default every other row received
 // from the backfill. An absent key is the only thing filled, so an explicit NULL
-// still does not select a default.
+// still does not select a default. Callers validate CHECK predicates separately
+// through validateRowConstraints/validateChecks so each write site can run them
+// against the final row image.
 func applySchemaDefaults(schema *Schema, row Row) error {
 	for _, column := range schema.Columns {
 		if _, ok := rowColumnValue(row, column.Name); ok {

pkg/storage/schema.go

@@ -25,6 +25,27 @@ type Schema struct {
 	CreatedAt     time.Time `json:"created_at"`
 	NextRowID     int64     `json:"next_rowid"`
 	AutoIncrement bool      `json:"autoincrement"`
+	// Checks holds the row predicates declared with CHECK, at column and table
+	// level alike. Inline and table-level declarations are equivalent once
+	// parsed, so both live in this one list.
+	Checks []CheckConstraint `json:"checks,omitempty"`
+}
+
+// CheckConstraint is a persisted CHECK predicate. The predicate is stored as
+// SQL text and re-parsed by the evaluator installed on the SchemaManager, so a
+// durable reload keeps enforcing it.
+type CheckConstraint struct {
+	Name       string `json:"name,omitempty"`
+	Expression string `json:"expression"`
+}
+
+// Label identifies the constraint in error messages and the catalog: the
+// declared name when there is one, otherwise the expression text.
+func (c CheckConstraint) Label() string {
+	if c.Name != "" {
+		return c.Name
+	}
+	return c.Expression
 }
 
 // Column represents a column definition.
@@ -84,10 +105,36 @@ type SchemaManager struct {
 	tableLocksMu    sync.Mutex
 	tableLocks      map[string]*sync.RWMutex
 
+	// checkMu guards checkEval, the pluggable evaluator for persisted CHECK
+	// predicates. It is set by the SQL executor so storage stays independent of
+	// the SQL front end.
+	checkMu   sync.RWMutex
+	checkEval CheckEvaluator
+
 	rowIDMu    sync.Mutex
 	rowIDAlloc map[string]*rowIDAllocator
 }
 
+// CheckEvaluator reports whether a persisted CHECK expression holds for a row.
+// A false result is a violation; NULL and other unknown results must be
+// reported as true, matching SQL three-valued logic. The SQL executor registers
+// one so the storage layer can enforce CHECK without importing the parser.
+type CheckEvaluator func(expression string, row Row) (bool, error)
+
+// SetCheckEvaluator installs the evaluator used to enforce persisted CHECK
+// predicates.
+func (m *SchemaManager) SetCheckEvaluator(eval CheckEvaluator) {
+	m.checkMu.Lock()
+	m.checkEval = eval
+	m.checkMu.Unlock()
+}
+
+func (m *SchemaManager) checkEvaluator() CheckEvaluator {
+	m.checkMu.RLock()
+	defer m.checkMu.RUnlock()
+	return m.checkEval
+}
+
 // BeginTransaction is retained for API compatibility. Buffered per-session
 // transactions no longer take a database-wide transaction lock; staged writes
 // are validated and committed atomically with CompareBatchWrite instead.
@@ -349,6 +396,7 @@ func cloneSchema(schema *Schema) *Schema {
 	}
 	cloned := *schema
 	cloned.Columns = append([]Column(nil), schema.Columns...)
+	cloned.Checks = append([]CheckConstraint(nil), schema.Checks...)
 	return &cloned
 }
 
@@ -943,27 +991,30 @@ func (m *SchemaManager) removeFromIndexList(name string) error {
 }
 
 // AddColumn adds a new column to a table. When unique is non-nil the column's
-// UNIQUE index is registered as part of the same statement.
+// UNIQUE index is registered as part of the same statement, and checks are
+// persisted with the column in the same schema snapshot.
 //
 // A constant DEFAULT is materialized into every existing row before the schema
-// is published. NOT NULL, UNIQUE, and index validation all read persisted row
-// values, so a default that only existed in the catalog would be invisible to
-// them: a UNIQUE index over the new column could not see the duplicate values
-// every existing row shares, and an UPDATE to NULL would look non-null. The
-// rewrite is the same shape as DropColumn: the table's exclusive gate keeps
-// point writes and CAS commits out while rows are scanned, validated, and
-// rewritten, the transform is idempotent, and the schema is saved last so an
-// interrupted operation resumes against the old schema.
+// is published. NOT NULL, UNIQUE, CHECK, and index validation all read
+// persisted row values, so a default that only existed in the catalog would be
+// invisible to them: a UNIQUE index over the new column could not see the
+// duplicate values every existing row shares, a CHECK predicate could not see a
+// violating default, and an UPDATE to NULL would look non-null. The rewrite is
+// the same shape as DropColumn: the table's exclusive gate keeps point writes
+// and CAS commits out while rows are scanned, validated, and rewritten, the
+// transform is idempotent, and the schema is saved last so an interrupted
+// operation resumes against the old schema.
 //
-// The gate also makes the validation authoritative: NOT NULL without a default
-// and duplicate UNIQUE values are rejected before any row is touched, and the
-// column and its index are published together in one batch write, so a crash
-// cannot leave the column visible without the constraint that validates it.
-// A nullable column without a default needs no rewrite because a missing row
-// field already reads as NULL. Like DROP COLUMN, the statement holds the catalog
-// lock for the whole scan, so the cost is one table scan (two for UNIQUE) and a
-// table-wide stall per ADD COLUMN that has a default or an index.
-func (m *SchemaManager) AddColumn(table string, column Column, unique *Index) error {
+// The gate also makes the validation authoritative: NOT NULL without a default,
+// duplicate UNIQUE values, and rows a new CHECK predicate rejects are rejected
+// before any row is touched, and the column, its checks, and its index are
+// published together in one batch write, so a crash cannot leave the column
+// visible without the constraints that validate it. A nullable column without a
+// default needs no rewrite because a missing row field already reads as NULL.
+// Like DROP COLUMN, the statement holds the catalog lock for the whole scan, so
+// the cost is one table scan per validating constraint plus a table-wide stall
+// per ADD COLUMN that has a default, an index, or a check.
+func (m *SchemaManager) AddColumn(table string, column Column, checks []CheckConstraint, unique *Index) error {
 	tableLock := m.tableLock(table)
 	tableLock.Lock()
 	defer tableLock.Unlock()
@@ -989,6 +1040,9 @@ func (m *SchemaManager) AddColumn(table string, column Column, unique *Index) er
 			return err
 		}
 	}
+	if err := m.validateAddColumnChecks(table, column, checks); err != nil {
+		return err
+	}
 
 	// A required column with nothing to write into rows that already exist
 	// cannot be declared on a populated table.
@@ -1014,16 +1068,18 @@ func (m *SchemaManager) AddColumn(table string, column Column, unique *Index) er
 		}
 	}
 
-	return m.publishAddedColumn(schema, column, unique)
+	return m.publishAddedColumn(schema, column, checks, unique)
 }
 
-// publishAddedColumn publishes a column and, when present, its unique index in
-// one atomic batch write (m.mu must be held): the schema snapshot, the index
-// definition, and the index list always become durable together or not at all.
-func (m *SchemaManager) publishAddedColumn(schema *Schema, column Column, unique *Index) error {
+// publishAddedColumn publishes a column, its checks, and, when present, its
+// unique index in one atomic batch write (m.mu must be held): the schema
+// snapshot, the index definition, and the index list always become durable
+// together or not at all.
+func (m *SchemaManager) publishAddedColumn(schema *Schema, column Column, checks []CheckConstraint, unique *Index) error {
 	// Publish a fresh snapshot instead of mutating readers' shared pointer.
 	schema = cloneSchema(schema)
 	schema.Columns = append(schema.Columns, column)
+	schema.Checks = append(schema.Checks, checks...)
 	schemaData, err := json.Marshal(schema)
 	if err != nil {
 		return fmt.Errorf("failed to serialize schema: %w", err)
@@ -1568,6 +1624,37 @@ func (m *SchemaManager) validateAddColumnUnique(table string, column Column, ind
 	})
 }
 
+// validateAddColumnChecks rejects an ADD COLUMN ... CHECK whose predicate is
+// already false for an existing row. Like validateAddColumnUnique it reads rows
+// as they will look after the constant-default backfill, and it runs before that
+// rewrite so a rejected DDL leaves no row carrying the new field. The caller
+// holds the table's exclusive gate, so no writer can insert a violating row
+// between this validation and the publication of the check.
+func (m *SchemaManager) validateAddColumnChecks(table string, column Column, checks []CheckConstraint) error {
+	if len(checks) == 0 {
+		return nil
+	}
+	eval := m.checkEvaluator()
+	if eval == nil {
+		return fmt.Errorf("cannot validate CHECK constraints on %s: no CHECK evaluator installed", table)
+	}
+	return m.applyRows(table, func(row Row) (bool, error) {
+		if _, ok := rowColumnValue(row, column.Name); !ok && column.Default != nil {
+			row[column.Name] = column.Default
+		}
+		for _, check := range checks {
+			holds, err := eval(check.Expression, row)
+			if err != nil {
+				return false, fmt.Errorf("evaluate CHECK constraint %s on %s: %w", check.Label(), table, err)
+			}
+			if !holds {
+				return false, &CheckViolationError{Table: table, Constraint: check.Label()}
+			}
+		}
+		return false, nil
+	})
+}
+
 // tableHasRows reports whether a table holds at least one durable row.
 func (m *SchemaManager) tableHasRows(table string) (bool, error) {
 	prefix := []byte(fmt.Sprintf("%s:_data:%s:", m.database, strings.ToLower(table)))

pkg/storage/schema_test.go

@@ -896,7 +896,7 @@ func TestAddColumnPublishesUniqueIndexInOneWrite(t *testing.T) {
 
 	index := &Index{Name: "uniq_t_code", Table: "t", Unique: true, Columns: []IndexColumn{{Name: "code"}}}
 	schemaWritesBefore := kv.writeCount(":_schema:")
-	if err := schemas.AddColumn("t", Column{Name: "code", Type: "TEXT", Nullable: true, Default: "x"}, index); err != nil {
+	if err := schemas.AddColumn("t", Column{Name: "code", Type: "TEXT", Nullable: true, Default: "x"}, nil, index); err != nil {
 		t.Fatalf("add column: %v", err)
 	}
 
@@ -957,7 +957,7 @@ func TestAddColumnRejectsDuplicateIndexName(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	err := schemas.AddColumn("t", Column{Name: "code", Type: "TEXT", Nullable: true, Default: "x"},
+	err := schemas.AddColumn("t", Column{Name: "code", Type: "TEXT", Nullable: true, Default: "x"}, nil,
 		&Index{Name: "uniq_t_code", Table: "t", Unique: true, Columns: []IndexColumn{{Name: "code"}}})
 	if err == nil {
 		t.Fatal("expected the existing index name to be rejected")

pkg/storage/table.go

@@ -552,7 +552,7 @@ func (m *TableManager) prepareInsert(table string, row Row) (Row, string, error)
 		normalizedRow["_rowid_"] = newRowID
 	}
 
-	if err := validateNotNull(schema, normalizedRow); err != nil {
+	if err := m.schema.validateRowConstraints(schema, normalizedRow); err != nil {
 		return nil, "", err
 	}
 	return normalizedRow, m.dataKey(table, pk), nil
@@ -598,6 +598,9 @@ func (m *TableManager) InsertWithRowID(table string, row Row) (int64, error) {
 	if err := applySchemaDefaults(schema, nr); err != nil {
 		return 0, err
 	}
+	if err := m.schema.validateChecks(schema, nr); err != nil {
+		return 0, err
+	}
 	data, err := encodeRow(nr)
 	if err != nil {
 		return 0, fmt.Errorf("failed to serialize row: %w", err)
@@ -763,7 +766,7 @@ func (m *TableManager) InsertBulkWithLastRowID(table string, rows []Row) (int, i
 			}
 		}
 		nr["_rowid_"] = rowid
-		if err := validateNotNull(schema, nr); err != nil {
+		if err := m.schema.validateRowConstraints(schema, nr); err != nil {
 			return 0, 0, err
 		}
 		if rowid > maxRowID {
@@ -1102,6 +1105,9 @@ func (m *TableManager) UpdateByPK(table, pk string, updateFn func(Row) (Row, err
 	if err := applySchemaDefaults(schema, row); err != nil {
 		return nil, false, err
 	}
+	if err := m.schema.validateChecks(schema, row); err != nil {
+		return nil, false, err
+	}
 	data, err := encodeRow(row)
 	if err != nil {
 		return nil, false, err

pkg/storage/tx.go

@@ -429,6 +429,9 @@ func (s *Session) applyOverlaySchemaDefaults() error {
 			if err := applySchemaDefaults(schema, entry.row); err != nil {
 				return err
 			}
+			if err := s.schema.validateChecks(schema, entry.row); err != nil {
+				return err
+			}
 		}
 	}
 	return nil
@@ -794,7 +797,7 @@ func (s *Session) UpdateByPK(table, pk string, updateFn func(Row) (Row, error))
 		}
 	}
 
-	if err := validateNotNull(schema, row); err != nil {
+	if err := s.schema.validateRowConstraints(schema, row); err != nil {
 		return nil, false, err
 	}
 	s.stagePut(table, key, row)
@@ -887,7 +890,7 @@ func (s *Session) UpdateFunc(table string, updateFn func(Row) (Row, error), filt
 				}
 			}
 		}
-		if err := validateNotNull(schema, row); err != nil {
+		if err := s.schema.validateRowConstraints(schema, row); err != nil {
 			return count, err
 		}
 		newKey := s.table.dataKey(table, fmt.Sprintf("%v", row[schema.PrimaryKey]))