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

Merge pull request #526 from molecula/rbf

Browse source 
Roaring Bitmap Format
This commit is contained in:
Ben Johnson 2020-07-08 13:39:37 -06:00 • committed by GitHub
commit 3c021fc2e3
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
28 changed files with 5891 additions and 49 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
.circleci/config.yml
View file

@ -13,8 +13,6 @@ executors:
- image: circleci/golang:<< parameters.version >>
resource_class: << parameters.resource_class >>
working_directory: /go/src/github.com/pilosa/pilosa
environment:
GO111MODULE: "on" # TODO: Only needed for Go <1.13, remove when dropping support for 1.11/1.12.
commands:
add-github-auth:
@ -177,7 +175,7 @@ workflows:
name: test-golang-<< matrix.golang_version >>
matrix:
parameters:
golang_version: ["1.14", "1.13", "1.12", "1.11"]
golang_version: ["1.14", "1.13"]
requires:
- setup
filters:

44
cmd/convert/main.go Normal file
View file

@ -0,0 +1,44 @@
// 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 main
import (
"log"
"os"
"github.com/pilosa/pilosa/v2"
"github.com/pilosa/pilosa/v2/rbf"
)
func main() {
if len(os.Args) != 3 {
log.Fatal("USAGE convert srcPath destPath")
}
holder := pilosa.NewHolder(256)
holder.Path = os.Args[1]
err := holder.Open()
if err != nil {
log.Fatal(err)
}
c := &pilosa.RBFConverter{
Dbs: make(map[string]*rbf.DB),
Base: os.Args[2],
}
holder.ConvertToRBF(c)
}

23
field.go
View file

