package storage import ( "errors" "fmt" "strconv" "strings" "sync" "time" "github.com/goccy/go-json" "github.com/danfragoso/pizzasql-next/pkg/analyzer" ) // ErrIndexNotFound is returned by GetIndex when an index is absent. It is // distinct from a storage/IO error so callers like ListTableIndexes can skip a // concurrently-dropped index without swallowing real read failures. var ErrIndexNotFound = errors.New("index not found") // Schema represents a table schema. type Schema struct { Name string `json:"name"` Columns []Column `json:"columns"` // PrimaryKey is the single primary-key column. It stays the source of truth // for pre-existing catalogs that do not carry PrimaryKeyColumns, and it is // empty for a composite key so a caller cannot mistake one member for the // whole key. Use PrimaryKeyList instead of reading either field directly. PrimaryKey string `json:"primary_key"` // PrimaryKeyColumns is the ordered primary-key column list for a composite // key. It is omitted for single-column keys so the catalog keeps its // historical shape; normalizePrimaryKey fills it in when a schema is loaded. PrimaryKeyColumns []string `json:"primary_key_columns,omitempty"` CreatedAt time.Time `json:"created_at"` NextRowID int64 `json:"next_rowid"` AutoIncrement bool `json:"autoincrement"` // Checks holds the row predicates declared with CHECK, at column and table // level alike. Inline and table-level declarations are equivalent once // parsed, so both live in this one list. Checks []CheckConstraint `json:"checks,omitempty"` // ForeignKeys holds the FOREIGN KEY declarations. Each constraint is // enforced on every write and re-validated at transaction commit; a // cascading action rewrites or removes the referencing rows as part of the // same statement. ForeignKeys []ForeignKeyConstraint `json:"foreign_keys,omitempty"` } // ForeignKeyConstraint is a persisted FOREIGN KEY declaration. Columns are the // child columns in declaration order; RefColumns are the parent columns they are // matched against. Both sides have the same length, and the parent side is // validated at DDL to be the parent's PRIMARY KEY or a UNIQUE index. type ForeignKeyConstraint struct { Name string `json:"name,omitempty"` Columns []string `json:"columns"` RefTable string `json:"ref_table"` RefColumns []string `json:"ref_columns"` // OnDelete and OnUpdate hold the declared referential action text (NO // ACTION, RESTRICT, CASCADE, SET NULL, or SET DEFAULT). They are empty when // the clause was omitted and are kept so sqlite_master can reproduce the // declaration. OnDelete string `json:"on_delete,omitempty"` OnUpdate string `json:"on_update,omitempty"` } // Label identifies the constraint in error messages: the declared name when // there is one, otherwise the child column list. func (fk ForeignKeyConstraint) Label() string { if fk.Name != "" { return fk.Name } return strings.Join(fk.Columns, ", ") + " -> " + fk.RefTable } // CheckConstraint is a persisted CHECK predicate. The predicate is stored as // SQL text and re-parsed by the evaluator installed on the SchemaManager, so a // durable reload keeps enforcing it. type CheckConstraint struct { Name string `json:"name,omitempty"` Expression string `json:"expression"` } // Label identifies the constraint in error messages and the catalog: the // declared name when there is one, otherwise the expression text. func (c CheckConstraint) Label() string { if c.Name != "" { return c.Name } return c.Expression } // Column represents a column definition. type Column struct { Name string `json:"name"` Type string `json:"type"` Nullable bool `json:"nullable"` // Default holds the evaluated constant DEFAULT. DefaultExpr holds the SQL // text of a non-constant DEFAULT (CURRENT_TIMESTAMP and the other // clock-reading forms) so it is evaluated on every write instead of being // frozen at DDL time. Only one of the two is set; DefaultExpr takes // precedence if a hand-edited catalog sets both. Default interface{} `json:"default,omitempty"` DefaultExpr string `json:"default_expr,omitempty"` PrimaryKey bool `json:"primary_key"` // GeneratedExpr holds the SQL text of a GENERATED ALWAYS AS (expr) column. // GeneratedStored reports whether the value is materialized on write // (STORED, the only form this engine persists) versus computed on read. GeneratedExpr string `json:"generated_expr,omitempty"` GeneratedStored bool `json:"generated_stored,omitempty"` } // DefaultLiteral returns the catalog representation of the column's default: // the evaluated constant value, or the SQL text of a non-constant default. // DefaultExpr takes precedence when a hand-edited catalog sets both. func (c Column) DefaultLiteral() interface{} { if c.DefaultExpr != "" { return c.DefaultExpr } if c.Default != nil { return c.Default } return nil } // Index represents an index definition. type Index struct { Name string `json:"name"` Table string `json:"table"` Columns []IndexColumn `json:"columns"` Unique bool `json:"unique"` CreatedAt time.Time `json:"created_at"` // OnConflict is the default conflict resolution declared for this index via // a UNIQUE(...) ON CONFLICT clause. Empty means the SQLite default (ABORT). OnConflict string `json:"on_conflict,omitempty"` } // IndexColumn represents a column in an index. Expression is set for expression // indexes (e.g. lower(email)); a plain column index leaves it empty and uses // Name. type IndexColumn struct { Name string `json:"name"` Desc bool `json:"desc"` Expression string `json:"expression,omitempty"` } // rowIDAllocator owns the next-ROWID state for a single table. It is a separate // mutex per table so allocating a ROWID on one table never serializes against // another table, and never contends with the SchemaManager catalog lock. type rowIDAllocator struct { mu sync.Mutex next int64 init bool } // SchemaManager manages table schemas. type SchemaManager struct { pool *KVPool database string cache map[string]*Schema indexCache map[string]*Index indexListCache []string indexListCached bool version uint64 mu sync.RWMutex tableLocksMu sync.Mutex tableLocks map[string]*sync.RWMutex // checkMu guards checkEval, the pluggable evaluator for persisted CHECK // predicates. It is set by the SQL executor so storage stays independent of // the SQL front end. checkMu sync.RWMutex checkEval CheckEvaluator rowIDMu sync.Mutex rowIDAlloc map[string]*rowIDAllocator // defaultEval evaluates a persisted non-constant DEFAULT expression. The // SQL executor registers one so storage can backfill ADD COLUMN rows and // complete row images without importing the parser or executor. // defaultClock supplies the instant a DDL statement samples once for every // expression default it touches. defaultEvalMu sync.RWMutex defaultEval DefaultEvaluator defaultClock DefaultClock // fkCatalogMu guards fkCatalog, the cached foreign-key metadata used to find // referencing tables without re-reading every schema on every write. fkCatalogMu sync.Mutex fkCatalog *foreignKeyCatalog } // DefaultEvaluator evaluates a persisted non-constant DEFAULT expression, such // as CURRENT_TIMESTAMP, at the supplied instant. Expression defaults are never // frozen at DDL time; the caller samples one instant per statement so every row // it touches sees the same value. type DefaultEvaluator func(expression string, now time.Time) (interface{}, error) // DefaultClock supplies the instant a statement's expression defaults use. The // SQL executor registers a clock backed by the session so a storage-side ADD // COLUMN backfill matches statement-time now(). type DefaultClock func() time.Time // SetDefaultEvaluator installs the evaluator used for expression defaults. func (m *SchemaManager) SetDefaultEvaluator(eval DefaultEvaluator) { m.defaultEvalMu.Lock() m.defaultEval = eval m.defaultEvalMu.Unlock() } // SetDefaultClock installs the clock sampled once per DDL statement for // expression defaults. func (m *SchemaManager) SetDefaultClock(clock DefaultClock) { m.defaultEvalMu.Lock() m.defaultClock = clock m.defaultEvalMu.Unlock() } func (m *SchemaManager) defaultEvaluator() DefaultEvaluator { m.defaultEvalMu.RLock() defer m.defaultEvalMu.RUnlock() return m.defaultEval } // defaultNow samples the registered clock, falling back to the wall clock. // Callers sample it once per statement and pass the instant down so validation // and the rewrite observe the same value. func (m *SchemaManager) defaultNow() time.Time { m.defaultEvalMu.RLock() clock := m.defaultClock m.defaultEvalMu.RUnlock() if clock != nil { return clock().UTC() } return time.Now().UTC() } // CheckEvaluator reports whether a persisted CHECK expression holds for a row. // A false result is a violation; NULL and other unknown results must be // reported as true, matching SQL three-valued logic. The SQL executor registers // one so the storage layer can enforce CHECK without importing the parser. type CheckEvaluator func(expression string, row Row) (bool, error) // SetCheckEvaluator installs the evaluator used to enforce persisted CHECK // predicates. func (m *SchemaManager) SetCheckEvaluator(eval CheckEvaluator) { m.checkMu.Lock() m.checkEval = eval m.checkMu.Unlock() } func (m *SchemaManager) checkEvaluator() CheckEvaluator { m.checkMu.RLock() defer m.checkMu.RUnlock() return m.checkEval } // BeginTransaction is retained for API compatibility. Buffered per-session // transactions no longer take a database-wide transaction lock; staged writes // are validated and committed atomically with CompareBatchWrite instead. func (m *SchemaManager) BeginTransaction() {} // EndTransaction is retained for API compatibility. func (m *SchemaManager) EndTransaction() {} // LockStatement is retained for API compatibility. Statement execution is now // serialized through per-table locks and optimistic validation, so no global // statement lock is required. func (m *SchemaManager) LockStatement() {} // UnlockStatement is retained for API compatibility. func (m *SchemaManager) UnlockStatement() {} // NewSchemaManager creates a new schema manager. func NewSchemaManager(pool *KVPool, database string) *SchemaManager { return &SchemaManager{ pool: pool, database: database, cache: make(map[string]*Schema), indexCache: make(map[string]*Index), tableLocks: make(map[string]*sync.RWMutex), rowIDAlloc: make(map[string]*rowIDAllocator), } } // rowIDAllocatorFor returns (creating if needed) the per-table ROWID allocator. func (m *SchemaManager) rowIDAllocatorFor(table string) *rowIDAllocator { key := strings.ToLower(table) m.rowIDMu.Lock() a, ok := m.rowIDAlloc[key] if !ok { a = &rowIDAllocator{} m.rowIDAlloc[key] = a } m.rowIDMu.Unlock() return a } func (m *SchemaManager) tableLock(table string) *sync.RWMutex { key := strings.ToLower(table) m.tableLocksMu.Lock() lock, ok := m.tableLocks[key] if !ok { lock = &sync.RWMutex{} m.tableLocks[key] = lock } m.tableLocksMu.Unlock() return lock } // GetDatabaseName returns the database name. func (m *SchemaManager) GetDatabaseName() string { return m.database } // GetPool returns the KV pool. func (m *SchemaManager) GetPool() *KVPool { return m.pool } // Version returns the in-process schema catalog version. It is incremented for // schema/index definition changes so cached executors can resync their analyzer // catalogs without scanning storage on every query. func (m *SchemaManager) Version() uint64 { m.mu.RLock() defer m.mu.RUnlock() return m.version } func (m *SchemaManager) bumpVersionLocked() { m.version++ } // schemaKey returns the key for a table schema. func (m *SchemaManager) schemaKey(table string) string { return fmt.Sprintf("%s:_schema:%s", m.database, strings.ToLower(table)) } // catalogKey returns the key for the table catalog. func (m *SchemaManager) catalogKey() string { return fmt.Sprintf("%s:_sys:tables", m.database) } // rowIDKey returns the key for a table's next ROWID counter. func (m *SchemaManager) rowIDKey(table string) string { return fmt.Sprintf("%s:_sys:rowid:%s", m.database, strings.ToLower(table)) } // readDurable reads a catalog key straight from durable storage, bypassing the // in-memory caches, and reports the LSN the store observed it at. Transactional // DDL uses it so a staged catalog change is pinned to the exact durable version // it was derived from, and a concurrent committed DDL fails the commit instead // of being silently overwritten. func (m *SchemaManager) readDurable(key string) (value string, lsn uint64, found bool, err error) { err = m.pool.WithClient(func(c *KVClient) error { var rerr error value, lsn, found, rerr = c.ReadWithLSN(key) return rerr }) return value, lsn, found, err } // catalogTablesDurable returns the committed table catalog and its LSN. func (m *SchemaManager) catalogTablesDurable() ([]string, uint64, error) { data, lsn, found, err := m.readDurable(m.catalogKey()) if err != nil { return nil, 0, err } if !found { return nil, lsn, nil } var tables []string if err := json.Unmarshal([]byte(data), &tables); err != nil { return nil, 0, fmt.Errorf("failed to parse catalog: %w", err) } return tables, lsn, nil } // indexListDurable returns the committed index catalog and its LSN. func (m *SchemaManager) indexListDurable() ([]string, uint64, error) { data, lsn, found, err := m.readDurable(m.indexListKey()) if err != nil { return nil, 0, err } if !found { return nil, lsn, nil } var indexes []string if err := json.Unmarshal([]byte(data), &indexes); err != nil { return nil, 0, fmt.Errorf("failed to parse index list: %w", err) } return indexes, lsn, nil } // invalidateCommittedDDL publishes a committed transactional DDL change to // readers: every cache derived from the catalog is dropped and the version is // bumped so the foreign-key metadata and session analyzers reload. The caches // are rebuilt lazily from the durable catalog, which the commit already wrote. func (m *SchemaManager) invalidateCommittedDDL() { m.mu.Lock() defer m.mu.Unlock() m.cache = make(map[string]*Schema) m.indexCache = make(map[string]*Index) m.indexListCache = nil m.indexListCached = false m.rowIDMu.Lock() m.rowIDAlloc = make(map[string]*rowIDAllocator) m.rowIDMu.Unlock() m.fkCatalogMu.Lock() m.fkCatalog = nil m.fkCatalogMu.Unlock() m.bumpVersionLocked() } // CreateTable creates a new table. func (m *SchemaManager) CreateTable(schema *Schema) error { m.mu.Lock() defer m.mu.Unlock() // Check if table already exists key := m.schemaKey(schema.Name) err := m.pool.WithClient(func(c *KVClient) error { _, err := c.Read(key) return err }) if err == nil { return fmt.Errorf("table already exists: %s", schema.Name) } // Keep the cached schema private so callers cannot mutate a published // catalog snapshot after this operation returns. schema = cloneSchema(schema) schema.CreatedAt = time.Now() // Determine the primary key if not set. A schema with no declared key uses // the synthetic _rowid_ so user columns remain unconstrained and can hold // duplicate or NULL values. schema.normalizePrimaryKey() // Serialize schema data, err := json.Marshal(schema) if err != nil { return fmt.Errorf("failed to serialize schema: %w", err) } // Write schema err = m.pool.WithClient(func(c *KVClient) error { return c.Write(key, string(data)) }) if err != nil { return fmt.Errorf("failed to write schema: %w", err) } // Update catalog if err := m.addToCatalog(schema.Name); err != nil { // Rollback schema write m.pool.WithClient(func(c *KVClient) error { return c.Delete(key) }) return err } // Update cache m.cache[strings.ToLower(schema.Name)] = schema m.bumpVersionLocked() return nil } // DropTable drops a table. func (m *SchemaManager) DropTable(name string) error { tableLock := m.tableLock(name) tableLock.Lock() defer tableLock.Unlock() m.mu.Lock() defer m.mu.Unlock() key := m.schemaKey(name) // Check if table exists err := m.pool.WithClient(func(c *KVClient) error { _, err := c.Read(key) return err }) if err != nil { return fmt.Errorf("table not found: %s", name) } // Delete all rows by scanning their actual keys and batch-deleting them, so // a table drop no longer leaks durable rows. if err := m.deleteKeysWithPrefix([]byte(fmt.Sprintf("%s:_data:%s:", m.database, strings.ToLower(name)))); err != nil { return err } // Delete schema err = m.pool.WithClient(func(c *KVClient) error { return c.Delete(key) }) if err != nil { return fmt.Errorf("failed to delete schema: %w", err) } // Delete ROWID state. m.pool.WithClient(func(c *KVClient) error { return c.Delete(m.rowIDKey(name)) }) // Update catalog if err := m.removeFromCatalog(name); err != nil { return err } // Update cache tableLower := strings.ToLower(name) delete(m.cache, tableLower) m.rowIDMu.Lock() delete(m.rowIDAlloc, tableLower) m.rowIDMu.Unlock() m.bumpVersionLocked() return nil } // GetSchema retrieves a table schema. func (m *SchemaManager) GetSchema(name string) (*Schema, error) { m.mu.RLock() if schema, ok := m.cache[strings.ToLower(name)]; ok { // Clone while still holding the read lock: the cached schema's // NextRowID field is mutated under the write lock, so cloning outside // the lock races with that mutation. cloned := cloneSchema(schema) m.mu.RUnlock() return cloned, nil } m.mu.RUnlock() m.mu.Lock() defer m.mu.Unlock() // Double-check after acquiring write lock if schema, ok := m.cache[strings.ToLower(name)]; ok { return cloneSchema(schema), nil } key := m.schemaKey(name) var data string err := m.pool.WithClient(func(c *KVClient) error { var err error data, err = c.Read(key) return err }) if err != nil { if err == ErrKeyNotFound { return nil, fmt.Errorf("table not found: %s", name) } return nil, err } var schema Schema if err := json.Unmarshal([]byte(data), &schema); err != nil { return nil, fmt.Errorf("failed to parse schema: %w", err) } schema.normalizePrimaryKey() m.cache[strings.ToLower(name)] = &schema return cloneSchema(&schema), nil } func cloneSchema(schema *Schema) *Schema { if schema == nil { return nil } cloned := *schema cloned.Columns = append([]Column(nil), schema.Columns...) cloned.Checks = append([]CheckConstraint(nil), schema.Checks...) cloned.PrimaryKeyColumns = append([]string(nil), schema.PrimaryKeyColumns...) if schema.ForeignKeys != nil { cloned.ForeignKeys = make([]ForeignKeyConstraint, len(schema.ForeignKeys)) for i, fk := range schema.ForeignKeys { cloned.ForeignKeys[i] = ForeignKeyConstraint{ Name: fk.Name, Columns: append([]string(nil), fk.Columns...), RefTable: fk.RefTable, RefColumns: append([]string(nil), fk.RefColumns...), OnDelete: fk.OnDelete, OnUpdate: fk.OnUpdate, } } } return &cloned } func cloneIndex(index *Index) *Index { if index == nil { return nil } cloned := *index cloned.Columns = append([]IndexColumn(nil), index.Columns...) return &cloned } // TableExists checks if a table exists. func (m *SchemaManager) TableExists(name string) bool { _, err := m.GetSchema(name) return err == nil } // ListTables returns all table names. func (m *SchemaManager) ListTables() ([]string, error) { var data string err := m.pool.WithClient(func(c *KVClient) error { var err error data, err = c.Read(m.catalogKey()) return err }) if err != nil { if err == ErrKeyNotFound { return nil, nil } return nil, err } var tables []string if err := json.Unmarshal([]byte(data), &tables); err != nil { return nil, fmt.Errorf("failed to parse catalog: %w", err) } return tables, nil } // addToCatalog adds a table to the catalog. func (m *SchemaManager) addToCatalog(name string) error { tables, err := m.ListTables() if err != nil && err != ErrKeyNotFound { return err } // Check if already exists lowerName := strings.ToLower(name) for _, t := range tables { if strings.ToLower(t) == lowerName { return nil } } tables = append(tables, name) data, err := json.Marshal(tables) if err != nil { return err } return m.pool.WithClient(func(c *KVClient) error { return c.Write(m.catalogKey(), string(data)) }) } // removeFromCatalog removes a table from the catalog. func (m *SchemaManager) removeFromCatalog(name string) error { tables, err := m.ListTables() if err != nil { return err } lowerName := strings.ToLower(name) newTables := make([]string, 0, len(tables)) for _, t := range tables { if strings.ToLower(t) != lowerName { newTables = append(newTables, t) } } data, err := json.Marshal(newTables) if err != nil { return err } return m.pool.WithClient(func(c *KVClient) error { return c.Write(m.catalogKey(), string(data)) }) } // InvalidateCache clears the cache for a table. func (m *SchemaManager) InvalidateCache(name string) { m.mu.Lock() defer m.mu.Unlock() tableLower := strings.ToLower(name) delete(m.cache, tableLower) m.rowIDMu.Lock() delete(m.rowIDAlloc, tableLower) m.rowIDMu.Unlock() } // ToAnalyzerTableInfo converts a Schema to analyzer.TableInfo. func (s *Schema) ToAnalyzerTableInfo() *analyzer.TableInfo { info := &analyzer.TableInfo{ Name: s.Name, } for _, col := range s.Columns { info.Columns = append(info.Columns, analyzer.ColumnInfo{ Name: col.Name, Type: analyzer.TypeFromName(col.Type), Nullable: col.Nullable, PrimaryKey: col.PrimaryKey, TableName: s.Name, Generated: col.GeneratedExpr != "", }) } return info } // normalizePrimaryKey derives the ordered primary-key column list from the // legacy single-column field or from the columns marked PRIMARY KEY, and keeps // both representations consistent. Primary-key columns are implicitly NOT NULL. func (s *Schema) normalizePrimaryKey() { if len(s.PrimaryKeyColumns) == 0 { if s.PrimaryKey != "" { s.PrimaryKeyColumns = []string{s.PrimaryKey} } else { for _, col := range s.Columns { if col.PrimaryKey { s.PrimaryKeyColumns = append(s.PrimaryKeyColumns, col.Name) } } if len(s.PrimaryKeyColumns) == 0 { s.PrimaryKeyColumns = []string{"_rowid_"} } } } if len(s.PrimaryKeyColumns) == 1 { // A single-column key keeps the historical catalog shape: only the // scalar field, so existing schema bytes stay unchanged. s.PrimaryKey = s.PrimaryKeyColumns[0] s.PrimaryKeyColumns = nil } else { s.PrimaryKey = "" } for i := range s.Columns { if s.IsPrimaryKeyColumn(s.Columns[i].Name) { s.Columns[i].PrimaryKey = true s.Columns[i].Nullable = false } } } // PrimaryKeyList returns the ordered primary-key columns. A schema always has // at least one: an explicit key, or the synthetic _rowid_. func (s *Schema) PrimaryKeyList() []string { if len(s.PrimaryKeyColumns) > 0 { return append([]string(nil), s.PrimaryKeyColumns...) } if s.PrimaryKey != "" { return []string{s.PrimaryKey} } return []string{"_rowid_"} } // UsesSyntheticRowID reports whether the table declares no primary key, so rows // are stored under the engine-assigned ROWID. func (s *Schema) UsesSyntheticRowID() bool { columns := s.PrimaryKeyList() return len(columns) == 1 && strings.EqualFold(columns[0], "_rowid_") } // IsCompositePrimaryKey reports whether the primary key spans more than one // column. func (s *Schema) IsCompositePrimaryKey() bool { return len(s.PrimaryKeyList()) > 1 } // IsSingleColumnPrimaryKey reports whether the primary key is exactly one // column, which is the only shape a durable point lookup may address with one // value. func (s *Schema) IsSingleColumnPrimaryKey() bool { return len(s.PrimaryKeyList()) == 1 } // IsPrimaryKeyColumn reports whether name is one of the primary-key columns. func (s *Schema) IsPrimaryKeyColumn(name string) bool { for _, column := range s.PrimaryKeyList() { if strings.EqualFold(column, name) { return true } } return false } // PrimaryKeyTargetMatches reports whether columns names the complete // primary-key column set. Order is not significant: ON CONFLICT (b, a) on a // PRIMARY KEY (a, b) addresses the same key, because the encoded key is built // in declaration order either way. SQLite instead requires the declaration // order of a constraint; this engine already accepts arbitrary conflict-target // lists, so it matches the set rather than raising. func (s *Schema) PrimaryKeyTargetMatches(columns []string) bool { primary := s.PrimaryKeyList() if len(primary) != len(columns) { return false } for _, column := range primary { found := false for _, candidate := range columns { if strings.EqualFold(column, candidate) { found = true break } } if !found { return false } } return true } // PrimaryKeyForRow returns the row's primary key encoded the way the table's // durable row keys encode it. A single-column key keeps the historical // "" form, so existing databases are read and written unchanged. A // composite key length-prefixes each component (":" repeated), a // prefix-free encoding, so ("ab","c") and ("a","bc") cannot collide. A NULL or // missing component is a NOT NULL violation: no rowid is auto-assigned. func (s *Schema) PrimaryKeyForRow(row Row) (string, error) { columns := s.PrimaryKeyList() parts := make([]string, len(columns)) for i, column := range columns { value, ok := rowColumnValue(row, column) if !ok || value == nil { return "", &NotNullViolationError{Table: s.Name, Column: column} } parts[i] = fmt.Sprintf("%v", value) } if len(parts) == 1 { return parts[0], nil } var b strings.Builder for _, part := range parts { b.WriteString(strconv.Itoa(len(part))) b.WriteByte(':') b.WriteString(part) } return b.String(), nil } // GetColumn returns a column by name. func (s *Schema) GetColumn(name string) (*Column, bool) { lowerName := strings.ToLower(name) for i := range s.Columns { if strings.ToLower(s.Columns[i].Name) == lowerName { return &s.Columns[i], true } } return nil, false } // GetNextRowID gets and increments the next ROWID for a table. Allocation is // serialized per table via the table's own allocator so inserts on different // tables never contend, and no global SchemaManager lock is held across the // durable derivation scan. func (m *SchemaManager) GetNextRowID(table string) (int64, error) { alloc := m.rowIDAllocatorFor(table) alloc.mu.Lock() defer alloc.mu.Unlock() next, err := m.nextRowIDLocked(alloc, table) if err != nil { return 0, err } alloc.next = next + 1 return next, nil } // UpdateMaxRowID updates the next ROWID if the provided value is higher. func (m *SchemaManager) UpdateMaxRowID(table string, rowid int64) error { alloc := m.rowIDAllocatorFor(table) alloc.mu.Lock() defer alloc.mu.Unlock() next, err := m.nextRowIDLocked(alloc, table) if err != nil { return err } if rowid >= next { alloc.next = rowid + 1 } return nil } // nextRowIDLocked returns the current next ROWID, deriving it from durable rows // on first use (must hold the per-table allocator lock). func (m *SchemaManager) nextRowIDLocked(alloc *rowIDAllocator, table string) (int64, error) { if alloc.init { return alloc.next, nil } next, err := m.deriveNextRowID(table) if err != nil { return 0, err } if next < 1 { next = 1 } alloc.next = next alloc.init = true return alloc.next, nil } // deriveNextRowID scans durable row keys to recover max(rowid)+1, streaming // each page through decodeRow instead of materializing every value. func (m *SchemaManager) deriveNextRowID(table string) (int64, error) { prefix := []byte(fmt.Sprintf("%s:_data:%s:", m.database, strings.ToLower(table))) var maxRowID int64 err := m.pool.WithClient(func(client *KVClient) (retErr error) { cursor, err := client.Scan(prefix) if err != nil { return err } defer func() { if err := cursor.Close(); retErr == nil { retErr = err } }() for { entries, done, err := cursor.Next() if err != nil { return err } for _, e := range entries { row, err := decodeRow(e.Value) if err != nil { return fmt.Errorf("failed to parse row while deriving ROWID: %w", err) } if rowid, ok := valueAsInt64(row["_rowid_"]); ok && rowid > maxRowID { maxRowID = rowid } } if done { return nil } } }) if err != nil { return 0, err } return maxRowID + 1, nil } // deleteKeysWithPrefix streams the keys with the given prefix one page at a // time and atomically batch-deletes them, so a bulk operation never leaves // durable rows behind. func (m *SchemaManager) deleteKeysWithPrefix(prefix []byte) error { return m.pool.WithClient(func(client *KVClient) (retErr error) { cursor, err := client.ScanKeys(prefix) if err != nil { return err } defer func() { if err := cursor.Close(); retErr == nil { retErr = err } }() ops := make([]BatchOp, 0, scanPageSize) batchBytes := 8 flush := func() error { if len(ops) == 0 { return nil } if _, err := client.BatchWrite(ops, nil); err != nil { return err } ops = ops[:0] batchBytes = 8 return nil } for { entries, done, err := cursor.Next() if err != nil { return err } for _, e := range entries { opBytes := 12 + len(e.Key) if len(ops) == maxOperations || batchBytes+opBytes > bulkBatchByteBudget { if err := flush(); err != nil { return err } } ops = append(ops, BatchOp{Op: batchDelete, Key: append([]byte(nil), e.Key...)}) batchBytes += opBytes } if done { return flush() } } }) } func valueAsInt64(value interface{}) (int64, bool) { switch v := value.(type) { case int64: return v, true case int: return int64(v), true case float64: return int64(v), true default: return 0, false } } // getSchemaLocked retrieves schema (must hold lock). func (m *SchemaManager) getSchemaLocked(name string) (*Schema, error) { if schema, ok := m.cache[strings.ToLower(name)]; ok { return schema, nil } key := m.schemaKey(name) var data string err := m.pool.WithClient(func(c *KVClient) error { var err error data, err = c.Read(key) return err }) if err != nil { if err == ErrKeyNotFound { return nil, fmt.Errorf("table not found: %s", name) } return nil, err } var schema Schema if err := json.Unmarshal([]byte(data), &schema); err != nil { return nil, fmt.Errorf("failed to parse schema: %w", err) } schema.normalizePrimaryKey() m.cache[strings.ToLower(name)] = &schema return &schema, nil } // saveSchemaLocked saves schema (must hold lock). func (m *SchemaManager) saveSchemaLocked(schema *Schema) error { data, err := json.Marshal(schema) if err != nil { return fmt.Errorf("failed to serialize schema: %w", err) } key := m.schemaKey(schema.Name) err = m.pool.WithClient(func(c *KVClient) error { return c.Write(key, string(data)) }) if err != nil { return fmt.Errorf("failed to write schema: %w", err) } m.cache[strings.ToLower(schema.Name)] = schema return nil } // Index management methods // indexKey returns the key for an index. func (m *SchemaManager) indexKey(name string) string { return fmt.Sprintf("%s:index:%s", m.database, strings.ToLower(name)) } // indexListKey returns the key for the index list. func (m *SchemaManager) indexListKey() string { return fmt.Sprintf("%s:indexes", m.database) } // CreateIndex creates a new index. func (m *SchemaManager) CreateIndex(index *Index) error { m.mu.Lock() defer m.mu.Unlock() // Check if index already exists key := m.indexKey(index.Name) err := m.pool.WithClient(func(c *KVClient) error { _, err := c.Read(key) return err }) if err == nil { return fmt.Errorf("index already exists: %s", index.Name) } // Verify table exists if _, err := m.getSchemaLocked(index.Table); err != nil { return fmt.Errorf("table not found: %s", index.Table) } // Save index index.CreatedAt = time.Now() data, err := json.Marshal(index) if err != nil { return fmt.Errorf("failed to serialize index: %w", err) } err = m.pool.WithClient(func(c *KVClient) error { return c.Write(key, string(data)) }) if err != nil { return fmt.Errorf("failed to write index: %w", err) } // Add to index list if err := m.addToIndexList(index.Name); err != nil { return err } m.indexCache[strings.ToLower(index.Name)] = cloneIndex(index) m.bumpVersionLocked() return nil } // DropIndex drops an index. func (m *SchemaManager) DropIndex(name string) error { m.mu.Lock() defer m.mu.Unlock() key := m.indexKey(name) err := m.pool.WithClient(func(c *KVClient) error { return c.Delete(key) }) if err != nil { return fmt.Errorf("failed to delete index: %w", err) } if err := m.removeFromIndexList(name); err != nil { return err } delete(m.indexCache, strings.ToLower(name)) m.bumpVersionLocked() return nil } // IndexExists checks if an index exists. 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 } key := m.indexKey(name) err := m.pool.WithClient(func(c *KVClient) error { _, err := c.Read(key) return err }) return err == nil } // GetIndex retrieves an index by name. func (m *SchemaManager) GetIndex(name string) (*Index, error) { m.mu.Lock() defer m.mu.Unlock() cacheKey := strings.ToLower(name) if index, ok := m.indexCache[cacheKey]; ok { return cloneIndex(index), nil } key := m.indexKey(name) var data string err := m.pool.WithClient(func(c *KVClient) error { var err error data, err = c.Read(key) return err }) if err != nil { if err == ErrKeyNotFound { return nil, fmt.Errorf("%w: %s", ErrIndexNotFound, name) } return nil, err } var index Index if err := json.Unmarshal([]byte(data), &index); err != nil { return nil, fmt.Errorf("failed to parse index: %w", err) } m.indexCache[cacheKey] = &index return cloneIndex(&index), nil } // ListIndexes returns all index names. 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 } key := m.indexListKey() var data string err := m.pool.WithClient(func(c *KVClient) error { var err error data, err = c.Read(key) return err }) if err == ErrKeyNotFound { m.indexListCache = nil m.indexListCached = true return []string{}, nil } if err != nil { return nil, err } var indexes []string if err := json.Unmarshal([]byte(data), &indexes); err != nil { return nil, fmt.Errorf("failed to parse index list: %w", err) } m.indexListCache = append([]string(nil), indexes...) m.indexListCached = true return indexes, nil } // ListTableIndexes returns all indexes for a table. func (m *SchemaManager) ListTableIndexes(table string) ([]*Index, error) { indexes, err := m.ListIndexes() if err != nil { return nil, err } var result []*Index for _, name := range indexes { idx, err := m.GetIndex(name) if err != nil { // An index dropped concurrently is simply absent; any other error // (storage/IO or corrupt data) must not be silently swallowed. if errors.Is(err, ErrIndexNotFound) { continue } return nil, err } if strings.EqualFold(idx.Table, table) { result = append(result, idx) } } return result, nil } // addToIndexList adds an index name to the list. func (m *SchemaManager) addToIndexList(name string) error { key := m.indexListKey() var indexes []string var data string err := m.pool.WithClient(func(c *KVClient) error { var err error data, err = c.Read(key) return err }) if err == nil { json.Unmarshal([]byte(data), &indexes) } indexes = append(indexes, name) newData, _ := json.Marshal(indexes) err = m.pool.WithClient(func(c *KVClient) error { return c.Write(key, string(newData)) }) if err == nil { m.indexListCache = append([]string(nil), indexes...) m.indexListCached = true } return err } // removeFromIndexList removes an index name from the list. func (m *SchemaManager) removeFromIndexList(name string) error { key := m.indexListKey() var indexes []string var data string err := m.pool.WithClient(func(c *KVClient) error { var err error data, err = c.Read(key) return err }) if err != nil { return nil } json.Unmarshal([]byte(data), &indexes) var newIndexes []string for _, idx := range indexes { if !strings.EqualFold(idx, name) { newIndexes = append(newIndexes, idx) } } newData, _ := json.Marshal(newIndexes) err = m.pool.WithClient(func(c *KVClient) error { return c.Write(key, string(newData)) }) if err == nil { m.indexListCache = append([]string(nil), newIndexes...) m.indexListCached = true } return err } // 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, and checks are // persisted with the column in the same schema snapshot. // // A DEFAULT is materialized into every existing row before the schema is // published. NOT NULL, UNIQUE, CHECK, 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, a CHECK predicate could not see a violating // default, 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, // duplicate UNIQUE values, and rows a new CHECK predicate rejects are rejected // before the column, its checks, and its index are published, and they are // published together in one batch write so a crash cannot leave the column // visible without the constraints that validate it. A constant default is // deterministic, so its UNIQUE prediction can run as a cheap pre-scan. A // clock-reading default is evaluated at one instant sampled for this ALTER // statement, and that instant is passed to every validation scan and to the // rewrite, so a clock crossing between them cannot publish a row that violates // the constraint the statement adds; the UNIQUE duplicate set is still // collected during the rewrite so it covers exactly the values the index will // see. 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 per validating // constraint plus a table-wide stall per ADD COLUMN that has a default, an // index, or a check. func (m *SchemaManager) AddColumn(table string, column Column, checks []CheckConstraint, unique *Index) error { tableLock := m.tableLock(table) tableLock.Lock() defer tableLock.Unlock() m.mu.Lock() defer m.mu.Unlock() // One instant for the whole ALTER statement: every validation scan and the // row rewrite must observe the same value, or a clock crossing between them // could publish rows that violate the CHECK or UNIQUE constraint this // statement adds. now := m.defaultNow() schema, err := m.getSchemaUnsafe(table) 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) } // A clock-reading default is validated during the rewrite below so the // duplicate set covers exactly the values this statement writes. The // pre-scan is kept for constant and absent defaults, which are // deterministic. if column.DefaultExpr == "" { if err := m.validateAddColumnUnique(table, column, unique, now); err != nil { return err } } } if err := m.validateAddColumnChecks(table, column, checks, now); err != nil { return err } // A required column with nothing to write into rows that already exist // cannot be declared on a populated table. if !column.Nullable && !columnHasDefault(column) { hasRows, err := m.tableHasRows(table) if err != nil { return err } if hasRows { return &NotNullViolationError{Table: schema.Name, Column: column.Name} } } // An expression default is evaluated at the ALTER statement's instant. When // the column is also UNIQUE, collect the values this rewrite writes and // abort on a duplicate, so the index is validated against exactly the values // it will see. var seen map[string]struct{} if unique != nil && column.DefaultExpr != "" { seen = make(map[string]struct{}) } if columnHasDefault(column) { if err := m.applyRows(table, func(row Row) (bool, error) { value, present := rowColumnValue(row, column.Name) if !present { var err error value, _, err = m.columnDefaultValue(column, now) if err != nil { return false, err } row[column.Name] = value } if seen != nil { if value == nil { return !present, nil } encoded := encodeUniqueScalar(value) if _, duplicate := seen[encoded]; duplicate { return false, &UniqueViolationError{Constraint: unique.Name} } seen[encoded] = struct{}{} } return !present, nil }); err != nil { return err } } return m.publishAddedColumn(schema, column, checks, unique) } // publishAddedColumn publishes a column, its checks, 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, checks []CheckConstraint, unique *Index) error { // Publish a fresh snapshot instead of mutating readers' shared pointer. schema = cloneSchema(schema) schema.Columns = append(schema.Columns, column) schema.Checks = append(schema.Checks, checks...) 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}} 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. func (m *SchemaManager) DropColumn(table, columnName string) error { tableLock := m.tableLock(table) tableLock.Lock() defer tableLock.Unlock() m.mu.Lock() defer m.mu.Unlock() schema, err := m.getSchemaUnsafe(table) if err != nil { return err } // Cannot drop primary key column if schema.IsPrimaryKeyColumn(columnName) { return fmt.Errorf("cannot drop primary key column: %s", columnName) } // Find and remove column newColumns := make([]Column, 0, len(schema.Columns)-1) found := false for _, col := range schema.Columns { if strings.EqualFold(col.Name, columnName) { found = true continue } newColumns = append(newColumns, col) } if !found { return fmt.Errorf("column not found: %s", columnName) } schema = cloneSchema(schema) schema.Columns = newColumns if err := m.rewriteRows(table, func(row Row) bool { for key := range row { if strings.EqualFold(key, columnName) { delete(row, key) return true } } return false }); err != nil { return err } // Update schema return m.updateSchemaUnsafe(schema) } // RenameTable renames a table. func (m *SchemaManager) RenameTable(oldName, newName string) error { m.mu.Lock() defer m.mu.Unlock() // Check if old table exists schema, err := m.getSchemaUnsafe(oldName) if err != nil { return err } // Check if new table name already exists _, err = m.getSchemaUnsafe(newName) if err == nil { return fmt.Errorf("table already exists: %s", newName) } schema = cloneSchema(schema) // Update schema name schema.Name = newName // Move durable rows from the old table's data prefix to the new one. Without // this, ALTER TABLE ... RENAME leaves every row under the old key and the // renamed table appears empty. if err := m.renameDataKeys(oldName, newName); err != nil { return err } // Repoint indexes that belonged to the old table. if err := m.repointIndexesLocked(oldName, newName); err != nil { return err } // Build the replacement catalog before switching schema names. The schema, // catalog, and stale ROWID state change in one batch, so a crash cannot // leave neither table name addressable after all row chunks have moved. tables, err := m.ListTables() if err != nil { return err } foundOld := false for i, name := range tables { if strings.EqualFold(name, oldName) { tables[i] = newName foundOld = true break } } if !foundOld { return fmt.Errorf("table not found in catalog: %s", oldName) } catalogData, err := json.Marshal(tables) if err != nil { return err } schemaData, err := json.Marshal(schema) if err != nil { return err } oldKey := m.schemaKey(oldName) newKey := m.schemaKey(newName) err = m.pool.WithClient(func(c *KVClient) error { _, err := c.BatchWrite([]BatchOp{ {Op: batchPut, Key: []byte(newKey), Value: schemaData}, {Op: batchDelete, Key: []byte(oldKey)}, {Op: batchDelete, Key: []byte(m.rowIDKey(oldName))}, {Op: batchPut, Key: []byte(m.catalogKey()), Value: catalogData}, }, nil) return err }) if err != nil { return err } // Update cache oldLower := strings.ToLower(oldName) newLower := strings.ToLower(newName) delete(m.cache, oldLower) m.rowIDMu.Lock() delete(m.rowIDAlloc, oldLower) m.rowIDMu.Unlock() // Update cache m.cache[newLower] = schema m.bumpVersionLocked() return nil } // renameDataKeys moves every durable row of a table from the old name's data // prefix to the new name's prefix in a single atomic batch per page. func (m *SchemaManager) renameDataKeys(oldName, newName string) error { oldPrefix := []byte(fmt.Sprintf("%s:_data:%s:", m.database, strings.ToLower(oldName))) newPrefix := fmt.Sprintf("%s:_data:%s:", m.database, strings.ToLower(newName)) return m.pool.WithClient(func(client *KVClient) (retErr error) { cursor, err := client.Scan(oldPrefix) if err != nil { return err } defer func() { if cerr := cursor.Close(); retErr == nil { retErr = cerr } }() ops := make([]BatchOp, 0, scanPageSize*2) batchBytes := 8 flush := func() error { if len(ops) == 0 { return nil } if _, err := client.BatchWrite(ops, nil); err != nil { return err } ops = ops[:0] batchBytes = 8 return nil } for { entries, done, err := cursor.Next() if err != nil { return err } for _, e := range entries { suffix := string(e.Key[len(oldPrefix):]) newKey := []byte(newPrefix + suffix) opBytes := 24 + len(newKey) + len(e.Value) + len(e.Key) if len(ops)+2 > maxOperations || batchBytes+opBytes > bulkBatchByteBudget { if err := flush(); err != nil { return err } } ops = append(ops, BatchOp{Op: batchPut, Key: newKey, Value: e.Value}, BatchOp{Op: batchDelete, Key: append([]byte(nil), e.Key...)}, ) batchBytes += opBytes } if done { return flush() } } }) } // updateIndexesOfTableLocked applies mutate to the durable definition of every // index on table and persists the definitions it changed. mutate reports // whether it modified the definition; unmodified indexes are left untouched. // The changed definitions are written in one batch, so a table's indexes never // split between the old and new shape. The caller holds m.mu; it reads the // durable index list directly rather than calling the self-locking ListIndexes // helper. func (m *SchemaManager) updateIndexesOfTableLocked(table string, mutate func(*Index) bool) error { var names []string if m.indexListCached { names = append([]string(nil), m.indexListCache...) } else { var data string err := m.pool.WithClient(func(c *KVClient) error { var rerr error data, rerr = c.Read(m.indexListKey()) return rerr }) if err == ErrKeyNotFound { return nil } if err != nil { return err } if err := json.Unmarshal([]byte(data), &names); err != nil { return err } } type indexUpdate struct { index Index key string value []byte } var updates []indexUpdate for _, name := range names { key := m.indexKey(name) var data string err := m.pool.WithClient(func(c *KVClient) error { var rerr error data, rerr = c.Read(key) return rerr }) if err != nil { if err == ErrKeyNotFound { continue } return err } var idx Index if err := json.Unmarshal([]byte(data), &idx); err != nil { return err } if !strings.EqualFold(idx.Table, table) { continue } if !mutate(&idx) { continue } updated, err := json.Marshal(&idx) if err != nil { return err } updates = append(updates, indexUpdate{index: idx, key: key, value: updated}) } if len(updates) == 0 { return nil } ops := make([]BatchOp, 0, len(updates)) for _, update := range updates { ops = append(ops, BatchOp{Op: batchPut, Key: []byte(update.key), Value: update.value}) } if err := m.pool.WithClient(func(c *KVClient) error { _, err := c.BatchWrite(ops, nil) return err }); err != nil { return fmt.Errorf("update index definitions for table %s: %w", table, err) } for _, update := range updates { if cached, ok := m.indexCache[strings.ToLower(update.index.Name)]; ok { *cached = update.index } } return nil } // repointIndexesLocked updates indexes whose table was renamed. func (m *SchemaManager) repointIndexesLocked(oldName, newName string) error { return m.updateIndexesOfTableLocked(oldName, func(idx *Index) bool { idx.Table = newName return true }) } // renameIndexColumnsLocked rewrites the stored definition of every index on // table so a reference to oldName becomes newName. Index names stay unchanged: // they are independent identifiers that SQLite does not derive from column // names, and callers may have chosen them explicitly. A rename to an existing // column name is rejected before this runs. Expression columns keep their SQL // text; callers must reject a rename that such an expression depends on, // because the text is not rewritten. func (m *SchemaManager) renameIndexColumnsLocked(table, oldName, newName string) error { return m.updateIndexesOfTableLocked(table, func(idx *Index) bool { changed := false for i := range idx.Columns { // Expression columns persist SQL text, not a column reference; their // text is left for the caller to drop and recreate the index. if idx.Columns[i].Expression == "" && strings.EqualFold(idx.Columns[i].Name, oldName) { idx.Columns[i].Name = newName changed = true } } return changed }) } // RenameColumn renames a column in a table. Column references in index // definitions are rewritten; expression-index text is not, so a SQL caller must // reject a rename that an expression depends on. func (m *SchemaManager) RenameColumn(table, oldName, newName string) error { tableLock := m.tableLock(table) tableLock.Lock() defer tableLock.Unlock() m.mu.Lock() defer m.mu.Unlock() schema, err := m.getSchemaUnsafe(table) if err != nil { return err } // Check if new column name already exists for _, col := range schema.Columns { if strings.EqualFold(col.Name, newName) { return fmt.Errorf("column already exists: %s", newName) } } schema = cloneSchema(schema) // Find and rename column found := false for i, col := range schema.Columns { if strings.EqualFold(col.Name, oldName) { schema.Columns[i].Name = newName found = true // Update primary key references if needed. Renaming keeps the column // order, so a composite key's encoding is unchanged. for i, pkColumn := range schema.PrimaryKeyColumns { if strings.EqualFold(pkColumn, oldName) { schema.PrimaryKeyColumns[i] = newName } } if strings.EqualFold(schema.PrimaryKey, oldName) { schema.PrimaryKey = newName } break } } if !found { return fmt.Errorf("column not found: %s", oldName) } if err := m.rewriteRows(table, func(row Row) bool { for key, value := range row { if strings.EqualFold(key, oldName) { delete(row, key) row[newName] = value return true } } return false }); err != nil { return err } // Publish the renamed schema before the index definitions: reads of the new // column name resolve against rewritten rows as soon as the schema lands. // Autocommit index DDL is not covered by this write, so a failure here leaves // stale index metadata for the caller to drop and recreate; inside a // transaction the whole statement is staged and published atomically. if err := m.updateSchemaUnsafe(schema); err != nil { return err } // Index metadata drives secondary-index values, UNIQUE enforcement, // ON CONFLICT target resolution, and catalog introspection, so it must name // the renamed column before any later statement uses the table. return m.renameIndexColumnsLocked(table, oldName, newName) } // rewriteRows applies an idempotent name-keyed row transformation in bounded // 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) if err != nil { return err } defer func() { if cerr := cursor.Close(); retErr == nil { retErr = cerr } }() ops := make([]BatchOp, 0, scanPageSize) batchBytes := 8 flush := func() error { if len(ops) == 0 { return nil } if _, err := client.BatchWrite(ops, nil); err != nil { return err } ops = ops[:0] batchBytes = 8 return nil } for { entries, done, err := cursor.Next() if err != nil { return err } for _, entry := range entries { row, err := decodeRow(entry.Value) if err != nil { return err } changed, err := transform(row) if err != nil { return err } if !changed { continue } value, err := encodeRow(row) if err != nil { return err } opBytes := 12 + len(entry.Key) + len(value) if len(ops) == maxOperations || batchBytes+opBytes > bulkBatchByteBudget { if err := flush(); err != nil { return err } } ops = append(ops, BatchOp{Op: batchPut, Key: append([]byte(nil), entry.Key...), Value: value}) batchBytes += opBytes } if done { return flush() } } }) } // 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. It is used only for a constant or absent // default; a clock-reading default is validated by the rewrite itself because // its values cannot be predicted. 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, now time.Time) error { seen := make(map[string]struct{}) return m.applyRows(table, func(row Row) (bool, error) { value, ok := rowColumnValue(row, column.Name) if !ok { var err error value, _, err = m.columnDefaultValue(column, now) if err != nil { return false, err } } 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 }) } // columnHasDefault reports whether the column contributes a value to rows that // omit it, whether that default is a stored constant or an expression. func columnHasDefault(column Column) bool { return column.Default != nil || column.DefaultExpr != "" } // validateAddColumnChecks rejects an ADD COLUMN ... CHECK whose predicate is // already false for an existing row. Like validateAddColumnUnique it reads rows // as they will look after the default backfill, and it runs before that rewrite // so a rejected DDL leaves no row carrying the new field. The default is // resolved through columnDefaultValue at the instant sampled once for the // ALTER statement, so a clock-reading default reaches the predicate as the value // the backfilled row will receive instead of NULL. The caller holds the table's // exclusive gate, so no writer can insert a violating row between this // validation and the publication of the check. func (m *SchemaManager) validateAddColumnChecks(table string, column Column, checks []CheckConstraint, now time.Time) error { if len(checks) == 0 { return nil } eval := m.checkEvaluator() if eval == nil { return fmt.Errorf("cannot validate CHECK constraints on %s: no CHECK evaluator installed", table) } return m.applyRows(table, func(row Row) (bool, error) { if _, ok := rowColumnValue(row, column.Name); !ok { value, hasDefault, err := m.columnDefaultValue(column, now) if err != nil { return false, err } if hasDefault { row[column.Name] = value } } for _, check := range checks { holds, err := eval(check.Expression, row) if err != nil { return false, fmt.Errorf("evaluate CHECK constraint %s on %s: %w", check.Label(), table, err) } if !holds { return false, &CheckViolationError{Table: table, Constraint: check.Label()} } } 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) // Check cache if schema, ok := m.cache[tableLower]; ok { return schema, nil } // Read from storage key := m.schemaKey(table) var data string err := m.pool.WithClient(func(c *KVClient) error { var err error data, err = c.Read(key) return err }) if err != nil { return nil, fmt.Errorf("table not found: %s", table) } var schema Schema if err := json.Unmarshal([]byte(data), &schema); err != nil { return nil, err } schema.normalizePrimaryKey() m.cache[tableLower] = &schema return &schema, nil } // updateSchemaUnsafe updates a schema without locking (internal use). func (m *SchemaManager) updateSchemaUnsafe(schema *Schema) error { key := m.schemaKey(schema.Name) data, _ := json.Marshal(schema) err := m.pool.WithClient(func(c *KVClient) error { return c.Write(key, string(data)) }) if err != nil { return err } // Update cache m.cache[strings.ToLower(schema.Name)] = schema m.bumpVersionLocked() return nil }