scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-05 08:10:50 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

debugstats and rbf tooling for enhanced debugging/diagnostics

Browse source
This commit is contained in:
Jason E. Aten 2020-12-08 22:21:01 +00:00
parent 9f7fa7eaad
commit 82d07bc123
12 changed files with 1406 additions and 146 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

142
debugstats/stats.go Normal file
View file

@ -0,0 +1,142 @@
// Copyright 2020 Pilosa Corp.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package debugstats
import (
"fmt"
"math"
//"os"
"runtime"
"sort"
"sync"
"time"
)
type CallStats struct {
// protect elap
mu sync.Mutex
// track how much time each call took.
elap map[string]*elapsed
}
type elapsed struct {
dur []float64
}
func NewCallStats() *CallStats {
w := &CallStats{}
w.Reset()
return w
}
func (w *CallStats) Reset() {
w.mu.Lock()
defer w.mu.Unlock()
w.elap = make(map[string]*elapsed)
}
type LineSorter struct {
Line string
Tot float64
}
type SortByTot []*LineSorter
func (p SortByTot) Len() int {
return len(p)
}
func (p SortByTot) Less(i, j int) bool {
return p[i].Tot < p[j].Tot
}
func (p SortByTot) Swap(i, j int) {
p[i], p[j] = p[j], p[i]
}
func (c *CallStats) Report(title string) (r string) {
//txsrc := os.Getenv("PILOSA_TXSRC")
r = fmt.Sprintf("CallStats: (%v)\n", title)
c.mu.Lock()
defer c.mu.Unlock()
var lines []*LineSorter
for id, elap := range c.elap {
slc := elap.dur
n := len(slc)
if n == 0 {
continue
}
mean, sd, totaltm := computeMeanSd(slc)
if n == 1 {
sd = 0
mean = slc[0]
totaltm = slc[0]
}
line := fmt.Sprintf(" %20v N=%8v avg/op: %12v sd: %12v total: %12v\n", id, n, time.Duration(mean), time.Duration(sd), time.Duration(totaltm))
lines = append(lines, &LineSorter{Line: line, Tot: totaltm})
}
sort.Sort(SortByTot(lines))
for i := range lines {
r += lines[i].Line
}
if false {
var m1 runtime.MemStats
runtime.ReadMemStats(&m1)
r += fmt.Sprintf("\n m1.TotalAlloc = %v\n", m1.TotalAlloc)
}
return
}
var NaN = math.NaN()
func computeMeanSd(slc []float64) (mean, sd, tot float64) {
if len(slc) < 2 {
return NaN, NaN, NaN
}
for _, v := range slc {
tot += v
}
n := float64(len(slc))
mean = tot / n
variance := 0.0
for _, v := range slc {
tmp := (v - mean)
variance += tmp * tmp
}
variance = variance / n // biased, but we don't care b/c we can have very small n
sd = math.Sqrt(variance)
if sd < 1e-8 {
// sd is super close to zero, NaN out the z-score rather than +/- Inf
sd = NaN
}
return
}
func (c *CallStats) Add(k string, dur time.Duration) {
if c == nil {
return
}
c.mu.Lock()
defer c.mu.Unlock()
e, ok := c.elap[k]
if !ok {
e = &elapsed{}
c.elap[k] = e
}
e.dur = append(e.dur, float64(dur))
}

54
debugstats/stats_test.go Normal file
View file

@ -0,0 +1,54 @@
// Copyright 2020 Pilosa Corp.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package debugstats
import (
"fmt"
"testing"
"time"
)
func TestCallStats(t *testing.T) {
callStats := NewCallStats()
for j := 0; j < 4; j++ {
t0 := time.Now()
doOperation0()
callStats.Add("op0", time.Since(t0))
t1 := time.Now()
doOperation1()
callStats.Add("op1", time.Since(t1))
t2 := time.Now()
doOperation2()
callStats.Add("op2", time.Since(t2))
}
fmt.Printf("report = \n%v\n", callStats.Report("test"))
}
func doOperation0() {
time.Sleep(50 * time.Millisecond)
}
func doOperation1() {
time.Sleep(100 * time.Millisecond)
}
func doOperation2() {
time.Sleep(200 * time.Millisecond)
}

205
rbf/cursor.go
View file