@ -85,11 +85,12 @@ var availableShardFileFlushDuration = &protected{
// Field represents a container for views.
type Field struct {
mu sync.RWMutex
createdAt int64
path string
index string
name string
mu sync.RWMutex
createdAt int64
path string
index string
name string
qualifiedName string
viewMap map[string]*view
@ -352,9 +353,10 @@ func newField(holder *Holder, path, index, name string, opts FieldOption) (*Fiel
}
f := &Field{
path: path,
index: index,
name: name,
path: path,
index: index,
name: name,
qualifiedName: FormatQualifiedFieldName(index, name),
viewMap: make(map[string]*view),
@ -2147,3 +2149,8 @@ func bitDepthInt64(v int64) uint {
}
return bitDepth(uint64(v))
}
// FormatQualifiedFieldName generates a qualified name for the field to be used with Tx operations.
func FormatQualifiedFieldName(index, field string) string {
return fmt.Sprintf("%s\x00%s\x00", index, field)
}

19
fragment.go
View file

@ -30,6 +30,7 @@ import (
"os"
"runtime/debug"
"sort"
"strconv"
"strings"
"sync"
"syscall"
@ -3611,3 +3612,21 @@ func (v *boolVector) Get(tx Tx, colID uint64) (uint64, bool, error) {
}
return 0, false, nil
}
// FormatQualifiedFragmentName generates a qualified name for the fragment to be used with Tx operations.
func FormatQualifiedFragmentName(index, field, view string, shard uint64) string {
return fmt.Sprintf("%s\x00%s\x00%s\x00%d", index, field, view, shard)
}
// ParseQualifiedFragmentName parses a qualified name into its parts.
func ParseQualifiedFragmentName(name string) (index, field, view string, shard uint64, err error) {
a := strings.Split(name, "\x00")
if len(a) < 4 {
return "", "", "", 0, fmt.Errorf("invalid qualified name: %q", name)
}
index, field, view = string(a[0]), string(a[1]), string(a[2])
if shard, err = strconv.ParseUint(a[3], 10, 64); err != nil {
return "", "", "", 0, fmt.Errorf("invalid qualified name: %q", name)
}
return index, field, view, shard, nil
}

48
generation.go
View file

@ -20,7 +20,7 @@ import (
"io/ioutil"
"os"
"runtime"
"runtime/debug"
// "runtime/debug"
"sync"
"syscall"
"time"
@ -169,29 +169,29 @@ func (m *mmapGeneration) Transaction(fileP *io.Writer, fn func() error) (transac
}
// We are done locking the generation itself for now.
m.mu.Unlock()
wouldPanic := debug.SetPanicOnFault(true)
defer func() {
debug.SetPanicOnFault(wouldPanic)
if r := recover(); r != nil {
if err, ok := r.(error); ok {
// special case: if we caught a page fault, we diagnose that directly. sadly,
// we can't see the actual values that were used to generate this, probably.
if err.Error() == "runtime error: invalid memory address or nil pointer dereference" {
if transactionErr == nil {
transactionErr = errors.New("invalid memory access during transaction")
} else {
transactionErr = fmt.Errorf("invalid memory access during transaction, previous error %v", transactionErr)
}
return
}
}
if transactionErr == nil {
transactionErr = fmt.Errorf("panic during transaction: %v", r)
} else {
transactionErr = fmt.Errorf("panic during erroring transaction: panic %v, previous error %v", r, transactionErr)
}
}
}()
// wouldPanic := debug.SetPanicOnFault(true)
// defer func() {
// debug.SetPanicOnFault(wouldPanic)
// if r := recover(); r != nil {
// if err, ok := r.(error); ok {
// // special case: if we caught a page fault, we diagnose that directly. sadly,
// // we can't see the actual values that were used to generate this, probably.
// if err.Error() == "runtime error: invalid memory address or nil pointer dereference" {
// if transactionErr == nil {
// transactionErr = errors.New("invalid memory access during transaction")
// } else {
// transactionErr = fmt.Errorf("invalid memory access during transaction, previous error %v", transactionErr)
// }
// return
// }
// }
// if transactionErr == nil {
// transactionErr = fmt.Errorf("panic during transaction: %v", r)
// } else {
// transactionErr = fmt.Errorf("panic during erroring transaction: panic %v, previous error %v", r, transactionErr)
// }
// }
// }()
return fn()
}

19
index.go
View file

@ -35,11 +35,12 @@ import (
// Index represents a container for fields.
type Index struct {
mu sync.RWMutex
createdAt int64
path string
name string
keys bool // use string keys
mu sync.RWMutex
createdAt int64
path string
name string
qualifiedName string
keys bool // use string keys
// Existence tracking.
trackExistence bool
@ -106,6 +107,9 @@ func (i *Index) CreatedAt() int64 {
// Name returns name of the index.
func (i *Index) Name() string { return i.name }
// QualifiedName returns the qualified name of the index.
func (i *Index) QualifiedName() string { return i.qualifiedName }
// Path returns the path the index was initialized with.
func (i *Index) Path() string { return i.path }
@ -611,3 +615,8 @@ type importValueData struct {
ColumnIDs []uint64
Values []int64
}
// FormatQualifiedIndexName generates a qualified name for the index to be used with Tx operations.
func FormatQualifiedIndexName(index string) string {
return fmt.Sprintf("%s\x00", index)
}

BIN
pilosa Executable file
View file

Binary file not shown.

110
rbf/README.md Normal file
View file

@ -0,0 +1,110 @@
Roaring B-tree Format
=====================
The RBF format represents a Roaring bitmap whose containers are stored in the
leafs of a b-tree. This allows the bitmap to be efficiently queried & updated.
## File Format
The RBF file is divided into equal 8KB pages. Each page after the meta page
is numbered incrementally from 1 to 1^31.
Pages can be one of the following types:
- Meta page: contains header information.
- Branch page: contains pointers to lower branch & leaf pages.
- Leaf page: contains array and RLE container data.
- Bitmap page: contains bitmap container data.
All integer values are little endian encoded.
## Page header
Every page type except the bitmap page contains the following header:
### Meta page
The meta page contains the following header:
[4] magic (\xFFRBF)
[4] flags
[4] page count
[8] wal ID
[4] root records pgno
[4] freelist pgno
### Root Records page
A list of all b-tree names & their respective root page numbers are stored in
root record pages. Once a bitmap root is created, it is never moved so the
root record pages only need to be rewritten when creating, renaming, or deleting
a b-tree. If records exceed the size of a page then they are overflowed to
additional pages.
[4] page number
[4] flags
[4] overflow pgno
[*] bitmap records
Each bitmap record is represented as:
[4] pgno
[2] name size
[*] name
All bitmap records are loaded into memory when the file is opened.
### Branch page
The branch page contains the following header:
[4] page number
[4] flags
[2] cell count
[*] cell index (2 * cell count)
[*] padding for 4-byte alignment
Each cell is formatted as:
[8] highbits
[4] flags
[4] page number
### Leaf page
The leaf page contains the following header:
[4] page number
[4] flags
[2] cell count
[*] cell index (2 * cell count)
The leaf page contains a series of cells with the header of:
[8] highbits
[4] flag
[4] child count
[*] array or RLE data
### Bitmap page
The data for the bitmap page takes up the entire 8KB.
## Proof of Concept Notes
The following are notes made that are temporary for the RBF format. This will
change as development progresses:
- Transaction support is deferred
- WAL support is deferred

69
rbf/array.go Normal file
View file

@ -0,0 +1,69 @@
// 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 (
"unsafe"
"github.com/pilosa/pilosa/v2/roaring"
)
// toArray16 converts a byte slice into a slice of uint16 values using unsafe.
func toArray16(a []byte) []uint16 {
return (*[4096]uint16)(unsafe.Pointer(&a[0]))[: len(a)/2 : len(a)/2]
}
// fromArray16 converts a slice of uint16 values into a byte slice using unsafe.
func fromArray16(a []uint16) []byte {
return (*[8192]byte)(unsafe.Pointer(&a[0]))[: len(a)*2 : len(a)*2]
}
// arrayIndex returns the insertion index of v in a. Returns true if exact match.
func arrayIndex(a []uint16, v uint16) (int, bool) {
return search(len(a), func(i int) int {
if a[i] == v {
return 0
} else if v < a[i] {
return -1
}
return 1
})
}
// toArray64 converts a byte slice into a slice of uint64 values using unsafe.
func toArray64(a []byte) []uint64 {
return (*[1024]uint64)(unsafe.Pointer(&a[0]))[:1024:1024]
}
// fromArray64 converts a slice of uint64 values into a byte slice using unsafe.
func fromArray64(a []uint64) []byte {
return (*[8192]byte)(unsafe.Pointer(&a[0]))[:8192:8192]
}
func cloneArray64(a []uint64) []uint64 {
other := make([]uint64, len(a))
copy(other, a)
return other
}
// toArray16 converts a byte slice into a slice of uint16 values using unsafe.
func toInterval16(a []byte) []roaring.Interval16 {
return (*[2048]roaring.Interval16)(unsafe.Pointer(&a[0]))[: len(a)/4 : len(a)/4]
}
// fromArray16 converts a slice of uint16 values into a byte slice using unsafe.
func fromInterval16(a []roaring.Interval16) []byte {
return (*[8192]byte)(unsafe.Pointer(&a[0]))[: len(a)*4 : len(a)*4]
}

1178
rbf/cursor.go Normal file
View file

File diff suppressed because it is too large Load diff

799
rbf/cursor_test.go Normal file
View file

@ -0,0 +1,799 @@
// 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_test
import (
"math/bits"
"math/rand"
"reflect"
"sort"
"testing"
"github.com/pilosa/pilosa/v2/rbf"
"github.com/pilosa/pilosa/v2/roaring"
)
func TestCursor_FirstNext(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)
} else if _, err := tx.Add("x", 0x00000001, 0x00000002, 0x00010003, 0x00030004); err != nil {
t.Fatal(err)
}
c, err := tx.Cursor("x")
if err != nil {
t.Fatal(err)
} else if err := c.First(); err != nil {
t.Fatal(err)
}
if err := c.Next(); err != nil {
t.Fatal(err)
} else if got, want := c.Key(), uint64(0); got != want {
t.Fatalf("Next()=%d, want %d", got, want)
} else if got, want := c.Values(), []uint16{1, 2}; !reflect.DeepEqual(got, want) {
t.Fatalf("Values()=%#v, want %#v", got, want)
}
if err := c.Next(); err != nil {
t.Fatal(err)
} else if got, want := c.Key(), uint64(1); got != want {
t.Fatalf("Next()=%d, want %d", got, want)
} else if got, want := c.Values(), []uint16{3}; !reflect.DeepEqual(got, want) {
t.Fatalf("Values()=%#v, want %#v", got, want)
}
if err := c.Next(); err != nil {
t.Fatal(err)
} else if got, want := c.Key(), uint64(3); got != want {
t.Fatalf("Next()=%d, want %d", got, want)
} else if got, want := c.Values(), []uint16{4}; !reflect.DeepEqual(got, want) {
t.Fatalf("Values()=%#v, want %#v", got, want)
}
}
func TestCursor_FirstNext_Quick(t *testing.T) {
if testing.Short() {
t.Skip("-short enabled, skipping")
} else if is32Bit() {
t.Skip("32-bit build, skipping quick check tests")
} else if rbf.RaceEnabled {
t.Skip("race detection enabled, skipping")
}
const n = 100000
QuickCheck(t, func(t *testing.T, rand *rand.Rand) {
t.Parallel()
// Generate sorted list of values.
values := make([]uint64, rand.Intn(n))
for i := range values {
values[i] = uint64(rand.Intn(rbf.ShardWidth))
}
sort.Slice(values, func(i, j int) bool { return values[i] < values[j] })
db := MustOpenDB(t)
defer MustCloseDB(t, db)
tx := MustBegin(t, db, true)
defer MustRollback(t, tx)
// Insert values in random order.
if err := tx.CreateBitmap("x"); err != nil {
t.Fatal(err)
}
for _, i := range rand.Perm(len(values)) {
v := values[i]
if _, err := tx.Add("x", v); err != nil {
t.Fatalf("Add(%d) i=%d err=%q", v, i, err)
}
}
// Generate unique bucketed values.
type Item struct {
key uint64
values []uint16
}
var items []Item
m := make(map[uint64]struct{})
for _, v := range values {
if _, ok := m[v]; ok {
continue
}
m[v] = struct{}{}
hi, lo := highbits(v), lowbits(v)
if len(items) == 0 || items[len(items)-1].key != hi {
items = append(items, Item{key: hi})
}
item := &items[len(items)-1]
item.values = append(item.values, lo)
}
// Verify cursor returns correct value groups.
c, err := tx.Cursor("x")
if err != nil {
t.Fatal(err)
} else if err := c.First(); err != nil {
t.Fatal(err)
}
for _, item := range items {
if err := c.Next(); err != nil {
t.Fatal(err)
} else if got, want := c.Key(), item.key; got != want {
t.Fatalf("Key()=%d, want %d", got, want)
} else if got, want := c.Values(), item.values; !reflect.DeepEqual(got, want) {
t.Fatalf("len(Values())=%v, want %v", len(got), len(want))
}
}
})
}
func TestCursor_LastPrev(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)
} else if _, err := tx.Add("x", 0x00000001, 0x00000002, 0x00010003, 0x00030004); err != nil {
t.Fatal(err)
}
c, err := tx.Cursor("x")
if err != nil {
t.Fatal(err)
} else if err := c.Last(); err != nil {
t.Fatal(err)
}
if err := c.Prev(); err != nil {
t.Fatal(err)
} else if got, want := c.Key(), uint64(3); got != want {
t.Fatalf("Prev()=%d, want %d", got, want)
} else if got, want := c.Values(), []uint16{4}; !reflect.DeepEqual(got, want) {
t.Fatalf("Values()=%#v, want %#v", got, want)
}
if err := c.Prev(); err != nil {
t.Fatal(err)
} else if got, want := c.Key(), uint64(1); got != want {
t.Fatalf("Prev()=%d, want %d", got, want)
} else if got, want := c.Values(), []uint16{3}; !reflect.DeepEqual(got, want) {
t.Fatalf("Values()=%#v, want %#v", got, want)
}
if err := c.Prev(); err != nil {
t.Fatal(err)
} else if got, want := c.Key(), uint64(0); got != want {
t.Fatalf("Prev()=%d, want %d", got, want)
} else if got, want := c.Values(), []uint16{1, 2}; !reflect.DeepEqual(got, want) {
t.Fatalf("Values()=%#v, want %#v", got, want)
}
}
func TestCursor_LastPrev_Quick(t *testing.T) {
if testing.Short() {
t.Skip("-short enabled, skipping")
} else if is32Bit() {
t.Skip("32-bit build, skipping quick check tests")
} else if rbf.RaceEnabled {
t.Skip("race detection enabled, skipping")
}
const n = 100000
QuickCheck(t, func(t *testing.T, rand *rand.Rand) {
t.Parallel()
// Generate sorted list of values.
values := make([]uint64, n)
for i := range values {
values[i] = uint64(rand.Intn(rbf.ShardWidth))
}
sort.Slice(values, func(i, j int) bool { return values[i] < values[j] })
db := MustOpenDB(t)
defer MustCloseDB(t, db)
tx := MustBegin(t, db, true)
defer MustRollback(t, tx)
// Insert values in random order.
if err := tx.CreateBitmap("x"); err != nil {
t.Fatal(err)
}
for _, i := range rand.Perm(len(values)) {
v := values[i]
if _, err := tx.Add("x", v); err != nil {
t.Fatalf("Add(%d) i=%d err=%q", v, i, err)
}
}
// Generate unique bucketed values.
type Item struct {
key uint64
values []uint16
}
var items []Item
m := make(map[uint64]struct{})
for _, v := range values {
if _, ok := m[v]; ok {
continue
}
m[v] = struct{}{}
hi, lo := highbits(v), lowbits(v)
if len(items) == 0 || items[len(items)-1].key != hi {
items = append(items, Item{key: hi})
}
item := &items[len(items)-1]
item.values = append(item.values, lo)
}
// Verify cursor returns correct value groups.
c, err := tx.Cursor("x")
if err != nil {
t.Fatal(err)
} else if err := c.Last(); err != nil {
t.Fatal(err)
}
for i := len(items) - 1; i >= 0; i-- {
if err := c.Prev(); err != nil {
t.Fatal(err)
} else if got, want := c.Key(), items[i].key; got != want {
t.Fatalf("Key()=%d, want %d", got, want)
} else if got, want := c.Values(), items[i].values; !reflect.DeepEqual(got, want) {
t.Fatalf("len(Values())=%v, want %v", len(got), len(want))
}
}
})
}
func TestCursor_Union(t *testing.T) {
t.Run("OK", func(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)
}
row := make([]uint64, rbf.ShardWidth/64)
if _, err := tx.Add("x", 1, 3); err != nil {
t.Fatal(err)
} else if _, err := tx.Add("x", rbf.ShardWidth+1, rbf.ShardWidth+2, rbf.ShardWidth+7); err != nil {
t.Fatal(err)
}
c, err := tx.Cursor("x")
if err != nil {
t.Fatal(err)
}
if err := c.Union(0, row); err != nil {
t.Fatal(err)
} else if row[0] != 0b00001010 {
t.Fatalf("unexpected row[0]: 0b%b", row[0])
}
if err := c.Union(1, row); err != nil {
t.Fatal(err)
} else if row[0] != 0b10001110 {
t.Fatalf("unexpected row[0]: 0b%b", row[0])
}
})
t.Run("Quick", func(t *testing.T) {
if testing.Short() {
t.Skip("-short enabled, skipping")
} else if is32Bit() {
t.Skip("32-bit build, skipping quick check tests")
} else if rbf.RaceEnabled {
t.Skip("race detection enabled, skipping")
}
QuickCheck(t, func(t *testing.T, rand *rand.Rand) {
t.Parallel()
db := MustOpenDB(t)
defer MustCloseDB(t, db)
tx := MustBegin(t, db, true)
defer MustRollback(t, tx)
values := GenerateValues(rand, 100000)
rows := ToRows(values)
if err := tx.CreateBitmap("x"); err != nil {
t.Fatal(err)
}
MustAddRandom(t, rand, tx, "x", values...)
// Iterate over rows and randomly choose another row to union.
for i, row0 := range rows {
row1 := rows[rand.Intn(len(rows))]
bitmap := row0.Bitmap()
c, err := tx.Cursor("x")
if err != nil {
t.Fatal(err)
} else if err := c.Union(row1.ID, bitmap); err != nil {
return
}
if got, want := len(rbf.RowValues(bitmap)), len(row0.Union(row1)); got != want {
t.Fatalf("%d. len()=%d, want %d", i, got, want)
}
}
})
})
}
func TestCursor_Intersect(t *testing.T) {
t.Run("OK", func(t *testing.T) {
db := MustOpenDB(t)
defer MustCloseDB(t, db)
tx := MustBegin(t, db, true)
defer MustRollback(t, tx)
row := make([]uint64, rbf.ShardWidth/64)
if err := tx.CreateBitmap("x"); err != nil {
t.Fatal(err)
}
if _, err := tx.Add("x", 1, 3); err != nil {
t.Fatal(err)
} else if _, err := tx.Add("x", rbf.ShardWidth+1, rbf.ShardWidth+2, rbf.ShardWidth+7); err != nil {
t.Fatal(err)
}
c, err := tx.Cursor("x")
if err != nil {
t.Fatal(err)
}
if err := c.Union(0, row); err != nil {
t.Fatal(err)
} else if row[0] != 0b00001010 {
t.Fatalf("unexpected row[0]: %#v", row[0])
}
if err := c.Intersect(1, row); err != nil {
t.Fatal(err)
} else if row[0] != 0b00000010 {
t.Fatalf("unexpected row[0]: %#v", row[0])
}
})
t.Run("Quick", func(t *testing.T) {
if testing.Short() {
t.Skip("-short enabled, skipping")
} else if is32Bit() {
t.Skip("32-bit build, skipping quick check tests")
} else if rbf.RaceEnabled {
t.Skip("race detection enabled, skipping")
}
QuickCheck(t, func(t *testing.T, rand *rand.Rand) {
t.Parallel()
db := MustOpenDB(t)
defer MustCloseDB(t, db)
tx := MustBegin(t, db, true)
defer MustRollback(t, tx)
values := GenerateValues(rand, rand.Intn(100000))
rows := ToRows(values)
if err := tx.CreateBitmap("x"); err != nil {
t.Fatal(err)
}
MustAddRandom(t, rand, tx, "x", values...)
// Iterate over rows and randomly choose another row to union.
for i, row0 := range rows {
row1 := rows[rand.Intn(len(rows))]
bitmap := row0.Bitmap()
if c, err := tx.Cursor("x"); err != nil {
t.Fatal(err)
} else if err := c.Intersect(row1.ID, bitmap); err != nil {
t.Fatal(err)
}
if got, want := len(rbf.RowValues(bitmap)), len(row0.Intersect(row1)); got != want {
t.Fatalf("%d. len()=%d, want %d", i, got, want)
}
}
})
})
}
func makeBitmap(bit []uint16) (n int, ret []uint64) {
ret = make([]uint64, 1024)
for _, v := range bit {
ret[v/64] |= 1 << uint64(v%64)
}
n = 0
for _, v := range ret {
n += bits.OnesCount64(v)
}
return
}
func TestCursor_AddRoaring(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)
}
tests := []struct {
name string
fieldview string
rb *roaring.Bitmap
wantChanged bool
wantErr bool
}{{
name: "no view",
fieldview: "a/standard",
rb: func() *roaring.Bitmap {
bm := roaring.NewBitmap()
return bm
}(),
wantChanged: false,
wantErr: true},
{
name: "initial Array",
fieldview: "x",
rb: func() *roaring.Bitmap {
bm := roaring.NewBitmap()
bm.Put(0, roaring.NewContainerArray([]uint16{1, 2}))
return bm
}(),
wantChanged: true,
wantErr: false}, {
name: "initial RLE",
fieldview: "x",
rb: func() *roaring.Bitmap {
bm := roaring.NewBitmap()
bm.Put(1, roaring.NewContainerRun([]roaring.Interval16{{Start: 10, Last: 20000}}))
return bm
}(),
wantChanged: true,
wantErr: false},
{
name: "initial Bitmap",
fieldview: "x",
rb: func() *roaring.Bitmap {
bm := roaring.NewBitmap()
bm.Put(3, roaring.NewContainerBitmap(makeBitmap([]uint16{4, 8, 12})))
return bm
}(),
wantChanged: true,
wantErr: false},
{
name: "merge Array exist",
fieldview: "x",
rb: func() *roaring.Bitmap {
bm := roaring.NewBitmap()
bm.Put(0, roaring.NewContainerArray([]uint16{1, 2}))
return bm
}(),
wantChanged: false,
wantErr: false}, {
name: "merge Array present",
fieldview: "x",
rb: func() *roaring.Bitmap {
bm := roaring.NewBitmap()
bm.Put(0, roaring.NewContainerArray([]uint16{3, 4}))
return bm
}(),
wantChanged: true,
wantErr: false},
{
name: "merge Bitmap exist",
fieldview: "x",
rb: func() *roaring.Bitmap {
bm := roaring.NewBitmap()
bm.Put(3, roaring.NewContainerBitmap(makeBitmap([]uint16{4, 8, 12})))
return bm
}(),
wantChanged: false,
wantErr: false},
{
name: "merge Bitmap ",
fieldview: "x",
rb: func() *roaring.Bitmap {
bm := roaring.NewBitmap()
bm.Put(3, roaring.NewContainerBitmap(makeBitmap([]uint16{75})))
return bm
}(),
wantChanged: true,
wantErr: false},
{
name: "merge BitmapArray ",
fieldview: "x",
rb: func() *roaring.Bitmap {
bm := roaring.NewBitmap()
bm.Put(0, roaring.NewContainerBitmap(makeBitmap([]uint16{75})))
return bm
}(),
wantChanged: true,
wantErr: false}, {
name: "too Big Array ",
fieldview: "x",
rb: func() *roaring.Bitmap {
bm := roaring.NewBitmap()
items := make([]uint16, rbf.ArrayMaxSize+2)
for i := 0; i < len(items); i++ {
items[i] = uint16(i)
}
bm.Put(10, roaring.NewContainerArray(items))
return bm
}(),
wantChanged: true,
wantErr: false},
{
name: "too Big RLE ",
fieldview: "x",
rb: func() *roaring.Bitmap {
bm := roaring.NewBitmap()
items := make([]roaring.Interval16, rbf.RLEMaxSize+2)
x := uint16(0)
for i := 0; i < len(items); i++ {
v := roaring.Interval16{Start: x, Last: x + 1}
x += 3
items[i] = v
}
bm.Put(10, roaring.NewContainerRun(items))
return bm
}(),
wantChanged: true,
wantErr: false}, {
name: "empty container ",
fieldview: "x",
rb: func() *roaring.Bitmap {
bm := roaring.NewBitmap()
bm.Put(11, roaring.NewContainerArray([]uint16{}))
return bm
}(),
wantChanged: false,
wantErr: false},
{
name: "merge RLE",
fieldview: "x",
rb: func() *roaring.Bitmap {
bm := roaring.NewBitmap()
bm.Put(1, roaring.NewContainerRun([]roaring.Interval16{{Start: 1, Last: 12}}))
return bm
}(),
wantChanged: true,
wantErr: false},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
gotChanged, err := tx.AddRoaring(tt.fieldview, tt.rb)
if (err != nil) != tt.wantErr {
t.Errorf("Cursor.AddRoaring() error = %v, wantErr %v", err, tt.wantErr)
return
}
if gotChanged != tt.wantChanged {
t.Errorf("Cursor.AddRoaring() = %v, want %v", gotChanged, tt.wantChanged)
}
})
}
}
func TestCursor_RLETesting(t *testing.T) {
db := MustOpenDB(t)
defer MustCloseDB(t, db)
tx := MustBegin(t, db, true)
defer MustRollback(t, tx)
//setup RLE
if err := tx.CreateBitmap("x"); err != nil {
t.Fatal(err)
}
rb := func() *roaring.Bitmap {
bm := roaring.NewBitmap()
bm.Put(0, roaring.NewContainerRun([]roaring.Interval16{{Start: 10, Last: 11}}))
return bm
}()
_, err := tx.AddRoaring("x", rb)
if err != nil {
t.Errorf("Add Roaring Failed %v", err)
}
//
tests := []struct {
name string
args []uint64
want []uint16
wantChanged bool
wantErr bool
}{{
name: "update run at Last",
args: []uint64{0x0000000c},
want: []uint16{0x0000000a, 0x0000000b, 0x0000000c},
wantChanged: true,
wantErr: false,
},
{
name: "update run at begining",
args: []uint64{0x00000001, 0x00000002},
want: []uint16{0x00000001, 0x00000002, 0x0000000a, 0x0000000b, 0x0000000c},
wantChanged: true,
wantErr: false,
},
{
name: "add run at end",
args: []uint64{0x0000000f},
want: []uint16{0x00000001, 0x00000002, 0x0000000a, 0x0000000b, 0x0000000c, 0x0000000f},
wantChanged: true,
wantErr: false,
},
{
name: "no change",
args: []uint64{0x0000000b},
want: []uint16{0x00000001, 0x00000002, 0x0000000a, 0x0000000b, 0x0000000c, 0x0000000f},
wantChanged: false,
wantErr: false,
}, {
name: "update start",
args: []uint64{0x0000000e},
want: []uint16{0x00000001, 0x00000002, 0x0000000a, 0x0000000b, 0x0000000c, 0x0000000e, 0x0000000f},
wantChanged: true,
wantErr: false,
}, {
name: "combine",
args: []uint64{0x0000000d},
want: []uint16{0x00000001, 0x00000002, 0x0000000a, 0x0000000b, 0x0000000c, 0x0000000d, 0x0000000e, 0x0000000f},
wantChanged: true,
wantErr: false,
}, {
name: "add end",
args: []uint64{0x0000ffff},
want: []uint16{0x00000001, 0x00000002, 0x0000000a, 0x0000000b, 0x0000000c, 0x0000000d, 0x0000000e, 0x0000000f, 0x0000ffff},
wantChanged: true,
wantErr: false,
},
{
name: "overflow container",
args: []uint64{0x00010000},
want: []uint16{0x00000001, 0x00000002, 0x0000000a, 0x0000000b, 0x0000000c, 0x0000000d, 0x0000000e, 0x0000000f, 0x0000ffff},
wantChanged: true,
wantErr: false,
},
}
c, err := tx.Cursor("x")
if err != nil {
t.Fatal(err)
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
changed, err := tx.Add("x", tt.args...)
if tt.wantErr && err == nil {
t.Errorf("No Error %v", err)
} else if tt.wantChanged && !changed {
t.Errorf("No Change %v", err)
} else if err != nil {
t.Fatal(err)
} else if err := c.First(); err != nil {
t.Fatal(err)
}
if got, want := c.Values(), tt.want; !reflect.DeepEqual(got, want) {
t.Fatalf("Values()=%#v, want %#v", got, want)
}
})
}
t.Run("overflow followup", func(t *testing.T) {
//verify than next container got created and is valid
if err := c.Next(); err != nil { //skip the buffered?
t.Fatal(err)
}
if err := c.Next(); err != nil {
t.Fatal(err)
}
want := []uint16{0}
if got, want := c.Values(), want; !reflect.DeepEqual(got, want) {
t.Fatalf("Values()=%#v, want %#v", got, want)
} else if got, want := c.Key(), uint64(1); !reflect.DeepEqual(got, want) {
t.Fatalf("Key()=%#v, want %#v", got, want)
}
})
}
func TestCursor_RLEConversion(t *testing.T) {
db := MustOpenDB(t)
defer MustCloseDB(t, db)
tx := MustBegin(t, db, true)
defer MustRollback(t, tx)
//setup RLE with full container
if err := tx.CreateBitmap("x"); err != nil {
t.Fatal(err)
}
want := make([]uint16, 0, rbf.ArrayMaxSize)
rb := func() *roaring.Bitmap {
bm := roaring.NewBitmap()
runs := make([]roaring.Interval16, rbf.RLEMaxSize)
x := uint16(1)
for i := range runs {
runs[i] = roaring.Interval16{Start: x, Last: x + 1}
want = append(want, x)
want = append(want, x+1)
x += 3
}
bm.Put(0, roaring.NewContainerRun(runs))
return bm
}()
_, err := tx.AddRoaring("x", rb)
if err != nil {
t.Errorf("Add Roaring Failed %v", err)
}
c, err := tx.Cursor("x")
if err != nil {
t.Fatal(err)
} else if err := c.First(); err != nil {
t.Fatal(err)
}
if c.CurrentPageType() != rbf.ContainerTypeRLE {
t.Fatalf("Should Be RLE but is: %v\n", c.CurrentPageType())
}
exists, err := c.Contains(0x7)
if err != nil {
t.Fatalf("ERR:%v", err)
}
if !exists {
t.Fatalf("Should Contain %v", 0x7)
}
//add a few bits to create another run
_, err = tx.Add("x",
func() []uint64 {
r := make([]uint64, 0, 128)
for x := uint64(65408); x < 65536; x++ {
r = append(r, x)
want = append(want, uint16(x))
}
return r
}()...)
if err != nil {
t.Fatalf("ERR adding bits: %v\n", err)
}
if err := c.First(); err != nil {
t.Fatal(err)
}
if got, want := c.Values(), want; !reflect.DeepEqual(got, want) {
t.Fatalf("Values()=%#v, want %#v", got, want)
} else if c.CurrentPageType() != rbf.ContainerTypeBitmap {
t.Fatalf("Should be bitmap but is %v", c.CurrentPageType())
}
}

