Merge pull request #30 from database-pizza/engine/a-add-column-backfill

feat: backfill ADD COLUMN constant defaults into existing 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, 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: true, Status: "rewrites_index_metadata", Notes: "column references in index definitions are rewritten in place, including composite and unique indexes; index names are unchanged; RENAME COLUMN is rejected when an expression index references the column (drop that index, rename, then recreate it)"},

pkg/capabilities/capabilities_test.go

@@ -56,3 +56,16 @@ func TestJSONArrowOperatorsAreAdvertised(t *testing.T) {
 		t.Fatalf("arrow operator status = %q, want a supported status", arrow.Status)
 	}
 }
+
+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,6 +1,7 @@
 package executor
 
 import (
+	"errors"
 	"strings"
 	"sync"
 	"testing"
@@ -8,9 +9,9 @@ import (
 	"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)")
@@ -23,8 +24,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])
 		}
 	}
 
@@ -35,6 +36,269 @@ 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. 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)
+	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")
+
+	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)
+	}
+
+	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)
+	}
+}
+
+// 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) {
+	_, 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

@@ -4113,22 +4113,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 {
@@ -4138,6 +4144,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)
@@ -4159,7 +4168,28 @@ func (e *Executor) executeAlterTableAddColumn(table string, action *parser.AddCo
 		}
 	}
 
-	if err := e.schema.AddColumn(table, col); err != nil {
+	var uniqueIndex *storage.Index
+	if uniqueConstraint != nil {
+		uniqueIndex = e.addColumnUniqueIndex(table, col.Name, uniqueConstraint)
+	}
+
+	// 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 {
+		stagedRows, err := e.session.CountFast(table)
+		if err != nil {
+			return nil, err
+		}
+		if stagedRows > 0 {
+			return nil, &storage.NotNullViolationError{Table: table, Column: col.Name}
+		}
+	}
+
+	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
 		}
@@ -4173,6 +4203,31 @@ func (e *Executor) executeAlterTableAddColumn(table string, action *parser.AddCo
 	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
+}
+
 // 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/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,8 +942,32 @@ func (m *SchemaManager) removeFromIndexList(name string) error {
 	return err
 }
 
-// AddColumn adds a new column to a table.
-func (m *SchemaManager) AddColumn(table string, column Column) error {
+// 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
+// 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, 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()
+
 	m.mu.Lock()
 	defer m.mu.Unlock()
 
@@ -939,20 +975,108 @@ func (m *SchemaManager) AddColumn(table string, column Column) error {
 	if err != nil {
 		return err
 	}
+	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
+		}
+	}
 
-	// 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)
+	// 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
 		}
 	}
 
+	return m.publishAddedColumn(schema, column, unique)
+}
+
+// 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 {
+	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.
@@ -1343,6 +1467,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)
@@ -1379,7 +1515,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)
@@ -1402,6 +1542,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