fix: close ADD COLUMN backfill races and publish its index atomically

Two problems in the previous commit:

1. A row image built before the DDL could be committed after it. A
   transaction stages a row normalized against the old schema, then a
   concurrent ADD COLUMN backfills and publishes (possibly a NOT NULL)
   column, then COMMIT persists the staged row. The compare-batch check
   cannot see it: a new key is validated as absent, so nothing conflicts.
   The same window exists between prepareInsert and the write gate for an
   autocommit INSERT, and between the schema read and the gate in
   UpdateByPK.

   Write paths now re-apply the schema that is current under the table
   gate: applySchemaDefaults fills catalog defaults for columns the row
   image predates and validates required columns. Commit does it for every
   staged row before uniqueness and cache planning, so those see the
   final values; INSERT does it after taking the gate and re-encodes the
   row afterwards; UpdateByPK reads its schema under the gate. A missing
   required column now fails with 23502 instead of becoming a durable
   NOT NULL violation, and a missing defaulted column receives the same
   value the backfill wrote.

2. The column and its UNIQUE index were published by separate writes, so
   a crash in between left the column visible but unconstrained. The
   whole statement now runs under the table's exclusive gate, validates
   NOT NULL and UNIQUE before touching rows, and publishes the schema
   snapshot, the index definition, and the index list in one atomic batch
   write. A rejected publication leaves neither the column nor a
   half-written index, and an existing index name is never overwritten.

ADD COLUMN IF NOT EXISTS with a different default stays a no-op.

pkg/capabilities/capabilities.go

@@ -176,7 +176,7 @@ func Current() *Document {
 
 		DDL: DDL{
 			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"},
+			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, ADD COLUMN ... UNIQUE conflicts with 23505 when backfilling a non-NULL default into more than one row, and the column is published together with its UNIQUE index in one atomic schema write"},
 			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/executor/alter_table_test.go

@@ -180,7 +180,8 @@ func TestAlterTableAddColumnUniqueWithDefaultRejectsPopulatedTable(t *testing.T)
 }
 
 // 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.
+// constraint is enforced for every later INSERT that omits the column. The
+// column and the index that validates it must be visible together.
 func TestAlterTableAddColumnUniqueWithDefaultEnforcesLaterInserts(t *testing.T) {
 	_, schema, table := newTestDB(t)
 	e := newExec(schema, table)
@@ -189,7 +190,25 @@ func TestAlterTableAddColumnUniqueWithDefaultEnforcesLaterInserts(t *testing.T)
 
 	execMust(t, e, "ALTER TABLE t ADD COLUMN code TEXT DEFAULT 'same' UNIQUE")
 
-	_, err := execSQL(e, "INSERT INTO t (id) VALUES (2)")
+	s, err := schema.GetSchema("t")
+	if err != nil {
+		t.Fatalf("get schema: %v", err)
+	}
+	if _, exists := s.GetColumn("code"); !exists {
+		t.Fatal("ADD COLUMN ... UNIQUE published no column")
+	}
+	indexes, err := schema.ListTableIndexes("t")
+	if err != nil {
+		t.Fatalf("list indexes: %v", err)
+	}
+	if len(indexes) != 1 {
+		t.Fatalf("indexes = %v, want the one unique index for column code", indexes)
+	}
+	if !indexes[0].Unique || len(indexes[0].Columns) != 1 || indexes[0].Columns[0].Name != "code" {
+		t.Fatalf("index = %+v, want a unique index over code", indexes[0])
+	}
+
+	_, 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)
@@ -202,6 +221,52 @@ func TestAlterTableAddColumnUniqueWithDefaultEnforcesLaterInserts(t *testing.T)
 	}
 }
 
