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

@@ -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
 		}

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()

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)
 		}

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 {

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")
 	}

like_test.go

@@ -20,7 +20,6 @@ import (
 )
 
 func TestPlanLike(t *testing.T) {
-	t.Parallel()
 
 	cases := []struct {
 		name            string

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

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 += "<empty bitmap>"
 			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
+}

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) {

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))
 }