diff --git a/fragment_internal_test.go b/fragment_internal_test.go index 871864d3d..fd47a36f2 100644 --- a/fragment_internal_test.go +++ b/fragment_internal_test.go @@ -4215,8 +4215,6 @@ func TestFragmentRowIterator(t *testing.T) { }) t.Run("skipped rows wrapped", func(t *testing.T) { - skipForRBF(t) - f, idx := mustOpenFragment("i", "f", "v", 0, CacheTypeRanked) _ = idx defer f.Clean(t) diff --git a/holder_test.go b/holder_test.go index 08b84acc6..079029484 100644 --- a/holder_test.go +++ b/holder_test.go @@ -32,7 +32,6 @@ import ( ) func TestHolder_Open(t *testing.T) { - t.Run("ErrIndexName", func(t *testing.T) { h := test.MustOpenHolder() @@ -404,8 +403,6 @@ func TestHolder_HasData(t *testing.T) { // Ensure holder can delete an index and its underlying files. func TestHolder_DeleteIndex(t *testing.T) { - skipForRBF(t) - hldr := test.MustOpenHolder() defer hldr.Close() @@ -700,8 +697,6 @@ func TestHolderSyncer_TimeQuantum(t *testing.T) { // Ensure holder can sync integer views with a remote holder. func TestHolderSyncer_IntField(t *testing.T) { - skipForRBF(t) - t.Run("BasicSync", func(t *testing.T) { c := test.MustNewCluster(t, 2) c[0].Config.Cluster.ReplicaN = 2 diff --git a/rbf/cursor.go b/rbf/cursor.go index ed0a9f849..2bc7c9b7d 100644 --- a/rbf/cursor.go +++ b/rbf/cursor.go @@ -428,10 +428,10 @@ func (c *Cursor) putLeafCell(in leafCell) (err error) { // Initialize a new root if we are currently the root page. if c.stack.index == 0 { - assert(newRoot) + assert(newRoot, "leaf write must be root when stack at root") return c.writeRoot(origPgno, parents) } - assert(!newRoot) + assert(!newRoot, "leaf write must NOT be root when stack not at root") // Otherwise update existing parent. return c.putBranchCells(c.stack.index-1, parents) @@ -552,10 +552,10 @@ func (c *Cursor) putBranchCells(stackIndex int, newCells []branchCell) (err erro // Initialize a new root if we are currently the root page. if stackIndex == 0 { - assert(newRoot) + assert(newRoot, "branch write must be root when stack at root") return c.writeRoot(origPgno, parents) } - assert(!newRoot) + assert(!newRoot, "branch write must NOT be root when stack at root") // Otherwise update existing parent. return c.putBranchCells(stackIndex-1, parents) @@ -813,7 +813,7 @@ func (c *Cursor) Seek(key uint64) (exact bool, err error) { c.buffered = true for c.stack.index = 0; ; c.stack.index++ { elem := &c.stack.elems[c.stack.index] - assert(elem.pgno != 0) + assert(elem.pgno != 0, "cursor should never point to page zero (meta)") buf, err := c.tx.readPage(elem.pgno) if err != nil { @@ -870,6 +870,11 @@ func (c *Cursor) Seek(key uint64) (exact bool, err error) { func (c *Cursor) Next() error { if c.buffered { c.buffered = false + + // Move to next available element if we are past the last cell in the page. + if elem := &c.stack.elems[c.stack.index]; elem.index >= readCellN(c.leafPage) { + return c.goNextPage() + } return nil } @@ -1041,6 +1046,9 @@ func (c *Cursor) Intersect(rowID uint64, row []uint64) error { // Values returns the values for the container the cursor is currently pointing to. func (c *Cursor) Values() []uint16 { elem := &c.stack.elems[c.stack.index] + if readCellN(c.leafPage[:]) == 0 { + return nil + } cell := readLeafCell(c.leafPage[:], elem.index) return cell.Values(c.tx) } diff --git a/rbf/cursor_test.go b/rbf/cursor_test.go index bd8e67203..ca946ef86 100644 --- a/rbf/cursor_test.go +++ b/rbf/cursor_test.go @@ -848,31 +848,27 @@ func TestCursor_UpdateBranchCells(t *testing.T) { if err != nil { t.Fatal(err) } + + changed, err := c.Add(1) if err != nil { t.Fatal(err) - } - changed, err := c.Add(1) - if changed { - - if err := c.First(); err != nil { - t.Fatal(err) - } - if got, want := c.Values(), []uint16{uint16(1)}; !reflect.DeepEqual(got, want) { - t.Fatal(err) - } - } else { + } else if !changed { t.Fatal("Expected Add Change") + } else if err := c.First(); err != nil { + t.Fatal(err) + } else if got, want := c.Values(), []uint16{uint16(1)}; !reflect.DeepEqual(got, want) { + t.Fatal(err) } + changed, err = c.Remove(1) - if changed { - if err := c.First(); err != nil && err != io.EOF { - t.Fatal(err) - } - if got, want := c.Values(), []uint16{}; !reflect.DeepEqual(got, want) { - t.Fatal(err) - } - } else { + if err != nil { + t.Fatal(err) + } else if !changed { t.Fatal("Expected Remove Change") + } else if err := c.First(); err != nil && err != io.EOF { + t.Fatal(err) + } else if got, want := c.Values(), ([]uint16)(nil); !reflect.DeepEqual(got, want) { + t.Fatal(err) } rb := func() *roaring.Bitmap { @@ -901,7 +897,7 @@ func TestCursor_UpdateBranchCells(t *testing.T) { if err := c.First(); err != nil && err != io.EOF { panic(err) } - if got, want := c.Values(), []uint16{}; !reflect.DeepEqual(got, want) { + if got, want := c.Values(), ([]uint16)(nil); !reflect.DeepEqual(got, want) { t.Fatal(err) } } else { diff --git a/rbf/rbf.go b/rbf/rbf.go index 911532d15..e92fa7f38 100644 --- a/rbf/rbf.go +++ b/rbf/rbf.go @@ -421,11 +421,12 @@ func (c *leafCell) countRange(start, end int32) (n int) { func readLeafCellKey(page []byte, i int) uint64 { offset := readCellOffset(page, i) - assert(offset < len(page)) + assert(offset < len(page), "cell read beyond page size: offset %d >= page size %d", offset, len(page)) return *(*uint64)(unsafe.Pointer(&page[offset])) } func readLeafCell(page []byte, i int) leafCell { + assert(i < readCellN(page), "cell index %d exceeds cell count %d", i, readCellN(page)) offset := readCellOffset(page, i) // cd ..; PILOSA_TXSRC=rbf go test -v -run TestFragment_TopN_IDs -tags=' shardwidth20' "-gcflags=all=-d=checkptr=0" @@ -475,7 +476,7 @@ func writeLeafCell(page []byte, i, offset int, cell leafCell) { *(*uint32)(unsafe.Pointer(&page[offset+8])) = uint32(cell.Type) *(*uint16)(unsafe.Pointer(&page[offset+12])) = uint16(cell.N) *(*uint16)(unsafe.Pointer(&page[offset+14])) = uint16(cell.BitN) - assert(offset+16+len(cell.Data) <= PageSize) + assert(offset+16+len(cell.Data) <= PageSize, "leaf cell write extends beyond page: offset %d + cell size %d > page size %d", offset, 16+len(cell.Data), PageSize) copy(page[offset+16:], cell.Data) } @@ -501,7 +502,8 @@ func readBranchCellKey(page []byte, i int) uint64 { } func readBranchCell(page []byte, i int) branchCell { - assert(i >= 0) + assert(i >= 0, "branch cell index must be zero or greater: index=%d", i) + assert(i < readCellN(page), "branch cell index %d must less than cell count %d", i, readCellN(page)) offset := readCellOffset(page, i) var cell branchCell @@ -612,9 +614,9 @@ func Walk(tx *Tx, pgno uint32, v func(uint32, []*RootRecord)) { } } -func assert(condition bool) { +func assert(condition bool, format string, args ...interface{}) { if !condition { - panic("assertion failed") + panic(fmt.Sprintf("assertion failed: "+format, args...)) } } diff --git a/rbf/tx.go b/rbf/tx.go index d11c34909..b4821bd98 100644 --- a/rbf/tx.go +++ b/rbf/tx.go @@ -965,27 +965,19 @@ func (tx *Tx) ContainerIterator(name string, key uint64) (citer roaring.Containe tx.mu.RLock() defer tx.mu.RUnlock() - var c *Cursor - c, err = tx.cursor(name) + c, err := tx.cursor(name) if c == nil && err == nil { - // nothing available. - citer = &emptyContainerIterator{} - return - } - if err != nil { - return + return &emptyContainerIterator{}, false, nil // nothing available. + } else if err != nil { + return nil, false, err } // INVAR: c is not nil - err = c.First() - if err != nil { - return + if _, err := c.Seek(key); err != nil { + return nil, false, err } - ci := &containerIterator{cursor: c} - citer = ci - - return citer, true, nil + return &containerIterator{cursor: c}, true, nil } func (tx *Tx) ForEach(name string, fn func(i uint64) error) error { diff --git a/rbf/wal.go b/rbf/wal.go index 501b4f63c..0f8461e16 100644 --- a/rbf/wal.go +++ b/rbf/wal.go @@ -143,7 +143,7 @@ func (s *WALSegment) ReadWALPage(walID int64) ([]byte, error) { // WriteWALPage writes a single page to the WAL segment and returns its WAL identifier. func (s *WALSegment) WriteWALPage(page []byte, isMeta bool) (walID int64, err error) { - assert(len(page) == PageSize) + assert(len(page) == PageSize, "invalid page size: %d", len(page)) // Initialize write file handle if not yet initialized. if s.w == nil { diff --git a/translator_test.go b/translator_test.go index 0acce9b66..9bb381533 100644 --- a/translator_test.go +++ b/translator_test.go @@ -377,7 +377,7 @@ func TestTranslation_Coordinator(t *testing.T) { // Ensure that field key translations requests sent to // non-coordinator nodes are forwarded to the coordinator. t.Run("ForwardFieldKey", func(t *testing.T) { - t.Skip("Short term skip to avoid go 1.13 test Should remove ASAP") + t.Skip("Short term skip to avoid go 1.13 test Should remove ASAP") // Start a 2-node cluster. c := test.MustRunCluster(t, 2, []server.CommandOption{ diff --git a/tx.go b/tx.go index cc78b501f..29f2da821 100644 --- a/tx.go +++ b/tx.go @@ -1052,6 +1052,11 @@ func rbfName(field, view string, shard uint64) string { return fmt.Sprintf("%s\x00%s\x00%d", field, view, shard) } +// rbfFieldPrefix returns a prefix for field keys in RBF. +func rbfFieldPrefix(field string) string { + return fmt.Sprintf("%s\x00", field) +} + // rbfFieldViewPrefix returns a NULL-separated prefix for keys in RBF. func rbfFieldViewPrefix(field, view string) string { return fmt.Sprintf("%s\x00%s\x00", field, view) diff --git a/txfactory.go b/txfactory.go index 816ed59c8..aa1384972 100644 --- a/txfactory.go +++ b/txfactory.go @@ -240,8 +240,21 @@ func (f *TxFactory) DeleteFieldFromStore(index, field, fieldPath string) error { case badgerTxn: return f.badgerDB.DeleteField(index, field, fieldPath) case rbfTxn: - //return f.rbfDB.DeleteField(index, field, fieldPath) - return nil + if err := os.RemoveAll(fieldPath); err != nil { + return errors.Wrap(err, "removing directory") + } + + tx, err := f.rbfDB.Begin(true) + if err != nil { + return err + } + defer tx.Rollback() + + if err := tx.DeleteBitmapsWithPrefix(rbfFieldPrefix(field)); err != nil { + return err + } + return tx.Commit() + case blueGreenBadgerRoaring: _ = f.badgerDB.DeleteField(index, field, fieldPath) return f.roaringDB.DeleteField(index, field, fieldPath)