|
|
@@ -0,0 +1,1589 @@
|
|
|
+package storage
|
|
|
+
|
|
|
+import (
|
|
|
+ "fmt"
|
|
|
+ "sort"
|
|
|
+ "strings"
|
|
|
+ "time"
|
|
|
+
|
|
|
+ "github.com/goccy/go-json"
|
|
|
+)
|
|
|
+
|
|
|
+// rowRewrite is one pending ALTER TABLE row transformation. It reports whether
|
|
|
+// it changed the row and must be idempotent: it runs again on every read of the
|
|
|
+// table while the transaction is open and once more on each durable row when
|
|
|
+// the transaction commits.
|
|
|
+type rowRewrite func(row Row) (bool, error)
|
|
|
+
|
|
|
+// SchemaTx is the transaction-scoped catalog overlay of one Session. DDL inside
|
|
|
+// an explicit transaction records its catalog and row effects here instead of
|
|
|
+// publishing them: the owning session reads through the overlay, every other
|
|
|
+// session keeps reading the committed catalog, ROLLBACK discards it, and COMMIT
|
|
|
+// merges the overlay into the session's single compare-and-swap batch write, so
|
|
|
+// a multi-statement migration becomes durable atomically or not at all.
|
|
|
+//
|
|
|
+// The overlay starts out empty and reads through to the durable catalog while
|
|
|
+// recording the LSN each touched key was observed at. Those LSNs become the
|
|
|
+// commit's compare checks, so a concurrent DDL that committed after the
|
|
|
+// transaction derived its catalog makes the commit fail with ErrSerialization
|
|
|
+// instead of overwriting the concurrent change.
|
|
|
+//
|
|
|
+// Every method assumes the owning session lock is held: a Session executes one
|
|
|
+// statement at a time, so the overlay needs no lock of its own and can share the
|
|
|
+// session's row overlay directly.
|
|
|
+type SchemaTx struct {
|
|
|
+ base *SchemaManager
|
|
|
+ sess *Session
|
|
|
+
|
|
|
+ // staged holds the transaction's definition of every touched table, keyed
|
|
|
+ // by lowercased name. A nil entry means the name is not visible to this
|
|
|
+ // transaction: it was dropped, or moved by RENAME TABLE.
|
|
|
+ staged map[string]*Schema
|
|
|
+ // stagedIndexes holds the transaction's definition of every touched index,
|
|
|
+ // keyed by lowercased name; nil means dropped.
|
|
|
+ stagedIndexes map[string]*Index
|
|
|
+ // committedSchemas and committedIndexes cache the definitions read from
|
|
|
+ // durable storage so the executor's repeated catalog lookups inside one
|
|
|
+ // statement do not re-read them. An entry is dropped when the transaction
|
|
|
+ // stages a replacement for the same name.
|
|
|
+ committedSchemas map[string]*Schema
|
|
|
+ committedIndexes map[string]*Index
|
|
|
+ // tables and indexList are the staged catalog lists. They are loaded from
|
|
|
+ // the committed catalog on first use and rewritten in place afterwards.
|
|
|
+ tables []string
|
|
|
+ indexList []string
|
|
|
+ // catalogDirty and indexListDirty report that the corresponding list is
|
|
|
+ // part of the commit.
|
|
|
+ catalogDirty bool
|
|
|
+ indexListDirty bool
|
|
|
+
|
|
|
+ // reads records the durable LSN every staged catalog key was derived from
|
|
|
+ // (0 for a key that was absent). commitChecks turns them into compare
|
|
|
+ // checks.
|
|
|
+ reads map[string]uint64
|
|
|
+
|
|
|
+ // rewrites holds pending row transformations keyed by the lowercased table
|
|
|
+ // name whose rows they apply to, in statement order.
|
|
|
+ rewrites map[string][]rowRewrite
|
|
|
+ // dataSources maps a staged table name to the durable table name whose data
|
|
|
+ // prefix still holds its rows: RENAME TABLE moves the definition without
|
|
|
+ // touching rows, so a later ALTER or DROP must look under the old prefix.
|
|
|
+ // An entry with an empty source is a table created by this transaction.
|
|
|
+ dataSources map[string]string
|
|
|
+ // drops holds durable table names whose data prefix is deleted at commit.
|
|
|
+ drops map[string]bool
|
|
|
+ // renames maps a durable table name to the staged name its rows move to at
|
|
|
+ // commit.
|
|
|
+ renames map[string]string
|
|
|
+
|
|
|
+ // rekeys records the overlay re-keyings this transaction performed, in
|
|
|
+ // order, so a savepoint can undo them.
|
|
|
+ rekeys []rekeyPair
|
|
|
+
|
|
|
+ version uint64
|
|
|
+}
|
|
|
+
|
|
|
+type rekeyPair struct {
|
|
|
+ from string
|
|
|
+ to string
|
|
|
+}
|
|
|
+
|
|
|
+// schemaTxSnapshot is the transaction catalog state at a savepoint.
|
|
|
+type schemaTxSnapshot struct {
|
|
|
+ version uint64
|
|
|
+ staged map[string]*Schema
|
|
|
+ stagedIndexes map[string]*Index
|
|
|
+ tables []string
|
|
|
+ indexList []string
|
|
|
+ catalogDirty bool
|
|
|
+ indexListDirty bool
|
|
|
+ rewrites map[string][]rowRewrite
|
|
|
+ dataSources map[string]string
|
|
|
+ drops map[string]bool
|
|
|
+ renames map[string]string
|
|
|
+ rekeys int
|
|
|
+}
|
|
|
+
|
|
|
+func newSchemaTx(s *Session) *SchemaTx {
|
|
|
+ return &SchemaTx{
|
|
|
+ base: s.schema,
|
|
|
+ sess: s,
|
|
|
+ staged: make(map[string]*Schema),
|
|
|
+ stagedIndexes: make(map[string]*Index),
|
|
|
+ committedSchemas: make(map[string]*Schema),
|
|
|
+ committedIndexes: make(map[string]*Index),
|
|
|
+ reads: make(map[string]uint64),
|
|
|
+ rewrites: make(map[string][]rowRewrite),
|
|
|
+ dataSources: make(map[string]string),
|
|
|
+ drops: make(map[string]bool),
|
|
|
+ renames: make(map[string]string),
|
|
|
+ version: s.schema.Version(),
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+// snapshot captures the catalog overlay. The maps are copied shallowly: staged
|
|
|
+// definitions are cloned before they are mutated, so the captured pointers stay
|
|
|
+// immutable.
|
|
|
+func (tx *SchemaTx) snapshot() schemaTxSnapshot {
|
|
|
+ snap := schemaTxSnapshot{
|
|
|
+ version: tx.version,
|
|
|
+ staged: make(map[string]*Schema, len(tx.staged)),
|
|
|
+ stagedIndexes: make(map[string]*Index, len(tx.stagedIndexes)),
|
|
|
+ tables: append([]string(nil), tx.tables...),
|
|
|
+ indexList: append([]string(nil), tx.indexList...),
|
|
|
+ catalogDirty: tx.catalogDirty,
|
|
|
+ indexListDirty: tx.indexListDirty,
|
|
|
+ rewrites: make(map[string][]rowRewrite, len(tx.rewrites)),
|
|
|
+ dataSources: make(map[string]string, len(tx.dataSources)),
|
|
|
+ drops: make(map[string]bool, len(tx.drops)),
|
|
|
+ renames: make(map[string]string, len(tx.renames)),
|
|
|
+ rekeys: len(tx.rekeys),
|
|
|
+ }
|
|
|
+ for k, v := range tx.staged {
|
|
|
+ snap.staged[k] = v
|
|
|
+ }
|
|
|
+ for k, v := range tx.stagedIndexes {
|
|
|
+ snap.stagedIndexes[k] = v
|
|
|
+ }
|
|
|
+ for k, v := range tx.rewrites {
|
|
|
+ snap.rewrites[k] = append([]rowRewrite(nil), v...)
|
|
|
+ }
|
|
|
+ for k, v := range tx.dataSources {
|
|
|
+ snap.dataSources[k] = v
|
|
|
+ }
|
|
|
+ for k, v := range tx.drops {
|
|
|
+ snap.drops[k] = v
|
|
|
+ }
|
|
|
+ for k, v := range tx.renames {
|
|
|
+ snap.renames[k] = v
|
|
|
+ }
|
|
|
+ return snap
|
|
|
+}
|
|
|
+
|
|
|
+// restore returns the overlay to a snapshot and undoes the overlay re-keyings
|
|
|
+// performed after it. The caller holds the session lock.
|
|
|
+func (tx *SchemaTx) restore(snap schemaTxSnapshot) {
|
|
|
+ for i := len(tx.rekeys) - 1; i >= snap.rekeys; i-- {
|
|
|
+ tx.sess.rekeyOverlayLocked(tx.rekeys[i].to, tx.rekeys[i].from)
|
|
|
+ }
|
|
|
+ tx.rekeys = tx.rekeys[:snap.rekeys]
|
|
|
+ tx.version = snap.version
|
|
|
+ tx.staged = snap.staged
|
|
|
+ tx.stagedIndexes = snap.stagedIndexes
|
|
|
+ tx.tables = snap.tables
|
|
|
+ tx.indexList = snap.indexList
|
|
|
+ tx.catalogDirty = snap.catalogDirty
|
|
|
+ tx.indexListDirty = snap.indexListDirty
|
|
|
+ tx.rewrites = snap.rewrites
|
|
|
+ tx.dataSources = snap.dataSources
|
|
|
+ tx.drops = snap.drops
|
|
|
+ tx.renames = snap.renames
|
|
|
+}
|
|
|
+
|
|
|
+// bump advances the transaction's catalog version so the executor's analyzer
|
|
|
+// catalog resynchronizes after every staged change.
|
|
|
+func (tx *SchemaTx) bump() {
|
|
|
+ tx.version++
|
|
|
+}
|
|
|
+
|
|
|
+func (tx *SchemaTx) recordRead(key string, lsn uint64) {
|
|
|
+ if _, ok := tx.reads[key]; !ok {
|
|
|
+ tx.reads[key] = lsn
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+// refreshCommitted drops the cached committed definitions before the commit's
|
|
|
+// final validation. Another session's DDL may have committed while this
|
|
|
+// transaction was open, so the rows it stages must be validated against the
|
|
|
+// catalog that is current under the commit gates; the staged overlay itself is
|
|
|
+// kept. The catalog lists keep their recorded LSN: a staged list write must
|
|
|
+// still fail its compare check rather than overwrite the concurrent change.
|
|
|
+func (tx *SchemaTx) refreshCommitted() {
|
|
|
+ tx.committedSchemas = make(map[string]*Schema)
|
|
|
+ tx.committedIndexes = make(map[string]*Index)
|
|
|
+ if !tx.indexListDirty {
|
|
|
+ tx.indexList = nil
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+// getSchema returns the transaction's view of a table.
|
|
|
+func (tx *SchemaTx) getSchema(name string) (*Schema, error) {
|
|
|
+ lower := strings.ToLower(name)
|
|
|
+ if schema, ok := tx.staged[lower]; ok {
|
|
|
+ if schema == nil {
|
|
|
+ return nil, fmt.Errorf("table not found: %s", name)
|
|
|
+ }
|
|
|
+ return cloneSchema(schema), nil
|
|
|
+ }
|
|
|
+ return tx.loadSchema(name)
|
|
|
+}
|
|
|
+
|
|
|
+// loadSchema reads the committed definition and records the LSN it observed.
|
|
|
+func (tx *SchemaTx) loadSchema(name string) (*Schema, error) {
|
|
|
+ lower := strings.ToLower(name)
|
|
|
+ if schema, ok := tx.committedSchemas[lower]; ok {
|
|
|
+ return cloneSchema(schema), nil
|
|
|
+ }
|
|
|
+ key := tx.base.schemaKey(name)
|
|
|
+ data, lsn, found, err := tx.base.readDurable(key)
|
|
|
+ if err != nil {
|
|
|
+ return nil, err
|
|
|
+ }
|
|
|
+ tx.recordRead(key, lsn)
|
|
|
+ if !found {
|
|
|
+ return nil, fmt.Errorf("table not found: %s", name)
|
|
|
+ }
|
|
|
+ var schema Schema
|
|
|
+ if err := json.Unmarshal([]byte(data), &schema); err != nil {
|
|
|
+ return nil, fmt.Errorf("failed to parse schema: %w", err)
|
|
|
+ }
|
|
|
+ schema.normalizePrimaryKey()
|
|
|
+ tx.committedSchemas[lower] = &schema
|
|
|
+ return cloneSchema(&schema), nil
|
|
|
+}
|
|
|
+
|
|
|
+func (tx *SchemaTx) tableExists(name string) bool {
|
|
|
+ _, err := tx.getSchema(name)
|
|
|
+ return err == nil
|
|
|
+}
|
|
|
+
|
|
|
+// listTables returns the staged table catalog in creation order.
|
|
|
+func (tx *SchemaTx) listTables() ([]string, error) {
|
|
|
+ if tx.tables == nil {
|
|
|
+ tables, lsn, err := tx.base.catalogTablesDurable()
|
|
|
+ if err != nil {
|
|
|
+ return nil, err
|
|
|
+ }
|
|
|
+ tx.recordRead(tx.base.catalogKey(), lsn)
|
|
|
+ tx.tables = tables
|
|
|
+ }
|
|
|
+ return append([]string(nil), tx.tables...), nil
|
|
|
+}
|
|
|
+
|
|
|
+func (tx *SchemaTx) stageTableList(tables []string) {
|
|
|
+ tx.tables = tables
|
|
|
+ tx.catalogDirty = true
|
|
|
+}
|
|
|
+
|
|
|
+// getIndex returns the transaction's view of an index definition.
|
|
|
+func (tx *SchemaTx) getIndex(name string) (*Index, error) {
|
|
|
+ lower := strings.ToLower(name)
|
|
|
+ if index, ok := tx.stagedIndexes[lower]; ok {
|
|
|
+ if index == nil {
|
|
|
+ return nil, fmt.Errorf("%w: %s", ErrIndexNotFound, name)
|
|
|
+ }
|
|
|
+ return cloneIndex(index), nil
|
|
|
+ }
|
|
|
+ if index, ok := tx.committedIndexes[lower]; ok {
|
|
|
+ return cloneIndex(index), nil
|
|
|
+ }
|
|
|
+ key := tx.base.indexKey(name)
|
|
|
+ data, lsn, found, err := tx.base.readDurable(key)
|
|
|
+ if err != nil {
|
|
|
+ return nil, err
|
|
|
+ }
|
|
|
+ tx.recordRead(key, lsn)
|
|
|
+ if !found {
|
|
|
+ return nil, fmt.Errorf("%w: %s", ErrIndexNotFound, name)
|
|
|
+ }
|
|
|
+ var index Index
|
|
|
+ if err := json.Unmarshal([]byte(data), &index); err != nil {
|
|
|
+ return nil, fmt.Errorf("failed to parse index: %w", err)
|
|
|
+ }
|
|
|
+ tx.committedIndexes[lower] = &index
|
|
|
+ return cloneIndex(&index), nil
|
|
|
+}
|
|
|
+
|
|
|
+func (tx *SchemaTx) indexExists(name string) bool {
|
|
|
+ _, err := tx.getIndex(name)
|
|
|
+ return err == nil
|
|
|
+}
|
|
|
+
|
|
|
+func (tx *SchemaTx) listIndexes() ([]string, error) {
|
|
|
+ if tx.indexList == nil {
|
|
|
+ indexes, lsn, err := tx.base.indexListDurable()
|
|
|
+ if err != nil {
|
|
|
+ return nil, err
|
|
|
+ }
|
|
|
+ tx.recordRead(tx.base.indexListKey(), lsn)
|
|
|
+ tx.indexList = indexes
|
|
|
+ }
|
|
|
+ return append([]string(nil), tx.indexList...), nil
|
|
|
+}
|
|
|
+
|
|
|
+func (tx *SchemaTx) stageIndexList(indexes []string) {
|
|
|
+ tx.indexList = indexes
|
|
|
+ tx.indexListDirty = true
|
|
|
+}
|
|
|
+
|
|
|
+func (tx *SchemaTx) listTableIndexes(table string) ([]*Index, error) {
|
|
|
+ names, err := tx.listIndexes()
|
|
|
+ if err != nil {
|
|
|
+ return nil, err
|
|
|
+ }
|
|
|
+ var result []*Index
|
|
|
+ for _, name := range names {
|
|
|
+ idx, err := tx.getIndex(name)
|
|
|
+ if err != nil {
|
|
|
+ if err == ErrIndexNotFound || strings.Contains(err.Error(), ErrIndexNotFound.Error()) {
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ return nil, err
|
|
|
+ }
|
|
|
+ if strings.EqualFold(idx.Table, table) {
|
|
|
+ result = append(result, idx)
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return result, nil
|
|
|
+}
|
|
|
+
|
|
|
+// stageSchema records a replacement table definition.
|
|
|
+func (tx *SchemaTx) stageSchema(schema *Schema) {
|
|
|
+ lower := strings.ToLower(schema.Name)
|
|
|
+ delete(tx.committedSchemas, lower)
|
|
|
+ tx.staged[lower] = cloneSchema(schema)
|
|
|
+ if _, ok := tx.dataSources[lower]; !ok {
|
|
|
+ tx.dataSources[lower] = lower
|
|
|
+ }
|
|
|
+ tx.recordRead(tx.base.schemaKey(schema.Name), 0)
|
|
|
+ tx.bump()
|
|
|
+}
|
|
|
+
|
|
|
+// stageDroppedTable records that a name stopped being addressable.
|
|
|
+func (tx *SchemaTx) stageDroppedTable(lower string) {
|
|
|
+ delete(tx.committedSchemas, lower)
|
|
|
+ tx.staged[lower] = nil
|
|
|
+ tx.recordRead(tx.base.schemaKey(lower), 0)
|
|
|
+ tx.bump()
|
|
|
+}
|
|
|
+
|
|
|
+// createTable stages CREATE TABLE.
|
|
|
+func (tx *SchemaTx) createTable(schema *Schema) error {
|
|
|
+ if tx.tableExists(schema.Name) {
|
|
|
+ return fmt.Errorf("table already exists: %s", schema.Name)
|
|
|
+ }
|
|
|
+ lower := strings.ToLower(schema.Name)
|
|
|
+ schema = cloneSchema(schema)
|
|
|
+ schema.CreatedAt = time.Now()
|
|
|
+ schema.normalizePrimaryKey()
|
|
|
+ tx.staged[lower] = schema
|
|
|
+ tx.dataSources[lower] = ""
|
|
|
+ tables, err := tx.listTables()
|
|
|
+ if err != nil {
|
|
|
+ return err
|
|
|
+ }
|
|
|
+ for _, name := range tables {
|
|
|
+ if strings.EqualFold(name, schema.Name) {
|
|
|
+ return fmt.Errorf("table already exists: %s", schema.Name)
|
|
|
+ }
|
|
|
+ }
|
|
|
+ tx.stageTableList(append(tables, schema.Name))
|
|
|
+ tx.bump()
|
|
|
+ return nil
|
|
|
+}
|
|
|
+
|
|
|
+// createIndex stages CREATE INDEX.
|
|
|
+func (tx *SchemaTx) createIndex(index *Index) error {
|
|
|
+ if tx.indexExists(index.Name) {
|
|
|
+ return fmt.Errorf("index already exists: %s", index.Name)
|
|
|
+ }
|
|
|
+ if !tx.tableExists(index.Table) {
|
|
|
+ return fmt.Errorf("table not found: %s", index.Table)
|
|
|
+ }
|
|
|
+ index = cloneIndex(index)
|
|
|
+ index.CreatedAt = time.Now()
|
|
|
+ delete(tx.committedIndexes, strings.ToLower(index.Name))
|
|
|
+ tx.stagedIndexes[strings.ToLower(index.Name)] = index
|
|
|
+ names, err := tx.listIndexes()
|
|
|
+ if err != nil {
|
|
|
+ return err
|
|
|
+ }
|
|
|
+ tx.stageIndexList(append(names, index.Name))
|
|
|
+ tx.recordRead(tx.base.indexKey(index.Name), 0)
|
|
|
+ tx.bump()
|
|
|
+ return nil
|
|
|
+}
|
|
|
+
|
|
|
+// dropIndex stages DROP INDEX.
|
|
|
+func (tx *SchemaTx) dropIndex(name string) error {
|
|
|
+ index, err := tx.getIndex(name)
|
|
|
+ if err != nil {
|
|
|
+ return fmt.Errorf("index not found: %s", name)
|
|
|
+ }
|
|
|
+ lower := strings.ToLower(name)
|
|
|
+ delete(tx.committedIndexes, lower)
|
|
|
+ tx.stagedIndexes[lower] = nil
|
|
|
+ names, err := tx.listIndexes()
|
|
|
+ if err != nil {
|
|
|
+ return err
|
|
|
+ }
|
|
|
+ filtered := make([]string, 0, len(names))
|
|
|
+ for _, existing := range names {
|
|
|
+ if !strings.EqualFold(existing, name) {
|
|
|
+ filtered = append(filtered, existing)
|
|
|
+ }
|
|
|
+ }
|
|
|
+ tx.stageIndexList(filtered)
|
|
|
+ tx.recordRead(tx.base.indexKey(index.Name), 0)
|
|
|
+ tx.bump()
|
|
|
+ return nil
|
|
|
+}
|
|
|
+
|
|
|
+// dropTable stages DROP TABLE. The durable rows are removed by the commit; the
|
|
|
+// session's staged rows for the table are discarded immediately because the
|
|
|
+// table is no longer visible to this transaction.
|
|
|
+func (tx *SchemaTx) dropTable(name string) error {
|
|
|
+ lower := strings.ToLower(name)
|
|
|
+ if _, err := tx.getSchema(name); err != nil {
|
|
|
+ return fmt.Errorf("table not found: %s", name)
|
|
|
+ }
|
|
|
+ source := tx.durableSource(lower)
|
|
|
+ if source != "" {
|
|
|
+ tx.drops[source] = true
|
|
|
+ }
|
|
|
+ delete(tx.rewrites, lower)
|
|
|
+ delete(tx.dataSources, lower)
|
|
|
+ // A name that was renamed inside this transaction has no durable record of
|
|
|
+ // its own; the rename already deleted the original key at commit time.
|
|
|
+ if _, renamed := tx.renames[lower]; !renamed {
|
|
|
+ tx.recordRead(tx.base.schemaKey(lower), 0)
|
|
|
+ }
|
|
|
+ tx.stageDroppedTable(lower)
|
|
|
+
|
|
|
+ for _, idx := range tx.tableIndexes(lower) {
|
|
|
+ if err := tx.dropIndex(idx.Name); err != nil {
|
|
|
+ return err
|
|
|
+ }
|
|
|
+ }
|
|
|
+ tables, err := tx.listTables()
|
|
|
+ if err != nil {
|
|
|
+ return err
|
|
|
+ }
|
|
|
+ filtered := make([]string, 0, len(tables))
|
|
|
+ for _, existing := range tables {
|
|
|
+ if !strings.EqualFold(existing, name) {
|
|
|
+ filtered = append(filtered, existing)
|
|
|
+ }
|
|
|
+ }
|
|
|
+ tx.stageTableList(filtered)
|
|
|
+ tx.sess.dropOverlayTableLocked(lower)
|
|
|
+ tx.sess.markTableScanned(lower)
|
|
|
+ return nil
|
|
|
+}
|
|
|
+
|
|
|
+// durableSource returns the committed table name whose data prefix holds the
|
|
|
+// rows of a staged table, or "" when the table was created by this transaction.
|
|
|
+func (tx *SchemaTx) durableSource(lower string) string {
|
|
|
+ if source, ok := tx.dataSources[lower]; ok {
|
|
|
+ return source
|
|
|
+ }
|
|
|
+ return lower
|
|
|
+}
|
|
|
+
|
|
|
+// tableIndexes returns the staged index definitions of a table.
|
|
|
+func (tx *SchemaTx) tableIndexes(lower string) []*Index {
|
|
|
+ names, err := tx.listIndexes()
|
|
|
+ if err != nil {
|
|
|
+ return nil
|
|
|
+ }
|
|
|
+ var result []*Index
|
|
|
+ for _, name := range names {
|
|
|
+ idx, err := tx.getIndex(name)
|
|
|
+ if err != nil || !strings.EqualFold(idx.Table, lower) {
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ result = append(result, idx)
|
|
|
+ }
|
|
|
+ return result
|
|
|
+}
|
|
|
+
|
|
|
+// addColumn stages ALTER TABLE ADD COLUMN. The new column, its CHECK
|
|
|
+// predicates, its UNIQUE index, and the default backfill of existing rows are
|
|
|
+// validated against the rows this session sees (durable rows plus its own
|
|
|
+// staged writes) before anything is staged, so a rejected DDL leaves the
|
|
|
+// transaction's catalog untouched.
|
|
|
+func (tx *SchemaTx) addColumn(table string, column Column, checks []CheckConstraint, unique *Index) error {
|
|
|
+ schema, err := tx.getSchema(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 tx.indexExists(unique.Name) {
|
|
|
+ return fmt.Errorf("index already exists: %s", unique.Name)
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ now := tx.base.defaultNow()
|
|
|
+ // Validation covers the committed rows, mirroring the autocommit DDL path:
|
|
|
+ // rows this transaction staged are re-validated against the staged schema at
|
|
|
+ // commit (applyOverlaySchemaDefaults), because a staged row may still be
|
|
|
+ // rewritten or removed before then.
|
|
|
+ rows, err := tx.durableRows(table)
|
|
|
+ if err != nil {
|
|
|
+ return err
|
|
|
+ }
|
|
|
+ if !column.Nullable && !columnHasDefault(column) && len(rows) > 0 {
|
|
|
+ return &NotNullViolationError{Table: schema.Name, Column: column.Name}
|
|
|
+ }
|
|
|
+ eval := tx.base.checkEvaluator()
|
|
|
+ if len(checks) > 0 && eval == nil {
|
|
|
+ return fmt.Errorf("cannot validate CHECK constraints on %s: no CHECK evaluator installed", table)
|
|
|
+ }
|
|
|
+ seen := make(map[string]struct{})
|
|
|
+ for _, row := range rows {
|
|
|
+ value, present := rowColumnValue(row, column.Name)
|
|
|
+ if !present {
|
|
|
+ if fallback, has, defErr := tx.base.columnDefaultValue(column, now); defErr != nil {
|
|
|
+ return defErr
|
|
|
+ } else if has {
|
|
|
+ value = fallback
|
|
|
+ present = true
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if len(checks) > 0 {
|
|
|
+ probe := cloneRow(row)
|
|
|
+ if present {
|
|
|
+ probe[column.Name] = value
|
|
|
+ }
|
|
|
+ for _, check := range checks {
|
|
|
+ holds, evalErr := eval(check.Expression, probe)
|
|
|
+ if evalErr != nil {
|
|
|
+ return fmt.Errorf("evaluate CHECK constraint %s on %s: %w", check.Label(), table, evalErr)
|
|
|
+ }
|
|
|
+ if !holds {
|
|
|
+ return &CheckViolationError{Table: table, Constraint: check.Label()}
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if unique != nil && value != nil {
|
|
|
+ encoded := encodeUniqueScalar(value)
|
|
|
+ if _, duplicate := seen[encoded]; duplicate {
|
|
|
+ return &UniqueViolationError{Constraint: unique.Name}
|
|
|
+ }
|
|
|
+ seen[encoded] = struct{}{}
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ schema = cloneSchema(schema)
|
|
|
+ schema.Columns = append(schema.Columns, column)
|
|
|
+ schema.Checks = append(schema.Checks, checks...)
|
|
|
+ tx.stageSchema(schema)
|
|
|
+ if unique != nil {
|
|
|
+ tx.stagedIndexes[strings.ToLower(unique.Name)] = cloneIndex(unique)
|
|
|
+ names, err := tx.listIndexes()
|
|
|
+ if err != nil {
|
|
|
+ return err
|
|
|
+ }
|
|
|
+ tx.stageIndexList(append(names, unique.Name))
|
|
|
+ tx.recordRead(tx.base.indexKey(unique.Name), 0)
|
|
|
+ }
|
|
|
+ if columnHasDefault(column) {
|
|
|
+ value, has, err := tx.base.columnDefaultValue(column, now)
|
|
|
+ if err != nil {
|
|
|
+ return err
|
|
|
+ }
|
|
|
+ if has {
|
|
|
+ name := column.Name
|
|
|
+ tx.addRewrite(table, func(row Row) (bool, error) {
|
|
|
+ if _, ok := rowColumnValue(row, name); ok {
|
|
|
+ return false, nil
|
|
|
+ }
|
|
|
+ row[name] = value
|
|
|
+ return true, nil
|
|
|
+ })
|
|
|
+ }
|
|
|
+ }
|
|
|
+ tx.sess.markTableScanned(strings.ToLower(table))
|
|
|
+ return nil
|
|
|
+}
|
|
|
+
|
|
|
+// durableRows returns the committed rows of a table, ignoring this
|
|
|
+// transaction's staged rows and pending rewrites.
|
|
|
+func (tx *SchemaTx) durableRows(table string) ([]Row, error) {
|
|
|
+ var rows []Row
|
|
|
+ prefix := tx.sess.table.dataPrefix(tx.durableSource(strings.ToLower(table)))
|
|
|
+ err := tx.sess.table.scanRowsAtPrefix(prefix, func(_ string, row Row, _ uint64) (bool, error) {
|
|
|
+ rows = append(rows, row)
|
|
|
+ return false, nil
|
|
|
+ })
|
|
|
+ return rows, err
|
|
|
+}
|
|
|
+
|
|
|
+// dropColumn stages ALTER TABLE DROP COLUMN.
|
|
|
+func (tx *SchemaTx) dropColumn(table, columnName string) error {
|
|
|
+ schema, err := tx.getSchema(table)
|
|
|
+ if err != nil {
|
|
|
+ return err
|
|
|
+ }
|
|
|
+ if schema.IsPrimaryKeyColumn(columnName) {
|
|
|
+ return fmt.Errorf("cannot drop primary key column: %s", columnName)
|
|
|
+ }
|
|
|
+ columns := make([]Column, 0, len(schema.Columns)-1)
|
|
|
+ found := false
|
|
|
+ for _, column := range schema.Columns {
|
|
|
+ if strings.EqualFold(column.Name, columnName) {
|
|
|
+ found = true
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ columns = append(columns, column)
|
|
|
+ }
|
|
|
+ if !found {
|
|
|
+ return fmt.Errorf("column not found: %s", columnName)
|
|
|
+ }
|
|
|
+ schema = cloneSchema(schema)
|
|
|
+ schema.Columns = columns
|
|
|
+ tx.stageSchema(schema)
|
|
|
+ tx.addRewrite(table, func(row Row) (bool, error) {
|
|
|
+ for key := range row {
|
|
|
+ if strings.EqualFold(key, columnName) {
|
|
|
+ delete(row, key)
|
|
|
+ return true, nil
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return false, nil
|
|
|
+ })
|
|
|
+ tx.sess.markTableScanned(strings.ToLower(table))
|
|
|
+ return nil
|
|
|
+}
|
|
|
+
|
|
|
+// renameColumn stages ALTER TABLE RENAME COLUMN, including the column
|
|
|
+// references of the table's index definitions.
|
|
|
+func (tx *SchemaTx) renameColumn(table, oldName, newName string) error {
|
|
|
+ schema, err := tx.getSchema(table)
|
|
|
+ if err != nil {
|
|
|
+ return err
|
|
|
+ }
|
|
|
+ for _, column := range schema.Columns {
|
|
|
+ if strings.EqualFold(column.Name, newName) {
|
|
|
+ return fmt.Errorf("column already exists: %s", newName)
|
|
|
+ }
|
|
|
+ }
|
|
|
+ schema = cloneSchema(schema)
|
|
|
+ found := false
|
|
|
+ for i := range schema.Columns {
|
|
|
+ if !strings.EqualFold(schema.Columns[i].Name, oldName) {
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ schema.Columns[i].Name = newName
|
|
|
+ found = true
|
|
|
+ for j, pk := range schema.PrimaryKeyColumns {
|
|
|
+ if strings.EqualFold(pk, oldName) {
|
|
|
+ schema.PrimaryKeyColumns[j] = newName
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if strings.EqualFold(schema.PrimaryKey, oldName) {
|
|
|
+ schema.PrimaryKey = newName
|
|
|
+ }
|
|
|
+ break
|
|
|
+ }
|
|
|
+ if !found {
|
|
|
+ return fmt.Errorf("column not found: %s", oldName)
|
|
|
+ }
|
|
|
+ // CHECK predicates persist SQL text and are not rewritten; the executor
|
|
|
+ // rejects a rename that one depends on, matching the autocommit path.
|
|
|
+ tx.stageSchema(schema)
|
|
|
+ for _, idx := range tx.tableIndexes(strings.ToLower(table)) {
|
|
|
+ changed := false
|
|
|
+ updated := cloneIndex(idx)
|
|
|
+ for i := range updated.Columns {
|
|
|
+ if updated.Columns[i].Expression == "" && strings.EqualFold(updated.Columns[i].Name, oldName) {
|
|
|
+ updated.Columns[i].Name = newName
|
|
|
+ changed = true
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if !changed {
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ tx.stagedIndexes[strings.ToLower(updated.Name)] = updated
|
|
|
+ tx.recordRead(tx.base.indexKey(updated.Name), 0)
|
|
|
+ }
|
|
|
+ tx.addRewrite(table, func(row Row) (bool, error) {
|
|
|
+ for key, value := range row {
|
|
|
+ if strings.EqualFold(key, oldName) {
|
|
|
+ delete(row, key)
|
|
|
+ row[newName] = value
|
|
|
+ return true, nil
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return false, nil
|
|
|
+ })
|
|
|
+ tx.sess.markTableScanned(strings.ToLower(table))
|
|
|
+ return nil
|
|
|
+}
|
|
|
+
|
|
|
+// renameTable stages ALTER TABLE RENAME TABLE. Rows stay under their committed
|
|
|
+// data prefix until commit; the session overlay is re-keyed immediately so the
|
|
|
+// transaction reads its own writes under the new name.
|
|
|
+func (tx *SchemaTx) renameTable(oldName, newName string) error {
|
|
|
+ schema, err := tx.getSchema(oldName)
|
|
|
+ if err != nil {
|
|
|
+ return err
|
|
|
+ }
|
|
|
+ if tx.tableExists(newName) {
|
|
|
+ return fmt.Errorf("table already exists: %s", newName)
|
|
|
+ }
|
|
|
+ oldLower := strings.ToLower(oldName)
|
|
|
+ newLower := strings.ToLower(newName)
|
|
|
+ source := tx.durableSource(oldLower)
|
|
|
+
|
|
|
+ schema = cloneSchema(schema)
|
|
|
+ schema.Name = newName
|
|
|
+ delete(tx.committedSchemas, oldLower)
|
|
|
+ delete(tx.committedSchemas, newLower)
|
|
|
+ tx.staged[oldLower] = nil
|
|
|
+ tx.staged[newLower] = schema
|
|
|
+ if source == "" {
|
|
|
+ tx.dataSources[newLower] = ""
|
|
|
+ } else {
|
|
|
+ tx.dataSources[newLower] = source
|
|
|
+ tx.renames[source] = newName
|
|
|
+ }
|
|
|
+ delete(tx.dataSources, oldLower)
|
|
|
+ // A drop staged before the rename keeps its durable source; the rename
|
|
|
+ // moves that source instead, so the drop no longer applies to it.
|
|
|
+ delete(tx.drops, newLower)
|
|
|
+ if rewrites, ok := tx.rewrites[oldLower]; ok {
|
|
|
+ delete(tx.rewrites, oldLower)
|
|
|
+ tx.rewrites[newLower] = rewrites
|
|
|
+ }
|
|
|
+
|
|
|
+ for _, idx := range tx.tableIndexes(oldLower) {
|
|
|
+ updated := cloneIndex(idx)
|
|
|
+ updated.Table = newName
|
|
|
+ tx.stagedIndexes[strings.ToLower(updated.Name)] = updated
|
|
|
+ tx.recordRead(tx.base.indexKey(updated.Name), 0)
|
|
|
+ }
|
|
|
+
|
|
|
+ tables, err := tx.listTables()
|
|
|
+ if err != nil {
|
|
|
+ return err
|
|
|
+ }
|
|
|
+ found := false
|
|
|
+ for i, name := range tables {
|
|
|
+ if strings.EqualFold(name, oldName) {
|
|
|
+ tables[i] = newName
|
|
|
+ found = true
|
|
|
+ break
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if !found {
|
|
|
+ return fmt.Errorf("table not found in catalog: %s", oldName)
|
|
|
+ }
|
|
|
+ tx.stageTableList(tables)
|
|
|
+
|
|
|
+ tx.rekeys = append(tx.rekeys, rekeyPair{from: oldLower, to: newLower})
|
|
|
+ tx.sess.rekeyOverlayLocked(oldLower, newLower)
|
|
|
+ source = tx.durableSource(newLower)
|
|
|
+ if source != "" {
|
|
|
+ tx.sess.markTableScanned(source)
|
|
|
+ }
|
|
|
+ tx.bump()
|
|
|
+ return nil
|
|
|
+}
|
|
|
+
|
|
|
+// addRewrite appends a row transformation for a table.
|
|
|
+func (tx *SchemaTx) addRewrite(table string, rewrite rowRewrite) {
|
|
|
+ lower := strings.ToLower(table)
|
|
|
+ tx.rewrites[lower] = append(tx.rewrites[lower], rewrite)
|
|
|
+}
|
|
|
+
|
|
|
+// rewriteRow applies every pending transformation of a table to a row.
|
|
|
+func (tx *SchemaTx) rewriteRow(table string, row Row) (bool, error) {
|
|
|
+ rewrites, ok := tx.rewrites[strings.ToLower(table)]
|
|
|
+ if !ok {
|
|
|
+ return false, nil
|
|
|
+ }
|
|
|
+ changed := false
|
|
|
+ for _, rewrite := range rewrites {
|
|
|
+ applied, err := rewrite(row)
|
|
|
+ if err != nil {
|
|
|
+ return changed, err
|
|
|
+ }
|
|
|
+ changed = changed || applied
|
|
|
+ }
|
|
|
+ return changed, nil
|
|
|
+}
|
|
|
+
|
|
|
+// commitCatalogOps returns the durable catalog writes of the transaction and
|
|
|
+// the compare checks that keep them from overwriting a concurrent commit. The
|
|
|
+// caller holds the session lock and the affected table gates.
|
|
|
+func (tx *SchemaTx) commitCatalogOps() ([]BatchOp, []CompareCheck, error) {
|
|
|
+ var ops []BatchOp
|
|
|
+ keys := make([]string, 0, len(tx.staged)+len(tx.stagedIndexes)+2)
|
|
|
+
|
|
|
+ for lower, schema := range tx.staged {
|
|
|
+ if schema == nil {
|
|
|
+ if _, renamed := tx.renames[lower]; renamed {
|
|
|
+ // RENAME TABLE names the source through renames; the delete of
|
|
|
+ // the old schema record is emitted below.
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ ops = append(ops,
|
|
|
+ BatchOp{Op: batchDelete, Key: []byte(tx.base.schemaKey(lower))},
|
|
|
+ BatchOp{Op: batchDelete, Key: []byte(tx.base.rowIDKey(lower))},
|
|
|
+ )
|
|
|
+ // The ROWID key is an in-memory allocator input, not catalog state a
|
|
|
+ // concurrent statement can change, so its delete needs no compare.
|
|
|
+ keys = append(keys, tx.base.schemaKey(lower))
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ data, err := json.Marshal(schema)
|
|
|
+ if err != nil {
|
|
|
+ return nil, nil, fmt.Errorf("failed to serialize schema: %w", err)
|
|
|
+ }
|
|
|
+ key := tx.base.schemaKey(schema.Name)
|
|
|
+ ops = append(ops, BatchOp{Op: batchPut, Key: []byte(key), Value: data})
|
|
|
+ keys = append(keys, key)
|
|
|
+ }
|
|
|
+ // A staged table definition whose name replaced an older one deletes the
|
|
|
+ // old record in the same batch.
|
|
|
+ for source, newName := range tx.renames {
|
|
|
+ ops = append(ops, BatchOp{Op: batchDelete, Key: []byte(tx.base.schemaKey(source))})
|
|
|
+ keys = append(keys, tx.base.schemaKey(source))
|
|
|
+ if !strings.EqualFold(source, newName) {
|
|
|
+ ops = append(ops, BatchOp{Op: batchDelete, Key: []byte(tx.base.rowIDKey(source))})
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ for lower, index := range tx.stagedIndexes {
|
|
|
+ if index == nil {
|
|
|
+ ops = append(ops, BatchOp{Op: batchDelete, Key: []byte(tx.base.indexKey(lower))})
|
|
|
+ keys = append(keys, tx.base.indexKey(lower))
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ data, err := json.Marshal(index)
|
|
|
+ if err != nil {
|
|
|
+ return nil, nil, fmt.Errorf("failed to serialize index: %w", err)
|
|
|
+ }
|
|
|
+ key := tx.base.indexKey(index.Name)
|
|
|
+ ops = append(ops, BatchOp{Op: batchPut, Key: []byte(key), Value: data})
|
|
|
+ keys = append(keys, key)
|
|
|
+ }
|
|
|
+
|
|
|
+ if tx.catalogDirty {
|
|
|
+ data, err := json.Marshal(tx.tables)
|
|
|
+ if err != nil {
|
|
|
+ return nil, nil, fmt.Errorf("failed to serialize catalog: %w", err)
|
|
|
+ }
|
|
|
+ ops = append(ops, BatchOp{Op: batchPut, Key: []byte(tx.base.catalogKey()), Value: data})
|
|
|
+ keys = append(keys, tx.base.catalogKey())
|
|
|
+ }
|
|
|
+ if tx.indexListDirty {
|
|
|
+ data, err := json.Marshal(tx.indexList)
|
|
|
+ if err != nil {
|
|
|
+ return nil, nil, fmt.Errorf("failed to serialize index list: %w", err)
|
|
|
+ }
|
|
|
+ ops = append(ops, BatchOp{Op: batchPut, Key: []byte(tx.base.indexListKey()), Value: data})
|
|
|
+ keys = append(keys, tx.base.indexListKey())
|
|
|
+ }
|
|
|
+
|
|
|
+ checks := make([]CompareCheck, 0, len(keys))
|
|
|
+ seen := make(map[string]bool, len(keys))
|
|
|
+ for _, key := range keys {
|
|
|
+ if seen[key] {
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ seen[key] = true
|
|
|
+ checks = append(checks, CompareCheck{Key: []byte(key), LSN: tx.reads[key]})
|
|
|
+ }
|
|
|
+ return ops, checks, nil
|
|
|
+}
|
|
|
+
|
|
|
+// commitDataOps returns the durable row writes of the transaction: dropped
|
|
|
+// tables lose every row, renamed tables move to their new prefix, and ALTER
|
|
|
+// TABLE transformations are applied to the committed row images. Rows this
|
|
|
+// session staged in its overlay are published from the overlay instead, so they
|
|
|
+// are skipped here. The caller holds the session lock and the exclusive table
|
|
|
+// gates of every affected table.
|
|
|
+func (tx *SchemaTx) commitDataOps() ([]BatchOp, map[string]map[string]*overlayEntry, error) {
|
|
|
+ entries := make(map[string]map[string]*overlayEntry)
|
|
|
+ var ops []BatchOp
|
|
|
+ if len(tx.drops) == 0 && len(tx.renames) == 0 && len(tx.rewrites) == 0 {
|
|
|
+ return nil, nil, nil
|
|
|
+ }
|
|
|
+
|
|
|
+ affected := make(map[string]bool)
|
|
|
+ for table := range tx.drops {
|
|
|
+ affected[table] = true
|
|
|
+ }
|
|
|
+ for table := range tx.renames {
|
|
|
+ affected[table] = true
|
|
|
+ }
|
|
|
+ for table := range tx.rewrites {
|
|
|
+ if source := tx.durableSource(table); source != "" && !tx.drops[source] {
|
|
|
+ affected[source] = true
|
|
|
+ }
|
|
|
+ }
|
|
|
+ names := make([]string, 0, len(affected))
|
|
|
+ for table := range affected {
|
|
|
+ names = append(names, table)
|
|
|
+ }
|
|
|
+ sort.Strings(names)
|
|
|
+
|
|
|
+ for _, source := range names {
|
|
|
+ if tx.drops[source] {
|
|
|
+ dropped, err := tx.scanRowsForDDL(source, func(key string, row Row) (bool, Row, error) {
|
|
|
+ if tx.sess.overlayHolds(strings.ToLower(source), key) {
|
|
|
+ return false, nil, nil
|
|
|
+ }
|
|
|
+ return true, nil, nil
|
|
|
+ })
|
|
|
+ if err != nil {
|
|
|
+ return nil, nil, err
|
|
|
+ }
|
|
|
+ ops = append(ops, dropped...)
|
|
|
+ continue
|
|
|
+ }
|
|
|
+
|
|
|
+ newName, renamed := tx.renames[source]
|
|
|
+ if renamed {
|
|
|
+ updated, err := tx.scanRowsForDDL(source, func(key string, row Row) (bool, Row, error) {
|
|
|
+ if _, err := tx.rewriteRow(newName, row); err != nil {
|
|
|
+ return false, nil, err
|
|
|
+ }
|
|
|
+ newKey := tx.sess.table.dataKey(newName, mustRowPK(tx, newName, row))
|
|
|
+ if tx.sess.overlayHolds(strings.ToLower(newName), newKey) {
|
|
|
+ return false, nil, nil
|
|
|
+ }
|
|
|
+ return true, row, nil
|
|
|
+ })
|
|
|
+ if err != nil {
|
|
|
+ return nil, nil, err
|
|
|
+ }
|
|
|
+ lower := strings.ToLower(newName)
|
|
|
+ if entries[lower] == nil {
|
|
|
+ entries[lower] = make(map[string]*overlayEntry)
|
|
|
+ }
|
|
|
+ for _, op := range updated {
|
|
|
+ oldKey := op.Key
|
|
|
+ suffix := strings.TrimPrefix(string(oldKey), tx.sess.table.dataPrefix(source))
|
|
|
+ newKey := []byte(tx.sess.table.dataPrefix(newName) + suffix)
|
|
|
+ row, err := decodeRow(op.Value)
|
|
|
+ if err != nil {
|
|
|
+ return nil, nil, err
|
|
|
+ }
|
|
|
+ entries[lower][string(newKey)] = &overlayEntry{row: row}
|
|
|
+ ops = append(ops,
|
|
|
+ BatchOp{Op: batchPut, Key: newKey, Value: op.Value},
|
|
|
+ BatchOp{Op: batchDelete, Key: oldKey},
|
|
|
+ )
|
|
|
+ }
|
|
|
+ continue
|
|
|
+ }
|
|
|
+
|
|
|
+ rewrites := tx.rewrites[source]
|
|
|
+ if len(rewrites) == 0 {
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ updated, err := tx.scanRowsForDDL(source, func(key string, row Row) (bool, Row, error) {
|
|
|
+ if tx.sess.overlayHolds(strings.ToLower(source), key) {
|
|
|
+ return false, nil, nil
|
|
|
+ }
|
|
|
+ changed, err := tx.rewriteRow(source, row)
|
|
|
+ if err != nil {
|
|
|
+ return false, nil, err
|
|
|
+ }
|
|
|
+ return changed, row, nil
|
|
|
+ })
|
|
|
+ if err != nil {
|
|
|
+ return nil, nil, err
|
|
|
+ }
|
|
|
+ lower := strings.ToLower(source)
|
|
|
+ if entries[lower] == nil {
|
|
|
+ entries[lower] = make(map[string]*overlayEntry)
|
|
|
+ }
|
|
|
+ for _, op := range updated {
|
|
|
+ row, err := decodeRow(op.Value)
|
|
|
+ if err != nil {
|
|
|
+ return nil, nil, err
|
|
|
+ }
|
|
|
+ entries[lower][string(op.Key)] = &overlayEntry{row: row}
|
|
|
+ ops = append(ops, op)
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return ops, entries, nil
|
|
|
+}
|
|
|
+
|
|
|
+// mustRowPK derives the primary-key string of a row the transaction's staged
|
|
|
+// schema describes.
|
|
|
+func mustRowPK(tx *SchemaTx, table string, row Row) string {
|
|
|
+ schema, err := tx.getSchema(table)
|
|
|
+ if err != nil {
|
|
|
+ return ""
|
|
|
+ }
|
|
|
+ pk, err := schema.PrimaryKeyForRow(row)
|
|
|
+ if err != nil {
|
|
|
+ return ""
|
|
|
+ }
|
|
|
+ return pk
|
|
|
+}
|
|
|
+
|
|
|
+// scanRowsForDDL streams a durable table's rows through transform. When
|
|
|
+// transform reports a change the row is re-encoded and returned as a put under
|
|
|
+// its existing key; when it reports removal the key is returned as a delete.
|
|
|
+func (tx *SchemaTx) scanRowsForDDL(table string, transform func(key string, row Row) (write bool, replacement Row, err error)) ([]BatchOp, error) {
|
|
|
+ var ops []BatchOp
|
|
|
+ prefix := []byte(tx.sess.table.dataPrefix(table))
|
|
|
+ err := tx.base.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
|
|
|
+ }
|
|
|
+ }()
|
|
|
+ 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
|
|
|
+ }
|
|
|
+ write, replacement, err := transform(string(entry.Key), row)
|
|
|
+ if err != nil {
|
|
|
+ return err
|
|
|
+ }
|
|
|
+ if !write {
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ if replacement == nil {
|
|
|
+ ops = append(ops, BatchOp{Op: batchDelete, Key: append([]byte(nil), entry.Key...)})
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ value, err := encodeRow(replacement)
|
|
|
+ if err != nil {
|
|
|
+ return err
|
|
|
+ }
|
|
|
+ ops = append(ops, BatchOp{Op: batchPut, Key: append([]byte(nil), entry.Key...), Value: value})
|
|
|
+ }
|
|
|
+ if done {
|
|
|
+ return nil
|
|
|
+ }
|
|
|
+ }
|
|
|
+ })
|
|
|
+ return ops, err
|
|
|
+}
|
|
|
+
|
|
|
+// touchedTables lists the lowercased table names whose caches the commit must
|
|
|
+// invalidate, including the source names rows moved away from.
|
|
|
+func (tx *SchemaTx) touchedTables() []string {
|
|
|
+ set := make(map[string]bool)
|
|
|
+ for lower, schema := range tx.staged {
|
|
|
+ set[lower] = true
|
|
|
+ if schema != nil {
|
|
|
+ set[strings.ToLower(schema.Name)] = true
|
|
|
+ if source, ok := tx.dataSources[lower]; ok && source != "" {
|
|
|
+ set[source] = true
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ for table := range tx.drops {
|
|
|
+ set[table] = true
|
|
|
+ }
|
|
|
+ for table := range tx.renames {
|
|
|
+ set[table] = true
|
|
|
+ }
|
|
|
+ for table := range tx.rewrites {
|
|
|
+ set[table] = true
|
|
|
+ if source := tx.durableSource(table); source != "" {
|
|
|
+ set[source] = true
|
|
|
+ }
|
|
|
+ }
|
|
|
+ names := make([]string, 0, len(set))
|
|
|
+ for table := range set {
|
|
|
+ names = append(names, table)
|
|
|
+ }
|
|
|
+ sort.Strings(names)
|
|
|
+ return names
|
|
|
+}
|
|
|
+
|
|
|
+// touchedIndexes lists the lowercased index names whose caches the commit must
|
|
|
+// reset.
|
|
|
+func (tx *SchemaTx) touchedIndexes() []string {
|
|
|
+ names := make([]string, 0, len(tx.stagedIndexes))
|
|
|
+ for lower := range tx.stagedIndexes {
|
|
|
+ names = append(names, lower)
|
|
|
+ }
|
|
|
+ sort.Strings(names)
|
|
|
+ return names
|
|
|
+}
|
|
|
+
|
|
|
+// hasDataEffects reports whether the commit must rewrite or remove durable rows.
|
|
|
+func (tx *SchemaTx) hasDataEffects() bool {
|
|
|
+ return len(tx.drops) > 0 || len(tx.renames) > 0 || len(tx.rewrites) > 0
|
|
|
+}
|
|
|
+
|
|
|
+// hasDDL reports whether the transaction staged a catalog change that the
|
|
|
+// commit has to publish.
|
|
|
+func (tx *SchemaTx) hasDDL() bool {
|
|
|
+ return len(tx.staged) > 0 || len(tx.stagedIndexes) > 0 || tx.catalogDirty || tx.indexListDirty || tx.hasDataEffects()
|
|
|
+}
|
|
|
+
|
|
|
+// Session catalog access
|
|
|
+//
|
|
|
+// A Session is the catalog view of one SQL connection. Outside an explicit
|
|
|
+// transaction every accessor delegates to the shared SchemaManager; while a
|
|
|
+// transaction is open the accessors resolve against its SchemaTx overlay, so
|
|
|
+// the session reads its own DDL without publishing it.
|
|
|
+
|
|
|
+// GetSchema returns the table definition this session observes.
|
|
|
+func (s *Session) GetSchema(name string) (*Schema, error) {
|
|
|
+ s.mu.Lock()
|
|
|
+ defer s.mu.Unlock()
|
|
|
+ return s.getSchemaLocked(name)
|
|
|
+}
|
|
|
+
|
|
|
+// getSchemaLocked resolves a table definition (caller holds the session lock).
|
|
|
+func (s *Session) getSchemaLocked(name string) (*Schema, error) {
|
|
|
+ if s.schemaTx != nil {
|
|
|
+ return s.schemaTx.getSchema(name)
|
|
|
+ }
|
|
|
+ return s.schema.GetSchema(name)
|
|
|
+}
|
|
|
+
|
|
|
+// TableExists reports whether the session can resolve a table.
|
|
|
+func (s *Session) TableExists(name string) bool {
|
|
|
+ _, err := s.GetSchema(name)
|
|
|
+ return err == nil
|
|
|
+}
|
|
|
+
|
|
|
+// ListTables returns the table catalog this session observes.
|
|
|
+func (s *Session) ListTables() ([]string, error) {
|
|
|
+ s.mu.Lock()
|
|
|
+ defer s.mu.Unlock()
|
|
|
+ if s.schemaTx != nil {
|
|
|
+ return s.schemaTx.listTables()
|
|
|
+ }
|
|
|
+ return s.schema.ListTables()
|
|
|
+}
|
|
|
+
|
|
|
+// GetIndex returns an index definition this session observes.
|
|
|
+func (s *Session) GetIndex(name string) (*Index, error) {
|
|
|
+ s.mu.Lock()
|
|
|
+ defer s.mu.Unlock()
|
|
|
+ if s.schemaTx != nil {
|
|
|
+ return s.schemaTx.getIndex(name)
|
|
|
+ }
|
|
|
+ return s.schema.GetIndex(name)
|
|
|
+}
|
|
|
+
|
|
|
+// IndexExists reports whether the session can resolve an index.
|
|
|
+func (s *Session) IndexExists(name string) bool {
|
|
|
+ _, err := s.GetIndex(name)
|
|
|
+ return err == nil
|
|
|
+}
|
|
|
+
|
|
|
+// ListIndexes returns the index catalog this session observes.
|
|
|
+func (s *Session) ListIndexes() ([]string, error) {
|
|
|
+ s.mu.Lock()
|
|
|
+ defer s.mu.Unlock()
|
|
|
+ if s.schemaTx != nil {
|
|
|
+ return s.schemaTx.listIndexes()
|
|
|
+ }
|
|
|
+ return s.schema.ListIndexes()
|
|
|
+}
|
|
|
+
|
|
|
+// ListTableIndexes returns the index definitions of a table as this session
|
|
|
+// observes them.
|
|
|
+func (s *Session) ListTableIndexes(table string) ([]*Index, error) {
|
|
|
+ s.mu.Lock()
|
|
|
+ defer s.mu.Unlock()
|
|
|
+ if s.schemaTx != nil {
|
|
|
+ return s.schemaTx.listTableIndexes(table)
|
|
|
+ }
|
|
|
+ return s.schema.ListTableIndexes(table)
|
|
|
+}
|
|
|
+
|
|
|
+// Version identifies the catalog generation this session observes, including
|
|
|
+// staged DDL, so the executor resynchronizes its analyzer catalog after a
|
|
|
+// transactional DDL statement and again after the overlay is discarded.
|
|
|
+func (s *Session) Version() uint64 {
|
|
|
+ s.mu.Lock()
|
|
|
+ defer s.mu.Unlock()
|
|
|
+ return s.versionLocked()
|
|
|
+}
|
|
|
+
|
|
|
+func (s *Session) versionLocked() uint64 {
|
|
|
+ if s.schemaTx != nil {
|
|
|
+ return s.schemaTx.version
|
|
|
+ }
|
|
|
+ return s.schema.Version()
|
|
|
+}
|
|
|
+
|
|
|
+// GetNextRowID allocates from the shared per-table ROWID counter. The counter is
|
|
|
+// derived from durable rows and is not part of the catalog, so a transaction
|
|
|
+// uses the same allocator as autocommit.
|
|
|
+func (s *Session) GetNextRowID(table string) (int64, error) {
|
|
|
+ return s.schema.GetNextRowID(table)
|
|
|
+}
|
|
|
+
|
|
|
+// UpdateMaxRowID raises the shared per-table ROWID counter.
|
|
|
+func (s *Session) UpdateMaxRowID(table string, rowid int64) error {
|
|
|
+ return s.schema.UpdateMaxRowID(table, rowid)
|
|
|
+}
|
|
|
+
|
|
|
+// CreateTable stages CREATE TABLE inside a transaction and publishes it
|
|
|
+// immediately otherwise.
|
|
|
+func (s *Session) CreateTable(schema *Schema) error {
|
|
|
+ s.mu.Lock()
|
|
|
+ defer s.mu.Unlock()
|
|
|
+ if s.schemaTx == nil {
|
|
|
+ return s.schema.CreateTable(schema)
|
|
|
+ }
|
|
|
+ return s.stageDDL(func(tx *SchemaTx) error { return tx.createTable(schema) })
|
|
|
+}
|
|
|
+
|
|
|
+// DropTable stages DROP TABLE inside a transaction and publishes it otherwise.
|
|
|
+func (s *Session) DropTable(name string) error {
|
|
|
+ s.mu.Lock()
|
|
|
+ defer s.mu.Unlock()
|
|
|
+ if s.schemaTx == nil {
|
|
|
+ return s.schema.DropTable(name)
|
|
|
+ }
|
|
|
+ return s.stageDDL(func(tx *SchemaTx) error { return tx.dropTable(name) })
|
|
|
+}
|
|
|
+
|
|
|
+// CreateIndex stages CREATE INDEX inside a transaction and publishes it
|
|
|
+// otherwise. A UNIQUE index is validated against the rows the session observes,
|
|
|
+// so a transaction that staged duplicate rows fails the statement, not the
|
|
|
+// commit.
|
|
|
+func (s *Session) CreateIndex(index *Index) error {
|
|
|
+ s.mu.Lock()
|
|
|
+ defer s.mu.Unlock()
|
|
|
+ if s.schemaTx == nil {
|
|
|
+ if index.Unique {
|
|
|
+ return s.table.CreateUniqueIndex(index)
|
|
|
+ }
|
|
|
+ return s.schema.CreateIndex(index)
|
|
|
+ }
|
|
|
+ return s.stageDDL(func(tx *SchemaTx) error {
|
|
|
+ if err := tx.createIndex(index); err != nil {
|
|
|
+ return err
|
|
|
+ }
|
|
|
+ if !index.Unique {
|
|
|
+ return nil
|
|
|
+ }
|
|
|
+ return s.validateUniqueIndexLocked(index)
|
|
|
+ })
|
|
|
+}
|
|
|
+
|
|
|
+// validateUniqueIndexLocked rejects a staged UNIQUE index whose indexed values
|
|
|
+// already collide in the rows this session observes (caller holds the session
|
|
|
+// lock).
|
|
|
+func (s *Session) validateUniqueIndexLocked(index *Index) error {
|
|
|
+ rows, err := s.selectLocked(index.Table, nil)
|
|
|
+ if err != nil {
|
|
|
+ return err
|
|
|
+ }
|
|
|
+ seen := make(map[string]struct{}, len(rows))
|
|
|
+ for _, row := range rows {
|
|
|
+ encoded, isNull, err := s.table.encodeUniqueValue(row, index.Columns)
|
|
|
+ if err != nil {
|
|
|
+ return err
|
|
|
+ }
|
|
|
+ if isNull {
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ if _, duplicate := seen[encoded]; duplicate {
|
|
|
+ return &UniqueViolationError{Constraint: index.Name}
|
|
|
+ }
|
|
|
+ seen[encoded] = struct{}{}
|
|
|
+ }
|
|
|
+ return nil
|
|
|
+}
|
|
|
+
|
|
|
+// DropIndex stages DROP INDEX inside a transaction and publishes it otherwise.
|
|
|
+func (s *Session) DropIndex(name string) error {
|
|
|
+ s.mu.Lock()
|
|
|
+ defer s.mu.Unlock()
|
|
|
+ if s.schemaTx == nil {
|
|
|
+ return s.schema.DropIndex(name)
|
|
|
+ }
|
|
|
+ return s.stageDDL(func(tx *SchemaTx) error { return tx.dropIndex(name) })
|
|
|
+}
|
|
|
+
|
|
|
+// AddColumn stages ALTER TABLE ADD COLUMN inside a transaction and publishes it
|
|
|
+// otherwise.
|
|
|
+func (s *Session) AddColumn(table string, column Column, checks []CheckConstraint, unique *Index) error {
|
|
|
+ s.mu.Lock()
|
|
|
+ defer s.mu.Unlock()
|
|
|
+ if s.schemaTx == nil {
|
|
|
+ return s.schema.AddColumn(table, column, checks, unique)
|
|
|
+ }
|
|
|
+ return s.stageDDL(func(tx *SchemaTx) error { return tx.addColumn(table, column, checks, unique) })
|
|
|
+}
|
|
|
+
|
|
|
+// DropColumn stages ALTER TABLE DROP COLUMN inside a transaction and publishes
|
|
|
+// it otherwise.
|
|
|
+func (s *Session) DropColumn(table, columnName string) error {
|
|
|
+ s.mu.Lock()
|
|
|
+ defer s.mu.Unlock()
|
|
|
+ if s.schemaTx == nil {
|
|
|
+ return s.schema.DropColumn(table, columnName)
|
|
|
+ }
|
|
|
+ return s.stageDDL(func(tx *SchemaTx) error { return tx.dropColumn(table, columnName) })
|
|
|
+}
|
|
|
+
|
|
|
+// RenameTable stages ALTER TABLE RENAME TABLE inside a transaction and
|
|
|
+// publishes it otherwise.
|
|
|
+func (s *Session) RenameTable(oldName, newName string) error {
|
|
|
+ s.mu.Lock()
|
|
|
+ defer s.mu.Unlock()
|
|
|
+ if s.schemaTx == nil {
|
|
|
+ return s.schema.RenameTable(oldName, newName)
|
|
|
+ }
|
|
|
+ return s.stageDDL(func(tx *SchemaTx) error { return tx.renameTable(oldName, newName) })
|
|
|
+}
|
|
|
+
|
|
|
+// RenameColumn stages ALTER TABLE RENAME COLUMN inside a transaction and
|
|
|
+// publishes it otherwise.
|
|
|
+func (s *Session) RenameColumn(table, oldName, newName string) error {
|
|
|
+ s.mu.Lock()
|
|
|
+ defer s.mu.Unlock()
|
|
|
+ if s.schemaTx == nil {
|
|
|
+ return s.schema.RenameColumn(table, oldName, newName)
|
|
|
+ }
|
|
|
+ return s.stageDDL(func(tx *SchemaTx) error { return tx.renameColumn(table, oldName, newName) })
|
|
|
+}
|
|
|
+
|
|
|
+// stageDDL runs one DDL statement against the transaction overlay and restores
|
|
|
+// the catalog, overlay, and read set if it fails, so a rejected statement
|
|
|
+// leaves the transaction exactly as it was (caller holds the session lock).
|
|
|
+func (s *Session) stageDDL(apply func(tx *SchemaTx) error) error {
|
|
|
+ savepoint := s.savepointLocked()
|
|
|
+ if err := apply(s.schemaTx); err != nil {
|
|
|
+ s.rollbackToSavepointLocked(savepoint)
|
|
|
+ return err
|
|
|
+ }
|
|
|
+ return nil
|
|
|
+}
|
|
|
+
|
|
|
+// overlayHolds reports whether the session staged the given durable key of a
|
|
|
+// table.
|
|
|
+func (s *Session) overlayHolds(lowerTable, key string) bool {
|
|
|
+ _, ok := s.overlay[lowerTable][key]
|
|
|
+ return ok
|
|
|
+}
|
|
|
+
|
|
|
+// markTableScanned pins the table's current generation as a read so the commit
|
|
|
+// fails with a serialization error if another session writes the table between
|
|
|
+// the DDL statement and the commit.
|
|
|
+func (s *Session) markTableScanned(table string) {
|
|
|
+ lower := strings.ToLower(table)
|
|
|
+ if _, ok := s.scanGens[lower]; !ok {
|
|
|
+ s.scanGens[lower] = s.table.generation(lower)
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+// dropOverlayTableLocked discards the staged rows of a table the transaction
|
|
|
+// dropped, including their mutation-log entries (caller holds the session lock).
|
|
|
+func (s *Session) dropOverlayTableLocked(lower string) {
|
|
|
+ delete(s.overlay, lower)
|
|
|
+ log := s.log[:0]
|
|
|
+ for _, mutation := range s.log {
|
|
|
+ if mutation.table != lower {
|
|
|
+ log = append(log, mutation)
|
|
|
+ }
|
|
|
+ }
|
|
|
+ s.log = log
|
|
|
+ delete(s.scanGens, lower)
|
|
|
+ for key, predicate := range s.predicateGens {
|
|
|
+ if predicate.table == lower {
|
|
|
+ delete(s.predicateGens, key)
|
|
|
+ }
|
|
|
+ }
|
|
|
+ prefix := s.table.dataPrefix(lower)
|
|
|
+ for key := range s.reads {
|
|
|
+ if strings.HasPrefix(key, prefix) {
|
|
|
+ delete(s.reads, key)
|
|
|
+ }
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+// rekeyOverlayLocked moves the staged rows of a renamed table to the new data
|
|
|
+// prefix so the transaction reads its own rows back under the new name
|
|
|
+// (caller holds the session lock).
|
|
|
+func (s *Session) rekeyOverlayLocked(oldTable, newTable string) {
|
|
|
+ oldLower := strings.ToLower(oldTable)
|
|
|
+ newLower := strings.ToLower(newTable)
|
|
|
+ if oldLower == newLower {
|
|
|
+ return
|
|
|
+ }
|
|
|
+ oldPrefix := s.table.dataPrefix(oldLower)
|
|
|
+ newPrefix := s.table.dataPrefix(newLower)
|
|
|
+
|
|
|
+ if entries, ok := s.overlay[oldLower]; ok {
|
|
|
+ delete(s.overlay, oldLower)
|
|
|
+ moved := make(map[string]*overlayEntry, len(entries))
|
|
|
+ for key, entry := range entries {
|
|
|
+ moved[newPrefix+strings.TrimPrefix(key, oldPrefix)] = entry
|
|
|
+ }
|
|
|
+ s.overlay[newLower] = moved
|
|
|
+ }
|
|
|
+ for i := range s.log {
|
|
|
+ if s.log[i].table != oldLower {
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ s.log[i].table = newLower
|
|
|
+ s.log[i].key = newPrefix + strings.TrimPrefix(s.log[i].key, oldPrefix)
|
|
|
+ }
|
|
|
+ if gen, ok := s.scanGens[oldLower]; ok {
|
|
|
+ delete(s.scanGens, oldLower)
|
|
|
+ s.scanGens[newLower] = gen
|
|
|
+ }
|
|
|
+ reads := make(map[string]uint64, len(s.reads))
|
|
|
+ for key, lsn := range s.reads {
|
|
|
+ if strings.HasPrefix(key, oldPrefix) {
|
|
|
+ reads[newPrefix+strings.TrimPrefix(key, oldPrefix)] = lsn
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ reads[key] = lsn
|
|
|
+ }
|
|
|
+ s.reads = reads
|
|
|
+ predicates := make(map[string]predicateRead, len(s.predicateGens))
|
|
|
+ for key, predicate := range s.predicateGens {
|
|
|
+ if predicate.table == oldLower {
|
|
|
+ predicate.table = newLower
|
|
|
+ }
|
|
|
+ if strings.HasPrefix(key, oldPrefix) {
|
|
|
+ key = newPrefix + strings.TrimPrefix(key, oldPrefix)
|
|
|
+ }
|
|
|
+ predicates[key] = predicate
|
|
|
+ }
|
|
|
+ s.predicateGens = predicates
|
|
|
+}
|
|
|
+
|
|
|
+// ForeignKeyParticipants reports whether a write to table can affect referential
|
|
|
+// integrity as this session observes the catalog.
|
|
|
+func (s *Session) ForeignKeyParticipants(table string) (bool, error) {
|
|
|
+ s.mu.Lock()
|
|
|
+ defer s.mu.Unlock()
|
|
|
+ if s.schemaTx == nil {
|
|
|
+ return s.schema.ForeignKeyParticipants(table)
|
|
|
+ }
|
|
|
+ catalog, err := s.foreignKeyCatalogLocked()
|
|
|
+ if err != nil {
|
|
|
+ return false, err
|
|
|
+ }
|
|
|
+ return catalog.participants[strings.ToLower(table)], nil
|
|
|
+}
|
|
|
+
|
|
|
+// ForeignKeyColumnDependency returns the constraints that keep a column in
|
|
|
+// place, as this session observes the catalog.
|
|
|
+func (s *Session) ForeignKeyColumnDependency(table, column string) ([]string, error) {
|
|
|
+ s.mu.Lock()
|
|
|
+ defer s.mu.Unlock()
|
|
|
+ if s.schemaTx == nil {
|
|
|
+ return s.schema.ForeignKeyColumnDependency(table, column)
|
|
|
+ }
|
|
|
+ catalog, err := s.foreignKeyCatalogLocked()
|
|
|
+ if err != nil {
|
|
|
+ return nil, err
|
|
|
+ }
|
|
|
+ schema, err := s.getSchemaLocked(table)
|
|
|
+ if err != nil {
|
|
|
+ return nil, err
|
|
|
+ }
|
|
|
+ seen := make(map[string]bool)
|
|
|
+ var labels []string
|
|
|
+ add := func(child string, fk ForeignKeyConstraint) {
|
|
|
+ label := foreignKeyReferenceName(child, fk)
|
|
|
+ if seen[label] {
|
|
|
+ return
|
|
|
+ }
|
|
|
+ seen[label] = true
|
|
|
+ labels = append(labels, label)
|
|
|
+ }
|
|
|
+ for _, fk := range schema.ForeignKeys {
|
|
|
+ for _, name := range fk.Columns {
|
|
|
+ if strings.EqualFold(name, column) {
|
|
|
+ add(schema.Name, fk)
|
|
|
+ break
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ for _, ref := range catalog.referencing[strings.ToLower(table)] {
|
|
|
+ for _, name := range ref.FK.RefColumns {
|
|
|
+ if strings.EqualFold(name, column) {
|
|
|
+ add(ref.Table, ref.FK)
|
|
|
+ break
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ sort.Strings(labels)
|
|
|
+ return labels, nil
|
|
|
+}
|
|
|
+
|
|
|
+// ForeignKeyTableDependency returns the constraints that name table as their
|
|
|
+// referenced parent, as this session observes the catalog.
|
|
|
+func (s *Session) ForeignKeyTableDependency(table string) ([]string, error) {
|
|
|
+ s.mu.Lock()
|
|
|
+ defer s.mu.Unlock()
|
|
|
+ if s.schemaTx == nil {
|
|
|
+ return s.schema.ForeignKeyTableDependency(table)
|
|
|
+ }
|
|
|
+ catalog, err := s.foreignKeyCatalogLocked()
|
|
|
+ if err != nil {
|
|
|
+ return nil, err
|
|
|
+ }
|
|
|
+ seen := make(map[string]bool)
|
|
|
+ var labels []string
|
|
|
+ for _, ref := range catalog.referencing[strings.ToLower(table)] {
|
|
|
+ label := foreignKeyReferenceName(ref.Table, ref.FK)
|
|
|
+ if seen[label] {
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ seen[label] = true
|
|
|
+ labels = append(labels, label)
|
|
|
+ }
|
|
|
+ sort.Strings(labels)
|
|
|
+ return labels, nil
|
|
|
+}
|
|
|
+
|
|
|
+// ForeignKeyClosure returns the constrained component of the given tables as
|
|
|
+// this session observes the catalog.
|
|
|
+func (s *Session) ForeignKeyClosure(tables []string) (map[string]bool, error) {
|
|
|
+ s.mu.Lock()
|
|
|
+ defer s.mu.Unlock()
|
|
|
+ if s.schemaTx == nil {
|
|
|
+ return s.schema.ForeignKeyClosure(tables)
|
|
|
+ }
|
|
|
+ catalog, err := s.foreignKeyCatalogLocked()
|
|
|
+ if err != nil {
|
|
|
+ return nil, err
|
|
|
+ }
|
|
|
+ return catalog.closure(tables), nil
|
|
|
+}
|
|
|
+
|
|
|
+// foreignKeyCatalogLocked builds the foreign-key metadata of the session's
|
|
|
+// catalog view (caller holds the session lock).
|
|
|
+func (s *Session) foreignKeyCatalogLocked() (*foreignKeyCatalog, error) {
|
|
|
+ version := s.versionLocked()
|
|
|
+ if s.fkCatalog != nil && s.fkVersion == version {
|
|
|
+ return s.fkCatalog, nil
|
|
|
+ }
|
|
|
+ tables, err := s.schemaTx.listTables()
|
|
|
+ if err != nil {
|
|
|
+ return nil, err
|
|
|
+ }
|
|
|
+ catalog, err := buildForeignKeyCatalog(version, tables, func(name string) (*Schema, error) {
|
|
|
+ return s.getSchemaLocked(name)
|
|
|
+ })
|
|
|
+ if err != nil {
|
|
|
+ return nil, err
|
|
|
+ }
|
|
|
+ s.fkCatalog = catalog
|
|
|
+ s.fkVersion = version
|
|
|
+ return catalog, nil
|
|
|
+}
|
Danilo Fragoso