From a3d802f8a3ed500df58fc9776cc8245d0b0519eb Mon Sep 17 00:00:00 2001 From: Jason Aten Date: Thu, 30 Jul 2020 14:06:42 -0400 Subject: [PATCH] rbf: OffsetRange, ImportRoaringBits, CountRange work green: TestFragment_RowsIteration/combinations TestFragment_RoaringImportTopN red: (needs Ben's attention) PILOSA_TXSRC=rbf go test -v -run TestFragment_TopN_IDs -tags=' shardwidth20' "-gcflags=all=-d=checkptr=0" also red: (one for Ben) TestCursor_FirstNext_Quick/9 is throwing panic: cannot find segment containing WAL page: 1 as we check the error back from checkpoint() in Rollback(). --- badger.go | 5 + bluegreentx.go | 2 +- executor_test.go | 4 +- fragment_internal_test.go | 3 +- index.go | 17 +- like_test.go | 1 - rbf/rbf.go | 27 +++- rbf/tx.go | 320 +++++++++++++++++++++++++++++++++----- tx.go | 3 +- txfactory.go | 91 ++++++----- 10 files changed, 380 insertions(+), 93 deletions(-) diff --git a/badger.go b/badger.go index f2fca4dc6..ce28de54e 100644 --- a/badger.go +++ b/badger.go @@ -1317,6 +1317,10 @@ func (tx *BadgerTx) UnionInPlace(index, field, view string, shard uint64, others // roaring.countRange counts the number of bits set between [start, end). func (tx *BadgerTx) CountRange(index, field, view string, shard uint64, start, end uint64) (n uint64, err error) { + if start >= end { + return 0, nil + } + skey := highbits(start) ekey := highbits(end) @@ -1415,6 +1419,7 @@ func (tx *BadgerTx) OffsetRange(index, field, view string, shard, offset, start, bkey := item.Key() k := badgerKeyExtractContainerKey(bkey) + // >= hi1 is correct b/c endx cannot have any lowbits set. if uint64(k) >= hi1 { break } diff --git a/bluegreentx.go b/bluegreentx.go index 9df2596a1..3084a2390 100644 --- a/bluegreentx.go +++ b/bluegreentx.go @@ -130,7 +130,6 @@ func (c *blueGreenTx) compareTxState(index, field, view string, shard uint64) { panic(fmt.Sprintf("compareTxState[%v]: B(%v) reported err %v at %v; but A(%v) did not", here, c.bs, bErr, c.as, stack())) } } - for aIter.Next() { aKey, aValue := aIter.Value() @@ -582,6 +581,7 @@ func (c *blueGreenTx) UnionInPlace(index, field, view string, shard uint64, othe func (c *blueGreenTx) CountRange(index, field, view string, shard uint64, start, end uint64) (n uint64, err error) { c.checker.see(index, field, view, shard) + //vv("CountRange start=0x%x, endx=0x%x", start, end) defer func() { if r := recover(); r != nil { c.Dump() diff --git a/executor_test.go b/executor_test.go index dd4927209..6fd5a7b61 100644 --- a/executor_test.go +++ b/executor_test.go @@ -4096,8 +4096,8 @@ func TestExecutor_Execute_SetRow(t *testing.T) { c := test.MustRunCluster(t, 1) defer c.Close() hldr := test.Holder{Holder: c[0].Server.Holder()} - index := hldr.MustCreateIndexIfNotExists("i", pilosa.IndexOptions{TrackExistence: true}) - _, err := index.CreateField("f", pilosa.OptFieldTypeDefault()) + idx := hldr.MustCreateIndexIfNotExists("i", pilosa.IndexOptions{TrackExistence: true}) + _, err := idx.CreateField("f", pilosa.OptFieldTypeDefault()) if err != nil { t.Fatal(err) } diff --git a/fragment_internal_test.go b/fragment_internal_test.go index a8851a1bb..871864d3d 100644 --- a/fragment_internal_test.go +++ b/fragment_internal_test.go @@ -3902,7 +3902,6 @@ func TestFragment_RoaringImport(t *testing.T) { defer tx.Rollback() for num, input := range test { - vv("num=%v, input='%#v'", num, input) buf := &bytes.Buffer{} bm := roaring.NewBitmap(input...) _, err := bm.WriteTo(buf) @@ -3913,7 +3912,6 @@ func TestFragment_RoaringImport(t *testing.T) { if err != nil { t.Fatalf("importing roaring: %v", err) } - tx.Dump() exp := calcExpected(test[:num+1]...) for row, expCols := range exp { @@ -3960,6 +3958,7 @@ func TestFragment_RoaringImportTopN(t *testing.T) { if err != nil { t.Fatalf("bulk importing ids: %v", err) } + expPairs := calcTop(test.rowIDs, test.colIDs) pairs, err := f.top(tx, topOptions{}) if err != nil { diff --git a/index.go b/index.go index 6007af4f8..d1a7770df 100644 --- a/index.go +++ b/index.go @@ -72,8 +72,21 @@ type Index struct { Txf *TxFactory } -// NewIndex returns a new instance of Index. +// OpenIndex opens or starts a new Index on path. Path +// can be empty. +func OpenIndex(holder *Holder, path, name string) (*Index, error) { + openExisting := true + return openOrCreateNewIndex(holder, path, name, openExisting) +} + +// NewIndex returns a new instance of Index at path. It will erase anything +// old already in path. func NewIndex(holder *Holder, path, name string) (*Index, error) { + openExisting := false + return openOrCreateNewIndex(holder, path, name, openExisting) +} + +func openOrCreateNewIndex(holder *Holder, path, name string, openExisting bool) (*Index, error) { // Emulate what the spf13/cobra does, letting env vars override // the defaults, because we may be under a simple "go test" run where @@ -104,7 +117,7 @@ func NewIndex(holder *Holder, path, name string) (*Index, error) { return nil, errors.Wrap(err, "validating name") } - txf, err := NewTxFactory(txsrc, holder.Path, name) + txf, err := NewTxFactory(txsrc, holder.Path, name, openExisting) if err != nil { return nil, errors.Wrap(err, "creating newTxFactory") } diff --git a/like_test.go b/like_test.go index 1382901ec..7e5268dea 100644 --- a/like_test.go +++ b/like_test.go @@ -20,7 +20,6 @@ import ( ) func TestPlanLike(t *testing.T) { - t.Parallel() cases := []struct { name string diff --git a/rbf/rbf.go b/rbf/rbf.go index 2aa3a473a..911532d15 100644 --- a/rbf/rbf.go +++ b/rbf/rbf.go @@ -262,8 +262,14 @@ func align8(offset int) int { // leafCell represents a leaf cell. type leafCell struct { Key uint64 - Type int - N int + Type int // container type + + // N is the number of "things" in Data: + // for an array container the number of integers in the array. + // for an RLE, number of intervals. + // etc. + N int + BitN int Data []byte } @@ -392,19 +398,22 @@ func (c *leafCell) lastValue() uint16 { } // countRange returns the bit count within the given range. -func (c *leafCell) countRange(start, end uint16) (n int) { +// We have to take int32 rather than uint16 because the interval is [start, end), +// and otherwise we have no way to ask to count the entire container (the +// high bit will be missed). +func (c *leafCell) countRange(start, end int32) (n int) { // If the full range is being queried, simply use the precalculated count. - if start == 0 && end == math.MaxUint16 { + if start == 0 && end > math.MaxUint16 { return c.BitN } switch c.Type { case ContainerTypeArray: - return int(roaring.ArrayCountRange(toArray16(c.Data), int32(start), int32(end))) + return int(roaring.ArrayCountRange(toArray16(c.Data), start, end)) case ContainerTypeRLE: - return int(roaring.RunCountRange(toInterval16(c.Data), int32(start), int32(end))) + return int(roaring.RunCountRange(toInterval16(c.Data), start, end)) case ContainerTypeBitmap: - return int(roaring.BitmapCountRange(toArray64(c.Data), int32(start), int32(end))) + return int(roaring.BitmapCountRange(toArray64(c.Data), start, end)) default: panic(fmt.Sprintf("invalid container type: %d", c.Type)) } @@ -412,11 +421,15 @@ func (c *leafCell) countRange(start, end uint16) (n int) { func readLeafCellKey(page []byte, i int) uint64 { offset := readCellOffset(page, i) + assert(offset < len(page)) return *(*uint64)(unsafe.Pointer(&page[offset])) } func readLeafCell(page []byte, i int) leafCell { offset := readCellOffset(page, i) + + // cd ..; PILOSA_TXSRC=rbf go test -v -run TestFragment_TopN_IDs -tags=' shardwidth20' "-gcflags=all=-d=checkptr=0" + // gives panic: runtime error: slice bounds out of range [16390:8192] here. buf := page[offset:] var cell leafCell diff --git a/rbf/tx.go b/rbf/tx.go index ca208fd1b..d11c34909 100644 --- a/rbf/tx.go +++ b/rbf/tx.go @@ -36,6 +36,15 @@ type Tx struct { pageMap *immutable.Map // mapping of database pages to WAL IDs writable bool // if true, tx can write dirty bool // if true, changes have been made + + // If Rollback() has already completed, don't do it again. + // Note db == nil means that commit has already been done. + rollbackDone bool + + // DeleteEmptyContainer lets us by default match the roaring + // behavior where an existing container has all its bits cleared + // but still sticks around in the database. + DeleteEmptyContainer bool } // Writable returns true if the transaction can mutate data. @@ -74,12 +83,17 @@ func (tx *Tx) Commit() error { func (tx *Tx) Rollback() { tx.mu.Lock() defer tx.mu.Unlock() - - // TODO(bbj): Invalidate DB if rollback fails. Possibly attempt reopen? - - if tx.db == nil { + // allow Rollback to be called more than once. + if tx.rollbackDone { return } + tx.rollbackDone = true + if tx.db == nil { + // Commit already done. + return + } + + // TODO(bbj): Invalidate DB if rollback fails. Possibly attempt reopen? // If any pages have been written, ensure we write a new meta page with // the rollback flag to mark the end of the transaction. This allows us to @@ -92,10 +106,18 @@ func (tx *Tx) Rollback() { } } - _ = tx.db.checkpoint() // TODO: Check error + // turn on these error checks! we see + // panic: cannot find segment containing WAL page: 1 + // when running go test -v + // TestCursor_FirstNext_Quick/6 + // + //panicOn(tx.db.checkpoint()) + //panicOn(tx.db.removeTx(tx)) + + _ = tx.db.checkpoint() // Disconnect transaction from DB. - _ = tx.db.removeTx(tx) // TODO: Check error + _ = tx.db.removeTx(tx) } // Root returns the root page number for a bitmap. Returns 0 if the bitmap does not exist. @@ -150,6 +172,8 @@ func (tx *Tx) CreateBitmap(name string) error { } func (tx *Tx) createBitmap(name string) error { + //vv("createBitmap(name='%v'", name) + if tx.db == nil { return ErrTxClosed } else if !tx.writable { @@ -425,6 +449,8 @@ func (tx *Tx) writeRootRecordPages(records []*RootRecord) (err error) { // Add sets a given bit on the bitmap. func (tx *Tx) Add(name string, a ...uint64) (changeCount int, err error) { + //vv("rbf Tx.Add(a='%#v')", a) + tx.mu.Lock() defer tx.mu.Unlock() @@ -589,14 +615,14 @@ func (tx *Tx) Container(name string, key uint64) (*roaring.Container, error) { } // PutContainer inserts a container into a bitmap. Overwrites if key already exists. -func (tx *Tx) PutContainer(name string, key uint64, cont *roaring.Container) error { +func (tx *Tx) PutContainer(name string, key uint64, ct *roaring.Container) error { tx.mu.Lock() defer tx.mu.Unlock() - cell := ConvertToLeafArgs(key, cont) - if cell.BitN == 0 { + if ct.N() == 0 { return nil } + cell := ConvertToLeafArgs(key, ct) if err := tx.createBitmapIfNotExists(name); err != nil { return err @@ -939,16 +965,27 @@ func (tx *Tx) ContainerIterator(name string, key uint64) (citer roaring.Containe tx.mu.RLock() defer tx.mu.RUnlock() - c, err := tx.cursor(name) - if err != nil { - // TODO(bbj): Don't return error if bitmap is simply not found? - return nil, false, err - } else if c == nil { - return nil, false, nil - } else if err := c.First(); err != nil { - return nil, false, err + var c *Cursor + c, err = tx.cursor(name) + if c == nil && err == nil { + // nothing available. + citer = &emptyContainerIterator{} + return } - return &containerIterator{cursor: c}, true, nil + if err != nil { + return + } + + // INVAR: c is not nil + + err = c.First() + if err != nil { + return + } + ci := &containerIterator{cursor: c} + citer = ci + + return citer, true, nil } func (tx *Tx) ForEach(name string, fn func(i uint64) error) error { @@ -1091,18 +1128,28 @@ func (tx *Tx) UnionInPlace(name string, others ...*roaring.Bitmap) error { panic("TODO") } +// roaring.countRange counts the number of bits set between [start, end). func (tx *Tx) CountRange(name string, start, end uint64) (uint64, error) { tx.mu.RLock() defer tx.mu.RUnlock() - c, err := tx.cursor(name) - if err != nil { - return 0, err - } else if c == nil { + if start >= end { return 0, nil } - if err := c.First(); err == io.EOF { + skey := highbits(start) + ekey := highbits(end) + + csr, err := tx.cursor(name) + if err != nil { + return 0, err + } else if csr == nil { + return 0, nil + } + + exact, err := csr.Seek(skey) + _ = exact + if err == io.EOF { return 0, nil } else if err != nil { return 0, err @@ -1110,37 +1157,94 @@ func (tx *Tx) CountRange(name string, start, end uint64) (uint64, error) { var n uint64 for { - if err := c.Next(); err == io.EOF { + if err := csr.Next(); err == io.EOF { break } else if err != nil { return 0, err } - cell := c.cell() - if cell.Key > highbits(end) { + c := csr.cell() + k := c.Key + if k > ekey { break } - if cell.Key == highbits(start) { - n += uint64(cell.countRange(lowbits(start), math.MaxUint16)) - } else if cell.Key == highbits(end) { - n += uint64(cell.countRange(0, lowbits(end))) - } else { - n += uint64(cell.BitN) + // If range is entirely in one container then just count that range. + if skey == ekey { + return uint64(c.countRange(int32(lowbits(start)), int32(lowbits(end)))), nil + } + // INVAR: skey < ekey + + // k > ekey handles the case when start > end and where start and end + // are in different containers. Same container case is already handled above. + if k > ekey { + break + } + if k == skey { + n += uint64(c.countRange(int32(lowbits(start)), roaring.MaxContainerVal+1)) + continue + } + if k < ekey { + n += uint64(c.BitN) + continue + } + if k == ekey { + n += uint64(c.countRange(0, int32(lowbits(end)))) + break } } return n, nil } -func (tx *Tx) OffsetRange(name string, offset, start, end uint64) (*roaring.Bitmap, error) { +func (tx *Tx) OffsetRange(name string, offset, start, endx uint64) (*roaring.Bitmap, error) { + if lowbits(offset) != 0 { + panic("offset must not contain low bits") + } else if lowbits(start) != 0 { + panic("range start must not contain low bits") + } else if lowbits(endx) != 0 { + panic("range endx must not contain low bits") + } + tx.mu.RLock() defer tx.mu.RUnlock() - b, err := tx.RoaringBitmap(name) + c, err := tx.cursor(name) if err != nil { return nil, err } - return b.OffsetRange(offset, start, end), nil + + other := roaring.NewSliceBitmap() + off := highbits(offset) + hi0, hi1 := highbits(start), highbits(endx) + + if c == nil { + // bitmap not found. Match what roaring does and return nil in this case. + return other, nil + } + + if _, err := c.Seek(hi0); err == io.EOF { + return other, nil + } else if err != nil { + return nil, err + } + + for { + if err := c.Next(); err == io.EOF { + break + } else if err != nil { + return nil, err + } + + cell := c.cell() + ckey := cell.Key + + // >= hi1 is correct b/c endx cannot have any lowbits set. + if ckey >= hi1 { + break + } + other.Containers.Put(off+(ckey-hi0), toContainer(cell, tx)) + } + return other, nil } // containerIterator wraps Cursor to implement roaring.ContainerIterator. @@ -1163,6 +1267,18 @@ func (itr *containerIterator) Value() (uint64, *roaring.Container) { return cell.Key, toContainer(cell, itr.cursor.tx) } +// always returns false for Next() +type emptyContainerIterator struct{} + +func (si *emptyContainerIterator) Close() {} + +func (si *emptyContainerIterator) Next() bool { + return false +} +func (si *emptyContainerIterator) Value() (uint64, *roaring.Container) { + panic("emptyContainerIterator never has any Values") +} + func (tx *Tx) Dump(index string) { fmt.Println(tx.DumpString(index)) } @@ -1177,7 +1293,7 @@ func (tx *Tx) DumpString(index string) (r string) { for _, rr := range records { c, err := tx.cursor(rr.Name) panicOn(err) - err = c.First() + err = c.First() // First will rewind to beginning. if err == io.EOF { r += "" n++ @@ -1185,7 +1301,7 @@ func (tx *Tx) DumpString(index string) (r string) { } panicOn(err) for { - err := c.Next() // hung here? + err := c.Next() if err == io.EOF { break } @@ -1322,3 +1438,133 @@ func stringOfCkeyCt(ckey uint64, ct *roaring.Container, rrName, index string) (s s += " ......." + srbm + "\n" return } + +func (tx *Tx) ImportRoaringBits(name string, itr roaring.RoaringIterator, clear bool, log bool, rowSize uint64, data []byte) (changed int, rowSet map[uint64]int, err error) { + + // begin write boilerplate + if tx.db == nil { + err = ErrTxClosed + return + } else if !tx.writable { + err = ErrTxNotWritable + return + } else if name == "" { + err = ErrBitmapNameRequired + return + } + + if err = tx.createBitmapIfNotExists(name); err != nil { + return + } + // end write boilerplate + + n := itr.Len() + if n == 0 { + return + } + rowSet = make(map[uint64]int) + + var currRow uint64 + + var oldC *roaring.Container + for itrKey, synthC := itr.NextContainer(); synthC != nil; itrKey, synthC = itr.NextContainer() { + if rowSize != 0 { + currRow = itrKey / rowSize + } + nsynth := int(synthC.N()) + if nsynth == 0 { + continue + } + // INVAR: nsynth > 0 + + oldC, err = tx.Container(name, itrKey) + panicOn(err) + if err != nil { + return + } + + if oldC == nil || oldC.N() == 0 { + // no container at the itrKey in badger (or all zero container). + if clear { + // changed of 0 and empty rowSet is perfect, no need to change the defaults. + continue + } else { + + changed += nsynth + rowSet[currRow] += nsynth + + err = tx.PutContainer(name, itrKey, synthC) + if err != nil { + return + } + continue + } + } + + if clear { + existN := oldC.N() // number of bits set in the old container + newC := oldC.Difference(synthC) + + // update rowSet and changes + if newC.N() == existN { + // INVAR: do changed need adjusting? nope. same bit count, + // so no change could have happened. + continue + } else { + changes := int(existN - newC.N()) + changed += changes + rowSet[currRow] -= changes + + if tx.DeleteEmptyContainer && newC.N() == 0 { + err = tx.RemoveContainer(name, itrKey) + if err != nil { + return + } + continue + } + err = tx.PutContainer(name, itrKey, newC) + if err != nil { + return + } + continue + } + } else { + // setting bits + + existN := oldC.N() + if existN == roaring.MaxContainerVal+1 { + // completely full container already, set will do nothing. so changed of 0 default is perfect. + continue + } + if existN == 0 { + // can nsynth be zero? No, because of the continue/invariant above where nsynth > 0 + changed += nsynth + rowSet[currRow] += nsynth + err = tx.PutContainer(name, itrKey, synthC) + if err != nil { + return + } + continue + } + + newC := oldC.UnionInPlace(synthC) + + if roaring.ContainerType(newC) == containerBitmap { + newC.Repair() // update the bit-count so .n is valid. b/c UnionInPlace doesn't update it. + } + if newC.N() != existN { + changes := int(newC.N() - existN) + changed += changes + rowSet[currRow] += changes + + err = tx.PutContainer(name, itrKey, newC) + if err != nil { + panicOn(err) + return + } + continue + } + } + } + return +} diff --git a/tx.go b/tx.go index ff544aca7..cc78b501f 100644 --- a/tx.go +++ b/tx.go @@ -993,8 +993,7 @@ func (tx *RBFTx) OffsetRange(index, field, view string, shard uint64, offset, st func (tx *RBFTx) IncrementOpN(index, field, view string, shard uint64, changedN int) {} func (tx *RBFTx) ImportRoaringBits(index, field, view string, shard uint64, rit roaring.RoaringIterator, clear bool, log bool, rowSize uint64, data []byte) (changed int, rowSet map[uint64]int, err error) { - // TODO: Implement RBFTX.ImportRoaringBits" - return 0, make(map[uint64]int), nil + return tx.tx.ImportRoaringBits(rbfName(field, view, shard), rit, clear, log, rowSize, data) } func (tx *RBFTx) RoaringBitmapReader(index, field, view string, shard uint64, fragmentPathForRoaring string) (r io.ReadCloser, sz int64, err error) { diff --git a/txfactory.go b/txfactory.go index 6120efe9f..de2e90c25 100644 --- a/txfactory.go +++ b/txfactory.go @@ -141,7 +141,7 @@ func MustTxsrcToTxtype(txsrc string) txtype { // always store files in a subdir of dir. If we are having one // database or many can depend on name. -func NewTxFactory(txsrc string, dir, name string) (f *TxFactory, err error) { +func NewTxFactory(txsrc string, dir, name string, openExisting bool) (f *TxFactory, err error) { ty := MustTxsrcToTxtype(txsrc) if ty < 1 || ty > 9 { @@ -162,14 +162,16 @@ func NewTxFactory(txsrc string, dir, name string) (f *TxFactory, err error) { // enables cross-index Tx, which are important and are tested for. path := dir + sep + "honeyBadger" - f.badgerDB, err = globalBadgerReg.openBadgerDBWrapper(path) - // TODO(jea): figure out what the appropriate error path is here. - //fmt.Printf("warning: could not open badgerdb on path '%v': '%v'. For safety, we are opening a new '%v-fallback' instead\n", path, err, path+"-fallback") - if err != nil { + if openExisting { + f.badgerDB, err = globalBadgerReg.openBadgerDBWrapper(path) + if err != nil { + return nil, errors.Wrap(err, fmt.Sprintf("cannot open badger db. path='%v'", path)) + } + } else { f.badgerDB, err = globalBadgerReg.newBadgerDBWrapper(path) - } - if err != nil { - return nil, errors.Wrap(err, fmt.Sprintf("cannot open badger db. path='%v'", path)) + if err != nil { + return nil, errors.Wrap(err, fmt.Sprintf("cannot create new badger db. path='%v'", path)) + } } // electric-fence like finding of access to mmapped data beyond // transaction end time. @@ -178,6 +180,7 @@ func NewTxFactory(txsrc string, dir, name string) (f *TxFactory, err error) { switch ty { case rbfTxn, blueGreenRBFRoaring, blueGreenRoaringRBF, blueGreenBadgerRBF, blueGreenRBFBadger: + f.rbfDB = rbf.NewDB(filepath.Join(dir, "db.rbf")) if err := f.rbfDB.Open(); err != nil { return nil, errors.Wrap(err, "cannot open rbf db") @@ -279,6 +282,18 @@ func (f *TxFactory) CloseIndex(idx *Index) error { return nil case blueGreenRoaringBadger: return nil + + case blueGreenRBFRoaring: + _ = f.rbfDB.Close() + return nil + case blueGreenRoaringRBF: + return f.rbfDB.Close() + case blueGreenBadgerRBF: + return f.rbfDB.Close() + case blueGreenRBFBadger: + _ = f.rbfDB.Close() + return nil + } panic(fmt.Sprintf("unknown f.typeOfTx type: '%v'", f.typeOfTx)) } @@ -300,7 +315,7 @@ func (f *TxFactory) NewTx(o Txo) Tx { if err != nil { panic(err) // TODO: Add error return on NewTx() } - return &RBFTx{tx: tx} + return &RBFTx{tx: tx, index: indexName} case blueGreenBadgerRoaring: btx := f.badgerDB.NewBadgerTx(o.Write, indexName) rtx := &RoaringTx{write: o.Write, Field: o.Field, Index: o.Index, fragment: o.Fragment} @@ -310,37 +325,35 @@ func (f *TxFactory) NewTx(o Txo) Tx { rtx := &RoaringTx{write: o.Write, Field: o.Field, Index: o.Index, fragment: o.Fragment} return newBlueGreenTx(rtx, btx, f.idx) - /* - case blueGreenBadgerRBF: - btx := f.badgerDB.NewBadgerTx(o.Write, indexName) - rbftx, err := f.rbfDB.Begin(o.Write) - if err != nil { - errors.Wrap(err, "rbfDB.Begin transaction errored") - } - return newBlueGreenTx(btx, rbftx, f.idx) - case blueGreenRBFBadger: - btx := f.badgerDB.NewBadgerTx(o.Write, indexName) - rbftx, err := f.rbfDB.Begin(o.Write) - if err != nil { - errors.Wrap(err, "rbfDB.Begin transaction errored") - } - return newBlueGreenTx(rbftx, btx, f.idx) + case blueGreenBadgerRBF: + btx := f.badgerDB.NewBadgerTx(o.Write, indexName) + rbftx, err := f.rbfDB.Begin(o.Write) + if err != nil { + panic(errors.Wrap(err, "rbfDB.Begin transaction errored")) + } + return newBlueGreenTx(btx, &RBFTx{tx: rbftx, index: indexName}, f.idx) + case blueGreenRBFBadger: + btx := f.badgerDB.NewBadgerTx(o.Write, indexName) + rbftx, err := f.rbfDB.Begin(o.Write) + if err != nil { + panic(errors.Wrap(err, "rbfDB.Begin transaction errored")) + } + return newBlueGreenTx(&RBFTx{tx: rbftx, index: indexName}, btx, f.idx) - case blueGreenRBFRoaring: - rbftx, err := f.rbfDB.Begin(o.Write) - if err != nil { - errors.Wrap(err, "rbfDB.Begin transaction errored") - } - rtx := &RoaringTx{write: o.Write, Field: o.Field, Index: o.Index, fragment: o.Fragment} - return newBlueGreenTx(rbftx, rtx, f.idx) - case blueGreenRoaringRBF: - rbftx, err := f.rbfDB.Begin(o.Write) - if err != nil { - errors.Wrap(err, "rbfDB.Begin transaction errored") - } - rtx := &RoaringTx{write: o.Write, Field: o.Field, Index: o.Index, fragment: o.Fragment} - return newBlueGreenTx(rtx, rbftx, f.idx) - */ + case blueGreenRBFRoaring: + rbftx, err := f.rbfDB.Begin(o.Write) + if err != nil { + panic(errors.Wrap(err, "rbfDB.Begin transaction errored")) + } + rtx := &RoaringTx{write: o.Write, Field: o.Field, Index: o.Index, fragment: o.Fragment} + return newBlueGreenTx(&RBFTx{tx: rbftx, index: indexName}, rtx, f.idx) + case blueGreenRoaringRBF: + rbftx, err := f.rbfDB.Begin(o.Write) + if err != nil { + panic(errors.Wrap(err, "rbfDB.Begin transaction errored")) + } + rtx := &RoaringTx{write: o.Write, Field: o.Field, Index: o.Index, fragment: o.Fragment} + return newBlueGreenTx(rtx, &RBFTx{tx: rbftx, index: indexName}, f.idx) } panic(fmt.Sprintf("unknown f.typeOfTx type: '%v'", f.typeOfTx)) }