Permalink
eccf3e9 Feb 28, 2018
3 contributors

Users who have contributed to this file

@cznic @gernest @dbrower
995 lines (849 sloc) 20.4 KB
// Copyright (c) 2014 ql Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package ql
import (
"fmt"
"strings"
)
type storage interface {
Acid() bool
BeginTransaction() error
Close() error
Commit() error
Create(data ...interface{}) (h int64, err error)
CreateIndex(unique bool) (handle int64, x btreeIndex, err error)
CreateTemp(asc bool) (bt temp, err error)
Delete(h int64, blobCols ...*col) error //LATER split the nil blobCols case
ID() (id int64, err error)
Name() string
OpenIndex(unique bool, handle int64) (btreeIndex, error) // Never called on the memory backend.
Read(dst []interface{}, h int64, cols ...*col) (data []interface{}, err error)
ResetID() (err error)
Rollback() error
Update(h int64, data ...interface{}) error
UpdateRow(h int64, blobCols []*col, data ...interface{}) error
Verify() (allocs int64, err error)
}
type btreeIterator interface {
Next() (k, v []interface{}, err error)
}
type temp interface {
BeginTransaction() error
Create(data ...interface{}) (h int64, err error)
Drop() (err error)
Get(k []interface{}) (v []interface{}, err error)
Read(dst []interface{}, h int64, cols ...*col) (data []interface{}, err error)
SeekFirst() (e btreeIterator, err error)
Set(k, v []interface{}) (err error)
}
type indexIterator interface {
Next() (k []interface{}, h int64, err error)
Prev() (k []interface{}, h int64, err error)
}
type btreeIndex interface {
Clear() error // supports truncate table statement
Create(indexedValues []interface{}, h int64) error // supports insert into statement
Delete(indexedValues []interface{}, h int64) error // supports delete from statement
Drop() error // supports drop table, drop index statements
Seek(indexedValues []interface{}) (iter indexIterator, hit bool, err error) // supports where clause
SeekFirst() (iter indexIterator, err error) // supports aggregate min / ascending order by
SeekLast() (iter indexIterator, err error) // supports aggregate max / descending order by
}
type indexedCol struct { // Column name or id() index.
name string
unique bool
x btreeIndex
xroot int64
}
type index2 struct { // Expression list index.
unique bool
x btreeIndex
xroot int64
sources []string
exprList []expression
}
func (x *index2) eval(ctx *execCtx, cols []*col, id int64, r []interface{}) ([]interface{}, error) {
f, isFile := ctx.db.store.(*file)
vlist := make([]interface{}, len(x.exprList))
m := map[interface{}]interface{}{"$id": id}
for _, col := range cols {
ci := col.index
v := interface{}(nil)
if ci < len(r) {
v = r[ci]
}
if b, ok := v.([]byte); ok && isFile {
var err error
if v, err = expand1(chunk{f: f, b: b}, nil); err != nil {
return nil, err
}
}
m[col.name] = v
}
for i, e := range x.exprList {
v, err := e.eval(ctx, m)
if err != nil {
return nil, err
}
if ok, typ := isBlobType(v); ok {
return nil, fmt.Errorf("value of a complex index cannot be of blob-like type: %v", typ)
}
vlist[i] = v
}
return vlist, nil
}
type indexKey struct {
value []interface{}
h int64
}
// storage fields
// 0: next int64
// 1: scols string
// 2: hhead int64
// 3: name string
// 4: indices string - optional
// 5: hxroots int64 - optional
type table struct {
cols []*col // logical
cols0 []*col // physical
h int64 //
head int64 // head of the single linked record list
hhead int64 // handle of the head of the single linked record list
hxroots int64
indices []*indexedCol
indices2 map[string]*index2
name string
next int64 // single linked table list
store storage
tnext *table
tprev *table
xroots []interface{}
constraints []*constraint
defaults []expression
}
func (t *table) hasIndices() bool { return len(t.indices) != 0 || len(t.indices2) != 0 }
func (t *table) constraintsAndDefaults(ctx *execCtx) error {
if isSystemName[t.name] {
return nil
}
_, ok := ctx.db.root.tables["__Column2"]
if !ok {
return nil
}
cols := t.cols
constraints := make([]*constraint, len(cols))
defaults := make([]expression, len(cols))
arg := []interface{}{t.name}
rs, err := selectColumn2.l[0].exec(newExecCtx(ctx.db, arg))
if err != nil {
return err
}
var rows [][]interface{}
ok = false
if err := rs.(recordset).do(
newExecCtx(ctx.db, arg),
func(id interface{}, data []interface{}) (more bool, err error) {
rows = append(rows, data)
return true, nil
},
); err != nil {
return err
}
for _, row := range rows {
nm := row[0].(string)
nonNull := row[1].(bool)
cexpr := row[2].(string)
dexpr := row[3].(string)
for i, c := range cols {
if c.name == nm {
var co *constraint
if nonNull || cexpr != "" {
co = &constraint{}
constraints[i] = co
if cexpr != "" {
if co.expr, err = ctx.db.str2expr(cexpr); err != nil {
return fmt.Errorf("constraint %q: %v", cexpr, err)
}
}
t.constraints = constraints
}
if dexpr != "" {
if defaults[i], err = ctx.db.str2expr(dexpr); err != nil {
return fmt.Errorf("constraint %q: %v", dexpr, err)
}
t.defaults = defaults
}
}
}
}
return nil
}
func (t *table) checkConstraintsAndDefaults(ctx *execCtx, row []interface{}, m map[interface{}]interface{}) error {
cols := t.cols
if len(t.defaults) != 0 {
// 1.
for _, c := range cols {
m[c.name] = row[c.index]
}
// 2.
for i, c := range cols {
if i >= len(t.defaults) {
break
}
val := row[c.index]
expr := t.defaults[i]
if val != nil || expr == nil {
continue
}
dval, err := expr.eval(ctx, m)
if err != nil {
return err
}
row[c.index] = dval
if err = typeCheck(row, []*col{c}); err != nil {
return err
}
}
}
if len(t.constraints) != 0 {
// 3.
for _, c := range cols {
m[c.name] = row[c.index]
}
// 4.
for i, c := range cols {
if i >= len(t.constraints) {
break
}
constraint := t.constraints[i]
if constraint == nil {
continue
}
val := row[c.index]
expr := constraint.expr
if expr == nil { // Constraint: NOT NULL
if val == nil {
return fmt.Errorf("column %s: constraint violation: NOT NULL", c.name)
}
continue
}
// Constraint is an expression
cval, err := expr.eval(ctx, m)
if err != nil {
return err
}
if cval == nil {
return fmt.Errorf("column %s: constraint violation: %s", c.name, expr)
}
bval, ok := cval.(bool)
if !ok {
return fmt.Errorf("column %s: non bool constraint expression: %s", c.name, expr)
}
if !bval {
return fmt.Errorf("column %s: constraint violation: %s", c.name, expr)
}
}
}
return nil
}
func (t *table) clone() *table {
r := &table{}
*r = *t
r.constraints = append([]*constraint(nil), t.constraints...)
r.defaults = append([]expression(nil), t.defaults...)
r.indices2 = nil
if n := len(t.indices2); n != 0 {
r.indices2 = make(map[string]*index2, n)
for k, v := range t.indices2 {
r.indices2[k] = v
}
}
r.cols = make([]*col, len(t.cols))
for i, v := range t.cols {
c := &col{}
*c = *v
r.cols[i] = c
}
r.cols0 = make([]*col, len(t.cols0))
for i, v := range t.cols0 {
c := &col{}
*c = *v
r.cols0[i] = c
}
r.indices = make([]*indexedCol, len(t.indices))
for i, v := range t.indices {
if v != nil {
c := &indexedCol{}
*c = *v
r.indices[i] = c
}
}
r.xroots = make([]interface{}, len(t.xroots))
copy(r.xroots, t.xroots)
r.tnext, r.tprev = nil, nil
return r
}
func (t *table) findIndexByColName(name string) (*col, *indexedCol) {
for i, v := range t.indices {
if v == nil {
continue
}
if i == 0 {
if name == "id()" {
return idCol, v
}
continue
}
if c := t.cols0[i-1]; c.name == name {
return c, v
}
}
return nil, nil
}
func (t *table) findIndexByName(name string) interface{} {
for _, v := range t.indices {
if v != nil && v.name == name {
return v
}
}
for k, v := range t.indices2 {
if k == name {
return v
}
}
return nil
}
func (t *table) load() (err error) {
data, err := t.store.Read(nil, t.h)
if err != nil {
return
}
var hasIndices bool
switch n := len(data); n {
case 4:
case 6:
hasIndices = true
default:
return fmt.Errorf("corrupted DB: table data len %d", n)
}
var ok bool
if t.next, ok = data[0].(int64); !ok {
return fmt.Errorf("corrupted DB: table data[0] of type %T", data[0])
}
scols, ok := data[1].(string)
if !ok {
return fmt.Errorf("corrupted DB: table data[1] of type %T", data[1])
}
if t.hhead, ok = data[2].(int64); !ok {
return fmt.Errorf("corrupted DB: table data[2] of type %T", data[2])
}
if t.name, ok = data[3].(string); !ok {
return fmt.Errorf("corrupted DB: table data[3] of type %T", data[3])
}
var head []interface{}
if head, err = t.store.Read(nil, t.hhead); err != nil {
return err
}
if len(head) != 1 {
return fmt.Errorf("corrupted DB: table head data len %d", len(head))
}
if t.head, ok = head[0].(int64); !ok {
return fmt.Errorf("corrupted DB: table head data[0] of type %T", head[0])
}
a := strings.Split(scols, "|")
t.cols0 = make([]*col, len(a))
for i, v := range a {
if len(v) < 1 {
return fmt.Errorf("corrupted DB: field info %q", v)
}
col := &col{name: v[1:], typ: int(v[0]), index: i}
t.cols0[i] = col
if col.name != "" {
t.cols = append(t.cols, col)
}
}
if !hasIndices {
return
}
if t.hxroots, ok = data[5].(int64); !ok {
return fmt.Errorf("corrupted DB: table data[5] of type %T", data[5])
}
xroots, err := t.store.Read(nil, t.hxroots)
if err != nil {
return err
}
if g, e := len(xroots), len(t.cols0)+1; g != e {
return fmt.Errorf("corrupted DB: got %d index roots, expected %d", g, e)
}
indices, ok := data[4].(string)
if !ok {
return fmt.Errorf("corrupted DB: table data[4] of type %T", data[4])
}
a = strings.Split(indices, "|")
if g, e := len(a), len(t.cols0)+1; g != e {
return fmt.Errorf("corrupted DB: got %d index definitions, expected %d", g, e)
}
t.indices = make([]*indexedCol, len(a))
for i, v := range a {
if v == "" {
continue
}
if len(v) < 2 {
return fmt.Errorf("corrupted DB: invalid index definition %q", v)
}
nm := v[1:]
h, ok := xroots[i].(int64)
if !ok {
return fmt.Errorf("corrupted DB: table index root of type %T", xroots[i])
}
if h == 0 {
return fmt.Errorf("corrupted DB: missing root for index %s", nm)
}
unique := v[0] == 'u'
x, err := t.store.OpenIndex(unique, h)
if err != nil {
return err
}
t.indices[i] = &indexedCol{nm, unique, x, h}
}
t.xroots = xroots
return
}
func newTable(store storage, name string, next int64, cols []*col, tprev, tnext *table) (t *table, err error) {
hhead, err := store.Create(int64(0))
if err != nil {
return
}
scols := cols2meta(cols)
h, err := store.Create(next, scols, hhead, name)
if err != nil {
return
}
t = &table{
cols0: cols,
h: h,
hhead: hhead,
name: name,
next: next,
store: store,
tnext: tnext,
tprev: tprev,
}
return t.updateCols(), nil
}
func (t *table) blobCols() (r []*col) {
for _, c := range t.cols0 {
switch c.typ {
case qBlob, qBigInt, qBigRat, qTime, qDuration:
r = append(r, c)
}
}
return
}
func (t *table) truncate() (err error) {
h := t.head
var rec []interface{}
blobCols := t.blobCols()
for h != 0 {
rec, err := t.store.Read(rec, h)
if err != nil {
return err
}
nh := rec[0].(int64)
if err = t.store.Delete(h, blobCols...); err != nil { //LATER remove double read for len(blobCols) != 0
return err
}
h = nh
}
if err = t.store.Update(t.hhead, 0); err != nil {
return
}
for _, v := range t.indices {
if v == nil {
continue
}
if err := v.x.Clear(); err != nil {
return err
}
}
for _, ix := range t.indices2 {
if err := ix.x.Clear(); err != nil {
return err
}
}
t.head = 0
return t.updated()
}
func (t *table) addIndex0(unique bool, indexName string, colIndex int) (int64, btreeIndex, error) {
switch len(t.indices) {
case 0:
indices := make([]*indexedCol, len(t.cols0)+1)
h, x, err := t.store.CreateIndex(unique)
if err != nil {
return -1, nil, err
}
indices[colIndex+1] = &indexedCol{indexName, unique, x, h}
xroots := make([]interface{}, len(indices))
xroots[colIndex+1] = h
hx, err := t.store.Create(xroots...)
if err != nil {
return -1, nil, err
}
t.hxroots, t.xroots, t.indices = hx, xroots, indices
return h, x, t.updated()
default:
ex := t.indices[colIndex+1]
if ex != nil && ex.name != "" {
colName := "id()"
if colIndex >= 0 {
colName = t.cols0[colIndex].name
}
return -1, nil, fmt.Errorf("column %s already has an index: %s", colName, ex.name)
}
h, x, err := t.store.CreateIndex(unique)
if err != nil {
return -1, nil, err
}
t.xroots[colIndex+1] = h
if err := t.store.Update(t.hxroots, t.xroots...); err != nil {
return -1, nil, err
}
t.indices[colIndex+1] = &indexedCol{indexName, unique, x, h}
return h, x, t.updated()
}
}
func (t *table) addIndex(unique bool, indexName string, colIndex int) (int64, error) {
hx, x, err := t.addIndex0(unique, indexName, colIndex)
if err != nil {
return -1, err
}
// Must fill the new index.
ncols := len(t.cols0)
h, store := t.head, t.store
for h != 0 {
rec, err := store.Read(nil, h, t.cols...)
if err != nil {
return -1, err
}
if n := ncols + 2 - len(rec); n > 0 {
rec = append(rec, make([]interface{}, n)...)
}
if err = x.Create([]interface{}{rec[colIndex+2]}, h); err != nil {
return -1, err
}
h = rec[0].(int64)
}
return hx, nil
}
func (t *table) addIndex2(execCtx *execCtx, unique bool, indexName string, exprList []expression) (int64, error) {
if _, ok := t.indices2[indexName]; ok {
panic("internal error 009")
}
hx, x, err := t.store.CreateIndex(unique)
if err != nil {
return -1, err
}
var a []string
for _, v := range exprList {
a = append(a, v.String())
}
x2 := &index2{unique, x, hx, a, exprList}
if t.indices2 == nil {
t.indices2 = map[string]*index2{}
}
t.indices2[indexName] = x2
// Must fill the new index.
m := map[interface{}]interface{}{}
h, store := t.head, t.store
for h != 0 {
rec, err := store.Read(nil, h, t.cols...)
if err != nil {
return -1, err
}
for _, col := range t.cols {
ci := col.index
v := interface{}(nil)
if ci < len(rec) {
v = rec[ci+2]
}
m[col.name] = v
}
id := rec[1].(int64)
vlist, err := x2.eval(execCtx, t.cols, id, rec[2:])
if err != nil {
return -1, err
}
if err := x2.x.Create(vlist, h); err != nil {
return -1, err
}
h = rec[0].(int64)
}
return hx, nil
}
func (t *table) dropIndex(xIndex int) error {
t.xroots[xIndex] = 0
if err := t.indices[xIndex].x.Drop(); err != nil {
return err
}
t.indices[xIndex] = nil
return t.updated()
}
func (t *table) updated() (err error) {
switch {
case len(t.indices) != 0:
a := []string{}
for _, v := range t.indices {
if v == nil {
a = append(a, "")
continue
}
s := "n"
if v.unique {
s = "u"
}
a = append(a, s+v.name)
}
return t.store.Update(t.h, t.next, cols2meta(t.updateCols().cols0), t.hhead, t.name, strings.Join(a, "|"), t.hxroots)
default:
return t.store.Update(t.h, t.next, cols2meta(t.updateCols().cols0), t.hhead, t.name)
}
}
// storage fields
// 0: next record handle int64
// 1: record id int64
// 2...: data row
func (t *table) addRecord(execCtx *execCtx, r []interface{}) (id int64, err error) {
if id, err = t.store.ID(); err != nil {
return
}
r = append([]interface{}{t.head, id}, r...)
h, err := t.store.Create(r...)
if err != nil {
return
}
for i, v := range t.indices {
if v == nil {
continue
}
if err = v.x.Create([]interface{}{r[i+1]}, h); err != nil {
return
}
}
for _, ix := range t.indices2 {
vlist, err := ix.eval(execCtx, t.cols, id, r[2:])
if err != nil {
return -1, err
}
if err := ix.x.Create(vlist, h); err != nil {
return -1, err
}
}
if err = t.store.Update(t.hhead, h); err != nil {
return
}
t.head = h
return
}
func (t *table) fieldNames() []string {
r := make([]string, len(t.cols))
for i, v := range t.cols {
r[i] = v.name
}
return r
}
func (t *table) updateCols() *table {
t.cols = t.cols[:0]
for i, c := range t.cols0 {
if c.name != "" {
c.index = i
t.cols = append(t.cols, c)
}
}
return t
}
func (t *table) row0(ctx *execCtx, h int64) ([]interface{}, error) {
rec, err := ctx.db.store.Read(nil, h, t.cols...)
if err != nil {
return nil, err
}
if d := len(t.cols) - (len(rec) - 2); d > 0 {
rec = append(rec, make([]interface{}, d)...)
}
return rec, nil
}
func (t *table) row(ctx *execCtx, h int64) (int64, []interface{}, error) {
rec, err := t.row0(ctx, h)
if err != nil {
return -1, nil, err
}
id := rec[1].(int64)
for i, c := range t.cols {
rec[i] = rec[c.index+2]
}
return id, rec[:len(t.cols)], nil
}
// storage fields
// 0: handle of first table in DB int64
type root struct {
head int64 // Single linked table list
lastInsertID int64
parent *root
//rowsAffected int64 //LATER implement
store storage
tables map[string]*table
thead *table
}
func newRoot(store storage) (r *root, err error) {
data, err := store.Read(nil, 1)
if err != nil {
return
}
switch len(data) {
case 0: // new empty DB, create empty table list
if err = store.BeginTransaction(); err != nil {
return
}
if err = store.Update(1, int64(0)); err != nil {
store.Rollback()
return
}
if err = store.Commit(); err != nil {
return
}
return &root{
store: store,
tables: map[string]*table{},
}, nil
case 1: // existing DB, load tables
if len(data) != 1 {
return nil, fmt.Errorf("corrupted DB: root is an %d-scalar", len(data))
}
p, ok := data[0].(int64)
if !ok {
return nil, fmt.Errorf("corrupted DB: root head has type %T", data[0])
}
r := &root{
head: p,
store: store,
tables: map[string]*table{},
}
var tprev *table
for p != 0 {
t := &table{
h: p,
store: store,
tprev: tprev,
}
if r.thead == nil {
r.thead = t
}
if tprev != nil {
tprev.tnext = t
}
tprev = t
if err = t.load(); err != nil {
return nil, err
}
if r.tables[t.name] != nil { // duplicate
return nil, fmt.Errorf("corrupted DB: duplicate table metadata for table %s", t.name)
}
r.tables[t.name] = t
p = t.next
}
return r, nil
default:
return nil, errIncompatibleDBFormat
}
}
func (r *root) findIndexByName(name string) (*table, interface{}) {
for _, t := range r.tables {
if i := t.findIndexByName(name); i != nil {
return t, i
}
}
return nil, nil
}
func (r *root) updated() (err error) {
return r.store.Update(1, r.head)
}
func (r *root) createTable(name string, cols []*col) (t *table, err error) {
if _, ok := r.tables[name]; ok {
panic("internal error 065")
}
if t, err = newTable(r.store, name, r.head, cols, nil, r.thead); err != nil {
return nil, err
}
if err = r.store.Update(1, t.h); err != nil {
return nil, err
}
if p := r.thead; p != nil {
p.tprev = t
}
r.tables[name], r.head, r.thead = t, t.h, t
return
}
func (r *root) dropTable(t *table) (err error) {
defer func() {
if err != nil {
return
}
delete(r.tables, t.name)
}()
if err = t.truncate(); err != nil {
return
}
if err = t.store.Delete(t.hhead); err != nil {
return
}
if err = t.store.Delete(t.h); err != nil {
return
}
for _, v := range t.indices {
if v != nil && v.x != nil {
if err = v.x.Drop(); err != nil {
return
}
}
}
for _, v := range t.indices2 {
if err = v.x.Drop(); err != nil {
return
}
}
if h := t.hxroots; h != 0 {
if err = t.store.Delete(h); err != nil {
return
}
}
switch {
case t.tprev == nil && t.tnext == nil:
r.head = 0
r.thead = nil
err = r.updated()
return errSet(&err, r.store.ResetID())
case t.tprev == nil && t.tnext != nil:
next := t.tnext
next.tprev = nil
r.head = next.h
r.thead = next
if err = r.updated(); err != nil {
return
}
return next.updated()
case t.tprev != nil && t.tnext == nil: // last in list
prev := t.tprev
prev.next = 0
prev.tnext = nil
return prev.updated()
default: //case t.tprev != nil && t.tnext != nil:
prev, next := t.tprev, t.tnext
prev.next = next.h
prev.tnext = next
next.tprev = prev
if err = prev.updated(); err != nil {
return
}
return next.updated()
}
}