func (c *columns) Range(fn func(column *column)) {
	cols := c.cols.Load().([]columnEntry)
	for _, v := range cols {
		fn(v.cols[0])
	}
}
Range iterates over columns in the registry.
package column
import (
	"context"
	"fmt"
	"math/bits"
	"reflect"
	"sync"
	"sync/atomic"
	"time"
	"github.com/kelindar/bitmap"
	"github.com/kelindar/column/commit"
	"github.com/kelindar/smutex"
)
type Object = map[string]interface{}
const (
	expireColumn = "expire"
	rowColumn    = "row"
)
type Collection struct {
	count  uint64              
	txns   *txnPool            
	lock   sync.RWMutex        
	slock  *smutex.SMutex128   
	cols   columns             
	fill   bitmap.Bitmap       
	size   int                 
	writer commit.Writer       
	pk     *columnKey          
	cancel context.CancelFunc  
}
type Options struct {
	Capacity int            
	Writer   commit.Writer  
	Vacuum   time.Duration  
}
func NewCollection(opts ...Options) *Collection {
	options := Options{
		Capacity: 1024,
		Vacuum:   1 * time.Second,
		Writer:   nil,
	}
	for _, o := range opts {
		if o.Capacity > 0 {
			options.Capacity = o.Capacity
		}
		if o.Vacuum > 0 {
			options.Vacuum = o.Vacuum
		}
		if o.Writer != nil {
			options.Writer = o.Writer
		}
	}
	ctx, cancel := context.WithCancel(context.Background())
	store := &Collection{
		cols:   makeColumns(8),
		txns:   newTxnPool(),
		size:   options.Capacity,
		slock:  new(smutex.SMutex128),
		fill:   make(bitmap.Bitmap, 0, options.Capacity>>6),
		writer: options.Writer,
		cancel: cancel,
	}
	store.CreateColumn(expireColumn, ForInt64())
	go store.vacuum(ctx, options.Vacuum)
	return store
}
func (c *Collection) next() uint32 {
	c.lock.Lock()
	idx := c.findFreeIndex(atomic.AddUint64(&c.count, 1))
	c.fill.Set(idx)
	c.lock.Unlock()
	return idx
}
func (c *Collection) findFreeIndex(count uint64) uint32 {
	fillSize := len(c.fill)
	if count > uint64(fillSize)<<6 {
		return uint32(len(c.fill)) << 6
	}
	if fillSize > 0 {
		if tail := c.fill[fillSize-1]; tail != 0xffffffffffffffff {
			return uint32((fillSize-1)<<6 + bits.TrailingZeros64(^tail))
		}
	}
	idx, _ := c.fill.MinZero()
	return idx
}
func (c *Collection) InsertObject(obj Object) (index uint32) {
	c.Query(func(txn *Txn) error {
		index, _ = txn.InsertObject(obj)
		return nil
	})
	return
}
func (c *Collection) InsertObjectWithTTL(obj Object, ttl time.Duration) (index uint32) {
	c.Query(func(txn *Txn) error {
		index, _ = txn.InsertObjectWithTTL(obj, ttl)
		return nil
	})
	return
}
func (c *Collection) Insert(columnName string, fn func(v Cursor) error) (index uint32, err error) {
	err = c.Query(func(txn *Txn) (innerErr error) {
		index, innerErr = txn.Insert(columnName, fn)
		return
	})
	return
}
func (c *Collection) InsertWithTTL(columnName string, ttl time.Duration, fn func(v Cursor) error) (index uint32, err error) {
	err = c.Query(func(txn *Txn) (innerErr error) {
		index, innerErr = txn.InsertWithTTL(columnName, ttl, fn)
		return
	})
	return
}
func (c *Collection) UpdateAt(idx uint32, columnName string, fn func(v Cursor) error) error {
	return c.Query(func(txn *Txn) error {
		return txn.UpdateAt(idx, columnName, fn)
	})
}
func (c *Collection) UpdateAtKey(key, columnName string, fn func(v Cursor) error) error {
	return c.Query(func(txn *Txn) error {
		return txn.UpdateAtKey(key, columnName, fn)
	})
}
func (c *Collection) SelectAt(idx uint32, fn func(v Selector)) bool {
	chunk := uint(idx >> chunkShift)
	if idx >= uint32(len(c.fill))<<6 || !c.fill.Contains(idx) {
		return false
	}
	c.slock.RLock(chunk)
	fn(Selector{idx: idx, col: c})
	c.slock.RUnlock(chunk)
	return true
}
func (c *Collection) SelectAtKey(key string, fn func(v Selector)) (found bool) {
	if c.pk == nil {
		return false
	}
	if idx, ok := c.pk.OffsetOf(key); ok {
		found = c.SelectAt(idx, fn)
	}
	return
}
func (c *Collection) DeleteAt(idx uint32) (deleted bool) {
	c.Query(func(txn *Txn) error {
		deleted = txn.DeleteAt(idx)
		return nil
	})
	return
}
func (c *Collection) Count() (count int) {
	return int(atomic.LoadUint64(&c.count))
}
func (c *Collection) createColumnKey(columnName string, column *columnKey) error {
	if c.pk != nil {
		return fmt.Errorf("column: unable to create key column '%s', another one exists", columnName)
	}
	c.pk = column
	c.pk.name = columnName
	return nil
}
func (c *Collection) CreateColumnsOf(object Object) error {
	for k, v := range object {
		c.CreateColumn(k, ForKind(reflect.TypeOf(v).Kind()))
	}
	return nil
}
func (c *Collection) CreateColumn(columnName string, column Column) error {
	column.Grow(uint32(c.size))
	c.cols.Store(columnName, columnFor(columnName, column))
	if pk, ok := column.(*columnKey); ok {
		return c.createColumnKey(columnName, pk)
	}
	return nil
}
func (c *Collection) DropColumn(columnName string) {
	c.cols.DeleteColumn(columnName)
}
func (c *Collection) CreateIndex(indexName, columnName string, fn func(r Reader) bool) error {
	if fn == nil || columnName == "" || indexName == "" {
		return fmt.Errorf("column: create index must specify name, column and function")
	}
	column, ok := c.cols.Load(columnName)
	if !ok {
		return fmt.Errorf("column: unable to create index, column '%v' does not exist", columnName)
	}
	index := newIndex(indexName, columnName, fn)
	c.lock.Lock()
	index.Grow(uint32(c.size))
	c.cols.Store(indexName, index)
	c.cols.Store(columnName, column, index)
	c.lock.Unlock()
	return c.Query(func(txn *Txn) error {
		impl := index.Column.(*columnIndex)
		return txn.With(columnName).Range(columnName, func(v Cursor) {
			impl.Update(&v)
		})
	})
}
func (c *Collection) DropIndex(indexName string) error {
	column, exists := c.cols.Load(indexName)
	if !exists {
		return fmt.Errorf("column: unable to drop index, index '%v' does not exist", indexName)
	}
	if _, ok := column.Column.(computed); !ok {
		return fmt.Errorf("column: unable to drop index, '%v' is not an index", indexName)
	}
	columnName := column.Column.(computed).Column()
	c.cols.DeleteIndex(columnName, indexName)
	c.cols.DeleteColumn(indexName)
	return nil
}
func (c *Collection) Query(fn func(txn *Txn) error) error {
	c.lock.RLock()
	txn := c.txns.acquire(c)
	c.lock.RUnlock()
	if err := fn(txn); err != nil {
		txn.rollback()
		c.txns.release(txn)
		return err
	}
	txn.commit()
	c.txns.release(txn)
	return nil
}
func (c *Collection) Close() error {
	c.cancel()
	return nil
}
func (c *Collection) vacuum(ctx context.Context, interval time.Duration) {
	ticker := time.NewTicker(interval)
	for {
		select {
		case <-ctx.Done():
			ticker.Stop()
			return
		case <-ticker.C:
			now := int(time.Now().UnixNano())
			c.Query(func(txn *Txn) error {
				return txn.With(expireColumn).Range(expireColumn, func(v Cursor) {
					if expirateAt := v.Int(); expirateAt != 0 && now >= v.Int() {
						v.Delete()
					}
				})
			})
		}
	}
}
func (c *Collection) Replay(change commit.Commit) error {
	return c.Query(func(txn *Txn) error {
		txn.dirty.Set(change.Chunk)
		for i := range change.Updates {
			if !change.Updates[i].IsEmpty() {
				txn.updates = append(txn.updates, change.Updates[i])
			}
		}
		return nil
	})
}
type columns struct {
	cols *atomic.Value
}
func makeColumns(capacity int) columns {
	data := columns{
		cols: &atomic.Value{},
	}
	data.cols.Store(make([]columnEntry, 0, capacity))
	return data
}
type columnEntry struct {
	name string     
	cols []*column  
}