151
rbf/cursorx.go Normal file
View file

@ -0,0 +1,151 @@
// 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 (
"bufio"
"fmt"
"io"
"math"
"os"
"github.com/pilosa/pilosa/v2/roaring"
)
//probably should just implement the container interface
// but for now i'll do it
func (c *Cursor) Rows() ([]uint64, error) {
shardVsContainerExponent := uint(4) //needs constant exported from roaring package
if err := c.First(); err != nil {
return nil, err
}
rows := make([]uint64, 0)
var err error
var lastRow uint64 = math.MaxUint64
for {
err := c.Next()
if err != nil {
break
}
cell := c.cell()
vRow := cell.Key >> shardVsContainerExponent
if vRow == lastRow {
continue
}
rows = append(rows, vRow)
lastRow = vRow
}
return rows, err
}
func (tx *Tx) FieldViews() []string {
r, _ := tx.rootRecords()
res := make([]string, len(r))
for i := range r {
res[i] = r[i].Name
}
return res
}
func (c *Cursor) DumpKeys() error {
if err := c.First(); err != nil {
return err
}
for {
err := c.Next()
if err == io.EOF {
return nil
} else if err != nil {
return err
}
cell := c.cell()
fmt.Println("key", cell.Key)
}
}
func (c *Cursor) DumpStack() {
fmt.Println("STACK")
for i := c.stack.index; i >= 0; i-- {
fmt.Printf("%+v\n", c.stack.elems[i])
}
fmt.Println()
}
func (c *Cursor) Dump() {
bufStdout := bufio.NewWriter(os.Stdout)
defer bufStdout.Flush()
fmt.Fprintf(bufStdout, "digraph RBF{\n")
fmt.Fprintf(bufStdout, "rankdir=\"LR\"\n")
fmt.Fprintf(bufStdout, "node [shape=record height=.1]\n")
dumpdot(c.tx, 0, " ", bufStdout)
fmt.Fprintf(bufStdout, "\n}")
}
func (c *Cursor) Row(rowID uint64) (*roaring.Bitmap, error) {
base := rowID * ShardWidth
offset := uint64(c.tx.db.Shard * ShardWidth)
off := highbits(offset)
hi0, hi1 := highbits(base), highbits((rowID+1)*ShardWidth)
c.stack.index = 0
ok, err := c.Seek(hi0)
if err != nil {
return nil, err
}
if !ok {
elem := &c.stack.elems[c.stack.index]
n := readCellN(c.leafPage)
if elem.index >= n {
if err := c.goNextPage(); err != nil {
return nil, err
}
}
}
other := roaring.NewSliceBitmap()
for {
err := c.Next()
if err == io.EOF {
break
} else if err != nil {
return nil, err
}
cell := c.cell()
if cell.Key >= hi1 {
break
}
other.Containers.Put(off+(cell.Key-hi0), toContainer(cell))
}
return other, nil
}
// 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() int {
cell := c.cell()
return cell.Type
}
func toContainer(l leafCell) *roaring.Container {
switch l.Type {
case ContainerTypeArray:
return roaring.NewContainerArray(toArray16(l.Data))
case ContainerTypeBitmap:
return roaring.NewContainerBitmap(l.N, toArray64(l.Data))
case ContainerTypeRLE:
return roaring.NewContainerRun(toInterval16(l.Data))
}
return nil
}

637
rbf/db.go Normal file
View file

@ -0,0 +1,637 @@
// 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 (
"errors"
"fmt"
"io"
"io/ioutil"
"os"
"path/filepath"
"sync"
"syscall"
"github.com/benbjohnson/immutable"
"github.com/pilosa/pilosa/v2/syswrap"
)
var (
ErrClosed = errors.New("rbf: database closed")
)
const (
// Maximum size of a single WAL segment.
// May exceed by one page if last page is a bitmap header + bitmap.
MaxWALSegmentFileSize = 10 * (1 << 20)
)
type DB struct {
data []byte // mmap data
file *os.File // file descriptor
segments []*WALSegment // write-ahead log
pageMap *immutable.Map // pgno-to-WALID mapping
txs map[*Tx]struct{} // active transactions
opened bool // true if open
mu sync.RWMutex // general mutex
rwmu sync.Mutex // mutex for restricting single writer
// Path represents the path to the database file.
Path string
// The maximum allowed database size. Required by mmap.
MaxSize int64
Shard int
}
// NewDB returns a new instance of DB.
func NewDB(path string) *DB {
return NewDBWithShard(path, 0)
}
func NewDBWithShard(path string, shard int) *DB {
return &DB{
txs: make(map[*Tx]struct{}),
pageMap: immutable.NewMap(&uint32Hasher{}),
Path: path,
MaxSize: DefaultMaxSize,
Shard: shard,
}
}
// DataPath returns the path to the data file for the DB.
func (db *DB) DataPath() string {
return filepath.Join(db.Path, "data")
}
// WALPath returns the path to the WAL directory.
func (db *DB) WALPath() string {
return filepath.Join(db.Path, "wal")
}
func CreateDirIfNotExist(path string) {
dir := filepath.Dir(path)
if _, err := os.Stat(dir); os.IsNotExist(err) {
err = os.MkdirAll(dir, 0755)
if err != nil {
panic(err)
}
}
}
// Open opens a database with the file specified in Path.
// Creates a new file if one does not already exist.
func (db *DB) Open() (err error) {
db.mu.Lock()
defer db.mu.Unlock()
if err := os.MkdirAll(filepath.Dir(db.Path), 0755); err != nil {
return err
} else if db.file, err = os.OpenFile(db.DataPath(), os.O_WRONLY|os.O_CREATE, 0666); err != nil {
return fmt.Errorf("open file: %w", err)
}
// Open read-only mmap.
if f, err := os.OpenFile(db.DataPath(), os.O_RDONLY, 0666); err != nil {
return fmt.Errorf("open mmap file: %w", err)
} else if db.data, err = syswrap.Mmap(int(f.Fd()), 0, int(db.MaxSize), syscall.PROT_READ, syscall.MAP_SHARED); err != nil {
f.Close()
return fmt.Errorf("open mmap file: %w", err)
} else if err := f.Close(); err != nil {
return fmt.Errorf("cannot close mmap file: %w", err)
}
// Initialize file if it is too small.
if fi, err := db.file.Stat(); err != nil {
return fmt.Errorf("stat: %w", err)
} else if fi.Size() < PageSize {
if err := db.init(); err != nil {
return fmt.Errorf("init: %w", err)
}
}
// TODO(BBJ): Obtain advisory lock on file.
// Ensure WAL directory exists.
if err := os.MkdirAll(db.WALPath(), 0777); err != nil {
return fmt.Errorf("create wal dir: %w", err)
}
// Open write-ahead log & checkpoint to the end since no transactions are open.
if err := db.openWALSegments(); err != nil {
return fmt.Errorf("wal open: %w", err)
} else if err := db.checkpoint(); err != nil {
return fmt.Errorf("checkpoint: %w", err)
}
db.opened = true
return nil
}
func (db *DB) openWALSegments() error {
fis, err := ioutil.ReadDir(db.WALPath())
if err != nil {
return fmt.Errorf("read dir: %w", err)
}
// Open all WAL segments.
for _, fi := range fis {
if filepath.Ext(fi.Name()) != ".wal" {
continue
}
segment := NewWALSegment(filepath.Join(db.WALPath(), fi.Name()))
if err := segment.Open(); err != nil {
_ = db.closeWALSegments()
return err
}
db.segments = append(db.segments, segment)
}
// Truncate last WAL page if it is a bitmap header.
if segment := db.activeWALSegment(); segment != nil {
if err := segment.trimBitmapHeaderTrailer(); err != nil {
return err
}
}
return nil
}
// checkpoint copies pages from WAL segments into the main DB file. This can
// only copy pages that aren't in use by an active transaction. The page map
// is rebuilt as well for all WAL pages still in use.
func (db *DB) checkpoint() error {
if !db.opened {
return nil
}
// Determine last checkpointed WAL ID.
page, err := db.readPage(nil, 0)
if err != nil {
return err
}
walID := readMetaWALID(page)
// Determine the high water mark for WAL pages that can be copied.
minActiveWALID := db.minActiveWALID()
// Loop over each transaction
walID++
pageMap := immutable.NewMap(&uint32Hasher{})
for {
// Determine last page of transaction.
metaWALID, metaFlags, err := db.findNextWALMetaPage(walID)
if err == io.EOF {
break
} else if err != nil {
return err
}
// If transaction was rolled back, skip it.
if metaFlags&MetaPageFlagCommit == 0 {
walID = metaWALID + 1
continue
}
// Loop over pages in the tranasction.
for ; walID <= metaWALID; walID++ {
canCheckpoint := minActiveWALID == 0 || walID <= minActiveWALID
page, err := db.readWALPage(walID)
if err != nil {
return err
}
isBitmapHeader := IsBitmapHeader(page)
// Determine page number. Meta pages are always on zero & bitmap
// headers specify the page number of the next page in the WAL.
// All other pages have their page number in the page data.
var pgno uint32
if isBitmapHeader {
pgno, walID = readPageNo(page), walID+1 // skip next page
} else if !IsMetaPage(page) {
pgno = readPageNo(page)
}
// If we can no longer checkpoint, map the page number to the WAL page.
if !canCheckpoint {
pageMap = pageMap.Set(pgno, walID)
continue
}
// Ensure we actually read the bitmap data in when we checkpoint.
// NOTE: The walID variable is incremented above in the pgno check.
if isBitmapHeader {
if page, err = db.readWALPage(walID); err != nil {
return err
}
}
// Write page data into main db file.
if err := db.writePage(pgno, page); err != nil {
return err
}
}
}
// Remove WAL segments that have been checkpointed.
for len(db.segments) > 1 {
segment := db.segments[0]
if minActiveWALID != 0 && segment.MaxWALID() >= minActiveWALID {
break
}
if err := segment.Close(); err != nil {
return err
}
db.segments, db.segments[0] = db.segments[1:], nil
}
db.pageMap = pageMap
return nil
}
func (db *DB) findNextWALMetaPage(walID int64) (metaWALID int64, metaFlags uint32, err error) {
maxWALID := db.maxWALID()
for ; walID <= maxWALID; walID++ {
// Read page data from WAL and return if it is a meta page (either commit or rollback)
page, err := db.readWALPage(walID)
if err != nil {
return walID, metaFlags, err
} else if IsMetaPage(page) {
return walID, readFlags(page), nil
}
// Skip over next page if this is a bitmap header.
if IsBitmapHeader(page) {
walID++
}
}
return -1, 0, io.EOF
}
// minActiveWALID returns the lowest WAL ID in use by any active transaction.
// Returns 0 if no transactions are active.
func (db *DB) minActiveWALID() int64 {
var walID int64
for tx := range db.txs {
if walID == 0 || walID > tx.walID {
walID = tx.walID
}
}
return walID
}
// ActiveWALSegment returns the most recent WAL segment.
func (db *DB) ActiveWALSegment() *WALSegment {
db.mu.RLock()
defer db.mu.RUnlock()
return db.activeWALSegment()
}
func (db *DB) activeWALSegment() *WALSegment {
if len(db.segments) == 0 {
return nil
}
return db.segments[len(db.segments)-1]
}
// MinWALID returns the lowest WAL ID available in the WAL.
func (db *DB) MinWALID() int64 {
db.mu.RLock()
defer db.mu.RUnlock()
return db.minWALID()
}
func (db *DB) minWALID() int64 {
if len(db.segments) == 0 {
return 0
}
return db.segments[0].MinWALID()
}
// MaxWALID returns the highest WAL ID available in the WAL.
func (db *DB) MaxWALID() int64 {
db.mu.RLock()
defer db.mu.RUnlock()
return db.maxWALID()
}
func (db *DB) maxWALID() int64 {
if len(db.segments) == 0 {
return 0
}
s := db.segments[len(db.segments)-1]
return s.MaxWALID()
}
// WALPageN returns the number of pages across all segments.
func (db *DB) WALPageN() int64 {
db.mu.RLock()
defer db.mu.RUnlock()
var n int64
for _, s := range db.segments {
n += int64(s.PageN())
}
return n
}
// SyncWAL flushes the active segment to disk.
func (db *DB) SyncWAL() error {
if s := db.ActiveWALSegment(); s != nil {
return s.Sync()
}
return nil
}
// readWALPage reads a single page at the given WAL ID.
func (db *DB) readWALPage(walID int64) ([]byte, error) {
// TODO(BBJ): Binary search for segment.
for _, s := range db.segments {
if walID >= s.MinWALID() && walID <= s.MaxWALID() {
return s.ReadWALPage(walID)
}
}
return nil, fmt.Errorf("cannot find segment containing WAL page: %d", walID)
}
func (db *DB) writeWALPage(page []byte, isMeta bool) (walID int64, err error) {
if err := db.ensureWritableWALSegment(); err != nil {
return 0, err
}
return db.activeWALSegment().WriteWALPage(page, isMeta)
}
func (db *DB) writeBitmapPage(pgno uint32, page []byte) (walID int64, err error) {
if err := db.ensureWritableWALSegment(); err != nil {
return 0, err
}
// Write header page for next bitmap page.
buf := make([]byte, PageSize)
writePageNo(buf[:], pgno)
writeFlags(buf[:], PageTypeBitmapHeader)
// TODO(BBJ): Write checksum.
if _, err := db.activeWALSegment().WriteWALPage(buf, false); err != nil {
return 0, fmt.Errorf("write bitmap header: %w", err)
}
// Write the bitmap page and return its WALID.
return db.activeWALSegment().WriteWALPage(page, false)
}
func (db *DB) ensureWritableWALSegment() error {
if s := db.activeWALSegment(); s != nil && s.Size() < MaxWALSegmentFileSize {
return nil
}
return db.addWALSegment()
}
// addWALSegment appends a new, writable segment and closing an existing segments for write.
func (db *DB) addWALSegment() error {
// Close previous last segment for writes.
base := int64(1)
if s := db.activeWALSegment(); s != nil {
base = s.MaxWALID() + 1
if err := s.CloseForWrite(); err != nil {
return err
}
}
// Create new segment file.
s := NewWALSegment(filepath.Join(db.WALPath(), FormatWALSegmentPath(base)))
if err := s.Open(); err != nil {
return fmt.Errorf("add wal segment: %w", err)
}
db.segments = append(db.segments, s)
return nil
}
// Close closes the database.
func (db *DB) Close() (err error) {
// TODO(bbj): Add wait group to hang until last Tx is complete.
db.mu.Lock()
defer db.mu.Unlock()
// Wait for writer lock.
db.rwmu.Lock()
defer db.rwmu.Unlock()
db.opened = false
// Close mmap handle.
if db.data != nil {
if e := syswrap.Munmap(db.data); e != nil && err == nil {
err = e
}
db.data = nil
}
// Close writer handler.
if db.file != nil {
if e := db.file.Close(); e != nil && err == nil {
err = e
}
db.file = nil
}
if e := db.closeWALSegments(); e != nil && err == nil {
err = e
}
return err
}
// closeWALSegments closes the WAL and all its segments.
func (db *DB) closeWALSegments() (err error) {
for _, s := range db.segments {
if e := s.Close(); e != nil && err == nil {
err = e
}
}
return err
}
// Size returns the size of the database & WAL, in bytes.
func (db *DB) Size() (int64, error) {
db.mu.RLock()
defer db.mu.RUnlock()
fi, err := os.Stat(db.Path)
if err != nil {
return 0, err
}
return db.walSize() + fi.Size(), nil
}
// WALSize returns the size of all WAL segments, in bytes.
func (db *DB) WALSize() int64 {
db.mu.RLock()
defer db.mu.RUnlock()
return db.walSize()
}
func (db *DB) walSize() int64 {
var sz int64
for _, s := range db.segments {
sz += s.Size()
}
return sz
}
// WALSegments returns the WAL segments currently on the DB.
// This should only be used for debugging & testing purposes.
func (db *DB) WALSegments() []*WALSegment {
db.mu.RLock()
defer db.mu.RUnlock()
return db.segments
}
// init initializes a new database file.
func (db *DB) init() error {
if err := db.initMetaPage(); err != nil {
return fmt.Errorf("meta: %w", err)
} else if err := db.initRootRecordPage(); err != nil {
return fmt.Errorf("root record page: %w", err)
} else if err := db.initFreelistPage(); err != nil {
return fmt.Errorf("freelist page: %w", err)
}
return nil
}
// initMetaPage initializes the meta page.
func (db *DB) initMetaPage() error {
page := make([]byte, PageSize)
writeMetaMagic(page)
writeMetaPageN(page, 3)
writeMetaRootRecordPageNo(page, 1)
writeMetaFreelistPageNo(page, 2)
_, err := db.file.WriteAt(page, 0*PageSize)
return err
}
// initRootRecordPage initializes the initial root record page.
func (db *DB) initRootRecordPage() error {
page := make([]byte, PageSize)
writePageNo(page, 1)
writeFlags(page, PageTypeRootRecord)
_, err := db.file.WriteAt(page, 1*PageSize)
return err
}
// initFreelistPage initializes the initial freelist btree page.
func (db *DB) initFreelistPage() error {
page := make([]byte, PageSize)
writePageNo(page, 2)
writeFlags(page, PageTypeLeaf)
_, err := db.file.WriteAt(page, 2*PageSize)
return err
}
// Begin starts a new transaction.
func (db *DB) Begin(writable bool) (_ *Tx, err error) {
// TODO(BBJ): Acquire write lock if writable.
db.mu.Lock()
defer db.mu.Unlock()
if !db.opened {
return nil, ErrClosed
}
tx := &Tx{db: db, pageMap: db.pageMap, writable: writable}
// Ensure only one writable transaction at a time.
if tx.writable {
db.rwmu.Lock()
}
// Copy meta page into transaction's buffer.
// This page is only written at the end of a dirty transaction.
page, err := db.readPage(db.pageMap, 0)
if err != nil {
_ = tx.Rollback()
return nil, err
}
copy(tx.meta[:], page)
// Attach starting WAL ID to transaction.
tx.walID = readMetaWALID(tx.meta[:])
// Track transaction with the DB.
db.txs[tx] = struct{}{}
return tx, nil
}
// removeTx removes an active transaction from the database.
func (db *DB) removeTx(tx *Tx) error {
// Release writer lock if tx is writable.
if tx.writable {
tx.db.rwmu.Unlock()
}
db.mu.Lock()
defer db.mu.Unlock()
// Write pages from WAL to DB.
// TODO(bbj): Move this to an async goroutine.
if err := db.checkpoint(); err != nil {
return err
}
delete(tx.db.txs, tx)
// Disassociate from db.
tx.db = nil
return nil
}
// Check performs an integrity check.
func (db *DB) Check() error {
tx, err := db.Begin(false)
if err != nil {
return err
}
defer func() { _ = tx.Rollback() }()
return tx.Check()
}
// writePage writes a page to the data file.
func (db *DB) writePage(pgno uint32, page []byte) error {
_, err := db.file.WriteAt(page, int64(pgno)*PageSize)
return err
}
func (db *DB) readPage(pageMap *immutable.Map, pgno uint32) ([]byte, error) {
// Check if page is currently in WAL.
if pageMap != nil {
if walID, ok := pageMap.Get(pgno); ok {
return db.readWALPage(walID.(int64))
}
}
// Otherwise read from the data file.
offset := int64(pgno) * PageSize
return db.data[offset : offset+PageSize], nil
}

