Merge pull request #637 from molecula/rbf-fixes

Multiple RBF test fixes

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)

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

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

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 {

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

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 {

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 {

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{

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)

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)