+// A transaction can stage a row image before a concurrent ADD COLUMN publishes
+// a required column. The commit must re-validate that row against the schema now
+// in effect instead of persisting a durable NOT NULL violation.
+func TestAlterTableAddColumnRejectsRowStagedBeforeDDL(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	writer := newExec(schema, table)
+	ddl := newExec(schema, table)
+	execMust(t, writer, "CREATE TABLE t (id INTEGER PRIMARY KEY)")
+
+	execMust(t, writer, "BEGIN")
+	execMust(t, writer, "INSERT INTO t (id) VALUES (1)")
+
+	// The table is durably empty, so the ALTER accepts the required column.
+	execMust(t, ddl, "ALTER TABLE t ADD COLUMN status TEXT NOT NULL")
+
+	_, err := execSQL(writer, "COMMIT")
+	var violation *storage.NotNullViolationError
+	if !errors.As(err, &violation) {
+		t.Fatalf("commit error = %v, want NotNullViolationError", err)
+	}
+	if got := execMust(t, ddl, "SELECT id FROM t").RowCount; got != 0 {
+		t.Fatalf("rows after rejected commit = %d, want 0", got)
+	}
+}
+
+// The same race is harmless when the new column has a default: the commit
+// materializes it into the staged row, exactly as the backfill did for rows that
+// were already durable.
+func TestAlterTableAddColumnDefaultsRowStagedBeforeDDL(t *testing.T) {
+	_, schema, table := newTestDB(t)
+	writer := newExec(schema, table)
+	ddl := newExec(schema, table)
+	execMust(t, writer, "CREATE TABLE t (id INTEGER PRIMARY KEY)")
+
+	execMust(t, writer, "BEGIN")
+	execMust(t, writer, "INSERT INTO t (id) VALUES (1)")
+
+	execMust(t, ddl, "ALTER TABLE t ADD COLUMN status TEXT NOT NULL DEFAULT 'pending'")
+	execMust(t, writer, "COMMIT")
+
+	res := execMust(t, ddl, "SELECT id, status FROM t")
+	if len(res.Rows) != 1 || res.Rows[0][1] != "pending" {
+		t.Fatalf("rows = %v, want [[1 pending]]", 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) {

pkg/executor/executor.go

@@ -4131,33 +4131,23 @@ func (e *Executor) executeAlterTableAddColumn(table string, action *parser.AddCo
 		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.
+	// The schema manager validates the new column against durable rows, rewrites
+	// them, and publishes the column together with its UNIQUE index while holding
+	// the table's exclusive gate, so a writer cannot slip a row in between the
+	// validation and the publication. Rows this session staged in an open
+	// transaction are invisible to that scan, so reject them here as well; the
+	// commit re-validates them against the schema that is current by then.
 	if !col.Nullable && col.Default == nil {
-		existingRows, err := e.session.CountFast(table)
+		stagedRows, err := e.session.CountFast(table)
 		if err != nil {
 			return nil, err
 		}
-		if existingRows > 0 {
+		if stagedRows > 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 err := e.schema.AddColumn(table, col, uniqueIndex); err != nil {
 		if action.IfNotExists && strings.Contains(err.Error(), "column already exists") {
 			return NewResult("ALTER TABLE"), nil
 		}
@@ -4167,12 +4157,6 @@ 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
 }
@@ -4202,34 +4186,6 @@ func (e *Executor) addColumnUniqueIndex(table, column string, constraint *parser
 	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/storage/constraints.go

@@ -1,6 +1,9 @@
 package storage
 
-import "fmt"
+import (
+	"fmt"
+	"strings"
+)
 
 // PrimaryKeyViolationError preserves duplicate-key identity through wrapped
 // executor errors so wire adapters can report integrity SQLSTATE 23505.
@@ -31,3 +34,41 @@ func validateNotNull(schema *Schema, row Row) error {
 	}
 	return nil
 }
+
+// applySchemaDefaults completes a row image that was built against an older
+// schema: columns the row predates receive their catalog default, and the row is
+// then validated against the required columns of the current schema.
+//
+// Write paths re-run it under the table gate because building a row image and
+// publishing it are separate steps. A transaction can normalize an INSERT before
+// a concurrent ADD COLUMN publishes a column and commit after it; the commit's
+// compare check cannot reject that row, because a new key is only checked for
+// absence. Re-applying the current schema keeps such a row from becoming a
+// durable NOT NULL violation, and gives it the default every other row received
+// from the backfill. An absent key is the only thing filled, so an explicit NULL
+// still does not select a default.
+func applySchemaDefaults(schema *Schema, row Row) error {
+	for _, column := range schema.Columns {
+		if _, ok := rowColumnValue(row, column.Name); ok {
+			continue
+		}
+		if column.Default != nil {
+			row[column.Name] = column.Default
+		}
+	}
+	return validateNotNull(schema, row)
+}
+
+// rowColumnValue returns the value a row holds for a column, matching the column
+// name case-insensitively.
+func rowColumnValue(row Row, name string) (interface{}, bool) {
+	if value, ok := row[name]; ok {
+		return value, true
+	}
+	for key, value := range row {
+		if strings.EqualFold(key, name) {
+			return value, true
+		}
+	}
+	return nil, false
+}

pkg/storage/schema.go

@@ -764,6 +764,12 @@ func (m *SchemaManager) DropIndex(name string) error {
 func (m *SchemaManager) IndexExists(name string) bool {
 	m.mu.RLock()
 	defer m.mu.RUnlock()
+	return m.indexExistsUnsafe(name)
+}
+
+// indexExistsUnsafe reports whether an index definition is durable (m.mu must be
+// held).
+func (m *SchemaManager) indexExistsUnsafe(name string) bool {
 	if _, ok := m.indexCache[strings.ToLower(name)]; ok {
 		return true
 	}
@@ -812,6 +818,12 @@ func (m *SchemaManager) GetIndex(name string) (*Index, error) {
 func (m *SchemaManager) ListIndexes() ([]string, error) {
 	m.mu.Lock()
 	defer m.mu.Unlock()
+	return m.indexListUnsafe()
+}
+
+// indexListUnsafe returns all index names, loading them from storage on first
+// use (m.mu must be held).
+func (m *SchemaManager) indexListUnsafe() ([]string, error) {
 	if m.indexListCached {
 		return append([]string(nil), m.indexListCache...), nil
 	}
@@ -930,7 +942,8 @@ func (m *SchemaManager) removeFromIndexList(name string) error {
 	return err
 }
 
-// AddColumn adds a new column to a table.
+// AddColumn adds a new column to a table. When unique is non-nil the column's
+// UNIQUE index is registered as part of the same statement.
 //
 // A constant DEFAULT is materialized into every existing row before the schema
 // is published. NOT NULL, UNIQUE, and index validation all read persisted row
@@ -938,28 +951,19 @@ func (m *SchemaManager) removeFromIndexList(name string) error {
 // 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)
-	}
-
+// point writes and CAS commits out while rows are scanned, validated, and
+// rewritten, the transform is idempotent, and the schema is saved last so an
+// interrupted operation resumes against the old schema.
+//
+// The gate also makes the validation authoritative: NOT NULL without a default
+// and duplicate UNIQUE values are rejected before any row is touched, and the
+// column and its index are published together in one batch write, so a crash
+// cannot leave the column visible without the constraint that validates it.
+// 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 scan, so the cost is one table scan (two for UNIQUE) and a
+// table-wide stall per ADD COLUMN that has a default or an index.
+func (m *SchemaManager) AddColumn(table string, column Column, unique *Index) error {
 	tableLock := m.tableLock(table)
 	tableLock.Lock()
 	defer tableLock.Unlock()
@@ -974,31 +978,96 @@ func (m *SchemaManager) AddColumn(table string, column Column) error {
 	if err := checkColumnAbsent(schema, column.Name); err != nil {
 		return err
 	}
+	if unique != nil {
+		if !strings.EqualFold(unique.Table, schema.Name) {
+			return fmt.Errorf("index %s belongs to table %s, not %s", unique.Name, unique.Table, schema.Name)
+		}
+		if m.indexExistsUnsafe(unique.Name) {
+			return fmt.Errorf("index already exists: %s", unique.Name)
+		}
+		if err := m.validateAddColumnUnique(table, column, unique); err != nil {
+			return err
+		}
+	}
 
-	if err := m.rewriteRows(table, func(row Row) bool {
-		for key := range row {
-			if strings.EqualFold(key, column.Name) {
+	// A required column with nothing to write into rows that already exist
+	// cannot be declared on a populated table.
+	if !column.Nullable && column.Default == nil {
+		hasRows, err := m.tableHasRows(table)
+		if err != nil {
+			return err
+		}
+		if hasRows {
+			return &NotNullViolationError{Table: schema.Name, Column: column.Name}
+		}
+	}
+
+	if column.Default != nil {
+		if err := m.rewriteRows(table, func(row Row) bool {
+			if _, ok := rowColumnValue(row, column.Name); ok {
 				return false
 			}
+			row[column.Name] = column.Default
+			return true
+		}); err != nil {
+			return err
 		}
-		row[column.Name] = column.Default
-		return true
-	}); err != nil {
-		return err
 	}
 
-	return m.publishAddedColumn(schema, column)
+	return m.publishAddedColumn(schema, column, unique)
 }
 
-// publishAddedColumn appends an absent column to the published schema snapshot
-// (m.mu must be held).
-func (m *SchemaManager) publishAddedColumn(schema *Schema, column Column) error {
+// publishAddedColumn publishes a column and, when present, its unique index in
+// one atomic batch write (m.mu must be held): the schema snapshot, the index
+// definition, and the index list always become durable together or not at all.
+func (m *SchemaManager) publishAddedColumn(schema *Schema, column Column, unique *Index) error {
 	// Publish a fresh snapshot instead of mutating readers' shared pointer.
 	schema = cloneSchema(schema)
 	schema.Columns = append(schema.Columns, column)
+	schemaData, err := json.Marshal(schema)
+	if err != nil {
+		return fmt.Errorf("failed to serialize schema: %w", err)
+	}
+	ops := []BatchOp{{Op: batchPut, Key: []byte(m.schemaKey(schema.Name)), Value: schemaData}}
 
-	// Update schema
-	return m.updateSchemaUnsafe(schema)
+	var indexList []string
+	if unique != nil {
+		unique.CreatedAt = time.Now()
+		indexData, err := json.Marshal(unique)
+		if err != nil {
+			return fmt.Errorf("failed to serialize index: %w", err)
+		}
+		indexList, err = m.indexListUnsafe()
+		if err != nil {
+			return err
+		}
+		indexList = append(indexList, unique.Name)
+		listData, err := json.Marshal(indexList)
+		if err != nil {
+			return fmt.Errorf("failed to serialize index list: %w", err)
+		}
+		ops = append(ops,
+			BatchOp{Op: batchPut, Key: []byte(m.indexKey(unique.Name)), Value: indexData},
+			BatchOp{Op: batchPut, Key: []byte(m.indexListKey()), Value: listData},
+		)
+	}
+
+	if err := m.pool.WithClient(func(c *KVClient) error {
+		_, err := c.BatchWrite(ops, nil)
+		return err
+	}); err != nil {
+		return fmt.Errorf("failed to publish column %s: %w", column.Name, err)
+	}
+
+	// Publish in-memory state only after the durable write.
+	m.cache[strings.ToLower(schema.Name)] = schema
+	if unique != nil {
+		m.indexCache[strings.ToLower(unique.Name)] = cloneIndex(unique)
+		m.indexListCache = append([]string(nil), indexList...)
+		m.indexListCached = true
+	}
+	m.bumpVersionLocked()
+	return nil
 }
 
 func checkColumnAbsent(schema *Schema, name string) error {
@@ -1331,6 +1400,18 @@ func (m *SchemaManager) RenameColumn(table, oldName, newName string) error {
 // batches. Column DDL updates the schema only after all rows are rewritten, so
 // an interrupted operation can safely resume against the old schema.
 func (m *SchemaManager) rewriteRows(table string, transform func(Row) bool) error {
+	return m.applyRows(table, func(row Row) (bool, error) {
+		return transform(row), nil
+	})
+}
+
+// applyRows scans a table once and applies transform to every row. Rows for
+// which transform reports a change are re-encoded and written in bounded
+// batches; an error from transform aborts the scan. Callers hold the table's
+// exclusive gate, so no writer can interleave between the scan and the schema
+// update that follows it. Aborting after a batch was flushed is safe only
+// because the DDL transforms are idempotent.
+func (m *SchemaManager) applyRows(table string, transform func(Row) (bool, error)) error {
 	prefix := []byte(fmt.Sprintf("%s:_data:%s:", m.database, strings.ToLower(table)))
 	return m.pool.WithClient(func(client *KVClient) (retErr error) {
 		cursor, err := client.Scan(prefix)
@@ -1367,7 +1448,11 @@ func (m *SchemaManager) rewriteRows(table string, transform func(Row) bool) erro
 				if err != nil {
 					return err
 				}
-				if !transform(row) {
+				changed, err := transform(row)
+				if err != nil {
+					return err
+				}
+				if !changed {
 					continue
 				}
 				value, err := encodeRow(row)
@@ -1390,6 +1475,56 @@ func (m *SchemaManager) rewriteRows(table string, transform func(Row) bool) erro
 	})
 }
 
+// validateAddColumnUnique rejects an ADD COLUMN ... UNIQUE whose backfilled
+// values would duplicate across existing rows. It reads rows as they will look
+// after the rewrite (a row that already carries the field from an interrupted
+// earlier attempt keeps its value) and mirrors ValidateUniqueIndex for a single
+// plain column: NULLs do not collide. The caller holds the table's exclusive
+// gate, so no writer can insert a conflicting row between this check and the
+// publication of the index.
+func (m *SchemaManager) validateAddColumnUnique(table string, column Column, index *Index) error {
+	seen := make(map[string]struct{})
+	return m.applyRows(table, func(row Row) (bool, error) {
+		value, ok := rowColumnValue(row, column.Name)
+		if !ok {
+			value = column.Default
+		}
+		if value == nil {
+			return false, nil
+		}
+		encoded := encodeUniqueScalar(value)
+		if _, duplicate := seen[encoded]; duplicate {
+			return false, &UniqueViolationError{Constraint: index.Name}
+		}
+		seen[encoded] = struct{}{}
+		return false, nil
+	})
+}
+
+// tableHasRows reports whether a table holds at least one durable row.
+func (m *SchemaManager) tableHasRows(table string) (bool, error) {
+	prefix := []byte(fmt.Sprintf("%s:_data:%s:", m.database, strings.ToLower(table)))
+	found := false
+	err := m.pool.WithClient(func(client *KVClient) (retErr error) {
+		cursor, err := client.ScanWithLimit(prefix, 1)
+		if err != nil {
+			return err
+		}
+		defer func() {
+			if cerr := cursor.Close(); retErr == nil {
+				retErr = cerr
+			}
+		}()
+		entries, _, err := cursor.Next()
+		if err != nil {
+			return err
+		}
+		found = len(entries) > 0
+		return nil
+	})
+	return found, err
+}
+
 // getSchemaUnsafe gets a schema without locking (internal use).
 func (m *SchemaManager) getSchemaUnsafe(table string) (*Schema, error) {
 	tableLower := strings.ToLower(table)

pkg/storage/schema_test.go

@@ -870,3 +870,110 @@ func TestListTableIndexesSkipsOnlyNotFound(t *testing.T) {
 		t.Fatalf("expected only uq_v, got %v", indexes)
 	}
 }
+
+// AddColumn publishes the column together with its unique index in one durable
+// batch: the schema key, the index definition, and the index list share a single
+// LSN, so a crash can never expose the column without the constraint that
+// validates it.
+func TestAddColumnPublishesUniqueIndexInOneWrite(t *testing.T) {
+	kv := newTestKVServer(t)
+	defer kv.close()
+	pool := newTestKVPool(kv, 4, 5*time.Second)
+	defer pool.Close()
+	schemas := NewSchemaManager(pool, "testdb")
+	tables := NewTableManager(pool, schemas, "testdb")
+
+	if err := schemas.CreateTable(&Schema{
+		Name:       "t",
+		PrimaryKey: "id",
+		Columns:    []Column{{Name: "id", Type: "INTEGER", PrimaryKey: true, Nullable: false}},
+	}); err != nil {
+		t.Fatal(err)
+	}
+	if err := tables.Insert("t", Row{"id": int64(1)}); err != nil {
+		t.Fatal(err)
+	}
+
+	index := &Index{Name: "uniq_t_code", Table: "t", Unique: true, Columns: []IndexColumn{{Name: "code"}}}
+	schemaWritesBefore := kv.writeCount(":_schema:")
+	if err := schemas.AddColumn("t", Column{Name: "code", Type: "TEXT", Nullable: true, Default: "x"}, index); err != nil {
+		t.Fatalf("add column: %v", err)
+	}
+
+	schema, err := schemas.GetSchema("t")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, ok := schema.GetColumn("code"); !ok {
+		t.Fatal("schema is missing the added column")
+	}
+	stored, err := schemas.GetIndex("uniq_t_code")
+	if err != nil {
+		t.Fatalf("get index: %v", err)
+	}
+	if !stored.Unique || len(stored.Columns) != 1 || stored.Columns[0].Name != "code" || stored.Table != "t" {
+		t.Fatalf("stored index = %+v, want a unique index over t.code", stored)
+	}
+	names, err := schemas.ListIndexes()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(names) != 1 || names[0] != "uniq_t_code" {
+		t.Fatalf("index list = %v, want [uniq_t_code]", names)
+	}
+
+	kv.mu.Lock()
+	schemaLSN := kv.lsns["testdb:_schema:t"]
+	indexLSN := kv.lsns["testdb:index:uniq_t_code"]
+	listLSN := kv.lsns["testdb:indexes"]
+	schemaWrites := kv.writes["testdb:_schema:t"]
+	indexWrites := kv.writes["testdb:index:uniq_t_code"]
+	kv.mu.Unlock()
+	if schemaLSN == 0 || schemaLSN != indexLSN || schemaLSN != listLSN {
+		t.Fatalf("schema/index/list LSNs = %d/%d/%d, want one shared LSN", schemaLSN, indexLSN, listLSN)
+	}
+	if schemaWrites != schemaWritesBefore+1 || indexWrites != 1 {
+		t.Fatalf("schema/index writes = %d/%d (schema had %d before), want one write each", schemaWrites, indexWrites, schemaWritesBefore)
+	}
+}
+
+// A rejected unique index must not leave the column behind, and an existing
+// index with the same name must be left intact.
+func TestAddColumnRejectsDuplicateIndexName(t *testing.T) {
+	kv := newTestKVServer(t)
+	defer kv.close()
+	pool := newTestKVPool(kv, 4, 5*time.Second)
+	defer pool.Close()
+	schemas := NewSchemaManager(pool, "testdb")
+
+	if err := schemas.CreateTable(&Schema{
+		Name:       "t",
+		PrimaryKey: "id",
+		Columns:    []Column{{Name: "id", Type: "INTEGER", PrimaryKey: true, Nullable: false}},
+	}); err != nil {
+		t.Fatal(err)
+	}
+	if err := schemas.CreateIndex(&Index{Name: "uniq_t_code", Table: "t", Unique: true, Columns: []IndexColumn{{Name: "id"}}}); err != nil {
+		t.Fatal(err)
+	}
+
+	err := schemas.AddColumn("t", Column{Name: "code", Type: "TEXT", Nullable: true, Default: "x"},
+		&Index{Name: "uniq_t_code", Table: "t", Unique: true, Columns: []IndexColumn{{Name: "code"}}})
+	if err == nil {
+		t.Fatal("expected the existing index name to be rejected")
+	}
+	schema, err := schemas.GetSchema("t")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, ok := schema.GetColumn("code"); ok {
+		t.Fatal("rejected ADD COLUMN published the column")
+	}
+	stored, err := schemas.GetIndex("uniq_t_code")
+	if err != nil {
+		t.Fatalf("existing index disappeared: %v", err)
+	}
+	if len(stored.Columns) != 1 || stored.Columns[0].Name != "id" {
+		t.Fatalf("existing index was overwritten: %+v", stored)
+	}
+}

pkg/storage/table.go

@@ -574,10 +574,6 @@ func (m *TableManager) InsertWithRowID(table string, row Row) (int64, error) {
 		return 0, err
 	}
 	rowid, _ := rowIDFromRow(nr)
-	data, err := encodeRow(nr)
-	if err != nil {
-		return 0, fmt.Errorf("failed to serialize row: %w", err)
-	}
 	wasInit := m.countInitialized(table)
 
 	// Point writers share this gate with each other. Transaction commits and
@@ -590,6 +586,23 @@ func (m *TableManager) InsertWithRowID(table string, row Row) (int64, error) {
 		return 0, err
 	}
 	defer unlock()
+
+	// prepareInsert normalized the row before this gate was held, so a
+	// concurrent ADD COLUMN may have published a column the row image predates.
+	// Re-apply the schema that is current under the gate; schema DDL needs the
+	// exclusive form of this gate, so it cannot change again before the write.
+	schema, err := m.schema.GetSchema(table)
+	if err != nil {
+		return 0, err
+	}
+	if err := applySchemaDefaults(schema, nr); err != nil {
+		return 0, err
+	}
+	data, err := encodeRow(nr)
+	if err != nil {
+		return 0, fmt.Errorf("failed to serialize row: %w", err)
+	}
+
 	var uniquePlan uniqueCachePlan
 	if uniq {
 		uniquePlan, err = m.prepareUniqueRows(table, map[string]*overlayEntry{key: {row: nr}})
@@ -623,9 +636,7 @@ func (m *TableManager) InsertWithRowID(table string, row Row) (int64, error) {
 		return 0, err
 	}
 	m.bumpGeneration(table)
-	if schema, serr := m.schema.GetSchema(table); serr == nil {
-		m.incrCount(table, schema.CreatedAt, 1, wasInit)
-	}
+	m.incrCount(table, schema.CreatedAt, 1, wasInit)
 	return rowid, nil
 }
 
@@ -1047,17 +1058,21 @@ func (m *TableManager) updateLocked(table string, updateFn func(Row) (Row, error
 // striped lock and a compare-and-swap write so a concurrent modification of the
 // same row fails with a serialization error instead of being silently lost.
 func (m *TableManager) UpdateByPK(table, pk string, updateFn func(Row) (Row, error)) (Row, bool, error) {
-	schema, err := m.schema.GetSchema(table)
+	key := m.dataKey(table, pk)
+	unlock, uniq, err := m.lockForWrite(table)
 	if err != nil {
 		return nil, false, err
 	}
+	defer unlock()
 
-	key := m.dataKey(table, pk)
-	unlock, uniq, err := m.lockForWrite(table)
+	// Read the schema under the gate so validation sees every column a
+	// concurrent ADD COLUMN has published; the row itself is read below, also
+	// under the gate.
+	schema, err := m.schema.GetSchema(table)
 	if err != nil {
 		return nil, false, err
 	}
-	defer unlock()
+
 	st := m.stripeKey(key)
 	st.Lock()
 	defer st.Unlock()
@@ -1084,7 +1099,7 @@ func (m *TableManager) UpdateByPK(table, pk string, updateFn func(Row) (Row, err
 		}
 	}
 
-	if err := validateNotNull(schema, row); err != nil {
+	if err := applySchemaDefaults(schema, row); err != nil {
 		return nil, false, err
 	}
 	data, err := encodeRow(row)

pkg/storage/tx.go

@@ -178,6 +178,19 @@ func (s *Session) CommitWithLSN() (uint64, error) {
 	}
 	defer unlockAll()
 
+	// A staged row was normalized against the schema as of its INSERT or UPDATE,
+	// which a concurrent ADD COLUMN can predate. Re-apply the schema that is
+	// current under the held gates before any derived state is computed: the CAS
+	// check cannot reject such a row (a new key is only checked for absence), so
+	// a NOT NULL column published after the row image was built would otherwise
+	// be violated durably, and a UNIQUE index would not see the default the row
+	// is about to receive. Run it before uniqueness and cache planning so both
+	// observe the final values.
+	if err := s.applyOverlaySchemaDefaults(); err != nil {
+		s.resetLocked()
+		return 0, err
+	}
+
 	// Validate scan generations for phantom detection.
 	for t, gen := range s.scanGens {
 		if s.table.generation(t) != gen {
@@ -399,6 +412,28 @@ func (s *Session) acquireCommitLocks() (locks []*sync.RWMutex, exclusive []bool,
 	}
 }
 
+// applyOverlaySchemaDefaults re-applies the current schema to every staged row
+// (the caller holds the per-table gates). Columns published after a row image
+// was built receive their catalog default, and required columns are validated,
+// so a commit cannot persist a row that violates the schema now in effect.
+func (s *Session) applyOverlaySchemaDefaults() error {
+	for table, entries := range s.overlay {
+		schema, err := s.schema.GetSchema(table)
+		if err != nil {
+			return err
+		}
+		for _, entry := range entries {
+			if entry.absent {
+				continue
+			}
+			if err := applySchemaDefaults(schema, entry.row); err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
 func (s *Session) resetLocked() {
 	s.inTx = false
 	s.aborted = false