fix: report typed primary-key violations as SQLSTATE 23505 (#24)

pkg/capabilities/capabilities.go

@@ -161,7 +161,7 @@ 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", Notes: "single-column only"},
+			PrimaryKey:    Feature{Supported: true, Status: "enforced", SQLState: "23505", Notes: "single-column only"},
 			CompositePK:   Feature{Supported: false, Status: "rejected", SQLState: "0A000", Notes: "declare a single-column primary key"},
 			ForeignKey:    Feature{Supported: false, Status: "rejected", SQLState: "0A000", Notes: "inline and table-level FOREIGN KEY are rejected; enforce referential integrity in the application, cascades are not implemented"},
 			Check:         Feature{Supported: false, Status: "rejected", SQLState: "0A000", Notes: "inline and table-level CHECK are rejected; enforce the rule in the application"},

pkg/executor/executor.go

@@ -3398,7 +3398,7 @@ func (e *Executor) executeUpdateFrom(stmt *parser.UpdateStmt, schema *storage.Sc
 			newPK := fmt.Sprintf("%v", newPKVal)
 			if newPK != plan.oldPK {
 				if _, gerr := e.session.GetByPK(tableName, newPK); gerr == nil {
-					return fmt.Errorf("duplicate primary key: %v", newPK)
+					return &storage.PrimaryKeyViolationError{Table: tableName, Key: newPK}
 				} else if gerr != storage.ErrKeyNotFound {
 					return gerr
 				}

pkg/pgserver/connection.go

@@ -410,6 +410,7 @@ func (c *Connection) handleQuery(msg *Message) error {
 		if err != nil {
 			code := ErrCodeInternalError
 			message := fmt.Sprintf("Execution error: %v", err)
+			var primaryKeyErr *storage.PrimaryKeyViolationError
 			var uniqueErr *storage.UniqueViolationError
 			var notNullErr *storage.NotNullViolationError
 			if errors.Is(err, storage.ErrSerialization) {
@@ -420,9 +421,11 @@ func (c *Connection) handleQuery(msg *Message) error {
 				if c.txStatus == TxStatusInBlock {
 					c.txStatus = TxStatusFailed
 				}
-			} else if errors.As(err, &uniqueErr) {
+			} else if errors.As(err, &uniqueErr) || errors.As(err, &primaryKeyErr) {
 				code = ErrCodeUniqueViolation
-				message = fmt.Sprintf("Execution error: UNIQUE constraint failed: %s", uniqueErr.Constraint)
+				if uniqueErr != nil {
+					message = fmt.Sprintf("Execution error: UNIQUE constraint failed: %s", uniqueErr.Constraint)
+				}
 				if _, commit := stmt.(*parser.CommitStmt); commit {
 					c.txStatus = TxStatusIdle
 				} else if c.txStatus == TxStatusInBlock {
@@ -616,6 +619,7 @@ func (c *Connection) handleExecute(msg *Message) error {
 	result, err := c.executor.Execute(stmt)
 	if err != nil {
 		code := ErrCodeInternalError
+		var primaryKeyErr *storage.PrimaryKeyViolationError
 		var uniqueErr *storage.UniqueViolationError
 		var notNullErr *storage.NotNullViolationError
 		if errors.Is(err, storage.ErrSerialization) {
@@ -626,7 +630,7 @@ func (c *Connection) handleExecute(msg *Message) error {
 			if c.txStatus == TxStatusInBlock {
 				c.txStatus = TxStatusFailed
 			}
-		} else if errors.As(err, &uniqueErr) {
+		} else if errors.As(err, &uniqueErr) || errors.As(err, &primaryKeyErr) {
 			code = ErrCodeUniqueViolation
 			if _, commit := stmt.(*parser.CommitStmt); commit {
 				c.txStatus = TxStatusIdle

pkg/pgserver/primary_key_test.go

@@ -0,0 +1,46 @@
+package pgserver
+
+import (
+	"fmt"
+	"testing"
+)
+
+func TestPrimaryKeySQLStateAcrossMutationAndProtocolPaths(t *testing.T) {
+	for _, extended := range []bool{false, true} {
+		for _, transaction := range []bool{false, true} {
+			for _, sql := range []string{"INSERT INTO t VALUES (1)", "INSERT INTO t VALUES (3),(3)", "UPDATE t SET id=2 WHERE id=1"} {
+				t.Run(fmt.Sprintf("extended=%v/transaction=%v/%s", extended, transaction, sql), func(t *testing.T) {
+					c, client := newDataConnection(t)
+					runQuery(t, c, client, &Message{Type: MsgQuery, Data: []byte("CREATE TABLE t (id INTEGER PRIMARY KEY)\x00")})
+					runQuery(t, c, client, &Message{Type: MsgQuery, Data: []byte("INSERT INTO t VALUES (1),(2)\x00")})
+					if transaction {
+						runQuery(t, c, client, &Message{Type: MsgQuery, Data: []byte("BEGIN\x00")})
+					}
+					var messages []*Message
+					if extended {
+						messages = runExtendedExecution(t, c, client, sql)
+					} else {
+						messages = runQuery(t, c, client, &Message{Type: MsgQuery, Data: []byte(sql + "\x00")})
+					}
+					if code := errorCode(messages[0]); code != ErrCodeUniqueViolation {
+						t.Fatalf("SQLSTATE=%s want 23505", code)
+					}
+					if transaction {
+						if c.txStatus != TxStatusFailed {
+							t.Fatalf("transaction status %c want E", c.txStatus)
+						}
+						blocked := runQuery(t, c, client, &Message{Type: MsgQuery, Data: []byte("SELECT 1\x00")})
+						if code := errorCode(blocked[0]); code != ErrCodeTransactionAborted {
+							t.Fatalf("aborted SQLSTATE=%s", code)
+						}
+						runQuery(t, c, client, &Message{Type: MsgQuery, Data: []byte("ROLLBACK\x00")})
+					}
+					count := runQuery(t, c, client, &Message{Type: MsgQuery, Data: []byte("SELECT COUNT(*) FROM t\x00")})
+					if values := dataRowValues(t, count[1]); len(values) != 1 || values[0] != "2" {
+						t.Fatalf("failed statement changed rows: %v", values)
+					}
+				})
+			}
+		}
+	}
+}

pkg/storage/constraints.go

@@ -2,6 +2,17 @@ package storage
 
 import "fmt"
 
+// PrimaryKeyViolationError preserves duplicate-key identity through wrapped
+// executor errors so wire adapters can report integrity SQLSTATE 23505.
+type PrimaryKeyViolationError struct {
+	Table string
+	Key   string
+}
+
+func (e *PrimaryKeyViolationError) Error() string {
+	return fmt.Sprintf("duplicate primary key: %s", e.Key)
+}
+
 // NotNullViolationError identifies the final row that violates a required column.
 type NotNullViolationError struct {
 	Table  string

pkg/storage/table.go

@@ -609,7 +609,7 @@ func (m *TableManager) InsertWithRowID(table string, row Row) (int64, error) {
 		return 0, err
 	}
 	if !committed {
-		return 0, fmt.Errorf("duplicate primary key: %v", row[schemaPrimaryKey(m.schema, table)])
+		return 0, &PrimaryKeyViolationError{Table: table, Key: strings.TrimPrefix(key, m.dataPrefix(table))}
 	}
 
 	m.applyUniquePlan(uniquePlan, nil)
@@ -629,14 +629,6 @@ func (m *TableManager) InsertWithRowID(table string, row Row) (int64, error) {
 	return rowid, nil
 }
 
-func schemaPrimaryKey(s *SchemaManager, table string) string {
-	schema, err := s.GetSchema(table)
-	if err != nil {
-		return "_rowid_"
-	}
-	return schema.PrimaryKey
-}
-
 // bulkBatchByteBudget bounds a single atomic BATCH_WRITE payload below the
 // PKBFI frame limit so a bulk insert never emits a frame the server rejects.
 // Each op contributes 12 header bytes plus its key and value.
@@ -791,7 +783,7 @@ func (m *TableManager) InsertBulkWithLastRowID(table string, rows []Row) (int, i
 	for i, op := range ops {
 		key := string(op.Key)
 		if _, duplicate := seen[key]; duplicate {
-			return 0, 0, fmt.Errorf("duplicate primary key: %s", key)
+			return 0, 0, &PrimaryKeyViolationError{Table: table, Key: key}
 		}
 		seen[key] = struct{}{}
 		keys[i] = op.Key
@@ -815,7 +807,7 @@ func (m *TableManager) InsertBulkWithLastRowID(table string, rows []Row) (int, i
 	}
 	for i, found := range existing {
 		if found {
-			return 0, 0, fmt.Errorf("duplicate primary key: %s", ops[i].Key)
+			return 0, 0, &PrimaryKeyViolationError{Table: table, Key: string(ops[i].Key)}
 		}
 	}
 

pkg/storage/tx.go

@@ -704,13 +704,13 @@ func (s *Session) insertLocked(table string, row Row) (int64, error) {
 
 	if e, ok := s.overlay[tl][key]; ok {
 		if !e.absent {
-			return 0, fmt.Errorf("duplicate primary key: %s", pk)
+			return 0, &PrimaryKeyViolationError{Table: table, Key: pk}
 		}
 	} else {
 		_, lsn, err := s.table.getByPKWithLSN(table, pk)
 		if err == nil {
 			s.reads[key] = lsn
-			return 0, fmt.Errorf("duplicate primary key: %s", pk)
+			return 0, &PrimaryKeyViolationError{Table: table, Key: pk}
 		}
 		if err != ErrKeyNotFound {
 			return 0, err
@@ -860,7 +860,7 @@ func (s *Session) UpdateFunc(table string, updateFn func(Row) (Row, error), filt
 			// 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, fmt.Errorf("duplicate primary key: %v", row[schema.PrimaryKey])
+				return count, &PrimaryKeyViolationError{Table: table, Key: fmt.Sprint(row[schema.PrimaryKey])}
 			} else if gerr != ErrKeyNotFound {
 				return count, gerr
 			}