schema.go 50 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824
  1. package storage
  2. import (
  3. "errors"
  4. "fmt"
  5. "strings"
  6. "sync"
  7. "time"
  8. "github.com/goccy/go-json"
  9. "github.com/danfragoso/pizzasql-next/pkg/analyzer"
  10. )
  11. // ErrIndexNotFound is returned by GetIndex when an index is absent. It is
  12. // distinct from a storage/IO error so callers like ListTableIndexes can skip a
  13. // concurrently-dropped index without swallowing real read failures.
  14. var ErrIndexNotFound = errors.New("index not found")
  15. // Schema represents a table schema.
  16. type Schema struct {
  17. Name string `json:"name"`
  18. Columns []Column `json:"columns"`
  19. PrimaryKey string `json:"primary_key"`
  20. CreatedAt time.Time `json:"created_at"`
  21. NextRowID int64 `json:"next_rowid"`
  22. AutoIncrement bool `json:"autoincrement"`
  23. // Checks holds the row predicates declared with CHECK, at column and table
  24. // level alike. Inline and table-level declarations are equivalent once
  25. // parsed, so both live in this one list.
  26. Checks []CheckConstraint `json:"checks,omitempty"`
  27. }
  28. // CheckConstraint is a persisted CHECK predicate. The predicate is stored as
  29. // SQL text and re-parsed by the evaluator installed on the SchemaManager, so a
  30. // durable reload keeps enforcing it.
  31. type CheckConstraint struct {
  32. Name string `json:"name,omitempty"`
  33. Expression string `json:"expression"`
  34. }
  35. // Label identifies the constraint in error messages and the catalog: the
  36. // declared name when there is one, otherwise the expression text.
  37. func (c CheckConstraint) Label() string {
  38. if c.Name != "" {
  39. return c.Name
  40. }
  41. return c.Expression
  42. }
  43. // Column represents a column definition.
  44. type Column struct {
  45. Name string `json:"name"`
  46. Type string `json:"type"`
  47. Nullable bool `json:"nullable"`
  48. // Default holds the evaluated constant DEFAULT. DefaultExpr holds the SQL
  49. // text of a non-constant DEFAULT (CURRENT_TIMESTAMP and the other
  50. // clock-reading forms) so it is evaluated on every write instead of being
  51. // frozen at DDL time. Only one of the two is set; DefaultExpr takes
  52. // precedence if a hand-edited catalog sets both.
  53. Default interface{} `json:"default,omitempty"`
  54. DefaultExpr string `json:"default_expr,omitempty"`
  55. PrimaryKey bool `json:"primary_key"`
  56. // GeneratedExpr holds the SQL text of a GENERATED ALWAYS AS (expr) column.
  57. // GeneratedStored reports whether the value is materialized on write
  58. // (STORED, the only form this engine persists) versus computed on read.
  59. GeneratedExpr string `json:"generated_expr,omitempty"`
  60. GeneratedStored bool `json:"generated_stored,omitempty"`
  61. }
  62. // DefaultLiteral returns the catalog representation of the column's default:
  63. // the evaluated constant value, or the SQL text of a non-constant default.
  64. // DefaultExpr takes precedence when a hand-edited catalog sets both.
  65. func (c Column) DefaultLiteral() interface{} {
  66. if c.DefaultExpr != "" {
  67. return c.DefaultExpr
  68. }
  69. if c.Default != nil {
  70. return c.Default
  71. }
  72. return nil
  73. }
  74. // Index represents an index definition.
  75. type Index struct {
  76. Name string `json:"name"`
  77. Table string `json:"table"`
  78. Columns []IndexColumn `json:"columns"`
  79. Unique bool `json:"unique"`
  80. CreatedAt time.Time `json:"created_at"`
  81. // OnConflict is the default conflict resolution declared for this index via
  82. // a UNIQUE(...) ON CONFLICT clause. Empty means the SQLite default (ABORT).
  83. OnConflict string `json:"on_conflict,omitempty"`
  84. }
  85. // IndexColumn represents a column in an index. Expression is set for expression
  86. // indexes (e.g. lower(email)); a plain column index leaves it empty and uses
  87. // Name.
  88. type IndexColumn struct {
  89. Name string `json:"name"`
  90. Desc bool `json:"desc"`
  91. Expression string `json:"expression,omitempty"`
  92. }
  93. // rowIDAllocator owns the next-ROWID state for a single table. It is a separate
  94. // mutex per table so allocating a ROWID on one table never serializes against
  95. // another table, and never contends with the SchemaManager catalog lock.
  96. type rowIDAllocator struct {
  97. mu sync.Mutex
  98. next int64
  99. init bool
  100. }
  101. // SchemaManager manages table schemas.
  102. type SchemaManager struct {
  103. pool *KVPool
  104. database string
  105. cache map[string]*Schema
  106. indexCache map[string]*Index
  107. indexListCache []string
  108. indexListCached bool
  109. version uint64
  110. mu sync.RWMutex
  111. tableLocksMu sync.Mutex
  112. tableLocks map[string]*sync.RWMutex
  113. // checkMu guards checkEval, the pluggable evaluator for persisted CHECK
  114. // predicates. It is set by the SQL executor so storage stays independent of
  115. // the SQL front end.
  116. checkMu sync.RWMutex
  117. checkEval CheckEvaluator
  118. rowIDMu sync.Mutex
  119. rowIDAlloc map[string]*rowIDAllocator
  120. // defaultEval evaluates a persisted non-constant DEFAULT expression. The
  121. // SQL executor registers one so storage can backfill ADD COLUMN rows and
  122. // complete row images without importing the parser or executor.
  123. defaultEvalMu sync.RWMutex
  124. defaultEval DefaultEvaluator
  125. }
  126. // DefaultEvaluator evaluates a persisted non-constant DEFAULT expression, such
  127. // as CURRENT_TIMESTAMP, to the value a row receives. Expression defaults are
  128. // never frozen at DDL time: each write reads the clock again.
  129. type DefaultEvaluator func(expression string) (interface{}, error)
  130. // SetDefaultEvaluator installs the evaluator used for expression defaults.
  131. func (m *SchemaManager) SetDefaultEvaluator(eval DefaultEvaluator) {
  132. m.defaultEvalMu.Lock()
  133. m.defaultEval = eval
  134. m.defaultEvalMu.Unlock()
  135. }
  136. func (m *SchemaManager) defaultEvaluator() DefaultEvaluator {
  137. m.defaultEvalMu.RLock()
  138. defer m.defaultEvalMu.RUnlock()
  139. return m.defaultEval
  140. }
  141. // CheckEvaluator reports whether a persisted CHECK expression holds for a row.
  142. // A false result is a violation; NULL and other unknown results must be
  143. // reported as true, matching SQL three-valued logic. The SQL executor registers
  144. // one so the storage layer can enforce CHECK without importing the parser.
  145. type CheckEvaluator func(expression string, row Row) (bool, error)
  146. // SetCheckEvaluator installs the evaluator used to enforce persisted CHECK
  147. // predicates.
  148. func (m *SchemaManager) SetCheckEvaluator(eval CheckEvaluator) {
  149. m.checkMu.Lock()
  150. m.checkEval = eval
  151. m.checkMu.Unlock()
  152. }
  153. func (m *SchemaManager) checkEvaluator() CheckEvaluator {
  154. m.checkMu.RLock()
  155. defer m.checkMu.RUnlock()
  156. return m.checkEval
  157. }
  158. // BeginTransaction is retained for API compatibility. Buffered per-session
  159. // transactions no longer take a database-wide transaction lock; staged writes
  160. // are validated and committed atomically with CompareBatchWrite instead.
  161. func (m *SchemaManager) BeginTransaction() {}
  162. // EndTransaction is retained for API compatibility.
  163. func (m *SchemaManager) EndTransaction() {}
  164. // LockStatement is retained for API compatibility. Statement execution is now
  165. // serialized through per-table locks and optimistic validation, so no global
  166. // statement lock is required.
  167. func (m *SchemaManager) LockStatement() {}
  168. // UnlockStatement is retained for API compatibility.
  169. func (m *SchemaManager) UnlockStatement() {}
  170. // NewSchemaManager creates a new schema manager.
  171. func NewSchemaManager(pool *KVPool, database string) *SchemaManager {
  172. return &SchemaManager{
  173. pool: pool,
  174. database: database,
  175. cache: make(map[string]*Schema),
  176. indexCache: make(map[string]*Index),
  177. tableLocks: make(map[string]*sync.RWMutex),
  178. rowIDAlloc: make(map[string]*rowIDAllocator),
  179. }
  180. }
  181. // rowIDAllocatorFor returns (creating if needed) the per-table ROWID allocator.
  182. func (m *SchemaManager) rowIDAllocatorFor(table string) *rowIDAllocator {
  183. key := strings.ToLower(table)
  184. m.rowIDMu.Lock()
  185. a, ok := m.rowIDAlloc[key]
  186. if !ok {
  187. a = &rowIDAllocator{}
  188. m.rowIDAlloc[key] = a
  189. }
  190. m.rowIDMu.Unlock()
  191. return a
  192. }
  193. func (m *SchemaManager) tableLock(table string) *sync.RWMutex {
  194. key := strings.ToLower(table)
  195. m.tableLocksMu.Lock()
  196. lock, ok := m.tableLocks[key]
  197. if !ok {
  198. lock = &sync.RWMutex{}
  199. m.tableLocks[key] = lock
  200. }
  201. m.tableLocksMu.Unlock()
  202. return lock
  203. }
  204. // GetDatabaseName returns the database name.
  205. func (m *SchemaManager) GetDatabaseName() string {
  206. return m.database
  207. }
  208. // GetPool returns the KV pool.
  209. func (m *SchemaManager) GetPool() *KVPool {
  210. return m.pool
  211. }
  212. // Version returns the in-process schema catalog version. It is incremented for
  213. // schema/index definition changes so cached executors can resync their analyzer
  214. // catalogs without scanning storage on every query.
  215. func (m *SchemaManager) Version() uint64 {
  216. m.mu.RLock()
  217. defer m.mu.RUnlock()
  218. return m.version
  219. }
  220. func (m *SchemaManager) bumpVersionLocked() {
  221. m.version++
  222. }
  223. // schemaKey returns the key for a table schema.
  224. func (m *SchemaManager) schemaKey(table string) string {
  225. return fmt.Sprintf("%s:_schema:%s", m.database, strings.ToLower(table))
  226. }
  227. // catalogKey returns the key for the table catalog.
  228. func (m *SchemaManager) catalogKey() string {
  229. return fmt.Sprintf("%s:_sys:tables", m.database)
  230. }
  231. // rowIDKey returns the key for a table's next ROWID counter.
  232. func (m *SchemaManager) rowIDKey(table string) string {
  233. return fmt.Sprintf("%s:_sys:rowid:%s", m.database, strings.ToLower(table))
  234. }
  235. // CreateTable creates a new table.
  236. func (m *SchemaManager) CreateTable(schema *Schema) error {
  237. m.mu.Lock()
  238. defer m.mu.Unlock()
  239. // Check if table already exists
  240. key := m.schemaKey(schema.Name)
  241. err := m.pool.WithClient(func(c *KVClient) error {
  242. _, err := c.Read(key)
  243. return err
  244. })
  245. if err == nil {
  246. return fmt.Errorf("table already exists: %s", schema.Name)
  247. }
  248. // Keep the cached schema private so callers cannot mutate a published
  249. // catalog snapshot after this operation returns.
  250. schema = cloneSchema(schema)
  251. schema.CreatedAt = time.Now()
  252. // Determine primary key if not set
  253. if schema.PrimaryKey == "" {
  254. for _, col := range schema.Columns {
  255. if col.PrimaryKey {
  256. schema.PrimaryKey = col.Name
  257. break
  258. }
  259. }
  260. // No explicit primary key declared — use synthetic _rowid_ so user
  261. // columns remain unconstrained and can hold duplicate or NULL values.
  262. if schema.PrimaryKey == "" {
  263. schema.PrimaryKey = "_rowid_"
  264. }
  265. }
  266. // Serialize schema
  267. data, err := json.Marshal(schema)
  268. if err != nil {
  269. return fmt.Errorf("failed to serialize schema: %w", err)
  270. }
  271. // Write schema
  272. err = m.pool.WithClient(func(c *KVClient) error {
  273. return c.Write(key, string(data))
  274. })
  275. if err != nil {
  276. return fmt.Errorf("failed to write schema: %w", err)
  277. }
  278. // Update catalog
  279. if err := m.addToCatalog(schema.Name); err != nil {
  280. // Rollback schema write
  281. m.pool.WithClient(func(c *KVClient) error {
  282. return c.Delete(key)
  283. })
  284. return err
  285. }
  286. // Update cache
  287. m.cache[strings.ToLower(schema.Name)] = schema
  288. m.bumpVersionLocked()
  289. return nil
  290. }
  291. // DropTable drops a table.
  292. func (m *SchemaManager) DropTable(name string) error {
  293. tableLock := m.tableLock(name)
  294. tableLock.Lock()
  295. defer tableLock.Unlock()
  296. m.mu.Lock()
  297. defer m.mu.Unlock()
  298. key := m.schemaKey(name)
  299. // Check if table exists
  300. err := m.pool.WithClient(func(c *KVClient) error {
  301. _, err := c.Read(key)
  302. return err
  303. })
  304. if err != nil {
  305. return fmt.Errorf("table not found: %s", name)
  306. }
  307. // Delete all rows by scanning their actual keys and batch-deleting them, so
  308. // a table drop no longer leaks durable rows.
  309. if err := m.deleteKeysWithPrefix([]byte(fmt.Sprintf("%s:_data:%s:", m.database, strings.ToLower(name)))); err != nil {
  310. return err
  311. }
  312. // Delete schema
  313. err = m.pool.WithClient(func(c *KVClient) error {
  314. return c.Delete(key)
  315. })
  316. if err != nil {
  317. return fmt.Errorf("failed to delete schema: %w", err)
  318. }
  319. // Delete ROWID state.
  320. m.pool.WithClient(func(c *KVClient) error {
  321. return c.Delete(m.rowIDKey(name))
  322. })
  323. // Update catalog
  324. if err := m.removeFromCatalog(name); err != nil {
  325. return err
  326. }
  327. // Update cache
  328. tableLower := strings.ToLower(name)
  329. delete(m.cache, tableLower)
  330. m.rowIDMu.Lock()
  331. delete(m.rowIDAlloc, tableLower)
  332. m.rowIDMu.Unlock()
  333. m.bumpVersionLocked()
  334. return nil
  335. }
  336. // GetSchema retrieves a table schema.
  337. func (m *SchemaManager) GetSchema(name string) (*Schema, error) {
  338. m.mu.RLock()
  339. if schema, ok := m.cache[strings.ToLower(name)]; ok {
  340. // Clone while still holding the read lock: the cached schema's
  341. // NextRowID field is mutated under the write lock, so cloning outside
  342. // the lock races with that mutation.
  343. cloned := cloneSchema(schema)
  344. m.mu.RUnlock()
  345. return cloned, nil
  346. }
  347. m.mu.RUnlock()
  348. m.mu.Lock()
  349. defer m.mu.Unlock()
  350. // Double-check after acquiring write lock
  351. if schema, ok := m.cache[strings.ToLower(name)]; ok {
  352. return cloneSchema(schema), nil
  353. }
  354. key := m.schemaKey(name)
  355. var data string
  356. err := m.pool.WithClient(func(c *KVClient) error {
  357. var err error
  358. data, err = c.Read(key)
  359. return err
  360. })
  361. if err != nil {
  362. if err == ErrKeyNotFound {
  363. return nil, fmt.Errorf("table not found: %s", name)
  364. }
  365. return nil, err
  366. }
  367. var schema Schema
  368. if err := json.Unmarshal([]byte(data), &schema); err != nil {
  369. return nil, fmt.Errorf("failed to parse schema: %w", err)
  370. }
  371. m.cache[strings.ToLower(name)] = &schema
  372. return cloneSchema(&schema), nil
  373. }
  374. func cloneSchema(schema *Schema) *Schema {
  375. if schema == nil {
  376. return nil
  377. }
  378. cloned := *schema
  379. cloned.Columns = append([]Column(nil), schema.Columns...)
  380. cloned.Checks = append([]CheckConstraint(nil), schema.Checks...)
  381. return &cloned
  382. }
  383. func cloneIndex(index *Index) *Index {
  384. if index == nil {
  385. return nil
  386. }
  387. cloned := *index
  388. cloned.Columns = append([]IndexColumn(nil), index.Columns...)
  389. return &cloned
  390. }
  391. // TableExists checks if a table exists.
  392. func (m *SchemaManager) TableExists(name string) bool {
  393. _, err := m.GetSchema(name)
  394. return err == nil
  395. }
  396. // ListTables returns all table names.
  397. func (m *SchemaManager) ListTables() ([]string, error) {
  398. var data string
  399. err := m.pool.WithClient(func(c *KVClient) error {
  400. var err error
  401. data, err = c.Read(m.catalogKey())
  402. return err
  403. })
  404. if err != nil {
  405. if err == ErrKeyNotFound {
  406. return nil, nil
  407. }
  408. return nil, err
  409. }
  410. var tables []string
  411. if err := json.Unmarshal([]byte(data), &tables); err != nil {
  412. return nil, fmt.Errorf("failed to parse catalog: %w", err)
  413. }
  414. return tables, nil
  415. }
  416. // addToCatalog adds a table to the catalog.
  417. func (m *SchemaManager) addToCatalog(name string) error {
  418. tables, err := m.ListTables()
  419. if err != nil && err != ErrKeyNotFound {
  420. return err
  421. }
  422. // Check if already exists
  423. lowerName := strings.ToLower(name)
  424. for _, t := range tables {
  425. if strings.ToLower(t) == lowerName {
  426. return nil
  427. }
  428. }
  429. tables = append(tables, name)
  430. data, err := json.Marshal(tables)
  431. if err != nil {
  432. return err
  433. }
  434. return m.pool.WithClient(func(c *KVClient) error {
  435. return c.Write(m.catalogKey(), string(data))
  436. })
  437. }
  438. // removeFromCatalog removes a table from the catalog.
  439. func (m *SchemaManager) removeFromCatalog(name string) error {
  440. tables, err := m.ListTables()
  441. if err != nil {
  442. return err
  443. }
  444. lowerName := strings.ToLower(name)
  445. newTables := make([]string, 0, len(tables))
  446. for _, t := range tables {
  447. if strings.ToLower(t) != lowerName {
  448. newTables = append(newTables, t)
  449. }
  450. }
  451. data, err := json.Marshal(newTables)
  452. if err != nil {
  453. return err
  454. }
  455. return m.pool.WithClient(func(c *KVClient) error {
  456. return c.Write(m.catalogKey(), string(data))
  457. })
  458. }
  459. // InvalidateCache clears the cache for a table.
  460. func (m *SchemaManager) InvalidateCache(name string) {
  461. m.mu.Lock()
  462. defer m.mu.Unlock()
  463. tableLower := strings.ToLower(name)
  464. delete(m.cache, tableLower)
  465. m.rowIDMu.Lock()
  466. delete(m.rowIDAlloc, tableLower)
  467. m.rowIDMu.Unlock()
  468. }
  469. // ToAnalyzerTableInfo converts a Schema to analyzer.TableInfo.
  470. func (s *Schema) ToAnalyzerTableInfo() *analyzer.TableInfo {
  471. info := &analyzer.TableInfo{
  472. Name: s.Name,
  473. }
  474. for _, col := range s.Columns {
  475. info.Columns = append(info.Columns, analyzer.ColumnInfo{
  476. Name: col.Name,
  477. Type: analyzer.TypeFromName(col.Type),
  478. Nullable: col.Nullable,
  479. PrimaryKey: col.PrimaryKey,
  480. TableName: s.Name,
  481. Generated: col.GeneratedExpr != "",
  482. })
  483. }
  484. return info
  485. }
  486. // GetColumn returns a column by name.
  487. func (s *Schema) GetColumn(name string) (*Column, bool) {
  488. lowerName := strings.ToLower(name)
  489. for i := range s.Columns {
  490. if strings.ToLower(s.Columns[i].Name) == lowerName {
  491. return &s.Columns[i], true
  492. }
  493. }
  494. return nil, false
  495. }
  496. // GetNextRowID gets and increments the next ROWID for a table. Allocation is
  497. // serialized per table via the table's own allocator so inserts on different
  498. // tables never contend, and no global SchemaManager lock is held across the
  499. // durable derivation scan.
  500. func (m *SchemaManager) GetNextRowID(table string) (int64, error) {
  501. alloc := m.rowIDAllocatorFor(table)
  502. alloc.mu.Lock()
  503. defer alloc.mu.Unlock()
  504. next, err := m.nextRowIDLocked(alloc, table)
  505. if err != nil {
  506. return 0, err
  507. }
  508. alloc.next = next + 1
  509. return next, nil
  510. }
  511. // UpdateMaxRowID updates the next ROWID if the provided value is higher.
  512. func (m *SchemaManager) UpdateMaxRowID(table string, rowid int64) error {
  513. alloc := m.rowIDAllocatorFor(table)
  514. alloc.mu.Lock()
  515. defer alloc.mu.Unlock()
  516. next, err := m.nextRowIDLocked(alloc, table)
  517. if err != nil {
  518. return err
  519. }
  520. if rowid >= next {
  521. alloc.next = rowid + 1
  522. }
  523. return nil
  524. }
  525. // nextRowIDLocked returns the current next ROWID, deriving it from durable rows
  526. // on first use (must hold the per-table allocator lock).
  527. func (m *SchemaManager) nextRowIDLocked(alloc *rowIDAllocator, table string) (int64, error) {
  528. if alloc.init {
  529. return alloc.next, nil
  530. }
  531. next, err := m.deriveNextRowID(table)
  532. if err != nil {
  533. return 0, err
  534. }
  535. if next < 1 {
  536. next = 1
  537. }
  538. alloc.next = next
  539. alloc.init = true
  540. return alloc.next, nil
  541. }
  542. // deriveNextRowID scans durable row keys to recover max(rowid)+1, streaming
  543. // each page through decodeRow instead of materializing every value.
  544. func (m *SchemaManager) deriveNextRowID(table string) (int64, error) {
  545. prefix := []byte(fmt.Sprintf("%s:_data:%s:", m.database, strings.ToLower(table)))
  546. var maxRowID int64
  547. err := m.pool.WithClient(func(client *KVClient) (retErr error) {
  548. cursor, err := client.Scan(prefix)
  549. if err != nil {
  550. return err
  551. }
  552. defer func() {
  553. if err := cursor.Close(); retErr == nil {
  554. retErr = err
  555. }
  556. }()
  557. for {
  558. entries, done, err := cursor.Next()
  559. if err != nil {
  560. return err
  561. }
  562. for _, e := range entries {
  563. row, err := decodeRow(e.Value)
  564. if err != nil {
  565. return fmt.Errorf("failed to parse row while deriving ROWID: %w", err)
  566. }
  567. if rowid, ok := valueAsInt64(row["_rowid_"]); ok && rowid > maxRowID {
  568. maxRowID = rowid
  569. }
  570. }
  571. if done {
  572. return nil
  573. }
  574. }
  575. })
  576. if err != nil {
  577. return 0, err
  578. }
  579. return maxRowID + 1, nil
  580. }
  581. // deleteKeysWithPrefix streams the keys with the given prefix one page at a
  582. // time and atomically batch-deletes them, so a bulk operation never leaves
  583. // durable rows behind.
  584. func (m *SchemaManager) deleteKeysWithPrefix(prefix []byte) error {
  585. return m.pool.WithClient(func(client *KVClient) (retErr error) {
  586. cursor, err := client.ScanKeys(prefix)
  587. if err != nil {
  588. return err
  589. }
  590. defer func() {
  591. if err := cursor.Close(); retErr == nil {
  592. retErr = err
  593. }
  594. }()
  595. ops := make([]BatchOp, 0, scanPageSize)
  596. batchBytes := 8
  597. flush := func() error {
  598. if len(ops) == 0 {
  599. return nil
  600. }
  601. if _, err := client.BatchWrite(ops, nil); err != nil {
  602. return err
  603. }
  604. ops = ops[:0]
  605. batchBytes = 8
  606. return nil
  607. }
  608. for {
  609. entries, done, err := cursor.Next()
  610. if err != nil {
  611. return err
  612. }
  613. for _, e := range entries {
  614. opBytes := 12 + len(e.Key)
  615. if len(ops) == maxOperations || batchBytes+opBytes > bulkBatchByteBudget {
  616. if err := flush(); err != nil {
  617. return err
  618. }
  619. }
  620. ops = append(ops, BatchOp{Op: batchDelete, Key: append([]byte(nil), e.Key...)})
  621. batchBytes += opBytes
  622. }
  623. if done {
  624. return flush()
  625. }
  626. }
  627. })
  628. }
  629. func valueAsInt64(value interface{}) (int64, bool) {
  630. switch v := value.(type) {
  631. case int64:
  632. return v, true
  633. case int:
  634. return int64(v), true
  635. case float64:
  636. return int64(v), true
  637. default:
  638. return 0, false
  639. }
  640. }
  641. // getSchemaLocked retrieves schema (must hold lock).
  642. func (m *SchemaManager) getSchemaLocked(name string) (*Schema, error) {
  643. if schema, ok := m.cache[strings.ToLower(name)]; ok {
  644. return schema, nil
  645. }
  646. key := m.schemaKey(name)
  647. var data string
  648. err := m.pool.WithClient(func(c *KVClient) error {
  649. var err error
  650. data, err = c.Read(key)
  651. return err
  652. })
  653. if err != nil {
  654. if err == ErrKeyNotFound {
  655. return nil, fmt.Errorf("table not found: %s", name)
  656. }
  657. return nil, err
  658. }
  659. var schema Schema
  660. if err := json.Unmarshal([]byte(data), &schema); err != nil {
  661. return nil, fmt.Errorf("failed to parse schema: %w", err)
  662. }
  663. m.cache[strings.ToLower(name)] = &schema
  664. return &schema, nil
  665. }
  666. // saveSchemaLocked saves schema (must hold lock).
  667. func (m *SchemaManager) saveSchemaLocked(schema *Schema) error {
  668. data, err := json.Marshal(schema)
  669. if err != nil {
  670. return fmt.Errorf("failed to serialize schema: %w", err)
  671. }
  672. key := m.schemaKey(schema.Name)
  673. err = m.pool.WithClient(func(c *KVClient) error {
  674. return c.Write(key, string(data))
  675. })
  676. if err != nil {
  677. return fmt.Errorf("failed to write schema: %w", err)
  678. }
  679. m.cache[strings.ToLower(schema.Name)] = schema
  680. return nil
  681. }
  682. // Index management methods
  683. // indexKey returns the key for an index.
  684. func (m *SchemaManager) indexKey(name string) string {
  685. return fmt.Sprintf("%s:index:%s", m.database, strings.ToLower(name))
  686. }
  687. // indexListKey returns the key for the index list.
  688. func (m *SchemaManager) indexListKey() string {
  689. return fmt.Sprintf("%s:indexes", m.database)
  690. }
  691. // CreateIndex creates a new index.
  692. func (m *SchemaManager) CreateIndex(index *Index) error {
  693. m.mu.Lock()
  694. defer m.mu.Unlock()
  695. // Check if index already exists
  696. key := m.indexKey(index.Name)
  697. err := m.pool.WithClient(func(c *KVClient) error {
  698. _, err := c.Read(key)
  699. return err
  700. })
  701. if err == nil {
  702. return fmt.Errorf("index already exists: %s", index.Name)
  703. }
  704. // Verify table exists
  705. if _, err := m.getSchemaLocked(index.Table); err != nil {
  706. return fmt.Errorf("table not found: %s", index.Table)
  707. }
  708. // Save index
  709. index.CreatedAt = time.Now()
  710. data, err := json.Marshal(index)
  711. if err != nil {
  712. return fmt.Errorf("failed to serialize index: %w", err)
  713. }
  714. err = m.pool.WithClient(func(c *KVClient) error {
  715. return c.Write(key, string(data))
  716. })
  717. if err != nil {
  718. return fmt.Errorf("failed to write index: %w", err)
  719. }
  720. // Add to index list
  721. if err := m.addToIndexList(index.Name); err != nil {
  722. return err
  723. }
  724. m.indexCache[strings.ToLower(index.Name)] = cloneIndex(index)
  725. m.bumpVersionLocked()
  726. return nil
  727. }
  728. // DropIndex drops an index.
  729. func (m *SchemaManager) DropIndex(name string) error {
  730. m.mu.Lock()
  731. defer m.mu.Unlock()
  732. key := m.indexKey(name)
  733. err := m.pool.WithClient(func(c *KVClient) error {
  734. return c.Delete(key)
  735. })
  736. if err != nil {
  737. return fmt.Errorf("failed to delete index: %w", err)
  738. }
  739. if err := m.removeFromIndexList(name); err != nil {
  740. return err
  741. }
  742. delete(m.indexCache, strings.ToLower(name))
  743. m.bumpVersionLocked()
  744. return nil
  745. }
  746. // IndexExists checks if an index exists.
  747. func (m *SchemaManager) IndexExists(name string) bool {
  748. m.mu.RLock()
  749. defer m.mu.RUnlock()
  750. return m.indexExistsUnsafe(name)
  751. }
  752. // indexExistsUnsafe reports whether an index definition is durable (m.mu must be
  753. // held).
  754. func (m *SchemaManager) indexExistsUnsafe(name string) bool {
  755. if _, ok := m.indexCache[strings.ToLower(name)]; ok {
  756. return true
  757. }
  758. key := m.indexKey(name)
  759. err := m.pool.WithClient(func(c *KVClient) error {
  760. _, err := c.Read(key)
  761. return err
  762. })
  763. return err == nil
  764. }
  765. // GetIndex retrieves an index by name.
  766. func (m *SchemaManager) GetIndex(name string) (*Index, error) {
  767. m.mu.Lock()
  768. defer m.mu.Unlock()
  769. cacheKey := strings.ToLower(name)
  770. if index, ok := m.indexCache[cacheKey]; ok {
  771. return cloneIndex(index), nil
  772. }
  773. key := m.indexKey(name)
  774. var data string
  775. err := m.pool.WithClient(func(c *KVClient) error {
  776. var err error
  777. data, err = c.Read(key)
  778. return err
  779. })
  780. if err != nil {
  781. if err == ErrKeyNotFound {
  782. return nil, fmt.Errorf("%w: %s", ErrIndexNotFound, name)
  783. }
  784. return nil, err
  785. }
  786. var index Index
  787. if err := json.Unmarshal([]byte(data), &index); err != nil {
  788. return nil, fmt.Errorf("failed to parse index: %w", err)
  789. }
  790. m.indexCache[cacheKey] = &index
  791. return cloneIndex(&index), nil
  792. }
  793. // ListIndexes returns all index names.
  794. func (m *SchemaManager) ListIndexes() ([]string, error) {
  795. m.mu.Lock()
  796. defer m.mu.Unlock()
  797. return m.indexListUnsafe()
  798. }
  799. // indexListUnsafe returns all index names, loading them from storage on first
  800. // use (m.mu must be held).
  801. func (m *SchemaManager) indexListUnsafe() ([]string, error) {
  802. if m.indexListCached {
  803. return append([]string(nil), m.indexListCache...), nil
  804. }
  805. key := m.indexListKey()
  806. var data string
  807. err := m.pool.WithClient(func(c *KVClient) error {
  808. var err error
  809. data, err = c.Read(key)
  810. return err
  811. })
  812. if err == ErrKeyNotFound {
  813. m.indexListCache = nil
  814. m.indexListCached = true
  815. return []string{}, nil
  816. }
  817. if err != nil {
  818. return nil, err
  819. }
  820. var indexes []string
  821. if err := json.Unmarshal([]byte(data), &indexes); err != nil {
  822. return nil, fmt.Errorf("failed to parse index list: %w", err)
  823. }
  824. m.indexListCache = append([]string(nil), indexes...)
  825. m.indexListCached = true
  826. return indexes, nil
  827. }
  828. // ListTableIndexes returns all indexes for a table.
  829. func (m *SchemaManager) ListTableIndexes(table string) ([]*Index, error) {
  830. indexes, err := m.ListIndexes()
  831. if err != nil {
  832. return nil, err
  833. }
  834. var result []*Index
  835. for _, name := range indexes {
  836. idx, err := m.GetIndex(name)
  837. if err != nil {
  838. // An index dropped concurrently is simply absent; any other error
  839. // (storage/IO or corrupt data) must not be silently swallowed.
  840. if errors.Is(err, ErrIndexNotFound) {
  841. continue
  842. }
  843. return nil, err
  844. }
  845. if strings.EqualFold(idx.Table, table) {
  846. result = append(result, idx)
  847. }
  848. }
  849. return result, nil
  850. }
  851. // addToIndexList adds an index name to the list.
  852. func (m *SchemaManager) addToIndexList(name string) error {
  853. key := m.indexListKey()
  854. var indexes []string
  855. var data string
  856. err := m.pool.WithClient(func(c *KVClient) error {
  857. var err error
  858. data, err = c.Read(key)
  859. return err
  860. })
  861. if err == nil {
  862. json.Unmarshal([]byte(data), &indexes)
  863. }
  864. indexes = append(indexes, name)
  865. newData, _ := json.Marshal(indexes)
  866. err = m.pool.WithClient(func(c *KVClient) error {
  867. return c.Write(key, string(newData))
  868. })
  869. if err == nil {
  870. m.indexListCache = append([]string(nil), indexes...)
  871. m.indexListCached = true
  872. }
  873. return err
  874. }
  875. // removeFromIndexList removes an index name from the list.
  876. func (m *SchemaManager) removeFromIndexList(name string) error {
  877. key := m.indexListKey()
  878. var indexes []string
  879. var data string
  880. err := m.pool.WithClient(func(c *KVClient) error {
  881. var err error
  882. data, err = c.Read(key)
  883. return err
  884. })
  885. if err != nil {
  886. return nil
  887. }
  888. json.Unmarshal([]byte(data), &indexes)
  889. var newIndexes []string
  890. for _, idx := range indexes {
  891. if !strings.EqualFold(idx, name) {
  892. newIndexes = append(newIndexes, idx)
  893. }
  894. }
  895. newData, _ := json.Marshal(newIndexes)
  896. err = m.pool.WithClient(func(c *KVClient) error {
  897. return c.Write(key, string(newData))
  898. })
  899. if err == nil {
  900. m.indexListCache = append([]string(nil), newIndexes...)
  901. m.indexListCached = true
  902. }
  903. return err
  904. }
  905. // AddColumn adds a new column to a table. When unique is non-nil the column's
  906. // UNIQUE index is registered as part of the same statement, and checks are
  907. // persisted with the column in the same schema snapshot.
  908. //
  909. // A DEFAULT is materialized into every existing row before the schema is
  910. // published. NOT NULL, UNIQUE, CHECK, and index validation all read persisted
  911. // row values, so a default that only existed in the catalog would be invisible
  912. // to them: a UNIQUE index over the new column could not see the duplicate
  913. // values every existing row shares, a CHECK predicate could not see a violating
  914. // default, and an UPDATE to NULL would look non-null. The rewrite is the same
  915. // shape as DropColumn: the table's exclusive gate keeps point writes and CAS
  916. // commits out while rows are scanned, validated, and rewritten, the transform
  917. // is idempotent, and the schema is saved last so an interrupted operation
  918. // resumes against the old schema.
  919. //
  920. // The gate also makes the validation authoritative: NOT NULL without a default,
  921. // duplicate UNIQUE values, and rows a new CHECK predicate rejects are rejected
  922. // before the column, its checks, and its index are published, and they are
  923. // published together in one batch write so a crash cannot leave the column
  924. // visible without the constraints that validate it. A constant default is
  925. // deterministic, so its UNIQUE prediction can run as a cheap pre-scan. A
  926. // clock-reading default cannot be predicted: the UNIQUE duplicate set is
  927. // collected during the rewrite itself, because a separate scan could observe a
  928. // different instant and publish an index over duplicate values; the CHECK
  929. // pre-scan resolves the same clock default for each row it validates. A
  930. // nullable column without a default needs no rewrite because a missing row
  931. // field already reads as NULL. Like DROP COLUMN, the statement holds the
  932. // catalog lock for the whole scan, so the cost is one table scan per validating
  933. // constraint plus a table-wide stall per ADD COLUMN that has a default, an
  934. // index, or a check.
  935. func (m *SchemaManager) AddColumn(table string, column Column, checks []CheckConstraint, unique *Index) error {
  936. tableLock := m.tableLock(table)
  937. tableLock.Lock()
  938. defer tableLock.Unlock()
  939. m.mu.Lock()
  940. defer m.mu.Unlock()
  941. schema, err := m.getSchemaUnsafe(table)
  942. if err != nil {
  943. return err
  944. }
  945. if err := checkColumnAbsent(schema, column.Name); err != nil {
  946. return err
  947. }
  948. if unique != nil {
  949. if !strings.EqualFold(unique.Table, schema.Name) {
  950. return fmt.Errorf("index %s belongs to table %s, not %s", unique.Name, unique.Table, schema.Name)
  951. }
  952. if m.indexExistsUnsafe(unique.Name) {
  953. return fmt.Errorf("index already exists: %s", unique.Name)
  954. }
  955. // A clock-reading default is validated during the rewrite below, which
  956. // samples the same values it writes. Predicting them in a separate scan
  957. // could disagree across a second boundary and publish duplicates.
  958. if column.DefaultExpr == "" {
  959. if err := m.validateAddColumnUnique(table, column, unique); err != nil {
  960. return err
  961. }
  962. }
  963. }
  964. if err := m.validateAddColumnChecks(table, column, checks); err != nil {
  965. return err
  966. }
  967. // A required column with nothing to write into rows that already exist
  968. // cannot be declared on a populated table.
  969. if !column.Nullable && !columnHasDefault(column) {
  970. hasRows, err := m.tableHasRows(table)
  971. if err != nil {
  972. return err
  973. }
  974. if hasRows {
  975. return &NotNullViolationError{Table: schema.Name, Column: column.Name}
  976. }
  977. }
  978. // An expression default is evaluated per rewritten row. When the column is
  979. // also UNIQUE, collect the values this rewrite writes and abort on a
  980. // duplicate, so validation cannot disagree with the published rows.
  981. var seen map[string]struct{}
  982. if unique != nil && column.DefaultExpr != "" {
  983. seen = make(map[string]struct{})
  984. }
  985. if columnHasDefault(column) {
  986. if err := m.applyRows(table, func(row Row) (bool, error) {
  987. value, present := rowColumnValue(row, column.Name)
  988. if !present {
  989. var err error
  990. value, _, err = m.columnDefaultValue(column)
  991. if err != nil {
  992. return false, err
  993. }
  994. row[column.Name] = value
  995. }
  996. if seen != nil {
  997. if value == nil {
  998. return !present, nil
  999. }
  1000. encoded := encodeUniqueScalar(value)
  1001. if _, duplicate := seen[encoded]; duplicate {
  1002. return false, &UniqueViolationError{Constraint: unique.Name}
  1003. }
  1004. seen[encoded] = struct{}{}
  1005. }
  1006. return !present, nil
  1007. }); err != nil {
  1008. return err
  1009. }
  1010. }
  1011. return m.publishAddedColumn(schema, column, checks, unique)
  1012. }
  1013. // publishAddedColumn publishes a column, its checks, and, when present, its
  1014. // unique index in one atomic batch write (m.mu must be held): the schema
  1015. // snapshot, the index definition, and the index list always become durable
  1016. // together or not at all.
  1017. func (m *SchemaManager) publishAddedColumn(schema *Schema, column Column, checks []CheckConstraint, unique *Index) error {
  1018. // Publish a fresh snapshot instead of mutating readers' shared pointer.
  1019. schema = cloneSchema(schema)
  1020. schema.Columns = append(schema.Columns, column)
  1021. schema.Checks = append(schema.Checks, checks...)
  1022. schemaData, err := json.Marshal(schema)
  1023. if err != nil {
  1024. return fmt.Errorf("failed to serialize schema: %w", err)
  1025. }
  1026. ops := []BatchOp{{Op: batchPut, Key: []byte(m.schemaKey(schema.Name)), Value: schemaData}}
  1027. var indexList []string
  1028. if unique != nil {
  1029. unique.CreatedAt = time.Now()
  1030. indexData, err := json.Marshal(unique)
  1031. if err != nil {
  1032. return fmt.Errorf("failed to serialize index: %w", err)
  1033. }
  1034. indexList, err = m.indexListUnsafe()
  1035. if err != nil {
  1036. return err
  1037. }
  1038. indexList = append(indexList, unique.Name)
  1039. listData, err := json.Marshal(indexList)
  1040. if err != nil {
  1041. return fmt.Errorf("failed to serialize index list: %w", err)
  1042. }
  1043. ops = append(ops,
  1044. BatchOp{Op: batchPut, Key: []byte(m.indexKey(unique.Name)), Value: indexData},
  1045. BatchOp{Op: batchPut, Key: []byte(m.indexListKey()), Value: listData},
  1046. )
  1047. }
  1048. if err := m.pool.WithClient(func(c *KVClient) error {
  1049. _, err := c.BatchWrite(ops, nil)
  1050. return err
  1051. }); err != nil {
  1052. return fmt.Errorf("failed to publish column %s: %w", column.Name, err)
  1053. }
  1054. // Publish in-memory state only after the durable write.
  1055. m.cache[strings.ToLower(schema.Name)] = schema
  1056. if unique != nil {
  1057. m.indexCache[strings.ToLower(unique.Name)] = cloneIndex(unique)
  1058. m.indexListCache = append([]string(nil), indexList...)
  1059. m.indexListCached = true
  1060. }
  1061. m.bumpVersionLocked()
  1062. return nil
  1063. }
  1064. func checkColumnAbsent(schema *Schema, name string) error {
  1065. for _, col := range schema.Columns {
  1066. if strings.EqualFold(col.Name, name) {
  1067. return fmt.Errorf("column already exists: %s", name)
  1068. }
  1069. }
  1070. return nil
  1071. }
  1072. // DropColumn removes a column from a table.
  1073. func (m *SchemaManager) DropColumn(table, columnName string) error {
  1074. tableLock := m.tableLock(table)
  1075. tableLock.Lock()
  1076. defer tableLock.Unlock()
  1077. m.mu.Lock()
  1078. defer m.mu.Unlock()
  1079. schema, err := m.getSchemaUnsafe(table)
  1080. if err != nil {
  1081. return err
  1082. }
  1083. // Cannot drop primary key column
  1084. if strings.EqualFold(schema.PrimaryKey, columnName) {
  1085. return fmt.Errorf("cannot drop primary key column: %s", columnName)
  1086. }
  1087. // Find and remove column
  1088. newColumns := make([]Column, 0, len(schema.Columns)-1)
  1089. found := false
  1090. for _, col := range schema.Columns {
  1091. if strings.EqualFold(col.Name, columnName) {
  1092. found = true
  1093. continue
  1094. }
  1095. newColumns = append(newColumns, col)
  1096. }
  1097. if !found {
  1098. return fmt.Errorf("column not found: %s", columnName)
  1099. }
  1100. schema = cloneSchema(schema)
  1101. schema.Columns = newColumns
  1102. if err := m.rewriteRows(table, func(row Row) bool {
  1103. for key := range row {
  1104. if strings.EqualFold(key, columnName) {
  1105. delete(row, key)
  1106. return true
  1107. }
  1108. }
  1109. return false
  1110. }); err != nil {
  1111. return err
  1112. }
  1113. // Update schema
  1114. return m.updateSchemaUnsafe(schema)
  1115. }
  1116. // RenameTable renames a table.
  1117. func (m *SchemaManager) RenameTable(oldName, newName string) error {
  1118. m.mu.Lock()
  1119. defer m.mu.Unlock()
  1120. // Check if old table exists
  1121. schema, err := m.getSchemaUnsafe(oldName)
  1122. if err != nil {
  1123. return err
  1124. }
  1125. // Check if new table name already exists
  1126. _, err = m.getSchemaUnsafe(newName)
  1127. if err == nil {
  1128. return fmt.Errorf("table already exists: %s", newName)
  1129. }
  1130. schema = cloneSchema(schema)
  1131. // Update schema name
  1132. schema.Name = newName
  1133. // Move durable rows from the old table's data prefix to the new one. Without
  1134. // this, ALTER TABLE ... RENAME leaves every row under the old key and the
  1135. // renamed table appears empty.
  1136. if err := m.renameDataKeys(oldName, newName); err != nil {
  1137. return err
  1138. }
  1139. // Repoint indexes that belonged to the old table.
  1140. if err := m.repointIndexesLocked(oldName, newName); err != nil {
  1141. return err
  1142. }
  1143. // Build the replacement catalog before switching schema names. The schema,
  1144. // catalog, and stale ROWID state change in one batch, so a crash cannot
  1145. // leave neither table name addressable after all row chunks have moved.
  1146. tables, err := m.ListTables()
  1147. if err != nil {
  1148. return err
  1149. }
  1150. foundOld := false
  1151. for i, name := range tables {
  1152. if strings.EqualFold(name, oldName) {
  1153. tables[i] = newName
  1154. foundOld = true
  1155. break
  1156. }
  1157. }
  1158. if !foundOld {
  1159. return fmt.Errorf("table not found in catalog: %s", oldName)
  1160. }
  1161. catalogData, err := json.Marshal(tables)
  1162. if err != nil {
  1163. return err
  1164. }
  1165. schemaData, err := json.Marshal(schema)
  1166. if err != nil {
  1167. return err
  1168. }
  1169. oldKey := m.schemaKey(oldName)
  1170. newKey := m.schemaKey(newName)
  1171. err = m.pool.WithClient(func(c *KVClient) error {
  1172. _, err := c.BatchWrite([]BatchOp{
  1173. {Op: batchPut, Key: []byte(newKey), Value: schemaData},
  1174. {Op: batchDelete, Key: []byte(oldKey)},
  1175. {Op: batchDelete, Key: []byte(m.rowIDKey(oldName))},
  1176. {Op: batchPut, Key: []byte(m.catalogKey()), Value: catalogData},
  1177. }, nil)
  1178. return err
  1179. })
  1180. if err != nil {
  1181. return err
  1182. }
  1183. // Update cache
  1184. oldLower := strings.ToLower(oldName)
  1185. newLower := strings.ToLower(newName)
  1186. delete(m.cache, oldLower)
  1187. m.rowIDMu.Lock()
  1188. delete(m.rowIDAlloc, oldLower)
  1189. m.rowIDMu.Unlock()
  1190. // Update cache
  1191. m.cache[newLower] = schema
  1192. m.bumpVersionLocked()
  1193. return nil
  1194. }
  1195. // renameDataKeys moves every durable row of a table from the old name's data
  1196. // prefix to the new name's prefix in a single atomic batch per page.
  1197. func (m *SchemaManager) renameDataKeys(oldName, newName string) error {
  1198. oldPrefix := []byte(fmt.Sprintf("%s:_data:%s:", m.database, strings.ToLower(oldName)))
  1199. newPrefix := fmt.Sprintf("%s:_data:%s:", m.database, strings.ToLower(newName))
  1200. return m.pool.WithClient(func(client *KVClient) (retErr error) {
  1201. cursor, err := client.Scan(oldPrefix)
  1202. if err != nil {
  1203. return err
  1204. }
  1205. defer func() {
  1206. if cerr := cursor.Close(); retErr == nil {
  1207. retErr = cerr
  1208. }
  1209. }()
  1210. ops := make([]BatchOp, 0, scanPageSize*2)
  1211. batchBytes := 8
  1212. flush := func() error {
  1213. if len(ops) == 0 {
  1214. return nil
  1215. }
  1216. if _, err := client.BatchWrite(ops, nil); err != nil {
  1217. return err
  1218. }
  1219. ops = ops[:0]
  1220. batchBytes = 8
  1221. return nil
  1222. }
  1223. for {
  1224. entries, done, err := cursor.Next()
  1225. if err != nil {
  1226. return err
  1227. }
  1228. for _, e := range entries {
  1229. suffix := string(e.Key[len(oldPrefix):])
  1230. newKey := []byte(newPrefix + suffix)
  1231. opBytes := 24 + len(newKey) + len(e.Value) + len(e.Key)
  1232. if len(ops)+2 > maxOperations || batchBytes+opBytes > bulkBatchByteBudget {
  1233. if err := flush(); err != nil {
  1234. return err
  1235. }
  1236. }
  1237. ops = append(ops,
  1238. BatchOp{Op: batchPut, Key: newKey, Value: e.Value},
  1239. BatchOp{Op: batchDelete, Key: append([]byte(nil), e.Key...)},
  1240. )
  1241. batchBytes += opBytes
  1242. }
  1243. if done {
  1244. return flush()
  1245. }
  1246. }
  1247. })
  1248. }
  1249. // updateIndexesOfTableLocked applies mutate to the durable definition of every
  1250. // index on table and persists the definitions it changed. mutate reports
  1251. // whether it modified the definition; unmodified indexes are left untouched.
  1252. // The changed definitions are written in one batch, so a table's indexes never
  1253. // split between the old and new shape. The caller holds m.mu; it reads the
  1254. // durable index list directly rather than calling the self-locking ListIndexes
  1255. // helper.
  1256. func (m *SchemaManager) updateIndexesOfTableLocked(table string, mutate func(*Index) bool) error {
  1257. var names []string
  1258. if m.indexListCached {
  1259. names = append([]string(nil), m.indexListCache...)
  1260. } else {
  1261. var data string
  1262. err := m.pool.WithClient(func(c *KVClient) error {
  1263. var rerr error
  1264. data, rerr = c.Read(m.indexListKey())
  1265. return rerr
  1266. })
  1267. if err == ErrKeyNotFound {
  1268. return nil
  1269. }
  1270. if err != nil {
  1271. return err
  1272. }
  1273. if err := json.Unmarshal([]byte(data), &names); err != nil {
  1274. return err
  1275. }
  1276. }
  1277. type indexUpdate struct {
  1278. index Index
  1279. key string
  1280. value []byte
  1281. }
  1282. var updates []indexUpdate
  1283. for _, name := range names {
  1284. key := m.indexKey(name)
  1285. var data string
  1286. err := m.pool.WithClient(func(c *KVClient) error {
  1287. var rerr error
  1288. data, rerr = c.Read(key)
  1289. return rerr
  1290. })
  1291. if err != nil {
  1292. if err == ErrKeyNotFound {
  1293. continue
  1294. }
  1295. return err
  1296. }
  1297. var idx Index
  1298. if err := json.Unmarshal([]byte(data), &idx); err != nil {
  1299. return err
  1300. }
  1301. if !strings.EqualFold(idx.Table, table) {
  1302. continue
  1303. }
  1304. if !mutate(&idx) {
  1305. continue
  1306. }
  1307. updated, err := json.Marshal(&idx)
  1308. if err != nil {
  1309. return err
  1310. }
  1311. updates = append(updates, indexUpdate{index: idx, key: key, value: updated})
  1312. }
  1313. if len(updates) == 0 {
  1314. return nil
  1315. }
  1316. ops := make([]BatchOp, 0, len(updates))
  1317. for _, update := range updates {
  1318. ops = append(ops, BatchOp{Op: batchPut, Key: []byte(update.key), Value: update.value})
  1319. }
  1320. if err := m.pool.WithClient(func(c *KVClient) error {
  1321. _, err := c.BatchWrite(ops, nil)
  1322. return err
  1323. }); err != nil {
  1324. return fmt.Errorf("update index definitions for table %s: %w", table, err)
  1325. }
  1326. for _, update := range updates {
  1327. if cached, ok := m.indexCache[strings.ToLower(update.index.Name)]; ok {
  1328. *cached = update.index
  1329. }
  1330. }
  1331. return nil
  1332. }
  1333. // repointIndexesLocked updates indexes whose table was renamed.
  1334. func (m *SchemaManager) repointIndexesLocked(oldName, newName string) error {
  1335. return m.updateIndexesOfTableLocked(oldName, func(idx *Index) bool {
  1336. idx.Table = newName
  1337. return true
  1338. })
  1339. }
  1340. // renameIndexColumnsLocked rewrites the stored definition of every index on
  1341. // table so a reference to oldName becomes newName. Index names stay unchanged:
  1342. // they are independent identifiers that SQLite does not derive from column
  1343. // names, and callers may have chosen them explicitly. A rename to an existing
  1344. // column name is rejected before this runs. Expression columns keep their SQL
  1345. // text; callers must reject a rename that such an expression depends on,
  1346. // because the text is not rewritten.
  1347. func (m *SchemaManager) renameIndexColumnsLocked(table, oldName, newName string) error {
  1348. return m.updateIndexesOfTableLocked(table, func(idx *Index) bool {
  1349. changed := false
  1350. for i := range idx.Columns {
  1351. // Expression columns persist SQL text, not a column reference; their
  1352. // text is left for the caller to drop and recreate the index.
  1353. if idx.Columns[i].Expression == "" && strings.EqualFold(idx.Columns[i].Name, oldName) {
  1354. idx.Columns[i].Name = newName
  1355. changed = true
  1356. }
  1357. }
  1358. return changed
  1359. })
  1360. }
  1361. // RenameColumn renames a column in a table. Column references in index
  1362. // definitions are rewritten; expression-index text is not, so a SQL caller must
  1363. // reject a rename that an expression depends on.
  1364. func (m *SchemaManager) RenameColumn(table, oldName, newName string) error {
  1365. tableLock := m.tableLock(table)
  1366. tableLock.Lock()
  1367. defer tableLock.Unlock()
  1368. m.mu.Lock()
  1369. defer m.mu.Unlock()
  1370. schema, err := m.getSchemaUnsafe(table)
  1371. if err != nil {
  1372. return err
  1373. }
  1374. // Check if new column name already exists
  1375. for _, col := range schema.Columns {
  1376. if strings.EqualFold(col.Name, newName) {
  1377. return fmt.Errorf("column already exists: %s", newName)
  1378. }
  1379. }
  1380. schema = cloneSchema(schema)
  1381. // Find and rename column
  1382. found := false
  1383. for i, col := range schema.Columns {
  1384. if strings.EqualFold(col.Name, oldName) {
  1385. schema.Columns[i].Name = newName
  1386. found = true
  1387. // Update primary key reference if needed
  1388. if strings.EqualFold(schema.PrimaryKey, oldName) {
  1389. schema.PrimaryKey = newName
  1390. }
  1391. break
  1392. }
  1393. }
  1394. if !found {
  1395. return fmt.Errorf("column not found: %s", oldName)
  1396. }
  1397. if err := m.rewriteRows(table, func(row Row) bool {
  1398. for key, value := range row {
  1399. if strings.EqualFold(key, oldName) {
  1400. delete(row, key)
  1401. row[newName] = value
  1402. return true
  1403. }
  1404. }
  1405. return false
  1406. }); err != nil {
  1407. return err
  1408. }
  1409. // Publish the renamed schema before the index definitions: reads of the new
  1410. // column name resolve against rewritten rows as soon as the schema lands.
  1411. // Index DDL is not transactional (as elsewhere in this engine), so a failure
  1412. // here leaves stale index metadata for the caller to drop and recreate.
  1413. if err := m.updateSchemaUnsafe(schema); err != nil {
  1414. return err
  1415. }
  1416. // Index metadata drives secondary-index values, UNIQUE enforcement,
  1417. // ON CONFLICT target resolution, and catalog introspection, so it must name
  1418. // the renamed column before any later statement uses the table.
  1419. return m.renameIndexColumnsLocked(table, oldName, newName)
  1420. }
  1421. // rewriteRows applies an idempotent name-keyed row transformation in bounded
  1422. // batches. Column DDL updates the schema only after all rows are rewritten, so
  1423. // an interrupted operation can safely resume against the old schema.
  1424. func (m *SchemaManager) rewriteRows(table string, transform func(Row) bool) error {
  1425. return m.applyRows(table, func(row Row) (bool, error) {
  1426. return transform(row), nil
  1427. })
  1428. }
  1429. // applyRows scans a table once and applies transform to every row. Rows for
  1430. // which transform reports a change are re-encoded and written in bounded
  1431. // batches; an error from transform aborts the scan. Callers hold the table's
  1432. // exclusive gate, so no writer can interleave between the scan and the schema
  1433. // update that follows it. Aborting after a batch was flushed is safe only
  1434. // because the DDL transforms are idempotent.
  1435. func (m *SchemaManager) applyRows(table string, transform func(Row) (bool, error)) error {
  1436. prefix := []byte(fmt.Sprintf("%s:_data:%s:", m.database, strings.ToLower(table)))
  1437. return m.pool.WithClient(func(client *KVClient) (retErr error) {
  1438. cursor, err := client.Scan(prefix)
  1439. if err != nil {
  1440. return err
  1441. }
  1442. defer func() {
  1443. if cerr := cursor.Close(); retErr == nil {
  1444. retErr = cerr
  1445. }
  1446. }()
  1447. ops := make([]BatchOp, 0, scanPageSize)
  1448. batchBytes := 8
  1449. flush := func() error {
  1450. if len(ops) == 0 {
  1451. return nil
  1452. }
  1453. if _, err := client.BatchWrite(ops, nil); err != nil {
  1454. return err
  1455. }
  1456. ops = ops[:0]
  1457. batchBytes = 8
  1458. return nil
  1459. }
  1460. for {
  1461. entries, done, err := cursor.Next()
  1462. if err != nil {
  1463. return err
  1464. }
  1465. for _, entry := range entries {
  1466. row, err := decodeRow(entry.Value)
  1467. if err != nil {
  1468. return err
  1469. }
  1470. changed, err := transform(row)
  1471. if err != nil {
  1472. return err
  1473. }
  1474. if !changed {
  1475. continue
  1476. }
  1477. value, err := encodeRow(row)
  1478. if err != nil {
  1479. return err
  1480. }
  1481. opBytes := 12 + len(entry.Key) + len(value)
  1482. if len(ops) == maxOperations || batchBytes+opBytes > bulkBatchByteBudget {
  1483. if err := flush(); err != nil {
  1484. return err
  1485. }
  1486. }
  1487. ops = append(ops, BatchOp{Op: batchPut, Key: append([]byte(nil), entry.Key...), Value: value})
  1488. batchBytes += opBytes
  1489. }
  1490. if done {
  1491. return flush()
  1492. }
  1493. }
  1494. })
  1495. }
  1496. // validateAddColumnUnique rejects an ADD COLUMN ... UNIQUE whose backfilled
  1497. // values would duplicate across existing rows. It reads rows as they will look
  1498. // after the rewrite (a row that already carries the field from an interrupted
  1499. // earlier attempt keeps its value) and mirrors ValidateUniqueIndex for a single
  1500. // plain column: NULLs do not collide. It is used only for a constant or absent
  1501. // default; a clock-reading default is validated by the rewrite itself because
  1502. // its values cannot be predicted. The caller holds the table's exclusive gate,
  1503. // so no writer can insert a conflicting row between this check and the
  1504. // publication of the index.
  1505. func (m *SchemaManager) validateAddColumnUnique(table string, column Column, index *Index) error {
  1506. seen := make(map[string]struct{})
  1507. return m.applyRows(table, func(row Row) (bool, error) {
  1508. value, ok := rowColumnValue(row, column.Name)
  1509. if !ok {
  1510. var err error
  1511. value, _, err = m.columnDefaultValue(column)
  1512. if err != nil {
  1513. return false, err
  1514. }
  1515. }
  1516. if value == nil {
  1517. return false, nil
  1518. }
  1519. encoded := encodeUniqueScalar(value)
  1520. if _, duplicate := seen[encoded]; duplicate {
  1521. return false, &UniqueViolationError{Constraint: index.Name}
  1522. }
  1523. seen[encoded] = struct{}{}
  1524. return false, nil
  1525. })
  1526. }
  1527. // columnHasDefault reports whether the column contributes a value to rows that
  1528. // omit it, whether that default is a stored constant or an expression.
  1529. func columnHasDefault(column Column) bool {
  1530. return column.Default != nil || column.DefaultExpr != ""
  1531. }
  1532. // validateAddColumnChecks rejects an ADD COLUMN ... CHECK whose predicate is
  1533. // already false for an existing row. Like validateAddColumnUnique it reads rows
  1534. // as they will look after the default backfill, and it runs before that rewrite
  1535. // so a rejected DDL leaves no row carrying the new field. The default is
  1536. // resolved through columnDefaultValue, so a clock-reading default reaches the
  1537. // predicate as the value the backfilled row will receive instead of NULL. The
  1538. // caller holds the table's exclusive gate, so no writer can insert a violating
  1539. // row between this validation and the publication of the check.
  1540. func (m *SchemaManager) validateAddColumnChecks(table string, column Column, checks []CheckConstraint) error {
  1541. if len(checks) == 0 {
  1542. return nil
  1543. }
  1544. eval := m.checkEvaluator()
  1545. if eval == nil {
  1546. return fmt.Errorf("cannot validate CHECK constraints on %s: no CHECK evaluator installed", table)
  1547. }
  1548. return m.applyRows(table, func(row Row) (bool, error) {
  1549. if _, ok := rowColumnValue(row, column.Name); !ok {
  1550. value, hasDefault, err := m.columnDefaultValue(column)
  1551. if err != nil {
  1552. return false, err
  1553. }
  1554. if hasDefault {
  1555. row[column.Name] = value
  1556. }
  1557. }
  1558. for _, check := range checks {
  1559. holds, err := eval(check.Expression, row)
  1560. if err != nil {
  1561. return false, fmt.Errorf("evaluate CHECK constraint %s on %s: %w", check.Label(), table, err)
  1562. }
  1563. if !holds {
  1564. return false, &CheckViolationError{Table: table, Constraint: check.Label()}
  1565. }
  1566. }
  1567. return false, nil
  1568. })
  1569. }
  1570. // tableHasRows reports whether a table holds at least one durable row.
  1571. func (m *SchemaManager) tableHasRows(table string) (bool, error) {
  1572. prefix := []byte(fmt.Sprintf("%s:_data:%s:", m.database, strings.ToLower(table)))
  1573. found := false
  1574. err := m.pool.WithClient(func(client *KVClient) (retErr error) {
  1575. cursor, err := client.ScanWithLimit(prefix, 1)
  1576. if err != nil {
  1577. return err
  1578. }
  1579. defer func() {
  1580. if cerr := cursor.Close(); retErr == nil {
  1581. retErr = cerr
  1582. }
  1583. }()
  1584. entries, _, err := cursor.Next()
  1585. if err != nil {
  1586. return err
  1587. }
  1588. found = len(entries) > 0
  1589. return nil
  1590. })
  1591. return found, err
  1592. }
  1593. // getSchemaUnsafe gets a schema without locking (internal use).
  1594. func (m *SchemaManager) getSchemaUnsafe(table string) (*Schema, error) {
  1595. tableLower := strings.ToLower(table)
  1596. // Check cache
  1597. if schema, ok := m.cache[tableLower]; ok {
  1598. return schema, nil
  1599. }
  1600. // Read from storage
  1601. key := m.schemaKey(table)
  1602. var data string
  1603. err := m.pool.WithClient(func(c *KVClient) error {
  1604. var err error
  1605. data, err = c.Read(key)
  1606. return err
  1607. })
  1608. if err != nil {
  1609. return nil, fmt.Errorf("table not found: %s", table)
  1610. }
  1611. var schema Schema
  1612. if err := json.Unmarshal([]byte(data), &schema); err != nil {
  1613. return nil, err
  1614. }
  1615. m.cache[tableLower] = &schema
  1616. return &schema, nil
  1617. }
  1618. // updateSchemaUnsafe updates a schema without locking (internal use).
  1619. func (m *SchemaManager) updateSchemaUnsafe(schema *Schema) error {
  1620. key := m.schemaKey(schema.Name)
  1621. data, _ := json.Marshal(schema)
  1622. err := m.pool.WithClient(func(c *KVClient) error {
  1623. return c.Write(key, string(data))
  1624. })
  1625. if err != nil {
  1626. return err
  1627. }
  1628. // Update cache
  1629. m.cache[strings.ToLower(schema.Name)] = schema
  1630. m.bumpVersionLocked()
  1631. return nil
  1632. }