feat: backfill ADD COLUMN constant defaults into existing rows

ALTER TABLE ... ADD COLUMN with a constant DEFAULT now writes the value
into every existing row instead of only recording catalog metadata. Rows
that predate the statement therefore read the default, matching the value
later INSERTs get when they omit the column.

The rewrite happens at DDL time rather than being materialized on read
because every integrity check in the engine consumes persisted row
values: CREATE UNIQUE INDEX validation, index scans, NOT NULL validation
on writes, and the ON CONFLICT paths. A read-time default would leave the
new column out of those checks, so a UNIQUE index over it could not see
the duplicate values the backfill creates. This is the same approach
DROP COLUMN and RENAME COLUMN already use.

Cost: one full table scan plus re-encode per ADD COLUMN with a default,
under the table's exclusive gate and the catalog lock, so the statement
stalls that table (and catalog lookups) for its duration. A nullable
column without a default is still metadata-only, and the transform is
idempotent so an interrupted rewrite resumes against the old schema.

Now that the value is durable, the constraints can be validated against
it:
- ADD COLUMN ... NOT NULL without a DEFAULT fails with SQLSTATE 23502 on
  a non-empty table (empty tables are still accepted).
- ADD COLUMN ... UNIQUE is enforced: it materializes a unique index using
  the CREATE TABLE naming rules, and backfilling one non-NULL constant
  into more than one row fails with 23505 before any durable change. A
  failed index build drops the column again instead of leaving a
  half-applied ALTER.
- IF NOT EXISTS stays a no-op when the column is already present.

The HTTP surface now maps NOT NULL violations to 23502 like the
PostgreSQL endpoint, and the capability document advertises
add_column_backfill as rewrites_rows.

pkg/capabilities/capabilities.go

@@ -175,8 +175,8 @@ func Current() *Document {
 		},
 
 		DDL: DDL{
-			AddColumn:              Feature{Supported: true, Status: "metadata_only"},
-			AddColumnBackfill:      Feature{Supported: false, Status: "no_backfill", Notes: "existing rows read NULL; a DEFAULT applies to later INSERTs only"},
+			AddColumn:              Feature{Supported: true, Status: "materializes_constant_default", Notes: "adds one nullable column; a constant DEFAULT is written into every existing row, so the rewrite cost is O(rows) and holds the table's exclusive gate"},
+			AddColumnBackfill:      Feature{Supported: true, Status: "rewrites_rows", Notes: "a constant DEFAULT is materialized into existing rows at DDL time; ADD COLUMN ... NOT NULL without a DEFAULT fails with 23502 on a non-empty table and ADD COLUMN ... UNIQUE conflicts with 23505 when backfilling a non-NULL default into more than one row"},
 			DropColumn:             Feature{Supported: true, Status: "rewrites_rows"},
 			RenameColumn:           Feature{Supported: true, Status: "rewrites_rows"},
 			RenameColumnIndexes:    Feature{Supported: false, Status: "stale_index_metadata", Notes: "drop and recreate indexes that reference a renamed column"},

pkg/capabilities/capabilities_test.go

@@ -40,3 +40,16 @@ func TestCurrentDocumentIsVersionedAndMarshalable(t *testing.T) {
 		t.Fatalf("unexpected notification limits: %+v", round.Notifications)
 	}
 }
+
+func TestAddColumnBackfillIsAdvertised(t *testing.T) {
+	doc := Current()
+	if !doc.DDL.AddColumnBackfill.Supported {
+		t.Fatalf("add_column_backfill = %+v, want supported", doc.DDL.AddColumnBackfill)
+	}
+	if doc.DDL.AddColumnBackfill.Status != "rewrites_rows" {
+		t.Fatalf("add_column_backfill status = %q, want rewrites_rows", doc.DDL.AddColumnBackfill.Status)
+	}
+	if doc.DDL.AddColumn.Status == "metadata_only" {
+		t.Fatalf("add_column status = %q, but a constant DEFAULT rewrites rows", doc.DDL.AddColumn.Status)
+	}
+}

pkg/executor/alter_table_test.go

@@ -1,13 +1,16 @@
 package executor
 
 import (
+	"errors"
 	"sync"
 	"testing"
+
+	"github.com/danfragoso/pizzasql-next/pkg/storage"
 )
 
