fix: close composite PK upsert ghosts and review gaps

- Reject ON CONFLICT DO UPDATE assignments to a primary-key column with
  0A000. The upsert updates the row in place, so accepting the assignment
  wrote the new key values under the old durable key, and a later full-key
  update split the row in two.
- Resolve constant catalog defaults in the executor's candidate row so a
  key component that comes from a DEFAULT participates in conflict
  matching. DO UPDATE, DO NOTHING, targetless ON CONFLICT, and INSERT OR
  REPLACE now hit the defaulted-key row instead of missing it and failing
  with a duplicate key; single-column keys follow the same path.
- Reject ALTER TABLE ADD COLUMN ... PRIMARY KEY with 0A000 (SQLite also
  rejects adding a PRIMARY KEY column).
- Reject a generated column in a primary key with 0A000.
- Write every resolved key-column value back into the caller's row, so
  INSERT ... RETURNING sees composite key components that came from a
  DEFAULT.
- Drop the unused conflictMatcher.name field and match a conflict target
  against the primary-key column set. ON CONFLICT (b, a) on a PRIMARY KEY
  (a, b) addresses the same key and is documented as such; SQLite requires
  declaration order instead.
- Document the additive primary_key_columns catalog field and the
  pre-composite rollback behavior in PKBFI_STORAGE_MIGRATION.md.
- Coverage: composite PK combined with a UNIQUE index (key move,
  ON CONFLICT on the unique column, INSERT OR REPLACE), last_insert_rowid()
  on a composite table, a capabilities assertion for composite_primary_key,
  and wire-level coverage for the new rejections and the defaulted-key
  upsert.

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

@@ -168,7 +168,7 @@ func Current() *Document {
 			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: "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"},
+			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

@@ -411,3 +411,293 @@ func TestCompositePrimaryKeyAlterTable(t *testing.T) {
 		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

@@ -2621,7 +2621,7 @@ 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 schema.PrimaryKeyMatches(columns) {
+	if schema.PrimaryKeyTargetMatches(columns) {
 		pk, err := schema.PrimaryKeyForRow(row)
 		if err != nil {
 			// A missing or NULL key component cannot conflict: the insert itself
@@ -2755,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
@@ -3740,6 +3755,12 @@ 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
 				inlinePrimaryKeyColumns = append(inlinePrimaryKeyColumns, col.Name)
@@ -3801,6 +3822,12 @@ func (e *Executor) executeCreateTable(stmt *parser.CreateTableStmt) (*Result, er
 						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",
@@ -4232,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:

pkg/executor/generated.go

@@ -111,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
 }
@@ -123,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 {
@@ -136,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
 }
@@ -180,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 {

pkg/pgserver/composite_primary_key_test.go

@@ -1,6 +1,7 @@
 package pgserver
 
 import (
+	"strings"
 	"testing"
 )
 
@@ -79,10 +80,53 @@ func TestCompositePrimaryKeyDDLRejections(t *testing.T) {
 		"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/storage/composite_key_test.go

@@ -316,3 +316,33 @@ func TestLegacySingleColumnSchemaLoads(t *testing.T) {
 		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/schema.go

@@ -670,15 +670,26 @@ func (s *Schema) IsPrimaryKeyColumn(name string) bool {
 	return false
 }
 
-// PrimaryKeyMatches reports whether columns is the complete primary-key column
-// list in declaration order.
-func (s *Schema) PrimaryKeyMatches(columns []string) bool {
+// 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 i := range primary {
-		if !strings.EqualFold(primary[i], columns[i]) {
+	for _, column := range primary {
+		found := false
+		for _, candidate := range columns {
+			if strings.EqualFold(column, candidate) {
+				found = true
+				break
+			}
+		}
+		if !found {
 			return false
 		}
 	}

pkg/storage/table.go

@@ -565,12 +565,13 @@ func (m *TableManager) normalizeInsertRow(table string, schema *Schema, row Row)
 	if err := m.schema.validateRowConstraints(schema, normalizedRow); err != nil {
 		return nil, "", 0, err
 	}
-	// Callers that read the assigned key back from the row they passed in (for
-	// example RETURNING on an auto-assigned INTEGER PRIMARY KEY) need the
-	// generated value; a composite key is never generated.
-	if columns := schema.PrimaryKeyList(); len(columns) == 1 {
-		if value, ok := normalizedRow[columns[0]]; ok {
-			row[columns[0]] = value
+	// 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