132
rbf/db_test.go Normal file
View file

@ -0,0 +1,132 @@
// 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_test
import (
"math/rand"
"os"
"testing"
"github.com/pilosa/pilosa/v2/rbf"
)
func TestDB_Open(t *testing.T) {
db := NewDB()
if err := db.Open(); err != nil {
t.Fatal(err)
} else if err := db.Close(); err != nil {
t.Fatal(err)
}
}
func TestDB_Checkpoint(t *testing.T) {
if testing.Short() {
t.Skip("-short enabled, skipping")
}
db := MustOpenDB(t)
defer MustCloseDB(t, db)
// Create bitmap.
if tx, err := db.Begin(true); err != nil {
t.Fatal(err)
} else if err := tx.CreateBitmap("x"); err != nil {
t.Fatal(err)
} else if err := tx.Commit(); err != nil {
t.Fatal(err)
}
// Create a bunch of transactions to generate WAL segments.
rand := rand.New(rand.NewSource(0))
for i := 0; i < 1000; i++ {
if tx, err := db.Begin(true); err != nil {
t.Fatal(err)
} else if _, err := tx.Add("x", rand.Uint64()); err != nil {
t.Fatal(err)
} else if _, err := tx.Add("x", rand.Uint64()); err != nil {
t.Fatal(err)
} else if err := tx.Commit(); err != nil {
t.Fatal(err)
}
}
// Ensure there is no more than two WAL segments.
if n := len(db.WALSegments()); n > 2 {
t.Fatalf("expected two or fewer WAL segments, got %d", n)
}
}
func TestDB_Recovery(t *testing.T) {
// Ensure a bitmap header written without a bitmap is truncated.
t.Run("TruncPartialWALBitmap", func(t *testing.T) {
db := MustOpenDB(t)
defer MustCloseDB(t, db)
a := make([]uint64, rbf.ArrayMaxSize+100)
for i := range a {
a[i] = uint64(i)
}
// Create bitmap & generate enough values to create a bitmap container.
if tx, err := db.Begin(true); err != nil {
t.Fatal(err)
} else if err := tx.CreateBitmap("x"); err != nil {
t.Fatal(err)
} else if _, err := tx.Add("x", a...); err != nil {
t.Fatal(err)
} else if err := tx.Commit(); err != nil {
t.Fatal(err)
}
// Add one additional bit in a second transaction.
if tx, err := db.Begin(true); err != nil {
t.Fatal(err)
} else if _, err := tx.Add("x", uint64(len(a))); err != nil {
t.Fatal(err)
} else if err := tx.Commit(); err != nil {
t.Fatal(err)
}
// Close database & truncate WAL to remove commit page & bitmap data page.
segment := db.ActiveWALSegment()
if err := db.Close(); err != nil {
t.Fatal(err)
} else if err := os.Truncate(segment.Path(), segment.Size()-(2*rbf.PageSize)); err != nil {
t.Fatal(err)
}
// Reopen database.
newDB := rbf.NewDB(db.Path)
if err := newDB.Open(); err != nil {
t.Fatal(err)
}
defer MustCloseDB(t, newDB)
// Verify last insert was not added.
tx, err := newDB.Begin(true)
if err != nil {
t.Fatal(err)
}
defer MustRollback(t, tx)
if exists, err := tx.Contains("x", uint64(len(a))); exists || err != nil {
t.Fatalf("Contains()=<%v,%#v>", exists, err)
} else if exists, err := tx.Contains("x", uint64(len(a)-1)); !exists || err != nil {
t.Fatalf("Contains()=<%v,%#v>", exists, err)
} else if err := tx.Rollback(); err != nil {
t.Fatal(err)
}
})
}

105
rbf/dot.go Normal file
View file

@ -0,0 +1,105 @@
// 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"
)
func dotCell(b []byte, parent string, writer io.Writer) {
pgno := readPageNo(b)
if pgno == Magic32() {
fmt.Fprintf(writer, "==META\n")
return
}
flags := readFlags(b)
cellN := readCellN(b)
switch {
case flags&PageTypeLeaf != 0:
fmt.Fprintf(writer, "cell%d [ shape=none label=<<table border=\"0\" cellspacing=\"0\">\n", pgno)
fmt.Fprintf(writer, "<tr><td border=\"1\">CELL</td></tr>\n")
for i := 0; i < cellN; i++ {
cell := readLeafCell(b, i)
switch cell.Type {
case ContainerTypeArray:
//fmt.Fprintf(os.Stderr, "[%d]: key=%d type=array n=%d elems=%v\n", i, cell.Key, cell.N, toArray16(cell.Data))
fmt.Fprintf(writer, "<tr><td border=\"1\" bgcolor=\"green\"><font color=\"white\">[%d]: key=%d type=array n=%d</font></td></tr>\n", i, cell.Key, cell.N)
case ContainerTypeRLE:
fmt.Fprintf(writer, "<tr><td border=\"1\" bgcolor=\"blue\"><font color=\"white\">[%d]: key=%d type=rle n=%d</font></td></tr>\n", i, cell.Key, cell.N)
default:
fmt.Fprintf(writer, "<tr><td border=\"1\" bgcolor=\"red\">[%d]: key=%d type=unknown<%d> n=%d</td></tr>\n", i, cell.Key, cell.Type, cell.N)
}
}
fmt.Fprintf(writer, "</table>>]\n")
fmt.Fprintf(writer, "%s -> cell%d\n", parent, pgno)
default:
//should not happen
fmt.Fprintf(writer, "==!PAGE %d flags=%d\n", pgno, flags)
}
}
// 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)
if err != nil {
panic(err)
}
if IsMetaPage(page) {
//fmt.Fprintf(writer, "META(%d)\n", pgno)
//fmt.Fprintf(writer, "└── <FREELIST>\n")
//treedump(tx, readMetaFreelistPageNo(page), indent+" ")
visitor := func(pgno uint32, records []*RootRecord) {
rr := fmt.Sprintf("rr%d", pgno)
fmt.Fprintf(writer, "%s[label=\"ROOT RECORD(%d): n=%d\"]\n", rr, pgno, len(records))
for _, record := range records {
root := fmt.Sprintf("root%d", record.Pgno)
fmt.Fprintf(writer, "%s[label=\"ROOT(%d)| %s\"]\n%s->%s\n", root, record.Pgno, record.Name, rr, root)
parent := fmt.Sprintf("root%d", record.Pgno)
dumpdot(tx, record.Pgno, parent, writer)
}
}
rrdump(tx, readMetaRootRecordPageNo(page), visitor)
return
}
// Handle
switch typ := readFlags(page); typ {
case PageTypeBranch:
p := fmt.Sprintf("branch%d", pgno)
fmt.Fprintf(writer, "%s[label=\"BRANCH(%d)| n=%d\"]\n %s->%s\n", p, pgno, readCellN(page), parent, p)
for i, n := 0, readCellN(page); i < n; i++ {
cell := readBranchCell(page, i)
if cell.Flags&ContainerTypeBitmap == 0 { // leaf/branch child page
dumpdot(tx, cell.Pgno, p, writer)
} else {
b := fmt.Sprintf("bm%d", cell.Pgno)
fmt.Fprintf(writer, "%s[label=\"BITMAP(%d)\"]\n %s -> %s\n", b, cell.Pgno, p, b)
}
}
case PageTypeLeaf:
p := fmt.Sprintf("leaf%d", pgno)
fmt.Fprintf(writer, "%s[label=\"LEAF(%d)| n=%d\"]\n%s->%s\n", p, pgno, readCellN(page), parent, p)
dotCell(page, p, writer)
default:
panic(err)
}
}

26
rbf/internal_test.go Normal file
View file

@ -0,0 +1,26 @@
// 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 "testing"
// This function exists to mark debugging helper function as "used" by the linter.
func TestUsed(t *testing.T) {
t.Skip("This function is always skipped")
dump(nil)
hexdump(nil)
pagedump(nil, "", nil)
treedump(nil, 0, "", nil)
}

22
rbf/os.go Normal file
View file

@ -0,0 +1,22 @@
// 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.
// +build !386
package rbf
// DefaultMaxSize is the default mmap size and therefore the maximum allowed
// size of the database. The size can be increased by updating the DB.MaxSize
// and reopening the database. This setting mainly affects virtual space usage.
const DefaultMaxSize = 100 * (1 << 30) // 100GB

20
rbf/os_386.go Normal file
View file

@ -0,0 +1,20 @@
// 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
// DefaultMaxSize is the default mmap size and therefore the maximum allowed
// size of the database. The size can be increased by updating the DB.MaxSize
// and reopening the database. This setting mainly affects virtual space usage.
const DefaultMaxSize = 256 * (1 << 20) // 256MB

620
rbf/rbf.go Normal file
View file

