소스 검색

fix(pgserver): return SQLSTATE 23505 for UNIQUE violations

Unique index violations now return a typed storage.UniqueViolationError
carrying the constraint name. The PostgreSQL wire server maps it to
SQLSTATE 23505 on both the simple and extended query paths and includes
the index name in the error message, so clients can use ON CONFLICT as an
idempotency primitive and distinguish conflicts from internal errors.
Danilo Fragoso 2 일 전
부모
커밋
401b0c4d59
5개의 변경된 파일과 157개의 추가작업 그리고 4개의 파일을 삭제
  1. 18 1
      pkg/pgserver/connection.go
  2. 99 0
      pkg/pgserver/features_test.go
  3. 1 0
      pkg/pgserver/protocol.go
  4. 14 3
      pkg/storage/table.go
  5. 25 0
      pkg/storage/unique_test.go

+ 18 - 1
pkg/pgserver/connection.go

@@ -320,13 +320,21 @@ func (c *Connection) handleQuery(msg *Message) error {
 		result, err := c.executor.Execute(stmt)
 		result, err := c.executor.Execute(stmt)
 		if err != nil {
 		if err != nil {
 			code := ErrCodeInternalError
 			code := ErrCodeInternalError
+			message := fmt.Sprintf("Execution error: %v", err)
+			var uniqueErr *storage.UniqueViolationError
 			if errors.Is(err, storage.ErrSerialization) {
 			if errors.Is(err, storage.ErrSerialization) {
 				code = ErrCodeSerializationFailure
 				code = ErrCodeSerializationFailure
 				c.txStatus = TxStatusIdle
 				c.txStatus = TxStatusIdle
+			} else if errors.As(err, &uniqueErr) {
+				code = ErrCodeUniqueViolation
+				message = fmt.Sprintf("Execution error: UNIQUE constraint failed: %s", uniqueErr.Constraint)
+				if c.txStatus == TxStatusInBlock {
+					c.txStatus = TxStatusFailed
+				}
 			} else if c.txStatus == TxStatusInBlock {
 			} else if c.txStatus == TxStatusInBlock {
 				c.txStatus = TxStatusFailed
 				c.txStatus = TxStatusFailed
 			}
 			}
-			c.sendError("ERROR", code, fmt.Sprintf("Execution error: %v", err))
+			c.sendError("ERROR", code, message)
 			return c.sendReadyForQuery()
 			return c.sendReadyForQuery()
 		}
 		}
 		if err := c.sendResult(result, stmt); err != nil {
 		if err := c.sendResult(result, stmt); err != nil {
@@ -489,12 +497,21 @@ func (c *Connection) handleExecute(msg *Message) error {
 	result, err := c.executor.Execute(stmt)
 	result, err := c.executor.Execute(stmt)
 	if err != nil {
 	if err != nil {
 		code := ErrCodeInternalError
 		code := ErrCodeInternalError
+		var uniqueErr *storage.UniqueViolationError
 		if errors.Is(err, storage.ErrSerialization) {
 		if errors.Is(err, storage.ErrSerialization) {
 			code = ErrCodeSerializationFailure
 			code = ErrCodeSerializationFailure
 			c.txStatus = TxStatusIdle
 			c.txStatus = TxStatusIdle
+		} else if errors.As(err, &uniqueErr) {
+			code = ErrCodeUniqueViolation
+			if c.txStatus == TxStatusInBlock {
+				c.txStatus = TxStatusFailed
+			}
 		} else if c.txStatus == TxStatusInBlock {
 		} else if c.txStatus == TxStatusInBlock {
 			c.txStatus = TxStatusFailed
 			c.txStatus = TxStatusFailed
 		}
 		}
+		if uniqueErr != nil {
+			return c.failExtended(code, fmt.Errorf("execution error: UNIQUE constraint failed: %s", uniqueErr.Constraint))
+		}
 		return c.failExtended(code, fmt.Errorf("execution error: %w", err))
 		return c.failExtended(code, fmt.Errorf("execution error: %w", err))
 	}
 	}
 	return c.sendResult(result, stmt)
 	return c.sendResult(result, stmt)

+ 99 - 0
pkg/pgserver/features_test.go

@@ -5,6 +5,7 @@ import (
 	"bytes"
 	"bytes"
 	"encoding/binary"
 	"encoding/binary"
 	"net"
 	"net"
+	"strings"
 	"testing"
 	"testing"
 
 
 	"github.com/danfragoso/pizzasql-next/pkg/executor"
 	"github.com/danfragoso/pizzasql-next/pkg/executor"
@@ -156,3 +157,101 @@ func TestGetCommandTagSelectUsesRowCount(t *testing.T) {
 		t.Fatalf("tag = %q, want SELECT 2", tag)
 		t.Fatalf("tag = %q, want SELECT 2", tag)
 	}
 	}
 }
 }
+
+// errorMessage extracts the human-readable message from an ErrorResponse.
+func errorMessage(msg *Message) string {
+	if msg == nil || msg.Type != MsgErrorResponse {
+		return ""
+	}
+	data := msg.Data
+	for i := 0; i < len(data); {
+		if data[i] == 0 {
+			break
+		}
+		field := data[i]
+		i++
+		end := bytes.IndexByte(data[i:], 0)
+		if end < 0 {
+			break
+		}
+		value := string(data[i : i+end])
+		if field == ErrorFieldMessage {
+			return value
+		}
+		i += end + 1
+	}
+	return ""
+}
+
+func TestSimpleQueryUniqueViolationErrorCode(t *testing.T) {
+	c, client := newDataConnection(t)
+	runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte("CREATE TABLE t (id INTEGER PRIMARY KEY, email TEXT UNIQUE)"), 0)})
+	runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte("INSERT INTO t VALUES (1, 'a@x')"), 0)})
+
+	msgs := runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte("INSERT INTO t VALUES (2, 'a@x')"), 0)})
+	if len(msgs) != 2 {
+		t.Fatalf("got %d messages, want ErrorResponse+ReadyForQuery", len(msgs))
+	}
+	if code := errorCode(msgs[0]); code != ErrCodeUniqueViolation {
+		t.Fatalf("error code = %q, want %q", code, ErrCodeUniqueViolation)
+	}
+	if message := errorMessage(msgs[0]); !strings.Contains(message, "uniq_t_email") {
+		t.Fatalf("error message = %q, want constraint uniq_t_email", message)
+	}
+}
+
+func TestExtendedQueryUniqueViolationErrorCode(t *testing.T) {
+	c, client := newDataConnection(t)
+	runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte("CREATE TABLE t (id INTEGER PRIMARY KEY, email TEXT UNIQUE)"), 0)})
+	runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte("INSERT INTO t VALUES (1, 'a@x')"), 0)})
+
+	parse := NewMessageBuilder()
+	parse.WriteString("")
+	parse.WriteString("INSERT INTO t VALUES (2, 'a@x')")
+	parse.WriteInt16(0)
+
+	bind := NewMessageBuilder()
+	bind.WriteString("")
+	bind.WriteString("")
+	bind.WriteInt16(0) // parameter format codes
+	bind.WriteInt16(0) // parameter values
+	bind.WriteInt16(0) // result format codes
+
+	execute := NewMessageBuilder()
+	execute.WriteString("")
+	execute.WriteInt32(0)
+
+	done := make(chan error, 1)
+	go func() {
+		if err := c.handleParse(&Message{Type: MsgParse, Data: parse.Bytes()}); err != nil {
+			done <- err
+			return
+		}
+		if err := c.handleBind(&Message{Type: MsgBind, Data: bind.Bytes()}); err != nil {
+			done <- err
+			return
+		}
+		done <- c.handleExecute(&Message{Type: MsgExecute, Data: execute.Bytes()})
+	}()
+
+	reader := bufio.NewReader(client)
+	var errMsg *Message
+	for errMsg == nil {
+		msg, err := ReadMessage(reader)
+		if err != nil {
+			t.Fatalf("read response: %v", err)
+		}
+		if msg.Type == MsgErrorResponse {
+			errMsg = msg
+		}
+	}
+	if err := <-done; err != nil {
+		t.Fatalf("extended query: %v", err)
+	}
+	if code := errorCode(errMsg); code != ErrCodeUniqueViolation {
+		t.Fatalf("error code = %q, want %q", code, ErrCodeUniqueViolation)
+	}
+	if message := errorMessage(errMsg); !strings.Contains(message, "uniq_t_email") {
+		t.Fatalf("error message = %q, want constraint uniq_t_email", message)
+	}
+}

