From b2b614a6869f25dad8f1cfd32281858491bee035 Mon Sep 17 00:00:00 2001
From: Todd Gruben <todd@pilosa.com>
Date: Mon, 13 Jul 2020 19:32:26 -0500
Subject: [PATCH] refactor cursor.GetBitmap->tx.GetBitmap;rename to
 WalkRootRecordPages;err check

---
 rbf/cursor.go      | 38 ++++++++++++++++++++++++--------------
 rbf/cursor_test.go | 36 ++++++++++++++++++++++++++++++++++++
 rbf/db.go          |  2 +-
 rbf/rbf.go         | 10 +++++-----
 rbf/tx.go          | 19 +++++++++++++++----
 5 files changed, 81 insertions(+), 24 deletions(-)

diff --git a/rbf/cursor.go b/rbf/cursor.go
index 56822341e..60727743d 100644
--- a/rbf/cursor.go
+++ b/rbf/cursor.go
@@ -9,6 +9,7 @@ import (
 	"unsafe"
 
 	"github.com/pilosa/pilosa/v2/roaring"
+	"github.com/pkg/errors"
 )
 
 const (
@@ -155,9 +156,10 @@ func (c *Cursor) Add(v uint64) (changed bool, err error) {
 		return false, nil
 	case ContainerTypeBitmap:
 		// Exit if bit set in bitmap container.
-		pgno, bm, err := cell.GetBitmap(c.tx)
+		pgno, bm, err := c.tx.GetBitmap(&cell)
+
 		if err != nil {
-			return false, err
+			return false, errors.Wrap(err, "cursor.Add")
 		}
 
 		a := cloneArray64(bm)
@@ -234,9 +236,9 @@ func (c *Cursor) Remove(v uint64) (changed bool, err error) {
 		}
 		return true, c.putLeafCell(leafArgs{Key: cell.Key, Type: ContainerTypeRLE, N: len(runs), Data: fromInterval16(runs)})
 	case ContainerTypeBitmap:
-		pgno, bm, err := cell.GetBitmap(c.tx)
+		pgno, bm, err := c.tx.GetBitmap(&cell)
 		if err != nil {
-			return false, err
+			return false, errors.Wrap(err, "cursor.add")
 		}
 		a := cloneArray64(bm)
 		if a[lo/64]&(1<<uint64(lo%64)) == 0 {
@@ -284,8 +286,10 @@ func (c *Cursor) Contains(v uint64) (exists bool, err error) {
 		}
 		return false, nil
 	case ContainerTypeBitmap:
-		_, a, err := cell.GetBitmap(c.tx)
-
+		_, a, err := c.tx.GetBitmap(&cell)
+		if err != nil {
+			return false, errors.Wrap(err, "cursor.Contains")
+		}
 		return a[lo/64]&(1<<uint64(lo%64)) != 0, err
 	default:
 		return false, fmt.Errorf("rbf.Cursor.Contains(): invalid container type: %d", cell.Type)
@@ -320,7 +324,10 @@ func (c *Cursor) putLeafCell(in leafArgs) (err error) {
 	} else if in.Type == ContainerTypeBitmap {
 		ecell := cells[elem.index]
 		if ecell.Type != ContainerTypeBitmap {
-			bitmapPgno, _ := c.tx.allocate()
+			bitmapPgno, err := c.tx.allocate()
+			if err != nil {
+				return errors.Wrap(err, "cursor.putLeafCell")
+			}
 			cell.Data = fromPgno(bitmapPgno)
 		} else {
 			cell.Data = ecell.Data //fill in the old pgno
@@ -936,7 +943,10 @@ func (c *Cursor) Union(rowID uint64, row []uint64) error {
 		case ContainerTypeRLE:
 			panic("TODO(BBJ): rbf.Bitmap.Union() RLE support")
 		case ContainerTypeBitmap:
-			_, bm, _ := cell.GetBitmap(c.tx)
+			_, bm, err := c.tx.GetBitmap(&cell)
+			if err != nil {
+				return errors.Wrap(err, "union")
+			}
 			for i, v := range bm {
 				row[(offset/64)+uint64(i)] |= v
 			}
@@ -981,9 +991,9 @@ func (c *Cursor) Intersect(rowID uint64, row []uint64) error {
 		case ContainerTypeRLE:
 			panic("TODO(BBJ): rbf.Bitmap.Intersect() RLE support")
 		case ContainerTypeBitmap:
-			_, bm, err := cell.GetBitmap(c.tx)
+			_, bm, err := c.tx.GetBitmap(&cell)
 			if err != nil {
-				return err
+				return errors.Wrap(err, "cursor.Intersect")
 			}
 			for i, v := range bm {
 				row[(offset/64)+uint64(i)] &= v
@@ -1107,9 +1117,9 @@ func (c *Cursor) merge(key uint64, data *roaring.Container) (bool, error) {
 		d := toArray16(cell.Data)
 		container = roaring.NewContainerArray(d)
 	case ContainerTypeBitmap:
-		_, d, err := cell.GetBitmap(c.tx)
+		_, d, err := c.tx.GetBitmap(&cell)
 		if err != nil {
-			return false, err
+			return false, errors.Wrap(err, "cursor.merge")
 		}
 		container = roaring.NewContainerBitmap(cell.N, d)
 	case ContainerTypeRLE:
@@ -1194,9 +1204,9 @@ func (c *Cursor) difference(key uint64, data *roaring.Container) (bool, error) {
 		d := toArray16(cell.Data)
 		container = roaring.NewContainerArray(d)
 	case ContainerTypeBitmap:
-		_, d, err := cell.GetBitmap(c.tx)
+		_, d, err := c.tx.GetBitmap(&cell)
 		if err != nil {
-			return false, err
+			return false, errors.Wrap(err, "cursor.difference")
 		}
 		container = roaring.NewContainerBitmap(cell.N, d)
 	case ContainerTypeRLE:
diff --git a/rbf/cursor_test.go b/rbf/cursor_test.go
index b6b8330d9..f5f206e2a 100644
--- a/rbf/cursor_test.go
+++ b/rbf/cursor_test.go
@@ -1048,3 +1048,39 @@ func TestCursor_OneBitmap(t *testing.T) {
 	cur.First()
 	cur.Dump("fun.dot")
 }
+func TestCursor_GenerateAll(t *testing.T) {
+	db := MustOpenDB(t)
+	defer MustCloseDB(t, db)
+	tx := MustBegin(t, db, true)
+	defer MustRollback(t, tx)
+	if err := tx.CreateBitmap("x"); err != nil {
+		t.Fatal(err)
+	}
+	ar := func() *roaring.Bitmap {
+		bm := roaring.NewBitmap()
+		bm.Put(11, roaring.NewContainerArray([]uint16{1, 2, 3}))
+		return bm
+	}()
+	tx.AddRoaring("x", ar)
+	rb := func() *roaring.Bitmap {
+		bm := roaring.NewBitmap()
+		bm.Put(1, roaring.NewContainerRun([]roaring.Interval16{{Start: 1, Last: 12}}))
+		return bm
+	}()
+	tx.AddRoaring("x", rb)
+	bb := func() *roaring.Bitmap {
+		bm := roaring.NewBitmap()
+		bm.Put(0, roaring.NewContainerBitmap(makeBitmap([]uint16{75})))
+		return bm
+	}()
+	tx.AddRoaring("x", bb)
+	if err := tx.CreateBitmap("field/view/"); err != nil {
+		t.Fatal(err)
+	}
+	tx.AddRoaring("field/view/", bb)
+	cur, err := tx.Cursor("field/view/")
+	if err != nil {
+		panic(err)
+	}
+	cur.Dump("fun.dot")
+}
diff --git a/rbf/db.go b/rbf/db.go
index 2df0b31c9..09f4b5aa9 100644
--- a/rbf/db.go
+++ b/rbf/db.go
@@ -207,7 +207,7 @@ func (db *DB) checkpoint() error {
 			continue
 		}
 
-		// Loop over pages in the tranasction.
+		// Loop over pages in the transaction.
 		for ; walID <= metaWALID; walID++ {
 			canCheckpoint := minActiveWALID == 0 || walID <= minActiveWALID
 
diff --git a/rbf/rbf.go b/rbf/rbf.go
index 5e5707d3a..c6b240604 100644
--- a/rbf/rbf.go
+++ b/rbf/rbf.go
@@ -108,7 +108,7 @@ func writeMetaChecksum(page []byte, chksum uint32) {
 
 // Root record page helpers
 
-func readRootRecordOverflowPgno(page []byte) uint32 { return binary.BigEndian.Uint32(page[8:]) }
+func WalkRootRecordPages(page []byte) uint32 { return binary.BigEndian.Uint32(page[8:]) }
 func writeRootRecordOverflowPgno(page []byte, pgno uint32) {
 	binary.BigEndian.PutUint32(page[8:], pgno)
 }
@@ -293,7 +293,7 @@ func (c *leafCell) Bitmap(tx *Tx) []uint64 {
 		}
 		return buf
 	case ContainerTypeBitmap:
-		_, bm, _ := c.GetBitmap(tx)
+		_, bm, _ := tx.GetBitmap(c)
 		return bm
 	default:
 		panic(fmt.Sprintf("invalid container type: %d", c.Type))
@@ -319,7 +319,7 @@ func (c *leafCell) Values(tx *Tx) []uint16 {
 		return a
 	case ContainerTypeBitmap:
 		a := make([]uint16, 0, BitmapN*64)
-		_, bm, _ := c.GetBitmap(tx)
+		_, bm, _ := tx.GetBitmap(c)
 		for i, v := range bm {
 			for j := uint(0); j < 64; j++ {
 				if v&(1<<j) != 0 {
@@ -540,7 +540,7 @@ func Walk(tx *Tx, pgno uint32, v func(uint32, []*RootRecord)) {
 		}
 		v(pgno, a)
 		// Read next overflow page number.
-		pgno = readRootRecordOverflowPgno(page)
+		pgno = WalkRootRecordPages(page)
 	}
 }
 
@@ -604,7 +604,7 @@ func rrdump(tx *Tx, pgno uint32, v func(uint32, []*RootRecord)) {
 		}
 		v(pgno, a)
 		// Read next overflow page number.
-		pgno = readRootRecordOverflowPgno(page)
+		pgno = WalkRootRecordPages(page)
 	}
 }
 
diff --git a/rbf/tx.go b/rbf/tx.go
index ca3cc897e..9320b900a 100644
--- a/rbf/tx.go
+++ b/rbf/tx.go
@@ -230,7 +230,7 @@ func (tx *Tx) rootRecords() ([]*RootRecord, error) {
 		records = append(records, a...)
 
 		// Read next overflow page number.
-		pgno = readRootRecordOverflowPgno(page)
+		pgno = WalkRootRecordPages(page)
 	}
 	return records, nil
 }
@@ -244,8 +244,11 @@ func (tx *Tx) writeRootRecordPages(records []*RootRecord) (err error) {
 			return err
 		}
 
-		tx.deallocate(pgno)
-		pgno = readRootRecordOverflowPgno(page)
+		err = tx.deallocate(pgno)
+		if err != nil {
+			return err
+		}
+		pgno = WalkRootRecordPages(page)
 	}
 
 	// Exit early if no records exist.
@@ -459,7 +462,7 @@ func (tx *Tx) inusePageSet() (map[uint32]struct{}, error) {
 		if err != nil {
 			return nil, err
 		}
-		pgno = readRootRecordOverflowPgno(page)
+		pgno = WalkRootRecordPages(page)
 	}
 
 	// Traverse freelist and mark pages as in-use.
@@ -673,3 +676,11 @@ func (tx *Tx) AddRoaring(name string, bm *roaring.Bitmap) (changed bool, err err
 	}
 	return c.AddRoaring(bm)
 }
+func (tx *Tx) GetBitmap(c *leafCell) (pgno uint32, bm []uint64, err error) {
+	pgno = toPgno(c.Data)
+	page, err := tx.readPage(pgno)
+	if err != nil {
+		return 0, nil, err
+	}
+	return pgno, toArray64(page), err
+}