@ -0,0 +1,620 @@
// 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 implements the roaring b-tree file format.
package rbf
import (
"bytes"
"encoding/binary"
"encoding/hex"
"errors"
"fmt"
"io"
"unsafe"
"github.com/pilosa/pilosa/v2/shardwidth"
)
const (
// Magic is the first 4 bytes of the RBF file.
Magic = "\xFFRBF"
// PageSize is the fixed size for every database page.
PageSize = 8192
// ShardWidth represents the number of bits per shard.
ShardWidth = 1 << shardwidth.Exponent
// RowValueMask masks the low bits for a row.
RowValueMask = ShardWidth - 1
// ArrayMaxSize represents the maximum size of array containers.
// This is sligtly less than roaring to accommodate the page header.
ArrayMaxSize = 4080
// RLEMaxSize represents the maximum size of run length encoded containers.
RLEMaxSize = 2040
)
// Page types.
const (
PageTypeRootRecord = 1
PageTypeLeaf = 2
PageTypeBranch = 4
PageTypeBitmapHeader = 8 // Only used by the WAL for marking next page
)
// Meta commit/rollback flags.
const (
MetaPageFlagCommit = 1
MetaPageFlagRollback = 2
)
// Container types.
const (
ContainerTypeNone = iota
ContainerTypeArray
ContainerTypeRLE
ContainerTypeBitmap
)
const (
rootRecordPageHeaderSize = 12
rootRecordHeaderSize = 4 + 2 // pgno, len(name)
leafCellHeaderSize = 8 + 4 + 4 // key, type, count
branchCellSize = 8 + 4 + 4 // key, flags, pgno
)
var (
ErrTxClosed = errors.New("transaction closed")
ErrTxNotWritable = errors.New("transaction not writable")
ErrBitmapNameRequired = errors.New("bitmap name required")
)
// Debug is just a temporary flag used for debugging.
var Debug bool
// Magic32 returns the magic bytes as a big endian encoded uint32.
func Magic32() uint32 {
return binary.BigEndian.Uint32([]byte(Magic))
}
// Meta page helpers
// IsMetaPage returns true if page is a meta page.
func IsMetaPage(page []byte) bool {
return bytes.Equal(readMetaMagic(page), []byte(Magic))
}
func readMetaMagic(page []byte) []byte { return page[0:4] }
func writeMetaMagic(page []byte) { copy(page, Magic) }
func readMetaPageN(page []byte) uint32 { return binary.BigEndian.Uint32(page[8:]) }
func writeMetaPageN(page []byte, n uint32) { binary.BigEndian.PutUint32(page[8:], n) }
func readMetaWALID(page []byte) int64 { return int64(binary.BigEndian.Uint64(page[12:])) }
func writeMetaWALID(page []byte, walID int64) { binary.BigEndian.PutUint64(page[12:], uint64(walID)) }
func readMetaRootRecordPageNo(page []byte) uint32 { return binary.BigEndian.Uint32(page[20:]) }
func writeMetaRootRecordPageNo(page []byte, pgno uint32) { binary.BigEndian.PutUint32(page[20:], pgno) }
func readMetaFreelistPageNo(page []byte) uint32 { return binary.BigEndian.Uint32(page[24:]) }
func writeMetaFreelistPageNo(page []byte, pgno uint32) { binary.BigEndian.PutUint32(page[24:], pgno) }
// func readMetaChecksum(page []byte) uint32 {
// return binary.BigEndian.Uint32(page[PageSize-4 : PageSize])
// }
// func writeMetaChecksum(page []byte, chksum uint32) {
// binary.BigEndian.PutUint32(page[PageSize-4:PageSize], chksum)
// }
// Root record page helpers
func readRootRecordOverflowPgno(page []byte) uint32 { return binary.BigEndian.Uint32(page[8:]) }
func writeRootRecordOverflowPgno(page []byte, pgno uint32) {
binary.BigEndian.PutUint32(page[8:], pgno)
}
func readRootRecords(page []byte) (records []*RootRecord, err error) {
for data := page[rootRecordPageHeaderSize:]; ; {
var rec *RootRecord
if rec, data, err = ReadRootRecord(data); err != nil {
return records, err
} else if rec == nil {
return records, nil
}
records = append(records, rec)
}
}
func writeRootRecords(page []byte, records []*RootRecord) (remaining []*RootRecord, err error) {
data := page[rootRecordPageHeaderSize:]
for i, rec := range records {
if data, err = WriteRootRecord(data, rec); err == io.ErrShortBuffer {
return records[i:], nil
} else if err != nil {
return records[i:], err
}
}
return nil, nil
}
// Branch & leaf page helpers
func readPageNo(page []byte) uint32 { return binary.BigEndian.Uint32(page[0:4]) }
func writePageNo(page []byte, v uint32) { binary.BigEndian.PutUint32(page[0:4], v) }
func readFlags(page []byte) uint32 { return binary.BigEndian.Uint32(page[4:8]) }
func writeFlags(page []byte, v uint32) { binary.BigEndian.PutUint32(page[4:8], v) }
func readCellN(page []byte) int { return int(binary.BigEndian.Uint16(page[8:10])) }
func writeCellN(page []byte, v int) { binary.BigEndian.PutUint16(page[8:10], uint16(v)) }
func readCellOffset(page []byte, i int) int {
assert(i < readCellN(page))
return int(binary.BigEndian.Uint16(page[10+(i*2):]))
}
func writeCellOffset(page []byte, i int, v int) {
binary.BigEndian.PutUint16(page[10+(i*2):], uint16(v))
}
func dataOffset(n int) int {
return align8(10 + (n * 2))
}
func IsBitmapHeader(page []byte) bool {
// TODO(BBJ): Verify checksum.
return readFlags(page) == PageTypeBitmapHeader
}
type RootRecord struct {
Name string
Pgno uint32
}
// ReadRootRecord reads the page number & name for a root record.
// If there is not enough space or the pgno is zero then a nil record is returned.
// Returns the remaining buffer.
func ReadRootRecord(data []byte) (rec *RootRecord, remaining []byte, err error) {
// Ensure there is enough space to read the pgno & name length.
if len(data) < rootRecordHeaderSize {
return nil, data, nil
}
// Read root page number.
rec = &RootRecord{}
rec.Pgno = binary.BigEndian.Uint32(data)
if rec.Pgno == 0 {
return nil, data, nil
}
data = data[4:]
// Read name length.
sz := int(binary.BigEndian.Uint16(data))
data = data[2:]
if len(data) < sz {
return nil, data, fmt.Errorf("short root record buffer")
}
// Read name and allocate as string on heap.
rec.Name, data = string(data[:sz]), data[sz:]
return rec, data, nil
}
// WriteRootRecord writes a root record with the pgno & name.
// Returns io.ErrShortBuffer if there is not enough space.
func WriteRootRecord(data []byte, rec *RootRecord) (remaining []byte, err error) {
// Ensure record data is valid.
if rec == nil {
return data, fmt.Errorf("root record required")
} else if rec.Name == "" {
return data, fmt.Errorf("root record name required")
} else if rec.Pgno == 0 {
return data, fmt.Errorf("invalid root record pgno: %d", rec.Pgno)
}
// Ensure there is enough space to write the full record.
if len(data) < rootRecordHeaderSize+len(rec.Name) {
return data, io.ErrShortBuffer
}
// Write root page number.
binary.BigEndian.PutUint32(data, rec.Pgno)
data = data[4:]
// Write name length.
binary.BigEndian.PutUint16(data, uint16(len(rec.Name)))
data = data[2:]
// Write name.
copy(data, rec.Name)
data = data[len(rec.Name):]
return data, nil
}
func align8(offset int) int {
if offset%8 == 0 {
return offset
}
return offset + (8 - (offset & 0x7))
}
// leafCell represents a leaf cell.
type leafCell struct {
Key uint64
Type int
N int
Data []byte
}
// Size returns the size of the leaf cell, in bytes.
func (c *leafCell) Size() int {
if c.Type == ContainerTypeBitmap {
return PageSize
}
return leafCellHeaderSize + len(c.Data)
}
// Bitmap returns a bitmap representation of the cell data.
func (c *leafCell) Bitmap() []uint64 {
switch c.Type {
case ContainerTypeArray:
buf := make([]uint64, PageSize/8)
for _, v := range toArray16(c.Data) {
buf[v/64] |= 1 << uint64(v%64)
}
return buf
case ContainerTypeRLE:
buf := make([]uint64, PageSize/8)
for _, iv := range toInterval16(c.Data) {
w1, w2 := iv.Start/64, iv.Last/64
b1, b2 := iv.Start&63, iv.Last&63
m1 := (uint64(1) << b1) - 1
m2 := (((uint64(1) << b2) - 1) << 1) | 1
if w1 == w2 {
buf[w1] |= (m2 &^ m1)
continue
}
buf[w2] |= m2
buf[w1] |= ^m1
words := buf[w1+1 : w2]
for i := range words {
words[i] = ^uint64(0)
}
}
return buf
case ContainerTypeBitmap:
return toArray64(c.Data)
default:
panic(fmt.Sprintf("invalid container type: %d", c.Type))
}
}
// Values returns a slice of 16-bit values from a container.
func (c *leafCell) Values() []uint16 {
switch c.Type {
case ContainerTypeArray:
return toArray16(c.Data)
case ContainerTypeRLE:
//a := make([]uint16, c.N)
a := make([]uint16, ArrayMaxSize)
n := int32(0)
for _, r := range toInterval16(c.Data) {
for v := int(r.Start); v <= int(r.Last); v++ {
a[n] = uint16(v)
n++
}
}
a = a[:n]
return a
case ContainerTypeBitmap:
a := make([]uint16, 0, ArrayMaxSize)
for i, v := range toArray64(c.Data) {
for j := uint(0); j < 64; j++ {
if v&(1<<j) != 0 {
a = append(a, (uint16(i)*64)+uint16(j))
}
}
}
return a
default:
panic(fmt.Sprintf("invalid container type: %d", c.Type))
}
}
// firstValue the first value from the container.
func (c *leafCell) firstValue() uint16 {
switch c.Type {
case ContainerTypeArray:
a := toArray16(c.Data)
return a[0]
case ContainerTypeRLE:
r := toInterval16(c.Data)
return r[0].Start
case ContainerTypeBitmap:
for i, v := range toArray64(c.Data) {
for j := uint(0); j < 64; j++ {
if v&(1<<j) != 0 {
return (uint16(i) * 64) + uint16(j)
}
}
}
panic(fmt.Sprintf("rbf.leafCell.firstValue(): no values set in bitmap container: key=%d", c.Key))
default:
panic(fmt.Sprintf("invalid container type: %d", c.Type))
}
}
func readLeafCellKey(page []byte, i int) uint64 {
offset := readCellOffset(page, i)
return *(*uint64)(unsafe.Pointer(&page[offset]))
}
func readLeafCell(page []byte, i int) leafCell {
offset := readCellOffset(page, i)
buf := page[offset:]
var cell leafCell
cell.Key = *(*uint64)(unsafe.Pointer(&buf[0]))
cell.Type = int(*(*uint32)(unsafe.Pointer(&buf[8])))
cell.N = int(*(*uint32)(unsafe.Pointer(&buf[12])))
switch cell.Type {
case ContainerTypeArray:
cell.Data = buf[16 : 16+(cell.N*2)]
case ContainerTypeRLE:
cell.Data = buf[16 : 16+(cell.N*4)]
default:
}
return cell
}
func readLeafCells(page []byte, isBitmap bool, buf []leafCell) []leafCell {
if isBitmap {
return []leafCell{{Type: ContainerTypeBitmap, Data: page}}
}
n := readCellN(page)
cells := buf[:n]
for i := 0; i < n; i++ {
cells[i] = readLeafCell(page, i)
}
return cells
}
// leafCellsPageSize returns the total page size required to hold cells.
func leafCellsPageSize(cells []leafCell) int {
sz := dataOffset(len(cells))
for i := range cells {
sz += align8(cells[i].Size())
}
return sz
}
func writeLeafCell(page []byte, i, offset int, cell leafCell) {
writeCellOffset(page, i, offset)
*(*uint64)(unsafe.Pointer(&page[offset])) = cell.Key
*(*uint32)(unsafe.Pointer(&page[offset+8])) = uint32(cell.Type)
*(*uint32)(unsafe.Pointer(&page[offset+12])) = uint32(cell.N)
assert(offset+16+len(cell.Data) <= PageSize)
copy(page[offset+16:], cell.Data)
}
// branchCell represents a branch cell.
type branchCell struct {
Key uint64
Flags uint32
Pgno uint32
}
// branchCellsPageSize returns the total page size required to hold cells.
func branchCellsPageSize(cells []branchCell) int {
sz := dataOffset(len(cells))
for range cells {
sz += align8(branchCellSize)
}
return sz
}
func readBranchCellKey(page []byte, i int) uint64 {
offset := readCellOffset(page, i)
return *(*uint64)(unsafe.Pointer(&page[offset]))
}
func readBranchCell(page []byte, i int) branchCell {
assert(i >= 0)
offset := readCellOffset(page, i)
var cell branchCell
cell.Key = *(*uint64)(unsafe.Pointer(&page[offset]))
cell.Flags = *(*uint32)(unsafe.Pointer(&page[offset+8]))
cell.Pgno = *(*uint32)(unsafe.Pointer(&page[offset+12]))
return cell
}
func readBranchCells(page []byte) []branchCell {
n := readCellN(page)
cells := make([]branchCell, n, n+1)
for i := 0; i < n; i++ {
cells[i] = readBranchCell(page, i)
}
return cells
}
func writeBranchCell(page []byte, i, offset int, cell branchCell) {
writeCellOffset(page, i, offset)
*(*uint64)(unsafe.Pointer(&page[offset+0])) = cell.Key
*(*uint32)(unsafe.Pointer(&page[offset+8])) = uint32(cell.Flags)
*(*uint32)(unsafe.Pointer(&page[offset+12])) = uint32(cell.Pgno)
}
func highbits(v uint64) uint64 { return v >> 16 }
func lowbits(v uint64) uint16 { return uint16(v & 0xFFFF) }
// search implements a binary search similar to sort.Search(), however,
// it returns the position as well as whether an exact match was made.
//
// The return value from f should be -1 for less than, 0 for equal, and 1 for
// greater than.
func search(n int, f func(int) int) (index int, exact bool) {
i, j := 0, n
for i < j {
h := int(uint(i+j) >> 1)
if cmp := f(h); cmp == 0 {
return h, true
} else if cmp > 0 {
i = h + 1
} else {
j = h
}
}
return i, false
}
func hexdump(b []byte) { println(hex.Dump(b)) }
func pagedump(b []byte, indent string, writer io.Writer) {
pgno := readPageNo(b)
if pgno == Magic32() {
fmt.Fprintf(writer, "==META\n")
return
}
flags := readFlags(b)
cellN := readCellN(b)
// NOTE(BBJ): There's no way to tell if a page is a bitmap container with
// the page alone so this will output !PAGE for bitmap pages & invalid pages.
switch {
case flags&PageTypeLeaf != 0:
for i := 0; i < cellN; i++ {
cell := readLeafCell(b, i)
switch cell.Type {
case ContainerTypeArray:
//fmt.Fprintf(os.Stderr, "[%d]: key=%d type=array n=%d elems=%v\n", i, cell.Key, cell.N, toArray16(cell.Data))
fmt.Fprintf(writer, "%s[%d]: key=%d type=array n=%d \n", indent, i, cell.Key, cell.N)
case ContainerTypeRLE:
fmt.Fprintf(writer, "%s[%d]: key=%d type=rle n=%d\n", indent, i, cell.Key, cell.N)
case ContainerTypeBitmap:
fmt.Fprintf(writer, "%s[%d]: key=%d type=bitmap n=%d\n", indent, i, cell.Key, cell.N)
default:
fmt.Fprintf(writer, "%s[%d]: key=%d type=unknown<%d> n=%d\n", indent, i, cell.Key, cell.Type, cell.N)
}
}
case flags&PageTypeBranch != 0:
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)
}
default:
fmt.Fprintf(writer, "==!PAGE %d flags=%d\n", pgno, flags)
}
}
// treedump recursively writes the tree representation starting from a given page to STDERR.
func treedump(tx *Tx, pgno uint32, indent string, writer io.Writer) {
page, err := tx.readPage(pgno)
if err != nil {
panic(err)
}
if IsMetaPage(page) {
fmt.Fprintf(writer, "META(%d)\n", pgno)
fmt.Fprintf(writer, "└── <FREELIST>\n")
//treedump(tx, readMetaFreelistPageNo(page), indent+" ")
visitor := func(pgno uint32, records []*RootRecord) {
fmt.Fprintf(writer, "└── ROOT RECORD(%d): n=%d\n", pgno, len(records))
for _, record := range records {
fmt.Fprintf(writer, "└── ROOT(%q) %d\n", record.Name, record.Pgno)
treedump(tx, record.Pgno, indent+" ", writer)
}
}
rrdump(tx, readMetaRootRecordPageNo(page), visitor)
return
}
// Handle
switch typ := readFlags(page); typ {
case PageTypeBranch:
fmt.Fprintf(writer, "%s BRANCH(%d) n=%d\n", fmtindent(indent), pgno, readCellN(page))
for i, n := 0, readCellN(page); i < n; i++ {
cell := readBranchCell(page, i)
if cell.Flags&ContainerTypeBitmap == 0 { // leaf/branch child page
treedump(tx, cell.Pgno, " "+indent, writer)
} else {
fmt.Fprintf(writer, "%s BITMAP(%d)\n", fmtindent(" "+indent), cell.Pgno)
}
}
case PageTypeLeaf:
fmt.Fprintf(writer, "%s LEAF(%d) n=%d\n", fmtindent(indent), pgno, readCellN(page))
pagedump(page, fmtindent(" "+indent), writer)
default:
panic(err)
}
}
func rrdump(tx *Tx, pgno uint32, v func(uint32, []*RootRecord)) {
for pgno := readMetaRootRecordPageNo(tx.meta[:]); pgno != 0; {
page, err := tx.readPage(pgno)
if err != nil {
panic(err)
}
// Read all records on the page.
a, err := readRootRecords(page)
if err != nil {
panic(err)
}
v(pgno, a)
// Read next overflow page number.
pgno = readRootRecordOverflowPgno(page)
}
}
func fmtindent(s string) string {
if s == "" {
return ""
}
return s + "└──"
}
// RowValues returns a list of integer values from a row bitmap.
func RowValues(b []uint64) []uint64 {
a := make([]uint64, 0)
for i, v := range b {
for j := uint(0); j < 64; j++ {
if v&(1<<j) != 0 {
a = append(a, (uint64(i)*64)+uint64(j))
}
}
}
return a
}
func assert(condition bool) {
if !condition {
panic("assertion failed")
}
}

20
rbf/rbf_norace.go Normal file
View file

@ -0,0 +1,20 @@
// 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.
// +build !race
package rbf
// RaceEnabled is true if the -race flag is enabled.
const RaceEnabled = false

20
rbf/rbf_race.go Normal file
View file

@ -0,0 +1,20 @@
// 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.
// +build race
package rbf
// RaceEnabled is true if the -race flag is enabled.
const RaceEnabled = true

233
rbf/rbf_test.go Normal file
View file

