fix: enforce NOT NULL on final inserted and updated rows (#17)

pkg/capabilities/capabilities.go

@@ -157,7 +157,7 @@ func Current() *Document {
 		},
 
 		Constraints: Constraints{
-			NotNull:       Feature{Supported: true, Status: "enforced"},
+			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"},
 			CompositePK:   Feature{Supported: false, Status: "rejected", SQLState: "0A000", Notes: "declare a single-column primary key"},
@@ -214,6 +214,7 @@ func Current() *Document {
 		},
 
 		ErrorCodes: map[string]string{
+			"23502": "not_null_violation",
 			"23505": "unique_violation",
 			"40001": "serialization_failure",
 			"0A000": "feature_not_supported",

pkg/executor/not_null_test.go

@@ -0,0 +1,44 @@
+package executor
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/danfragoso/pizzasql-next/pkg/storage"
+)
+
+func TestNotNullFinalRowValidation(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE t (id INTEGER PRIMARY KEY, v INTEGER NOT NULL DEFAULT 7, w INTEGER NOT NULL)")
+	execMust(t, e, "INSERT INTO t (id,w) VALUES (1,2)")
+	for _, explicit := range []bool{false, true} {
+		if explicit {
+			execMust(t, e, "BEGIN")
+		}
+		for _, sql := range []string{
+			"INSERT INTO t VALUES (2,NULL,3)",
+			"INSERT INTO t (id) VALUES (2)",
+			"INSERT INTO t VALUES (2,7,3),(3,7,NULL)",
+			"UPDATE t SET v=NULL WHERE id=1",
+			"UPDATE t SET w=NULL",
+			"INSERT INTO t VALUES (1,7,4) ON CONFLICT (id) DO UPDATE SET w=NULL",
+		} {
+			_, err := execSQL(e, sql)
+			var violation *storage.NotNullViolationError
+			if !errors.As(err, &violation) {
+				t.Fatalf("tx=%v %s: expected typed NOT NULL violation, got %v", explicit, sql, err)
+			}
+			rows := execMust(t, e, "SELECT id,v,w FROM t").Rows
+			if len(rows) != 1 || rows[0][1] != int64(7) || rows[0][2] != int64(2) {
+				t.Fatalf("failed mutation changed rows: %v", rows)
+			}
+		}
+		if explicit {
+			execMust(t, e, "COMMIT")
+		}
+	}
+	if _, err := table.InsertBulk("t", []storage.Row{{"id": int64(2), "v": nil, "w": int64(3)}}); err == nil {
+		t.Fatal("bulk insert accepted explicit NULL")
+	}
+}

pkg/pgserver/connection.go

@@ -410,9 +410,15 @@ func (c *Connection) handleQuery(msg *Message) error {
 			code := ErrCodeInternalError
 			message := fmt.Sprintf("Execution error: %v", err)
 			var uniqueErr *storage.UniqueViolationError
+			var notNullErr *storage.NotNullViolationError
 			if errors.Is(err, storage.ErrSerialization) {
 				code = ErrCodeSerializationFailure
 				c.txStatus = TxStatusIdle
+			} else if errors.As(err, &notNullErr) {
+				code = "23502"
+				if c.txStatus == TxStatusInBlock {
+					c.txStatus = TxStatusFailed
+				}
 			} else if errors.As(err, &uniqueErr) {
 				code = ErrCodeUniqueViolation
 				message = fmt.Sprintf("Execution error: UNIQUE constraint failed: %s", uniqueErr.Constraint)
@@ -610,9 +616,15 @@ func (c *Connection) handleExecute(msg *Message) error {
 	if err != nil {
 		code := ErrCodeInternalError
 		var uniqueErr *storage.UniqueViolationError
+		var notNullErr *storage.NotNullViolationError
 		if errors.Is(err, storage.ErrSerialization) {
 			code = ErrCodeSerializationFailure
 			c.txStatus = TxStatusIdle
+		} else if errors.As(err, &notNullErr) {
+			code = "23502"
+			if c.txStatus == TxStatusInBlock {
+				c.txStatus = TxStatusFailed
+			}
 		} else if errors.As(err, &uniqueErr) {
 			code = ErrCodeUniqueViolation
 			if _, commit := stmt.(*parser.CommitStmt); commit {

pkg/pgserver/not_null_test.go

@@ -0,0 +1,57 @@
+package pgserver
+
+import (
+	"bufio"
+	"net"
+	"testing"
+)
+
+func runExtendedExecution(t *testing.T, c *Connection, client net.Conn, sql string) []*Message {
+	t.Helper()
+	c.portals[""] = &portal{query: sql}
+	done := make(chan error, 1)
+	go func() {
+		if err := c.handleExecute(&Message{Type: MsgExecute, Data: []byte{0, 0, 0, 0, 0}}); err != nil {
+			done <- err
+			return
+		}
+		done <- c.handleMessage(&Message{Type: MsgSync})
+	}()
+	reader := bufio.NewReader(client)
+	var messages []*Message
+	for {
+		m, err := ReadMessage(reader)
+		if err != nil {
+			t.Fatal(err)
+		}
+		messages = append(messages, m)
+		if m.Type == MsgReadyForQuery {
+			break
+		}
+	}
+	if err := <-done; err != nil {
+		t.Fatal(err)
+	}
+	return messages
+}
+
+func TestNotNullSQLStateAndTransactionState(t *testing.T) {
+	for _, extended := range []bool{false, true} {
+		c, client := newDataConnection(t)
+		runQuery(t, c, client, &Message{Type: MsgQuery, Data: []byte("CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT NOT NULL DEFAULT 'x')\x00")})
+		runQuery(t, c, client, &Message{Type: MsgQuery, Data: []byte("BEGIN\x00")})
+		var messages []*Message
+		if extended {
+			messages = runExtendedExecution(t, c, client, "INSERT INTO t VALUES (1,NULL)")
+		} else {
+			messages = runQuery(t, c, client, &Message{Type: MsgQuery, Data: []byte("INSERT INTO t VALUES (1,NULL)\x00")})
+		}
+		if code := errorCode(messages[0]); code != "23502" {
+			t.Fatalf("extended=%v SQLSTATE=%q", extended, code)
+		}
+		if c.txStatus != TxStatusFailed {
+			t.Fatalf("extended=%v transaction not failed: %c", extended, c.txStatus)
+		}
+		runQuery(t, c, client, &Message{Type: MsgQuery, Data: []byte("ROLLBACK\x00")})
+	}
+}

pkg/storage/constraints.go

@@ -0,0 +1,22 @@
+package storage
+
+import "fmt"
+
+// NotNullViolationError identifies the final row that violates a required column.
+type NotNullViolationError struct {
+	Table  string
+	Column string
+}
+
+func (e *NotNullViolationError) Error() string {
+	return fmt.Sprintf("NOT NULL constraint failed: %s.%s", e.Table, e.Column)
+}
+
+func validateNotNull(schema *Schema, row Row) error {
+	for _, column := range schema.Columns {
+		if !column.Nullable && row[column.Name] == nil {
+			return &NotNullViolationError{Table: schema.Name, Column: column.Name}
+		}
+	}
+	return nil
+}

pkg/storage/table.go

@@ -512,24 +512,6 @@ func (m *TableManager) prepareInsert(table string, row Row) (Row, string, error)
 
 	pk := fmt.Sprintf("%v", pkValue)
 
-	for _, col := range schema.Columns {
-		if !col.Nullable && col.Default == nil {
-			val, hasVal := row[col.Name]
-			if !hasVal {
-				for k, v := range row {
-					if strings.EqualFold(k, col.Name) {
-						val = v
-						hasVal = true
-						break
-					}
-				}
-			}
-			if !hasVal || val == nil {
-				return nil, "", fmt.Errorf("missing required column: %s", col.Name)
-			}
-		}
-	}
-
 	normalizedRow := make(Row)
 	for _, col := range schema.Columns {
 		for k, v := range row {
@@ -555,6 +537,9 @@ func (m *TableManager) prepareInsert(table string, row Row) (Row, string, error)
 		normalizedRow["_rowid_"] = newRowID
 	}
 
+	if err := validateNotNull(schema, normalizedRow); err != nil {
+		return nil, "", err
+	}
 	return normalizedRow, m.dataKey(table, pk), nil
 }
 
@@ -756,6 +741,9 @@ func (m *TableManager) InsertBulkWithLastRowID(table string, rows []Row) (int, i
 			}
 		}
 		nr["_rowid_"] = rowid
+		if err := validateNotNull(schema, nr); err != nil {
+			return 0, 0, err
+		}
 		if rowid > maxRowID {
 			maxRowID = rowid
 		}
@@ -1062,6 +1050,9 @@ func (m *TableManager) UpdateByPK(table, pk string, updateFn func(Row) (Row, err
 		}
 	}
 
+	if err := validateNotNull(schema, row); err != nil {
+		return nil, false, err
+	}
 	data, err := encodeRow(row)
 	if err != nil {
 		return nil, false, err

pkg/storage/tx.go

@@ -719,6 +719,9 @@ func (s *Session) UpdateByPK(table, pk string, updateFn func(Row) (Row, error))
 		}
 	}
 
+	if err := validateNotNull(schema, row); err != nil {
+		return nil, false, err
+	}
 	s.stagePut(table, key, row)
 	return oldRow, true, nil
 }
@@ -809,6 +812,9 @@ func (s *Session) UpdateFunc(table string, updateFn func(Row) (Row, error), filt
 				}
 			}
 		}
+		if err := validateNotNull(schema, row); err != nil {
+			return count, err
+		}
 		newKey := s.table.dataKey(table, fmt.Sprintf("%v", row[schema.PrimaryKey]))
 		if newKey != oldKey {
 			// The primary key changed: move the staged row instead of leaving a