scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-11 23:31:03 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Add sync.Pool for RBF pages

Browse source
This commit is contained in:
Ben Johnson 2020-12-03 07:32:36 -07:00
parent 1ce15a51c9
commit 79e6156003
7 changed files with 261 additions and 84 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

145
rbf/cursor.go
View file

@ -33,7 +33,6 @@ type Cursor struct {
buffered bool
// buffers
leafPage []byte
array [ArrayMaxSize + 1]uint16
rle [RLEMaxSize + 1]roaring.Interval16
leafCells [PageSize / 8]leafCell
@ -140,7 +139,13 @@ func (c *Cursor) Add(v uint64) (changed bool, err error) {
}
// If the container exists and bit is not set then update the page.
cell := c.cell()
elem := &c.stack.elems[c.stack.index]
leafPage, _, err := c.tx.readPage(elem.pgno)
if err != nil {
return false, err
}
cell := readLeafCell(leafPage, elem.index)
switch cell.Type {
case ContainerTypeArray:
// Exit if value exists in array container.
@ -205,7 +210,13 @@ func (c *Cursor) Remove(v uint64) (changed bool, err error) {
}
// If the container exists and bit is not set then update the page.
cell := c.cell()
elem := &c.stack.elems[c.stack.index]
leafPage, _, err := c.tx.readPage(elem.pgno)
if err != nil {
return false, err
}
cell := readLeafCell(leafPage, elem.index)
switch cell.Type {
case ContainerTypeArray:
// Exit if value does not exists in array container.
@ -315,7 +326,13 @@ func (c *Cursor) Contains(v uint64) (exists bool, err error) {
}
// If the container exists then check for low bits existence.
cell := c.cell()
elem := &c.stack.elems[c.stack.index]
leafPage, _, err := c.tx.readPage(elem.pgno)
if err != nil {
return false, err
}
cell := readLeafCell(leafPage, elem.index)
switch cell.Type {
case ContainerTypeArray:
a := toArray16(cell.Data)
@ -351,14 +368,16 @@ func toPgno(val []byte) uint32 {
}
func (c *Cursor) putLeafCell(in leafCell) (err error) {
leafPage := c.leafPage // the last read leaf page
elem := &c.stack.elems[c.stack.index]
leafPage, isHeap, err := c.tx.readPage(elem.pgno) // the last read leaf page
if err != nil {
return err
}
cellN := readCellN(leafPage)
// Determine if the insert/update will overflow the page.
// If it doesn't then we can do an optimized write where we don't deserialize.
isInsert := elem.index >= cellN || c.Key() != in.Key
isInsert := elem.index >= cellN || pageKeyAt(leafPage, elem.index) != in.Key
newEstPageSize := leafPageSize(leafPage)
if isInsert {
newEstPageSize += in.Size() + leafCellIndexElemSize
@ -477,6 +496,11 @@ func (c *Cursor) putLeafCell(in leafCell) (err error) {
parents = append(parents, parent)
}
// Free the source page once we've finished with it if it is on heap.
if isHeap {
freePage(leafPage)
}
// TODO(BBJ): Update page in buffer & cursor stack.
// If this is not a split then exit now.
@ -498,8 +522,11 @@ func (c *Cursor) putLeafCell(in leafCell) (err error) {
// 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) {
src := c.leafPage
elem := &c.stack.elems[c.stack.index]
src, isHeap, err := c.tx.readPage(elem.pgno)
if err != nil {
return err
}
srcCellN := readCellN(src)
// Determine the cell count of the new page.
@ -509,7 +536,7 @@ func (c *Cursor) putLeafCellFast(in leafCell, isInsert bool) (err error) {
}
// Write page header.
dst := make([]byte, PageSize)
dst := allocPage() // make([]byte, PageSize)
writePageNo(dst, readPageNo(src))
writeFlags(dst, PageTypeLeaf)
writeCellN(dst, dstCellN)
@ -557,15 +584,28 @@ func (c *Cursor) putLeafCellFast(in leafCell, isInsert bool) (err error) {
}
// Write new page to dirty page cache.
return c.tx.writePage(dst)
if err := c.tx.writePage(dst); err != nil {
return err
}
// Free page if on heap.
if isHeap {
freePage(src)
}
return nil
}
// deleteLeafCell removes a cell from the currently positioned page & index.
func (c *Cursor) deleteLeafCell(key uint64) (err error) {
cells := readLeafCells(c.leafPage, c.leafCells[:])
elem := &c.stack.elems[c.stack.index]
leafPage, _, err := c.tx.readPage(elem.pgno)
if err != nil {
return err
}
cells := readLeafCells(leafPage, c.leafCells[:])
oldPageKey := cells[0].Key
cell := c.cell()
cell := readLeafCell(leafPage, elem.index)
if cell.Type == ContainerTypeBitmapPtr {
if err := c.tx.freePgno(toPgno(cell.Data)); err != nil {
return err
@ -612,7 +652,7 @@ func (c *Cursor) putBranchCells(stackIndex int, newCells []branchCell) (err erro
elem := &c.stack.elems[stackIndex]
// Read branch page from disk. The current buffer is the leaf page.
page, err := c.tx.readPage(elem.pgno)
page, _, err := c.tx.readPage(elem.pgno)
if err != nil {
return err
}
@ -692,7 +732,7 @@ func (c *Cursor) updateBranchCell(stackIndex int, newKey uint64) (err error) {
elem := &c.stack.elems[stackIndex]
// Read branch page from disk. The current buffer is the leaf page.
page, err := c.tx.readPage(elem.pgno)
page, _, err := c.tx.readPage(elem.pgno)
if err != nil {
return err
}
@ -728,7 +768,7 @@ func (c *Cursor) deleteBranchCell(stackIndex int, key uint64) (err error) {
elem := &c.stack.elems[stackIndex]
// Read branch page from disk. The current buffer is the leaf page.
page, err := c.tx.readPage(elem.pgno)
page, _, err := c.tx.readPage(elem.pgno)
if err != nil {
return err
}
@ -742,7 +782,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].Pgno)
if err != nil {
return err
}
@ -849,16 +889,10 @@ func splitBranchCells(cells []branchCell) [][]branchCell {
return slices
}
// Key returns the key that the cursor is currently positioned over.
func (c *Cursor) Key() uint64 {
elem := &c.stack.elems[c.stack.index]
offset := readCellOffset(c.leafPage, elem.index)
return *(*uint64)(unsafe.Pointer(&c.leafPage[offset]))
}
func (c *Cursor) cell() leafCell {
elem := &c.stack.elems[c.stack.index]
return readLeafCell(c.leafPage[:], elem.index)
// pageKeyAt returns the key at the given index of the page.
func pageKeyAt(page []byte, index int) uint64 {
offset := readCellOffset(page, index)
return *(*uint64)(unsafe.Pointer(&page[offset]))
}
// First moves to the first element of the btree.
@ -868,7 +902,7 @@ func (c *Cursor) First() error {
for c.stack.index = 0; ; c.stack.index++ {
elem := &c.stack.elems[c.stack.index]
buf, err := c.tx.readPage(elem.pgno)
buf, _, err := c.tx.readPage(elem.pgno)
if err != nil {
return err
}
@ -886,7 +920,6 @@ func (c *Cursor) First() error {
}
case PageTypeLeaf:
c.leafPage = buf
elem.index = 0
if readCellN(buf) == 0 {
return io.EOF // root leaf with no elements
@ -906,7 +939,7 @@ func (c *Cursor) Last() error {
for c.stack.index = 0; ; c.stack.index++ {
elem := &c.stack.elems[c.stack.index]
buf, err := c.tx.readPage(elem.pgno)
buf, _, err := c.tx.readPage(elem.pgno)
if err != nil {
return err
}
@ -924,7 +957,6 @@ func (c *Cursor) Last() error {
case PageTypeLeaf:
elem.index = readCellN(buf) - 1
c.leafPage = buf
if readCellN(buf) == 0 {
return io.EOF // root leaf with no elements
}
@ -944,7 +976,7 @@ func (c *Cursor) Seek(key uint64) (exact bool, err error) {
elem := &c.stack.elems[c.stack.index]
assert(elem.pgno != 0) // cursor should never point to page zero (meta)
buf, err := c.tx.readPage(elem.pgno)
buf, _, err := c.tx.readPage(elem.pgno)
if err != nil {
return false, err
}
@ -986,7 +1018,6 @@ func (c *Cursor) Seek(key uint64) (exact bool, err error) {
return 1
})
elem.index = index
c.leafPage = buf
return xact, nil
default:
@ -997,18 +1028,24 @@ 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]
leafPage, _, err := c.tx.readPage(elem.pgno)
if err != nil {
return err
}
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) {
if elem.index >= readCellN(leafPage) {
return c.goNextPage()
}
return nil
}
// Move forward to the next leaf element if available.
if elem := &c.stack.elems[c.stack.index]; elem.index < readCellN(c.leafPage)-1 {
if elem.index < readCellN(leafPage)-1 {
elem.index++
return nil
}
@ -1047,7 +1084,7 @@ func (c *Cursor) Prev() error {
for ; ; c.stack.index++ {
elem := &c.stack.elems[c.stack.index]
buf, err := c.tx.readPage(elem.pgno)
buf, _, err := c.tx.readPage(elem.pgno)
if err != nil {
return err
}
@ -1063,7 +1100,6 @@ func (c *Cursor) Prev() error {
case PageTypeLeaf:
elem.index = readCellN(buf) - 1
c.leafPage = buf
return nil
default:
return fmt.Errorf("rbf.Cursor.Prev(): invalid page type: pgno=%d type=%d", elem.pgno, typ)
@ -1071,13 +1107,25 @@ 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]
leafPage, _, _ := c.tx.readPage(elem.pgno)
if readCellN(leafPage[:]) == 0 {
return 0
}
cell := readLeafCell(leafPage, elem.index)
return cell.Key
}
// 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 {
leafPage, _, _ := c.tx.readPage(elem.pgno)
if readCellN(leafPage[:]) == 0 {
return nil
}
cell := readLeafCell(c.leafPage[:], elem.index)
cell := readLeafCell(leafPage, elem.index)
return cell.Values(c.tx)
}
@ -1091,7 +1139,7 @@ type stackElem struct {
func (c *Cursor) goNextPage() error {
for c.stack.index--; c.stack.index >= 0; c.stack.index-- {
elem := &c.stack.elems[c.stack.index]
if buf, err := c.tx.readPage(elem.pgno); err != nil {
if buf, _, err := c.tx.readPage(elem.pgno); err != nil {
return err
} else if n := readCellN(buf); elem.index+1 < n {
elem.index++
@ -1108,7 +1156,7 @@ func (c *Cursor) goNextPage() error {
// Traverse back down the stack to find the first element in each page.
for ; ; c.stack.index++ {
elem := &c.stack.elems[c.stack.index]
buf, err := c.tx.readPage(elem.pgno)
buf, _, err := c.tx.readPage(elem.pgno)
if err != nil {
return err
}
@ -1122,7 +1170,6 @@ func (c *Cursor) goNextPage() error {
}
case PageTypeLeaf:
elem.index = 0
c.leafPage = buf
return nil
default:
return fmt.Errorf("rbf.Cursor.Next(): invalid page type: pgno=%d type=%d", elem.pgno, typ)
@ -1175,7 +1222,13 @@ func ConvertToLeafArgs(key uint64, c *roaring.Container) (result leafCell) {
}
func (c *Cursor) merge(key uint64, data *roaring.Container) (bool, error) {
cell := c.cell()
elem := &c.stack.elems[c.stack.index]
leafPage, _, err := c.tx.readPage(elem.pgno)
if err != nil {
return false, err
}
cell := readLeafCell(leafPage, elem.index)
var container *roaring.Container
switch cell.Type {
case ContainerTypeArray:
@ -1258,7 +1311,13 @@ func (c *Cursor) RemoveRoaring(bm *roaring.Bitmap) (changed bool, err error) {
}
func (c *Cursor) difference(key uint64, data *roaring.Container) (bool, error) {
cell := c.cell()
elem := &c.stack.elems[c.stack.index]
leafPage, _, err := c.tx.readPage(elem.pgno)
if err != nil {
return false, err
}
cell := readLeafCell(leafPage, elem.index)
var container *roaring.Container
switch cell.Type {
case ContainerTypeArray:

35
rbf/cursorx.go
View file

@ -61,7 +61,14 @@ func (c *Cursor) Rows() ([]uint64, error) {
if err != nil {
break
}
cell := c.cell()
elem := &c.stack.elems[c.stack.index]
leafPage, _, err := c.tx.readPage(elem.pgno)
if err != nil {
return nil, err
}
cell := readLeafCell(leafPage, elem.index)
vRow := cell.Key >> shardVsContainerExponent
if vRow == lastRow {
continue
@ -92,7 +99,12 @@ func (c *Cursor) DumpKeys() {
if err == io.EOF {
break
}
cell := c.cell()
elem := &c.stack.elems[c.stack.index]
leafPage, _, err := c.tx.readPage(elem.pgno)
if err != nil {
return
}
cell := readLeafCell(leafPage, elem.index)
fmt.Println("key", cell.Key)
}
}
@ -129,7 +141,11 @@ func (c *Cursor) Row(shard, rowID uint64) (*roaring.Bitmap, error) {
}
if !ok {
elem := &c.stack.elems[c.stack.index]
n := readCellN(c.leafPage)
leafPage, _, err := c.tx.readPage(elem.pgno)
if err != nil {
return nil, err
}
n := readCellN(leafPage)
if elem.index >= n {
if err := c.goNextPage(); err != nil {
return nil, errors.Wrap(err, "row")
@ -145,7 +161,12 @@ func (c *Cursor) Row(shard, rowID uint64) (*roaring.Bitmap, error) {
return nil, err
}
cell := c.cell()
elem := &c.stack.elems[c.stack.index]
leafPage, _, err := c.tx.readPage(elem.pgno)
if err != nil {
return nil, err
}
cell := readLeafCell(leafPage, elem.index)
if cell.Key >= hi1 {
break
}
@ -157,7 +178,9 @@ 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 {
cell := c.cell()
elem := &c.stack.elems[c.stack.index]
leafPage, _, _ := c.tx.readPage(elem.pgno)
cell := readLeafCell(leafPage, elem.index)
return cell.Type
}
@ -219,7 +242,7 @@ type Walker interface {
}
func WalkPage(tx *Tx, pgno uint32, walker Walker) {
page, err := tx.readPage(pgno)
page, _, err := tx.readPage(pgno)
if err != nil {
panic(err)
}

18
rbf/db.go
View file

@ -599,3 +599,21 @@ func (db *DB) getCursor(tx *Tx) (c *Cursor) {
return nil
}
}
// Shared pool for in-memory database pages.
// These are used before being flushed to disk.
var pagePool = &sync.Pool{
New: func() interface{} {
page := make([]byte, PageSize)
return &page
},
}
func allocPage() []byte {
page := pagePool.Get().(*[]byte)
return *page
}
func freePage(page []byte) {
pagePool.Put(&page)
}

2
rbf/dot.go
View file

@ -62,7 +62,7 @@ func dotCell(b []byte, parent string, writer io.Writer) {
// dumpdot recursively writes the tree representation starting from a given page to STDERR.
func dumpdot(tx *Tx, pgno uint32, parent string, writer io.Writer) {
page, err := tx.readPage(pgno)
page, _, err := tx.readPage(pgno)
if err != nil {
panic(err)
}

2
rbf/rbf.go
View file

@ -682,7 +682,7 @@ func Pagedump(b []byte, indent string, writer io.Writer) {
func Walk(tx *Tx, pgno uint32, v func(uint32, []*RootRecord)) {
for pgno := readMetaRootRecordPageNo(tx.meta[:]); pgno != 0; {
page, err := tx.readPage(pgno)
page, _, err := tx.readPage(pgno)
if err != nil {
panic(err)
}

141
rbf/tx.go
View file

@ -373,7 +373,7 @@ func (tx *Tx) RootRecords() (records *immutable.SortedMap, err error) {
records = immutable.NewSortedMap(nil)
for pgno := readMetaRootRecordPageNo(tx.meta[:]); pgno != 0; {
page, err := tx.readPage(pgno)
page, _, err := tx.readPage(pgno)
if err != nil {
return nil, err
}
@ -401,7 +401,7 @@ func (tx *Tx) writeRootRecordPages(records *immutable.SortedMap) (err error) {
// Release all existing root record pages.
for pgno := readMetaRootRecordPageNo(tx.meta[:]); pgno != 0; {
page, err := tx.readPage(pgno)
page, _, err := tx.readPage(pgno)
if err != nil {
return err
}
@ -598,7 +598,12 @@ func (tx *Tx) RoaringBitmap(name string) (*roaring.Bitmap, error) {
return nil, err
}
cell := c.cell()
elem := &c.stack.elems[c.stack.index]
leafPage, _, err := c.tx.readPage(elem.pgno)
if err != nil {
return nil, err
}
cell := readLeafCell(leafPage, elem.index)
other.Containers.Put(cell.Key, toContainer(cell, tx))
}
}
@ -629,7 +634,14 @@ func (tx *Tx) container(name string, key uint64) (*roaring.Container, error) {
return nil, err
}
return toContainer(c.cell(), tx), nil
elem := &c.stack.elems[c.stack.index]
leafPage, _, err := c.tx.readPage(elem.pgno)
if err != nil {
return nil, err
}
cell := readLeafCell(leafPage, elem.index)
return toContainer(cell, tx), nil
}
// PutContainer inserts a container into a bitmap. Overwrites if key already exists.
@ -733,7 +745,7 @@ func (tx *Tx) checkPageAllocations() error {
if isInuse && isFree {
return fmt.Errorf("page in-use & free: pgno=%d", pgno)
} else if !isInuse && !isFree {
page, err := tx.readPage(pgno)
page, _, err := tx.readPage(pgno)
if err != nil {
return err
}
@ -767,7 +779,13 @@ func (tx *Tx) freePageSet() (map[uint32]struct{}, error) {
return m, err
}
cell := c.cell()
elem := &c.stack.elems[c.stack.index]
leafPage, _, err := c.tx.readPage(elem.pgno)
if err != nil {
return nil, err
}
cell := readLeafCell(leafPage, elem.index)
for _, v := range cell.Values(tx) {
pgno := uint32((cell.Key << 16) & uint64(v))
m[pgno] = struct{}{}
@ -784,7 +802,7 @@ func (tx *Tx) inusePageSet() (map[uint32]struct{}, error) {
for pgno := readMetaRootRecordPageNo(tx.meta[:]); pgno != 0; {
m[pgno] = struct{}{}
page, err := tx.readPage(pgno)
page, _, err := tx.readPage(pgno)
if err != nil {
return nil, err
}
@ -822,7 +840,7 @@ func (tx *Tx) inusePageSet() (map[uint32]struct{}, error) {
// walkTree recursively iterates over a page and all its children.
func (tx *Tx) walkTree(pgno, parent uint32, fn func(pgno, parent, typ uint32) error) error {
// Read page and iterate over children.
page, err := tx.readPage(pgno)
page, _, err := tx.readPage(pgno)
if err != nil {
return err
}
@ -892,7 +910,13 @@ func (tx *Tx) nextFreelistPageNo() (uint32, error) {
return 0, err
}
cell := c.cell()
elem := &c.stack.elems[c.stack.index]
leafPage, _, err := c.tx.readPage(elem.pgno)
if err != nil {
return 0, err
}
cell := readLeafCell(leafPage, elem.index)
v := cell.firstValue(tx)
pgno := uint32((cell.Key << 16) | uint64(v))
@ -914,7 +938,7 @@ func (tx *Tx) freePgno(pgno uint32) error {
// deallocateTree recursively all pages in a btree.
func (tx *Tx) deallocateTree(pgno uint32) error {
page, err := tx.readPage(pgno)
page, _, err := tx.readPage(pgno)
if err != nil {
return err
}
@ -936,34 +960,36 @@ func (tx *Tx) deallocateTree(pgno uint32) error {
}
}
func (tx *Tx) readPage(pgno uint32) ([]byte, error) {
func (tx *Tx) readPage(pgno uint32) (_ []byte, isHeap bool, err error) {
// Meta page is always cached on the transaction.
if pgno == 0 {
return tx.meta[:], nil
return tx.meta[:], false, nil
}
// Verify page number requested is within current size of database.
pageN := readMetaPageN(tx.meta[:])
if pgno > pageN {
return nil, fmt.Errorf("rbf: page read out of bounds: pgno=%d max=%d", pgno, pageN)
return nil, false, fmt.Errorf("rbf: page read out of bounds: pgno=%d max=%d", pgno, pageN)
}
// Check if page has been updated in this tx.
if tx.writable {
if page := tx.dirtyPages[pgno]; page != nil {
return page, nil
return page, true, nil
} else if page := tx.dirtyBitmapPages[pgno]; page != nil {
return page, nil
return page, true, nil
}
}
// Check if page is remapped in WAL.
if walID, ok := tx.pageMap.Get(pgno); ok {
return tx.db.readWALPageByID(walID)
buf, err := tx.db.readWALPageByID(walID)
return buf, false, err
}
// Otherwise read directly from DB.
return tx.db.readDBPage(pgno)
buf, err := tx.db.readDBPage(pgno)
return buf, false, err
}
func (tx *Tx) writePage(page []byte) error {
@ -1001,7 +1027,7 @@ func (tx *Tx) AddRoaring(name string, bm *roaring.Bitmap) (changed bool, err err
}
func (tx *Tx) leafCellBitmap(pgno uint32) (uint32, []uint64, error) {
page, err := tx.readPage(pgno)
page, _, err := tx.readPage(pgno)
if err != nil {
return 0, nil, err
}
@ -1053,7 +1079,14 @@ func (tx *Tx) ForEachRange(name string, start, end uint64, fn func(uint64) error
return err
}
switch cell := c.cell(); cell.Type {
elem := &c.stack.elems[c.stack.index]
leafPage, _, err := c.tx.readPage(elem.pgno)
if err != nil {
return err
}
cell := readLeafCell(leafPage, elem.index)
switch cell.Type {
case ContainerTypeArray:
for _, lo := range toArray16(cell.Data) {
v := cell.Key<<16 | uint64(lo)
@ -1146,7 +1179,14 @@ func (tx *Tx) Count(name string) (uint64, error) {
return 0, err
}
n += uint64(c.cell().BitN)
elem := &c.stack.elems[c.stack.index]
leafPage, _, err := c.tx.readPage(elem.pgno)
if err != nil {
return 0, err
}
cell := readLeafCell(leafPage, elem.index)
n += uint64(cell.BitN)
}
return n, nil
}
@ -1169,7 +1209,13 @@ func (tx *Tx) Max(name string) (uint64, error) {
return 0, err
}
cell := c.cell()
elem := &c.stack.elems[c.stack.index]
leafPage, _, err := c.tx.readPage(elem.pgno)
if err != nil {
return 0, err
}
cell := readLeafCell(leafPage, elem.index)
return uint64((cell.Key << 16) | uint64(cell.lastValue(tx))), nil
}
@ -1191,7 +1237,13 @@ func (tx *Tx) Min(name string) (uint64, bool, error) {
return 0, false, err
}
cell := c.cell()
elem := &c.stack.elems[c.stack.index]
leafPage, _, err := c.tx.readPage(elem.pgno)
if err != nil {
return 0, false, err
}
cell := readLeafCell(leafPage, elem.index)
return uint64((cell.Key << 16) | uint64(cell.firstValue(tx))), true, nil
}
@ -1252,7 +1304,13 @@ func (tx *Tx) CountRange(name string, start, end uint64) (uint64, error) {
return 0, err
}
c := csr.cell()
elem := &csr.stack.elems[csr.stack.index]
leafPage, _, err := csr.tx.readPage(elem.pgno)
if err != nil {
return 0, err
}
c := readLeafCell(leafPage, elem.index)
k := c.Key
if k > ekey {
break
@ -1324,7 +1382,12 @@ func (tx *Tx) OffsetRange(name string, offset, start, endx uint64) (*roaring.Bit
return nil, err
}
cell := c.cell()
elem := &c.stack.elems[c.stack.index]
leafPage, _, err := c.tx.readPage(elem.pgno)
if err != nil {
return nil, err
}
cell := readLeafCell(leafPage, elem.index)
ckey := cell.Key
// >= hi1 is correct b/c endx cannot have any lowbits set.
@ -1356,7 +1419,9 @@ func (itr *containerIterator) Next() bool {
// Value returns the current key & container.
func (itr *containerIterator) Value() (uint64, *roaring.Container) {
cell := itr.cursor.cell()
elem := &itr.cursor.stack.elems[itr.cursor.stack.index]
leafPage, _, _ := itr.cursor.tx.readPage(elem.pgno)
cell := readLeafCell(leafPage, elem.index)
return cell.Key, toContainer(cell, itr.cursor.tx)
}
@ -1404,7 +1469,12 @@ func (tx *Tx) DumpString(short bool, shard uint64) (r string) {
break
}
panicOn(err)
cell := c.cell()
elem := &c.stack.elems[c.stack.index]
leafPage, _, err := c.tx.readPage(elem.pgno)
panicOn(err)
cell := readLeafCell(leafPage, elem.index)
ckey := cell.Key
ct := toContainer(cell, tx)
@ -1532,7 +1602,13 @@ 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 {
oldC = toContainer(cur.cell(), tx)
elem := &cur.stack.elems[cur.stack.index]
leafPage, _, err := cur.tx.readPage(elem.pgno)
if err != nil {
return changed, rowSet, err
}
cell := readLeafCell(leafPage, elem.index)
oldC = toContainer(cell, tx)
}
if oldC == nil || oldC.N() == 0 {
@ -1687,7 +1763,7 @@ func (tx *Tx) Pages(pgnos []uint32) ([]Page, error) {
// Loop over each requested page number and extract additional data.
var pages []Page
for _, pgno := range pgnos {
buf, err := tx.readPage(pgno)
buf, _, err := tx.readPage(pgno)
if err != nil {
return nil, err
}
@ -1805,7 +1881,7 @@ func (tx *Tx) PageInfos() ([]PageInfo, error) {
// metaPageInfo returns page metadata for the meta page.
func (tx *Tx) metaPageInfo() (*MetaPageInfo, error) {
buf, err := tx.readPage(0)
buf, _, err := tx.readPage(0)
if err != nil {
return nil, err
}
@ -1822,7 +1898,7 @@ func (tx *Tx) metaPageInfo() (*MetaPageInfo, error) {
// rootRecordPageInfo returns page metadata for a root record page.
func (tx *Tx) rootRecordPageInfo(pgno uint32) (*RootRecordPageInfo, error) {
buf, err := tx.readPage(pgno)
buf, _, err := tx.readPage(pgno)
if err != nil {
return nil, err
}
@ -1835,7 +1911,7 @@ func (tx *Tx) rootRecordPageInfo(pgno uint32) (*RootRecordPageInfo, error) {
func (tx *Tx) walkPageInfo(infos []PageInfo, root uint32, name string) error {
return tx.walkTree(root, 0, func(pgno, parent, typ uint32) error {
buf, err := tx.readPage(pgno)
buf, _, err := tx.readPage(pgno)
if err != nil {
return err
}
@ -1873,7 +1949,8 @@ func (tx *Tx) walkPageInfo(infos []PageInfo, root uint32, name string) error {
// PageData returns the raw page data for a single page.
func (tx *Tx) PageData(pgno uint32) ([]byte, error) {
return tx.readPage(pgno)
buf, _, err := tx.readPage(pgno)
return buf, err
}
type PageInfo interface {

2
rbf/util_test.go
View file

@ -46,7 +46,7 @@ func (c_orig *Cursor) DebugSlowCheckAllPages() {
// checkElemNBitN recursively writes the tree representation starting from a given page to STDERR.
func checkElemNBitN(tx *Tx, pgno uint32) {
page, err := tx.readPage(pgno)
page, _, err := tx.readPage(pgno)
if err != nil {
panic(err)
}