@ -0,0 +1,233 @@
// 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_test
import (
"flag"
"fmt"
"io"
"io/ioutil"
"math/rand"
"os"
"runtime"
"sort"
"testing"
"github.com/pilosa/pilosa/v2/rbf"
)
var quickCheckN *int = flag.Int("quickchecks", 10, "The number of iterations for each quickcheck")
// Ensure root record helper functions work to read & write root records.
func TestReadWriteRootRecord(t *testing.T) {
t.Run("OK", func(t *testing.T) {
buf := make([]byte, 26)
// Write records.
if remaining, err := rbf.WriteRootRecord(buf, &rbf.RootRecord{Pgno: 10, Name: "foo"}); err != nil {
t.Fatal(err)
} else if remaining, err := rbf.WriteRootRecord(remaining, &rbf.RootRecord{Pgno: 11, Name: "bar"}); err != nil {
t.Fatal(err)
} else if _, err := rbf.WriteRootRecord(remaining, &rbf.RootRecord{Pgno: 12, Name: "baz"}); err != io.ErrShortBuffer {
t.Fatalf("unexpected error: %#v", err) // buffer too short
}
// Read records back.
if rec, remaining, err := rbf.ReadRootRecord(buf); err != nil {
t.Fatal(err)
} else if got, want := *rec, (rbf.RootRecord{Pgno: 10, Name: "foo"}); got != want {
t.Fatalf("ReadRootRecord=%#v, want %#v", got, want)
} else if rec, remaining, err := rbf.ReadRootRecord(remaining); err != nil {
t.Fatal(err)
} else if got, want := *rec, (rbf.RootRecord{Pgno: 11, Name: "bar"}); got != want {
t.Fatalf("ReadRootRecord=%#v, want %#v", got, want)
} else if rec, _, _ := rbf.ReadRootRecord(remaining); rec != nil {
t.Fatalf("expected nil record, got %#v", rec)
}
})
}
// NewDB returns a new instance of DB with a temporary path.
func NewDB() *rbf.DB {
path, err := ioutil.TempDir("", "")
if err != nil {
panic(err)
}
db := rbf.NewDB(path)
return db
}
// MustOpenDB returns a db opened on a temporary file. On error, fail test.
func MustOpenDB(tb testing.TB) *rbf.DB {
tb.Helper()
db := NewDB()
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 *rbf.DB) {
tb.Helper()
if err := db.Check(); err != nil && err != rbf.ErrClosed {
tb.Fatal(err)
} else if err := db.Close(); err != nil && err != rbf.ErrClosed {
tb.Fatal(err)
} else if err := os.RemoveAll(db.Path); err != nil {
tb.Fatal(err)
}
}
// MustReopenDB closes and reopens a database.
func MustReopenDB(tb testing.TB, db *rbf.DB) *rbf.DB {
tb.Helper()
if err := db.Check(); err != nil {
tb.Fatal(err)
} else if err := db.Close(); err != nil {
tb.Fatal(err)
}
other := rbf.NewDB(db.Path)
if err := other.Open(); err != nil {
tb.Fatal(err)
}
return other
}
// MustBegin returns a new transaction or fails.
func MustBegin(tb testing.TB, db *rbf.DB, writable bool) *rbf.Tx {
tb.Helper()
tx, err := db.Begin(writable)
if err != nil {
tb.Fatal(err)
}
return tx
}
// MustRollback rolls back a transaction or fails.
func MustRollback(tb testing.TB, tx *rbf.Tx) {
tb.Helper()
if err := tx.Rollback(); err != nil && err != rbf.ErrTxClosed {
tb.Logf("rollback error: %q", err)
}
}
// MustAddRandom adds values to a bitmap in a random order.
func MustAddRandom(tb testing.TB, rand *rand.Rand, tx *rbf.Tx, name string, values ...uint64) {
tb.Helper()
for _, i := range rand.Perm(len(values)) {
v := values[i]
if _, err := tx.Add(name, v); err != nil {
tb.Fatalf("Add(%d) i=%d err=%q", v, i, err)
}
}
}
// GenerateValues returns a sorted list of random values.
func GenerateValues(rand *rand.Rand, n int) []uint64 {
a := make([]uint64, n)
for i := range a {
a[i] = uint64(rand.Intn(rbf.ShardWidth))
}
sort.Slice(a, func(i, j int) bool { return a[i] < a[j] })
return a
}
// ToRows returns a sorted list of rows from a set of values.
func ToRows(values []uint64) []*Row {
m := make(map[uint64][]uint64)
for _, v := range values {
id := v / rbf.ShardWidth
m[id] = append(m[id], v&rbf.RowValueMask)
}
a := make([]*Row, 0, len(m))
for id, values := range m {
a = append(a, &Row{ID: id, Values: values})
}
sort.Slice(a, func(i, j int) bool { return a[i].ID < a[j].ID })
return a
}
type Row struct {
ID uint64
Values []uint64
}
func (r *Row) Bitmap() []uint64 {
a := make([]uint64, rbf.ShardWidth/64)
for _, v := range r.Values {
a[v/64] |= 1 << (v % 64)
}
return a
}
// Union returns the union of r and other's values.
func (r *Row) Union(other *Row) []uint64 {
m := make(map[uint64]struct{})
for _, v := range r.Values {
m[v] = struct{}{}
}
for _, v := range other.Values {
m[v] = struct{}{}
}
a := make([]uint64, 0, len(m))
for v := range m {
a = append(a, v)
}
sort.Slice(a, func(i, j int) bool { return a[i] < a[j] })
return a
}
// Intersect returns the intersection of r & other's values.
func (r *Row) Intersect(other *Row) []uint64 {
m := make(map[uint64]struct{})
for _, v := range r.Values {
m[v] = struct{}{}
}
a := make([]uint64, 0)
used := make(map[uint64]struct{})
for _, v := range other.Values {
if _, ok := used[v]; ok {
continue
}
if _, ok := m[v]; ok {
used[v] = struct{}{}
a = append(a, v)
}
}
sort.Slice(a, func(i, j int) bool { return a[i] < a[j] })
return a
}
// QuickCheck executes fn multiple times with a different PRNG.
func QuickCheck(t *testing.T, fn func(t *testing.T, rand *rand.Rand)) {
for i := 0; i < *quickCheckN; i++ {
t.Run(fmt.Sprint(i), func(t *testing.T) {
fn(t, rand.New(rand.NewSource(int64(i))))
})
}
}
func highbits(v uint64) uint64 { return v >> 16 }
func lowbits(v uint64) uint16 { return uint16(v & 0xFFFF) }
// is32Bit returns true if the architecture is 32-bit.
func is32Bit() bool { return runtime.GOARCH == "386" }

672
rbf/tx.go Normal file
View file

@ -0,0 +1,672 @@
// 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"
"sort"
"github.com/benbjohnson/immutable"
"github.com/pilosa/pilosa/v2/roaring"
)
// Tx represents a transaction.
type Tx struct {
db *DB // parent db
meta [PageSize]byte // copy of current meta page
walID int64 // max WAL ID at start of tx
pageMap *immutable.Map // mapping of database pages to WAL IDs
writable bool // if true, tx can write
dirty bool // if true, changes have been made
}
// Commit completes the transaction and persists data changes.
func (tx *Tx) Commit() error {
if tx.db == nil {
return ErrTxClosed
}
// If any pages have been written, ensure we write a new meta page with
// the commit flag to mark the end of the transaction.
if tx.dirty {
if err := tx.writeMetaPage(MetaPageFlagCommit); err != nil {
return err
} else if err := tx.db.SyncWAL(); err != nil {
return err
}
tx.db.pageMap = tx.pageMap
}
// Disconnect transaction from DB.
return tx.db.removeTx(tx)
}
func (tx *Tx) Rollback() error {
if tx.db == nil {
return ErrTxClosed
}
// If any pages have been written, ensure we write a new meta page with
// the rollback flag to mark the end of the transaction. This allows us to
// discard pages in the transaction during playback of the WAL on open.
if tx.dirty {
if err := tx.writeMetaPage(MetaPageFlagRollback); err != nil {
return err
} else if err := tx.db.SyncWAL(); err != nil {
return err
}
}
// Disconnect transaction from DB.
return tx.db.removeTx(tx)
}
// Root returns the root page number for a bitmap. Returns 0 if the bitmap does not exist.
func (tx *Tx) Root(name string) (uint32, error) {
records, err := tx.rootRecords()
if err != nil {
return 0, err
}
i := sort.Search(len(records), func(i int) bool { return records[i].Name >= name })
if i >= len(records) || records[i].Name != name {
return 0, fmt.Errorf("bitmap not found: %q", name)
}
return records[i].Pgno, nil
}
// CreateBitmap creates a new empty bitmap with the given name.
// Returns an error if the bitmap already exists.
func (tx *Tx) CreateBitmap(name string) error {
if tx.db == nil {
return ErrTxClosed
} else if !tx.writable {
return ErrTxNotWritable
} else if name == "" {
return ErrBitmapNameRequired
}
// Read list of root records.
records, err := tx.rootRecords()
if err != nil {
return err
}
// Find btree by name. Exit if already exists.
index := sort.Search(len(records), func(i int) bool { return records[i].Name >= name })
if index < len(records) && records[index].Name == name {
return fmt.Errorf("bitmap already exists: %q", name)
}
//fmt.Println("CREATE BITMAP", name, index)
// Allocate new root page.
pgno, err := tx.allocate()
//fmt.Println("CREATE BITMAP @ PGNO", pgno)
if err != nil {
return err
}
// Write root page.
page := make([]byte, PageSize)
writePageNo(page, pgno)
writeFlags(page, PageTypeLeaf)
writeCellN(page, 0)
if err := tx.writePage(page); err != nil {
return err
}
// Insert into correct index.
records = append(records, nil)
copy(records[index+1:], records[index:])
records[index] = &RootRecord{Name: name, Pgno: pgno}
if err := tx.writeRootRecordPages(records); err != nil {
return fmt.Errorf("write bitmaps: %w", err)
}
return nil
}
func dump(r []*RootRecord) {
for _, i := range r {
fmt.Println("RECORD", i.Name, i.Pgno)
}
}
// DeleteBitmap removes a bitmap with the given name.
// Returns an error if the bitmap does not exist.
func (tx *Tx) DeleteBitmap(name string) error {
if tx.db == nil {
return ErrTxClosed
} else if !tx.writable {
return ErrTxNotWritable
} else if name == "" {
return ErrBitmapNameRequired
}
// Read list of root records.
records, err := tx.rootRecords()
if err != nil {
return err
}
// Find btree by name. Exit if it doesn't exist.
index := sort.Search(len(records), func(i int) bool { return records[i].Name >= name })
if index >= len(records) || records[index].Name != name {
return fmt.Errorf("bitmap does not exist: %q", name)
}
pgno := records[index].Pgno
// Deallocate all pages in the tree.
if err := tx.deallocateTree(pgno); err != nil {
return err
}
// Delete from record list & rewrite record pages.
records = append(records[:index], records[index+1:]...)
if err := tx.writeRootRecordPages(records); err != nil {
return fmt.Errorf("write bitmaps: %w", err)
}
return nil
}
// RenameBitmap updates the name of an existing bitmap.
// Returns an error if the bitmap does not exist.
func (tx *Tx) RenameBitmap(oldname, newname string) error {
if tx.db == nil {
return ErrTxClosed
} else if !tx.writable {
return ErrTxNotWritable
} else if oldname == "" || newname == "" {
return ErrBitmapNameRequired
}
// Read list of root records.
records, err := tx.rootRecords()
if err != nil {
return err
}
// Find btree by name. Exit if it doesn't exist.
index := sort.Search(len(records), func(i int) bool { return records[i].Name >= oldname })
if index >= len(records) || records[index].Name != oldname {
return fmt.Errorf("bitmap does not exist: %q", oldname)
}
// Update record name & rewrite record pages.
records[index].Name = newname
if err := tx.writeRootRecordPages(records); err != nil {
return fmt.Errorf("write bitmaps: %w", err)
}
return nil
}
// rootRecords returns a list of root records.
func (tx *Tx) rootRecords() ([]*RootRecord, error) {
var records []*RootRecord
for pgno := readMetaRootRecordPageNo(tx.meta[:]); pgno != 0; {
page, err := tx.readPage(pgno)
if err != nil {
return nil, err
}
// Read all records on the page.
a, err := readRootRecords(page)
if err != nil {
return nil, err
}
records = append(records, a...)
// Read next overflow page number.
pgno = readRootRecordOverflowPgno(page)
}
return records, nil
}
// writeRootRecordPages writes a list of root record pages.
func (tx *Tx) writeRootRecordPages(records []*RootRecord) (err error) {
// Release all existing root record pages.
for pgno := readMetaRootRecordPageNo(tx.meta[:]); pgno != 0; {
page, err := tx.readPage(pgno)
if err != nil {
return err
}
if err := tx.deallocate(pgno); err != nil {
return err
}
pgno = readRootRecordOverflowPgno(page)
}
// Exit early if no records exist.
if len(records) == 0 {
writeMetaRootRecordPageNo(tx.meta[:], 0)
return nil
}
// Ensure records are in sorted order.
sort.Slice(records, func(i, j int) bool { return records[i].Name < records[j].Name })
// Allocate initial root record page.
pgno, err := tx.allocate()
if err != nil {
return err
}
writeMetaRootRecordPageNo(tx.meta[:], pgno)
// Write new root record pages.
for i := 0; len(records) != 0; i++ {
// Initialize page & write as many records as will fit.
page := make([]byte, PageSize)
writePageNo(page, pgno)
writeFlags(page, PageTypeRootRecord)
if records, err = writeRootRecords(page, records); err != nil {
return err
}
// Allocate next and write overflow if we have remaining records.
if len(records) != 0 {
if pgno, err = tx.allocate(); err != nil {
return err
}
writeRootRecordOverflowPgno(page, pgno)
}
// Write page to disk.
if err := tx.writePage(page); err != nil {
return err
}
}
return nil
}
// Add sets a given bit on the bitmap.
func (tx *Tx) Add(name string, a ...uint64) (changed bool, err error) {
if tx.db == nil {
return false, ErrTxClosed
} else if !tx.writable {
return false, ErrTxNotWritable
} else if name == "" {
return false, ErrBitmapNameRequired
}
c, err := tx.Cursor(name)
if err != nil {
return false, err
}
for _, v := range a {
if vchanged, err := c.Add(v); err != nil {
return changed, err
} else if vchanged {
changed = true
}
}
return changed, nil
}
// Remove unsets a given bit on the bitmap.
func (tx *Tx) Remove(name string, a ...uint64) (changed bool, err error) {
if tx.db == nil {
return false, ErrTxClosed
} else if !tx.writable {
return false, ErrTxNotWritable
} else if name == "" {
return false, ErrBitmapNameRequired
}
c, err := tx.Cursor(name)
if err != nil {
return false, err
}
for _, v := range a {
if vchanged, err := c.Remove(v); err != nil {
return changed, err
} else if vchanged {
changed = true
}
}
return changed, nil
}
// Contains returns true if the given bit is set on the bitmap.
func (tx *Tx) Contains(name string, v uint64) (bool, error) {
if tx.db == nil {
return false, ErrTxClosed
} else if name == "" {
return false, ErrBitmapNameRequired
}
c, err := tx.Cursor(name)
if err != nil {
return false, err
}
return c.Contains(v)
}
// Cursor returns an instance of a cursor this bitmap.
func (tx *Tx) Cursor(name string) (*Cursor, error) {
if tx.db == nil {
return nil, ErrTxClosed
} else if name == "" {
return nil, ErrBitmapNameRequired
}
root, err := tx.Root(name)
if err != nil {
return nil, err
}
c := Cursor{tx: tx}
c.stack.elems[0] = stackElem{pgno: root}
return &c, nil
}
// Check verifies the integrity of the database.
func (tx *Tx) Check() error {
if tx.db == nil {
return ErrTxClosed
}
if err := tx.checkPageAllocations(); err != nil {
return fmt.Errorf("page allocations: %w", err)
}
return nil
}
// checkPageAllocations ensures that all pages are either in-use or on the freelist.
func (tx *Tx) checkPageAllocations() error {
freePageSet, err := tx.freePageSet()
if err != nil {
return err
}
inusePageSet, err := tx.inusePageSet()
if err != nil {
return err
}
// Iterate over all pages and ensure they are either in-use or free.
// They should not be BOTH in-use or free or NEITHER in-use or free.
pageN := readMetaPageN(tx.meta[:])
for pgno := uint32(1); pgno < pageN; pgno++ {
_, isInuse := inusePageSet[pgno]
_, isFree := freePageSet[pgno]
if isInuse && isFree {
return fmt.Errorf("page in-use & free: pgno=%d", pgno)
} else if !isInuse && !isFree {
return fmt.Errorf("page not in-use & not free: pgno=%d", pgno)
}
}
return nil
}
// freePageSet returns the set of pages in the freelist.
func (tx *Tx) freePageSet() (map[uint32]struct{}, error) {
m := make(map[uint32]struct{})
c := Cursor{tx: tx}
c.stack.elems[0] = stackElem{pgno: readMetaFreelistPageNo(tx.meta[:])}
if err := c.First(); err == io.EOF {
return m, nil
} else if err != nil {
return m, err
}
for {
if err := c.Next(); err == io.EOF {
return m, nil
} else if err != nil {
return m, err
}
cell := c.cell()
for _, v := range cell.Values() {
pgno := uint32((cell.Key << 16) & uint64(v))
m[pgno] = struct{}{}
}
}
}
// inusePageSet returns the set of pages in use by the root records or b-trees.
func (tx *Tx) inusePageSet() (map[uint32]struct{}, error) {
m := make(map[uint32]struct{})
m[0] = struct{}{} // meta page
// Traverse root record linked list and mark each page as in-use.
for pgno := readMetaRootRecordPageNo(tx.meta[:]); pgno != 0; {
m[pgno] = struct{}{}
page, err := tx.readPage(pgno)
if err != nil {
return nil, err
}
pgno = readRootRecordOverflowPgno(page)
}
// Traverse freelist and mark pages as in-use.
if err := tx.walkTree(readMetaFreelistPageNo(tx.meta[:]), func(pgno uint32) error {
m[pgno] = struct{}{}
return nil
}); err != nil {
return m, err
}
// Traverse every b-tree and mark pages as in-use.
records, err := tx.rootRecords()
if err != nil {
return m, err
}
for _, record := range records {
if err := tx.walkTree(record.Pgno, func(pgno uint32) error {
m[pgno] = struct{}{}
return nil
}); err != nil {
return m, err
}
}
return m, nil
}
// walkTree recursively iterates over a page and all its children.
func (tx *Tx) walkTree(pgno uint32, fn func(uint32) error) error {
// Execute callback.
if err := fn(pgno); err != nil {
return err
}
// Read page and iterate over children.
page, err := tx.readPage(pgno)
if err != nil {
return err
}
switch typ := readFlags(page); typ {
case PageTypeBranch:
for i, n := 0, readCellN(page); i < n; i++ {
cell := readBranchCell(page, i)
if cell.Flags&ContainerTypeBitmap != 0 { // bitmap cell (cannot traverse into)
if err := fn(cell.Pgno); err != nil {
return err
}
} else {
if err := tx.walkTree(cell.Pgno, fn); err != nil {
return err
}
}
}
return nil
case PageTypeLeaf:
return nil
default:
return fmt.Errorf("rbf.Tx.forEachTreePage(): invalid page type: pgno=%d type=%d", pgno, typ)
}
}
// allocate returns a page number for a new available page. This page may be
// pulled from the free list or, if no free pages are available, it will be
// created by extending the file size.
func (tx *Tx) allocate() (uint32, error) {
// Attempt to find page in freelist.
pgno, err := tx.nextFreelistPageNo()
if err != nil {
return 0, err
} else if pgno != 0 {
c := Cursor{tx: tx}
c.stack.elems[0] = stackElem{pgno: readMetaFreelistPageNo(tx.meta[:])}
if changed, err := c.Remove(uint64(pgno)); err != nil {
return 0, err
} else if !changed {
panic(fmt.Sprintf("tx.Tx.allocate(): double alloc: %d", pgno))
}
return pgno, nil
}
// Increment the total page count by one and return the last page.
pgno = readMetaPageN(tx.meta[:])
writeMetaPageN(tx.meta[:], pgno+1)
return pgno, nil
}
func (tx *Tx) nextFreelistPageNo() (uint32, error) {
c := Cursor{tx: tx}
c.stack.elems[0] = stackElem{pgno: readMetaFreelistPageNo(tx.meta[:])}
if err := c.First(); err == io.EOF {
return 0, nil
} else if err != nil {
return 0, err
}
cell := c.cell()
v := cell.firstValue()
pgno := uint32((cell.Key << 16) | uint64(v))
return pgno, nil
}
// deallocate releases a page number to the freelist.
func (tx *Tx) deallocate(pgno uint32) error {
c := Cursor{tx: tx}
c.stack.elems[0] = stackElem{pgno: readMetaFreelistPageNo(tx.meta[:])}
if changed, err := c.Add(uint64(pgno)); err != nil {
return err
} else if !changed {
panic(fmt.Sprintf("rbf.Tx.deallocate(): double free: %d", pgno))
}
return nil
}
// deallocateTree recursively all pages in a btree.
func (tx *Tx) deallocateTree(pgno uint32) error {
page, err := tx.readPage(pgno)
if err != nil {
return err
}
switch typ := readFlags(page); typ {
case PageTypeBranch:
for i, n := 0, readCellN(page); i < n; i++ {
cell := readBranchCell(page, i)
if cell.Flags&ContainerTypeBitmap == 0 { // leaf/branch child page
if err := tx.deallocateTree(cell.Pgno); err != nil {
return err
}
} else {
if err := tx.deallocate(cell.Pgno); err != nil { // bitmap child page
return err
}
}
}
return nil
case PageTypeLeaf:
return tx.deallocate(pgno)
default:
return fmt.Errorf("rbf.Tx.deallocateTree(): invalid page type: pgno=%d type=%d", pgno, typ)
}
}
func (tx *Tx) readPage(pgno uint32) ([]byte, error) {
// fmt.Println("readPage", pgno)
// Meta page is always cached on the transaction.
if pgno == 0 {
return tx.meta[:], nil
}
pageN := readMetaPageN(tx.meta[:])
if pgno > pageN {
return nil, fmt.Errorf("rbf: page read out of bounds: pgno=%d max=%d", pgno, pageN)
}
return tx.db.readPage(tx.pageMap, pgno)
}
func (tx *Tx) writePage(page []byte) error {
// fmt.Println("writePage", readPageNo(page))
// Write page to WAL and obtain position in WAL.
walID, err := tx.db.writeWALPage(page, false)
if err != nil {
return err
}
// Mark transaction as dirty so we write a meta page on commit/rollback.
tx.dirty = true
// Update page map with WAL position.
tx.pageMap = tx.pageMap.Set(readPageNo(page), walID)
return nil
}
func (tx *Tx) writeBitmapPage(pgno uint32, page []byte) error {
// Write bitmap to WAL and obtain WAL position of the actual page data (not the prefix page).
walID, err := tx.db.writeBitmapPage(pgno, page)
if err != nil {
return err
}
// Mark transaction as dirty so we write a meta page on commit/rollback.
tx.dirty = true
// Update page map with WAL position.
tx.pageMap = tx.pageMap.Set(pgno, walID)
return nil
}
func (tx *Tx) writeMetaPage(flag uint32) error {
// Set meta flags.
writeFlags(tx.meta[:], flag)
// Write page to WAL and obtain position in WAL.
walID, err := tx.db.writeWALPage(tx.meta[:], true)
if err != nil {
return err
}
tx.pageMap = tx.pageMap.Set(uint32(0), walID)
return nil
}
func (tx *Tx) AddRoaring(name string, bm *roaring.Bitmap) (changed bool, err error) {
c, err := tx.Cursor(name)
if err != nil {
return false, err
}
return c.AddRoaring(bm)
}

