mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-09-07 17:15:56 +00:00
Merge branch 'master' into fast-write
This commit is contained in:
Ben Johnson
GitHub
commit
33334511c4
17 changed files with 277 additions and 203 deletions
1
.gitignore
vendored
1
.gitignore
vendored
|
|
@ -10,3 +10,4 @@ release-pilosa-fsck.*.*.tar.gz
|
|||
/log.*
|
||||
/tourna.log.*
|
||||
pilosa
|
||||
*.dot
|
||||
|
|
|
|||
|
|
@ -15,6 +15,7 @@
|
|||
package ctl
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/pilosa/pilosa/v2"
|
||||
|
|
@ -89,7 +90,9 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) {
|
|||
flags.IntVar(&srv.Config.Profile.MutexFraction, "profile.mutex-fraction", srv.Config.Profile.MutexFraction, "Sampling fraction for mutex contention profiling. Sample 1/<rate> of events.")
|
||||
|
||||
// Transactional storage engine
|
||||
flags.StringVarP(&srv.Config.Txsrc, "tx", "", pilosa.DefaultTxsrc, "transaction/storage to use: one of roaring, rbf, bolt, lmdb, or a blue-green setup: rbf_roaring, roaring_rbf, bolt_roaring, roaring_bolt, bolt_rbf, etc.")
|
||||
// Note: the default for --tx must be kept "" empty string. Otherwise we
|
||||
// cannot detect and honor the PILOSA_TXSRC env var over-ride.
|
||||
flags.StringVarP(&srv.Config.Txsrc, "txsrc", "", "", fmt.Sprintf("transaction/storage to use: one of roaring, rbf, bolt, lmdb, or a blue-green setup: rbf_roaring, roaring_rbf, bolt_roaring, roaring_bolt, bolt_rbf, etc. The default is: %v. The env var PILOSA_TXSRC is over-ridden by --tx option on the command line.", pilosa.DefaultTxsrc))
|
||||
|
||||
// RowcacheOff
|
||||
flags.BoolVarP((&srv.Config.RowcacheOff), "rowcache-off", "", srv.Config.RowcacheOff, "turn off the rowcache for all backends (reduces memory use)")
|
||||
|
|
|
|||
|
|
@ -35,3 +35,14 @@ func TestBuildServerFlags(t *testing.T) {
|
|||
t.Fatal("log-path flag is required")
|
||||
}
|
||||
}
|
||||
|
||||
func TestServerDefaultTxsrcFlags(t *testing.T) {
|
||||
cm := &cobra.Command{}
|
||||
buf := bytes.Buffer{}
|
||||
stdin, stdout, stderr := GetIO(buf)
|
||||
Server := server.NewCommand(stdin, stdout, stderr)
|
||||
BuildServerFlags(cm, Server)
|
||||
if cm.Flags().Lookup("txsrc").DefValue != "" {
|
||||
t.Fatal("cannot set the txsrc default in ctl/server.go, otherwise we won't know to let the environment override the lack of --txsrc on the command line. We want explicit command line --txsrc to override the env value.")
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1450,6 +1450,7 @@ func executeDistinctShardSet(ctx context.Context, qcx *Qcx, idx *Index, fieldNam
|
|||
if err != nil {
|
||||
return SignedRow{}, errors.Wrap(err, "getting fragment data")
|
||||
}
|
||||
defer fragData.Close()
|
||||
// We can't grab the containers "for each row" from the set-type field,
|
||||
// because we don't know how many rows there are, and some of them
|
||||
// might be empty, so really, we're going to iterate through the
|
||||
|
|
|
|||
1
go.mod
1
go.mod
|
|
@ -14,7 +14,6 @@ require (
|
|||
github.com/dustin/go-humanize v1.0.0
|
||||
github.com/glycerine/idem v0.0.0-20190127113923-7a8083893311
|
||||
github.com/glycerine/lmdb-go v1.9.34
|
||||
github.com/glycerine/rbtree v0.0.0-20190406191118-ceb71889d809
|
||||
github.com/go-ole/go-ole v1.2.4 // indirect
|
||||
github.com/gogo/protobuf v1.2.1
|
||||
github.com/golang/protobuf v1.3.3
|
||||
|
|
|
|||
2
go.sum
2
go.sum
|
|
@ -56,8 +56,6 @@ github.com/glycerine/idem v0.0.0-20190127113923-7a8083893311 h1:AAXH0ZvYIHHqU06A
|
|||
github.com/glycerine/idem v0.0.0-20190127113923-7a8083893311/go.mod h1:B72P/ZM99sNiCmaQJflpmMAF5LsDzStpLdWzn0+Vr2Y=
|
||||
github.com/glycerine/lmdb-go v1.9.34 h1:0lymJjpdelYnIMcNzsKROfIaApt99zhaHtjDJTHjGkE=
|
||||
github.com/glycerine/lmdb-go v1.9.34/go.mod h1:DrPeeTGooMg6B7cjNSP14perptTJzzdBy5YoosthrRs=
|
||||
github.com/glycerine/rbtree v0.0.0-20190406191118-ceb71889d809 h1:wBr8MeUUS+Xi4oweFspffWBlDw8s1rGmRBwM4fUjxrc=
|
||||
github.com/glycerine/rbtree v0.0.0-20190406191118-ceb71889d809/go.mod h1:tf1G9WLJXoNEQ5TWYvCSkqsOepuCNCJebECwJ/B/64I=
|
||||
github.com/go-kit/kit v0.8.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2as=
|
||||
github.com/go-kit/kit v0.9.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2as=
|
||||
github.com/go-logfmt/logfmt v0.3.0/go.mod h1:Qt1PoO58o5twSAckw1HlFXLmHsOX5/0LbT9GBnD5lWE=
|
||||
|
|
|
|||
|
|
@ -44,6 +44,10 @@ type Config struct {
|
|||
|
||||
// for mmap correctness testing.
|
||||
DoAllocZero bool
|
||||
|
||||
// CursorCacheSize is the number of copies of Cursor{} to keep in our
|
||||
// readyCursorCh arena to avoid GC pressure.
|
||||
CursorCacheSize int64
|
||||
}
|
||||
|
||||
func NewDefaultConfig() *Config {
|
||||
|
|
@ -53,6 +57,10 @@ func NewDefaultConfig() *Config {
|
|||
MinWALCheckpointSize: DefaultMinWALCheckpointSize,
|
||||
MaxWALCheckpointSize: DefaultMaxWALCheckpointSize,
|
||||
FsyncEnabled: true,
|
||||
|
||||
// CI passed with 20. 50 was too big for CI, even on X-large instances.
|
||||
// For now we default to 0, which means use sync.Pool.
|
||||
CursorCacheSize: 0,
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -65,4 +73,6 @@ func (cfg *Config) DefineFlags(flags *pflag.FlagSet) {
|
|||
|
||||
// renamed from --rbf-fsync to just --fsync because now it applies to all Tx backends.
|
||||
flags.BoolVar(&cfg.FsyncEnabled, "fsync", default0.FsyncEnabled, "enable fsync fully safe flush-to-disk")
|
||||
flags.Int64Var(&cfg.CursorCacheSize, "rbf-cursor-cache", default0.CursorCacheSize, "how big a Cursor arena to maintain. 0 means use sync.Pool with dynamic sizing. Note that <= 20 is needed to pass CI. Controls the memory footprint of rbf.")
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1275,3 +1275,21 @@ func (c *Cursor) difference(key uint64, data *roaring.Container) (bool, error) {
|
|||
|
||||
return false, nil
|
||||
}
|
||||
|
||||
func (c *Cursor) Close() {
|
||||
if c == nil {
|
||||
panic("cannot Close nil Cursor")
|
||||
}
|
||||
tx := c.tx
|
||||
c.tx = nil // allow tx to be garbage collected.
|
||||
|
||||
if tx.db.cfg.CursorCacheSize == 0 {
|
||||
globalCursorSyncPool.Put(c)
|
||||
return
|
||||
}
|
||||
|
||||
select {
|
||||
case tx.db.cursorArenaCh <- c:
|
||||
case <-tx.db.cursorCleaner.ReqStop.Chan:
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1024,8 +1024,8 @@ func TestCursor_PlayContainer(t *testing.T) {
|
|||
}
|
||||
cur, _ := tx.Cursor("x")
|
||||
offset := uint64(0)
|
||||
many(&cur, 0, rbf.ArrayMaxSize+offset)
|
||||
many(&cur, 65536, rbf.ArrayMaxSize+offset)
|
||||
many(cur, 0, rbf.ArrayMaxSize+offset)
|
||||
many(cur, 65536, rbf.ArrayMaxSize+offset)
|
||||
/*
|
||||
many(cur, 2*65536, rbf.ArrayMaxSize+offset)
|
||||
many(cur, 3*65536, rbf.ArrayMaxSize) //+offset)
|
||||
|
|
|
|||
|
|
@ -71,9 +71,15 @@ func (c *Cursor) Rows() ([]uint64, error) {
|
|||
}
|
||||
return rows, err
|
||||
}
|
||||
|
||||
func (tx *Tx) FieldViews() []string {
|
||||
rr, _ := tx.RootRecords()
|
||||
return rr.sliceOfNames()
|
||||
records, _ := tx.RootRecords()
|
||||
a := make([]string, 0, records.Len())
|
||||
for itr := records.Iterator(); !itr.Done(); {
|
||||
name, _ := itr.Next()
|
||||
a = append(a, name.(string))
|
||||
}
|
||||
return a
|
||||
}
|
||||
|
||||
func (c *Cursor) DumpKeys() {
|
||||
|
|
|
|||
67
rbf/db.go
67
rbf/db.go
|
|
@ -24,26 +24,34 @@ import (
|
|||
"syscall"
|
||||
|
||||
"github.com/benbjohnson/immutable"
|
||||
"github.com/pilosa/pilosa/v2/syswrap"
|
||||
|
||||
"github.com/glycerine/idem"
|
||||
rbfcfg "github.com/pilosa/pilosa/v2/rbf/cfg"
|
||||
"github.com/pilosa/pilosa/v2/syswrap"
|
||||
)
|
||||
|
||||
var (
|
||||
ErrClosed = errors.New("rbf: database closed")
|
||||
)
|
||||
|
||||
// global in the sense that it is shared among all instances
|
||||
// of rbf.DBs in this process. This is deliberate.
|
||||
var globalCursorSyncPool = &sync.Pool{
|
||||
New: func() interface{} {
|
||||
return &Cursor{}
|
||||
},
|
||||
}
|
||||
|
||||
// DB options like MaxSize, FsyncEnabled, DoAllocZero
|
||||
// can be set before calling DB.Open().
|
||||
type DB struct {
|
||||
cfg rbfcfg.Config
|
||||
|
||||
data []byte // database mmap
|
||||
file *os.File // database file descriptor
|
||||
rootRecords *rr // cached root records
|
||||
pageMap *immutable.Map // pgno-to-WALID mapping
|
||||
txs map[*Tx]struct{} // active transactions
|
||||
opened bool // true if open
|
||||
data []byte // database mmap
|
||||
file *os.File // database file descriptor
|
||||
rootRecords *immutable.SortedMap // cached root records
|
||||
pageMap *immutable.Map // pgno-to-WALID mapping
|
||||
txs map[*Tx]struct{} // active transactions
|
||||
opened bool // true if open
|
||||
|
||||
wal []byte // wal mmap
|
||||
walFile *os.File // wal file descriptor
|
||||
|
|
@ -55,6 +63,9 @@ type DB struct {
|
|||
|
||||
// Path represents the path to the database file.
|
||||
Path string
|
||||
|
||||
cursorArenaCh chan *Cursor
|
||||
cursorCleaner *idem.Halter
|
||||
}
|
||||
|
||||
// NewDB returns a new instance of DB.
|
||||
|
|
@ -68,8 +79,15 @@ func NewDB(path string, cfg *rbfcfg.Config) *DB {
|
|||
txs: make(map[*Tx]struct{}),
|
||||
pageMap: immutable.NewMap(&uint32Hasher{}),
|
||||
Path: path,
|
||||
|
||||
cursorArenaCh: make(chan *Cursor, cfg.CursorCacheSize),
|
||||
cursorCleaner: idem.NewHalter(),
|
||||
}
|
||||
for i := int64(0); i < cfg.CursorCacheSize; i++ {
|
||||
db.cursorArenaCh <- &Cursor{}
|
||||
}
|
||||
db.haltCond = sync.NewCond(&db.mu)
|
||||
|
||||
return db
|
||||
}
|
||||
|
||||
|
|
@ -254,6 +272,8 @@ func (db *DB) Close() (err error) {
|
|||
db.mu.Lock()
|
||||
defer db.mu.Unlock()
|
||||
|
||||
defer db.cursorCleaner.RequestStop()
|
||||
|
||||
db.opened = false
|
||||
|
||||
// Close mmap handle.
|
||||
|
|
@ -330,13 +350,15 @@ func (db *DB) HasData(requireOneHotBit bool) (hasAnyRecords bool, err error) {
|
|||
// Loop over each bitmap and attempt to move to the first cell.
|
||||
// If we can move to a cell then we have at least one record.
|
||||
|
||||
for it := records.tree.Min(); it != records.tree.Limit(); it = it.Next() {
|
||||
record := it.Item().(RootRecord)
|
||||
for itr := records.Iterator(); !itr.Done(); {
|
||||
name, _ := itr.Next()
|
||||
// Fetch cursor for bitmap.
|
||||
cur, err := tx.Cursor(record.Name)
|
||||
cur, err := tx.Cursor(name.(string))
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
defer cur.Close()
|
||||
|
||||
if !requireOneHotBit {
|
||||
return true, nil
|
||||
}
|
||||
|
|
@ -497,9 +519,6 @@ func (db *DB) removeTx(tx *Tx) error {
|
|||
tx.db.rwmu.Unlock()
|
||||
}
|
||||
|
||||
db.mu.Lock()
|
||||
defer db.mu.Unlock()
|
||||
|
||||
delete(tx.db.txs, tx)
|
||||
|
||||
// Disassociate from db.
|
||||
|
|
@ -560,3 +579,23 @@ func (db *DB) readMetaPage() ([]byte, error) {
|
|||
}
|
||||
return db.readDBPage(0)
|
||||
}
|
||||
|
||||
func (db *DB) getCursor(tx *Tx) (c *Cursor) {
|
||||
if db.cfg.CursorCacheSize == 0 {
|
||||
c = globalCursorSyncPool.Get().(*Cursor)
|
||||
c.tx = tx
|
||||
return
|
||||
}
|
||||
|
||||
n := len(db.cursorArenaCh)
|
||||
if n < 10 {
|
||||
vv("warning, db.cursorArenaCh is low! %v left", n)
|
||||
}
|
||||
select {
|
||||
case c = <-db.cursorArenaCh:
|
||||
c.tx = tx
|
||||
return
|
||||
case <-db.cursorCleaner.ReqStop.Chan:
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -129,6 +129,7 @@ func TestIngest_lots_of_views(t *testing.T) {
|
|||
} else if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
defer c.Close()
|
||||
c.Dump("one.bitmap.dot.dump")
|
||||
}
|
||||
}
|
||||
|
|
|
|||
17
rbf/rbf.go
17
rbf/rbf.go
|
|
@ -27,7 +27,7 @@ import (
|
|||
"time"
|
||||
"unsafe"
|
||||
|
||||
"github.com/glycerine/rbtree"
|
||||
"github.com/benbjohnson/immutable"
|
||||
"github.com/pilosa/pilosa/v2/roaring"
|
||||
"github.com/pilosa/pilosa/v2/shardwidth"
|
||||
)
|
||||
|
|
@ -177,18 +177,19 @@ func readRootRecords(page []byte) (records []*RootRecord, err error) {
|
|||
// We can return io.ErrShortBuffer in err. If we still have records
|
||||
// to write that don't fit on page, remain will point to the next
|
||||
// record that hasn't yet been written.
|
||||
func writeRootRecords(page []byte, recit, limit rbtree.Iterator) (remain rbtree.Iterator, err error) {
|
||||
func writeRootRecords(page []byte, itr *immutable.SortedMapIterator) (err error) {
|
||||
data := page[rootRecordPageHeaderSize:]
|
||||
|
||||
for recit != limit {
|
||||
rec := recit.Item().(RootRecord)
|
||||
data, err = WriteRootRecord(data, &rec)
|
||||
for !itr.Done() {
|
||||
name, pgno := itr.Next()
|
||||
|
||||
data, err = WriteRootRecord(data, &RootRecord{Name: name.(string), Pgno: pgno.(uint32)})
|
||||
if err != nil {
|
||||
return recit, err
|
||||
itr.Seek(name)
|
||||
return err
|
||||
}
|
||||
recit = recit.Next()
|
||||
}
|
||||
return recit, nil
|
||||
return nil
|
||||
}
|
||||
|
||||
// Branch & leaf page helpers
|
||||
|
|
|
|||
287
rbf/tx.go
287
rbf/tx.go
|
|
@ -23,7 +23,6 @@ import (
|
|||
"sync"
|
||||
|
||||
"github.com/benbjohnson/immutable"
|
||||
"github.com/glycerine/rbtree"
|
||||
"github.com/pilosa/pilosa/v2/hash"
|
||||
"github.com/pilosa/pilosa/v2/roaring"
|
||||
"github.com/pilosa/pilosa/v2/txkey"
|
||||
|
|
@ -34,11 +33,11 @@ var _ = txkey.ToString
|
|||
// Tx represents a transaction.
|
||||
type Tx struct {
|
||||
mu sync.RWMutex
|
||||
db *DB // parent db
|
||||
meta [PageSize]byte // copy of current meta page
|
||||
walID int64 // max WAL ID at start of tx
|
||||
walPageN int // wal page count
|
||||
rootRecords *rr // read-only cache of root records
|
||||
db *DB // parent db
|
||||
meta [PageSize]byte // copy of current meta page
|
||||
walID int64 // max WAL ID at start of tx
|
||||
walPageN int // wal page count
|
||||
rootRecords *immutable.SortedMap // read-only cache of root records
|
||||
|
||||
// pageMap holds WAL pages that have not yet been transferred
|
||||
// into the database pages. So it can be empty, if the whole previous
|
||||
|
|
@ -59,70 +58,6 @@ type Tx struct {
|
|||
DeleteEmptyContainer bool
|
||||
}
|
||||
|
||||
type rr struct {
|
||||
tree rbtree.Tree
|
||||
}
|
||||
|
||||
func newRR() *rr {
|
||||
return &rr{
|
||||
tree: *rbtree.NewTree(
|
||||
func(a, b rbtree.Item) int {
|
||||
an := a.(RootRecord).Name
|
||||
bn := b.(RootRecord).Name
|
||||
if an == bn {
|
||||
return 0
|
||||
}
|
||||
if an < bn {
|
||||
return -1
|
||||
}
|
||||
return 1
|
||||
}),
|
||||
}
|
||||
}
|
||||
|
||||
func (s *rr) size() int {
|
||||
return s.tree.Len()
|
||||
}
|
||||
|
||||
func (s *rr) add(r RootRecord) {
|
||||
s.tree.Insert(r)
|
||||
}
|
||||
|
||||
func (s *rr) addAll(recs []*RootRecord) {
|
||||
for _, r := range recs {
|
||||
s.add(*r)
|
||||
}
|
||||
}
|
||||
|
||||
func (s *rr) remove(it rbtree.Iterator) {
|
||||
s.tree.DeleteWithIterator(it)
|
||||
}
|
||||
|
||||
func iterToRootRecord(it rbtree.Iterator) RootRecord {
|
||||
return it.Item().(RootRecord)
|
||||
}
|
||||
|
||||
func (s *rr) sliceOfNames() (res []string) {
|
||||
res = make([]string, s.size())
|
||||
|
||||
i := 0
|
||||
for it := s.tree.Min(); it != s.tree.Limit(); it = it.Next() {
|
||||
res[i] = it.Item().(RootRecord).Name
|
||||
i++
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
func (s *rr) find(name string) (r RootRecord, iter rbtree.Iterator, exact bool) {
|
||||
iter = s.tree.FindGE(RootRecord{Name: name})
|
||||
if iter.Limit() {
|
||||
return
|
||||
}
|
||||
r = iter.Item().(RootRecord)
|
||||
exact = (r.Name == name)
|
||||
return
|
||||
}
|
||||
|
||||
func (tx *Tx) DBPath() string {
|
||||
return tx.db.Path
|
||||
}
|
||||
|
|
@ -169,13 +104,16 @@ func (tx *Tx) Commit() error {
|
|||
// avoid race detector firing on a write race here
|
||||
// vs the read of rootRecords at db.Begin()
|
||||
tx.db.mu.Lock()
|
||||
defer tx.db.mu.Unlock()
|
||||
tx.db.rootRecords = tx.rootRecords
|
||||
tx.db.pageMap = tx.pageMap
|
||||
tx.db.walPageN = tx.walPageN
|
||||
tx.db.mu.Unlock()
|
||||
return tx.db.removeTx(tx)
|
||||
}
|
||||
|
||||
// Disconnect transaction from DB.
|
||||
tx.db.mu.Lock()
|
||||
defer tx.db.mu.Unlock()
|
||||
return tx.db.removeTx(tx)
|
||||
}
|
||||
|
||||
|
|
@ -194,6 +132,8 @@ func (tx *Tx) Rollback() {
|
|||
}
|
||||
|
||||
// Disconnect transaction from DB.
|
||||
tx.db.mu.Lock()
|
||||
defer tx.db.mu.Unlock()
|
||||
panicOn(tx.db.removeTx(tx))
|
||||
}
|
||||
|
||||
|
|
@ -210,11 +150,11 @@ func (tx *Tx) root(name string) (uint32, error) {
|
|||
return 0, err
|
||||
}
|
||||
|
||||
_, it, exactHit := records.find(name)
|
||||
if !exactHit {
|
||||
pgno, ok := records.Get(name)
|
||||
if !ok {
|
||||
return 0, ErrBitmapNotFound
|
||||
}
|
||||
return iterToRootRecord(it).Pgno, nil
|
||||
return pgno.(uint32), nil
|
||||
}
|
||||
|
||||
// BitmapNames returns a list of all bitmap names.
|
||||
|
|
@ -231,7 +171,13 @@ func (tx *Tx) BitmapNames() ([]string, error) {
|
|||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return records.sliceOfNames(), nil
|
||||
|
||||
a := make([]string, 0, records.Len())
|
||||
for itr := records.Iterator(); !itr.Done(); {
|
||||
k, _ := itr.Next()
|
||||
a = append(a, k.(string))
|
||||
}
|
||||
return a, nil
|
||||
}
|
||||
|
||||
// CreateBitmap creates a new empty bitmap with the given name.
|
||||
|
|
@ -258,15 +204,12 @@ func (tx *Tx) createBitmap(name string) error {
|
|||
}
|
||||
|
||||
// Find btree by name. Exit if already exists.
|
||||
_, _, exact := records.find(name)
|
||||
if exact {
|
||||
if _, ok := records.Get(name); ok {
|
||||
return ErrBitmapExists
|
||||
}
|
||||
//fmt.Println("CREATE BITMAP", name, index)
|
||||
|
||||
// Allocate new root page.
|
||||
pgno, err := tx.allocatePgno()
|
||||
//fmt.Println("CREATE BITMAP @ PGNO", pgno)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
|
@ -281,7 +224,7 @@ func (tx *Tx) createBitmap(name string) error {
|
|||
}
|
||||
|
||||
// Insert into correct index.
|
||||
records.add(RootRecord{Name: name, Pgno: pgno})
|
||||
records = records.Set(name, pgno)
|
||||
if err := tx.writeRootRecordPages(records); err != nil {
|
||||
return fmt.Errorf("write bitmaps: %w", err)
|
||||
}
|
||||
|
|
@ -305,15 +248,6 @@ func (tx *Tx) createBitmapIfNotExists(name string) error {
|
|||
return nil
|
||||
}
|
||||
|
||||
/*
|
||||
func dump(r []*RootRecord) {
|
||||
for _, i := range r {
|
||||
fmt.Println("RECORD", i.Name, i.Pgno)
|
||||
}
|
||||
|
||||
}
|
||||
*/
|
||||
|
||||
// DeleteBitmap removes a bitmap with the given name.
|
||||
// Returns an error if the bitmap does not exist.
|
||||
func (tx *Tx) DeleteBitmap(name string) error {
|
||||
|
|
@ -335,21 +269,18 @@ func (tx *Tx) DeleteBitmap(name string) error {
|
|||
}
|
||||
|
||||
// Find btree by name. Exit if it doesn't exist.
|
||||
record, it, ok := records.find(name)
|
||||
|
||||
pgno, ok := records.Get(name)
|
||||
if !ok {
|
||||
return fmt.Errorf("bitmap does not exist: %q", name)
|
||||
}
|
||||
pgno := record.Pgno
|
||||
|
||||
// Deallocate all pages in the tree.
|
||||
if err := tx.deallocateTree(pgno); err != nil {
|
||||
if err := tx.deallocateTree(pgno.(uint32)); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
records.remove(it)
|
||||
|
||||
// Delete from record list & rewrite record pages.
|
||||
records = records.Delete(name)
|
||||
if err := tx.writeRootRecordPages(records); err != nil {
|
||||
return fmt.Errorf("write bitmaps: %w", err)
|
||||
}
|
||||
|
|
@ -373,24 +304,21 @@ func (tx *Tx) DeleteBitmapsWithPrefix(prefix string) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for it := records.tree.Min(); it != records.tree.Limit(); {
|
||||
record := it.Item().(RootRecord)
|
||||
|
||||
for itr := records.Iterator(); !itr.Done(); {
|
||||
name, pgno := itr.Next()
|
||||
|
||||
// Skip bitmaps without matching prefix.
|
||||
if !strings.HasPrefix(record.Name, prefix) {
|
||||
if !strings.HasPrefix(name.(string), prefix) {
|
||||
continue
|
||||
}
|
||||
|
||||
// Deallocate all pages in the tree.
|
||||
if err := tx.deallocateTree(record.Pgno); err != nil {
|
||||
if err := tx.deallocateTree(pgno.(uint32)); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// as long we've advanced it past delme, we can
|
||||
// delete delme without affecting it.
|
||||
delme := it
|
||||
it = it.Next()
|
||||
records.remove(delme)
|
||||
records = records.Delete(name.(string))
|
||||
}
|
||||
|
||||
// Rewrite record pages.
|
||||
|
|
@ -422,16 +350,14 @@ func (tx *Tx) RenameBitmap(oldname, newname string) error {
|
|||
}
|
||||
|
||||
// Find btree by name. Exit if it doesn't exist.
|
||||
rec, it, exactHit := records.find(oldname)
|
||||
if !exactHit {
|
||||
pgno, ok := records.Get(oldname)
|
||||
if !ok {
|
||||
return fmt.Errorf("bitmap does not exist: %q", oldname)
|
||||
}
|
||||
|
||||
// Update record name & rewrite record pages.
|
||||
rec2 := rec
|
||||
rec2.Name = newname
|
||||
records.remove(it)
|
||||
records.add(rec2)
|
||||
records = records.Delete(oldname)
|
||||
records = records.Set(newname, pgno)
|
||||
if err := tx.writeRootRecordPages(records); err != nil {
|
||||
return fmt.Errorf("write bitmaps: %w", err)
|
||||
}
|
||||
|
|
@ -440,12 +366,12 @@ func (tx *Tx) RenameBitmap(oldname, newname string) error {
|
|||
}
|
||||
|
||||
// RootRecords returns a list of root records.
|
||||
func (tx *Tx) RootRecords() (records *rr, err error) {
|
||||
func (tx *Tx) RootRecords() (records *immutable.SortedMap, err error) {
|
||||
if tx.rootRecords != nil {
|
||||
return tx.rootRecords, nil
|
||||
}
|
||||
|
||||
records = newRR()
|
||||
records = immutable.NewSortedMap(nil)
|
||||
for pgno := readMetaRootRecordPageNo(tx.meta[:]); pgno != 0; {
|
||||
page, err := tx.readPage(pgno)
|
||||
if err != nil {
|
||||
|
|
@ -457,7 +383,9 @@ func (tx *Tx) RootRecords() (records *rr, err error) {
|
|||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
records.addAll(a)
|
||||
for _, rec := range a {
|
||||
records = records.Set(rec.Name, rec.Pgno)
|
||||
}
|
||||
|
||||
// Read next overflow page number.
|
||||
pgno = WalkRootRecordPages(page)
|
||||
|
|
@ -469,7 +397,7 @@ func (tx *Tx) RootRecords() (records *rr, err error) {
|
|||
}
|
||||
|
||||
// writeRootRecordPages writes a list of root record pages.
|
||||
func (tx *Tx) writeRootRecordPages(records *rr) (err error) {
|
||||
func (tx *Tx) writeRootRecordPages(records *immutable.SortedMap) (err error) {
|
||||
|
||||
// Release all existing root record pages.
|
||||
for pgno := readMetaRootRecordPageNo(tx.meta[:]); pgno != 0; {
|
||||
|
|
@ -486,7 +414,7 @@ func (tx *Tx) writeRootRecordPages(records *rr) (err error) {
|
|||
}
|
||||
|
||||
// Exit early if no records exist.
|
||||
if records.size() == 0 {
|
||||
if records.Len() == 0 {
|
||||
writeMetaRootRecordPageNo(tx.meta[:], 0)
|
||||
return nil
|
||||
}
|
||||
|
|
@ -499,30 +427,19 @@ func (tx *Tx) writeRootRecordPages(records *rr) (err error) {
|
|||
writeMetaRootRecordPageNo(tx.meta[:], pgno)
|
||||
|
||||
// Write new root record pages.
|
||||
limit := records.tree.Limit()
|
||||
it := records.tree.Min()
|
||||
for it != limit {
|
||||
|
||||
for itr := records.Iterator(); !itr.Done(); {
|
||||
// Initialize page & write as many records as will fit.
|
||||
page := make([]byte, PageSize)
|
||||
writePageNo(page, pgno)
|
||||
writeFlags(page, PageTypeRootRecord)
|
||||
|
||||
// writeRootRecords does it = it.Next() for us after
|
||||
// each successful write to the page.
|
||||
it, err = writeRootRecords(page, it, limit)
|
||||
|
||||
switch err {
|
||||
case nil:
|
||||
// nothing to do, all the rest of the records fit on the page.
|
||||
|
||||
case io.ErrShortBuffer:
|
||||
if err := writeRootRecords(page, itr); err == io.ErrShortBuffer {
|
||||
// Allocate next pgno and write overflow if we have remaining records.
|
||||
if pgno, err = tx.allocatePgno(); err != nil {
|
||||
return err
|
||||
}
|
||||
writeRootRecordOverflowPgno(page, pgno)
|
||||
default:
|
||||
} else if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
|
|
@ -559,6 +476,8 @@ func (tx *Tx) Add(name string, a ...uint64) (changeCount int, err error) {
|
|||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
defer c.Close()
|
||||
|
||||
for _, v := range a {
|
||||
if vchanged, err := c.Add(v); err != nil {
|
||||
return changeCount, err
|
||||
|
|
@ -588,6 +507,8 @@ func (tx *Tx) Remove(name string, a ...uint64) (changeCount int, err error) {
|
|||
} else if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
defer c.Close()
|
||||
|
||||
for _, v := range a {
|
||||
if vchanged, err := c.Remove(v); err != nil {
|
||||
return changeCount, err
|
||||
|
|
@ -615,29 +536,31 @@ func (tx *Tx) Contains(name string, v uint64) (bool, error) {
|
|||
} else if err != nil {
|
||||
return false, err
|
||||
}
|
||||
defer c.Close()
|
||||
|
||||
return c.Contains(v)
|
||||
}
|
||||
|
||||
// Cursor returns an instance of a cursor this bitmap.
|
||||
func (tx *Tx) Cursor(name string) (Cursor, error) {
|
||||
func (tx *Tx) Cursor(name string) (*Cursor, error) {
|
||||
tx.mu.RLock()
|
||||
defer tx.mu.RUnlock()
|
||||
return tx.cursor(name)
|
||||
}
|
||||
|
||||
func (tx *Tx) cursor(name string) (Cursor, error) {
|
||||
func (tx *Tx) cursor(name string) (*Cursor, error) {
|
||||
if tx.db == nil {
|
||||
return Cursor{}, ErrTxClosed
|
||||
return nil, ErrTxClosed
|
||||
} else if name == "" {
|
||||
return Cursor{}, ErrBitmapNameRequired
|
||||
return nil, ErrBitmapNameRequired
|
||||
}
|
||||
|
||||
root, err := tx.root(name)
|
||||
if err != nil {
|
||||
return Cursor{}, err
|
||||
return nil, err
|
||||
}
|
||||
|
||||
c := Cursor{tx: tx}
|
||||
c := tx.db.getCursor(tx)
|
||||
c.stack.elems[0] = stackElem{pgno: root}
|
||||
return c, nil
|
||||
}
|
||||
|
|
@ -659,6 +582,7 @@ func (tx *Tx) RoaringBitmap(name string) (*roaring.Bitmap, error) {
|
|||
} else if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer c.Close()
|
||||
|
||||
other := roaring.NewSliceBitmap()
|
||||
if err := c.First(); err == io.EOF {
|
||||
|
|
@ -698,9 +622,13 @@ func (tx *Tx) container(name string, key uint64) (*roaring.Container, error) {
|
|||
return nil, nil
|
||||
} else if err != nil {
|
||||
return nil, err
|
||||
} else if exact, err := c.Seek(key); err != nil || !exact {
|
||||
}
|
||||
defer c.Close()
|
||||
|
||||
if exact, err := c.Seek(key); err != nil || !exact {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return toContainer(c.cell(), tx), nil
|
||||
}
|
||||
|
||||
|
|
@ -725,9 +653,13 @@ func (tx *Tx) putContainer(name string, key uint64, ct *roaring.Container) error
|
|||
c, err := tx.cursor(name)
|
||||
if err != nil {
|
||||
return err
|
||||
} else if _, err := c.Seek(cell.Key); err != nil {
|
||||
}
|
||||
defer c.Close()
|
||||
|
||||
if _, err := c.Seek(cell.Key); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return c.putLeafCell(cell)
|
||||
}
|
||||
|
||||
|
|
@ -754,9 +686,13 @@ func (tx *Tx) removeContainer(name string, key uint64) error {
|
|||
return nil
|
||||
} else if err != nil {
|
||||
return err
|
||||
} else if exact, err := c.Seek(key); err != nil || !exact {
|
||||
}
|
||||
defer c.Close()
|
||||
|
||||
if exact, err := c.Seek(key); err != nil || !exact {
|
||||
return err
|
||||
}
|
||||
|
||||
return c.deleteLeafCell(key)
|
||||
}
|
||||
|
||||
|
|
@ -869,10 +805,10 @@ func (tx *Tx) inusePageSet() (map[uint32]struct{}, error) {
|
|||
return m, err
|
||||
}
|
||||
|
||||
for it := records.tree.Min(); it != records.tree.Limit(); it = it.Next() {
|
||||
record := it.Item().(RootRecord)
|
||||
for itr := records.Iterator(); !itr.Done(); {
|
||||
_, pgno := itr.Next()
|
||||
|
||||
if err := tx.walkTree(record.Pgno, 0, func(pgno, parent, typ uint32) error {
|
||||
if err := tx.walkTree(pgno.(uint32), 0, func(pgno, parent, typ uint32) error {
|
||||
m[pgno] = struct{}{}
|
||||
return nil
|
||||
}); err != nil {
|
||||
|
|
@ -1059,6 +995,8 @@ func (tx *Tx) AddRoaring(name string, bm *roaring.Bitmap) (changed bool, err err
|
|||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
defer c.Close()
|
||||
|
||||
return c.AddRoaring(bm)
|
||||
}
|
||||
|
||||
|
|
@ -1101,7 +1039,10 @@ func (tx *Tx) ForEachRange(name string, start, end uint64, fn func(uint64) error
|
|||
return nil
|
||||
} else if err != nil {
|
||||
return err
|
||||
} else if _, err := c.Seek(highbits(start)); err != nil {
|
||||
}
|
||||
defer c.Close()
|
||||
|
||||
if _, err := c.Seek(highbits(start)); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
|
|
@ -1190,7 +1131,10 @@ func (tx *Tx) Count(name string) (uint64, error) {
|
|||
return 0, nil
|
||||
} else if err != nil {
|
||||
return 0, err
|
||||
} else if err := c.First(); err != nil {
|
||||
}
|
||||
defer c.Close()
|
||||
|
||||
if err := c.First(); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
|
|
@ -1216,7 +1160,10 @@ func (tx *Tx) Max(name string) (uint64, error) {
|
|||
return 0, nil
|
||||
} else if err != nil {
|
||||
return 0, err
|
||||
} else if err := c.Last(); err == io.EOF {
|
||||
}
|
||||
defer c.Close()
|
||||
|
||||
if err := c.Last(); err == io.EOF {
|
||||
return 0, nil
|
||||
} else if err != nil {
|
||||
return 0, err
|
||||
|
|
@ -1235,7 +1182,10 @@ func (tx *Tx) Min(name string) (uint64, bool, error) {
|
|||
return 0, false, nil
|
||||
} else if err != nil {
|
||||
return 0, false, err
|
||||
} else if err := c.First(); err == io.EOF {
|
||||
}
|
||||
defer c.Close()
|
||||
|
||||
if err := c.First(); err == io.EOF {
|
||||
return 0, false, nil
|
||||
} else if err != nil {
|
||||
return 0, false, err
|
||||
|
|
@ -1284,6 +1234,7 @@ func (tx *Tx) CountRange(name string, start, end uint64) (uint64, error) {
|
|||
} else if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
defer csr.Close()
|
||||
|
||||
exact, err := csr.Seek(skey)
|
||||
_ = exact
|
||||
|
|
@ -1354,6 +1305,7 @@ func (tx *Tx) OffsetRange(name string, offset, start, endx uint64) (*roaring.Bit
|
|||
} else if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer c.Close()
|
||||
|
||||
other := roaring.NewSliceBitmap()
|
||||
off := highbits(offset)
|
||||
|
|
@ -1386,11 +1338,15 @@ func (tx *Tx) OffsetRange(name string, offset, start, endx uint64) (*roaring.Bit
|
|||
|
||||
// containerIterator wraps Cursor to implement roaring.ContainerIterator.
|
||||
type containerIterator struct {
|
||||
cursor Cursor
|
||||
cursor *Cursor
|
||||
}
|
||||
|
||||
// Close is a no-op. It exists to implement the roaring.ContainerIterator interface.
|
||||
func (itr *containerIterator) Close() {}
|
||||
// Close must be called when the client is done
|
||||
// with the containerIterator so that the internal
|
||||
// Cursor can be recycled.
|
||||
func (itr *containerIterator) Close() {
|
||||
itr.cursor.Close()
|
||||
}
|
||||
|
||||
// Next moves the iterator to the next container.
|
||||
func (itr *containerIterator) Next() bool {
|
||||
|
|
@ -1428,11 +1384,13 @@ func (tx *Tx) DumpString(short bool, shard uint64) (r string) {
|
|||
panicOn(err)
|
||||
n := 0
|
||||
|
||||
for it := records.tree.Min(); it != records.tree.Limit(); it = it.Next() {
|
||||
rr := it.Item().(RootRecord)
|
||||
for itr := records.Iterator(); !itr.Done(); {
|
||||
name, _ := itr.Next()
|
||||
|
||||
c, err := tx.cursor(rr.Name)
|
||||
c, err := tx.cursor(name.(string))
|
||||
panicOn(err)
|
||||
defer c.Close()
|
||||
|
||||
err = c.First() // First will rewind to beginning.
|
||||
if err == io.EOF {
|
||||
r += "<empty bitmap>"
|
||||
|
|
@ -1450,7 +1408,7 @@ func (tx *Tx) DumpString(short bool, shard uint64) (r string) {
|
|||
ckey := cell.Key
|
||||
ct := toContainer(cell, tx)
|
||||
|
||||
s := stringOfCkeyCt(ckey, ct, rr.Name, short)
|
||||
s := stringOfCkeyCt(ckey, ct, name.(string), short)
|
||||
r += s
|
||||
n++
|
||||
}
|
||||
|
|
@ -1557,6 +1515,7 @@ func (tx *Tx) ImportRoaringBits(name string, itr roaring.RoaringIterator, clear
|
|||
if err != nil {
|
||||
return changed, rowSet, err
|
||||
}
|
||||
defer cur.Close()
|
||||
|
||||
for itrKey, synthC := itr.NextContainer(); synthC != nil; itrKey, synthC = itr.NextContainer() {
|
||||
if rowSize != 0 {
|
||||
|
|
@ -1585,7 +1544,7 @@ func (tx *Tx) ImportRoaringBits(name string, itr roaring.RoaringIterator, clear
|
|||
changed += nsynth
|
||||
rowSet[currRow] += nsynth
|
||||
|
||||
if err := tx.putContainerWithCursor(&cur, itrKey, synthC); err != nil {
|
||||
if err := tx.putContainerWithCursor(cur, itrKey, synthC); err != nil {
|
||||
return changed, rowSet, err
|
||||
}
|
||||
continue
|
||||
|
|
@ -1606,7 +1565,7 @@ func (tx *Tx) ImportRoaringBits(name string, itr roaring.RoaringIterator, clear
|
|||
changes := int(existN - newC.N())
|
||||
changed += changes
|
||||
rowSet[currRow] -= changes
|
||||
err = tx.putContainerWithCursor(&cur, itrKey, newC)
|
||||
err = tx.putContainerWithCursor(cur, itrKey, newC)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
|
@ -1624,7 +1583,7 @@ func (tx *Tx) ImportRoaringBits(name string, itr roaring.RoaringIterator, clear
|
|||
// can nsynth be zero? No, because of the continue/invariant above where nsynth > 0
|
||||
changed += nsynth
|
||||
rowSet[currRow] += nsynth
|
||||
err = tx.putContainerWithCursor(&cur, itrKey, synthC)
|
||||
err = tx.putContainerWithCursor(cur, itrKey, synthC)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
|
@ -1641,7 +1600,7 @@ func (tx *Tx) ImportRoaringBits(name string, itr roaring.RoaringIterator, clear
|
|||
changed += changes
|
||||
rowSet[currRow] += changes
|
||||
|
||||
err = tx.putContainerWithCursor(&cur, itrKey, newC)
|
||||
err = tx.putContainerWithCursor(cur, itrKey, newC)
|
||||
if err != nil {
|
||||
panicOn(err)
|
||||
return
|
||||
|
|
@ -1656,7 +1615,6 @@ func (tx *Tx) ImportRoaringBits(name string, itr roaring.RoaringIterator, clear
|
|||
// flush writes the dirty pages & meta page to the WAL.
|
||||
func (tx *Tx) flush() error {
|
||||
w := bufio.NewWriterSize(tx.db.walFile, 65536)
|
||||
builder := immutable.NewMapBuilder(tx.pageMap)
|
||||
|
||||
// Write non-bitmap pages to WAL.
|
||||
for _, pgno := range dirtyPageMapKeys(tx.dirtyPages) {
|
||||
|
|
@ -1664,7 +1622,7 @@ func (tx *Tx) flush() error {
|
|||
if err != nil {
|
||||
return fmt.Errorf("write page to wal: %w", err)
|
||||
}
|
||||
builder.Set(pgno, walID)
|
||||
tx.pageMap = tx.pageMap.Set(pgno, walID)
|
||||
}
|
||||
|
||||
// Write bitmap headers & pages to WAL.
|
||||
|
|
@ -1682,7 +1640,7 @@ func (tx *Tx) flush() error {
|
|||
if err != nil {
|
||||
return fmt.Errorf("write bitmap page to wal: %w", err)
|
||||
}
|
||||
builder.Set(pgno, walID)
|
||||
tx.pageMap = tx.pageMap.Set(pgno, walID)
|
||||
}
|
||||
|
||||
// Write meta page to WAL.
|
||||
|
|
@ -1690,7 +1648,7 @@ func (tx *Tx) flush() error {
|
|||
if err != nil {
|
||||
return fmt.Errorf("write meta page to wal: %w", err)
|
||||
}
|
||||
builder.Set(uint32(0), walID)
|
||||
tx.pageMap = tx.pageMap.Set(uint32(0), walID)
|
||||
|
||||
// Flush & sync WAL.
|
||||
if err := w.Flush(); err != nil {
|
||||
|
|
@ -1699,9 +1657,6 @@ func (tx *Tx) flush() error {
|
|||
return fmt.Errorf("sync wal: %w", err)
|
||||
}
|
||||
|
||||
// Save page map for new WAL pages.
|
||||
tx.pageMap = builder.Map()
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
|
|
@ -1828,10 +1783,10 @@ func (tx *Tx) PageInfos() ([]PageInfo, error) {
|
|||
return nil, err
|
||||
}
|
||||
|
||||
for it := records.tree.Min(); it != records.tree.Limit(); it = it.Next() {
|
||||
record := it.Item().(RootRecord)
|
||||
for itr := records.Iterator(); !itr.Done(); {
|
||||
name, pgno := itr.Next()
|
||||
|
||||
if err := tx.walkPageInfo(infos, record.Pgno, record.Name); err != nil {
|
||||
if err := tx.walkPageInfo(infos, pgno.(uint32), name.(string)); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -361,11 +361,11 @@ func TestTx_CursorCrashArray(t *testing.T) {
|
|||
}
|
||||
//setArray(t, 0, 2379, &c)
|
||||
//setArray(t, 1, 2337, &c)
|
||||
setArray(t, 32, 1216, &c)
|
||||
setArray(t, 33, 1195, &c)
|
||||
setArray(t, 48, 1186, &c)
|
||||
setArray(t, 49, 1223, &c)
|
||||
setArray(t, 50, 1223, &c)
|
||||
setArray(t, 32, 1216, c)
|
||||
setArray(t, 33, 1195, c)
|
||||
setArray(t, 48, 1186, c)
|
||||
setArray(t, 49, 1223, c)
|
||||
setArray(t, 50, 1223, c)
|
||||
|
||||
}
|
||||
|
||||
|
|
@ -385,8 +385,8 @@ func TestTx_CursorCrashBitmap(t *testing.T) {
|
|||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
setArray(t, 0, 22510, &c)
|
||||
setArray(t, 1, 23584, &c)
|
||||
setArray(t, 0, 22510, c)
|
||||
setArray(t, 1, 23584, c)
|
||||
}
|
||||
|
||||
func setArray(tb testing.TB, key, num int, c *rbf.Cursor) {
|
||||
|
|
@ -488,6 +488,30 @@ func TestTx_Dump(t *testing.T) {
|
|||
}
|
||||
}
|
||||
|
||||
func TestTx_CreateBitmap(t *testing.T) {
|
||||
t.Run("Bulk", func(t *testing.T) {
|
||||
db := MustOpenDB(t)
|
||||
defer MustCloseDB(t, db)
|
||||
|
||||
tx, err := db.Begin(true)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer tx.Rollback()
|
||||
|
||||
if err := tx.CreateBitmap(fmt.Sprintf("%4000x", 0)); err != nil {
|
||||
t.Fatal(err)
|
||||
} else if err := tx.CreateBitmap(fmt.Sprintf("%4000x", 1)); err != nil {
|
||||
t.Fatal(err)
|
||||
} else if err := tx.CreateBitmap(fmt.Sprintf("%4000x", 2)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := tx.Commit(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func rbfName(index, field, view string, shard uint64) string {
|
||||
return string(txkey.Prefix(index, field, view, shard))
|
||||
}
|
||||
|
|
|
|||
|
|
@ -19,6 +19,7 @@ import (
|
|||
"io"
|
||||
"math"
|
||||
"os"
|
||||
"runtime"
|
||||
"sort"
|
||||
"sync"
|
||||
"time"
|
||||
|
|
@ -108,6 +109,11 @@ func (c *callStats) report() (r string) {
|
|||
for i := range lines {
|
||||
r += lines[i].Line
|
||||
}
|
||||
|
||||
var m1 runtime.MemStats
|
||||
runtime.ReadMemStats(&m1)
|
||||
r += fmt.Sprintf("\n m1.TotalAlloc = %v\n", m1.TotalAlloc)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
|
|
|
|||
1
tx.go
1
tx.go
|
|
@ -109,6 +109,7 @@ type Tx interface {
|
|||
// ContainerIterator must not have side-effects. blueGreenTx will
|
||||
// call it at the very beginning of commit to verify db contents.
|
||||
//
|
||||
// citer.Close() must be called when the client is done using it.
|
||||
ContainerIterator(index, field, view string, shard uint64, ckey uint64) (citer roaring.ContainerIterator, found bool, err error)
|
||||
|
||||
// RoaringBitmap retreives the roaring.Bitmap for the entire shard.
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue