engine: implement CHECK constraints

Accept inline and table-level CHECK declarations, named or unnamed, at
CREATE TABLE and ALTER TABLE ADD COLUMN, persist the predicate SQL in the
schema, and enforce it on every write path with SQLSTATE 23514.

Enforcement lives in the storage layer next to NOT NULL: prepareInsert,
bulk insert, point and scan updates, and transaction commit all validate
persisted predicates through an evaluator the executor registers, so the
rule also covers rows staged before a concurrent ADD COLUMN publishes a
constraint. The evaluator treats NULL/unknown results as passing, matching
SQL three-valued logic.

ADD COLUMN ... CHECK validates existing rows against the value the
constant-default backfill would write, before any row is rewritten, and
publishes the column, its checks, and any UNIQUE index in one schema
write. A violating row rejects the DDL with no column, constraint, or
row rewrite published.

Only deterministic, subquery-free predicates over unqualified columns are
accepted (column references, literals, arithmetic, comparisons, IS NULL,
LIKE, CAST, and the deterministic scalar allowlist shared with generated
columns and expression indexes); subqueries, aggregates, window functions,
non-deterministic functions, and qualified references are rejected with
0A000. Predicates are stored as SQL text, so DROP/RENAME COLUMN of a
referenced column is rejected instead of leaving a constraint that would
silently evaluate to NULL.

The emulated sqlite_master/sqlite_schema table SQL renders the constraints;
pg_constraint and information_schema.check_constraints are not emulated.
The HTTP API and PostgreSQL protocol map violations to 23514, and the
capability document advertises CHECK as enforced.

pkg/capabilities/capabilities.go

@@ -170,7 +170,7 @@ 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"},
 		},
 
@@ -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

@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"testing"
 
+	"github.com/danfragoso/pizzasql-next/pkg/pgerror"
 	"github.com/danfragoso/pizzasql-next/pkg/version"
 )
 
@@ -69,3 +70,23 @@ 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)
+	}
+}

pkg/executor/check_constraint_test.go

@@ -0,0 +1,470 @@
+package executor
+
+import (
+	"errors"
+	"strings"
+	"testing"
+
+	"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: "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, "")
+			}
+		})
+	}
+}

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,123 @@
+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,
+		)
+	}
+	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

@@ -113,11 +113,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 +3712,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 +3725,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 +3744,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 +3757,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 +3767,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 +4141,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 +4163,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 +4180,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 +4196,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 +4242,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 +4276,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
 	}

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

@@ -446,6 +446,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,

pkg/httpserver/sqlstate_test.go

@@ -46,6 +46,31 @@ func TestHTTPUniqueViolationReturns23505(t *testing.T) {
 	}
 }
 
+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 +80,7 @@ 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"},
 		{"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"},

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_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

@@ -413,6 +413,7 @@ func (c *Connection) handleQuery(msg *Message) error {
 			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
@@ -421,6 +422,11 @@ func (c *Connection) handleQuery(msg *Message) error {
 				if c.txStatus == TxStatusInBlock {
 					c.txStatus = TxStatusFailed
 				}
+			} else if errors.As(err, &checkErr) {
+				code = ErrCodeCheckViolation
+				if c.txStatus == TxStatusInBlock {
+					c.txStatus = TxStatusFailed
+				}
 			} else if errors.As(err, &uniqueErr) || errors.As(err, &primaryKeyErr) {
 				code = ErrCodeUniqueViolation
 				if uniqueErr != nil {
@@ -656,6 +662,7 @@ func (c *Connection) handleExecute(msg *Message) error {
 		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
@@ -664,6 +671,11 @@ func (c *Connection) handleExecute(msg *Message) error {
 			if c.txStatus == TxStatusInBlock {
 				c.txStatus = TxStatusFailed
 			}
+		} else if errors.As(err, &checkErr) {
+			code = ErrCodeCheckViolation
+			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/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]))