465
rbf/tx_test.go Normal file
View file

@ -0,0 +1,465 @@
// 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_test
import (
"fmt"
"math/rand"
"testing"
"time"
"github.com/pilosa/pilosa/v2/rbf"
)
func TestTx_CommitRollback(t *testing.T) {
t.Run("NoReopen", func(t *testing.T) {
db := MustOpenDB(t)
defer MustCloseDB(t, db)
// Create bitmap in transaction but rollback.
if tx, err := db.Begin(true); err != nil {
t.Fatal(err)
} else if err := tx.CreateBitmap("x"); err != nil {
t.Fatal(err)
} else if err := tx.Rollback(); err != nil {
t.Fatal(err)
}
// Create bitmap in transaction again but commit.
if tx, err := db.Begin(true); err != nil {
t.Fatal(err)
} else if err := tx.CreateBitmap("x"); err != nil {
t.Fatal(err)
} else if err := tx.Commit(); err != nil {
t.Fatal(err)
}
// Create bitmap again but it should fail as it already exists.
if tx, err := db.Begin(true); err != nil {
t.Fatal(err)
} else if err := tx.CreateBitmap("x"); err == nil || err.Error() != `bitmap already exists: "x"` {
_ = tx.Rollback()
t.Fatal(err)
} else if err := tx.Commit(); err != nil {
t.Fatal(err)
}
})
t.Run("Reopen", func(t *testing.T) {
db := MustOpenDB(t)
defer func() { MustCloseDB(t, db) }()
// Create bitmap in transaction but rollback.
if tx, err := db.Begin(true); err != nil {
t.Fatal(err)
} else if err := tx.CreateBitmap("x"); err != nil {
t.Fatal(err)
} else if err := tx.Rollback(); err != nil {
t.Fatal(err)
}
db = MustReopenDB(t, db)
// Create bitmap in transaction again but commit.
if tx, err := db.Begin(true); err != nil {
t.Fatal(err)
} else if err := tx.CreateBitmap("x"); err != nil {
t.Fatal(err)
} else if err := tx.Commit(); err != nil {
t.Fatal(err)
}
db = MustReopenDB(t, db)
// Create bitmap again but it should fail as it already exists.
if tx, err := db.Begin(true); err != nil {
t.Fatal(err)
} else if err := tx.CreateBitmap("x"); err == nil || err.Error() != `bitmap already exists: "x"` {
_ = tx.Rollback()
t.Fatal(err)
} else if err := tx.Commit(); err != nil {
t.Fatal(err)
}
})
t.Run("SingleWriter", func(t *testing.T) {
db := MustOpenDB(t)
defer MustCloseDB(t, db)
// Start write transaction.
ch0 := make(chan struct{})
tx0 := MustBegin(t, db, true)
go func() {
<-ch0
_ = tx0.Rollback()
}()
// Start separate write transaction in different goroutine.
ch1 := make(chan struct{})
go func() {
tx1 := MustBegin(t, db, true)
close(ch1)
_ = tx1.Commit()
}()
// Ensure second tx doesn't start.
select {
case <-ch1:
t.Fatal("second tx started while first tx active")
case <-time.After(10 * time.Millisecond):
}
// Finish first transaction.
close(ch0)
select {
case <-ch1:
case <-time.After(10 * time.Millisecond):
t.Fatal("second tx should have started after first tx closed")
}
})
}
func TestTx_Add(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)
}
if _, err := tx.Add("x", 1); err != nil {
t.Fatal(err)
} else if _, err := tx.Add("x", 10); err != nil {
t.Fatal(err)
} else if _, err := tx.Add("x", 3); err != nil {
t.Fatal(err)
}
for _, v := range []uint64{1, 3, 10} {
if ok, err := tx.Contains("x", v); err != nil {
t.Fatal(err)
} else if !ok {
t.Fatalf("Tx.Contains(%d): expected true", v)
}
}
if ok, err := tx.Contains("x", 2); err != nil {
t.Fatal(err)
} else if ok {
t.Fatal("Tx.Contains(): expected false")
}
}
func TestTx_DeleteBitmap(t *testing.T) {
db := MustOpenDB(t)
defer MustCloseDB(t, db)
tx := MustBegin(t, db, true)
defer MustRollback(t, tx)
// Create bitmap & add value.
if err := tx.CreateBitmap("x"); err != nil {
t.Fatal(err)
} else if _, err := tx.Add("x", 1); err != nil {
t.Fatal(err)
}
// Recreate bitmap & ensure value does not exist.
if err := tx.DeleteBitmap("x"); err != nil {
t.Fatal(err)
} else if err := tx.CreateBitmap("x"); err != nil {
t.Fatal(err)
} else if ok, err := tx.Contains("x", 1); err != nil {
t.Fatal(err)
} else if ok {
t.Fatal("expected no value in recreated bitmap")
}
}
func TestTx_RenameBitmap(t *testing.T) {
db := MustOpenDB(t)
defer MustCloseDB(t, db)
tx := MustBegin(t, db, true)
defer MustRollback(t, tx)
// Create bitmap & add value.
if err := tx.CreateBitmap("x"); err != nil {
t.Fatal(err)
} else if _, err := tx.Add("x", 1); err != nil {
t.Fatal(err)
}
// Rename bitmap & ensure value still exists.
if err := tx.RenameBitmap("x", "y"); err != nil {
t.Fatal(err)
} else if ok, err := tx.Contains("y", 1); err != nil {
t.Fatal(err)
} else if !ok {
t.Fatal("expected value in renamed bitmap")
}
}
func TestTx_Add_Quick(t *testing.T) {
if testing.Short() {
t.Skip("-short enabled, skipping")
} else if is32Bit() {
t.Skip("32-bit build, skipping quick check tests")
} else if rbf.RaceEnabled {
t.Skip("race detection enabled, skipping")
}
QuickCheck(t, func(t *testing.T, rand *rand.Rand) {
t.Parallel()
db := MustOpenDB(t)
defer MustCloseDB(t, db)
tx := MustBegin(t, db, true)
defer MustRollback(t, tx)
values := GenerateValues(rand, 100000)
if err := tx.CreateBitmap("x"); err != nil {
t.Fatal(err)
}
// Insert values in random order.
for _, i := range rand.Perm(len(values)) {
v := values[i]
if _, err := tx.Add("x", v); err != nil {
t.Fatalf("Add(%d) i=%d err=%q", v, i, err)
}
}
// Verify all bits are written.
for i, v := range values {
if ok, err := tx.Contains("x", v); !ok || err != nil {
t.Fatalf("Contains(%d)=(%v,%v) i=%d hi=%d lo=%d", v, ok, err, i, highbits(v), lowbits(v))
}
}
})
}
func TestTx_AddRemove_Quick(t *testing.T) {
if testing.Short() {
t.Skip("-short enabled, skipping")
} else if is32Bit() {
t.Skip("32-bit build, skipping quick check tests")
} else if rbf.RaceEnabled {
t.Skip("race detection enabled, skipping")
}
QuickCheck(t, func(t *testing.T, rand *rand.Rand) {
t.Parallel()
db := MustOpenDB(t)
defer MustCloseDB(t, db)
tx := MustBegin(t, db, true)
defer MustRollback(t, tx)
values := GenerateValues(rand, 100000)
if err := tx.CreateBitmap("x"); err != nil {
t.Fatal(err)
}
// Insert values in random order.
for _, i := range rand.Perm(len(values)) {
if _, err := tx.Add("x", values[i]); err != nil {
t.Fatalf("Add(%d) i=%d err=%q", values[i], i, err)
}
}
// Remove half the values in random order.
for _, i := range rand.Perm(len(values)) {
if _, err := tx.Remove("x", values[i]); err != nil {
t.Fatalf("Remove(%d) i=%d err=%q", values[i], i, err)
}
}
// Verify all bits are removed.
for i, v := range values {
if ok, err := tx.Contains("x", v); ok || err != nil {
t.Fatalf("Contains(%d)=(%v,%v) i=%d hi=%d lo=%d", v, ok, err, i, highbits(v), lowbits(v))
}
}
// Re-add those values back in.
for _, i := range rand.Perm(len(values)) {
if _, err := tx.Add("x", values[i]); err != nil {
t.Fatalf("Re-Add(%d) i=%d err=%q", values[i], i, err)
}
}
// Verify all bits are written.
for i, v := range values {
if ok, err := tx.Contains("x", v); !ok || err != nil {
t.Fatalf("Contains(%d)=(%v,%v) i=%d hi=%d lo=%d", v, ok, err, i, highbits(v), lowbits(v))
}
}
})
}
func TestTx_Multiple_CreateBitmap(t *testing.T) {
rand := rand.New(rand.NewSource(0))
db := MustOpenDB(t)
defer MustCloseDB(t, db)
tx := MustBegin(t, db, true)
defer MustRollback(t, tx)
values := GenerateValues(rand, 2)
if err := tx.CreateBitmap("x/1"); err != nil {
t.Fatal(err)
}
// Insert values in random order.
for _, i := range rand.Perm(len(values)) {
if _, err := tx.Add("x/1", values[i]); err != nil {
t.Fatalf("Add(%d) i=%d err=%q", values[i], i, err)
}
}
if err := tx.Commit(); err != nil {
t.Fatalf("Commit 1 err=%q", err)
}
tx1 := MustBegin(t, db, true)
defer func() { _ = tx1.Rollback() }()
if err := tx1.CreateBitmap("x/2"); err != nil {
t.Fatal(err)
}
// Insert values in random order.
for _, i := range rand.Perm(len(values)) {
if _, err := tx1.Add("x/2", values[i]); err != nil {
t.Fatalf("Add(%d) i=%d err=%q", values[i], i, err)
}
}
if err := tx1.Commit(); err != nil {
t.Fatalf("Commit 2 err=%q", err)
}
}
func TestTx_CursorCrashArray(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)
}
c, err := tx.Cursor("x")
if err != nil {
t.Fatal(err)
}
//setArray(t, 0, 2379, c)
//setArray(t, 1, 2337, c)
setArray(t, 32, 1216, c)
setArray(t, 33, 1195, c)
setArray(t, 48, 1186, c)
setArray(t, 49, 1223, c)
setArray(t, 50, 1223, c)
}
func TestTx_CursorCrashBitmap(t *testing.T) {
if testing.Short() {
t.Skip("-short enabled, skipping")
}
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)
}
c, err := tx.Cursor("x")
if err != nil {
t.Fatal(err)
}
setArray(t, 0, 22510, c)
setArray(t, 1, 23584, c)
}
func setArray(tb testing.TB, key, num int, c *rbf.Cursor) {
for i := uint64(0); i < uint64(num); i++ {
v := i | (uint64(key) << 16)
if _, err := c.Add(v); err != nil {
tb.Fatal(err)
}
}
}
func BenchmarkTx_Add(b *testing.B) {
for _, n := range []int{10000, 100000, 1000000} {
b.Run(fmt.Sprint(n), func(b *testing.B) {
rand := rand.New(rand.NewSource(0))
values := make([]uint64, n)
for i := range values {
values[i] = uint64(rand.Intn(rbf.ShardWidth))
}
b.ResetTimer()
t := time.Now()
b.ReportAllocs()
for i := 0; i < b.N; i++ {
func() {
db := MustOpenDB(b)
defer MustCloseDB(b, db)
tx := MustBegin(b, db, true)
defer MustRollback(b, tx)
for _, v := range values {
if _, err := tx.Add("x", v); err != nil {
b.Fatalf("Add(%d) i=%d err=%q", v, i, err)
}
}
}()
}
b.ReportMetric(float64(time.Since(t).Nanoseconds())/float64(n*b.N), "ns/op")
})
}
}
func BenchmarkTx_Contains(b *testing.B) {
for _, n := range []int{10000, 100000, 1000000} {
b.Run(fmt.Sprint(n), func(b *testing.B) {
rand := rand.New(rand.NewSource(0))
values := make([]uint64, n)
for i := range values {
values[i] = uint64(rand.Intn(rbf.ShardWidth))
}
db := MustOpenDB(b)
defer MustCloseDB(b, db)
tx := MustBegin(b, db, true)
defer MustRollback(b, tx)
b.ResetTimer()
t := time.Now()
b.ReportAllocs()
for i := 0; i < b.N; i++ {
for _, v := range values {
if _, err := tx.Contains("x", v); err != nil {
b.Fatalf("Contains(%d) i=%d err=%q", v, i, err)
}
}
}
b.ReportMetric(float64(time.Since(t).Nanoseconds())/float64(n*b.N), "ns/op")
})
}
}

