Merge pull request #34 from database-pizza/engine/b-composite-pk

feat: composite PRIMARY KEY (a, b) support

PKBFI_STORAGE_MIGRATION.md

@@ -39,6 +39,17 @@ The migration command exits after writing and verifying the destination. It refu
   pre-CHECK engine ignores it on read and drops it on the next schema write, so
   rolling back to such an engine silently stops enforcing CHECK constraints;
   restore the newer engine (or re-create the constraints) after a rollback.
+- `Schema.PrimaryKeyColumns` (`primary_key_columns`) is an additive, optional
+  JSON field for composite primary keys. A single-column catalog still carries
+  only `primary_key`; a composite catalog persists `primary_key` as an empty
+  string plus the per-column `primary_key` flags and the ordered
+  `primary_key_columns` list. A pre-composite engine ignores the new field, so
+  rolling back to one collapses key derivation: reads, duplicate detection, and
+  updates then address the row by at most one column (or by the hidden rowid),
+  which can return the wrong row or report spurious duplicates. Restore the
+  newer engine before serving traffic from such a tenant. No row migration is
+  needed moving forward: rows written by this engine keep their keys, and
+  single-column key bytes are unchanged for older engines.
 - 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

@@ -167,8 +167,8 @@ func Current() *Document {
 		Constraints: Constraints{
 			NotNull:       Feature{Supported: true, Status: "enforced", SQLState: "23502", Notes: "validated after defaults and assignments; explicit NULL does not select a default"},
 			Unique:        Feature{Supported: true, Status: "enforced", SQLState: "23505", Notes: "unique indexes, INLINE UNIQUE, and ON CONFLICT are enforced atomically"},
-			PrimaryKey:    Feature{Supported: true, Status: "enforced", SQLState: "23505", Notes: "single-column only"},
-			CompositePK:   Feature{Supported: false, Status: "rejected", SQLState: "0A000", Notes: "declare a single-column primary key"},
+			PrimaryKey:    Feature{Supported: true, Status: "enforced", SQLState: "23505", Notes: "inline and table-level single-column keys; INTEGER PRIMARY KEY remains the rowid alias"},
+			CompositePK:   Feature{Supported: true, Status: "enforced", SQLState: "23505", Notes: "table-level PRIMARY KEY (a, b) with ordered columns; key columns are implicitly NOT NULL (23502) and a composite key is never a rowid alias; two inline PRIMARY KEY declarations or an inline/table-level mix are rejected with 0A000; the key is durable under a length-prefixed encoding and reported by PRAGMA table_info pk ordinals and sqlite_master table SQL, but no implicit index is created for it; a conflict target that names the key columns matches them in any order, and ON CONFLICT DO UPDATE cannot assign a key column (0A000)"},
 			ForeignKey:    Feature{Supported: false, Status: "rejected", SQLState: "0A000", Notes: "inline and table-level FOREIGN KEY are rejected; enforce referential integrity in the application, cascades are not implemented"},
 			Check:         Feature{Supported: true, Status: "enforced", SQLState: "23514", Notes: "inline and table-level CHECK are evaluated on INSERT, multi-row INSERT, UPDATE, and ON CONFLICT DO UPDATE; NULL/unknown results pass; predicates are restricted to deterministic, subquery-free expressions over unqualified columns; exposed in sqlite_master/sqlite_schema table SQL only (pg_constraint and information_schema.check_constraints are not emulated)"},
 			DefaultValues: Feature{Supported: true, Status: "evaluated_at_write_time", SQLState: "0A000", Notes: "constant defaults are stored as their evaluated value; CURRENT_TIMESTAMP, CURRENT_DATE, CURRENT_TIME, datetime('now'), date('now'), and time('now') are stored as SQL text and evaluated once per statement on write (once per ALTER statement for an ADD COLUMN backfill), so they are never frozen at DDL time; any other non-constant default is rejected with 0A000"},

pkg/capabilities/capabilities_test.go

@@ -120,3 +120,30 @@ func TestDDLNotesDescribeCheckDependencies(t *testing.T) {
 		t.Fatalf("add_column_backfill notes = %q, want the CHECK backfill failure", doc.DDL.AddColumnBackfill.Notes)
 	}
 }
+
+// Composite primary keys are advertised as supported, with the SQLSTATE callers
+// should expect for a duplicate key.
+func TestCompositePrimaryKeyIsAdvertised(t *testing.T) {
+	doc := Current()
+	composite := doc.Constraints.CompositePK
+	if !composite.Supported {
+		t.Fatalf("composite_primary_key = %+v, want supported", composite)
+	}
+	if composite.Status == "rejected" {
+		t.Fatalf("composite_primary_key status = %q, want a supported status", composite.Status)
+	}
+	if composite.SQLState != "23505" {
+		t.Fatalf("composite_primary_key SQLSTATE = %q, want 23505", composite.SQLState)
+	}
+	if !doc.Constraints.PrimaryKey.Supported {
+		t.Fatalf("primary_key = %+v, want supported", doc.Constraints.PrimaryKey)
+	}
+
+	body, err := json.Marshal(doc)
+	if err != nil {
+		t.Fatalf("marshal: %v", err)
+	}
+	if !strings.Contains(string(body), `"composite_primary_key":{"supported":true`) {
+		t.Fatalf("document does not advertise composite_primary_key as supported: %s", body)
+	}
+}

pkg/executor/composite_primary_key_test.go

@@ -1,36 +1,76 @@
 package executor
 
 import (
+	"errors"
 	"strings"
 	"testing"
+
+	"github.com/danfragoso/pizzasql-next/pkg/pgerror"
+	"github.com/danfragoso/pizzasql-next/pkg/storage"
 )
 
 func TestCreateTablePrimaryKeyConstraints(t *testing.T) {
 	tests := []struct {
-		name           string
-		sql            string
-		wantErr        string
-		wantPrimaryKey string
+		name            string
+		sql             string
+		wantErr         string
+		wantErrSQLState bool
+		wantPrimaryKey  string
+		wantKeyColumns  []string
 	}{
 		{
-			name:    "composite table-level primary key is rejected",
-			sql:     "CREATE TABLE t (a TEXT, b TEXT, PRIMARY KEY (a, b))",
-			wantErr: "composite primary key",
+			name:           "composite table-level primary key is accepted",
+			sql:            "CREATE TABLE t (a TEXT, b TEXT, PRIMARY KEY (a, b))",
+			wantPrimaryKey: "",
+			wantKeyColumns: []string{"a", "b"},
 		},
 		{
-			name:    "duplicate column composite primary key is rejected",
-			sql:     "CREATE TABLE t (a TEXT, PRIMARY KEY (a, a))",
-			wantErr: "composite primary key",
+			name:           "three-column table-level primary key is accepted in declaration order",
+			sql:            "CREATE TABLE t (a TEXT, b TEXT, c TEXT, PRIMARY KEY (c, a, b))",
+			wantPrimaryKey: "",
+			wantKeyColumns: []string{"c", "a", "b"},
 		},
 		{
 			name:           "single-column table-level primary key is accepted",
 			sql:            "CREATE TABLE t (a TEXT, PRIMARY KEY (a))",
 			wantPrimaryKey: "a",
+			wantKeyColumns: []string{"a"},
 		},
 		{
 			name:           "inline primary key is accepted",
 			sql:            "CREATE TABLE t (a TEXT PRIMARY KEY)",
 			wantPrimaryKey: "a",
+			wantKeyColumns: []string{"a"},
+		},
+		{
+			name:            "two inline primary keys are rejected",
+			sql:             "CREATE TABLE t (a TEXT PRIMARY KEY, b TEXT PRIMARY KEY)",
+			wantErr:         "more than one inline PRIMARY KEY",
+			wantErrSQLState: true,
+		},
+		{
+			name:            "inline and table-level primary keys are rejected together",
+			sql:             "CREATE TABLE t (a TEXT PRIMARY KEY, b TEXT, PRIMARY KEY (b))",
+			wantErr:         "mixes an inline PRIMARY KEY with a table-level PRIMARY KEY",
+			wantErrSQLState: true,
+		},
+		{
+			name:            "duplicate column in a composite primary key is rejected",
+			sql:             "CREATE TABLE t (a TEXT, PRIMARY KEY (a, a))",
+			wantErr:         "appears more than once in the PRIMARY KEY",
+			wantErrSQLState: true,
+		},
+		{
+			name:            "unknown primary key column is rejected",
+			sql:             "CREATE TABLE t (a TEXT, PRIMARY KEY (a, missing))",
+			wantErr:         `PRIMARY KEY column "missing" does not exist`,
+			wantErrSQLState: true,
+		},
+		{
+			name:            "two table-level primary keys are rejected",
+			sql:             "CREATE TABLE t (a TEXT, b TEXT, PRIMARY KEY (a), PRIMARY KEY (b))",
+			wantErr:         "declares more than one PRIMARY KEY",
+			wantErrSQLState: true,
 		},
 	}
 
@@ -47,6 +87,9 @@ func TestCreateTablePrimaryKeyConstraints(t *testing.T) {
 				if !strings.Contains(err.Error(), tt.wantErr) {
 					t.Fatalf("error = %q, want it to contain %q", err, tt.wantErr)
 				}
+				if tt.wantErrSQLState && !errors.Is(err, pgerror.ErrFeatureNotSupported) {
+					t.Fatalf("error = %v, want SQLSTATE 0A000 (feature not supported)", err)
+				}
 				if schema.TableExists("t") {
 					t.Fatal("rejected CREATE TABLE left a schema behind")
 				}
@@ -66,7 +109,595 @@ func TestCreateTablePrimaryKeyConstraints(t *testing.T) {
 			if got.PrimaryKey != tt.wantPrimaryKey {
 				t.Fatalf("PrimaryKey = %q, want %q", got.PrimaryKey, tt.wantPrimaryKey)
 			}
-			execMust(t, e, "INSERT INTO t (a) VALUES ('x')")
+			if strings.Join(got.PrimaryKeyList(), ",") != strings.Join(tt.wantKeyColumns, ",") {
+				t.Fatalf("PrimaryKeyList = %v, want %v", got.PrimaryKeyList(), tt.wantKeyColumns)
+			}
+			for _, col := range got.Columns {
+				if got.IsPrimaryKeyColumn(col.Name) && col.Nullable {
+					t.Fatalf("primary key column %s is nullable", col.Name)
+				}
+			}
 		})
 	}
 }