+ 1 - 0
pkg/pgserver/protocol.go

@@ -92,6 +92,7 @@ const (
 	ErrCodeFeatureNotSupported  = "0A000"
 	ErrCodeFeatureNotSupported  = "0A000"
 	ErrCodeTransactionAborted   = "25P02"
 	ErrCodeTransactionAborted   = "25P02"
 	ErrCodeSerializationFailure = "40001"
 	ErrCodeSerializationFailure = "40001"
+	ErrCodeUniqueViolation      = "23505"
 )
 )
 
 
 // Message represents a PostgreSQL protocol message
 // Message represents a PostgreSQL protocol message

+ 14 - 3
pkg/storage/table.go

@@ -14,6 +14,17 @@ import (
 // Row represents a database row.
 // Row represents a database row.
 type Row map[string]interface{}
 type Row map[string]interface{}
 
 
+// UniqueViolationError reports a duplicate value on a UNIQUE index. The
+// constraint name identifies the violated index so protocol layers can surface a
+// machine-detectable error.
+type UniqueViolationError struct {
+	Constraint string
+}
+
+func (e *UniqueViolationError) Error() string {
+	return fmt.Sprintf("UNIQUE constraint failed: %s", e.Constraint)
+}
+
 // TableManager manages table data operations.
 // TableManager manages table data operations.
 type TableManager struct {
 type TableManager struct {
 	pool     *KVPool
 	pool     *KVPool
@@ -1804,7 +1815,7 @@ func (m *TableManager) validateUniqueRows(table string, pending []Row, excludedK
 				seen[idx.Name] = perIndex
 				seen[idx.Name] = perIndex
 			}
 			}
 			if perIndex[encoded] {
 			if perIndex[encoded] {
-				return fmt.Errorf("UNIQUE constraint failed: %s", idx.Name)
+				return &UniqueViolationError{Constraint: idx.Name}
 			}
 			}
 			perIndex[encoded] = true
 			perIndex[encoded] = true
 		}
 		}