241
rbf/wal.go Normal file
View file

@ -0,0 +1,241 @@
// 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"
"os"
"path/filepath"
"syscall"
"github.com/pilosa/pilosa/v2/syswrap"
)
// WALSegment represents a single file in the WAL.
type WALSegment struct {
minWALID int64 // base WALID; calculated from path
path string // path to file
w *os.File // write handle
data []byte // read-only mmap data
pageN int // number of written pages
}
// NewWALSegment returns a new instance of WALSegment for a given path.
func NewWALSegment(path string) *WALSegment {
return &WALSegment{
path: path,
}
}
// Path returns the path the segment was initialized with.
func (s *WALSegment) Path() string { return s.path }
// MinWALID returns the initial WAL ID of the segment. Only available after Open().
func (s *WALSegment) MinWALID() int64 { return s.minWALID }
// MaxWALID returns the maximum WAL ID of the segment. Only available after Open().
func (s *WALSegment) MaxWALID() int64 {
return s.minWALID + int64(s.pageN) - 1
}
// PageN returns the number of pages in the segment.
func (s *WALSegment) PageN() int { return s.pageN }
// Size returns the current size of the segment, in bytes.
func (s *WALSegment) Size() int64 { return int64(s.pageN) * PageSize }
func (s *WALSegment) Open() (err error) {
// Extract base WAL ID and validate path.
if s.minWALID, err = ParseWALSegmentPath(s.path); err != nil {
return err
}
// Determine file size & create if necessary.
var sz int64
if fi, err := os.Stat(s.path); os.IsNotExist(err) {
if f, err := os.OpenFile(s.path, os.O_RDWR|os.O_CREATE, 0666); err != nil {
return fmt.Errorf("touch wal segment file: %w", err)
} else if err := f.Close(); err != nil {
return fmt.Errorf("close touched wal segment file: %w", err)
}
} else if err != nil {
return fmt.Errorf("stat wal segment file: %w", err)
} else {
sz = fi.Size()
}
// Determine page count & truncate if a partial page is written.
s.pageN = int(sz / PageSize)
if sz%PageSize != 0 {
sz = int64(s.pageN * PageSize)
if err := os.Truncate(s.path, sz); err != nil {
return fmt.Errorf("truncate wal segment file: %w", err)
}
}
// Default the mmap size to the max size plus a page of padding for bitmap pages.
// If the actual size is larger, then increase to that size.
mmapSize := int64(MaxWALSegmentFileSize + PageSize)
if sz > mmapSize {
mmapSize = sz
}
// Open file as a read-only memory map.
if f, err := os.OpenFile(s.path, os.O_RDONLY, 0666); err != nil {
return fmt.Errorf("open wal segment file: %w", err)
} else if s.data, err = syswrap.Mmap(int(f.Fd()), 0, int(mmapSize), syscall.PROT_READ, syscall.MAP_SHARED); err != nil {
f.Close()
return fmt.Errorf("mmap wal segment: %w", err)
} else if err := f.Close(); err != nil {
return fmt.Errorf("close wal segment mmap file: %w", err)
}
return nil
}
// Close closes the write handle and the read-only mmap.
func (s *WALSegment) Close() error {
if err := s.CloseForWrite(); err != nil {
return err
}
if s.data != nil {
if err := syswrap.Munmap(s.data); err != nil {
return err
}
s.data = nil
}
return nil
}
// CloseForWrite closes the write handle, if initialized.
func (s *WALSegment) CloseForWrite() error {
if s.w != nil {
if err := s.w.Close(); err != nil {
return err
}
s.w = nil
}
return nil
}
// ReadWALPage reads a single page at the given WAL ID.
func (s *WALSegment) ReadWALPage(walID int64) ([]byte, error) {
// Ensure requested ID is contained in this file.
if walID < s.minWALID || walID > s.minWALID+int64(s.pageN) {
return nil, fmt.Errorf("wal segment page read out of range: id=%d base=%d pageN=%d", walID, s.minWALID, s.pageN)
}
offset := (walID - s.minWALID) * PageSize
return s.data[offset : offset+PageSize], nil
}
// WriteWALPage writes a single page to the WAL segment and returns its WAL identifier.
func (s *WALSegment) WriteWALPage(page []byte, isMeta bool) (walID int64, err error) {
assert(len(page) == PageSize)
// Initialize write file handle if not yet initialized.
if s.w == nil {
if s.w, err = os.OpenFile(s.path, os.O_WRONLY, 0666); err != nil {
return 0, fmt.Errorf("open wal segment write handle: %w", err)
}
}
// Determine current WAL position.
walID = s.minWALID + int64(s.pageN)
// Write WAL ID if this is a meta page.
if isMeta {
writeMetaWALID(page, walID)
// TODO: Write meta page checksum
}
// Write page at position & increment page count.
if _, err := s.w.WriteAt(page, int64(s.pageN*PageSize)); err != nil {
return 0, fmt.Errorf("wal segment write: %w", err)
}
s.pageN++
return walID, nil
}
// Sync flushes all changes to disk.
func (s *WALSegment) Sync() error {
if s.w == nil {
return nil
}
return s.w.Sync()
}
// trimBitmapHeaderTrailer removes the last page if the last page is a bitmap header.
// This should only be called on the last segment during recovery. A bitmap
// header write is a 2-page write so a partial write would corrupt the WAL.
func (s *WALSegment) trimBitmapHeaderTrailer() error {
// Skip if there are no pages in this segment.
if s.PageN() == 0 {
return nil
}
// Skip if this is not a bitmap header page.
if page, err := s.ReadWALPage(s.MaxWALID()); err != nil {
return err
} else if !IsBitmapHeader(page) {
return nil
}
// Truncate last page and reduce page count.
if err := os.Truncate(s.Path(), s.Size()-PageSize); err != nil {
return err
}
s.pageN--
return nil
}
// FormatWALSegmentPath returns a path for a WAL segment using a WAL ID.
func FormatWALSegmentPath(walID int64) string {
return fmt.Sprintf("%016x.wal", walID)
}
// ParseWALSegmentPath returns the WAL ID for a given WAL segment path.
func ParseWALSegmentPath(s string) (walID int64, err error) {
if _, err = fmt.Sscanf(filepath.Base(s), "%016x.wal", &walID); err != nil {
return 0, fmt.Errorf("invalid WAL path: %s", s)
}
return walID, nil
}
// uint32Hasher implements Hasher for uint32 keys.
type uint32Hasher struct{}
// Hash returns a hash for key.
func (h *uint32Hasher) Hash(key interface{}) uint32 {
return hashUint64(uint64(key.(uint32)))
}
// Equal returns true if a is equal to b. Otherwise returns false.
// Panics if a and b are not ints.
func (h *uint32Hasher) Equal(a, b interface{}) bool {
return a.(uint32) == b.(uint32)
}
// hashUint64 returns a 32-bit hash for a 64-bit value.
func hashUint64(value uint64) uint32 {
hash := value
for value > 0xffffffff {
value /= 0xffffffff
hash ^= value
}
return uint32(hash)
}

138
rbf/wal_test.go Normal file
View file

@ -0,0 +1,138 @@
// 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_test
import (
"bytes"
"encoding/hex"
"io/ioutil"
"math/rand"
"os"
"path/filepath"
"testing"
"github.com/pilosa/pilosa/v2/rbf"
)
func TestWALSegment_Open(t *testing.T) {
t.Run("OK", func(t *testing.T) {
s := MustOpenWALSegment(t, 10)
defer MustCloseWALSegment(t, s)
if got, want := s.MinWALID(), int64(10); got != want {
t.Fatalf("Base()=%d, want %d", got, want)
} else if got, want := s.PageN(), 0; got != want {
t.Fatalf("PageN()=%d, want %d", got, want)
}
})
// TODO(BBJ): Test open w/ partially written pages.
}
func TestWALSegment_WritePage(t *testing.T) {
rand := rand.New(rand.NewSource(0))
s := MustOpenWALSegment(t, 10)
defer MustCloseWALSegment(t, s)
pages := [][]byte{
make([]byte, rbf.PageSize),
make([]byte, rbf.PageSize),
}
rand.Read(pages[0])
rand.Read(pages[1])
// Write first page.
if walID, err := s.WriteWALPage(pages[0], false); err != nil {
t.Fatal(err)
} else if got, want := walID, int64(10); got != want {
t.Fatalf("WALID=%d, want %d", got, want)
} else if got, want := s.PageN(), 1; got != want {
t.Fatalf("PageN()=%d, want %d", got, want)
}
// Write second page.
if walID, err := s.WriteWALPage(pages[1], false); err != nil {
t.Fatal(err)
} else if got, want := walID, int64(11); got != want {
t.Fatalf("WALID=%d, want %d", got, want)
} else if got, want := s.PageN(), 2; got != want {
t.Fatalf("PageN()=%d, want %d", got, want)
}
// Read & verify first page.
if buf, err := s.ReadWALPage(10); err != nil {
t.Fatal(err)
} else if !bytes.Equal(pages[0], buf) {
t.Fatalf("unexpected first page:\n%s", hex.Dump(buf))
}
// Read & verify second page.
if buf, err := s.ReadWALPage(11); err != nil {
t.Fatal(err)
} else if !bytes.Equal(pages[1], buf) {
t.Fatal("unexpected second page")
}
}
func TestFormatWALSegmentPath(t *testing.T) {
if got, want := rbf.FormatWALSegmentPath(1234), "00000000000004d2.wal"; got != want {
t.Fatalf("FormatWALSegmentPath()=%q, want %q", got, want)
}
}
func TestParseWALSegmentPath(t *testing.T) {
t.Run("OK", func(t *testing.T) {
if walID, err := rbf.ParseWALSegmentPath("/tmp/00000000000004d2.wal"); err != nil {
t.Fatal(err)
} else if got, want := walID, int64(1234); got != want {
t.Fatalf("ParseWALSegmentPath()=%q, want %q", got, want)
}
})
t.Run("ErrInvalidWALPath", func(t *testing.T) {
if _, err := rbf.ParseWALSegmentPath("/tmp/xyz"); err == nil || err.Error() != "invalid WAL path: /tmp/xyz" {
t.Fatalf("unexpected error: %#v", err)
}
})
}
// MustOpenWALSegment opens a WAL segment in a temporary path. Fails on error.
func MustOpenWALSegment(tb testing.TB, walID int64) *rbf.WALSegment {
tb.Helper()
dir, err := ioutil.TempDir("", "")
if err != nil {
tb.Fatal(err)
}
path := filepath.Join(dir, rbf.FormatWALSegmentPath(walID))
if err := ioutil.WriteFile(path, nil, 0666); err != nil {
tb.Fatal(err)
}
s := rbf.NewWALSegment(path)
if err := s.Open(); err != nil {
tb.Fatal(err)
}
return s
}
// MustCloseWALSegment closes s. Fails on error.
func MustCloseWALSegment(tb testing.TB, s *rbf.WALSegment) {
tb.Helper()
if err := s.Close(); err != nil {
tb.Fatal(err)
} else if err := os.Remove(s.Path()); err != nil {
tb.Fatal(err)
}
}

25
view.go
View file

@ -42,11 +42,12 @@ const (
// view represents a container for field data.
type view struct {
mu sync.RWMutex
path string
index string
field string
name string
mu sync.RWMutex
path string
index string
field string
name string
qualifiedName string
holder *Holder
@ -68,10 +69,11 @@ type view struct {
// newView returns a new instance of View.
func newView(holder *Holder, path, index, field, name string, fieldOptions FieldOptions) *view {
return &view{
path: path,
index: index,
field: field,
name: name,
path: path,
index: index,
field: field,
name: name,
qualifiedName: FormatQualifiedViewName(index, field, name),
holder: holder,
@ -512,3 +514,8 @@ type viewInfoSlice []*ViewInfo
func (p viewInfoSlice) Swap(i, j int) { p[i], p[j] = p[j], p[i] }
func (p viewInfoSlice) Len() int { return len(p) }
func (p viewInfoSlice) Less(i, j int) bool { return p[i].Name < p[j].Name }
// FormatQualifiedViewName generates a qualified name for the view to be used with Tx operations.
func FormatQualifiedViewName(index, field, view string) string {
return fmt.Sprintf("%s\x00%s\x00%s\x00", index, field, view)
}

100
xrbrsupport.go Normal file
View file

@ -0,0 +1,100 @@
// 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 pilosa
import (
"fmt"
"github.com/pilosa/pilosa/v2/rbf"
"github.com/pilosa/pilosa/v2/roaring"
)
type Converter interface {
Convert(index, field, view string, shard uint64, rb *roaring.Bitmap) error
Shutdown()
}
type RBFConverter struct {
Dbs map[string]*rbf.DB
Base string
}
func (rbc *RBFConverter) GetOrCreateDB(index string, shard uint64) (*rbf.DB, error) {
key := fmt.Sprintf("%s/%d", index, shard)
db, found := rbc.Dbs[key]
if found {
return db, nil
}
path := rbc.Base + "/" + key
db = rbf.NewDB(path)
err := db.Open()
if err != nil {
return nil, err
}
rbc.Dbs[key] = db
return db, nil
}
func (rbc *RBFConverter) Shutdown() {
for key, db := range rbc.Dbs {
fmt.Println("Shutdown", key)
db.Close()
}
}
func (rbc *RBFConverter) Convert(index, field, view string, shard uint64, rb *roaring.Bitmap) error {
fmt.Println("CONVERT", index, field, view, shard)
db, err := rbc.GetOrCreateDB(index, shard)
if err != nil {
return err
}
tx, err := db.Begin(true)
if err != nil {
return err
}
defer func() { _ = tx.Rollback() }()
name := fmt.Sprintf("%s/%s", field, view)
err = tx.CreateBitmap(name)
if err != nil {
return err
}
_, err = tx.AddRoaring(name, rb)
if err != nil {
return err
}
return tx.Commit()
}
func (h *Holder) ConvertToRBF(c Converter) {
/*
for idxname, idx := range h.indexes {
for fieldName, field := range idx.fields {
for _, view := range field.views() {
for shard, fragment := range view.fragments {
panic("NEED bitmap from storage")
junk := roaring.NewBitmap()
err := c.Convert(idxname, fieldName, view.name, shard, junk)
if err != nil {
fmt.Println("ERR", err, fragment.shard) //just added shard for compile
}
}
}
}
}
c.Shutdown()
*/
}