+
+func TestCompositePrimaryKeyDML(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE pairs (a TEXT, b INTEGER, v TEXT, PRIMARY KEY (a, b))")
+
+	execMust(t, e, "INSERT INTO pairs (a, b, v) VALUES ('x', 1, 'one')")
+	execMust(t, e, "INSERT INTO pairs (a, b, v) VALUES ('x', 2, 'two')")
+	execMust(t, e, "INSERT INTO pairs (a, b, v) VALUES ('y', 1, 'other')")
+
+	// A full-key predicate resolves exactly one row.
+	res := execMust(t, e, "SELECT v FROM pairs WHERE a = 'x' AND b = 1")
+	if res.RowCount != 1 || res.Rows[0][0] != "one" {
+		t.Fatalf("full-key select = %#v, want one row 'one'", res.Rows)
+	}
+
+	// A single key column is only a prefix: it must not be treated as a point
+	// lookup that returns whichever row was stored first.
+	res = execMust(t, e, "SELECT b FROM pairs WHERE a = 'x' ORDER BY b")
+	if res.RowCount != 2 {
+		t.Fatalf("prefix select returned %d rows, want 2", res.RowCount)
+	}
+
+	// Duplicate full key.
+	if _, err := execSQL(e, "INSERT INTO pairs (a, b, v) VALUES ('x', 1, 'dup')"); err == nil {
+		t.Fatal("expected a duplicate composite key to fail")
+	} else if !isConflictError(err) {
+		t.Fatalf("duplicate error = %v, want a primary-key conflict", err)
+	}
+
+	// Missing component: NOT NULL violation, never a rowid auto-assign.
+	_, err := execSQL(e, "INSERT INTO pairs (a, v) VALUES ('z', 'no b')")
+	var notNull *storage.NotNullViolationError
+	if !errors.As(err, &notNull) || notNull.Column != "b" {
+		t.Fatalf("missing component error = %v, want a NOT NULL violation on b", err)
+	}
+	_, err = execSQL(e, "INSERT INTO pairs (a, b, v) VALUES ('z', NULL, 'null b')")
+	if !errors.As(err, &notNull) || notNull.Column != "b" {
+		t.Fatalf("NULL component error = %v, want a NOT NULL violation on b", err)
+	}
+	if res := execMust(t, e, "SELECT count(*) FROM pairs"); res.Rows[0][0] != int64(3) {
+		t.Fatalf("rejected inserts stored rows: count = %v", res.Rows[0][0])
+	}
+
+	// Update by full key.
+	res = execMust(t, e, "UPDATE pairs SET v = 'ONE' WHERE a = 'x' AND b = 1")
+	if res.RowsAffected != 1 {
+		t.Fatalf("update affected %d rows, want 1", res.RowsAffected)
+	}
+	if res = execMust(t, e, "SELECT v FROM pairs WHERE a = 'x' AND b = 1"); res.RowCount != 1 || res.Rows[0][0] != "ONE" {
+		t.Fatalf("row after update = %#v, want 'ONE'", res.Rows)
+	}
+
+	// Moving one key component moves the durable row.
+	execMust(t, e, "UPDATE pairs SET b = 3 WHERE a = 'x' AND b = 1")
+	if res = execMust(t, e, "SELECT v FROM pairs WHERE a = 'x' AND b = 3"); res.RowCount != 1 || res.Rows[0][0] != "ONE" {
+		t.Fatalf("row under the new key = %#v, want 'ONE'", res.Rows)
+	}
+	if res = execMust(t, e, "SELECT v FROM pairs WHERE a = 'x' AND b = 1"); res.RowCount != 0 {
+		t.Fatalf("old key still resolves %d rows", res.RowCount)
+	}
+
+	// Moving onto an existing key is a duplicate.
+	if _, err := execSQL(e, "UPDATE pairs SET b = 2 WHERE a = 'x' AND b = 3"); err == nil {
+		t.Fatal("expected a key move onto an existing row to fail")
+	} else if !isConflictError(err) {
+		t.Fatalf("key move error = %v, want a primary-key conflict", err)
+	}
+
+	// Clearing a key component is a NOT NULL violation, not a key move.
+	_, err = execSQL(e, "UPDATE pairs SET b = NULL WHERE a = 'x' AND b = 2")
+	if !errors.As(err, &notNull) || notNull.Column != "b" {
+		t.Fatalf("NULL key update error = %v, want a NOT NULL violation on b", err)
+	}
+
+	// Delete by full key removes exactly one row.
+	res = execMust(t, e, "DELETE FROM pairs WHERE a = 'x' AND b = 2")
+	if res.RowsAffected != 1 {
+		t.Fatalf("delete affected %d rows, want 1", res.RowsAffected)
+	}
+	if res = execMust(t, e, "SELECT count(*) FROM pairs"); res.Rows[0][0] != int64(2) {
+		t.Fatalf("row count after delete = %v, want 2", res.Rows[0][0])
+	}
+}
+
+func TestCompositePrimaryKeyOnConflict(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE counters (a TEXT, b TEXT, n INTEGER, PRIMARY KEY (a, b))")
+	execMust(t, e, "INSERT INTO counters (a, b, n) VALUES ('x', 'y', 1)")
+
+	execMust(t, e, "INSERT INTO counters (a, b, n) VALUES ('x', 'y', 2) ON CONFLICT (a, b) DO UPDATE SET n = excluded.n")
+	if res := execMust(t, e, "SELECT n FROM counters WHERE a = 'x' AND b = 'y'"); res.RowCount != 1 || res.Rows[0][0] != int64(2) {
+		t.Fatalf("upsert result = %#v, want one row with n = 2", res.Rows)
+	}
+
+	execMust(t, e, "INSERT INTO counters (a, b, n) VALUES ('x', 'y', 9) ON CONFLICT (a, b) DO NOTHING")
+	if res := execMust(t, e, "SELECT n FROM counters WHERE a = 'x' AND b = 'y'"); res.RowCount != 1 || res.Rows[0][0] != int64(2) {
+		t.Fatalf("DO NOTHING result = %#v, want the existing row with n = 2", res.Rows)
+	}
+
+	// A targetless conflict clause uses the full primary-key column list.
+	execMust(t, e, "INSERT INTO counters (a, b, n) VALUES ('x', 'y', 9) ON CONFLICT DO NOTHING")
+	if res := execMust(t, e, "SELECT count(*) FROM counters"); res.Rows[0][0] != int64(1) {
+		t.Fatalf("targetless DO NOTHING inserted a row: count = %v", res.Rows[0][0])
+	}
+
+	// REPLACE resolves the conflict on the full key.
+	execMust(t, e, "INSERT OR REPLACE INTO counters (a, b, n) VALUES ('x', 'y', 5)")
+	if res := execMust(t, e, "SELECT n FROM counters WHERE a = 'x' AND b = 'y'"); res.RowCount != 1 || res.Rows[0][0] != int64(5) {
+		t.Fatalf("REPLACE result = %#v, want one row with n = 5", res.Rows)
+	}
+
+	// A different key is not a conflict.
+	execMust(t, e, "INSERT INTO counters (a, b, n) VALUES ('x', 'z', 7) ON CONFLICT (a, b) DO NOTHING")
+	if res := execMust(t, e, "SELECT n FROM counters WHERE a = 'x' AND b = 'z'"); res.RowCount != 1 || res.Rows[0][0] != int64(7) {
+		t.Fatalf("new key result = %#v, want one row with n = 7", res.Rows)
+	}
+
+	// ON CONFLICT UPDATE on a full key with the columns in the other order is a
+	// distinct target and must not match the declared key.
+	execMust(t, e, "INSERT INTO counters (a, b, n) VALUES ('q', 'r', 1) ON CONFLICT (b, a) DO UPDATE SET n = excluded.n")
+	if res := execMust(t, e, "SELECT count(*) FROM counters WHERE a = 'q' AND b = 'r'"); res.Rows[0][0] != int64(1) {
+		t.Fatalf("reordered conflict target rows = %v, want 1", res.Rows[0][0])
+	}
+}
+
+func TestCompositePrimaryKeyCatalog(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE pairs (a TEXT, b INTEGER, v TEXT, PRIMARY KEY (a, b))")
+
+	res := execMust(t, e, "SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'pairs'")
+	if res.RowCount != 1 {
+		t.Fatalf("sqlite_master rows = %d, want 1", res.RowCount)
+	}
+	createSQL, _ := res.Rows[0][0].(string)
+	if !strings.Contains(createSQL, "PRIMARY KEY (`a`, `b`)") {
+		t.Fatalf("sqlite_master SQL = %q, want a table-level PRIMARY KEY (a, b)", createSQL)
+	}
+	if strings.Count(createSQL, "PRIMARY KEY") != 1 {
+		t.Fatalf("sqlite_master SQL = %q, want exactly one PRIMARY KEY clause", createSQL)
+	}
+
+	res = execMust(t, e, "PRAGMA table_info(pairs)")
+	if res.RowCount != 3 {
+		t.Fatalf("table_info rows = %d, want 3", res.RowCount)
+	}
+	wantPK := []int64{1, 2, 0}
+	for i, row := range res.Rows {
+		if row[1] != []string{"a", "b", "v"}[i] {
+			t.Fatalf("table_info row %d column = %v", i, row[1])
+		}
+		if row[5] != wantPK[i] {
+			t.Fatalf("table_info pk ordinal for %v = %v, want %d", row[1], row[5], wantPK[i])
+		}
+	}
+	// Both key columns are implicitly NOT NULL.
+	if res.Rows[0][3] != int64(1) || res.Rows[1][3] != int64(1) {
+		t.Fatalf("table_info notnull = %v/%v, want 1/1", res.Rows[0][3], res.Rows[1][3])
+	}
+
+	res = execMust(t, e, "PRAGMA table_xinfo(pairs)")
+	if res.RowCount != 3 || res.Rows[1][5] != int64(2) {
+		t.Fatalf("table_xinfo = %#v, want pk ordinal 2 on the second key column", res.Rows)
+	}
+}
+
+func TestCompositePrimaryKeyDurableReload(t *testing.T) {
+	pool, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE pairs (a TEXT, b TEXT, v TEXT, PRIMARY KEY (a, b))")
+	execMust(t, e, "INSERT INTO pairs (a, b, v) VALUES ('x', 'y', 'one')")
+	execMust(t, e, "INSERT INTO pairs (a, b, v) VALUES ('x', 'z', 'two')")
+
+	// Simulate a process restart: new managers share only the durable KV state.
+	reloadedSchema := storage.NewSchemaManager(pool, "testdb")
+	reloadedTable := storage.NewTableManager(pool, reloadedSchema, "testdb")
+	reloaded := newExec(reloadedSchema, reloadedTable)
+
+	got, err := reloadedSchema.GetSchema("pairs")
+	if err != nil {
+		t.Fatalf("get reloaded schema: %v", err)
+	}
+	if got.PrimaryKey != "" || strings.Join(got.PrimaryKeyList(), ",") != "a,b" {
+		t.Fatalf("reloaded key = %q/%v, want composite a,b", got.PrimaryKey, got.PrimaryKeyList())
+	}
+
+	res := execMust(t, reloaded, "SELECT v FROM pairs WHERE a = 'x' AND b = 'y'")
+	if res.RowCount != 1 || res.Rows[0][0] != "one" {
+		t.Fatalf("reloaded full-key read = %#v, want one row 'one'", res.Rows)
+	}
+	if res = execMust(t, reloaded, "SELECT count(*) FROM pairs"); res.Rows[0][0] != int64(2) {
+		t.Fatalf("reloaded count = %v, want 2", res.Rows[0][0])
+	}
+
+	execMust(t, reloaded, "UPDATE pairs SET v = 'updated' WHERE a = 'x' AND b = 'z'")
+	if res = execMust(t, reloaded, "SELECT v FROM pairs WHERE a = 'x' AND b = 'z'"); res.RowCount != 1 || res.Rows[0][0] != "updated" {
+		t.Fatalf("reloaded update = %#v, want one row 'updated'", res.Rows)
+	}
+}
+
+func TestCompositePrimaryKeyRejectsDuplicateInsert(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE pairs (a TEXT, b TEXT, PRIMARY KEY (a, b))")
+	execMust(t, e, "INSERT INTO pairs (a, b) VALUES ('x', 'y')")
+	if _, err := execSQL(e, "INSERT INTO pairs (a, b) VALUES ('x', 'y')"); err == nil {
+		t.Fatal("expected a duplicate composite key to fail")
+	} else if !isConflictError(err) {
+		t.Fatalf("duplicate error = %v, want a primary-key conflict", err)
+	}
+	execMust(t, e, "INSERT INTO pairs (a, b) VALUES ('x', 'z')")
+	execMust(t, e, "INSERT INTO pairs (a, b) VALUES ('y', 'y')")
+	if res := execMust(t, e, "SELECT count(*) FROM pairs"); res.Rows[0][0] != int64(3) {
+		t.Fatalf("row count = %v, want 3", res.Rows[0][0])
+	}
+}
+
+func TestIntegerPrimaryKeyRemainsRowIDAlias(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT)")
+	execMust(t, e, "INSERT INTO t (v) VALUES ('auto')")
+	res := execMust(t, e, "SELECT id, rowid FROM t")
+	if res.RowCount != 1 || res.Rows[0][0] != int64(1) || res.Rows[0][1] != int64(1) {
+		t.Fatalf("INTEGER PRIMARY KEY row = %#v, want id and rowid 1", res.Rows)
+	}
+
+	// A composite key is never a rowid alias: the key columns keep the supplied
+	// values and the hidden rowid is engine-assigned.
+	execMust(t, e, "CREATE TABLE pairs (a INTEGER, b TEXT, PRIMARY KEY (a, b))")
+	execMust(t, e, "INSERT INTO pairs (a, b) VALUES (7, 'x')")
+	res = execMust(t, e, "SELECT a, b, rowid FROM pairs")
+	if res.RowCount != 1 || res.Rows[0][0] != int64(7) || res.Rows[0][2] != int64(1) {
+		t.Fatalf("composite key row = %#v, want a=7 b='x' rowid=1", res.Rows)
+	}
+}
+
+func TestCompositePrimaryKeyInsertSelect(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE src (a TEXT, b TEXT)")
+	execMust(t, e, "INSERT INTO src (a, b) VALUES ('x', 'y'), ('x', 'z'), ('q', 'r')")
+	execMust(t, e, "CREATE TABLE pairs (a TEXT, b TEXT, v TEXT, PRIMARY KEY (a, b))")
+
+	execMust(t, e, "INSERT INTO pairs (a, b, v) SELECT a, b, 'copied' FROM src")
+	if res := execMust(t, e, "SELECT count(*) FROM pairs"); res.Rows[0][0] != int64(3) {
+		t.Fatalf("bulk insert stored %v rows, want 3", res.Rows[0][0])
+	}
+	if res := execMust(t, e, "SELECT v FROM pairs WHERE a = 'x' AND b = 'z'"); res.RowCount != 1 || res.Rows[0][0] != "copied" {
+		t.Fatalf("bulk row = %#v, want one row 'copied'", res.Rows)
+	}
+
+	// A bulk insert that repeats a key fails atomically.
+	if _, err := execSQL(e, "INSERT INTO pairs (a, b, v) SELECT a, b, 'again' FROM src"); err == nil {
+		t.Fatal("expected a duplicate-key bulk insert to fail")
+	} else if !isConflictError(err) {
+		t.Fatalf("bulk duplicate error = %v, want a key conflict", err)
+	}
+	if res := execMust(t, e, "SELECT count(*) FROM pairs"); res.Rows[0][0] != int64(3) {
+		t.Fatalf("failed bulk insert changed the row count: %v", res.Rows[0][0])
+	}
+}
+
+func TestCompositePrimaryKeyAlterTable(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE pairs (a TEXT, b TEXT, v TEXT, PRIMARY KEY (a, b))")
+	execMust(t, e, "INSERT INTO pairs (a, b, v) VALUES ('x', 'y', 'one')")
+
+	// Renaming a key column keeps the key list and the durable row keys.
+	execMust(t, e, "ALTER TABLE pairs RENAME COLUMN a TO k")
+	got, err := schema.GetSchema("pairs")
+	if err != nil {
+		t.Fatalf("get schema: %v", err)
+	}
+	if strings.Join(got.PrimaryKeyList(), ",") != "k,b" {
+		t.Fatalf("key list after rename = %v, want [k b]", got.PrimaryKeyList())
+	}
+	if res := execMust(t, e, "SELECT v FROM pairs WHERE k = 'x' AND b = 'y'"); res.RowCount != 1 || res.Rows[0][0] != "one" {
+		t.Fatalf("read after rename = %#v, want one row 'one'", res.Rows)
+	}
+
+	// A key column cannot be dropped, so the key arity is immutable.
+	if _, err := execSQL(e, "ALTER TABLE pairs DROP COLUMN b"); err == nil {
+		t.Fatal("expected DROP COLUMN of a key column to fail")
+	} else if !strings.Contains(err.Error(), "cannot drop primary key column") {
+		t.Fatalf("DROP COLUMN error = %v, want a primary-key rejection", err)
+	}
+}
+
+// An upsert updates the conflicting row in place, so it cannot move that row to
+// a different key. Accepting the assignment would leave the durable row under
+// the old key while its body carries the new key values, and a later full-key
+// update would create a second row.
+func TestUpsertRejectsPrimaryKeyAssignment(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE pairs (a TEXT, b TEXT, v TEXT, PRIMARY KEY (a, b))")
+	execMust(t, e, "INSERT INTO pairs (a, b, v) VALUES ('x', 'y', 'one')")
+
+	for _, sql := range []string{
+		"INSERT INTO pairs (a, b, v) VALUES ('x', 'y', 'two') ON CONFLICT (a, b) DO UPDATE SET v = excluded.v, a = 'moved'",
+		"INSERT INTO pairs (a, b, v) VALUES ('new', 'key', 'v') ON CONFLICT (a, b) DO UPDATE SET b = excluded.b",
+	} {
+		_, err := execSQL(e, sql)
+		if err == nil {
+			t.Fatalf("%q succeeded, want a primary-key rejection", sql)
+		}
+		if !errors.Is(err, pgerror.ErrFeatureNotSupported) {
+			t.Fatalf("%q error = %v, want SQLSTATE 0A000", sql, err)
+		}
+		if !strings.Contains(err.Error(), "cannot assign primary key column") {
+			t.Fatalf("%q error = %v, want a primary-key assignment message", sql, err)
+		}
+	}
+
+	// No ghost row: the original row kept both its key and its value.
+	if res := execMust(t, e, "SELECT a, b, v FROM pairs"); res.RowCount != 1 || res.Rows[0][0] != "x" || res.Rows[0][1] != "y" || res.Rows[0][2] != "one" {
+		t.Fatalf("rows after rejected upserts = %#v, want the original ('x','y','one')", res.Rows)
+	}
+	if res := execMust(t, e, "SELECT v FROM pairs WHERE a = 'x' AND b = 'y'"); res.RowCount != 1 || res.Rows[0][0] != "one" {
+		t.Fatalf("full-key read after rejected upserts = %#v", res.Rows)
+	}
+	execMust(t, e, "UPDATE pairs SET v = 'updated' WHERE a = 'x' AND b = 'y'")
+	if res := execMust(t, e, "SELECT count(*) FROM pairs"); res.Rows[0][0] != int64(1) {
+		t.Fatalf("row count after full-key update = %v, want 1 (no split row)", res.Rows[0][0])
+	}
+
+	// A single-column key gets the same rejection.
+	execMust(t, e, "CREATE TABLE single (id INTEGER PRIMARY KEY, v TEXT)")
+	execMust(t, e, "INSERT INTO single (v) VALUES ('one')")
+	if _, err := execSQL(e, "INSERT INTO single (id, v) VALUES (1, 'two') ON CONFLICT (id) DO UPDATE SET id = excluded.id"); err == nil {
+		t.Fatal("single-column upsert key assignment succeeded, want a rejection")
+	} else if !errors.Is(err, pgerror.ErrFeatureNotSupported) {
+		t.Fatalf("single-column upsert error = %v, want SQLSTATE 0A000", err)
+	}
+}
+
+// A key column that takes its value from a constant DEFAULT must be resolved
+// before conflict matching, otherwise the candidate key is encoded from a
+// missing component and the upsert misses its own row.
+func TestUpsertWithConstantDefaultKeyColumns(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE konst (a TEXT DEFAULT 'd', b TEXT DEFAULT 'e', v TEXT, PRIMARY KEY (a, b))")
+
+	execMust(t, e, "INSERT INTO konst (b, v) VALUES ('y', 'two')")
+	if res := execMust(t, e, "SELECT a, b, v FROM konst"); res.RowCount != 1 || res.Rows[0][0] != "d" || res.Rows[0][1] != "y" {
+		t.Fatalf("defaulted key row = %#v, want ('d','y')", res.Rows)
+	}
+
+	// DO UPDATE hits the row whose key came from the DEFAULT.
+	execMust(t, e, "INSERT INTO konst (b, v) VALUES ('y', 'updated') ON CONFLICT (a, b) DO UPDATE SET v = excluded.v")
+	if res := execMust(t, e, "SELECT v FROM konst WHERE a = 'd' AND b = 'y'"); res.RowCount != 1 || res.Rows[0][0] != "updated" {
+		t.Fatalf("DO UPDATE result = %#v, want one row 'updated'", res.Rows)
+	}
+	// DO NOTHING suppresses the duplicate.
+	execMust(t, e, "INSERT INTO konst (b, v) VALUES ('y', 'ignored') ON CONFLICT (a, b) DO NOTHING")
+	if res := execMust(t, e, "SELECT v FROM konst WHERE a = 'd' AND b = 'y'"); res.RowCount != 1 || res.Rows[0][0] != "updated" {
+		t.Fatalf("DO NOTHING result = %#v, want the existing 'updated' row", res.Rows)
+	}
+	// Targetless ON CONFLICT uses the defaulted key too.
+	execMust(t, e, "INSERT INTO konst (b, v) VALUES ('y', 'ignored') ON CONFLICT DO NOTHING")
+	if res := execMust(t, e, "SELECT count(*) FROM konst"); res.Rows[0][0] != int64(1) {
+		t.Fatalf("targetless DO NOTHING inserted a row: count = %v", res.Rows[0][0])
+	}
+	// INSERT OR REPLACE replaces the defaulted-key row instead of duplicating it.
+	execMust(t, e, "INSERT OR REPLACE INTO konst (b, v) VALUES ('y', 'replaced')")
+	if res := execMust(t, e, "SELECT count(*) FROM konst"); res.Rows[0][0] != int64(1) {
+		t.Fatalf("OR REPLACE left %v rows, want 1", res.Rows[0][0])
+	}
+	if res := execMust(t, e, "SELECT v FROM konst WHERE a = 'd' AND b = 'y'"); res.RowCount != 1 || res.Rows[0][0] != "replaced" {
+		t.Fatalf("OR REPLACE result = %#v, want one row 'replaced'", res.Rows)
+	}
+
+	// A single-column TEXT key with a DEFAULT follows the same path and must not
+	// fall back to an engine-assigned rowid key.
+	execMust(t, e, "CREATE TABLE single (id TEXT DEFAULT 'k', v TEXT, PRIMARY KEY (id))")
+	execMust(t, e, "INSERT INTO single (v) VALUES ('a')")
+	if res := execMust(t, e, "SELECT id FROM single"); res.RowCount != 1 || res.Rows[0][0] != "k" {
+		t.Fatalf("single defaulted key = %#v, want one row with id 'k'", res.Rows)
+	}
+	execMust(t, e, "INSERT INTO single (v) VALUES ('b') ON CONFLICT (id) DO UPDATE SET v = excluded.v")
+	if res := execMust(t, e, "SELECT v FROM single"); res.RowCount != 1 || res.Rows[0][0] != "b" {
+		t.Fatalf("single DO UPDATE = %#v, want one row 'b'", res.Rows)
+	}
+	execMust(t, e, "INSERT INTO single (v) VALUES ('c') ON CONFLICT DO NOTHING")
+	if res := execMust(t, e, "SELECT count(*) FROM single"); res.Rows[0][0] != int64(1) {
+		t.Fatalf("single targetless DO NOTHING count = %v, want 1", res.Rows[0][0])
+	}
+	execMust(t, e, "INSERT OR REPLACE INTO single (v) VALUES ('d')")
+	if res := execMust(t, e, "SELECT v FROM single"); res.RowCount != 1 || res.Rows[0][0] != "d" {
+		t.Fatalf("single OR REPLACE = %#v, want one row 'd'", res.Rows)
+	}
+}
+
+func TestAlterTableAddColumnPrimaryKeyRejected(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE t (a TEXT)")
+	execMust(t, e, "INSERT INTO t (a) VALUES ('x')")
+
+	_, err := execSQL(e, "ALTER TABLE t ADD COLUMN id INTEGER PRIMARY KEY")
+	if err == nil {
+		t.Fatal("ALTER TABLE ADD COLUMN PRIMARY KEY succeeded, want a rejection")
+	}
+	if !errors.Is(err, pgerror.ErrFeatureNotSupported) {
+		t.Fatalf("error = %v, want SQLSTATE 0A000", err)
+	}
+	if !strings.Contains(err.Error(), "cannot declare PRIMARY KEY") {
+		t.Fatalf("error = %v, want an ADD COLUMN PRIMARY KEY message", err)
+	}
+	got, err := schema.GetSchema("t")
+	if err != nil {
+		t.Fatalf("get schema: %v", err)
+	}
+	if _, exists := got.GetColumn("id"); exists {
+		t.Fatal("rejected ADD COLUMN published the column")
+	}
+}
+
+func TestGeneratedColumnInPrimaryKeyRejected(t *testing.T) {
+	for _, sql := range []string{
+		"CREATE TABLE g (a TEXT, b TEXT GENERATED ALWAYS AS (a) STORED, PRIMARY KEY (b))",
+		"CREATE TABLE g (a TEXT, b TEXT GENERATED ALWAYS AS (a) STORED PRIMARY KEY)",
+	} {
+		_, schema, table := newTestDB(t)
+		e := newExec(schema, table)
+		_, err := execSQL(e, sql)
+		if err == nil {
+			t.Fatalf("%q succeeded, want a generated-column rejection", sql)
+		}
+		if !errors.Is(err, pgerror.ErrFeatureNotSupported) {
+			t.Fatalf("%q error = %v, want SQLSTATE 0A000", sql, err)
+		}
+		if !strings.Contains(err.Error(), "cannot be part of a PRIMARY KEY") {
+			t.Fatalf("%q error = %v, want a PRIMARY KEY message", sql, err)
+		}
+		if schema.TableExists("g") {
+			t.Fatalf("%q left a schema behind", sql)
+		}
+	}
+}
+
+// RETURNING must observe key components that the storage layer resolved from a
+// catalog DEFAULT; the candidate row does not carry them until then.
+func TestCompositePrimaryKeyReturningWithDefaultedKey(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE pairs (a TEXT DEFAULT 'd', b TEXT DEFAULT 'e', v TEXT, PRIMARY KEY (a, b))")
+
+	res := execMust(t, e, "INSERT INTO pairs (v) VALUES ('one') RETURNING a, b, v")
+	if res.RowCount != 1 || res.Rows[0][0] != "d" || res.Rows[0][1] != "e" || res.Rows[0][2] != "one" {
+		t.Fatalf("RETURNING = %#v, want ('d','e','one')", res.Rows)
+	}
+
+	// A partially defaulted key also returns the stored values.
+	res = execMust(t, e, "INSERT INTO pairs (b, v) VALUES ('y', 'two') RETURNING *")
+	if res.RowCount != 1 {
+		t.Fatalf("RETURNING * rows = %d, want 1", res.RowCount)
+	}
+	values := map[string]interface{}{}
+	for i, name := range res.Columns {
+		values[name] = res.Rows[0][i]
+	}
+	if values["a"] != "d" || values["b"] != "y" || values["v"] != "two" {
+		t.Fatalf("RETURNING * row = %#v, want a='d' b='y' v='two'", values)
+	}
+}
+
+// ON CONFLICT targets address the key column set, not its declaration order:
+// (b, a) and (a, b) name the same PRIMARY KEY (a, b). SQLite requires the
+// constraint's order; this engine already accepts arbitrary target lists and
+// documents the set match here.
+func TestOnConflictReorderedPrimaryKeyTarget(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE pairs (a TEXT, b TEXT, v TEXT, PRIMARY KEY (a, b))")
+	execMust(t, e, "INSERT INTO pairs (a, b, v) VALUES ('x', 'y', 'one')")
+
+	execMust(t, e, "INSERT INTO pairs (a, b, v) VALUES ('x', 'y', 'two') ON CONFLICT (b, a) DO UPDATE SET v = excluded.v")
+	res := execMust(t, e, "SELECT count(*) FROM pairs")
+	if res.Rows[0][0] != int64(1) {
+		t.Fatalf("reordered target left %v rows, want 1", res.Rows[0][0])
+	}
+	res = execMust(t, e, "SELECT v FROM pairs WHERE a = 'x' AND b = 'y'")
+	if res.RowCount != 1 || res.Rows[0][0] != "two" {
+		t.Fatalf("reordered target result = %#v, want the updated row 'two'", res.Rows)
+	}
+
+	execMust(t, e, "INSERT INTO pairs (a, b, v) VALUES ('x', 'y', 'three') ON CONFLICT (b, a) DO NOTHING")
+	res = execMust(t, e, "SELECT v FROM pairs WHERE a = 'x' AND b = 'y'")
+	if res.RowCount != 1 || res.Rows[0][0] != "two" {
+		t.Fatalf("reordered DO NOTHING result = %#v, want the existing 'two' row", res.Rows)
+	}
+}
+
+// A composite primary key must coexist with scan-validated UNIQUE indexes:
+// key moves keep index lookups correct, ON CONFLICT (unique_col) upserts the
+// row, and INSERT OR REPLACE resolves both the unique index and the primary key.
+func TestCompositePrimaryKeyWithUniqueIndex(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE members (org TEXT, user_id INTEGER, email TEXT UNIQUE, v TEXT, PRIMARY KEY (org, user_id))")
+	execMust(t, e, "INSERT INTO members (org, user_id, email, v) VALUES ('o', 1, 'a@x', 'one')")
+
+	// ON CONFLICT (unique_col) updates the row under its composite key.
+	execMust(t, e, "INSERT INTO members (org, user_id, email, v) VALUES ('o', 1, 'a@x', 'two') ON CONFLICT (email) DO UPDATE SET v = excluded.v")
+	if res := execMust(t, e, "SELECT v FROM members WHERE org = 'o' AND user_id = 1"); res.RowCount != 1 || res.Rows[0][0] != "two" {
+		t.Fatalf("unique-target upsert = %#v, want one row 'two'", res.Rows)
+	}
+
+	// A unique conflict on a different key is suppressed by DO NOTHING.
+	execMust(t, e, "INSERT INTO members (org, user_id, email, v) VALUES ('o', 2, 'a@x', 'duplicate email') ON CONFLICT (email) DO NOTHING")
+	if res := execMust(t, e, "SELECT count(*) FROM members"); res.Rows[0][0] != int64(1) {
+		t.Fatalf("unique DO NOTHING count = %v, want 1", res.Rows[0][0])
+	}
+
+	// Moving a key component keeps the unique index entry attached to the row.
+	execMust(t, e, "UPDATE members SET user_id = 3 WHERE org = 'o' AND user_id = 1")
+	if res := execMust(t, e, "SELECT v FROM members WHERE org = 'o' AND user_id = 3"); res.RowCount != 1 || res.Rows[0][0] != "two" {
+		t.Fatalf("row after key move = %#v, want one row 'two'", res.Rows)
+	}
+	if res := execMust(t, e, "SELECT v FROM members WHERE email = 'a@x'"); res.RowCount != 1 || res.Rows[0][0] != "two" {
+		t.Fatalf("unique index after key move = %#v, want one row 'two'", res.Rows)
+	}
+
+	// OR REPLACE resolves the unique index by removing the old row and keeping
+	// exactly the new key.
+	execMust(t, e, "INSERT OR REPLACE INTO members (org, user_id, email, v) VALUES ('o', 9, 'a@x', 'replaced')")
+	if res := execMust(t, e, "SELECT count(*) FROM members"); res.Rows[0][0] != int64(1) {
+		t.Fatalf("unique OR REPLACE count = %v, want 1", res.Rows[0][0])
+	}
+	if res := execMust(t, e, "SELECT v FROM members WHERE org = 'o' AND user_id = 9"); res.RowCount != 1 || res.Rows[0][0] != "replaced" {
+		t.Fatalf("unique OR REPLACE row = %#v, want one row 'replaced'", res.Rows)
+	}
+	if res := execMust(t, e, "SELECT v FROM members WHERE org = 'o' AND user_id = 3"); res.RowCount != 0 {
+		t.Fatalf("old composite key still resolves %d rows", res.RowCount)
+	}
+
+	// OR REPLACE resolving the primary key works with a unique index present.
+	execMust(t, e, "INSERT OR REPLACE INTO members (org, user_id, email, v) VALUES ('o', 9, 'b@x', 'key replaced')")
+	if res := execMust(t, e, "SELECT email, v FROM members WHERE org = 'o' AND user_id = 9"); res.RowCount != 1 || res.Rows[0][0] != "b@x" || res.Rows[0][1] != "key replaced" {
+		t.Fatalf("PK OR REPLACE row = %#v, want ('b@x','key replaced')", res.Rows)
+	}
+	if res := execMust(t, e, "SELECT v FROM members WHERE email = 'a@x'"); res.RowCount != 0 {
+		t.Fatalf("stale unique index entry still resolves %d rows", res.RowCount)
+	}
+}
+
+// A composite table has no INTEGER PRIMARY KEY alias, so last_insert_rowid()
+// reports the engine-assigned hidden rowid and advances per inserted row.
+func TestLastInsertRowIDOnCompositeTable(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE pairs (a TEXT, b TEXT, PRIMARY KEY (a, b))")
+
+	res := execMust(t, e, "INSERT INTO pairs (a, b) VALUES ('x', 'y')")
+	if res.LastInsertID != 1 {
+		t.Fatalf("first LastInsertID = %d, want 1", res.LastInsertID)
+	}
+	if res := execMust(t, e, "SELECT last_insert_rowid()"); res.RowCount != 1 || res.Rows[0][0] != int64(1) {
+		t.Fatalf("last_insert_rowid() = %#v, want 1", res.Rows)
+	}
+
+	res = execMust(t, e, "INSERT INTO pairs (a, b) VALUES ('x', 'z')")
+	if res.LastInsertID != 2 {
+		t.Fatalf("second LastInsertID = %d, want 2", res.LastInsertID)
+	}
+	if res := execMust(t, e, "SELECT last_insert_rowid()"); res.RowCount != 1 || res.Rows[0][0] != int64(2) {
+		t.Fatalf("last_insert_rowid() after second insert = %#v, want 2", res.Rows)
+	}
+
+	// A suppressed conflict does not advance last_insert_rowid().
+	execMust(t, e, "INSERT INTO pairs (a, b) VALUES ('x', 'z') ON CONFLICT (a, b) DO NOTHING")
+	if res := execMust(t, e, "SELECT last_insert_rowid()"); res.Rows[0][0] != int64(2) {
+		t.Fatalf("last_insert_rowid() after upsert = %v, want 2", res.Rows[0][0])
+	}
+}