@@ -1842,7 +1853,7 @@ func (m *TableManager) validateUniqueRows(table string, pending []Row, excludedK
 						continue
 						continue
 					}
 					}
 					if perIndex := seen[idx.Name]; perIndex != nil && perIndex[encoded] {
 					if perIndex := seen[idx.Name]; perIndex != nil && perIndex[encoded] {
-						return fmt.Errorf("UNIQUE constraint failed: %s", idx.Name)
+						return &UniqueViolationError{Constraint: idx.Name}
 					}
 					}
 				}
 				}
 			}
 			}
@@ -1915,7 +1926,7 @@ func (m *TableManager) ValidateUniqueIndex(index *Index) error {
 			return false, nil
 			return false, nil
 		}
 		}
 		if seen[encoded] {
 		if seen[encoded] {
-			return true, fmt.Errorf("UNIQUE constraint failed: %s", index.Name)
+			return true, &UniqueViolationError{Constraint: index.Name}
 		}
 		}
 		seen[encoded] = true
 		seen[encoded] = true
 		return false, nil
 		return false, nil

+ 25 - 0
pkg/storage/unique_test.go

@@ -1,6 +1,7 @@
 package storage
 package storage
 
 
 import (
 import (
+	"errors"
 	"fmt"
 	"fmt"
 	"strings"
 	"strings"
 	"sync"
 	"sync"
@@ -77,6 +78,30 @@ func TestUniqueIndexEnforcesOnInsert(t *testing.T) {
 	}
 	}
 }
 }
 
 
+func TestUniqueViolationErrorCarriesConstraint(t *testing.T) {
+	schemas, tables := uniqueTable(t,
+		Column{Name: "id", Type: "INTEGER", PrimaryKey: true},
+		Column{Name: "email", Type: "TEXT"},
+	)
+	if err := schemas.CreateIndex(&Index{Name: "uq_email", Table: "t", Unique: true, Columns: []IndexColumn{{Name: "email"}}}); err != nil {
+		t.Fatal(err)
+	}
+	if err := tables.Insert("t", Row{"id": int64(1), "email": "a@x"}); err != nil {
+		t.Fatal(err)
+	}
+	err := tables.Insert("t", Row{"id": int64(2), "email": "a@x"})
+	var violation *UniqueViolationError
+	if !errors.As(err, &violation) {
+		t.Fatalf("error %v is not a *UniqueViolationError", err)
+	}
+	if violation.Constraint != "uq_email" {
+		t.Fatalf("constraint = %q, want uq_email", violation.Constraint)
+	}
+	if err.Error() != "UNIQUE constraint failed: uq_email" {
+		t.Fatalf("message = %q, want UNIQUE constraint failed: uq_email", err.Error())
+	}
+}
+
 func TestUniqueIndexAllowsMultipleNulls(t *testing.T) {
 func TestUniqueIndexAllowsMultipleNulls(t *testing.T) {
 	schemas, tables := uniqueTable(t,
 	schemas, tables := uniqueTable(t,
 		Column{Name: "id", Type: "INTEGER", PrimaryKey: true},
 		Column{Name: "id", Type: "INTEGER", PrimaryKey: true},