func (txn *Txn) InsertObjectWithTTL(object Object, ttl time.Duration) (uint32, error) {
	return txn.insertObject(object, time.Now().Add(ttl).UnixNano())
}
InsertObjectWithTTL adds an object to a collection, sets the expiration time
based on the specified time-to-live and returns the allocated index.
package column
import (
	"errors"
	"sync"
	"sync/atomic"
	"time"
	"github.com/kelindar/bitmap"
	"github.com/kelindar/column/commit"
)
var (
	errNoKey = errors.New("column: collection does not have a key column")
)
type txnPool struct {
	txns  sync.Pool
	pages sync.Pool
}
func newTxnPool() *txnPool {
	return &txnPool{
		txns: sync.Pool{
			New: func() interface{} {
				return &Txn{
					index:   make(bitmap.Bitmap, 0, 4),
					dirty:   make(bitmap.Bitmap, 0, 4),
					updates: make([]*commit.Buffer, 0, 256),
					columns: make([]columnCache, 0, 16),
					reader:  commit.NewReader(),
				}
			},
		},
		pages: sync.Pool{
			New: func() interface{} {
				return commit.NewBuffer(chunkSize)
			},
		},
	}
}
func (p *txnPool) acquire(owner *Collection) *Txn {
	txn := p.txns.Get().(*Txn)
	txn.owner = owner
	txn.writer = owner.writer
	txn.index.Grow(uint32(owner.size))
	owner.fill.Clone(&txn.index)
	return txn
}
func (p *txnPool) release(txn *Txn) {
	p.txns.Put(txn)
}
func (p *txnPool) acquirePage(columnName string) *commit.Buffer {
	page := p.pages.Get().(*commit.Buffer)
	page.Reset(columnName)
	return page
}
func (p *txnPool) releasePage(buffer *commit.Buffer) {
	buffer.Reset("")
	p.pages.Put(buffer)
}
type Txn struct {
	owner   *Collection       
	index   bitmap.Bitmap     
	dirty   bitmap.Bitmap     
	updates []*commit.Buffer  
	columns []columnCache     
	writer  commit.Writer     
	reader  *commit.Reader    
}
func (txn *Txn) reset() {
	for i := range txn.updates {
		txn.owner.txns.releasePage(txn.updates[i])
	}
	txn.dirty.Clear()
	txn.reader.Rewind()
	txn.columns = txn.columns[:0]
	txn.updates = txn.updates[:0]
}
type columnCache struct {
	name string   
	col  *column  
}
func (txn *Txn) columnAt(columnName string) (*column, bool) {
	for _, v := range txn.columns {
		if v.name == columnName {
			return v.col, true
		}
	}
	column, ok := txn.owner.cols.Load(columnName)
	if !ok {
		return nil, false
	}
	txn.columns = append(txn.columns, columnCache{
		name: columnName,
		col:  column,
	})
	return column, true
}
func (txn *Txn) With(columns ...string) *Txn {
	for _, columnName := range columns {
		if idx, ok := txn.columnAt(columnName); ok {
			txn.rangeReadPair(*idx.Column.Index(), func(dst, src bitmap.Bitmap) {
				dst.And(src)
			})
		} else {
			txn.index.Clear()
		}
	}
	return txn
}
func (txn *Txn) Without(columns ...string) *Txn {
	for _, columnName := range columns {
		if idx, ok := txn.columnAt(columnName); ok {
			txn.rangeReadPair(*idx.Column.Index(), func(dst, src bitmap.Bitmap) {
				dst.AndNot(src)
			})
		}
	}
	return txn
}
func (txn *Txn) Union(columns ...string) *Txn {
	for _, columnName := range columns {
		if idx, ok := txn.columnAt(columnName); ok {
			txn.rangeReadPair(*idx.Column.Index(), func(dst, src bitmap.Bitmap) {
				dst.Or(src)
			})
		}
	}
	return txn
}
func (txn *Txn) WithValue(column string, predicate func(v interface{}) bool) *Txn {
	c, ok := txn.columnAt(column)
	if !ok {
		txn.index.Clear()
		return txn
	}
	txn.rangeRead(func(offset uint32, index bitmap.Bitmap) {
		index.Filter(func(x uint32) (match bool) {
			if v, ok := c.Value(offset + x); ok {
				match = predicate(v)
			}
			return
		})
	})
	return txn
}
func (txn *Txn) WithFloat(column string, predicate func(v float64) bool) *Txn {
	c, ok := txn.columnAt(column)
	if !ok || !c.IsNumeric() {
		txn.index.Clear()
		return txn
	}
	txn.rangeRead(func(offset uint32, index bitmap.Bitmap) {
		c.Column.(Numeric).FilterFloat64(offset, index, predicate)
	})
	return txn
}
func (txn *Txn) WithInt(column string, predicate func(v int64) bool) *Txn {
	c, ok := txn.columnAt(column)
	if !ok || !c.IsNumeric() {
		txn.index.Clear()
		return txn
	}
	txn.rangeRead(func(offset uint32, index bitmap.Bitmap) {
		c.Column.(Numeric).FilterInt64(offset, index, predicate)
	})
	return txn
}
func (txn *Txn) WithUint(column string, predicate func(v uint64) bool) *Txn {
	c, ok := txn.columnAt(column)
	if !ok || !c.IsNumeric() {
		txn.index.Clear()
		return txn
	}
	txn.rangeRead(func(offset uint32, index bitmap.Bitmap) {
		c.Column.(Numeric).FilterUint64(offset, index, predicate)
	})
	return txn
}
func (txn *Txn) WithString(column string, predicate func(v string) bool) *Txn {
	c, ok := txn.columnAt(column)
	if !ok || !c.IsTextual() {
		txn.index.Clear()
		return txn
	}
	txn.rangeRead(func(offset uint32, index bitmap.Bitmap) {
		c.Column.(Textual).FilterString(offset, index, predicate)
	})
	return txn
}
func (txn *Txn) Count() int {
	return int(txn.index.Count())
}
func (txn *Txn) UpdateAtKey(key, columnName string, fn func(v Cursor) error) error {
	if txn.owner.pk == nil {
		return errNoKey
	}
	if idx, ok := txn.owner.pk.OffsetOf(key); ok {
		return txn.UpdateAt(idx, columnName, fn)
	}
	idx, err := txn.insert(columnName, fn, 0)
	txn.bufferFor(txn.owner.pk.name).PutString(commit.Put, idx, key)
	return err
}
func (txn *Txn) UpdateAt(index uint32, columnName string, fn func(v Cursor) error) error {
	cursor, err := txn.cursorFor(columnName)
	if err != nil {
		return err
	}
	cursor.idx = index
	return fn(cursor)
}
func (txn *Txn) SelectAt(index uint32, fn func(v Selector)) bool {
	return txn.owner.SelectAt(index, fn)
}
func (txn *Txn) SelectAtKey(key string, fn func(v Selector)) (found bool) {
	return txn.owner.SelectAtKey(key, fn)
}
func (txn *Txn) DeleteAt(index uint32) bool {
	if !txn.index.Contains(index) {
		return false
	}
	txn.deleteAt(index)
	return true
}
func (txn *Txn) deleteAt(idx uint32) {
	txn.bufferFor(rowColumn).PutOperation(commit.Delete, idx)
}
func (txn *Txn) InsertObject(object Object) (uint32, error) {
	return txn.insertObject(object, 0)
}