pkg/executor/ddl_defaults.go

@@ -154,29 +154,41 @@ func evalDefaultExpressionAt(text string, now time.Time) (interface{}, error) {
 	return nil, pgerror.Unsupported("stored DEFAULT expression %q is not supported", text)
 }
 
-// statementDefaults evaluates every expression default in the schema once for
-// the statement. A multi-row INSERT therefore stamps all its rows with one
-// value, while a later statement reads the clock again. Constant defaults stay
-// in the catalog and are applied by the storage layer.
+// statementDefaults resolves every default in the schema for the statement.
+// Expression defaults are evaluated once, so a multi-row INSERT stamps all its
+// rows with one value while a later statement reads the clock again. Constant
+// catalog defaults are included too: the conflict-matching and RETURNING paths
+// inspect the candidate row before storage fills missing columns, so a key
+// column that comes from a DEFAULT must already be present in it.
 func (e *Executor) statementDefaults(schema *storage.Schema) (map[string]interface{}, error) {
 	// One instant per statement: two clock-default columns in the same row must
 	// not disagree because the loop sampled the clock twice. The clock is read
 	// lazily so a schema without expression defaults never samples it.
 	var now time.Time
+	sampledClock := false
 	var defaults map[string]interface{}
+	add := func(name string, value interface{}) {
+		if defaults == nil {
+			defaults = make(map[string]interface{})
+		}
+		defaults[name] = value
+	}
 	for _, column := range schema.Columns {
 		if column.DefaultExpr == "" {
+			if column.Default != nil {
+				add(column.Name, column.Default)
+			}
 			continue
 		}
-		if defaults == nil {
+		if !sampledClock {
 			now = e.statementNow()
-			defaults = make(map[string]interface{})
+			sampledClock = true
 		}
 		value, err := evalDefaultExpressionAt(column.DefaultExpr, now)
 		if err != nil {
 			return nil, fmt.Errorf("evaluate DEFAULT for column %q: %w", column.Name, err)
 		}
-		defaults[column.Name] = value
+		add(column.Name, value)
 	}
 	return defaults, nil
 }
@@ -191,8 +203,8 @@ func (e *Executor) statementNow() time.Time {
 	return time.Now().UTC()
 }
 
-// applyRowDefaults fills the expression-defaulted columns a statement omitted.
-// An explicit NULL is already present in the row, so it wins over the default.
+// applyRowDefaults fills the defaulted columns a statement omitted. An explicit
+// NULL is already present in the row, so it wins over the default.
 func applyRowDefaults(row storage.Row, defaults map[string]interface{}) {
 	for name, value := range defaults {
 		if _, present := lookupRowValue(row, name); present {

pkg/executor/executor.go

@@ -555,7 +555,7 @@ func (e *Executor) executeSelect(stmt *parser.SelectStmt) (*Result, error) {
 		for _, condition := range splitANDClauses(effectiveWhere) {
 			colName, colValue, isEquality := e.extractSelectLookupCondition(condition, stmt.From[0])
 			if isEquality {
-				if strings.EqualFold(schema.PrimaryKey, colName) && safePrimaryKeyLiteral(colValue) {
+				if schema.IsSingleColumnPrimaryKey() && strings.EqualFold(schema.PrimaryKey, colName) && safePrimaryKeyLiteral(colValue) {
 					row, getErr := e.session.GetByPK(tableName, fmt.Sprintf("%v", colValue))
 					if getErr == nil {
 						normalizeRowBySchema(row, schema)
@@ -2104,7 +2104,7 @@ func (e *Executor) executeJoinsWithMode(tableRef parser.TableRef, leftRows []sto
 		if err != nil {
 			return nil, err
 		}
-		if canHash && strings.EqualFold(rightSchema.PrimaryKey, rightKey) && len(leftRows) <= 256 {
+		if canHash && rightSchema.IsSingleColumnPrimaryKey() && strings.EqualFold(rightSchema.PrimaryKey, rightKey) && len(leftRows) <= 256 {
 			seen := make(map[string]bool, len(leftRows))
 			for _, left := range leftRows {
 				key := joinKeyString(left, leftKey)
@@ -2621,12 +2621,14 @@ func isConflictError(err error) bool {
 // target uses an indexed lookup; only an unindexed composite target falls back
 // to a scan.
 func (e *Executor) findConflictRow(table string, schema *storage.Schema, row storage.Row, columns []string) (storage.Row, bool, error) {
-	if len(columns) == 1 && strings.EqualFold(columns[0], schema.PrimaryKey) {
-		pkValue, ok := lookupRowValue(row, schema.PrimaryKey)
-		if !ok || pkValue == nil {
+	if schema.PrimaryKeyTargetMatches(columns) {
+		pk, err := schema.PrimaryKeyForRow(row)
+		if err != nil {
+			// A missing or NULL key component cannot conflict: the insert itself
+			// fails with the NOT NULL violation.
 			return nil, false, nil
 		}
-		found, err := e.session.GetByPK(table, fmt.Sprintf("%v", pkValue))
+		found, err := e.session.GetByPK(table, pk)
 		if err == storage.ErrKeyNotFound {
 			return nil, false, nil
 		}
@@ -2699,7 +2701,10 @@ func (e *Executor) applyUpsert(table string, schema *storage.Schema, existing, r
 		context["excluded."+k] = v
 	}
 
-	pk := fmt.Sprintf("%v", existing[schema.PrimaryKey])
+	pk, err := schema.PrimaryKeyForRow(existing)
+	if err != nil {
+		return err
+	}
 	_, updated, err := e.session.UpdateByPK(table, pk, func(current storage.Row) (storage.Row, error) {
 		updates := make(storage.Row)
 		merged := make(storage.Row, len(current))
@@ -2750,9 +2755,24 @@ func (e *Executor) executeInsert(stmt *parser.InsertStmt) (*Result, error) {
 			}
 		}
 	}
-	// Expression defaults are evaluated once per statement, so all rows of a
-	// multi-row INSERT share one value and the next statement reads the clock
-	// again. Constant defaults are applied by storage from the catalog.
+	// An upsert is applied as an in-place update of the conflicting row, so it
+	// cannot move that row to a new primary key. Accepting the assignment would
+	// leave the durable row under its old key with the new key values in its
+	// body, which makes later full-key reads and updates split the row in two.
+	for _, assignment := range stmt.ConflictUpdate {
+		if schema.IsPrimaryKeyColumn(assignment.Column) {
+			return nil, pgerror.Unsupported(
+				"ON CONFLICT DO UPDATE cannot assign primary key column %q of table %s; delete and re-insert the row to change its key",
+				assignment.Column, tableName,
+			)
+		}
+	}
+	// Defaults are resolved once per statement: expression defaults are
+	// evaluated at the statement instant so a multi-row INSERT shares one value,
+	// and constant catalog defaults are copied into the candidate row. Resolving
+	// both here lets conflict matching and RETURNING see the same key the storage
+	// layer will write; storage still applies them as a fallback for rows that
+	// reach it another way.
 	defaults, err := e.statementDefaults(schema)
 	if err != nil {
 		return nil, err
@@ -2918,8 +2938,8 @@ func (e *Executor) insertOneRow(tableName string, schema *storage.Schema, row st
 	}
 	if stmt.ConflictDoNothing || len(stmt.ConflictUpdate) > 0 {
 		columns := stmt.ConflictTarget
-		if len(columns) == 0 && schema.PrimaryKey != "" && schema.PrimaryKey != "_rowid_" {
-			columns = []string{schema.PrimaryKey}
+		if len(columns) == 0 && !schema.UsesSyntheticRowID() {
+			columns = schema.PrimaryKeyList()
 		}
 		if len(columns) > 0 {
 			existing, found, err := e.findConflictRow(tableName, schema, row, columns)
@@ -2936,13 +2956,13 @@ func (e *Executor) insertOneRow(tableName string, schema *storage.Schema, row st
 						return 0, false, false, nil, fmt.Errorf("cannot UPDATE generated column %s", assignment.Column)
 					}
 				}
-				existingPK, _ := lookupRowValue(existing, schema.PrimaryKey)
+				existingPK, _ := schema.PrimaryKeyForRow(existing)
 				if err := e.applyUpsert(tableName, schema, existing, row, stmt.ConflictUpdate); err != nil {
 					return 0, false, false, nil, err
 				}
 				var stored storage.Row
 				if wantStored {
-					stored = e.fetchByPK(tableName, fmt.Sprintf("%v", existingPK))
+					stored = e.fetchByPK(tableName, existingPK)
 				}
 				return 0, false, true, stored, nil
 			}
@@ -2966,9 +2986,9 @@ func (e *Executor) insertOneRow(tableName string, schema *storage.Schema, row st
 			case parser.ConflictIgnore:
 				return 0, false, false, nil, nil
 			case parser.ConflictReplace:
-				pkValue := row[schema.PrimaryKey]
-				if pkValue != nil {
-					if _, deleted, delErr := e.session.DeleteByPK(tableName, fmt.Sprintf("%v", pkValue)); delErr == nil && deleted {
+				pk, pkErr := schema.PrimaryKeyForRow(row)
+				if pkErr == nil {
+					if _, deleted, delErr := e.session.DeleteByPK(tableName, pk); delErr == nil && deleted {
 						replacedID, replaceErr := e.session.InsertWithRowID(tableName, row)
 						if replaceErr != nil {
 							return 0, false, false, nil, replaceErr
@@ -2992,11 +3012,11 @@ func (e *Executor) fetchIfRequested(tableName string, schema *storage.Schema, ro
 	if !want {
 		return nil
 	}
-	pkValue, ok := lookupRowValue(row, schema.PrimaryKey)
-	if !ok || pkValue == nil {
+	pk, err := schema.PrimaryKeyForRow(row)
+	if err != nil {
 		return row
 	}
-	return e.fetchByPK(tableName, fmt.Sprintf("%v", pkValue))
+	return e.fetchByPK(tableName, pk)
 }
 
 // fetchByPK reads a stored row, falling back to nil on a read error; RETURNING
@@ -3120,11 +3140,11 @@ func (e *Executor) isPointMutation(table string, where parser.Expr, assignments
 		return false
 	}
 	schema, err := e.schema.GetSchema(table)
-	if err != nil || !strings.EqualFold(column, schema.PrimaryKey) {
+	if err != nil || !schema.IsSingleColumnPrimaryKey() || !strings.EqualFold(column, schema.PrimaryKey) {
 		return false
 	}
 	for _, assignment := range assignments {
-		if strings.EqualFold(assignment.Column, schema.PrimaryKey) {
+		if schema.IsPrimaryKeyColumn(assignment.Column) {
 			return false
 		}
 	}
@@ -3293,12 +3313,12 @@ func (e *Executor) executeUpdate(stmt *parser.UpdateStmt) (*Result, error) {
 		column, value, equality := e.extractIndexableCondition(stmt.Where)
 		updatesPrimaryKey := false
 		for _, assignment := range stmt.Set {
-			if strings.EqualFold(assignment.Column, schema.PrimaryKey) {
+			if schema.IsPrimaryKeyColumn(assignment.Column) {
 				updatesPrimaryKey = true
 				break
 			}
 		}
-		if equality && safePrimaryKeyLiteral(value) && strings.EqualFold(column, schema.PrimaryKey) && !updatesPrimaryKey {
+		if equality && safePrimaryKeyLiteral(value) && schema.IsSingleColumnPrimaryKey() && strings.EqualFold(column, schema.PrimaryKey) && !updatesPrimaryKey {
 			pk := fmt.Sprintf("%v", value)
 			_, updated, err := e.session.UpdateByPK(tableName, pk, updateFn)
 			if err != nil {
@@ -3330,11 +3350,11 @@ func (e *Executor) executeUpdate(stmt *parser.UpdateStmt) (*Result, error) {
 			if merr != nil {
 				return nil, merr
 			}
-			key, ok := lookupRowValue(merged, schema.PrimaryKey)
-			if !ok || key == nil {
-				key, _ = lookupRowValue(r, schema.PrimaryKey)
+			key, keyErr := schema.PrimaryKeyForRow(merged)
+			if keyErr != nil {
+				key, _ = schema.PrimaryKeyForRow(r)
 			}
-			returningPKs = append(returningPKs, fmt.Sprintf("%v", key))
+			returningPKs = append(returningPKs, key)
 		}
 	}
 
@@ -3456,8 +3476,12 @@ func (e *Executor) executeUpdateFrom(stmt *parser.UpdateStmt, schema *storage.Sc
 		if err := e.applyGeneratedColumns(schema, row); err != nil {
 			return nil, err
 		}
+		oldPK, err := schema.PrimaryKeyForRow(target)
+		if err != nil {
+			return nil, err
+		}
 		plans = append(plans, updateFromPlan{
-			oldPK: fmt.Sprintf("%v", target[schema.PrimaryKey]),
+			oldPK: oldPK,
 			row:   row,
 		})
 	}
@@ -3466,8 +3490,10 @@ func (e *Executor) executeUpdateFrom(stmt *parser.UpdateStmt, schema *storage.Sc
 	var affected []storage.Row
 	err = e.runDMLAtomicForce(tableName, true, func() error {
 		for _, plan := range plans {
-			newPKVal, _ := lookupRowValue(plan.row, schema.PrimaryKey)
-			newPK := fmt.Sprintf("%v", newPKVal)
+			newPK, newPKErr := schema.PrimaryKeyForRow(plan.row)
+			if newPKErr != nil {
+				return newPKErr
+			}
 			if newPK != plan.oldPK {
 				if _, gerr := e.session.GetByPK(tableName, newPK); gerr == nil {
 					return &storage.PrimaryKeyViolationError{Table: tableName, Key: newPK}
@@ -3648,7 +3674,7 @@ func (e *Executor) executeDelete(stmt *parser.DeleteStmt) (*Result, error) {
 
 	if stmt.Where != nil {
 		column, value, equality := e.extractIndexableCondition(stmt.Where)
-		if equality && safePrimaryKeyLiteral(value) && strings.EqualFold(column, schema.PrimaryKey) {
+		if equality && safePrimaryKeyLiteral(value) && schema.IsSingleColumnPrimaryKey() && strings.EqualFold(column, schema.PrimaryKey) {
 			deletedRow, deleted, err := e.session.DeleteByPK(tableName, fmt.Sprintf("%v", value))
 			if err != nil {
 				return nil, err
@@ -3703,6 +3729,10 @@ func (e *Executor) executeCreateTable(stmt *parser.CreateTableStmt) (*Result, er
 		Name: stmt.Table.Name,
 	}
 
+	// A column-level PRIMARY KEY is recorded separately so more than one, or a
+	// mix with a table-level PRIMARY KEY, is rejected instead of silently
+	// choosing one of them.
+	var inlinePrimaryKeyColumns []string
 	for _, colDef := range stmt.Columns {
 		col := storage.Column{
 			Name:     colDef.Name,
@@ -3725,9 +3755,15 @@ func (e *Executor) executeCreateTable(stmt *parser.CreateTableStmt) (*Result, er
 		for _, constraint := range colDef.Constraints {
 			switch constraint.Type {
 			case parser.ConstraintPrimaryKey:
+				if col.GeneratedExpr != "" {
+					return nil, pgerror.Unsupported(
+						"generated column %q cannot be part of a PRIMARY KEY",
+						col.Name,
+					)
+				}
 				col.PrimaryKey = true
 				col.Nullable = false
-				schema.PrimaryKey = col.Name
+				inlinePrimaryKeyColumns = append(inlinePrimaryKeyColumns, col.Name)
 			case parser.ConstraintNotNull:
 				col.Nullable = false
 			case parser.ConstraintDefault:
@@ -3758,20 +3794,48 @@ func (e *Executor) executeCreateTable(stmt *parser.CreateTableStmt) (*Result, er
 	// Handle table-level constraints. CHECK predicates are persisted and
 	// enforced; FOREIGN KEY is still rejected because the engine never validates
 	// referential integrity.
+	var tablePrimaryKeyColumns []string
 	for _, constraint := range stmt.Constraints {
 		switch constraint.Type {
 		case parser.ConstraintPrimaryKey:
-			if len(constraint.Columns) > 0 {
-				if len(constraint.Columns) > 1 {
-					return nil, fmt.Errorf("%w: composite primary keys are not supported; declare a single-column PRIMARY KEY", ErrUnsupportedFeature)
-				}
-				schema.PrimaryKey = constraint.Columns[0]
-				for i := range schema.Columns {
-					if strings.EqualFold(schema.Columns[i].Name, schema.PrimaryKey) {
-						schema.Columns[i].PrimaryKey = true
-						schema.Columns[i].Nullable = false
-					}
-				}
+			if len(constraint.Columns) == 0 {
+				break
+			}
+			if len(tablePrimaryKeyColumns) > 0 {
+				return nil, pgerror.Unsupported(
+					"table %s declares more than one PRIMARY KEY; declare exactly one column list",
+					stmt.Table.Name,
+				)
+			}
+			if len(inlinePrimaryKeyColumns) > 0 {
+				return nil, pgerror.Unsupported(
+					"table %s mixes an inline PRIMARY KEY with a table-level PRIMARY KEY; use one form, and list every key column in the table-level PRIMARY KEY (a, b) for a composite key",
+					stmt.Table.Name,
+				)
+			}
+			seen := make(map[string]bool, len(constraint.Columns))
+			for _, name := range constraint.Columns {
+				column, ok := schema.GetColumn(name)
+				if !ok {
+					return nil, pgerror.Unsupported(
+						"PRIMARY KEY column %q does not exist in table %s",
+						name, stmt.Table.Name,
+					)
+				}
+				if column.GeneratedExpr != "" {
+					return nil, pgerror.Unsupported(
+						"generated column %q cannot be part of a PRIMARY KEY",
+						column.Name,
+					)
+				}
+				if seen[strings.ToLower(column.Name)] {
+					return nil, pgerror.Unsupported(
+						"column %q appears more than once in the PRIMARY KEY of table %s",
+						column.Name, stmt.Table.Name,
+					)
+				}
+				seen[strings.ToLower(column.Name)] = true
+				tablePrimaryKeyColumns = append(tablePrimaryKeyColumns, column.Name)
 			}
 		case parser.ConstraintCheck:
 			// Handled with the column-level predicates once every column is known.
@@ -3782,6 +3846,22 @@ func (e *Executor) executeCreateTable(stmt *parser.CreateTableStmt) (*Result, er
 		}
 	}
 
+	if len(inlinePrimaryKeyColumns) > 1 {
+		return nil, pgerror.Unsupported(
+			"table %s declares more than one inline PRIMARY KEY; use a table-level PRIMARY KEY (%s) instead",
+			stmt.Table.Name, strings.Join(inlinePrimaryKeyColumns, ", "),
+		)
+	}
+	if len(tablePrimaryKeyColumns) > 0 {
+		schema.PrimaryKeyColumns = tablePrimaryKeyColumns
+		for i := range schema.Columns {
+			if schema.IsPrimaryKeyColumn(schema.Columns[i].Name) {
+				schema.Columns[i].PrimaryKey = true
+				schema.Columns[i].Nullable = false
+			}
+		}
+	}
+
 	// Translate inline/table UNIQUE constraints (Gogs/GORM style) into the same
 	// scan-validated unique Index metadata that explicit CREATE UNIQUE INDEX
 	// produces, so both enforcement and the catalog observe them.
@@ -4179,8 +4259,10 @@ func (e *Executor) executeAlterTableAddColumn(table string, action *parser.AddCo
 	for _, constraint := range action.Column.Constraints {
 		switch constraint.Type {
 		case parser.ConstraintPrimaryKey:
-			col.PrimaryKey = true
-			col.Nullable = false
+			return nil, pgerror.Unsupported(
+				"ALTER TABLE ADD COLUMN cannot declare PRIMARY KEY on column %q; declare the key in CREATE TABLE (SQLite also rejects adding a PRIMARY KEY column)",
+				col.Name,
+			)
 		case parser.ConstraintNotNull:
 			col.Nullable = false
 		case parser.ConstraintUnique:
@@ -4755,11 +4837,7 @@ func (e *Executor) pragmaTableInfo(tableName string) (*Result, error) {
 		if !col.Nullable {
 			notnull = 1
 		}
-		pk := 0
-		if col.PrimaryKey {
-			pk = 1
-		}
-		result.AddRow(int64(i), col.Name, col.Type, int64(notnull), col.DefaultLiteral(), int64(pk))
+		result.AddRow(int64(i), col.Name, col.Type, int64(notnull), col.DefaultLiteral(), int64(primaryKeyOrdinal(schema, col.Name)))
 	}
 
 	return result, nil

pkg/executor/generated.go

@@ -70,7 +70,9 @@ func generatedColumnSet(schema *storage.Schema) map[string]bool {
 // integer primary key are left to the storage layer, which assigns the hidden
 // _rowid_ during the insert.
 func (e *Executor) ensureGeneratedRowID(tableName string, schema *storage.Schema, row storage.Row) error {
-	if schema.PrimaryKey == "" || schema.PrimaryKey == "_rowid_" {
+	// Only a single-column INTEGER PRIMARY KEY is generated before evaluation; a
+	// composite key is never a rowid alias.
+	if !schema.IsSingleColumnPrimaryKey() || schema.UsesSyntheticRowID() {
 		return nil
 	}
 	if v, ok := lookupRowValue(row, schema.PrimaryKey); ok && v != nil {
@@ -109,7 +111,6 @@ func isIntegerColumnType(typeName string) bool {
 // conflictMatcher describes one uniqueness constraint used to resolve an
 // INSERT conflict. index is nil for the primary key.
 type conflictMatcher struct {
-	name       string
 	index      *storage.Index
 	primaryKey bool
 }
@@ -121,7 +122,7 @@ type conflictMatcher struct {
 func (e *Executor) insertConflictMatchers(tableName string, schema *storage.Schema, stmtReplace bool) ([]conflictMatcher, error) {
 	var matchers []conflictMatcher
 	if stmtReplace {
-		matchers = append(matchers, conflictMatcher{name: schema.PrimaryKey, primaryKey: true})
+		matchers = append(matchers, conflictMatcher{primaryKey: true})
 	}
 	indexes, err := e.schema.ListTableIndexes(tableName)
 	if err != nil {
@@ -134,7 +135,7 @@ func (e *Executor) insertConflictMatchers(tableName string, schema *storage.Sche
 		if !stmtReplace && !strings.EqualFold(idx.OnConflict, "REPLACE") {
 			continue
 		}
-		matchers = append(matchers, conflictMatcher{name: idx.Name, index: idx})
+		matchers = append(matchers, conflictMatcher{index: idx})
 	}
 	return matchers, nil
 }
@@ -144,11 +145,13 @@ func (e *Executor) insertConflictMatchers(tableName string, schema *storage.Sche
 // index-key lookup rather than a full table scan.
 func (e *Executor) matcherConflicts(tableName string, schema *storage.Schema, m conflictMatcher, candidate storage.Row) ([]storage.Row, error) {
 	if m.primaryKey {
-		pkValue, ok := lookupRowValue(candidate, schema.PrimaryKey)
-		if !ok || pkValue == nil {
+		pk, err := schema.PrimaryKeyForRow(candidate)
+		if err != nil {
+			// A missing or NULL key component cannot conflict; the insert fails
+			// its NOT NULL constraint instead.
 			return nil, nil
 		}
-		row, err := e.session.GetByPK(tableName, fmt.Sprintf("%v", pkValue))
+		row, err := e.session.GetByPK(tableName, pk)
 		if err == storage.ErrKeyNotFound {
 			return nil, nil
 		}
@@ -176,14 +179,14 @@ func (e *Executor) matcherConflicts(tableName string, schema *storage.Schema, m
 // statement-level OR IGNORE and targetless DO NOTHING apply to any conflict,
 // not just the primary key.
 func (e *Executor) hasAnyInsertConflict(tableName string, schema *storage.Schema, candidate storage.Row) (bool, error) {
-	matchers := []conflictMatcher{{name: schema.PrimaryKey, primaryKey: true}}
+	matchers := []conflictMatcher{{primaryKey: true}}
 	indexes, err := e.schema.ListTableIndexes(tableName)
 	if err != nil {
 		return false, err
 	}
 	for _, index := range indexes {
 		if index.Unique {
-			matchers = append(matchers, conflictMatcher{name: index.Name, index: index})
+			matchers = append(matchers, conflictMatcher{index: index})
 		}
 	}
 	for _, matcher := range matchers {
@@ -219,15 +222,19 @@ func (e *Executor) resolveInsertConflicts(tableName string, schema *storage.Sche
 			return err
 		}
 		for _, row := range rows {
-			toDelete[fmt.Sprintf("%v", row[schema.PrimaryKey])] = row
+			pk, err := schema.PrimaryKeyForRow(row)
+			if err != nil {
+				return err
+			}
+			toDelete[pk] = row
 		}
 	}
 	if len(toDelete) == 0 {
 		return nil
 	}
 
-	for _, row := range toDelete {
-		if _, deleted, derr := e.session.DeleteByPK(tableName, fmt.Sprintf("%v", row[schema.PrimaryKey])); derr != nil {
+	for pk := range toDelete {
+		if _, deleted, derr := e.session.DeleteByPK(tableName, pk); derr != nil {
 			return derr
 		} else if !deleted {
 			return fmt.Errorf("ON CONFLICT REPLACE row disappeared during delete")

pkg/executor/sqlite_catalog.go

@@ -290,10 +290,7 @@ func (e *Executor) pragmaTableXInfo(table string) (*Result, error) {
 		if !col.Nullable {
 			notnull = 1
 		}
-		pk := int64(0)
-		if col.PrimaryKey {
-			pk = 1
-		}
+		pk := int64(primaryKeyOrdinal(schema, col.Name))
 		// Hidden flag: 0 normal, 2 generated VIRTUAL, 3 generated STORED.
 		hidden := int64(0)
 		if col.GeneratedExpr != "" {
@@ -312,14 +309,25 @@ 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)+len(s.Checks))
+	compositeKey := s.IsCompositePrimaryKey()
+	cols := make([]string, 0, len(s.Columns)+len(s.Checks)+1)
 	for _, col := range s.Columns {
 		// Strip only the engine-injected hidden rowid column, never user
 		// columns that happen to be named oid/rowid/_rowid_.
-		if s.PrimaryKey == "_rowid_" && col.Name == "_rowid_" {
+		if s.UsesSyntheticRowID() && col.Name == "_rowid_" {
 			continue
 		}
-		cols = append(cols, recreateColumnDef(s, col))
+		cols = append(cols, recreateColumnDef(s, col, compositeKey))
+	}
+	// A composite key is rendered as one table-level constraint, as it was
+	// declared; a single-column key keeps its inline PRIMARY KEY form.
+	if compositeKey {
+		columns := s.PrimaryKeyList()
+		quoted := make([]string, len(columns))
+		for i, name := range columns {
+			quoted[i] = quoteIdent(name)
+		}
+		cols = append(cols, "PRIMARY KEY ("+strings.Join(quoted, ", ")+")")
 	}
 	// Inline and table-level CHECK declarations are stored uniformly, so they
 	// are all rendered as table-level constraints: `CREATE TABLE t (a INTEGER,
@@ -334,15 +342,16 @@ func recreateCreateTableSQL(s *storage.Schema) string {
 	return fmt.Sprintf("CREATE TABLE %s (%s)", quoteIdent(s.Name), strings.Join(cols, ", "))
 }
 
-func recreateColumnDef(s *storage.Schema, col storage.Column) string {
+func recreateColumnDef(s *storage.Schema, col storage.Column, compositeKey bool) string {
 	var b strings.Builder
 	b.WriteString(quoteIdent(col.Name))
 	b.WriteString(" ")
 	b.WriteString(col.Type)
-	if col.PrimaryKey {
+	inlineKey := col.PrimaryKey && !compositeKey
+	if inlineKey {
 		b.WriteString(" PRIMARY KEY")
 	}
-	if col.PrimaryKey && s.AutoIncrement {
+	if inlineKey && s.AutoIncrement {
 		b.WriteString(" AUTOINCREMENT")
 	}
 	if !col.Nullable {
@@ -370,6 +379,18 @@ func recreateColumnDef(s *storage.Schema, col storage.Column) string {
 	return b.String()
 }
 
+// primaryKeyOrdinal returns the column's 1-based position in the primary key,
+// or 0 when it is not part of it. PRAGMA table_info/table_xinfo report the
+// ordinal, so a composite key is distinguishable from several single keys.
+func primaryKeyOrdinal(s *storage.Schema, columnName string) int {
+	for i, name := range s.PrimaryKeyList() {
+		if strings.EqualFold(name, columnName) {
+			return i + 1
+		}
+	}
+	return 0
+}
+
 func recreateCreateIndexSQL(idx *storage.Index) string {
 	unique := ""
 	if idx.Unique {

pkg/pgserver/compat_suite_test.go

@@ -122,7 +122,11 @@ func TestCompatibilitySuite(t *testing.T) {
 	expectRows("upsert.result", `SELECT plan FROM accounts WHERE id = 2`, "pro")
 
 	expectCode("constraints.unique_23505", `INSERT INTO accounts (id, email) VALUES (3, 'a@x')`, "23505")
-	expectCode("ddl.composite_pk_0A000", `CREATE TABLE bad_pk (a INTEGER, b INTEGER, PRIMARY KEY (a, b))`, "0A000")
+	expectOK("ddl.composite_pk", `CREATE TABLE composite_pk (a INTEGER, b TEXT, PRIMARY KEY (a, b))`)
+	expectOK("dml.composite_pk_insert", `INSERT INTO composite_pk (a, b) VALUES (1, 'x')`)
+	expectRows("dml.composite_pk_select", `SELECT count(*) FROM composite_pk WHERE a = 1 AND b = 'x'`, "1")
+	expectCode("constraints.composite_pk_23505", `INSERT INTO composite_pk (a, b) VALUES (1, 'x')`, "23505")
+	expectCode("constraints.composite_pk_23502", `INSERT INTO composite_pk (a) VALUES (2)`, "23502")
 	expectOK("ddl.timestamp_default", `CREATE TABLE stamped (id INTEGER PRIMARY KEY, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP)`)
 	expectOK("ddl.timestamp_default_insert", `INSERT INTO stamped (id) VALUES (1)`)
 	expectRows("ddl.timestamp_default_applied", `SELECT count(*) FROM stamped WHERE created_at IS NOT NULL`, "1")

pkg/pgserver/composite_primary_key_test.go

@@ -0,0 +1,132 @@
+package pgserver
+
+import (
+	"strings"
+	"testing"
+)
+
+// TestCompositePrimaryKeyWireBehavior drives a composite key through the
+// simple-query protocol: DDL, duplicate and NOT NULL SQLSTATEs, full-key reads,
+// PRAGMA table_info ordinals, and the information_schema key columns.
+func TestCompositePrimaryKeyWireBehavior(t *testing.T) {
+	c, client := newDataConnection(t)
+	query := func(sql string) []*Message {
+		return runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte(sql), 0)})
+	}
+
+	if msgs := query("CREATE TABLE pairs (a TEXT, b TEXT, v TEXT, PRIMARY KEY (a, b))"); errorCode(msgs[0]) != "" {
+		t.Fatalf("CREATE TABLE failed: %s", errorMessage(msgs[0]))
+	}
+	if msgs := query("INSERT INTO pairs (a, b, v) VALUES ('x', 'y', 'one')"); errorCode(msgs[0]) != "" {
+		t.Fatalf("INSERT failed: %s", errorMessage(msgs[0]))
+	}
+	if code := errorCode(query("INSERT INTO pairs (a, b, v) VALUES ('x', 'y', 'two')")[0]); code != ErrCodeUniqueViolation {
+		t.Fatalf("duplicate composite key SQLSTATE = %q, want 23505", code)
+	}
+	if code := errorCode(query("INSERT INTO pairs (a, v) VALUES ('x', 'missing b')")[0]); code != "23502" {
+		t.Fatalf("missing key component SQLSTATE = %q, want 23502", code)
+	}
+
+	// Only the accepted row survived the two rejected inserts.
+	selectRows := query("SELECT v FROM pairs WHERE a = 'x' AND b = 'y'")
+	var dataRows []*Message
+	for _, msg := range selectRows {
+		if msg.Type == MsgDataRow {
+			dataRows = append(dataRows, msg)
+		}
+	}
+	if len(dataRows) != 1 {
+		t.Fatalf("full-key select returned %d rows, want 1", len(dataRows))
+	}
+	if values := dataRowValues(t, dataRows[0]); len(values) != 1 || values[0] != "one" {
+		t.Fatalf("full-key select values = %v, want [one]", values)
+	}
+
+	// PRAGMA table_info reports the pk ordinal, not a bare flag.
+	var info [][]string
+	for _, msg := range query("PRAGMA table_info(pairs)") {
+		if msg.Type == MsgDataRow {
+			info = append(info, dataRowValues(t, msg))
+		}
+	}
+	if len(info) != 3 {
+		t.Fatalf("table_info rows = %d, want 3", len(info))
+	}
+	wantPK := []string{"1", "2", "0"}
+	for i, row := range info {
+		if row[5] != wantPK[i] {
+			t.Fatalf("table_info row %d pk = %s, want %s", i, row[5], wantPK[i])
+		}
+	}
+
+	// KEY_COLUMN_USAGE lists one row per key column with its ordinal.
+	var usage [][]string
+	for _, msg := range query("SELECT column_name, ordinal_position FROM INFORMATION_SCHEMA.KEY_COLUMN_USAGE WHERE table_name = 'pairs'") {
+		if msg.Type == MsgDataRow {
+			usage = append(usage, dataRowValues(t, msg))
+		}
+	}
+	if len(usage) != 2 {
+		t.Fatalf("KEY_COLUMN_USAGE rows = %d, want 2", len(usage))
+	}
+	if usage[0][0] != "a" || usage[0][1] != "1" || usage[1][0] != "b" || usage[1][1] != "2" {
+		t.Fatalf("KEY_COLUMN_USAGE = %v, want [[a 1] [b 2]]", usage)
+	}
+}
+
+func TestCompositePrimaryKeyDDLRejections(t *testing.T) {
+	c, client := newDataConnection(t)
+	for _, sql := range []string{
+		"CREATE TABLE t1 (a TEXT PRIMARY KEY, b TEXT PRIMARY KEY)",
+		"CREATE TABLE t2 (a TEXT PRIMARY KEY, b TEXT, PRIMARY KEY (b))",
+		"CREATE TABLE t3 (a TEXT, PRIMARY KEY (a, a))",
+		"CREATE TABLE t4 (a TEXT, b TEXT GENERATED ALWAYS AS (a) STORED, PRIMARY KEY (b))",
+		"CREATE TABLE t5 (a TEXT)",
+		"ALTER TABLE t5 ADD COLUMN id INTEGER PRIMARY KEY",
+	} {
+		msgs := runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte(sql), 0)})
+		if strings.HasPrefix(sql, "CREATE TABLE t5") {
+			if code := errorCode(msgs[0]); code != "" {
+				t.Fatalf("%s SQLSTATE = %q, want success", sql, code)
+			}
+			continue
+		}
+		if code := errorCode(msgs[0]); code != ErrCodeFeatureNotSupported {
+			t.Fatalf("%s SQLSTATE = %q, want 0A000", sql, code)
+		}
+	}
+}
+
+// A key column that takes its value from a constant DEFAULT must resolve before
+// conflict matching, including over the wire.
+func TestCompositePrimaryKeyDefaultedKeyUpsertOverWire(t *testing.T) {
+	c, client := newDataConnection(t)
+	query := func(sql string) []*Message {
+		return runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte(sql), 0)})
+	}
+
+	query("CREATE TABLE konst (a TEXT DEFAULT 'd', b TEXT DEFAULT 'e', v TEXT, PRIMARY KEY (a, b))")
+	query("INSERT INTO konst (b, v) VALUES ('y', 'one')")
+	if code := errorCode(query("INSERT INTO konst (b, v) VALUES ('y', 'two') ON CONFLICT (a, b) DO UPDATE SET v = excluded.v")[0]); code != "" {
+		t.Fatalf("defaulted-key upsert SQLSTATE = %q, want success", code)
+	}
+
+	rows := query("SELECT v FROM konst WHERE a = 'd' AND b = 'y'")
+	var dataRows []*Message
+	for _, msg := range rows {
+		if msg.Type == MsgDataRow {
+			dataRows = append(dataRows, msg)
+		}
+	}
+	if len(dataRows) != 1 {
+		t.Fatalf("upsert read returned %d rows, want 1", len(dataRows))
+	}
+	if values := dataRowValues(t, dataRows[0]); len(values) != 1 || values[0] != "two" {
+		t.Fatalf("upsert result = %v, want [two]", values)
+	}
+
+	// Assigning a key column is rejected, not applied as a ghost row.
+	if code := errorCode(query("INSERT INTO konst (b, v) VALUES ('y', 'x') ON CONFLICT (a, b) DO UPDATE SET a = 'other'")[0]); code != ErrCodeFeatureNotSupported {
+		t.Fatalf("key-assignment upsert SQLSTATE = %q, want 0A000", code)
+	}
+}

pkg/pgserver/connection.go

@@ -773,22 +773,26 @@ func (c *Connection) catalogResult(sql string) (*executor.Result, bool, error) {
 	case "INFORMATION_SCHEMA.TABLE_CONSTRAINTS", "INFORMATION_SCHEMA.KEY_COLUMN_USAGE":
 		for _, table := range tables {
 			schema, schemaErr := c.schema.GetSchema(table)
-			if schemaErr != nil || schema.PrimaryKey == "" || schema.PrimaryKey == "_rowid_" {
+			if schemaErr != nil || schema.UsesSyntheticRowID() {
 				continue
 			}
-			row := map[string]interface{}{
-				"constraint_catalog": c.database, "constraint_schema": "public", "constraint_name": table + "_pkey",
-				"table_catalog": c.database, "table_schema": "public", "table_name": table,
-			}
+			columns := schema.PrimaryKeyList()
 			if source == "INFORMATION_SCHEMA.TABLE_CONSTRAINTS" {
-				row["constraint_type"] = "PRIMARY KEY"
-				row["is_deferrable"] = "NO"
-				row["initially_deferred"] = "NO"
-			} else {
-				row["column_name"] = schema.PrimaryKey
-				row["ordinal_position"] = int64(1)
+				rows = append(rows, map[string]interface{}{
+					"constraint_catalog": c.database, "constraint_schema": "public", "constraint_name": table + "_pkey",
+					"table_catalog": c.database, "table_schema": "public", "table_name": table,
+					"constraint_type": "PRIMARY KEY", "is_deferrable": "NO", "initially_deferred": "NO",
+				})
+				continue
+			}
+			// KEY_COLUMN_USAGE reports one row per key column, in key order.
+			for i, column := range columns {
+				rows = append(rows, map[string]interface{}{
+					"constraint_catalog": c.database, "constraint_schema": "public", "constraint_name": table + "_pkey",
+					"table_catalog": c.database, "table_schema": "public", "table_name": table,
+					"column_name": column, "ordinal_position": int64(i + 1),
+				})
 			}
-			rows = append(rows, row)
 		}
 	case "PG_INDEXES":
 		for _, table := range tables {

pkg/pgserver/features_test.go

@@ -274,18 +274,3 @@ func TestExtendedQueryUniqueViolationErrorCode(t *testing.T) {
 		t.Fatalf("error message = %q, want constraint uniq_t_email", message)
 	}
 }
-
-func TestCompositePrimaryKeyDDLErrorCode(t *testing.T) {
-	c, client := newDataConnection(t)
-
-	msgs := runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte("CREATE TABLE t (a TEXT, b TEXT, PRIMARY KEY (a,b))"), 0)})
-	if len(msgs) != 2 {
-		t.Fatalf("got %d messages, want ErrorResponse+ReadyForQuery", len(msgs))
-	}
-	if code := errorCode(msgs[0]); code != ErrCodeFeatureNotSupported {
-		t.Fatalf("error code = %q, want %q", code, ErrCodeFeatureNotSupported)
-	}
-	if message := errorMessage(msgs[0]); !strings.Contains(message, "composite primary key") {
-		t.Fatalf("error message = %q, want a composite primary key error", message)
-	}
-}

pkg/sqlexport/export.go

@@ -106,57 +106,52 @@ func ExportTable(schema *storage.SchemaManager, table *storage.TableManager, tab
 
 // generateCreateTable generates a CREATE TABLE statement from schema.
 func generateCreateTable(schema *storage.Schema) string {
-	var sb strings.Builder
-
-	sb.WriteString(fmt.Sprintf("CREATE TABLE %s (\n", quoteIdentifier(schema.Name)))
-
-	// Generate column definitions
-	for i, col := range schema.Columns {
+	compositeKey := schema.IsCompositePrimaryKey()
+	defs := make([]string, 0, len(schema.Columns)+1)
+	for _, col := range schema.Columns {
 		// Skip internal _rowid_ column
 		if col.Name == "_rowid_" {
 			continue
 		}
 
-		sb.WriteString("    ")
-		sb.WriteString(quoteIdentifier(col.Name))
-		sb.WriteString(" ")
-		sb.WriteString(col.Type)
+		var b strings.Builder
+		b.WriteString(quoteIdentifier(col.Name))
+		b.WriteString(" ")
+		b.WriteString(col.Type)
 
-		// Add PRIMARY KEY constraint
-		if col.PrimaryKey {
-			sb.WriteString(" PRIMARY KEY")
+		// A single-column key keeps its inline PRIMARY KEY; a composite key is
+		// emitted once at the end so no column claims the key by itself.
+		inlineKey := col.PrimaryKey && !compositeKey
+		if inlineKey {
+			b.WriteString(" PRIMARY KEY")
 		}
 
-		// Add NOT NULL constraint
-		if !col.Nullable && !col.PrimaryKey {
-			sb.WriteString(" NOT NULL")
+		if !col.Nullable && !inlineKey {
+			b.WriteString(" NOT NULL")
 		}
 
 		// Add DEFAULT value. A non-constant default is emitted as its stored SQL
 		// text so the dump keeps evaluating it instead of freezing a value.
 		if col.DefaultExpr != "" {
-			sb.WriteString(" DEFAULT ")
-			sb.WriteString(col.DefaultExpr)
+			b.WriteString(" DEFAULT ")
+			b.WriteString(col.DefaultExpr)
 		} else if col.Default != nil {
-			sb.WriteString(" DEFAULT ")
-			sb.WriteString(formatValue(col.Default, col.Type))
+			b.WriteString(" DEFAULT ")
+			b.WriteString(formatValue(col.Default, col.Type))
 		}
 
-		// Add comma if not last column
-		if i < len(schema.Columns)-1 {
-			// Check if next column is _rowid_
-			if i+1 < len(schema.Columns) && schema.Columns[i+1].Name != "_rowid_" {
-				sb.WriteString(",")
-			} else if i+2 < len(schema.Columns) {
-				sb.WriteString(",")
-			}
+		defs = append(defs, "    "+b.String())
+	}
+	if compositeKey {
+		columns := schema.PrimaryKeyList()
+		quoted := make([]string, len(columns))
+		for i, name := range columns {
+			quoted[i] = quoteIdentifier(name)
 		}
-		sb.WriteString("\n")
+		defs = append(defs, "    PRIMARY KEY ("+strings.Join(quoted, ", ")+")")
 	}
 
-	sb.WriteString(");")
-
-	return sb.String()
+	return fmt.Sprintf("CREATE TABLE %s (\n%s\n);", quoteIdentifier(schema.Name), strings.Join(defs, ",\n"))
 }
 
 // generateInserts generates INSERT statements for rows.

pkg/sqlexport/export_test.go

@@ -45,3 +45,37 @@ func TestExportPreservesDefaultExpression(t *testing.T) {
 		t.Fatalf("dump lost the default expression:\n%s", dump)
 	}
 }
+
+// A composite key must be dumped as one table-level constraint; emitting an
+// inline PRIMARY KEY on each key column would describe two keys.
+func TestExportCompositePrimaryKey(t *testing.T) {
+	kv := testkv.New(t)
+	pool := kv.Pool(4)
+	t.Cleanup(func() { pool.Close() })
+
+	schemaManager := storage.NewSchemaManager(pool, "testdb")
+	tableManager := storage.NewTableManager(pool, schemaManager, "testdb")
+
+	if err := schemaManager.CreateTable(&storage.Schema{
+		Name:              "pairs",
+		PrimaryKeyColumns: []string{"a", "b"},
+		Columns: []storage.Column{
+			{Name: "a", Type: "TEXT", Nullable: false},
+			{Name: "b", Type: "TEXT", Nullable: false},
+			{Name: "v", Type: "TEXT", Nullable: true},
+		},
+	}); err != nil {
+		t.Fatalf("create table: %v", err)
+	}
+
+	dump, err := ExportDatabase(schemaManager, tableManager, ExportOptions{Tables: []string{"pairs"}})
+	if err != nil {
+		t.Fatalf("export: %v", err)
+	}
+	if got := strings.Count(dump, "PRIMARY KEY"); got != 1 {
+		t.Fatalf("dump has %d PRIMARY KEY clauses, want 1:\n%s", got, dump)
+	}
+	if !strings.Contains(dump, "PRIMARY KEY (a, b)") {
+		t.Fatalf("dump lost the composite key:\n%s", dump)
+	}
+}

pkg/storage/composite_key_test.go

@@ -0,0 +1,348 @@
+package storage
+
+import (
+	"errors"
+	"strings"
+	"testing"
+)
+
+// compositeTestTable creates a table whose primary key is (a, b). The key
+// columns are declared nullable on purpose: CreateTable must make them NOT NULL.
+func compositeTestTable(t *testing.T, schemas *SchemaManager) {
+	t.Helper()
+	if err := schemas.CreateTable(&Schema{
+		Name:              "t",
+		PrimaryKeyColumns: []string{"a", "b"},
+		Columns: []Column{
+			{Name: "a", Type: "TEXT", Nullable: true},
+			{Name: "b", Type: "TEXT", Nullable: true},
+			{Name: "v", Type: "TEXT", Nullable: true},
+		},
+	}); err != nil {
+		t.Fatalf("create table: %v", err)
+	}
+}
+
+func TestCompositePrimaryKeyNormalizesSchema(t *testing.T) {
+	_, _, schemas, _ := newTestSession(t)
+	compositeTestTable(t, schemas)
+
+	schema, err := schemas.GetSchema("t")
+	if err != nil {
+		t.Fatalf("get schema: %v", err)
+	}
+	if schema.PrimaryKey != "" {
+		t.Fatalf("PrimaryKey = %q, want empty for a composite key", schema.PrimaryKey)
+	}
+	if strings.Join(schema.PrimaryKeyList(), ",") != "a,b" {
+		t.Fatalf("PrimaryKeyList = %v, want [a b]", schema.PrimaryKeyList())
+	}
+	if !schema.IsCompositePrimaryKey() || schema.IsSingleColumnPrimaryKey() {
+		t.Fatal("composite key predicates disagree with the column list")
+	}
+	for _, column := range schema.Columns {
+		if schema.IsPrimaryKeyColumn(column.Name) {
+			if column.Nullable {
+				t.Fatalf("key column %s is nullable", column.Name)
+			}
+			if !column.PrimaryKey {
+				t.Fatalf("key column %s is not marked PrimaryKey", column.Name)
+			}
+		}
+	}
+}
+
+func TestCompositePrimaryKeyEncodingIsUnambiguous(t *testing.T) {
+	kv, _, schemas, tables := newTestSession(t)
+	compositeTestTable(t, schemas)
+
+	if err := tables.Insert("t", Row{"a": "ab", "b": "c", "v": "first"}); err != nil {
+		t.Fatalf("insert first: %v", err)
+	}
+	if err := tables.Insert("t", Row{"a": "a", "b": "bc", "v": "second"}); err != nil {
+		t.Fatalf("insert second: %v", err)
+	}
+
+	// Each component is length-prefixed, so the two rows cannot share a key.
+	if !kv.hasKey("testdb:_data:t:2:ab1:c") {
+		t.Fatal("expected the first composite key to be 2:ab1:c")
+	}
+	if !kv.hasKey("testdb:_data:t:1:a2:bc") {
+		t.Fatal("expected the second composite key to be 1:a2:bc")
+	}
+
+	row, err := tables.GetByPK("t", "2:ab1:c")
+	if err != nil {
+		t.Fatalf("get by composite key: %v", err)
+	}
+	if row["v"] != "first" {
+		t.Fatalf("row under 2:ab1:c = %v, want first", row["v"])
+	}
+
+	// A single component, or a truncated encoding, is not a row key.
+	for _, key := range []string{"ab", "2:ab", "1:a", "2:ab1:", "2:ab4:bc12"} {
+		if _, err := tables.GetByPK("t", key); err != ErrKeyNotFound {
+			t.Fatalf("GetByPK(%q) error = %v, want ErrKeyNotFound", key, err)
+		}
+	}
+}
+
+func TestCompositePrimaryKeyInsertErrors(t *testing.T) {
+	_, _, schemas, tables := newTestSession(t)
+	compositeTestTable(t, schemas)
+
+	var notNull *NotNullViolationError
+	if err := tables.Insert("t", Row{"a": "x"}); !errors.As(err, &notNull) || notNull.Column != "b" {
+		t.Fatalf("missing component error = %v, want NOT NULL violation on b", err)
+	}
+	notNull = nil
+	if err := tables.Insert("t", Row{"a": "x", "b": nil}); !errors.As(err, &notNull) || notNull.Column != "b" {
+		t.Fatalf("NULL component error = %v, want NOT NULL violation on b", err)
+	}
+
+	if err := tables.Insert("t", Row{"a": "x", "b": "y", "v": "one"}); err != nil {
+		t.Fatalf("insert: %v", err)
+	}
+	var duplicate *PrimaryKeyViolationError
+	if err := tables.Insert("t", Row{"a": "x", "b": "y", "v": "again"}); !errors.As(err, &duplicate) {
+		t.Fatalf("duplicate error = %v, want PrimaryKeyViolationError", err)
+	}
+
+	// Rejected inserts must not consume the row. Only the accepted row exists.
+	rows, err := tables.Select("t", nil)
+	if err != nil {
+		t.Fatalf("select: %v", err)
+	}
+	if len(rows) != 1 || rows[0]["v"] != "one" {
+		t.Fatalf("rows after rejected inserts = %v", rows)
+	}
+}
+
+func TestCompositePrimaryKeyUpdateAndDelete(t *testing.T) {
+	_, _, schemas, tables := newTestSession(t)
+	compositeTestTable(t, schemas)
+
+	if err := tables.Insert("t", Row{"a": "x", "b": "y", "v": "one"}); err != nil {
+		t.Fatalf("insert: %v", err)
+	}
+	if err := tables.Insert("t", Row{"a": "x", "b": "z", "v": "two"}); err != nil {
+		t.Fatalf("insert: %v", err)
+	}
+
+	old, ok, err := tables.UpdateByPK("t", "1:x1:y", func(Row) (Row, error) {
+		return Row{"v": "updated"}, nil
+	})
+	if err != nil || !ok {
+		t.Fatalf("UpdateByPK = ok %v, err %v", ok, err)
+	}
+	if old["v"] != "one" {
+		t.Fatalf("UpdateByPK old row = %v, want one", old["v"])
+	}
+	row, err := tables.GetByPK("t", "1:x1:y")
+	if err != nil || row["v"] != "updated" {
+		t.Fatalf("row after update = %v, err %v", row, err)
+	}
+
+	// Moving a key component through the scan path moves the durable key.
+	session := NewSession(schemas, tables)
+	if err := session.Begin(); err != nil {
+		t.Fatalf("begin: %v", err)
+	}
+	count, err := session.UpdateFunc("t",
+		func(Row) (Row, error) { return Row{"b": "w"}, nil },
+		func(row Row) bool { return row["b"] == "y" },
+	)
+	if err != nil {
+		t.Fatalf("update key component: %v", err)
+	}
+	if err := session.Commit(); err != nil {
+		t.Fatalf("commit: %v", err)
+	}
+	if count != 1 {
+		t.Fatalf("moved rows = %d, want 1", count)
+	}
+	if _, err := tables.GetByPK("t", "1:x1:y"); err != ErrKeyNotFound {
+		t.Fatalf("old key error = %v, want ErrKeyNotFound", err)
+	}
+	row, err = tables.GetByPK("t", "1:x1:w")
+	if err != nil || row["v"] != "updated" {
+		t.Fatalf("row under moved key = %v, err %v", row, err)
+	}
+
+	deleted, ok, err := tables.DeleteByPK("t", "1:x1:w")
+	if err != nil || !ok {
+		t.Fatalf("DeleteByPK = ok %v, err %v", ok, err)
+	}
+	if deleted["v"] != "updated" {
+		t.Fatalf("DeleteByPK row = %v, want updated", deleted["v"])
+	}
+	if _, err := tables.GetByPK("t", "1:x1:w"); err != ErrKeyNotFound {
+		t.Fatalf("deleted key error = %v, want ErrKeyNotFound", err)
+	}
+	if rows, err := tables.Select("t", nil); err != nil || len(rows) != 1 {
+		t.Fatalf("rows after delete = %v, err %v", rows, err)
+	}
+}
+
+func TestCompositePrimaryKeyDurableReload(t *testing.T) {
+	kv, pool, schemas, tables := newTestSession(t)
+	compositeTestTable(t, schemas)
+	if err := tables.Insert("t", Row{"a": "x", "b": "y", "v": "one"}); err != nil {
+		t.Fatalf("insert: %v", err)
+	}
+	if err := tables.Insert("t", Row{"a": "x", "b": "z", "v": "two"}); err != nil {
+		t.Fatalf("insert: %v", err)
+	}
+
+	var stored string
+	if err := pool.WithClient(func(c *KVClient) error {
+		var err error
+		stored, err = c.Read("testdb:_schema:t")
+		return err
+	}); err != nil {
+		t.Fatalf("read schema: %v", err)
+	}
+	if !strings.Contains(stored, `"primary_key_columns":["a","b"]`) {
+		t.Fatalf("persisted schema = %s, want a primary_key_columns list", stored)
+	}
+
+	// A new manager pair keeps only the durable state, as after a restart.
+	reloadedSchemas := NewSchemaManager(pool, "testdb")
+	reloadedTables := NewTableManager(pool, reloadedSchemas, "testdb")
+
+	schema, err := reloadedSchemas.GetSchema("t")
+	if err != nil {
+		t.Fatalf("get reloaded schema: %v", err)
+	}
+	if strings.Join(schema.PrimaryKeyList(), ",") != "a,b" || schema.PrimaryKey != "" {
+		t.Fatalf("reloaded key = %q/%v, want a,b", schema.PrimaryKey, schema.PrimaryKeyList())
+	}
+
+	row, err := reloadedTables.GetByPK("t", "1:x1:y")
+	if err != nil || row["v"] != "one" {
+		t.Fatalf("reloaded row = %v, err %v", row, err)
+	}
+	rows, err := reloadedTables.Select("t", nil)
+	if err != nil || len(rows) != 2 {
+		t.Fatalf("reloaded select = %v, err %v", rows, err)
+	}
+
+	// Writes after the reload keep the same encoding.
+	if err := reloadedTables.Insert("t", Row{"a": "x", "b": "y", "v": "again"}); err == nil {
+		t.Fatal("expected a duplicate key to fail after reload")
+	}
+	if err := reloadedTables.Insert("t", Row{"a": "q", "b": "r", "v": "three"}); err != nil {
+		t.Fatalf("insert after reload: %v", err)
+	}
+	if !kv.hasKey("testdb:_data:t:1:q1:r") {
+		t.Fatal("insert after reload did not use the composite encoding")
+	}
+}
+
+func TestSingleColumnPrimaryKeyEncodingUnchanged(t *testing.T) {
+	kv, _, schemas, tables := newTestSession(t)
+
+	if err := schemas.CreateTable(&Schema{
+		Name:    "ints",
+		Columns: []Column{{Name: "id", Type: "INTEGER", PrimaryKey: true}, {Name: "v", Type: "TEXT"}},
+	}); err != nil {
+		t.Fatalf("create ints: %v", err)
+	}
+	if err := tables.Insert("ints", Row{"id": int64(7), "v": "seven"}); err != nil {
+		t.Fatalf("insert ints: %v", err)
+	}
+	if !kv.hasKey("testdb:_data:ints:7") {
+		t.Fatal("INTEGER primary key row key changed from the bare value")
+	}
+	if row, err := tables.GetByPK("ints", "7"); err != nil || row["v"] != "seven" {
+		t.Fatalf("GetByPK = %v, err %v", row, err)
+	}
+
+	if err := schemas.CreateTable(&Schema{
+		Name:    "texts",
+		Columns: []Column{{Name: "name", Type: "TEXT", PrimaryKey: true}},
+	}); err != nil {
+		t.Fatalf("create texts: %v", err)
+	}
+	if err := tables.Insert("texts", Row{"name": "alpha"}); err != nil {
+		t.Fatalf("insert texts: %v", err)
+	}
+	if !kv.hasKey("testdb:_data:texts:alpha") {
+		t.Fatal("TEXT primary key row key changed from the bare value")
+	}
+
+	var stored string
+	if err := schemas.GetPool().WithClient(func(c *KVClient) error {
+		var err error
+		stored, err = c.Read("testdb:_schema:ints")
+		return err
+	}); err != nil {
+		t.Fatalf("read schema: %v", err)
+	}
+	if !strings.Contains(stored, `"primary_key":"id"`) {
+		t.Fatalf("persisted schema = %s, want the legacy primary_key field", stored)
+	}
+	if strings.Contains(stored, "primary_key_columns") {
+		t.Fatalf("persisted schema = %s, want no composite field for a single-column key", stored)
+	}
+}
+
+func TestLegacySingleColumnSchemaLoads(t *testing.T) {
+	kv, pool, schemas, tables := newTestSession(t)
+
+	// A catalog written before composite keys only has "primary_key".
+	const legacy = `{"name":"legacy","columns":[{"name":"id","type":"INTEGER","nullable":false,"primary_key":true},{"name":"v","type":"TEXT","nullable":true}],"primary_key":"id","created_at":"2026-01-01T00:00:00Z","next_rowid":0,"autoincrement":false}`
+	if err := pool.WithClient(func(c *KVClient) error {
+		return c.Write("testdb:_schema:legacy", legacy)
+	}); err != nil {
+		t.Fatalf("write legacy schema: %v", err)
+	}
+
+	schema, err := schemas.GetSchema("legacy")
+	if err != nil {
+		t.Fatalf("get legacy schema: %v", err)
+	}
+	if schema.PrimaryKey != "id" || strings.Join(schema.PrimaryKeyList(), ",") != "id" {
+		t.Fatalf("legacy key = %q/%v, want id", schema.PrimaryKey, schema.PrimaryKeyList())
+	}
+
+	if err := tables.Insert("legacy", Row{"id": int64(5), "v": "five"}); err != nil {
+		t.Fatalf("insert into legacy table: %v", err)
+	}
+	if !kv.hasKey("testdb:_data:legacy:5") {
+		t.Fatal("legacy INTEGER primary key did not use the bare-value key")
+	}
+	if row, err := tables.GetByPK("legacy", "5"); err != nil || row["v"] != "five" {
+		t.Fatalf("GetByPK = %v, err %v", row, err)
+	}
+}
+
+// Insert resolves catalog defaults for key components and writes every resolved
+// key value back into the caller's row, so callers (RETURNING, conflict
+// matching) observe the key the row was stored under.
+func TestCompositePrimaryKeyInsertWritesResolvedKeyBack(t *testing.T) {
+	_, _, schemas, tables := newTestSession(t)
+	if err := schemas.CreateTable(&Schema{
+		Name:              "t",
+		PrimaryKeyColumns: []string{"a", "b"},
+		Columns: []Column{
+			{Name: "a", Type: "TEXT", Default: "d"},
+			{Name: "b", Type: "TEXT"},
+			{Name: "v", Type: "TEXT"},
+		},
+	}); err != nil {
+		t.Fatalf("create table: %v", err)
+	}
+
+	row := Row{"b": "y", "v": "one"}
+	if err := tables.Insert("t", row); err != nil {
+		t.Fatalf("insert: %v", err)
+	}
+	if row["a"] != "d" || row["b"] != "y" {
+		t.Fatalf("caller row after insert = %v, want a='d' b='y'", row)
+	}
+	stored, err := tables.GetByPK("t", "1:d1:y")
+	if err != nil || stored["v"] != "one" {
+		t.Fatalf("stored row = %v, err %v", stored, err)
+	}
+}

pkg/storage/pizzakv_integration_test.go

@@ -2,10 +2,12 @@ package storage
 
 import (
 	"bytes"
+	"errors"
 	"fmt"
 	"os"
 	"os/exec"
 	"path/filepath"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -141,6 +143,80 @@ func TestPizzaKVIntegration(t *testing.T) {
 	}
 }
 
+// A composite key must survive a real engine restart: the encoded key is the
+// durable row key, so both reads and duplicate detection keep working after the
+// schema and rows are reloaded from PizzaKV.
+func TestPizzaKVCompositePrimaryKeyIntegration(t *testing.T) {
+	binary := os.Getenv("PIZZAKV_BIN")
+	if binary == "" {
+		t.Skip("PIZZAKV_BIN is not set")
+	}
+
+	dir := t.TempDir()
+	socket := shortPizzaKVSocket(t)
+	database := filepath.Join(dir, "composite.pkvdb")
+	stop := startPizzaKVTest(t, binary, socket, database)
+	pool := waitPizzaKVPool(t, socket)
+
+	schemas := NewSchemaManager(pool, "composite")
+	tables := NewTableManager(pool, schemas, "composite")
+	if err := schemas.CreateTable(&Schema{
+		Name:              "pairs",
+		PrimaryKeyColumns: []string{"a", "b"},
+		Columns: []Column{
+			{Name: "a", Type: "TEXT", Nullable: false},
+			{Name: "b", Type: "INTEGER", Nullable: false},
+			{Name: "v", Type: "TEXT", Nullable: true},
+		},
+	}); err != nil {
+		t.Fatalf("create table: %v", err)
+	}
+	for _, row := range []Row{
+		{"a": "x", "b": int64(1), "v": "one"},
+		{"a": "x", "b": int64(2), "v": "two"},
+	} {
+		if err := tables.Insert("pairs", row); err != nil {
+			t.Fatalf("insert %v: %v", row, err)
+		}
+	}
+
+	if err := pool.Close(); err != nil {
+		t.Fatalf("close pool: %v", err)
+	}
+	stop()
+	startPizzaKVTest(t, binary, socket, database)
+	pool = waitPizzaKVPool(t, socket)
+	defer pool.Close()
+
+	schemas = NewSchemaManager(pool, "composite")
+	tables = NewTableManager(pool, schemas, "composite")
+	schema, err := schemas.GetSchema("pairs")
+	if err != nil {
+		t.Fatalf("reload schema: %v", err)
+	}
+	if strings.Join(schema.PrimaryKeyList(), ",") != "a,b" {
+		t.Fatalf("reloaded key = %v, want [a b]", schema.PrimaryKeyList())
+	}
+	row, err := tables.GetByPK("pairs", "1:x1:1")
+	if err != nil || row["v"] != "one" {
+		t.Fatalf("point read after restart = %v, err %v", row, err)
+	}
+	rows, err := tables.Select("pairs", nil)
+	if err != nil || len(rows) != 2 {
+		t.Fatalf("scan after restart = %v, err %v", rows, err)
+	}
+	var duplicate *PrimaryKeyViolationError
+	if err := tables.Insert("pairs", Row{"a": "x", "b": int64(1), "v": "again"}); !errors.As(err, &duplicate) {
+		t.Fatalf("duplicate after restart = %v, want PrimaryKeyViolationError", err)
+	}
+	if err := tables.Insert("pairs", Row{"a": "x", "b": int64(3), "v": "three"}); err != nil {
+		t.Fatalf("insert after restart: %v", err)
+	}
+	if count, err := tables.CountFast("pairs"); err != nil || count != 3 {
+		t.Fatalf("count after restart = %d, err %v", count, err)
+	}
+}
+
 func TestPizzaKVCompareBatchIntegration(t *testing.T) {
 	binary := os.Getenv("PIZZAKV_BIN")
 	if binary == "" {

pkg/storage/schema.go

@@ -3,6 +3,7 @@ package storage
 import (
 	"errors"
 	"fmt"
+	"strconv"
 	"strings"
 	"sync"
 	"time"
@@ -19,12 +20,20 @@ var ErrIndexNotFound = errors.New("index not found")
 
 // Schema represents a table schema.
 type Schema struct {
-	Name          string    `json:"name"`
-	Columns       []Column  `json:"columns"`
-	PrimaryKey    string    `json:"primary_key"`
-	CreatedAt     time.Time `json:"created_at"`
-	NextRowID     int64     `json:"next_rowid"`
-	AutoIncrement bool      `json:"autoincrement"`
+	Name    string   `json:"name"`
+	Columns []Column `json:"columns"`
+	// PrimaryKey is the single primary-key column. It stays the source of truth
+	// for pre-existing catalogs that do not carry PrimaryKeyColumns, and it is
+	// empty for a composite key so a caller cannot mistake one member for the
+	// whole key. Use PrimaryKeyList instead of reading either field directly.
+	PrimaryKey string `json:"primary_key"`
+	// PrimaryKeyColumns is the ordered primary-key column list for a composite
+	// key. It is omitted for single-column keys so the catalog keeps its
+	// historical shape; normalizePrimaryKey fills it in when a schema is loaded.
+	PrimaryKeyColumns []string  `json:"primary_key_columns,omitempty"`
+	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.
@@ -319,20 +328,10 @@ func (m *SchemaManager) CreateTable(schema *Schema) error {
 	schema = cloneSchema(schema)
 	schema.CreatedAt = time.Now()
 
-	// Determine primary key if not set
-	if schema.PrimaryKey == "" {
-		for _, col := range schema.Columns {
-			if col.PrimaryKey {
-				schema.PrimaryKey = col.Name
-				break
-			}
-		}
-		// No explicit primary key declared — use synthetic _rowid_ so user
-		// columns remain unconstrained and can hold duplicate or NULL values.
-		if schema.PrimaryKey == "" {
-			schema.PrimaryKey = "_rowid_"
-		}
-	}
+	// Determine the primary key if not set. A schema with no declared key uses
+	// the synthetic _rowid_ so user columns remain unconstrained and can hold
+	// duplicate or NULL values.
+	schema.normalizePrimaryKey()
 
 	// Serialize schema
 	data, err := json.Marshal(schema)
@@ -458,6 +457,7 @@ func (m *SchemaManager) GetSchema(name string) (*Schema, error) {
 	if err := json.Unmarshal([]byte(data), &schema); err != nil {
 		return nil, fmt.Errorf("failed to parse schema: %w", err)
 	}
+	schema.normalizePrimaryKey()
 
 	m.cache[strings.ToLower(name)] = &schema
 	return cloneSchema(&schema), nil
@@ -470,6 +470,7 @@ func cloneSchema(schema *Schema) *Schema {
 	cloned := *schema
 	cloned.Columns = append([]Column(nil), schema.Columns...)
 	cloned.Checks = append([]CheckConstraint(nil), schema.Checks...)
+	cloned.PrimaryKeyColumns = append([]string(nil), schema.PrimaryKeyColumns...)
 	return &cloned
 }
 
@@ -593,6 +594,136 @@ func (s *Schema) ToAnalyzerTableInfo() *analyzer.TableInfo {
 	return info
 }
 
+// normalizePrimaryKey derives the ordered primary-key column list from the
+// legacy single-column field or from the columns marked PRIMARY KEY, and keeps
+// both representations consistent. Primary-key columns are implicitly NOT NULL.
+func (s *Schema) normalizePrimaryKey() {
+	if len(s.PrimaryKeyColumns) == 0 {
+		if s.PrimaryKey != "" {
+			s.PrimaryKeyColumns = []string{s.PrimaryKey}
+		} else {
+			for _, col := range s.Columns {
+				if col.PrimaryKey {
+					s.PrimaryKeyColumns = append(s.PrimaryKeyColumns, col.Name)
+				}
+			}
+			if len(s.PrimaryKeyColumns) == 0 {
+				s.PrimaryKeyColumns = []string{"_rowid_"}
+			}
+		}
+	}
+	if len(s.PrimaryKeyColumns) == 1 {
+		// A single-column key keeps the historical catalog shape: only the
+		// scalar field, so existing schema bytes stay unchanged.
+		s.PrimaryKey = s.PrimaryKeyColumns[0]
+		s.PrimaryKeyColumns = nil
+	} else {
+		s.PrimaryKey = ""
+	}
+	for i := range s.Columns {
+		if s.IsPrimaryKeyColumn(s.Columns[i].Name) {
+			s.Columns[i].PrimaryKey = true
+			s.Columns[i].Nullable = false
+		}
+	}
+}
+
+// PrimaryKeyList returns the ordered primary-key columns. A schema always has
+// at least one: an explicit key, or the synthetic _rowid_.
+func (s *Schema) PrimaryKeyList() []string {
+	if len(s.PrimaryKeyColumns) > 0 {
+		return append([]string(nil), s.PrimaryKeyColumns...)
+	}
+	if s.PrimaryKey != "" {
+		return []string{s.PrimaryKey}
+	}
+	return []string{"_rowid_"}
+}
+
+// UsesSyntheticRowID reports whether the table declares no primary key, so rows
+// are stored under the engine-assigned ROWID.
+func (s *Schema) UsesSyntheticRowID() bool {
+	columns := s.PrimaryKeyList()
+	return len(columns) == 1 && strings.EqualFold(columns[0], "_rowid_")
+}
+
+// IsCompositePrimaryKey reports whether the primary key spans more than one
+// column.
+func (s *Schema) IsCompositePrimaryKey() bool {
+	return len(s.PrimaryKeyList()) > 1
+}
+
+// IsSingleColumnPrimaryKey reports whether the primary key is exactly one
+// column, which is the only shape a durable point lookup may address with one
+// value.
+func (s *Schema) IsSingleColumnPrimaryKey() bool {
+	return len(s.PrimaryKeyList()) == 1
+}
+
+// IsPrimaryKeyColumn reports whether name is one of the primary-key columns.
+func (s *Schema) IsPrimaryKeyColumn(name string) bool {
+	for _, column := range s.PrimaryKeyList() {
+		if strings.EqualFold(column, name) {
+			return true
+		}
+	}
+	return false
+}
+
+// PrimaryKeyTargetMatches reports whether columns names the complete
+// primary-key column set. Order is not significant: ON CONFLICT (b, a) on a
+// PRIMARY KEY (a, b) addresses the same key, because the encoded key is built
+// in declaration order either way. SQLite instead requires the declaration
+// order of a constraint; this engine already accepts arbitrary conflict-target
+// lists, so it matches the set rather than raising.
+func (s *Schema) PrimaryKeyTargetMatches(columns []string) bool {
+	primary := s.PrimaryKeyList()
+	if len(primary) != len(columns) {
+		return false
+	}
+	for _, column := range primary {
+		found := false
+		for _, candidate := range columns {
+			if strings.EqualFold(column, candidate) {
+				found = true
+				break
+			}
+		}
+		if !found {
+			return false
+		}
+	}
+	return true
+}
+
+// PrimaryKeyForRow returns the row's primary key encoded the way the table's
+// durable row keys encode it. A single-column key keeps the historical
+// "<value>" form, so existing databases are read and written unchanged. A
+// composite key length-prefixes each component ("<len>:<value>" repeated), a
+// prefix-free encoding, so ("ab","c") and ("a","bc") cannot collide. A NULL or
+// missing component is a NOT NULL violation: no rowid is auto-assigned.
+func (s *Schema) PrimaryKeyForRow(row Row) (string, error) {
+	columns := s.PrimaryKeyList()
+	parts := make([]string, len(columns))
+	for i, column := range columns {
+		value, ok := rowColumnValue(row, column)
+		if !ok || value == nil {
+			return "", &NotNullViolationError{Table: s.Name, Column: column}
+		}
+		parts[i] = fmt.Sprintf("%v", value)
+	}
+	if len(parts) == 1 {
+		return parts[0], nil
+	}
+	var b strings.Builder
+	for _, part := range parts {
+		b.WriteString(strconv.Itoa(len(part)))
+		b.WriteByte(':')
+		b.WriteString(part)
+	}
+	return b.String(), nil
+}
+
 // GetColumn returns a column by name.
 func (s *Schema) GetColumn(name string) (*Column, bool) {
 	lowerName := strings.ToLower(name)
@@ -781,6 +912,7 @@ func (m *SchemaManager) getSchemaLocked(name string) (*Schema, error) {
 	if err := json.Unmarshal([]byte(data), &schema); err != nil {
 		return nil, fmt.Errorf("failed to parse schema: %w", err)
 	}
+	schema.normalizePrimaryKey()
 
 	m.cache[strings.ToLower(name)] = &schema
 	return &schema, nil
@@ -1265,7 +1397,7 @@ func (m *SchemaManager) DropColumn(table, columnName string) error {
 	}
 
 	// Cannot drop primary key column
-	if strings.EqualFold(schema.PrimaryKey, columnName) {
+	if schema.IsPrimaryKeyColumn(columnName) {
 		return fmt.Errorf("cannot drop primary key column: %s", columnName)
 	}
 
@@ -1596,7 +1728,13 @@ func (m *SchemaManager) RenameColumn(table, oldName, newName string) error {
 			schema.Columns[i].Name = newName
 			found = true
 
-			// Update primary key reference if needed
+			// Update primary key references if needed. Renaming keeps the column
+			// order, so a composite key's encoding is unchanged.
+			for i, pkColumn := range schema.PrimaryKeyColumns {
+				if strings.EqualFold(pkColumn, oldName) {
+					schema.PrimaryKeyColumns[i] = newName
+				}
+			}
 			if strings.EqualFold(schema.PrimaryKey, oldName) {
 				schema.PrimaryKey = newName
 			}
@@ -1840,6 +1978,7 @@ func (m *SchemaManager) getSchemaUnsafe(table string) (*Schema, error) {
 	if err := json.Unmarshal([]byte(data), &schema); err != nil {
 		return nil, err
 	}
+	schema.normalizePrimaryKey()
 
 	m.cache[tableLower] = &schema
 	return &schema, nil

pkg/storage/table.go

@@ -468,65 +468,26 @@ func (m *TableManager) dataPrefix(table string) string {
 // prepareInsert validates and normalizes an insert row, generating the ROWID
 // when required. It returns the normalized row and the full data key without
 // writing anything, so both the autocommit path (compare-and-swap) and the
-// buffered transaction path (staging) can share it. The input row has its
-// primary key populated as a side effect.
+// buffered transaction path (staging) can share it.
 func (m *TableManager) prepareInsert(table string, row Row) (Row, string, error) {
 	schema, err := m.schema.GetSchema(table)
 	if err != nil {
 		return nil, "", err
 	}
-
-	// Get primary key value
-	pkValue, ok := row[schema.PrimaryKey]
-	if !ok {
-		for k, v := range row {
-			if strings.EqualFold(k, schema.PrimaryKey) {
-				pkValue = v
-				ok = true
-				break
-			}
-		}
-	}
-
-	pkCol, _ := schema.GetColumn(schema.PrimaryKey)
-	isIntegerPK := pkCol != nil && isIntegerType(pkCol.Type)
-
-	var rowid int64
-	if !ok || pkValue == nil {
-		if isIntegerPK || !ok {
-			rowid, err = m.schema.GetNextRowID(table)
-			if err != nil {
-				return nil, "", err
-			}
-			pkValue = rowid
-			row[schema.PrimaryKey] = rowid
-			ok = true
-		} else {
-			return nil, "", fmt.Errorf("missing primary key: %s", schema.PrimaryKey)
-		}
-	} else if isIntegerPK {
-		switch v := pkValue.(type) {
-		case int64:
-			rowid = v
-		case float64:
-			if math.Trunc(v) != v {
-				return nil, "", fmt.Errorf("invalid integer primary key: %v", v)
-			}
-			rowid = int64(v)
-		case int:
-			rowid = int64(v)
-		default:
-			rowid = 0
-		}
-		if rowid > 0 {
-			if err := m.schema.UpdateMaxRowID(table, rowid); err != nil {
-				return nil, "", err
-			}
-		}
+	normalizedRow, key, _, err := m.normalizeInsertRow(table, schema, row)
+	if err != nil {
+		return nil, "", err
 	}
+	return normalizedRow, key, nil
+}
 
-	pk := fmt.Sprintf("%v", pkValue)
-
+// normalizeInsertRow completes a row image for an insert: it copies the caller's
+// columns case-insensitively, applies catalog defaults, assigns the ROWID (or
+// the INTEGER PRIMARY KEY rowid alias), encodes the primary key, and validates
+// the row constraints. It is shared by the point insert and bulk insert paths so
+// both key rows identically. The returned rowid is the row's ROWID alias, or 0
+// when the caller must not treat the primary key as a rowid.
+func (m *TableManager) normalizeInsertRow(table string, schema *Schema, row Row) (Row, string, int64, error) {
 	normalizedRow := make(Row)
 	for _, col := range schema.Columns {
 		for k, v := range row {
@@ -542,27 +503,78 @@ func (m *TableManager) prepareInsert(table string, row Row) (Row, string, error)
 		}
 		value, hasDefault, err := m.schema.columnDefaultValue(col, m.schema.defaultNow())
 		if err != nil {
-			return nil, "", err
+			return nil, "", 0, err
 		}
 		if hasDefault {
 			normalizedRow[col.Name] = value
 		}
 	}
 
+	// Only a single-column INTEGER PRIMARY KEY is a ROWID alias. A composite key
+	// is never an alias, so its rows always carry an engine-assigned _rowid_.
+	var rowid int64
+	if schema.IsSingleColumnPrimaryKey() && !schema.UsesSyntheticRowID() {
+		pkColumn := schema.PrimaryKey
+		pkCol, _ := schema.GetColumn(pkColumn)
+		if pkCol != nil && isIntegerType(pkCol.Type) {
+			pkValue := normalizedRow[pkColumn]
+			if pkValue == nil {
+				nextRowID, err := m.schema.GetNextRowID(table)
+				if err != nil {
+					return nil, "", 0, err
+				}
+				rowid = nextRowID
+				normalizedRow[pkColumn] = rowid
+			} else {
+				switch v := pkValue.(type) {
+				case int64:
+					rowid = v
+				case float64:
+					if math.Trunc(v) != v {
+						return nil, "", 0, fmt.Errorf("invalid integer primary key: %v", v)
+					}
+					rowid = int64(v)
+				case int:
+					rowid = int64(v)
+				default:
+					rowid = 0
+				}
+				if rowid > 0 {
+					if err := m.schema.UpdateMaxRowID(table, rowid); err != nil {
+						return nil, "", 0, err
+					}
+				}
+			}
+		}
+	}
+
 	if rowid > 0 {
 		normalizedRow["_rowid_"] = rowid
 	} else {
 		newRowID, err := m.schema.GetNextRowID(table)
 		if err != nil {
-			return nil, "", err
+			return nil, "", 0, err
 		}
 		normalizedRow["_rowid_"] = newRowID
 	}
 
+	key, err := schema.PrimaryKeyForRow(normalizedRow)
+	if err != nil {
+		return nil, "", 0, err
+	}
 	if err := m.schema.validateRowConstraints(schema, normalizedRow); err != nil {
-		return nil, "", err
+		return nil, "", 0, err
 	}
-	return normalizedRow, m.dataKey(table, pk), nil
+	// Callers read the resolved key back from the row they passed in: RETURNING
+	// needs an auto-assigned INTEGER PRIMARY KEY, and a composite table needs key
+	// components that came from a catalog DEFAULT because the conflict-matching
+	// and fetch paths run before the next storage call.
+	for _, column := range schema.PrimaryKeyList() {
+		if value, ok := normalizedRow[column]; ok {
+			row[column] = value
+		}
+	}
+	return normalizedRow, m.dataKey(table, key), rowid, nil
 }
 
 // Insert inserts a new row. The duplicate check and write are one atomic
@@ -708,102 +720,29 @@ func (m *TableManager) InsertBulkWithLastRowID(table string, rows []Row) (int, i
 	}
 	wasInit := m.countInitialized(table)
 
-	pkCol, _ := schema.GetColumn(schema.PrimaryKey)
-	isIntegerPK := pkCol != nil && isIntegerType(pkCol.Type)
-
-	// Normalize rows and assign _rowid_.
+	// Normalize rows, assigning the ROWID and encoding the primary key exactly as
+	// the point insert path does.
 	normalized := make([]Row, 0, len(rows))
-	var maxRowID int64
+	dataKeys := make([]string, 0, len(rows))
 	for _, row := range rows {
-		nr := make(Row)
-		for _, col := range schema.Columns {
-			for k, v := range row {
-				if strings.EqualFold(k, col.Name) {
-					nr[col.Name] = v
-					break
-				}
-			}
-		}
-		for _, col := range schema.Columns {
-			if _, ok := nr[col.Name]; ok {
-				continue
-			}
-			value, hasDefault, err := m.schema.columnDefaultValue(col, m.schema.defaultNow())
-			if err != nil {
-				return 0, 0, err
-			}
-			if hasDefault {
-				nr[col.Name] = value
-			}
-		}
-		var rowid int64
-		var hasRowid bool
-		if isIntegerPK {
-			switch v := nr[schema.PrimaryKey].(type) {
-			case float64:
-				if math.Trunc(v) != v {
-					return 0, 0, fmt.Errorf("invalid integer primary key: %v", v)
-				}
-				rowid = int64(v)
-				hasRowid = true
-			case int64:
-				rowid = v
-				hasRowid = true
-			case int:
-				rowid = int64(v)
-				hasRowid = true
-			}
-		}
-		if !hasRowid {
-			// Auto-generate for an INTEGER primary key or the synthetic _rowid_,
-			// mirroring prepareInsert so INSERT ... SELECT works on AUTOINCREMENT
-			// tables.
-			if isIntegerPK || schema.PrimaryKey == "_rowid_" {
-				rowid, err = m.schema.GetNextRowID(table)
-				if err != nil {
-					return 0, 0, err
-				}
-				if isIntegerPK {
-					nr[schema.PrimaryKey] = rowid
-				} else {
-					nr["_rowid_"] = rowid
-				}
-			} else {
-				pk, ok := nr[schema.PrimaryKey]
-				if !ok || pk == nil {
-					return 0, 0, fmt.Errorf("missing primary key: %s", schema.PrimaryKey)
-				}
-				rowid, err = m.schema.GetNextRowID(table)
-				if err != nil {
-					return 0, 0, err
-				}
-			}
-		}
-		nr["_rowid_"] = rowid
-		if err := m.schema.validateRowConstraints(schema, nr); err != nil {
+		nr, key, _, err := m.normalizeInsertRow(table, schema, row)
+		if err != nil {
 			return 0, 0, err
 		}
-		if rowid > maxRowID {
-			maxRowID = rowid
-		}
 		normalized = append(normalized, nr)
-	}
-
-	if maxRowID > 0 {
-		m.schema.UpdateMaxRowID(table, maxRowID)
+		dataKeys = append(dataKeys, key)
 	}
 
 	// Serialize all rows into batch operations. A parallel slice keeps the
 	// normalized Row for each op for index maintenance after the write.
 	ops := make([]BatchOp, 0, len(normalized))
 	encoded := make([]Row, 0, len(normalized))
-	for _, nr := range normalized {
-		pk := fmt.Sprintf("%v", nr[schema.PrimaryKey])
+	for i, nr := range normalized {
 		data, err := encodeRow(nr)
 		if err != nil {
 			return 0, 0, err
 		}
-		ops = append(ops, BatchOp{Op: batchPut, Key: []byte(m.dataKey(table, pk)), Value: data})
+		ops = append(ops, BatchOp{Op: batchPut, Key: []byte(dataKeys[i]), Value: data})
 		encoded = append(encoded, nr)
 	}
 	keys := make([][]byte, len(ops))
@@ -1425,8 +1364,12 @@ func (m *TableManager) ensureIndex(index *Index) error {
 			return true, err
 		}
 		valueKey := formatIndexValue(colValue)
+		pk, err := tableSchema.PrimaryKeyForRow(row)
+		if err != nil {
+			return true, err
+		}
 		values[valueKey] = append(values[valueKey], rowid)
-		rowKeys[rowid] = fmt.Sprintf("%v", row[tableSchema.PrimaryKey])
+		rowKeys[rowid] = pk
 		return false, nil
 	}); err != nil {
 		return err
@@ -1571,8 +1514,12 @@ func (m *TableManager) BuildIndex(indexName, tableName string, columns []string)
 		if err != nil {
 			return true, err
 		}
+		pk, err := tableSchema.PrimaryKeyForRow(row)
+		if err != nil {
+			return true, err
+		}
 		values[formatIndexValue(colValue)] = append(values[formatIndexValue(colValue)], rowid)
-		rowKeys[rowid] = fmt.Sprintf("%v", row[tableSchema.PrimaryKey])
+		rowKeys[rowid] = pk
 		return false, nil
 	}); err != nil {
 		return err

pkg/storage/tx.go

@@ -545,7 +545,11 @@ func (s *Session) selectLocked(table string, filter func(Row) bool) ([]Row, erro
 
 	var rows []Row
 	err = s.table.scanRowsWithLSN(table, func(row Row, lsn uint64) (bool, error) {
-		key := s.table.dataKey(table, fmt.Sprintf("%v", row[schema.PrimaryKey]))
+		pk, err := schema.PrimaryKeyForRow(row)
+		if err != nil {
+			return true, err
+		}
+		key := s.table.dataKey(table, pk)
 		if _, ok := overlay[key]; ok {
 			return false, nil
 		}
@@ -733,12 +737,10 @@ func (s *Session) insertLocked(table string, row Row) (int64, error) {
 		return 0, err
 	}
 	rowid, _ := rowIDFromRow(nr)
-	schema, err := s.schema.GetSchema(table)
-	if err != nil {
-		return 0, err
-	}
-	pk := fmt.Sprintf("%v", nr[schema.PrimaryKey])
 	tl := strings.ToLower(table)
+	// prepareInsert returned the durable key; duplicate reporting shows the same
+	// encoded primary-key component the key carries.
+	pk := strings.TrimPrefix(key, s.table.dataPrefix(table))
 
 	if e, ok := s.overlay[tl][key]; ok {
 		if !e.absent {
@@ -877,7 +879,11 @@ func (s *Session) UpdateFunc(table string, updateFn func(Row) (Row, error), filt
 	count := 0
 	for _, row := range rows {
 		oldRow := cloneRow(row)
-		oldKey := s.table.dataKey(table, fmt.Sprintf("%v", row[schema.PrimaryKey]))
+		oldPK, err := schema.PrimaryKeyForRow(row)
+		if err != nil {
+			return count, err
+		}
+		oldKey := s.table.dataKey(table, oldPK)
 		updates, err := s.runUpdateFn(updateFn, row)
 		if err != nil {
 			return count, err
@@ -893,12 +899,16 @@ func (s *Session) UpdateFunc(table string, updateFn func(Row) (Row, error), filt
 		if err := s.schema.validateRowConstraints(schema, row); err != nil {
 			return count, err
 		}
-		newKey := s.table.dataKey(table, fmt.Sprintf("%v", row[schema.PrimaryKey]))
+		newPK, err := schema.PrimaryKeyForRow(row)
+		if err != nil {
+			return count, err
+		}
+		newKey := s.table.dataKey(table, newPK)
 		if newKey != oldKey {
 			// The primary key changed: move the staged row instead of leaving a
 			// stale copy under the old key, and refuse to overwrite another row.
-			if _, gerr := s.getByPKLocked(table, fmt.Sprintf("%v", row[schema.PrimaryKey])); gerr == nil {
-				return count, &PrimaryKeyViolationError{Table: table, Key: fmt.Sprint(row[schema.PrimaryKey])}
+			if _, gerr := s.getByPKLocked(table, newPK); gerr == nil {
+				return count, &PrimaryKeyViolationError{Table: table, Key: newPK}
 			} else if gerr != ErrKeyNotFound {
 				return count, gerr
 			}
@@ -929,8 +939,11 @@ func (s *Session) Delete(table string, filter func(Row) bool) (int, error) {
 	}
 	count := 0
 	for _, row := range rows {
-		key := s.table.dataKey(table, fmt.Sprintf("%v", row[schema.PrimaryKey]))
-		s.stageDelete(table, key, row)
+		pk, err := schema.PrimaryKeyForRow(row)
+		if err != nil {
+			return count, err
+		}
+		s.stageDelete(table, s.table.dataKey(table, pk), row)
 		count++
 	}
 	return count, nil