-// ADD COLUMN only appends schema metadata. Existing rows are not rewritten, so
-// a declared DEFAULT applies to later INSERTs only.
-func TestAlterTableAddColumnDoesNotBackfillPopulatedTable(t *testing.T) {
+// ADD COLUMN with a constant DEFAULT materializes the value into rows that
+// already exist, so the column reads the default everywhere.
+func TestAlterTableAddColumnBackfillsPopulatedTable(t *testing.T) {
 	_, schema, table := newTestDB(t)
 	e := newExec(schema, table)
 	execMust(t, e, "CREATE TABLE t (id INTEGER PRIMARY KEY)")
@@ -20,8 +23,8 @@ func TestAlterTableAddColumnDoesNotBackfillPopulatedTable(t *testing.T) {
 		t.Fatalf("row count = %d, want 2", res.RowCount)
 	}
 	for _, row := range res.Rows {
-		if row[1] != nil {
-			t.Fatalf("existing row %v note = %v, want NULL (no backfill)", row[0], row[1])
+		if row[1] != "later" {
+			t.Fatalf("existing row %v note = %v, want later", row[0], row[1])
 		}
 	}
 
@@ -32,6 +35,204 @@ func TestAlterTableAddColumnDoesNotBackfillPopulatedTable(t *testing.T) {
 	}
 }
 
+// A DEFAULT declared on an empty table has no rows to rewrite; later INSERTs
+// that omit the column take it, and an explicit NULL still wins.
+func TestAlterTableAddColumnDefaultOnEmptyTable(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE t (id INTEGER PRIMARY KEY)")
+
+	execMust(t, e, "ALTER TABLE t ADD COLUMN note TEXT DEFAULT 'later'")
+
+	execMust(t, e, "INSERT INTO t (id) VALUES (1)")
+	execMust(t, e, "INSERT INTO t (id, note) VALUES (2, NULL)")
+	res := execMust(t, e, "SELECT id, note FROM t ORDER BY id")
+	if len(res.Rows) != 2 || res.Rows[0][1] != "later" || res.Rows[1][1] != nil {
+		t.Fatalf("rows = %v, want [[1 later] [2 NULL]]", res.Rows)
+	}
+}
+
+// ADD COLUMN without a DEFAULT leaves existing rows NULL.
+func TestAlterTableAddColumnWithoutDefaultLeavesExistingRowsNull(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE t (id INTEGER PRIMARY KEY)")
+	execMust(t, e, "INSERT INTO t (id) VALUES (1), (2)")
+
+	execMust(t, e, "ALTER TABLE t ADD COLUMN note TEXT")
+
+	res := execMust(t, e, "SELECT id, note FROM t ORDER BY id")
+	if res.RowCount != 2 {
+		t.Fatalf("row count = %d, want 2", res.RowCount)
+	}
+	for _, row := range res.Rows {
+		if row[1] != nil {
+			t.Fatalf("existing row %v note = %v, want NULL", row[0], row[1])
+		}
+	}
+}
+
+// A NOT NULL column with a constant DEFAULT backfills existing rows, so the
+// constraint holds for rows that predate the statement.
+func TestAlterTableAddColumnNotNullWithDefaultBackfills(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE t (id INTEGER PRIMARY KEY)")
+	execMust(t, e, "INSERT INTO t (id) VALUES (1), (2)")
+
+	execMust(t, e, "ALTER TABLE t ADD COLUMN status TEXT NOT NULL DEFAULT 'pending'")
+
+	fresh := newExec(schema, table)
+	res := execMust(t, fresh, "SELECT status FROM t ORDER BY id")
+	if res.RowCount != 2 {
+		t.Fatalf("row count = %d, want 2", res.RowCount)
+	}
+	for _, row := range res.Rows {
+		if row[0] != "pending" {
+			t.Fatalf("existing status = %v, want pending", row[0])
+		}
+	}
+
+	execMust(t, e, "INSERT INTO t (id) VALUES (3)")
+	res = execMust(t, e, "SELECT status FROM t WHERE id = 3")
+	if got := res.Rows[0][0]; got != "pending" {
+		t.Fatalf("new row status = %v, want pending", got)
+	}
+}
+
+// A NOT NULL column without a DEFAULT has nothing to write into existing rows
+// and must not report the table as migrated.
+func TestAlterTableAddColumnNotNullWithoutDefaultRejectsPopulatedTable(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE t (id INTEGER PRIMARY KEY)")
+	execMust(t, e, "INSERT INTO t (id) VALUES (1)")
+
+	_, err := execSQL(e, "ALTER TABLE t ADD COLUMN status TEXT NOT NULL")
+	var violation *storage.NotNullViolationError
+	if !errors.As(err, &violation) {
+		t.Fatalf("error = %v, want NotNullViolationError", err)
+	}
+
+	s, err := schema.GetSchema("t")
+	if err != nil {
+		t.Fatalf("get schema: %v", err)
+	}
+	if _, exists := s.GetColumn("status"); exists {
+		t.Fatal("rejected ADD COLUMN left the column in the schema")
+	}
+	if rows := execMust(t, e, "SELECT id FROM t").Rows; len(rows) != 1 {
+		t.Fatalf("rows = %v, want the existing row untouched", rows)
+	}
+}
+
+// The same statement is legal on an empty table; the constraint is then
+// enforced on the INSERTs that follow.
+func TestAlterTableAddColumnNotNullWithoutDefaultOnEmptyTable(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE t (id INTEGER PRIMARY KEY)")
+
+	execMust(t, e, "ALTER TABLE t ADD COLUMN status TEXT NOT NULL")
+
+	_, err := execSQL(e, "INSERT INTO t (id) VALUES (1)")
+	var violation *storage.NotNullViolationError
+	if !errors.As(err, &violation) {
+		t.Fatalf("insert without status error = %v, want NotNullViolationError", err)
+	}
+
+	execMust(t, e, "INSERT INTO t (id, status) VALUES (1, 'active')")
+	res := execMust(t, e, "SELECT status FROM t WHERE id = 1")
+	if got := res.Rows[0][0]; got != "active" {
+		t.Fatalf("status = %v, want active", got)
+	}
+}
+
+// Backfilling one constant into several rows would duplicate the value, so the
+// UNIQUE constraint must reject the ALTER instead of creating a silently
+// non-unique column.
+func TestAlterTableAddColumnUniqueWithDefaultRejectsPopulatedTable(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE t (id INTEGER PRIMARY KEY)")
+	execMust(t, e, "INSERT INTO t (id) VALUES (1), (2)")
+
+	_, err := execSQL(e, "ALTER TABLE t ADD COLUMN code TEXT DEFAULT 'same' UNIQUE")
+	var violation *storage.UniqueViolationError
+	if !errors.As(err, &violation) {
+		t.Fatalf("error = %v, want UniqueViolationError", err)
+	}
+
+	s, err := schema.GetSchema("t")
+	if err != nil {
+		t.Fatalf("get schema: %v", err)
+	}
+	if _, exists := s.GetColumn("code"); exists {
+		t.Fatal("rejected ADD COLUMN left the column in the schema")
+	}
+	indexes, err := schema.ListTableIndexes("t")
+	if err != nil {
+		t.Fatalf("list indexes: %v", err)
+	}
+	if len(indexes) != 0 {
+		t.Fatalf("indexes = %v, want none", indexes)
+	}
+}
+
+// A single-row table can take a UNIQUE column with a default, and the
+// constraint is enforced for every later INSERT that omits the column.
+func TestAlterTableAddColumnUniqueWithDefaultEnforcesLaterInserts(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE t (id INTEGER PRIMARY KEY)")
+	execMust(t, e, "INSERT INTO t (id) VALUES (1)")
+
+	execMust(t, e, "ALTER TABLE t ADD COLUMN code TEXT DEFAULT 'same' UNIQUE")
+
+	_, err := execSQL(e, "INSERT INTO t (id) VALUES (2)")
+	var violation *storage.UniqueViolationError
+	if !errors.As(err, &violation) {
+		t.Fatalf("insert with duplicated default error = %v, want UniqueViolationError", err)
+	}
+
+	execMust(t, e, "INSERT INTO t (id, code) VALUES (2, 'other')")
+	res := execMust(t, e, "SELECT code FROM t ORDER BY id")
+	if len(res.Rows) != 2 || res.Rows[0][0] != "same" || res.Rows[1][0] != "other" {
+		t.Fatalf("codes = %v, want [same other]", res.Rows)
+	}
+}
+
+// IF NOT EXISTS stays idempotent on a populated table: the second statement is
+// a no-op even when it declares a different default.
+func TestAlterTableAddColumnIfNotExistsKeepsBackfilledValues(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	e := newExec(schema, table)
+	execMust(t, e, "CREATE TABLE t (id INTEGER PRIMARY KEY)")
+	execMust(t, e, "INSERT INTO t (id) VALUES (1)")
+
+	execMust(t, e, "ALTER TABLE t ADD COLUMN IF NOT EXISTS note TEXT DEFAULT 'first'")
+	execMust(t, e, "ALTER TABLE t ADD COLUMN IF NOT EXISTS note TEXT DEFAULT 'second'")
+
+	res := execMust(t, e, "SELECT note FROM t WHERE id = 1")
+	if got := res.Rows[0][0]; got != "first" {
+		t.Fatalf("note = %v, want the first default to stand", got)
+	}
+
+	s, err := schema.GetSchema("t")
+	if err != nil {
+		t.Fatalf("get schema: %v", err)
+	}
+	count := 0
+	for _, col := range s.Columns {
+		if col.Name == "note" {
+			count++
+		}
+	}
+	if count != 1 {
+		t.Fatalf("column note appears %d times, want 1", count)
+	}
+}
+
 func TestAlterTableAddColumnIfNotExistsIsIdempotent(t *testing.T) {
 	_, schema, table := newTestDB(t)
 	e := newExec(schema, table)

pkg/executor/ddl_defaults_test.go

@@ -59,7 +59,7 @@ func TestConstantDefaultsStillApplyOnInsert(t *testing.T) {
 	}
 }
 
-func TestAlterTableAddColumnDoesNotBackfill(t *testing.T) {
+func TestAlterTableAddColumnBackfillsExistingRows(t *testing.T) {
 	_, schema, table := newTestDB(t)
 	e := newExec(schema, table)
 	execMust(t, e, "CREATE TABLE t (id INTEGER PRIMARY KEY)")
@@ -71,8 +71,8 @@ func TestAlterTableAddColumnDoesNotBackfill(t *testing.T) {
 	if res.RowCount != 1 {
 		t.Fatalf("row count = %d, want 1", res.RowCount)
 	}
-	if got := res.Rows[0][1]; got != nil {
-		t.Fatalf("existing row status = %v, want NULL (ADD COLUMN does not backfill)", got)
+	if got := res.Rows[0][1]; got != "pending" {
+		t.Fatalf("existing row status = %v, want pending (ADD COLUMN backfills)", got)
 	}
 
 	execMust(t, e, "INSERT INTO t (id) VALUES (2)")

pkg/executor/executor.go

@@ -4071,22 +4071,28 @@ func (e *Executor) executeAlterTable(stmt *parser.AlterTableStmt) (*Result, erro
 	}
 }
 
-// executeAlterTableAddColumn adds a column to a table.
+// executeAlterTableAddColumn adds a column to a table. A constant DEFAULT is
+// written into every existing row (see SchemaManager.AddColumn), so the column
+// reads the default on rows that predate the statement as well as on later
+// INSERTs that omit it.
 func (e *Executor) executeAlterTableAddColumn(table string, action *parser.AddColumnAction) (*Result, error) {
-	if action.IfNotExists {
-		schema, err := e.schema.GetSchema(table)
-		if err != nil {
-			return nil, err
-		}
-		if _, exists := schema.GetColumn(action.Column.Name); exists {
+	schema, err := e.schema.GetSchema(table)
+	if err != nil {
+		return nil, err
+	}
+	if _, exists := schema.GetColumn(action.Column.Name); exists {
+		if action.IfNotExists {
 			return NewResult("ALTER TABLE"), nil
 		}
+		return nil, fmt.Errorf("column already exists: %s", action.Column.Name)
 	}
+
 	col := storage.Column{
 		Name:     action.Column.Name,
 		Type:     action.Column.Type.Name,
 		Nullable: true,
 	}
+	var uniqueConstraint *parser.ColumnConstraint
 
 	// Process column constraints
 	for _, constraint := range action.Column.Constraints {
@@ -4096,6 +4102,9 @@ func (e *Executor) executeAlterTableAddColumn(table string, action *parser.AddCo
 			col.Nullable = false
 		case parser.ConstraintNotNull:
 			col.Nullable = false
+		case parser.ConstraintUnique:
+			unique := constraint
+			uniqueConstraint = &unique
 		case parser.ConstraintDefault:
 			if constraint.Default != nil {
 				val, err := constantDefaultValue(e, col.Name, constraint.Default)
@@ -4117,6 +4126,37 @@ func (e *Executor) executeAlterTableAddColumn(table string, action *parser.AddCo
 		}
 	}
 
+	var uniqueIndex *storage.Index
+	if uniqueConstraint != nil {
+		uniqueIndex = e.addColumnUniqueIndex(table, col.Name, uniqueConstraint)
+	}
+
+	// A required column with no default has no value for rows that already
+	// exist. Refusing before the rewrite leaves the table untouched instead of
+	// failing NOT NULL validation on a later read or write.
+	if !col.Nullable && col.Default == nil {
+		existingRows, err := e.session.CountFast(table)
+		if err != nil {
+			return nil, err
+		}
+		if existingRows > 0 {
+			return nil, &storage.NotNullViolationError{Table: table, Column: col.Name}
+		}
+	}
+
+	// Every existing row receives the same constant, so a non-NULL default
+	// duplicates immediately once the table holds more than one row. Check
+	// before rewriting so a rejected UNIQUE column leaves no durable changes.
+	if uniqueIndex != nil && col.Default != nil {
+		rows, err := e.session.CountFast(table)
+		if err != nil {
+			return nil, err
+		}
+		if rows > 1 {
+			return nil, &storage.UniqueViolationError{Constraint: uniqueIndex.Name}
+		}
+	}
+
 	if err := e.schema.AddColumn(table, col); err != nil {
 		if action.IfNotExists && strings.Contains(err.Error(), "column already exists") {
 			return NewResult("ALTER TABLE"), nil
@@ -4127,10 +4167,69 @@ func (e *Executor) executeAlterTableAddColumn(table string, action *parser.AddCo
 	// Update catalog
 	e.SyncCatalog()
 
+	if uniqueIndex != nil {
+		if err := e.createAddedColumnUniqueIndex(uniqueIndex); err != nil {
+			return nil, err
+		}
+	}
+
 	result := NewResult("ALTER TABLE")
 	return result, nil
 }
 
+// addColumnUniqueIndex names the index an ADD COLUMN ... UNIQUE constraint
+// materializes, using the declared constraint name when there is one and the
+// same deterministic internal name CREATE TABLE gives an unnamed column-level
+// UNIQUE constraint otherwise, suffixing collisions.
+func (e *Executor) addColumnUniqueIndex(table, column string, constraint *parser.ColumnConstraint) *storage.Index {
+	base := "uniq_" + strings.ToLower(table) + "_" + strings.ToLower(column)
+	if constraint.Name != "" {
+		base = constraint.Name
+	}
+	name := base
+	for n := 2; e.schema.IndexExists(name); n++ {
+		name = fmt.Sprintf("%s_%d", base, n)
+	}
+	index := &storage.Index{
+		Name:    name,
+		Table:   table,
+		Unique:  true,
+		Columns: []storage.IndexColumn{{Name: column}},
+	}
+	if constraint.HasOnConflict {
+		index.OnConflict = conflictActionName(constraint.OnConflict)
+	}
+	return index
+}
+
+// createAddedColumnUniqueIndex registers and populates the unique index for an
+// ADD COLUMN. Uniqueness is validated after the backfill has written the new
+// values, so the check sees what later INSERTs and index scans will see. DDL is
+// not transactional, but a failure here drops the column again (best effort)
+// instead of leaving a half-applied ADD COLUMN that a retry would reject as an
+// existing column.
+func (e *Executor) createAddedColumnUniqueIndex(index *storage.Index) error {
+	if err := e.table.CreateUniqueIndex(index); err != nil {
+		e.rollbackAddedColumn(index.Table, index.Columns[0].Name)
+		return err
+	}
+	columns := []string{index.Columns[0].Name}
+	if err := e.session.BuildIndex(index.Name, index.Table, columns); err != nil {
+		e.schema.DropIndex(index.Name)
+		e.rollbackAddedColumn(index.Table, index.Columns[0].Name)
+		return fmt.Errorf("failed to build index: %w", err)
+	}
+	return nil
+}
+
+// rollbackAddedColumn removes a column whose unique index could not be built.
+func (e *Executor) rollbackAddedColumn(table, column string) {
+	if err := e.schema.DropColumn(table, column); err != nil {
+		return
+	}
+	e.SyncCatalog()
+}
+
 // executeAlterTableDropColumn drops a column from a table.
 func (e *Executor) executeAlterTableDropColumn(table string, action *parser.DropColumnAction) (*Result, error) {
 	if err := e.schema.DropColumn(table, action.Column); err != nil {

pkg/httpserver/handler.go

@@ -423,8 +423,8 @@ func (s *Server) handleVersion(w http.ResponseWriter, r *http.Request) {
 
 // sqlStateHTTPError maps engine errors onto the same SQLSTATE codes the
 // PostgreSQL endpoint reports. The executor returns plain errors on released
-// builds, so the unique-violation and unsupported-feature markers are matched
-// on their stable message prefixes.
+// builds, so the unique-violation, not-null, and unsupported-feature markers are
+// matched on their stable message prefixes.
 func sqlStateHTTPError(err error) *HTTPError {
 	if err == nil {
 		return nil
@@ -439,6 +439,13 @@ func sqlStateHTTPError(err error) *HTTPError {
 			Details: map[string]interface{}{"constraint": constraint},
 		}
 	}
+	if strings.HasPrefix(message, "NOT NULL constraint failed: ") {
+		return &HTTPError{
+			Status:  http.StatusBadRequest,
+			Code:    "23502",
+			Message: message,
+		}
+	}
 	if errors.Is(err, storage.ErrSerialization) {
 		return &HTTPError{
 			Status:  http.StatusConflict,

pkg/httpserver/sqlstate_test.go

@@ -54,6 +54,7 @@ func TestSQLStateHTTPErrorMapping(t *testing.T) {
 		code   string
 	}{
 		{"unique", fmt.Errorf("UNIQUE constraint failed: uniq_users_email"), http.StatusConflict, "23505"},
+		{"not null", fmt.Errorf("NOT NULL constraint failed: t.v"), http.StatusBadRequest, "23502"},
 		{"serialization", fmt.Errorf("commit: %w", storage.ErrSerialization), http.StatusConflict, "40001"},
 		{"unsupported", fmt.Errorf("feature not supported: DEFAULT is not constant"), http.StatusBadRequest, "0A000"},
 		{"other", errors.New("boom"), http.StatusInternalServerError, "EXECUTION_ERROR"},

pkg/pgserver/alter_table_test.go

@@ -15,8 +15,8 @@ func TestAlterTableAddColumnOverSimpleQuery(t *testing.T) {
 
 	msgs := runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte("SELECT id, note FROM t ORDER BY id"), 0)})
 	values := dataRowValues(t, msgs[1])
-	if len(values) != 2 || values[0] != "1" || values[1] != "<null>" {
-		t.Fatalf("existing row after ADD COLUMN = %v, want [1 <null>]", values)
+	if len(values) != 2 || values[0] != "1" || values[1] != "later" {
+		t.Fatalf("existing row after ADD COLUMN = %v, want [1 later]", values)
 	}
 
 	runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte("INSERT INTO t (id) VALUES (2)"), 0)})
@@ -26,6 +26,38 @@ func TestAlterTableAddColumnOverSimpleQuery(t *testing.T) {
 	}
 }
 
+func TestAlterTableAddColumnErrorsOverSimpleQuery(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 (id) VALUES (1), (2)\x00")})
+
+	// A required column with no default cannot be materialized for existing
+	// rows.
+	msgs := runQuery(t, c, client, &Message{Type: MsgQuery, Data: []byte("ALTER TABLE t ADD COLUMN note TEXT NOT NULL\x00")})
+	if code := errorCode(msgs[0]); code != "23502" {
+		t.Fatalf("NOT NULL without default SQLSTATE = %q, want 23502", code)
+	}
+
+	// Backfilling one constant into two rows violates the declared UNIQUE
+	// constraint.
+	msgs = runQuery(t, c, client, &Message{Type: MsgQuery, Data: []byte("ALTER TABLE t ADD COLUMN code TEXT DEFAULT 'same' UNIQUE\x00")})
+	if code := errorCode(msgs[0]); code != "23505" {
+		t.Fatalf("UNIQUE backfill SQLSTATE = %q, want 23505", code)
+	}
+
+	msgs = runQuery(t, c, client, &Message{Type: MsgQuery, Data: []byte("ALTER TABLE t ADD COLUMN note TEXT NOT NULL DEFAULT 'kept'\x00")})
+	if code := errorCode(msgs[0]); code != "" {
+		t.Fatalf("NOT NULL with default SQLSTATE = %q, want success", code)
+	}
+	msgs = runQuery(t, c, client, &Message{Type: MsgQuery, Data: []byte("SELECT note FROM t ORDER BY id\x00")})
+	for i, values := range []string{"kept", "kept"} {
+		got := dataRowValues(t, msgs[1+i])
+		if len(got) != 1 || got[0] != values {
+			t.Fatalf("row %d note = %v, want [%s]", i, got, values)
+		}
+	}
+}
+
 func TestAlterTableDropAndRenameColumnOverSimpleQuery(t *testing.T) {
 	c, client := newDataConnection(t)
 	runQuery(t, c, client, &Message{Type: MsgQuery, Data: append([]byte("CREATE TABLE t (id INTEGER PRIMARY KEY, keep TEXT, dropme TEXT)"), 0)})

pkg/storage/schema.go

@@ -931,7 +931,39 @@ func (m *SchemaManager) removeFromIndexList(name string) error {
 }
 
 // AddColumn adds a new column to a table.
+//
+// A constant DEFAULT is materialized into every existing row before the schema
+// is published. NOT NULL, UNIQUE, and index validation all read persisted row
+// values, so a default that only existed in the catalog would be invisible to
+// them: a UNIQUE index over the new column could not see the duplicate values
+// every existing row shares, and an UPDATE to NULL would look non-null. The
+// rewrite is the same shape as DropColumn: the table's exclusive gate keeps
+// point writes and CAS commits out while rows are scanned and rewritten, the
+// transform is idempotent, and the schema is saved last so an interrupted
+// operation resumes against the old schema. A nullable column without a default
+// needs no rewrite because a missing row field already reads as NULL. Like
+// DROP COLUMN, the statement holds the catalog lock for the whole rewrite, so
+// the cost is one table scan and a table-wide stall per ADD COLUMN with a
+// default.
 func (m *SchemaManager) AddColumn(table string, column Column) error {
+	if column.Default == nil {
+		m.mu.Lock()
+		defer m.mu.Unlock()
+
+		schema, err := m.getSchemaUnsafe(table)
+		if err != nil {
+			return err
+		}
+		if err := checkColumnAbsent(schema, column.Name); err != nil {
+			return err
+		}
+		return m.publishAddedColumn(schema, column)
+	}
+
+	tableLock := m.tableLock(table)
+	tableLock.Lock()
+	defer tableLock.Unlock()
+
 	m.mu.Lock()
 	defer m.mu.Unlock()
 
@@ -939,14 +971,28 @@ func (m *SchemaManager) AddColumn(table string, column Column) error {
 	if err != nil {
 		return err
 	}
+	if err := checkColumnAbsent(schema, column.Name); err != nil {
+		return err
+	}
 
-	// Check if column already exists
-	for _, col := range schema.Columns {
-		if strings.EqualFold(col.Name, column.Name) {
-			return fmt.Errorf("column already exists: %s", column.Name)
+	if err := m.rewriteRows(table, func(row Row) bool {
+		for key := range row {
+			if strings.EqualFold(key, column.Name) {
+				return false
+			}
 		}
+		row[column.Name] = column.Default
+		return true
+	}); err != nil {
+		return err
 	}
 
+	return m.publishAddedColumn(schema, column)
+}
+
+// publishAddedColumn appends an absent column to the published schema snapshot
+// (m.mu must be held).
+func (m *SchemaManager) publishAddedColumn(schema *Schema, column Column) error {
 	// Publish a fresh snapshot instead of mutating readers' shared pointer.
 	schema = cloneSchema(schema)
 	schema.Columns = append(schema.Columns, column)
@@ -955,6 +1001,15 @@ func (m *SchemaManager) AddColumn(table string, column Column) error {
 	return m.updateSchemaUnsafe(schema)
 }
 
+func checkColumnAbsent(schema *Schema, name string) error {
+	for _, col := range schema.Columns {
+		if strings.EqualFold(col.Name, name) {
+			return fmt.Errorf("column already exists: %s", name)
+		}
+	}
+	return nil
+}
+
 // DropColumn removes a column from a table.
 func (m *SchemaManager) DropColumn(table, columnName string) error {
 	tableLock := m.tableLock(table)