@ -38,10 +38,12 @@ type Cursor struct {
leafCells [PageSize / 8]leafCell
// stack holds branches
stack struct {
index int
elems [32]stackElem
}
stack searchStack
}
type searchStack struct {
top int
elems [32]stackElem
}
func runAdd(runs []roaring.Interval16, v uint16) ([]roaring.Interval16, bool) {
@ -139,7 +141,7 @@ func (c *Cursor) Add(v uint64) (changed bool, err error) {
}
// If the container exists and bit is not set then update the page.
elem := &c.stack.elems[c.stack.index]
elem := &c.stack.elems[c.stack.top]
leafPage, _, err := c.tx.readPage(elem.pgno)
if err != nil {
return false, err
@ -210,7 +212,7 @@ func (c *Cursor) Remove(v uint64) (changed bool, err error) {
}
// If the container exists and bit is not set then update the page.
elem := &c.stack.elems[c.stack.index]
elem := &c.stack.elems[c.stack.top]
leafPage, _, err := c.tx.readPage(elem.pgno)
if err != nil {
return false, err
@ -326,7 +328,7 @@ func (c *Cursor) Contains(v uint64) (exists bool, err error) {
}
// If the container exists then check for low bits existence.
elem := &c.stack.elems[c.stack.index]
elem := &c.stack.elems[c.stack.top]
leafPage, _, err := c.tx.readPage(elem.pgno)
if err != nil {
return false, err
@ -368,7 +370,7 @@ func toPgno(val []byte) uint32 {
}
func (c *Cursor) putLeafCell(in leafCell) (err error) {
elem := &c.stack.elems[c.stack.index]
elem := &c.stack.elems[c.stack.top]
leafPage, isHeap, err := c.tx.readPage(elem.pgno) // the last read leaf page
if err != nil {
return err
@ -453,18 +455,18 @@ func (c *Cursor) putLeafCell(in leafCell) (err error) {
}
// Write each group to a separate page.
newRoot := (len(groups) > 1) && (c.stack.index == 0)
newRoot := (len(groups) > 1) && (c.stack.top == 0)
var parents []branchCell
origPgno := elem.pgno
// newRoot if split occured and bottom of the stack
for i, group := range groups {
// First page should overwrite the original.
// Subsequent pages should allocate new pages.
parent := branchCell{Key: group[0].Key} //<<< this is the key spot for making sure that key is correct
parent := branchCell{LeftKey: group[0].Key} //<<< this is the key spot for making sure that key is correct
if i == 0 && !newRoot {
parent.Pgno = origPgno
parent.ChildPgno = origPgno
} else {
if parent.Pgno, err = c.tx.allocatePgno(); err != nil {
if parent.ChildPgno, err = c.tx.allocatePgno(); err != nil {
return fmt.Errorf("cannot allocate leaf: %w", err)
}
}
@ -479,7 +481,7 @@ func (c *Cursor) putLeafCell(in leafCell) (err error) {
}
var buf [PageSize]byte
// Write cells to page.
writePageNo(buf[:], parent.Pgno)
writePageNo(buf[:], parent.ChildPgno)
writeFlags(buf[:], PageTypeLeaf)
writeCellN(buf[:], len(group))
@ -509,20 +511,20 @@ func (c *Cursor) putLeafCell(in leafCell) (err error) {
}
// Initialize a new root if we are currently the root page.
if c.stack.index == 0 {
if c.stack.top == 0 {
assert(newRoot) // leaf write must be root when stack at root
return c.writeRoot(origPgno, parents)
}
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)
return c.putBranchCells(c.stack.top-1, parents)
}
// putLeafCellFast quickly insert or updates a cell on a leaf page.
// It works by shifting bytes around instead of deserializing. This must not overflow.
func (c *Cursor) putLeafCellFast(in leafCell, isInsert bool) (err error) {
elem := &c.stack.elems[c.stack.index]
elem := &c.stack.elems[c.stack.top]
src, isHeap, err := c.tx.readPage(elem.pgno)
if err != nil {
return err
@ -597,7 +599,7 @@ func (c *Cursor) putLeafCellFast(in leafCell, isInsert bool) (err error) {
// deleteLeafCell removes a cell from the currently positioned page & index.
func (c *Cursor) deleteLeafCell(key uint64) (err error) {
elem := &c.stack.elems[c.stack.index]
elem := &c.stack.elems[c.stack.top]
leafPage, _, err := c.tx.readPage(elem.pgno)
if err != nil {
return err
@ -613,11 +615,11 @@ func (c *Cursor) deleteLeafCell(key uint64) (err error) {
}
// If no more cells exist and we have a parent, remove from parent.
if c.stack.index > 0 && len(cells) == 1 {
if c.stack.top > 0 && len(cells) == 1 {
if err := c.tx.freePgno(elem.pgno); err != nil {
return err
}
return c.deleteBranchCell(c.stack.index-1, cells[0].Key)
return c.deleteBranchCell(c.stack.top-1, cells[0].Key)
}
// Remove matching cell from list.
@ -641,8 +643,8 @@ func (c *Cursor) deleteLeafCell(key uint64) (err error) {
}
// Update the parent's reference key if it's changed.
if c.stack.index > 0 && oldPageKey != cells[0].Key {
return c.updateBranchCell(c.stack.index-1, cells[0].Key)
if c.stack.top > 0 && oldPageKey != cells[0].Key {
return c.updateBranchCell(c.stack.top-1, cells[0].Key)
}
return nil
}
@ -659,6 +661,10 @@ func (c *Cursor) putBranchCells(stackIndex int, newCells []branchCell) (err erro
cells := readBranchCells(page)
if len(cells) == 0 {
cells = make([]branchCell, 1)
}
// Update current cell & insert additional cells after it.
cells[elem.index] = newCells[0]
if len(newCells) > 1 {
@ -680,11 +686,11 @@ func (c *Cursor) putBranchCells(stackIndex int, newCells []branchCell) (err erro
for i, group := range groups {
// First page should overwrite the original.
// Subsequent pages should allocate new pages.
parent := branchCell{Key: group[0].Key}
parent := branchCell{LeftKey: group[0].LeftKey}
if i == 0 && !newRoot {
parent.Pgno = origPgno
parent.ChildPgno = origPgno
} else {
if parent.Pgno, err = c.tx.allocatePgno(); err != nil {
if parent.ChildPgno, err = c.tx.allocatePgno(); err != nil {
return fmt.Errorf("cannot allocate branch: %w", err)
}
}
@ -692,7 +698,7 @@ func (c *Cursor) putBranchCells(stackIndex int, newCells []branchCell) (err erro
// Write cells to page.
var buf [PageSize]byte
writePageNo(buf[:], parents[i].Pgno)
writePageNo(buf[:], parents[i].ChildPgno)
writeFlags(buf[:], PageTypeBranch)
writeCellN(buf[:], len(group))
@ -737,10 +743,10 @@ func (c *Cursor) updateBranchCell(stackIndex int, newKey uint64) (err error) {
return err
}
cells := readBranchCells(page)
oldPageKey := cells[0].Key
oldPageKey := cells[0].LeftKey
// Update key in branch cell.
cells[elem.index].Key = newKey
cells[elem.index].LeftKey = newKey
// Write cells to page.
var buf [PageSize]byte
@ -757,8 +763,8 @@ func (c *Cursor) updateBranchCell(stackIndex int, newKey uint64) (err error) {
return err
}
if stackIndex > 0 && oldPageKey != cells[0].Key {
return c.updateBranchCell(stackIndex-1, cells[0].Key)
if stackIndex > 0 && oldPageKey != cells[0].LeftKey {
return c.updateBranchCell(stackIndex-1, cells[0].LeftKey)
}
return nil
}
@ -773,7 +779,7 @@ func (c *Cursor) deleteBranchCell(stackIndex int, key uint64) (err error) {
return err
}
cells := readBranchCells(page)
oldPageKey := cells[0].Key
oldPageKey := cells[0].LeftKey
// Remove cell from branch.
copy(cells[elem.index:], cells[elem.index+1:])
@ -782,7 +788,7 @@ func (c *Cursor) deleteBranchCell(stackIndex int, key uint64) (err error) {
// If the root only has one node, replace it with its child.
if stackIndex == 0 && len(cells) == 1 {
target, _, err := c.tx.readPage(cells[0].Pgno)
target, _, err := c.tx.readPage(cells[0].ChildPgno)
if err != nil {
return err
}
@ -791,7 +797,7 @@ func (c *Cursor) deleteBranchCell(stackIndex int, key uint64) (err error) {
copy(buf, target)
writePageNo(buf[:], elem.pgno)
if err := c.tx.freePgno(cells[0].Pgno); err != nil {
if err := c.tx.freePgno(cells[0].ChildPgno); err != nil {
return err
}
return c.tx.writePage(buf[:])
@ -812,8 +818,8 @@ func (c *Cursor) deleteBranchCell(stackIndex int, key uint64) (err error) {
return err
}
if stackIndex > 0 && len(cells) > 0 && oldPageKey != cells[0].Key {
return c.updateBranchCell(stackIndex-1, cells[0].Key)
if stackIndex > 0 && len(cells) > 0 && oldPageKey != cells[0].LeftKey {
return c.updateBranchCell(stackIndex-1, cells[0].LeftKey)
}
return nil
}
@ -849,7 +855,8 @@ func splitLeafCells(cells []leafCell) [][]leafCell {
// If there is at least one cell on the slice & we've exceeded
// half a page then create a new group of cells.
if cellN != 0 && (dataOffset(cellN+1)+dataSize+sz) > (PageSize*60)/100 {
thresh := int(float64(PageSize) * globalBranchFillPct)
if cellN != 0 && (dataOffset(cellN+1)+dataSize+sz) > thresh {
slices, dataSize = append(slices, nil), 0
} else if cellN != 0 && cell.Type == ContainerTypeArray && cell.ElemN > ArrayMaxSize {
slices, dataSize = append(slices, nil), 0
@ -864,6 +871,8 @@ func splitLeafCells(cells []leafCell) [][]leafCell {
return slices
}
var globalBranchFillPct = 0.60
// splitBranchCells splits cells into roughly equal parts. It's a naive
// implementation that splits cells whenever a page is 60% full.
func splitBranchCells(cells []branchCell) [][]branchCell {
@ -877,7 +886,9 @@ func splitBranchCells(cells []branchCell) [][]branchCell {
// If there is at least one cell on the slice & we've exceeded
// half a page then create a new group of cells.
if cellN != 0 && (dataOffset(cellN+1)+dataSize+sz) > (PageSize*60)/100 {
thresh := int(float64(PageSize) * globalBranchFillPct)
if cellN != 0 && (dataOffset(cellN+1)+dataSize+sz) > thresh {
slices, dataSize = append(slices, nil), 0
}
@ -899,8 +910,8 @@ func pageKeyAt(page []byte, index int) uint64 {
func (c *Cursor) First() error {
c.buffered = true
for c.stack.index = 0; ; c.stack.index++ {
elem := &c.stack.elems[c.stack.index]
for c.stack.top = 0; ; c.stack.top++ {
elem := &c.stack.elems[c.stack.top]
buf, _, err := c.tx.readPage(elem.pgno)
if err != nil {
@ -914,9 +925,9 @@ func (c *Cursor) First() error {
// Read cell pgno into the next stack level.
cell := readBranchCell(buf, elem.index)
c.stack.elems[c.stack.index+1] = stackElem{
pgno: cell.Pgno,
key: cell.Key,
c.stack.elems[c.stack.top+1] = stackElem{
pgno: cell.ChildPgno,
key: cell.LeftKey,
}
case PageTypeLeaf:
@ -936,8 +947,8 @@ func (c *Cursor) Last() error {
// c.stack.elems[0].pgno = c.root
c.buffered = true
for c.stack.index = 0; ; c.stack.index++ {
elem := &c.stack.elems[c.stack.index]
for c.stack.top = 0; ; c.stack.top++ {
elem := &c.stack.elems[c.stack.top]
buf, _, err := c.tx.readPage(elem.pgno)
if err != nil {
@ -950,9 +961,9 @@ func (c *Cursor) Last() error {
// Read cell pgno into the next stack level.
cell := readBranchCell(buf, elem.index)
c.stack.elems[c.stack.index+1] = stackElem{
pgno: cell.Pgno,
key: cell.Key,
c.stack.elems[c.stack.top+1] = stackElem{
pgno: cell.ChildPgno,
key: cell.LeftKey,
}
case PageTypeLeaf:
@ -972,8 +983,8 @@ func (c *Cursor) Last() error {
func (c *Cursor) Seek(key uint64) (exact bool, err error) {
// c.stack.elems[0].pgno = c.bitmap.root
c.buffered = true
for c.stack.index = 0; ; c.stack.index++ {
elem := &c.stack.elems[c.stack.index]
for c.stack.top = 0; ; c.stack.top++ {
elem := &c.stack.elems[c.stack.top]
assert(elem.pgno != 0) // cursor should never point to page zero (meta)
buf, _, err := c.tx.readPage(elem.pgno)
@ -1002,9 +1013,9 @@ func (c *Cursor) Seek(key uint64) (exact bool, err error) {
cell := readBranchCell(buf, elem.index)
c.stack.elems[c.stack.index+1] = stackElem{
pgno: cell.Pgno,
key: cell.Key,
c.stack.elems[c.stack.top+1] = stackElem{
pgno: cell.ChildPgno,
key: cell.LeftKey,
}
case PageTypeLeaf:
@ -1028,7 +1039,7 @@ func (c *Cursor) Seek(key uint64) (exact bool, err error) {
// Next moves to the next element of the btree. Returns EOF if no more elements exist.
func (c *Cursor) Next() error {
elem := &c.stack.elems[c.stack.index]
elem := &c.stack.elems[c.stack.top]
leafPage, _, err := c.tx.readPage(elem.pgno)
if err != nil {
return err
@ -1060,14 +1071,14 @@ func (c *Cursor) Prev() error {
}
// Move forward to the next leaf element if available.
if elem := &c.stack.elems[c.stack.index]; elem.index > 0 {
if elem := &c.stack.elems[c.stack.top]; elem.index > 0 {
elem.index--
return nil
}
// Move up the stack until we can move forward one element.
for c.stack.index--; c.stack.index >= 0; c.stack.index-- {
elem := &c.stack.elems[c.stack.index]
for c.stack.top--; c.stack.top >= 0; c.stack.top-- {
elem := &c.stack.elems[c.stack.top]
if elem.index > 0 {
elem.index--
break
@ -1075,14 +1086,14 @@ func (c *Cursor) Prev() error {
}
// No more elements, return EOF.
if c.stack.index == -1 {
c.stack.index = 0
if c.stack.top == -1 {
c.stack.top = 0
return io.EOF
}
// Traverse back down the stack to find the first element in each page.
for ; ; c.stack.index++ {
elem := &c.stack.elems[c.stack.index]
for ; ; c.stack.top++ {
elem := &c.stack.elems[c.stack.top]
buf, _, err := c.tx.readPage(elem.pgno)
if err != nil {
@ -1093,9 +1104,9 @@ func (c *Cursor) Prev() error {
case PageTypeBranch:
cell := readBranchCell(buf, elem.index)
c.stack.elems[c.stack.index+1] = stackElem{
pgno: cell.Pgno,
key: cell.Key,
c.stack.elems[c.stack.top+1] = stackElem{
pgno: cell.ChildPgno,
key: cell.LeftKey,
}
case PageTypeLeaf:
@ -1109,7 +1120,7 @@ func (c *Cursor) Prev() error {
// Key returns the key for the container the cursor is currently pointing to.
func (c *Cursor) Key() uint64 {
elem := &c.stack.elems[c.stack.index]
elem := &c.stack.elems[c.stack.top]
leafPage, _, _ := c.tx.readPage(elem.pgno)
if readCellN(leafPage[:]) == 0 {
return 0
@ -1120,7 +1131,7 @@ func (c *Cursor) Key() uint64 {
// 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]
elem := &c.stack.elems[c.stack.top]
leafPage, _, _ := c.tx.readPage(elem.pgno)
if readCellN(leafPage[:]) == 0 {
return nil
@ -1136,9 +1147,36 @@ type stackElem struct {
key uint64 // element key
}
func (se *stackElem) equal(se2 *stackElem) bool {
if se.pgno != se2.pgno {
return false
}
if se.index != se2.index {
return false
}
if se.key != se2.key {
return false
}
return true
}
func (se *stackElem) String() string {
return fmt.Sprintf("stackElem{pgno:%v index:%v key:%v}", int(se.pgno) /*,tx.pageTypeDesc(se.pgno)*/, se.index, int(se.key))
}
func (se *stackElem) clear() {
se.pgno = 0
se.index = 0
se.key = 0
}
var _ = (&stackElem{}).clear
var _ = (&stackElem{}).String
var _ = (&stackElem{}).equal
func (c *Cursor) goNextPage() error {
for c.stack.index--; c.stack.index >= 0; c.stack.index-- {
elem := &c.stack.elems[c.stack.index]
for c.stack.top--; c.stack.top >= 0; c.stack.top-- {
elem := &c.stack.elems[c.stack.top]
if buf, _, err := c.tx.readPage(elem.pgno); err != nil {
return err
} else if n := readCellN(buf); elem.index+1 < n {
@ -1148,14 +1186,14 @@ func (c *Cursor) goNextPage() error {
}
// No more elements, return EOF.
if c.stack.index == -1 {
c.stack.index = 0
if c.stack.top == -1 {
c.stack.top = 0
return io.EOF
}
// Traverse back down the stack to find the first element in each page.
for ; ; c.stack.index++ {
elem := &c.stack.elems[c.stack.index]
for ; ; c.stack.top++ {
elem := &c.stack.elems[c.stack.top]
buf, _, err := c.tx.readPage(elem.pgno)
if err != nil {
return err
@ -1164,9 +1202,9 @@ func (c *Cursor) goNextPage() error {
switch typ := readFlags(buf); typ {
case PageTypeBranch:
cell := readBranchCell(buf, elem.index)
c.stack.elems[c.stack.index+1] = stackElem{
pgno: cell.Pgno,
key: cell.Key,
c.stack.elems[c.stack.top+1] = stackElem{
pgno: cell.ChildPgno,
key: cell.LeftKey,
}
case PageTypeLeaf:
elem.index = 0
@ -1222,7 +1260,7 @@ func ConvertToLeafArgs(key uint64, c *roaring.Container) (result leafCell) {
}
func (c *Cursor) merge(key uint64, data *roaring.Container) (bool, error) {
elem := &c.stack.elems[c.stack.index]
elem := &c.stack.elems[c.stack.top]
leafPage, _, err := c.tx.readPage(elem.pgno)
if err != nil {
return false, err
@ -1311,7 +1349,7 @@ func (c *Cursor) RemoveRoaring(bm *roaring.Bitmap) (changed bool, err error) {
}
func (c *Cursor) difference(key uint64, data *roaring.Container) (bool, error) {
elem := &c.stack.elems[c.stack.index]
elem := &c.stack.elems[c.stack.top]
leafPage, _, err := c.tx.readPage(elem.pgno)
if err != nil {
return false, err
@ -1365,3 +1403,20 @@ func (c *Cursor) Close() {
case <-tx.db.cursorCleaner.ReqStop.Chan:
}
}
func keysFromParents(parents []branchCell) (ckeys []int) {
for _, par := range parents {
ckeys = append(ckeys, int(par.LeftKey))
}
return
}
var _ = (&Cursor{}).showCursorStack
func (c *Cursor) showCursorStack() (r string) {
r = fmt.Sprintf("top = %v\n", c.stack.top)
for i := 0; i <= c.stack.top; i++ {
r += fmt.Sprintf(" [%02v] %v\n", i, c.stack.elems[i].String())
}
return
}

605
rbf/cursor_internal_test.go Normal file
View file

@ -0,0 +1,605 @@
// Copyright 2017 Pilosa Corp.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package rbf
import (
"bytes"
"fmt"
"testing"
"github.com/pilosa/pilosa/v2/roaring"
)
func getRoaringIter(bitsToSet ...uint64) roaring.RoaringIterator {
b := roaring.NewBitmap()
changed := b.DirectAddN(bitsToSet...)
n := len(bitsToSet)
if changed != n {
panic(fmt.Sprintf("changed=%v but bitsToSet len = %v", changed, n))
}
buf := bytes.NewBuffer(make([]byte, 0, 100000))
_, err := b.WriteTo(buf)
if err != nil {
panic(err)
}
itr, err := roaring.NewRoaringIterator(buf.Bytes())
panicOn(err)
return itr
}
func TestCursor_RoaringImport(t *testing.T) {
itr := getRoaringIter([]uint64{1}...)
db := testHelperMustOpenNewDB(t)
defer MustCloseDB(t, db)
index := "i"
field := "f"
view := "v"
shard := uint64(0)
name := rbfName(index, field, view, shard)
clear := false
rowSize := uint64(0)
tx := MustBegin(t, db, true)
defer tx.Rollback()
changed, rowSet, err := tx.ImportRoaringBits(name, itr, clear, false, rowSize, nil)
panicOn(err) // writeTheTailOfLeafCellsFromIter has to be AFTER any pre-existing data. ckey=0 was found already in db.
_ = rowSet
if changed != 1 {
t.Fatalf("expected 1 changed, got %v", changed)
}
if false {
cur, err := tx.cursor(name)
panicOn(err)
cur.dump()
_ = cur.tx.dumpAllPages(true)
}
}
func TestCursor_RoaringImport_clear_bits(t *testing.T) {
itr := getRoaringIter([]uint64{1}...)
db := testHelperMustOpenNewDB(t)
defer MustCloseDB(t, db)
index := "i"
field := "f"
view := "v"
shard := uint64(0)
name := rbfName(index, field, view, shard)
clear := false
rowSize := uint64(0)
tx := MustBegin(t, db, true)
defer tx.Rollback()
changed, rowSet, err := tx.ImportRoaringBits(name, itr, clear, false, rowSize, nil)
panicOn(err)
_ = rowSet
if changed != 1 {
t.Fatalf("expected 1 changed, got %v", changed)
}
// now clear
clear = true
itr2 := getRoaringIter([]uint64{1}...)
changed, rowSet, err = tx.ImportRoaringBits(name, itr2, clear, false, rowSize, nil)
panicOn(err)
_ = rowSet
if changed != 1 {
t.Fatalf("expected 1 changed on clear true, got %v", changed)
}
if false {
cur, err := tx.cursor(name)
panicOn(err)
cur.dump()
_ = cur.tx.dumpAllPages(true)
}
}
func TestCursor_RoaringImport_two_leaves(t *testing.T) {
// make enough for 2 leaves, so then we'll have
// to make a branch too.
want := make([]uint64, 0, ArrayMaxSize)
for x := uint64(0); x < 6000; x += 2 {
want = append(want, x)
}
for x := uint64(0); x < 6000; x += 2 {
want = append(want, x+ShardWidth)
}
itr := getRoaringIter(want...)
db := testHelperMustOpenNewDB(t)
defer MustCloseDB(t, db)
index := "i"
field := "f"
view := "v"
shard := uint64(0)
name := rbfName(index, field, view, shard)
clear := false
rowSize := uint64(0)
tx := MustBegin(t, db, true)
defer tx.Rollback()
changed, rowSet, err := tx.ImportRoaringBits(name, itr, clear, false, rowSize, nil)
panicOn(err) // writeTheTailOfLeafCellsFromIter has to be AFTER any pre-existing data. ckey=0 was found already in db.
_ = rowSet
if changed != 6000 {
t.Fatalf("expected 6000 bits changed, got %v", changed)
}
if false {
cur, err := tx.cursor(name)
panicOn(err)
cur.dump()
_ = cur.tx.dumpAllPages(true)
}
}
func TestCursor_RoaringImport_many_leaves_manual_split(t *testing.T) {
// does the right split:
// make enough for so many leaves that we have to
// make a branch too. so 513 or more cells, because
// 512 16-byte branchCells should
//biggerFactor := 2
NbranchCells := int(maxBranchCellsPerPage) + 1 // * biggerFactor
want := make([]uint64, 0, ArrayMaxSize)
expectedBitsChanged := 0
m := make(map[int]bool)
for i := 0; i < NbranchCells; i++ {
for x := 0; x < 6000; x += 2 {
rowID := i
columnID := x
value := (rowID * ShardWidth) + (columnID % ShardWidth)
want = append(want, uint64(value)) // x+uint64(i)*ShardWidth)
m[value] = true
expectedBitsChanged++
}
}
itr := getRoaringIter(want[:len(want)-3000]...)
itr2 := getRoaringIter(want[len(want)-3000:]...)
db := testHelperMustOpenNewDB(t)
defer MustCloseDB(t, db)
index := "i"
field := "f"
view := "v"
shard := uint64(0)
name := rbfName(index, field, view, shard)
clear := false
rowSize := uint64(0)
tx := MustBegin(t, db, true)
defer tx.Rollback()
//vv("DONE WITH Add()")
changed, _, err := tx.ImportRoaringBits(name, itr, clear, false, rowSize, nil)
panicOn(err) // writeTheTailOfLeafCellsFromIter has to be AFTER any pre-existing data. ckey=0 was found already in db.
if changed != expectedBitsChanged-3000 {
t.Fatalf("expected %v bits changed, got %v", expectedBitsChanged-3000, changed)
}
//vv("changed on set is %v", changed)
//vv("about to do itr2, that starts with key %v", itr2.ContainerKeys()[0])
changed, _, err = tx.ImportRoaringBits(name, itr2, clear, false, rowSize, nil)
panicOn(err)
if changed != 3000 {
t.Fatalf("expected %v bits changed, got %v", 3000, changed)
}
//vv("done with itr2")
dump := func() {
cur, err := tx.cursor(name)
panicOn(err)
//cur.dump()
_ = cur.tx.dumpAllPages(true)
}
_ = dump
//dump()
//vv("now clear")
// now clear
clear = true
//itr3 := getRoaringIter(want[len(want)-3000:]...)
itr3 := getRoaringIter(want...)
changed, _, err = tx.ImportRoaringBits(name, itr3, clear, false, rowSize, nil)
panicOn(err)
if changed != expectedBitsChanged {
// cursor_internal_test.go:235: expected 2,724,000 bits changed, got 2,721,000
t.Fatalf("expected %v bits changed, got %v", expectedBitsChanged, changed)
}
//dump()
}
func TestCursor_RoaringImport_auto_many_leaves(t *testing.T) {
// make enough for so many leaves that we have to
// make a branch too. so 513 or more cells, because
// 512 16-byte branchCells should
//biggerFactor := 2
NbranchCells := int(maxBranchCellsPerPage) + 1 // * biggerFactor
want := make([]uint64, 0, ArrayMaxSize)
expectedBitsChanged := 0
m := make(map[int]bool)
for i := 0; i < NbranchCells; i++ {
for x := 0; x < 6000; x += 2 {
rowID := i
columnID := x
value := (rowID * ShardWidth) + (columnID % ShardWidth)
want = append(want, uint64(value)) // x+uint64(i)*ShardWidth)
m[value] = true
expectedBitsChanged++
}
}
itr := getRoaringIter(want...)
db := testHelperMustOpenNewDB(t)
defer MustCloseDB(t, db)
index := "i"
field := "f"
view := "v"
shard := uint64(0)
name := rbfName(index, field, view, shard)
clear := false
rowSize := uint64(0)
tx := MustBegin(t, db, true)
defer tx.Rollback()
//vv("DONE WITH Add()")
changed, _, err := tx.ImportRoaringBits(name, itr, clear, false, rowSize, nil)
panicOn(err) // writeTheTailOfLeafCellsFromIter has to be AFTER any pre-existing data. ckey=0 was found already in db.
if changed != expectedBitsChanged {
t.Fatalf("expected %v bits changed, got %v", expectedBitsChanged, changed)
}
//vv("changed on set is %v", changed)
dump := func() {
cur, err := tx.cursor(name)
panicOn(err)
//cur.dump()
_ = cur.tx.dumpAllPages(true)
}
_ = dump
//dump()
//vv("now clear")
// now clear
clear = true
itr2 := getRoaringIter(want...)
changed, _, err = tx.ImportRoaringBits(name, itr2, clear, false, rowSize, nil)
panicOn(err)
if changed != expectedBitsChanged {
t.Fatalf("expected %v bits changed, got %v", expectedBitsChanged, changed)
}
//dump()
}
func TestCursor_putBranchCellsHandlesLotsOfNewBranchesAtTheRoot(t *testing.T) {
const maxBranchCells = 2
prev := globalBranchFillPct
defer func() {
globalBranchFillPct = prev
}()
// force there to be so many branch cells that the root cannot handle
// them without making extra branch levels.
globalBranchFillPct = float64(maxBranchCells+1) / float64(maxBranchCellsPerPage)
biggerFactor := 3 //4 // int(maxBranchCellsPerPage) + 1
_ = biggerFactor
NbranchCells := 10 // (int(maxBranchCellsPerPage) + 1) * biggerFactor
want := make([]uint64, 0, ArrayMaxSize)
expectedBitsChanged := 0
m := make(map[int]bool)
for i := 0; i < NbranchCells; i++ {
//if i%10000 == 0 {
//vv("i = %v, NbranchCells = %v", i, NbranchCells)
//}
for x := 0; x < 6000; x += 2 {
rowID := i
columnID := x
value := (rowID * ShardWidth) + (columnID % ShardWidth)
want = append(want, uint64(value)) // x+uint64(i)*ShardWidth)
m[value] = true
expectedBitsChanged++
}
}
itr := getRoaringIter(want...)
db := testHelperMustOpenNewDB(t)
defer MustCloseDB(t, db)
index := "i"
field := "f"
view := "v"
shard := uint64(0)
name := rbfName(index, field, view, shard)
tx := MustBegin(t, db, true)
defer tx.Rollback()
if err := tx.CreateBitmap(name); err != nil {
t.Fatal(err)
}
c, err := tx.cursor(name)
panicOn(err)
ikeys := itr.ContainerKeys()
for _, ckey := range ikeys {
_, err := c.Seek(ckey)
panicOn(err)
break // after the first seek
}
var leafcells []leafCell
for ckey, ct := itr.NextContainer(); ct != nil; ckey, ct = itr.NextContainer() {
newN := int(ct.N())
if newN == 0 {
continue
}
lc := ConvertToLeafArgs(ckey, ct)
leafcells = append(leafcells, lc)
} // end ckey loop
// INVAR: leafcells is ready to go
groups := splitLeafCells(leafcells)
var branches []branchCell
for i, group := range groups {
_ = i
// First page should overwrite the original.
// Subsequent pages should allocate new pages.
branch := branchCell{LeftKey: group[0].Key}
branch.ChildPgno, err = c.tx.allocatePgno()
panicOn(err)
branches = append(branches, branch)
//vv("on group i=%v of %v, group[0].Key = %v; branch.ChildPgno=%v; branch.Key=%v", i, len(groups.slc), int(group[0].Key), (branch.ChildPgno), int(branch.Key))
var buf [PageSize]byte
// Write child page.
writePageNo(buf[:], branch.ChildPgno)
writeFlags(buf[:], PageTypeLeaf)
writeCellN(buf[:], len(group))
offset := dataOffset(len(group))
for j, cell := range group {
writeLeafCell(buf[:], j, offset, cell)
offset += align8(cell.Size())
}
err = c.tx.writePage(buf[:])
panicOn(err)
}
//vv("branches ckeys = '%#v'", keysFromParents(branches))
err = c.putBranchCells(0, branches)
panicOn(err)
//c.tx.dumpAllPages(true)
}
func TestCursor_incrementally_add_pages_and_view_them(t *testing.T) {
const maxBranchCells = 2
prev := globalBranchFillPct
defer func() {
globalBranchFillPct = prev
}()
// force there to be so many branch cells that the root cannot handle
// them without making extra branch levels.
globalBranchFillPct = float64(maxBranchCells+1) / float64(maxBranchCellsPerPage)
NbranchCells := 10
want := make([]uint64, 0, ArrayMaxSize)
expectedBitsChanged := 0
m := make(map[int]bool)
for i := 0; i < NbranchCells; i++ {
for x := 0; x < 6000; x += 2 {
rowID := i
columnID := x
value := (rowID * ShardWidth) + (columnID % ShardWidth)
want = append(want, uint64(value)) // x+uint64(i)*ShardWidth)
m[value] = true
expectedBitsChanged++
}
}
db := testHelperMustOpenNewDB(t)
defer MustCloseDB(t, db)
index := "i"
field := "f"
view := "v"
shard := uint64(0)
name := rbfName(index, field, view, shard)
clear := false
rowSize := uint64(0)
tx := MustBegin(t, db, true)
defer tx.Rollback()
dump := func() {
cur, err := tx.cursor(name)
panicOn(err)
//cur.dump()
_ = cur.tx.dumpAllPages(true)
}
_ = dump
for i := 0; i < NbranchCells; i++ {
itr := getRoaringIter(want[i*3000 : (i+1)*3000]...)
changed, _, err := tx.ImportRoaringBits(name, itr, clear, false, rowSize, nil)
panicOn(err) // writeTheTailOfLeafCellsFromIter has to be AFTER any pre-existing data. ckey=0 was found already in db.
if changed != 3000 {
t.Fatalf("expected %v bits changed, got %v", 3000, changed)
}
//vv("changed on set is %v", changed)
//dump()
}
//vv("now clear")
// now clear
clear = true
for i := 0; i < NbranchCells; i++ {
itr := getRoaringIter(want[i*3000 : (i+1)*3000]...)
changed, _, err := tx.ImportRoaringBits(name, itr, clear, false, rowSize, nil)
panicOn(err) // writeTheTailOfLeafCellsFromIter has to be AFTER any pre-existing data. ckey=0 was found already in db.
if changed != 3000 {
t.Fatalf("expected %v bits changed, got %v", 3000, changed)
}
//vv("changed on set is %v", changed)
//dump()
}
}
// Useful for understanding the split patterns. This
// is how the README.md split pattern docs were obtained.
func TestCursor_from_B_to_C(t *testing.T) {
const maxBranchCells = 2
prev := globalBranchFillPct
defer func() {
globalBranchFillPct = prev
}()
// force there to be so many branch cells that the root cannot handle
// them without making extra branch levels.
globalBranchFillPct = float64(maxBranchCells+1) / float64(maxBranchCellsPerPage)
NbranchCells := 3
want := make([]uint64, 0, ArrayMaxSize)
expectedBitsChanged := 0
m := make(map[int]bool)
for i := 0; i < NbranchCells; i++ {
for x := 0; x < 6000; x += 2 {
rowID := i
columnID := x
value := (rowID * ShardWidth) + (columnID % ShardWidth)
want = append(want, uint64(value)) // x+uint64(i)*ShardWidth)
m[value] = true
expectedBitsChanged++
}
}
db := testHelperMustOpenNewDB(t)
defer MustCloseDB(t, db)
index := "i"
field := "f"
view := "v"
shard := uint64(0)
name := rbfName(index, field, view, shard)
clear := false
rowSize := uint64(0)
tx := MustBegin(t, db, true)
defer tx.Rollback()
dump := func() {
cur, err := tx.cursor(name)
panicOn(err)
//cur.dump()
_ = cur.tx.dumpAllPages(true)
}
_ = dump
itr := getRoaringIter(want[:6000]...)
changed, _, err := tx.ImportRoaringBits(name, itr, clear, false, rowSize, nil)
panicOn(err) // writeTheTailOfLeafCellsFromIter has to be AFTER any pre-existing data. ckey=0 was found already in db.
if changed != 6000 {
t.Fatalf("expected %v bits changed, got %v", 6000, changed)
}
//vv("changed on set is %v", changed)
//dump()
//vv("STARTING TO ADD C")
itr = getRoaringIter(want[6000:9000]...)
changed, _, err = tx.ImportRoaringBits(name, itr, clear, false, rowSize, nil)
panicOn(err) // writeTheTailOfLeafCellsFromIter has to be AFTER any pre-existing data. ckey=0 was found already in db.
if changed != 3000 {
t.Fatalf("expected %v bits changed, got %v", 3000, changed)
}
//vv("changed on set is %v", changed)
//dump()
//vv("now clear")
// now clear
clear = true
for i := 0; i < NbranchCells; i++ {
itr := getRoaringIter(want[i*3000 : (i+1)*3000]...)
changed, _, err := tx.ImportRoaringBits(name, itr, clear, false, rowSize, nil)
panicOn(err) // writeTheTailOfLeafCellsFromIter has to be AFTER any pre-existing data. ckey=0 was found already in db.
if changed != 3000 {
t.Fatalf("expected %v bits changed, got %v", 3000, changed) // failing here got 0
}
//vv("changed on clear is %v", changed)
//dump()
}
}

16
rbf/cursorx.go
View file

@ -62,7 +62,7 @@ func (c *Cursor) Rows() ([]uint64, error) {
break
}
elem := &c.stack.elems[c.stack.index]
elem := &c.stack.elems[c.stack.top]
leafPage, _, err := c.tx.readPage(elem.pgno)
if err != nil {
return nil, err
@ -99,7 +99,7 @@ func (c *Cursor) DumpKeys() {
if err == io.EOF {
break
}
elem := &c.stack.elems[c.stack.index]
elem := &c.stack.elems[c.stack.top]
leafPage, _, err := c.tx.readPage(elem.pgno)
if err != nil {
return
@ -110,7 +110,7 @@ func (c *Cursor) DumpKeys() {
}
func (c *Cursor) DumpStack() {
fmt.Println("STACK")
for i := c.stack.index; i >= 0; i-- {
for i := c.stack.top; i >= 0; i-- {
fmt.Printf("%+v\n", c.stack.elems[i])
}
fmt.Println()
@ -134,13 +134,13 @@ func (c *Cursor) Row(shard, rowID uint64) (*roaring.Bitmap, error) {
offset := uint64(shard * ShardWidth)
off := highbits(offset)
hi0, hi1 := highbits(base), highbits((rowID+1)*ShardWidth)
c.stack.index = 0
c.stack.top = 0
ok, err := c.Seek(hi0)
if err != nil {
return nil, err
}
if !ok {
elem := &c.stack.elems[c.stack.index]
elem := &c.stack.elems[c.stack.top]
leafPage, _, err := c.tx.readPage(elem.pgno)
if err != nil {
return nil, err
@ -161,7 +161,7 @@ func (c *Cursor) Row(shard, rowID uint64) (*roaring.Bitmap, error) {
return nil, err
}
elem := &c.stack.elems[c.stack.index]
elem := &c.stack.elems[c.stack.top]
leafPage, _, err := c.tx.readPage(elem.pgno)
if err != nil {
return nil, err
@ -178,7 +178,7 @@ func (c *Cursor) Row(shard, rowID uint64) (*roaring.Bitmap, error) {
// CurrentPageType returns the type of the container currently pointed to by cursor used in testing
// sometimes the cursor needs to be positions prior to this call with First/Last etc.
func (c *Cursor) CurrentPageType() ContainerType {
elem := &c.stack.elems[c.stack.index]
elem := &c.stack.elems[c.stack.top]
leafPage, _, _ := c.tx.readPage(elem.pgno)
cell := readLeafCell(leafPage, elem.index)
return cell.Type
@ -258,7 +258,7 @@ func WalkPage(tx *Tx, pgno uint32, walker Walker) {
walker.Visit(pgno, Branch)
for i, n := 0, readCellN(page); i < n; i++ {
cell := readBranchCell(page, i)
WalkPage(tx, cell.Pgno, walker)
WalkPage(tx, cell.ChildPgno, walker)
}
case PageTypeLeaf:
walker.Visit(pgno, Leaf)

6
rbf/dot.go
View file

@ -96,10 +96,10 @@ func dumpdot(tx *Tx, pgno uint32, parent string, writer io.Writer) {
for i, n := 0, readCellN(page); i < n; i++ {
cell := readBranchCell(page, i)
if cell.Flags&uint32(ContainerTypeBitmap) == 0 { // leaf/branch child page
dumpdot(tx, cell.Pgno, p, writer)
dumpdot(tx, cell.ChildPgno, p, writer)
} else {
b := fmt.Sprintf("bm%d", cell.Pgno)
fmt.Fprintf(writer, "%s[label=\"BITMAP(%d) key=%d \"]\n %s -> %s\n", b, cell.Pgno, cell.Key, p, b)
b := fmt.Sprintf("bm%d", cell.ChildPgno)
fmt.Fprintf(writer, "%s[label=\"BITMAP(%d) key=%d \"]\n %s -> %s\n", b, cell.ChildPgno, cell.LeftKey, p, b)
}
}
case PageTypeLeaf:

20
rbf/rbf.go
View file

@ -53,6 +53,8 @@ const (
RLEMaxSize = 2039
)
const maxBranchCellsPerPage = int((PageSize - branchPageHeaderSize) / (branchCellIndexElemSize + unsafe.Sizeof(branchCell{})))
// Page types.
const (
PageTypeRootRecord = 1
@ -103,7 +105,9 @@ const (
leafCellHeaderSize = 8 + 4 + 6 // key, type, count
leafPageHeaderSize = 4 + 4 + 2 // pgno, flags, cell n
leafCellIndexElemSize = 2
branchPageHeaderSize = 4 + 4 + 2 // pgno, flags, cell n
branchCellSize = 8 + 4 + 4 // key, flags, pgno
branchCellIndexElemSize = 2
)
var (
@ -566,9 +570,9 @@ func writeLeafCell(page []byte, i, offset int, cell leafCell) {
// branchCell represents a branch cell.
type branchCell struct {
Key uint64
Flags uint32
Pgno uint32
LeftKey uint64 // smallest key on ChildPgno
Flags uint32
ChildPgno uint32
}
// branchCellsPageSize returns the total page size required to hold cells.
@ -591,9 +595,9 @@ func readBranchCell(page []byte, i int) branchCell {
offset := readCellOffset(page, i)
var cell branchCell
cell.Key = *(*uint64)(unsafe.Pointer(&page[offset]))
cell.LeftKey = *(*uint64)(unsafe.Pointer(&page[offset]))
cell.Flags = *(*uint32)(unsafe.Pointer(&page[offset+8]))
cell.Pgno = *(*uint32)(unsafe.Pointer(&page[offset+12]))
cell.ChildPgno = *(*uint32)(unsafe.Pointer(&page[offset+12]))
return cell
}
@ -608,9 +612,9 @@ func readBranchCells(page []byte) []branchCell {
func writeBranchCell(page []byte, i, offset int, cell branchCell) {
writeCellOffset(page, i, offset)
*(*uint64)(unsafe.Pointer(&page[offset+0])) = cell.Key
*(*uint64)(unsafe.Pointer(&page[offset+0])) = cell.LeftKey
*(*uint32)(unsafe.Pointer(&page[offset+8])) = uint32(cell.Flags)
*(*uint32)(unsafe.Pointer(&page[offset+12])) = uint32(cell.Pgno)
*(*uint32)(unsafe.Pointer(&page[offset+12])) = uint32(cell.ChildPgno)
}
func highbits(v uint64) uint64 { return v >> 16 }
@ -673,7 +677,7 @@ func Pagedump(b []byte, indent string, writer io.Writer) {
fmt.Fprintf(writer, "==BRANCH pgno=%d flags=%d n=%d\n", pgno, flags, cellN)
for i := 0; i < cellN; i++ {
cell := readBranchCell(b, i)
fmt.Fprintf(writer, "[%d]: key=%d flags=%d pgno=%d\n", i, cell.Key, cell.Flags, cell.Pgno)
fmt.Fprintf(writer, "[%d]: key=%d flags=%d pgno=%d\n", i, cell.LeftKey, cell.Flags, cell.ChildPgno)
}
default:
fmt.Fprintf(writer, "==!PAGE %d flags=%d\n", pgno, flags)

55
rbf/tx.go
View file

@ -598,7 +598,7 @@ func (tx *Tx) RoaringBitmap(name string) (*roaring.Bitmap, error) {
return nil, err
}
elem := &c.stack.elems[c.stack.index]
elem := &c.stack.elems[c.stack.top]
leafPage, _, err := c.tx.readPage(elem.pgno)
if err != nil {
return nil, err
@ -634,7 +634,7 @@ func (tx *Tx) container(name string, key uint64) (*roaring.Container, error) {
return nil, err
}
elem := &c.stack.elems[c.stack.index]
elem := &c.stack.elems[c.stack.top]
leafPage, _, err := c.tx.readPage(elem.pgno)
if err != nil {
return nil, err
@ -779,7 +779,7 @@ func (tx *Tx) freePageSet() (map[uint32]struct{}, error) {
return m, err
}
elem := &c.stack.elems[c.stack.index]
elem := &c.stack.elems[c.stack.top]
leafPage, _, err := c.tx.readPage(elem.pgno)
if err != nil {
return nil, err
@ -855,7 +855,7 @@ func (tx *Tx) walkTree(pgno, parent uint32, fn func(pgno, parent, typ uint32) er
case PageTypeBranch:
for i, n := 0, readCellN(page); i < n; i++ {
cell := readBranchCell(page, i)
if err := tx.walkTree(cell.Pgno, pgno, fn); err != nil {
if err := tx.walkTree(cell.ChildPgno, pgno, fn); err != nil {
return err
}
}
@ -910,7 +910,7 @@ func (tx *Tx) nextFreelistPageNo() (uint32, error) {
return 0, err
}
elem := &c.stack.elems[c.stack.index]
elem := &c.stack.elems[c.stack.top]
leafPage, _, err := c.tx.readPage(elem.pgno)
if err != nil {
return 0, err
@ -947,7 +947,7 @@ func (tx *Tx) deallocateTree(pgno uint32) error {
case PageTypeBranch:
for i, n := 0, readCellN(page); i < n; i++ {
cell := readBranchCell(page, i)
if err := tx.deallocateTree(cell.Pgno); err != nil {
if err := tx.deallocateTree(cell.ChildPgno); err != nil {
return err
}
}
@ -1079,7 +1079,7 @@ func (tx *Tx) ForEachRange(name string, start, end uint64, fn func(uint64) error
return err
}
elem := &c.stack.elems[c.stack.index]
elem := &c.stack.elems[c.stack.top]
leafPage, _, err := c.tx.readPage(elem.pgno)
if err != nil {
return err
@ -1179,7 +1179,7 @@ func (tx *Tx) Count(name string) (uint64, error) {
return 0, err
}
elem := &c.stack.elems[c.stack.index]
elem := &c.stack.elems[c.stack.top]
leafPage, _, err := c.tx.readPage(elem.pgno)
if err != nil {
return 0, err
@ -1209,7 +1209,7 @@ func (tx *Tx) Max(name string) (uint64, error) {
return 0, err
}
elem := &c.stack.elems[c.stack.index]
elem := &c.stack.elems[c.stack.top]
leafPage, _, err := c.tx.readPage(elem.pgno)
if err != nil {
return 0, err
@ -1237,7 +1237,7 @@ func (tx *Tx) Min(name string) (uint64, bool, error) {
return 0, false, err
}
elem := &c.stack.elems[c.stack.index]
elem := &c.stack.elems[c.stack.top]
leafPage, _, err := c.tx.readPage(elem.pgno)
if err != nil {
return 0, false, err
@ -1304,7 +1304,7 @@ func (tx *Tx) CountRange(name string, start, end uint64) (uint64, error) {
return 0, err
}
elem := &csr.stack.elems[csr.stack.index]
elem := &csr.stack.elems[csr.stack.top]
leafPage, _, err := csr.tx.readPage(elem.pgno)
if err != nil {
return 0, err
@ -1382,7 +1382,7 @@ func (tx *Tx) OffsetRange(name string, offset, start, endx uint64) (*roaring.Bit
return nil, err
}
elem := &c.stack.elems[c.stack.index]
elem := &c.stack.elems[c.stack.top]
leafPage, _, err := c.tx.readPage(elem.pgno)
if err != nil {
return nil, err
@ -1419,7 +1419,7 @@ func (itr *containerIterator) Next() bool {
// Value returns the current key & container.
func (itr *containerIterator) Value() (uint64, *roaring.Container) {
elem := &itr.cursor.stack.elems[itr.cursor.stack.index]
elem := &itr.cursor.stack.elems[itr.cursor.stack.top]
leafPage, _, _ := itr.cursor.tx.readPage(elem.pgno)
cell := readLeafCell(leafPage, elem.index)
return cell.Key, toContainer(cell, itr.cursor.tx)
@ -1470,7 +1470,7 @@ func (tx *Tx) DumpString(short bool, shard uint64) (r string) {
}
panicOn(err)
elem := &c.stack.elems[c.stack.index]
elem := &c.stack.elems[c.stack.top]
leafPage, _, err := c.tx.readPage(elem.pgno)
panicOn(err)
cell := readLeafCell(leafPage, elem.index)
@ -1478,7 +1478,7 @@ func (tx *Tx) DumpString(short bool, shard uint64) (r string) {
ckey := cell.Key
ct := toContainer(cell, tx)
s := stringOfCkeyCt(ckey, ct, name.(string), short)
s := stringOfCkeyCt(ckey, ct, name.(string), short, true)
r += s
n++
}
@ -1532,20 +1532,27 @@ func bitmapAsString(rbm *roaring.Bitmap) (r string) {
return r + ")"
}
func stringOfCkeyCt(ckey uint64, ct *roaring.Container, rrName string, short bool) (s string) {
func stringOfCkeyCt(ckey uint64, ct *roaring.Container, rrName string, short, showHash bool) (s string) {
by := containerToBytes(ct)
hash := hash.Blake3sum16(by)
hsh := ""
if showHash {
by := containerToBytes(ct)
hsh = hash.Blake3sum16(by)
}
cts := roaring.NewSliceContainers()
cts.Put(ckey, ct)
rbm := &roaring.Bitmap{Containers: cts}
srbm := bitmapAsString(rbm)
pre := txkey.PrefixToString([]byte(rrName))
var pre string
if len(rrName) > 0 {
pre = txkey.PrefixToString([]byte(rrName))
}
bkey := pre + fmt.Sprintf("ckey@%020d", ckey)
s = fmt.Sprintf("%v -> %v (%v hot)\n", bkey, hash, ct.N())
s = fmt.Sprintf("%v -> %v (%v hot)\n", bkey, hsh, ct.N())
if !short {
s += " ......." + srbm + "\n"
}
@ -1553,6 +1560,7 @@ func stringOfCkeyCt(ckey uint64, ct *roaring.Container, rrName string, short boo
}
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
@ -1588,6 +1596,7 @@ func (tx *Tx) ImportRoaringBits(name string, itr roaring.RoaringIterator, clear
defer cur.Close()
for itrKey, synthC := itr.NextContainer(); synthC != nil; itrKey, synthC = itr.NextContainer() {
if rowSize != 0 {
currRow = itrKey / rowSize
}
@ -1602,7 +1611,7 @@ func (tx *Tx) ImportRoaringBits(name string, itr roaring.RoaringIterator, clear
if exact, err := cur.Seek(itrKey); err != nil {
return changed, rowSet, err
} else if exact {
elem := &cur.stack.elems[cur.stack.index]
elem := &cur.stack.elems[cur.stack.top]
leafPage, _, err := cur.tx.readPage(elem.pgno)
if err != nil {
return changed, rowSet, err
@ -1803,9 +1812,9 @@ func (tx *Tx) Pages(pgnos []uint32) ([]Page, error) {
page := &BranchPage{BranchPageInfo: info}
for _, cell := range readBranchCells(buf) {
page.Cells = append(page.Cells, &BranchCell{
Key: cell.Key,
Key: cell.LeftKey,
Flags: cell.Flags,
Pgno: cell.Pgno,
Pgno: cell.ChildPgno,
})
}
pages = append(pages, page)

273
rbf/util.go Normal file
View file

@ -0,0 +1,273 @@
// Copyright 2017 Pilosa Corp.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package rbf
import (
"fmt"
"io"
"strings"
"github.com/pilosa/pilosa/v2/txkey"
)
func (tx *Tx) dumpAllPages(showLeaves bool) error {
infos, err := tx.PageInfos()
if err != nil {
return err
}
// Write header.
fmt.Printf("Pgno ")
fmt.Printf("TYPE ")
spc := strings.Repeat(" ", 29)
fmt.Printf("TREE " + spc)
fmt.Printf("EXTRA\n")
fmt.Printf("======== ")
fmt.Printf("========== ")
fmt.Printf("============================== " + spc)
fmt.Printf("====================\n")
// Print one line for each page.
for pgno, info := range infos {
switch info := info.(type) {
case *MetaPageInfo:
fmt.Printf("Pgno:%-8d ", pgno)
fmt.Printf("%-10s ", "meta")
fmt.Printf("%-54s ", "")
fmt.Printf("pageN=%d,walid=%d,rootrec=%d,freelist=%d\n", info.PageN, info.WALID, info.RootRecordPageNo, info.FreelistPageNo)
case *RootRecordPageInfo:
fmt.Printf("Pgno:%-8d ", pgno)
fmt.Printf("%-10s ", "rootrec")
fmt.Printf("%-54s ", "")
fmt.Printf("next=%d\n", info.Next)
page, _, err := tx.readPage(uint32(pgno))
panicOn(err)
rootRecords, err := readRootRecords(page)
panicOn(err)
for k, rr := range rootRecords {
fmt.Printf(" [%02v] Name:'%v' pgno:%v\n", k, prefixToString(rr.Name), rr.Pgno)
}
case *LeafPageInfo:
if !showLeaves {
continue
}
fmt.Printf("Pgno:%-8d ", pgno)
fmt.Printf("%-10s ", "leaf")
fmt.Printf("%-54q ", prefixToString(info.Tree))
fmt.Printf("flags=x%x,celln=%d\n", info.Flags, info.CellN)
page, _, err := tx.readPage(uint32(pgno))
panicOn(err)
var leafCells [PageSize / 8]leafCell
cells := readLeafCells(page, leafCells[:])
for k, cell := range cells {
fmt.Printf(" [%02v] : (container)Key:%v Type:%v BitN:%v len(Data):%v\n", k, cell.Key, cell.Type, cell.BitN, len(cell.Data))
}
case *BranchPageInfo:
fmt.Printf("Pgno:%-8d ", pgno)
fmt.Printf("%-10s ", "branch")
fmt.Printf("%-54q ", prefixToString(info.Tree))
fmt.Printf("flags=x%x,celln=%d\n", info.Flags, info.CellN)
page, _, err := tx.readPage(uint32(pgno))
panicOn(err)
cells := readBranchCells(page)
for i, cell := range cells {
fmt.Printf(" [%02v] : (ChildPages's smallest) Key:%05v -> (Child) pgno:%v\n", i, cell.LeftKey, cell.ChildPgno)
}
case *BitmapPageInfo:
fmt.Printf("Pgno:%-8d ", pgno)
fmt.Printf("%-10s ", "bitmap")
fmt.Printf("%-54q ", prefixToString(info.Tree))
fmt.Printf("-\n")
case *FreePageInfo:
fmt.Printf("Pgno:%-8d ", pgno)
fmt.Printf("%-10s ", "free")
fmt.Printf("%-54s ", "")
fmt.Printf("-\n")
case nil:
fmt.Printf("Pgno:%-8d ", pgno)
fmt.Printf("%-10s ", "<nil> problem, corrupt page set")
fmt.Printf("%-54s ", "")
fmt.Printf("-\n")
default:
panic(fmt.Sprintf("unexpected page info type %T at pgno %v", info, pgno))
}
}
return nil
}
func (tx *Tx) dumpPages(pgnos []uint32) error {
// Fetch the page.
pages, err := tx.Pages(pgnos)
if err != nil {
return err
}
for _, page := range pages {
switch page := page.(type) {
case *MetaPage:
printMetaPage(page)
case *RootRecordPage:
printRootRecordPage(page)
case *LeafPage:
printLeafPage(page)
case *BranchPage:
printBranchPage(page)
case *BitmapPage:
printBitmapPage(page)
case *FreePage:
printFreePage(page)
default:
return fmt.Errorf("unexpected page type %T", page)
}
fmt.Printf("\n")
}
return nil
}
var _ = (&Tx{}).dumpPages
func printMetaPage(page *MetaPage) {
fmt.Printf("Pgno: %d\n", page.Pgno)
fmt.Printf("Type: meta\n")
fmt.Printf("PageN: %d\n", page.PageN)
fmt.Printf("WALID: %d\n", page.WALID)
fmt.Printf("Root Record Pgno: %d\n", page.RootRecordPageNo)
fmt.Printf("Freelist Pgno: %d\n", page.FreelistPageNo)
}
func printRootRecordPage(page *RootRecordPage) {
fmt.Printf("Pgno: %d\n", page.Pgno)
fmt.Printf("Type: root record\n")
fmt.Printf("Next: %d\n", page.Next)
fmt.Printf("Records: n=%d\n", len(page.Records))
for i, rec := range page.Records {
fmt.Printf("[%d]: name=%q pgno=%d\n", i, rec.Name, rec.Pgno)
}
}
func printLeafPage(page *LeafPage) {
fmt.Printf("Pgno: %d\n", page.Pgno)
fmt.Printf("Type: leaf\n")
fmt.Printf("Cells: n=%d\n", len(page.Cells))
for i, cell := range page.Cells {
if cell.Type == ContainerTypeBitmapPtr {
fmt.Printf("[%d]: ckey=%d type=%s pgno=%d\n", i, cell.Key, cell.Type, cell.Pgno)
} else {
fmt.Printf("[%d]: ckey=%d type=%s values=%v\n", i, cell.Key, cell.Type, cell.Values)
}
}
}
func printBranchPage(page *BranchPage) {
fmt.Printf("Pgno: %d\n", page.Pgno)
fmt.Printf("Type: branch\n")
fmt.Printf("Cells: n=%d\n", len(page.Cells))
for i, cell := range page.Cells {
fmt.Printf("[%d]: ckey=%d flags=%d pgno=%d\n", i, cell.Key, cell.Flags, cell.Pgno)
}
}
func printBitmapPage(page *BitmapPage) {
fmt.Printf("Pgno: %d\n", page.Pgno)
fmt.Printf("Type: bitmap\n")
fmt.Printf("Values: %v\n", page.Values)
}
func printFreePage(page *FreePage) {
fmt.Printf("Pgno: %d\n", page.Pgno)
fmt.Printf("Type: free\n")
}
func prefixToString(s string) (ret string) {
defer func() {
if err := recover(); err != nil {
ret = s
}
}()
return txkey.PrefixToString([]byte(s))
}
func (c *Cursor) dump() {
fmt.Printf("\n Cursor %p has bitmaps:\n%v\n", c, c.debugStringBitmaps())
}
var _ = (&Cursor{}).dump
var _ = (&Cursor{}).debugStringBitmaps
func (c_orig *Cursor) debugStringBitmaps() (r string) {
// work with a totally new Cursor, so we don't impact our current cursor
// so any test using the cursor isn't disturbed.
c2 := Cursor{tx: c_orig.tx}
c2.stack.elems[0] = c_orig.stack.elems[0]
err := c2.First()
if err != nil {
if err == io.EOF {
// ok, can be empty
return "<empty cursor/tx>"
} else {
panic(err)
}
}
n := 0
for {
err := c2.Next()
if err == io.EOF {
break
}
panicOn(err)
//instead of cell := c2.cell()
elem := &c2.stack.elems[c2.stack.top]
leafPage, _, err := c2.tx.readPage(elem.pgno)
panicOn(err)
cell := readLeafCell(leafPage, elem.index)
ckey := cell.Key
ct := toContainer(cell, c2.tx)
const short = true
s := stringOfCkeyCt(ckey, ct, "", short, true)
r += s
n++
}
if n == 0 {
return ""
}
return
}
///////////////// happy linter
var _ = printMetaPage
var _ = printRootRecordPage
var _ = printLeafPage
var _ = printBranchPage
var _ = printBitmapPage
var _ = printFreePage
var _ = prefixToString

56
rbf/util_test.go
View file

@ -15,8 +15,13 @@ package rbf
import (
"fmt"
"github.com/pilosa/pilosa/v2/roaring"
"io"
"io/ioutil"
"os"
"testing"
rbfcfg "github.com/pilosa/pilosa/v2/rbf/cfg"
"github.com/pilosa/pilosa/v2/roaring"
)
// util_test adds reusable utilities for testing.
@ -24,6 +29,8 @@ import (
// scanning all data under an rbf-root (logically equivalent
// to a single roaring.Bitmap with multiple rows).
var _ = keysFromParents // linter happy
// verify that BitN and ElemN are correct.
func (c_orig *Cursor) DebugSlowCheckAllPages() {
@ -67,7 +74,7 @@ func checkElemNBitN(tx *Tx, pgno uint32) {
for i, n := 0, readCellN(page); i < n; i++ {
cell := readBranchCell(page, i)
if cell.Flags&uint32(ContainerTypeBitmap) == 0 { // leaf/branch child page
checkElemNBitN(tx, cell.Pgno)
checkElemNBitN(tx, cell.ChildPgno)
}
// else is a bitmap
}
@ -133,3 +140,48 @@ func verifyElemNBitN(tx *Tx, lc leafCell) {
panic(fmt.Sprintf("lc.ElemN(%v) != obsElemN(%v); typ='%v'", lc.ElemN, obsElemN, typ))
}
}
func testHelperMustOpenNewDB(tb testing.TB, cfg ...*rbfcfg.Config) *DB {
tb.Helper()
path, err := ioutil.TempDir("", "")
if err != nil {
panic(err)
}
var cfg0 *rbfcfg.Config
if len(cfg) > 0 {
cfg0 = cfg[0]
}
db := NewDB(path, cfg0)
if err := db.Open(); err != nil {
tb.Fatal(err)
}
return db
}
// MustCloseDB closes db. On error, fail test.
// This function also also performs an integrity check on the DB.
func MustCloseDB(tb testing.TB, db *DB) {
tb.Helper()
if err := db.Check(); err != nil && err != ErrClosed {
tb.Fatal(err)
} else if n := db.TxN(); n != 0 {
tb.Fatalf("db still has %d active transactions; must closed before closing db", n)
} else if err := db.Close(); err != nil && err != ErrClosed {
tb.Fatal(err)
} else if err := os.RemoveAll(db.Path); err != nil {
tb.Fatal(err)
}
}
// MustBegin returns a new transaction or fails.
func MustBegin(tb testing.TB, db *DB, writable bool) *Tx {
tb.Helper()
tx, err := db.Begin(writable)
if err != nil {
tb.Fatal(err)
}
return tx
}

47
roaring/roaring.go
View file

@ -1757,13 +1757,16 @@ type RoaringIterator interface {
// It may well share much underlying data.
Clone() RoaringIterator
// ContainerKeySpan provides the smallest and largest
// ContainerKeys provides all the
// container keys that the iterator will return.
// The current implementation requires that the underlying header
// lists the keys in ascending order.
// Iff there no keys, then empty will be returned true.
// If there is only a single key, then ckeyLast will equal ckeyFirst.
ContainerKeySpan() (ckeyFirst, ckeyLast uint64, empty bool)
// If there are no keys, then an empty slice will be returned.
ContainerKeys() (slc []uint64)
// Skip will move the iterator forward by 1 without
// materializing the container.
Skip()
}
// baseRoaringIterator holds values used by both Pilosa and official Roaring
@ -1932,22 +1935,42 @@ func (r *baseRoaringIterator) Done(err error) {
r.currentDataOffset = 0
}
func (r *baseRoaringIterator) ContainerKeySpan() (ckeyFirst, ckeyLast uint64, empty bool) {
func (r *pilosaRoaringIterator) ContainerKeys() (slc []uint64) {
n := r.keys
if n == 0 {
empty = true
return
}
ckeyFirst = binary.LittleEndian.Uint64(r.headers[0:8])
if n == 1 {
ckeyLast = ckeyFirst
return
for i := int64(0); i < n; i++ {
beg := i * 12
slc = append(slc, binary.LittleEndian.Uint64(r.headers[beg:beg+8]))
}
beg := (n - 1) * 12
ckeyLast = binary.LittleEndian.Uint64(r.headers[beg : beg+8])
return
}
func (r *officialRoaringIterator) ContainerKeys() (slc []uint64) {
n := r.keys
if n == 0 {
return
}
for i := int64(0); i < n; i++ {
beg := i * 4
slc = append(slc, uint64(binary.LittleEndian.Uint16(r.headers[beg:beg+2])))
}
return
}
func (r *baseRoaringIterator) Skip() {
if r.currentIdx >= r.keys {
// we're already done
return
}
r.currentIdx++
if r.currentIdx == r.keys {
// this is the last key. transition state to the finalized state
r.Done(io.EOF)
}
}
// Len() indicates the total number of containers the iterator expects to have.
func (r *baseRoaringIterator) Len() int64 {
return r.keys

73
roaring/roaring_internal_test.go
View file

@ -4430,15 +4430,15 @@ func TestCloneRoaringIterator(t *testing.T) {
itr2 := itr.Clone()
firstCkey, lastCkey, empty := itr2.ContainerKeySpan()
if empty {
ikeys := itr2.ContainerKeys()
if len(ikeys) == 0 {
t.Fatalf("should not be empty")
}
if firstCkey != 0 {
t.Fatalf("firstCkey should be 0")
if ikeys[0] != 0 {
t.Fatalf("first ikeys should be 0")
}
if lastCkey != 10001 {
t.Fatalf("lastCkey should be 10001")
if ikeys[len(ikeys)-1] != 10001 {
t.Fatalf("last ikeys should be 10001")
}
var keys []uint64
@ -4457,7 +4457,7 @@ func TestCloneRoaringIterator(t *testing.T) {
}
}
func TestRoaringIteratorContainerKeySpan(t *testing.T) {
func TestRoaringIteratorContainerKeys(t *testing.T) {
ca := NewContainerArray([]uint16{1, 10, 100, 1000})
ba := NewFileBitmap()
@ -4475,15 +4475,15 @@ func TestRoaringIteratorContainerKeySpan(t *testing.T) {
t.Fatalf("error NewRoaringIterator(buf.Bytes()): %v", err)
}
firstCkey, lastCkey, empty := itr.ContainerKeySpan()
if empty {
ikeys := itr.ContainerKeys()
if len(ikeys) == 0 {
t.Fatalf("should not be empty")
}
if firstCkey != 10 {
t.Fatalf("firstCkey should be 10")
if ikeys[0] != 10 {
t.Fatalf("first ikeys should be 10")
}
if lastCkey != 10001 {
t.Fatalf("lastCkey should be 10001")
if ikeys[len(ikeys)-1] != 10001 {
t.Fatalf("last ikeys should be 10001")
}
// make and check empty bitmap
@ -4499,12 +4499,55 @@ func TestRoaringIteratorContainerKeySpan(t *testing.T) {
if err != nil {
t.Fatalf("error NewRoaringIterator(bufEmpty.Bytes()): %v", err)
}
_, _, empty = itrEmpty.ContainerKeySpan()
if !empty {
ikeys = itrEmpty.ContainerKeys()
if len(ikeys) != 0 {
t.Fatalf("should be empty")
}
}
func TestRoaringIteratorSkip(t *testing.T) {
ca := NewContainerArray([]uint16{1, 10, 100, 1000})
ba := NewFileBitmap()
ba.Containers.Put(101, ca)
ba.Containers.Put(10, ca)
ba.Containers.Put(10001, ca)
var buf bytes.Buffer
_, err := ba.WriteTo(&buf)
if err != nil {
t.Fatalf("error writing: %v", err)
}
itr, err := NewRoaringIterator(buf.Bytes())
if err != nil {
t.Fatalf("error NewRoaringIterator(buf.Bytes()): %v", err)
}
itr.Skip()
ckey1, ct := itr.NextContainer()
_ = ct
if ckey1 != 101 {
t.Fatalf("expected to skip 10 and get 101 but got: %v", ckey1)
}
// make and check empty bitmap
baEmpty := NewFileBitmap()
var bufEmpty bytes.Buffer
_, err = baEmpty.WriteTo(&bufEmpty)
if err != nil {
t.Fatalf("error writing: %v", err)
}
itrEmpty, err := NewRoaringIterator(bufEmpty.Bytes())
if err != nil {
t.Fatalf("error NewRoaringIterator(bufEmpty.Bytes()): %v", err)
}
itrEmpty.Skip()
// should not have panic-ed.
}
// we were seeing unionInterval16InPlace() returning too
// large an run container, which was causing problems when
// we write to the transactional backends. Verify that