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

initial implementation of RBF backend

Browse source 
This provides us with most of the existing Tx interface, split
across QueryRead and QueryWrite. The functions not included here
are the ones that are used *only* for anti-entropy (ForEach
and ForEachRange).

We add additional testing to verify that TxStores are getting
closed correctly, to go with cleaning up the test directories they're
made in.

We also introduce some test wrappers that can automatically
fail tests on error, so tests don't need to be full of error
checks.

Also, now that I'm starting to think more about the flow of
writing tests using QueryScope, we add the missing "full
database" scope option, and make the Add methods return
their operand so (1) you can chain them, (2) you can use
the AddIndex(...) inline in a NewWriteQueryContext.

Also addressed a plausible performance concern in shardList,
and some comments that were stale or incorrect.

The test coverage here is skimpy on the actual RBF-calling
functions because those are trivial. We do, however, significantly
expand coverage in the random write requests, which are now
a mix of random writes and random reads, and add test cases
that at least hit a lot of the error checks once.

The Error() method is changed to be like (testing.T).Error(),
taking ...interface{} and using fmt.Sprint on them.

There's also some minor tweaks such as making the visualizations
more consistent, testing visualization generation on two kinds
of keysplitter, and so on.
This commit is contained in:
Seebs 2022-10-26 12:23:34 -05:00 • committed by seebs
parent b4eb1b8bdc
commit 7b434cd11c
8 changed files with 1004 additions and 231 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

86
querycontext/dot_test.go
View file

@ -8,61 +8,35 @@ import (
)
func TestDot(t *testing.T) {
// txs, err := newRbfTxStore("foo", &flexibleKeySplitter{splitIndexes: map[string]struct{}{"i": {}}})
txs, err := newRbfTxStore("/data/foo", nil)
if err != nil {
t.Fatalf("creating tx store: %v", err)
}
file, err := os.Create("test.dot")
if err != nil {
t.Fatalf("creating file: %v", err)
}
defer file.Close()
scope := txs.Scope()
scope.AddIndex("i")
q, err := txs.NewWriteQueryContext(context.Background(), scope)
if err != nil {
t.Fatalf("creating query context: %v", err)
}
defer q.Release()
_, err = q.NewWrite("i", "f", "v", 0)
if err != nil {
t.Fatalf("creating write: %v", err)
}
_, err = q.NewWrite("i", "g", "v", 0)
if err != nil {
t.Fatalf("creating write: %v", err)
}
_, err = q.NewWrite("i", "f", "v", 1)
if err != nil {
t.Fatalf("creating write: %v", err)
}
// read, but it's in a writable thing, so still creates rbfQueryWrite
_, err = q.NewRead("i", "f", "v", 2)
if err != nil {
t.Fatalf("creating read: %v", err)
}
_, err = q.NewRead("j", "f", "v", 0)
if err != nil {
t.Fatalf("creating read: %v", err)
}
q2, err := txs.NewQueryContext(context.Background())
if err != nil {
t.Fatalf("creating query context: %v", err)
}
_, err = q2.NewRead("i", "f", "v", 0)
if err != nil {
t.Fatalf("creating write: %v", err)
}
_, err = q2.NewRead("i", "g", "v", 0)
if err != nil {
t.Fatalf("creating write: %v", err)
}
var dg dotGraph
dg.enqueue(txs)
dg.build(5)
err = dg.Write(file)
if err != nil {
t.Fatalf("writing dot: %v", err)
// test this with two splitters, one of which is the default indexShardKeySplitter. the second one will overwrite
// the output from the first one.
for _, splitter := range []KeySplitter{&flexibleKeySplitter{splitIndexes: map[IndexName]struct{}{"i": {}}}, nil} {
txs := testTxStore(t, "foo", splitter)
file, err := os.Create("test.dot")
if err != nil {
t.Fatalf("creating file: %v", err)
}
defer file.Close()
q, _ := txs.NewWriteQueryContext(context.Background(), txs.Scope().AddIndex("i").AddIndexShards("k", 0))
defer q.Release()
q.NewWrite("i", "f", "v", 0)
q.NewWrite("i", "g", "v", 0)
q.NewWrite("i", "f", "v", 1)
// read, but it's in a writable thing, so still creates rbfQueryWrite
q.NewRead("i", "f", "v", 2)
q.NewRead("j", "f", "v", 0)
q2, _ := txs.NewQueryContext(context.Background())
defer q2.Release()
q2.NewRead("i", "f", "v", 0)
q2.NewRead("i", "g", "v", 0)
var dg dotGraph
// we know what we actually have here...
inner := txs.inner
dg.enqueue(inner.(*rbfTxStore))
dg.build(5)
err = dg.Write(file)
if err != nil {
t.Fatalf("writing dot: %v", err)
}
}
}

225
querycontext/query_context.go
View file

@ -4,6 +4,8 @@ package querycontext
import (
"sort"
"strings"
"github.com/molecula/featurebase/v3/roaring"
)
// QueryContext represents the lifespan of a query or similar thing which
@ -32,7 +34,7 @@ type QueryContext interface {
NewWrite(IndexName, FieldName, ViewName, ShardID) (QueryWrite, error)
// Error sets a persistent error state and indicates that this QueryContext
// must not commit its writes.
Error(error)
Error(...interface{})
// Errorf is a convenience function equivalent to Error(fmt.Errorf(...))
Errorf(string, ...interface{})
// Release releases resources held by this QueryContext without committing
@ -48,12 +50,117 @@ type QueryContext interface {
Commit() error
}
// QueryRead represents read access to a fragment.
// QueryRead represents read access to a fragment. When functions in
// this interface return an error, the error indicates a failed operation,
// such as an I/O error. Empty or nonexistent data is not an error.
// For example, the Container method can return a nil pointer if no such
// container exists, but would also return a nil error in that case. An
// error would be returned only if the attempt to determine whether the
// container exists failed for some reason.
type QueryRead interface {
// ContainerIterator yields a container iterator starting at
// the given key. The found bool return indicates whether that
// exact container was present. The iterator's Close() function
// must be called when done using it.
ContainerIterator(ckey uint64) (citer roaring.ContainerIterator, found bool, err error)
// ApplyFilter applies a roaring.BitmapFilter to the fragment, starting
// at the given container key. The container objects passed to the
// filter's ConsiderData method are transient objects; both the
// container header and the data associated with the container can be
// overwritten by the filter after each call. If you need the Container
// objects, or the data they reference, after that method is called,
// you must clone them.
ApplyFilter(ckey uint64, filter roaring.BitmapFilter) (err error)
// Container returns the *roaring.Container for the container key,
// which may be a nil if the container isn't present. The container
// returned is valid for the life of the query context.
Container(ckey uint64) (*roaring.Container, error)
// Contains determines whether the bit is set.
Contains(v uint64) (exists bool, err error)
// Count returns the count of bits set in the fragment.
Count() (uint64, error)
// Max returns the highest bit set in the fragment.
Max() (uint64, error)
// Min returns the lowest bit set in the fragment.
Min() (uint64, bool, error)
// CountRange returns the count of set bits in the range [start, end)
// in this fragment. The lower bound is inclusive, the upper bound is
// exclusive.
CountRange(start, end uint64) (uint64, error)
// OffsetRange returns a bitmap containing the containers covering the
// range (in bits) from start (inclusive) to end (exclusive). Despite
// the range being specified in bits, all three parameters must be multiples
// of 65,536 (the size of a Container).
//
// The bits returned will have their offsets adjusted by (offset-start).
// For instance, if start is 0, and offset is 65536, all bits will be
// 65536 higher (which is to say, all container keys will be one higher
// than they were in the fragment).
//
// OffsetRange is used to translate from a row of a fragment to a shard
// of a database-wide Row. For instance:
//
// OffsetRange(3 * ShardWidth, 4 * ShardWidth, 7 * ShardWidth)
//
// would yield the third "row" of a fragment, with its container keys adjusted
// to reflect the range covered by shard 7 of the index.
//
// The resulting bitmap is valid for the lifespan of the QueryContext.
OffsetRange(offset, start, end uint64) (*roaring.Bitmap, error)
// RoaringBitmap produces a roaring.Bitmap representing the entire fragment.
// The resulting bitmap is valid for the lifespan of the QueryContext.
RoaringBitmap() (*roaring.Bitmap, error)
}
// QueryWrite represents write access to a fragment.
// QueryWrite represents write access to a fragment. As with QueryRead,
// errors indicate an unexpected error. For instance, if you try to
// remove a container that doesn't exist, that's not an "error", but if
// you try to remove a container and get a disk write error or something
// like that, that's an error.
type QueryWrite interface {
QueryRead
// PutContainer stores c under the given key in the fragment.
PutContainer(ckey uint64, c *roaring.Container) error
// RemoveContainer deletes the roaring.Container under the given key
// in the fragment.
RemoveContainer(ckey uint64) error
// Add sets the given bits in the fragment, and reports how many bits
// actually changed.
Add(a ...uint64) (changeCount int, err error)
// Remove clears the given bits in the fragment, and reports how many
// bits actually changed.
Remove(a ...uint64) (changeCount int, err error)
// ApplyRewriter applies a roaring.BitmapRewriter to a specified shard,
// starting at the given container key. The filter's ConsiderData
// method may be called with transient Container objects which *must
// not* be retained or referenced after that function exits. Similarly,
// their data must not be retained. If you need the data later, you
// must copy it into some other memory. However, it is safe to overwrite
// the returned container; for instance, you can DifferenceInPlace on
// it.
ApplyRewriter(ckey uint64, filter roaring.BitmapRewriter) (err error)
// ImportRoaringBits does efficient bulk import using a roaring.RoaringIterator.
//
// See the roaring package for details of the RoaringIterator.
//
// If clear is true, the bits from rit are cleared, otherwise they are set in the
// specifed fragment.
ImportRoaringBits(rit roaring.RoaringIterator, clear bool, rowSize uint64) (changed int, rowSet map[uint64]int, err error)
}
type IndexName string
@ -74,6 +181,12 @@ type ShardID uint64
// at least one of these fragment identifiers, and the TxStore's
// KeySplitter would produce the same database key for those fragment
// identifiers.
//
// The Add functions return the scope to allow things like
//
// txs.NewWriteQueryContext(ctx, txs.Scope().AddIndex("i"))
//
// and chaining add operations in simple cases.
type QueryScope interface {
// Allowed determines whether a specific fragment
// is covered by this QueryScope.
@ -85,10 +198,11 @@ type QueryScope interface {
// both objects.
Overlap(QueryScope) bool
AddIndex(IndexName)
AddField(IndexName, FieldName)
AddIndexShards(IndexName, ...ShardID)
AddFieldShards(IndexName, FieldName, ...ShardID)
AddAll() QueryScope
AddIndex(IndexName) QueryScope
AddField(IndexName, FieldName) QueryScope
AddIndexShards(IndexName, ...ShardID) QueryScope
AddFieldShards(IndexName, FieldName, ...ShardID) QueryScope
String() string
}
@ -99,6 +213,7 @@ type QueryScope interface {
// an absent key indicates no shards. Shard lists are stored sorted.
type indexShardQueryScope struct {
shards map[IndexName]shardList
all bool
}
var _ QueryScope = &indexShardQueryScope{}
@ -107,9 +222,9 @@ func (i *indexShardQueryScope) String() string {
var scopes []string
for index, shards := range i.shards {
if shards.all {
scopes = append(scopes, string(index+"*"))
scopes = append(scopes, string(index))
} else {
scopes = append(scopes, string(index+"+"))
scopes = append(scopes, string(index+"#"))
}
}
// ensure consistent order for reader benefit
@ -117,17 +232,27 @@ func (i *indexShardQueryScope) String() string {
return strings.Join(scopes, ",")
}
// AddAll adds the whole database
func (i *indexShardQueryScope) AddAll() QueryScope {
i.all = true
return i
}
// AddIndex adds the given index, with all shards writable.
func (i *indexShardQueryScope) AddIndex(index IndexName) {
func (i *indexShardQueryScope) AddIndex(index IndexName) QueryScope {
if i.shards == nil {
i.shards = map[IndexName]shardList{index: {all: true}}
return
return i
}
i.shards[index] = shardList{all: true}
return i
}
// AddIndexShards adds the given index for the given shards.
func (i *indexShardQueryScope) AddIndexShards(index IndexName, shards ...ShardID) {
func (i *indexShardQueryScope) AddIndexShards(index IndexName, shards ...ShardID) QueryScope {
if i.all {
return i
}
if i.shards == nil {
i.shards = map[IndexName]shardList{}
}
@ -136,20 +261,21 @@ func (i *indexShardQueryScope) AddIndexShards(index IndexName, shards ...ShardID
// if not, just use a new {any: shards} shardlist, but we want to verify
// shard lists are sorted.
if existing.all {
return
return i
}
for _, shard := range shards {
existing.Add(shard)
}
i.shards[index] = existing
return i
}
func (i *indexShardQueryScope) AddField(index IndexName, _ FieldName) {
i.AddIndex(index)
func (i *indexShardQueryScope) AddField(index IndexName, _ FieldName) QueryScope {
return i.AddIndex(index)
}
func (i *indexShardQueryScope) AddFieldShards(index IndexName, field FieldName, shards ...ShardID) {
i.AddIndexShards(index, shards...)
func (i *indexShardQueryScope) AddFieldShards(index IndexName, field FieldName, shards ...ShardID) QueryScope {
return i.AddIndexShards(index, shards...)
}
func (i *indexShardQueryScope) Allowed(index IndexName, _ FieldName, _ ViewName, shard ShardID) bool {
@ -292,6 +418,7 @@ func (scope *indexScope) Complexity() string {
//
// If no flexibleKeySplitter is provided, every index is split.
type flexibleQueryScope struct {
all bool
splitter *flexibleKeySplitter
indexes map[IndexName]*indexScope
}
@ -299,6 +426,9 @@ type flexibleQueryScope struct {
var _ QueryScope = &flexibleQueryScope{}
func (i *flexibleQueryScope) String() string {
if i.all {
return "*"
}
descrs := make([]string, 0, len(i.indexes))
for index, scope := range i.indexes {
// Show the index's name plus something indicating the
@ -310,17 +440,30 @@ func (i *flexibleQueryScope) String() string {
return strings.Join(descrs, ",")
}
// AddAll does what it sounds like.
func (i *flexibleQueryScope) AddAll() QueryScope {
i.all = true
return i
}
// AddIndex adds the given index, with all shards writable.
func (i *flexibleQueryScope) AddIndex(index IndexName) {
func (i *flexibleQueryScope) AddIndex(index IndexName) QueryScope {
if i.all {
return i
}
if i.indexes == nil {
i.indexes = map[IndexName]*indexScope{index: {all: shardList{all: true}}}
return
return i
}
i.indexes[index] = &indexScope{all: shardList{all: true}}
return i
}
// AddIndexShards adds the given index for the given shards.
func (i *flexibleQueryScope) AddIndexShards(index IndexName, shards ...ShardID) {
func (i *flexibleQueryScope) AddIndexShards(index IndexName, shards ...ShardID) QueryScope {
if i.all {
return i
}
if i.indexes == nil {
i.indexes = map[IndexName]*indexScope{}
}
@ -333,21 +476,24 @@ func (i *flexibleQueryScope) AddIndexShards(index IndexName, shards ...ShardID)
i.indexes[index] = scope
}
if scope.all.all {
return
return i
}
for _, shard := range shards {
scope.all.Add(shard)
}
return i
}
// AddField adds the given field, with all shards writable. If the field
// is in an unsplit index, the entire index is covered.
func (i *flexibleQueryScope) AddField(index IndexName, field FieldName) {
func (i *flexibleQueryScope) AddField(index IndexName, field FieldName) QueryScope {
if i.all {
return i
}
if i.splitter != nil {
if _, ok := i.splitter.splitIndexes[index]; !ok {
// ignore field because this index isn't split
i.AddIndex(index)
return
return i.AddIndex(index)
}
}
if i.indexes == nil {
@ -359,15 +505,18 @@ func (i *flexibleQueryScope) AddField(index IndexName, field FieldName) {
i.indexes[index] = scope
}
scope.AddField(field)
return i
}
// AddFieldShards adds the given index for the given shards.
func (i *flexibleQueryScope) AddFieldShards(index IndexName, field FieldName, shards ...ShardID) {
func (i *flexibleQueryScope) AddFieldShards(index IndexName, field FieldName, shards ...ShardID) QueryScope {
if i.all {
return i
}
if i.splitter != nil {
if _, ok := i.splitter.splitIndexes[index]; !ok {
// ignore field because this index isn't split
i.AddIndexShards(index, shards...)
return
return i.AddIndexShards(index, shards...)
}
}
if i.indexes == nil {
@ -379,9 +528,13 @@ func (i *flexibleQueryScope) AddFieldShards(index IndexName, field FieldName, sh
i.indexes[index] = scope
}
scope.AddFieldShards(field, shards...)
return i
}
func (i *flexibleQueryScope) Allowed(index IndexName, field FieldName, _ ViewName, shard ShardID) bool {
if i.all {
return true
}
if i.splitter != nil {
// unsplit index: we can't have stored fields so we don't check them
if _, ok := i.splitter.splitIndexes[index]; !ok {
@ -413,6 +566,9 @@ func (i *flexibleQueryScope) Overlap(qw QueryScope) (out bool) {
// overlap occurs if there's an overlap of indexes, or of fields.
// We compare indexes against the other side's corresponding fields,
// and fields against the other side's corresponding indexes.
if (i.all && len(other.indexes) > 0) || (other.all && len(i.indexes) > 0) {
return true
}
for index, scope := range i.indexes {
// if the other has this index as an unsplit index, overlap
// there counts
@ -495,12 +651,23 @@ func (s *shardList) Overlap(other shardList) bool {
return false
}
// Add returns a list because if you call it on a nil shardList,
// it needs to return a new one.
// Add adds the given shard to the shardlist, maintaining
// sorted order.
func (s *shardList) Add(shard ShardID) {
if s.all {
return
}
// short circuit for empty lists or the case where the
// new item is the largest, so sorted lists are O(n)
// instead of O(n log n).
if len(s.any) == 0 {
s.any = []ShardID{shard}
return
}
if s.any[len(s.any)-1] < shard {
s.any = append(s.any, shard)
return
}
pos := s.findShard(shard)
if pos >= 0 {
return

362
querycontext/query_context_test.go
View file

@ -4,14 +4,84 @@ package querycontext
import (
"context"
"errors"
"fmt"
"math/rand"
"path/filepath"
"runtime"
"testing"
"time"
rbfcfg "github.com/molecula/featurebase/v3/rbf/cfg"
"golang.org/x/sync/errgroup"
)
var rbfTestConfig = func() *rbfcfg.Config {
cfg := rbfcfg.NewDefaultConfig()
cfg.FsyncEnabled = false
cfg.MaxSize = (1 << 28)
cfg.MaxWALSize = (1 << 28)
return cfg
}()
// testTxStore creates a testTxStoreWrapper with the provided path (relative to a TempDir)
// and KeySplitter, or fails the test. It also registers a cleanup function
// which closes the TxStore, and fails the test if that close doesn't succeed,
// for instance if the test leaves a QueryContext open. Neat, huh!
func testTxStore(tb testing.TB, path string, ks KeySplitter) *testTxStoreWrapper {
dir := filepath.Join(tb.TempDir(), path)
txs, err := NewRBFTxStore(dir, rbfTestConfig, ks)
if err != nil {
tb.Fatalf("opening TxStore: %v", err)
}
tb.Cleanup(func() {
err := txs.Close()
if err != nil {
tb.Errorf("closing TxStore: %v", err)
}
})
return newTestTxStoreWrapper(tb, txs)
}
func TestTxStoreClose(t *testing.T) {
dir := filepath.Join(t.TempDir(), "foo")
ctx := context.Background()
txs, err := NewRBFTxStore(dir, rbfTestConfig, nil)
if err != nil {
t.Fatalf("opening TxStore: %v", err)
}
qcx, err := txs.NewQueryContext(ctx)
if err != nil {
t.Fatalf("creating initial query context: %v", err)
}
_, err = qcx.NewWrite("i", "f", "v", 0)
if err == nil {
t.Fatalf("should get error requesting a write from a read qcx")
}
err = txs.Close() // should fail because of the qcx...
if err == nil {
t.Fatalf("should have failed to close TxStore because of open query")
}
qcx.Release()
err = txs.Close()
if err != nil {
t.Fatalf("should have closed TxStore successfully")
}
qcx, err = txs.NewQueryContext(ctx)
if err == nil {
qcx.Release()
t.Fatalf("should have failed to open query context on closed TxStore")
}
qcx, err = txs.NewWriteQueryContext(ctx, txs.Scope())
if err == nil {
qcx.Release()
t.Fatalf("should have failed to open query context on closed TxStore")
}
err = txs.Close()
if err == nil {
t.Fatalf("should get error on double-close")
}
}
func TestShardList(t *testing.T) {
rng := rand.New(rand.NewSource(3))
prev := &shardList{}
@ -62,98 +132,121 @@ func TestShardList(t *testing.T) {
func TestTxStore(t *testing.T) {
ctx := context.Background()
txs, err := newRbfTxStore("foo", nil)
if err != nil {
t.Fatalf("creating Tx store: %v", err)
}
q, err := txs.NewQueryContext(ctx)
if err != nil {
t.Fatalf("creating read context: %v", err)
}
err = txs.Close()
if err == nil {
t.Fatalf("shouldn't close a database with unused QueryContext live")
}
_, err = q.NewRead("i", "f", "v", 0)
if err != nil {
t.Fatalf("creating read query: %v", err)
}
err = txs.Close()
txs := testTxStore(t, "foo", nil)
q, _ := txs.NewQueryContext(ctx)
_, _ = q.NewRead("i", "f", "v", 0)
txs.expect(1)
err := txs.Close()
if err == nil {
t.Fatalf("shouldn't close a database while still open")
}
q.Release()
err = txs.Close()
if err != nil {
t.Fatalf("closing database: %v", err)
q.expect(1)
err = q.Commit()
if err == nil {
t.Fatalf("shouldn't be able to commit after releasing")
}
q, _ = txs.NewWriteQueryContext(ctx, txs.Scope())
q.Error("oops")
q.expect(1)
err = q.Commit()
if err == nil {
t.Fatalf("shouldn't have been able to commit with error)")
}
nctx, cancel := context.WithCancel(ctx)
q, _ = txs.NewWriteQueryContext(nctx, txs.Scope())
cancel()
q.expect(1)
err = q.Commit()
if err == nil {
t.Fatalf("shouldn't have been able to commit after context cancelled")
}
}
// writeReq represents a possible write request.
// writeReq represents a possible write request. some of them will
// actually just read, and don't need to be within write scope.
type writeReq struct {
index IndexName
field FieldName
shard ShardID
index IndexName
field FieldName
shard ShardID
justRead bool
}
// satisfyWriteRequest tries to process the writes in a list of
// writeReqs, by requesting them in order, and returns an error
// if it encounters any errors.
func satisfyWriteRequest(t *testing.T, txs TxStore, writeList []writeReq) error {
if len(writeList) == 0 {
func satisfyWriteRequest(t *testing.T, txs *testTxStoreWrapper, requests []writeReq) error {
if len(requests) == 0 {
return nil
}
writes := txs.Scope()
prevIndex := writeList[0].index
prevField := writeList[0].field
shards := []ShardID{writeList[0].shard}
scope := txs.Scope()
prevIndex := requests[0].index
prevField := requests[0].field
shards := []ShardID{requests[0].shard}
add := func(index IndexName, field FieldName, shards ...ShardID) {
if field == "" {
if len(shards) == 1 && shards[0] == 0 {
writes.AddIndex(index)
scope.AddIndex(index)
} else {
writes.AddIndexShards(index, shards...)
scope.AddIndexShards(index, shards...)
}
} else {
if len(shards) == 1 && shards[0] == 0 {
writes.AddField(index, field)
scope.AddField(index, field)
} else {
writes.AddFieldShards(index, field, shards...)
scope.AddFieldShards(index, field, shards...)
}
}
}
// batch shards together. this is mostly irrelevant, but if we happened to get
// shard lists in an unsorted order, and didn't handle that correctly, this would
// catch that.
for _, write := range writeList[1:] {
if write.index == prevIndex && write.field == prevField {
shards = append(shards, write.shard)
for _, req := range requests[1:] {
// don't add scope for a request that isn't a write
if req.justRead {
continue
}
if req.index == prevIndex && req.field == prevField {
shards = append(shards, req.shard)
} else {
add(prevIndex, prevField, shards...)
shards = append(shards[:0], write.shard)
prevIndex = write.index
prevField = write.field
shards = append(shards[:0], req.shard)
prevIndex = req.index
prevField = req.field
}
}
add(prevIndex, prevField, shards...)
qcx, err := txs.NewWriteQueryContext(context.Background(), writes)
defer qcx.Release()
qcx, err := txs.NewWriteQueryContext(context.Background(), scope)
if err != nil {
return err
}
for _, i := range rand.Perm(len(writeList)) {
write := writeList[i]
_, err := qcx.NewWrite(write.index, write.field, "v", write.shard)
if err != nil {
return err
}
// try reading something that could be outside our scope, because that
// should still be allowed
_, err = qcx.NewRead(write.index, write.field, "v", write.shard+5)
if err != nil {
return err
defer qcx.Release()
for _, i := range rand.Perm(len(requests)) {
req := requests[i]
if req.justRead {
qr, err := qcx.NewRead(req.index, req.field, "v", req.shard)
if err != nil {
return err
}
_, _ = qr.Contains(1)
if !scope.Allowed(req.index, req.field, "v", req.shard) {
qcx.expect(1)
_, err := qcx.NewWrite(req.index, req.field, "v", req.shard)
if err == nil {
qcx.Error("write was allowed when it should have been out of scope")
}
}
} else {
qw, err := qcx.NewWrite(req.index, req.field, "v", req.shard)
if err != nil {
return err
}
_, _ = qw.Add(1)
}
}
if t.Failed() {
qcx.Error("an error occurred during read or write ops")
}
// When testing this, we're spawning a number of satisfyWriteRequest tasks
// at once. We want to be checking the overlap case more than we care about
// the no-overlap case, but we don't want to actually spend *time* sleeping.
@ -166,7 +259,7 @@ func satisfyWriteRequest(t *testing.T, txs TxStore, writeList []writeReq) error
// delayWriteRequest is like satisfyWriteRequest, but doesn't commit
// the write request until the provided channel is closed, allowing us
// to verify that new requests can still happen before this one is closed.
func delayWriteRequest(t *testing.T, txs TxStore, writeList []writeReq, ch chan struct{}) (func() error, error) {
func delayWriteRequest(t *testing.T, txs *testTxStoreWrapper, writeList []writeReq, ch chan struct{}) (func() error, error) {
writes := txs.Scope()
for _, write := range writeList {
writes.AddFieldShards(write.index, write.field, write.shard)
@ -177,11 +270,12 @@ func delayWriteRequest(t *testing.T, txs TxStore, writeList []writeReq, ch chan
}
for _, i := range rand.Perm(len(writeList)) {
write := writeList[i]
_, err := qcx.NewWrite(write.index, write.field, "v", write.shard)
qw, err := qcx.NewWrite(write.index, write.field, "v", write.shard)
if err != nil {
qcx.Release()
return nil, err
}
qw.Add(1)
}
return func() error {
<-ch
@ -198,10 +292,7 @@ func testSomeWriteRequests(t *testing.T, writeRequests [][]writeReq) {
} else {
splitter = &indexShardKeySplitter{}
}
txs, err := newRbfTxStore("foo", splitter)
if err != nil {
t.Fatalf("creating Tx store: %v", err)
}
txs := testTxStore(t, "foo", splitter)
eg := errgroup.Group{}
for i := range writeRequests {
req := writeRequests[i]
@ -209,7 +300,7 @@ func testSomeWriteRequests(t *testing.T, writeRequests [][]writeReq) {
return satisfyWriteRequest(t, txs, req)
})
}
err = eg.Wait()
err := eg.Wait()
if err != nil {
t.Logf("write requests (%d):", len(writeRequests))
for _, req := range writeRequests {
@ -217,10 +308,6 @@ func testSomeWriteRequests(t *testing.T, writeRequests [][]writeReq) {
}
t.Fatalf("running reqs: %v", err)
}
err = txs.Close()
if err != nil {
t.Fatalf("closing tx store: %v", err)
}
}
// overlappingWriteReqs represents a test case for allowing write requests to
@ -237,26 +324,72 @@ type overlappingWriteReqs struct {
//
// the timeout is returned as an error, other failures cause the test to fail.
func testOverlappingWriteRequests(t *testing.T, write overlappingWriteReqs) error {
txs, err := newRbfTxStore("foo", write.splitter)
if err != nil {
t.Fatalf("creating Tx store: %v", err)
}
txs := testTxStore(t, "foo", write.splitter)
// we spawn goroutines to open each of these, but wait until doneCh
// closes before closing any of them, so they're all open at once.
doneCh := make(chan struct{})
eg := errgroup.Group{}
closeFuncs := make([]func() error, len(write.reqs))
closeFuncs := make(chan func() error, len(write.reqs))
for i := range write.reqs {
i := i
req := write.reqs[i]
eg.Go(func() (err error) {
closeFuncs[i], err = delayWriteRequest(t, txs, req, doneCh)
fn, err := delayWriteRequest(t, txs, req, doneCh)
if err == nil {
closeFuncs <- fn
}
return err
})
}
errCh := make(chan error)
go func() {
errCh <- eg.Wait()
close(errCh)
}()
defer func() {
close(doneCh)
// we have an obligation to close every QueryContext before we close
// the TxStore. but we couldn't open them all! So. We check whether
// we actually got a response from errCh. If that times out, some of
// the delayWriteRequest calls aren't done yet, which means they haven't
// updated their entry in closeFuncs. So, we close all the existing
// QueryContexts, and try again. At least something ought to succeed,
// but it's possible not everything will, so we keep trying until we
// actually got them all done, call the remaining closeFuncs, and
// exit.
pending := true
expected := len(write.reqs)
for pending {
// wait up to 50ms for errCh to provide a result; if it doesn't,
// we need to try again
select {
case <-errCh:
pending = false
case <-time.After(50 * time.Millisecond):
pending = true
}
// run any pending close funcs. If errCh is closed, we should
// have all of our closeFuncs. If it isn't, we might only have
// some, because others haven't been created yet. So we grab
// everything currently available.
gotClose := true
for gotClose {
select {
case fn := <-closeFuncs:
if fn != nil {
if err := fn(); err != nil {
t.Errorf("closing transaction: %v", err)
}
}
expected--
if expected == 0 {
pending = false
}
default:
gotClose = false
}
}
}
}()
select {
case err := <-errCh:
@ -268,27 +401,22 @@ func testOverlappingWriteRequests(t *testing.T, write overlappingWriteReqs) erro
t.Fatalf("running reqs: %v", err)
}
case <-time.After(50 * time.Millisecond):
t.Logf("write requests (%d):", len(write.reqs))
for _, req := range write.reqs {
t.Logf("> %v", req)
if !write.shouldError {
// Only log this if we didn't expect an error.
t.Logf("write requests (%d):", len(write.reqs))
for _, req := range write.reqs {
t.Logf("> %v", req)
}
}
return errors.New("timed out trying to create write requests")
}
// all the writes should still be pending
err = txs.Close()
txs.expect(1)
err := txs.Close()
if err == nil {
t.Fatalf("close completed while transactions were pending")
}
close(doneCh)
for _, fn := range closeFuncs {
if err := fn(); err != nil {
t.Fatalf("closing transaction: %v", err)
}
}
err = txs.Close()
if err != nil {
t.Fatalf("closing tx store: %v", err)
}
return nil
}
@ -299,47 +427,65 @@ func TestOverlappingWriteRequests(t *testing.T) {
{
splitter: nil,
reqs: [][]writeReq{
{{"a", "f", 0}},
{{"b", "f", 0}},
{{"a", "f", 1}},
{{"a", "f", 0, false}},
{{"b", "f", 0, false}},
{{"a", "f", 1, false}},
},
},
{
// we should be able to do two fields in i, but we wouldn't in j.
splitter: NewFlexibleKeySplitter("i"),
reqs: [][]writeReq{
{{"i", "f", 0}},
{{"i", "g", 0}},
{{"j", "f", 0}},
{{"i", "f", 0, false}},
{{"i", "g", 0, false}},
{{"j", "f", 0, false}},
},
},
{
// this should fail
splitter: NewFlexibleKeySplitter("i"),
reqs: [][]writeReq{
{{"i", "f", 0}},
{{"i", "g", 0}},
{{"j", "f", 0}},
{{"j", "g", 0}},
{{"i", "f", 0, false}},
{{"i", "g", 0, false}},
{{"j", "f", 0, false}},
{{"j", "g", 0, false}},
},
shouldError: true,
},
{
// this should fail more than once
splitter: NewFlexibleKeySplitter("i"),
reqs: [][]writeReq{
{{"i", "f", 0, false}},
{{"i", "g", 0, false}},
{{"j", "f", 0, false}},
{{"j", "g", 0, false}},
{{"i", "f", 0, false}},
{{"i", "g", 0, false}},
{{"i", "f", 0, false}},
},
shouldError: true,
},
}
for i, testCase := range overlapping {
err := testOverlappingWriteRequests(t, testCase)
if testCase.shouldError {
if err == nil {
t.Fatalf("expected error for case %d, but didn't get it", i)
// run these in separate test cases, because testTxStore is closing in
// t.Cleanup, and we want them wrapped up as we go.
t.Run(fmt.Sprintf("case-%d", i), func(t *testing.T) {
err := testOverlappingWriteRequests(t, testCase)
if testCase.shouldError {
if err == nil {
t.Fatalf("expected error for case %d, but didn't get it", i)
}
} else {
if (err != nil) != testCase.shouldError {
t.Fatalf("case %d: unexpected error %v", i, err)
}
}
} else {
if (err != nil) != testCase.shouldError {
t.Fatalf("case %d: unexpected error %v", i, err)
}
}
})
}
}
func writeRequestsFromBytes(data []byte) [][]writeReq {
func buildRequestsFromBytes(data []byte) [][]writeReq {
// To fuzz these, we'll use []byte. each pair of bytes is an index/shard pair, and
// we break the list into batches of up to 50 such that the sum of their values caps
// at 256.
@ -358,7 +504,11 @@ func writeRequestsFromBytes(data []byte) [][]writeReq {
}
shard := ShardID(data[1] % 8)
total += int(shard)
sub = append(sub, writeReq{index, field, shard})
// some requests are just reads. reads won't be added to our scope, so if there's
// no corresponding write, they'll be outside of scope, so we're verifying that we
// can read outside of scope.
justRead := ((data[1] >> 4) % 4) != 0
sub = append(sub, writeReq{index, field, shard, justRead})
if total > 10 {
out = append(out, sub)
sub = []writeReq{}
@ -384,7 +534,7 @@ func FuzzWriteRequests(f *testing.F) {
f.Add(seed)
}
f.Fuzz(func(t *testing.T, data []byte) {
writeRequests := writeRequestsFromBytes(data)
writeRequests := buildRequestsFromBytes(data)
testSomeWriteRequests(t, writeRequests)
})
}

142
querycontext/rbf.go
View file

@ -11,15 +11,18 @@ import (
"sync"
"github.com/molecula/featurebase/v3/rbf"
rbfcfg "github.com/molecula/featurebase/v3/rbf/cfg"
"github.com/molecula/featurebase/v3/roaring"
)
// rbfDBQueryContexts represents an actual backend DB, and a map of the
// rbfQueryContexts associated with that DB. this lets us check for
// outstanding transactions before closing a database.
type rbfDBQueryContexts struct {
key dbKey
db *rbf.DB
mu sync.Mutex
key dbKey
dbPath string // dbPath relative to root
db *rbf.DB
mu sync.Mutex
// mutex used to pretend we're using the database even though we're
// not wired up to it yet
lockCheck sync.Mutex
@ -38,8 +41,10 @@ func (r *rbfDBQueryContexts) writeTx(rq *rbfQueryContext) (*rbfTxWrappers, error
if !ok {
return nil, errors.New("locking error: tried for write Tx when it was already held")
}
// still nil for now
var tx *rbf.Tx
tx, err := r.db.Begin(true)
if err != nil {
return nil, err
}
q := &rbfTxWrappers{db: r, tx: tx, key: r.key, queries: make(map[fragKey]QueryRead), writeTx: true}
r.queryContexts[q] = rq
return q, nil
@ -49,7 +54,10 @@ func (r *rbfDBQueryContexts) writeTx(rq *rbfQueryContext) (*rbfTxWrappers, error
func (r *rbfDBQueryContexts) readTx(rq *rbfQueryContext) (*rbfTxWrappers, error) {
r.mu.Lock()
defer r.mu.Unlock()
var tx *rbf.Tx
tx, err := r.db.Begin(false)
if err != nil {
return nil, err
}
q := &rbfTxWrappers{db: r, tx: tx, key: r.key, queries: make(map[fragKey]QueryRead)}
r.queryContexts[q] = rq
return q, nil
@ -101,11 +109,21 @@ type rbfTxStore struct {
dbs map[dbKey]*rbfDBQueryContexts
writeScopes map[*rbfQueryContext]QueryScope
queries map[*rbfQueryContext]struct{}
cfg *rbfcfg.Config
closed bool
}
// newRbfTxStore creates a new RBF-backed TxStore in the given directory.
func newRbfTxStore(path string, splitter KeySplitter) (*rbfTxStore, error) {
// NewRBFTxStore creates a new RBF-backed TxStore in the given directory. If
// cfg is nil, it will use a `NewDefaultConfig`. All databases will be opened
// using the same config. If splitter is nil, it uses an index/shard splitter.
//
// With the index/shard key splitter, database directory paths look like
// `path/indexes/i/shards/00000000`, with each shard directory containing
// data/wal files.
func NewRBFTxStore(path string, cfg *rbfcfg.Config, splitter KeySplitter) (*rbfTxStore, error) {
if cfg == nil {
cfg = rbfcfg.NewDefaultConfig()
}
if splitter == nil {
splitter = &indexShardKeySplitter{}
}
@ -115,17 +133,12 @@ func newRbfTxStore(path string, splitter KeySplitter) (*rbfTxStore, error) {
dbs: make(map[dbKey]*rbfDBQueryContexts),
writeScopes: make(map[*rbfQueryContext]QueryScope),
queries: make(map[*rbfQueryContext]struct{}),
cfg: cfg,
}
r.writeQueue = sync.NewCond(&r.mu)
return r, nil
}
// dbPath yields the filesystem path to be used with a given dbKey to
// open the underlying database.
func (r *rbfTxStore) dbPath(d dbKey) string {
return filepath.Join(r.rootPath, d.Path())
}
// getDB gets or creates the db for the given key. call only when you hold
// the rbfTxStore's lock.
func (r *rbfTxStore) getDB(dbk dbKey) (*rbfDBQueryContexts, error) {
@ -133,10 +146,21 @@ func (r *rbfTxStore) getDB(dbk dbKey) (*rbfDBQueryContexts, error) {
if ok {
return db, nil
}
// if we were opening a database here, it could fail, which is why we return an error here
db = &rbfDBQueryContexts{key: dbk, queryContexts: make(map[*rbfTxWrappers]*rbfQueryContext)}
// use this function so staticcheck is happy
_ = r.dbPath(dbk)
dbPath, err := r.dbPath(dbk)
if err != nil {
return nil, err
}
db = &rbfDBQueryContexts{key: dbk, dbPath: dbPath, queryContexts: make(map[*rbfTxWrappers]*rbfQueryContext)}
path := filepath.Join(r.rootPath, db.dbPath)
// note: we don't need to create the directory here, because rbf.Open
// creates the directory for us. I was slightly surprised by this and
// I'm not sure I like it. Note also that the database path is actually
// a directory containing files named "data" and "wal".
db.db = rbf.NewDB(path, r.cfg)
err = db.db.Open()
if err != nil {
return nil, err
}
r.dbs[dbk] = db
return db, nil
}
@ -246,6 +270,9 @@ func (r *rbfTxStore) NewWriteQueryContext(ctx context.Context, scope QueryScope)
func (r *rbfTxStore) Close() error {
r.mu.Lock()
defer r.mu.Unlock()
if r.closed {
return errors.New("double-close of TxStore")
}
var firstErr error
// A QueryContext can be live without having any transactions open yet.
if len(r.queries) > 0 {
@ -256,7 +283,7 @@ func (r *rbfTxStore) Close() error {
r.closed = true
for key, db := range r.dbs {
if len(db.queryContexts) > 0 {
firstErr = fmt.Errorf("%d transaction(s) still open", len(db.queryContexts))
firstErr = fmt.Errorf("db %q: %d transaction(s) still open", db.dbPath, len(db.queryContexts))
continue
}
if db.db != nil {
@ -273,9 +300,14 @@ func (r *rbfTxStore) Close() error {
// rbfTxWrappers is the per-dbKey part of a QueryContext, representing the set of
// fragment-specific query reads (or writes) associated with a given rbf.Tx. The objects
// are stored here in a map of QueryRead, but they may actually be QueryScope. (We
// are stored here in a map of QueryRead, but they may actually be QueryWrite. (We
// have to grab the write transaction initially, because we can only grab a transaction
// once for each dbKey, and any Allowed fragment could later request a write.)
//
// Underlying Tx are internally locked with an RWMutex on the RBF side, so we don't
// do locking on our side. If multiple QueryRead/QueryWrite are simultaneously
// operating, that's fine, they'll still be serialized at that point. In theory,
// though, that's probably a logic error. Possibly we should check for it.
type rbfTxWrappers struct {
key dbKey
db *rbfDBQueryContexts
@ -385,10 +417,10 @@ func (rq *rbfQueryContext) NewWrite(index IndexName, field FieldName, view ViewN
return queries.writeKey(fk)
}
func (rq *rbfQueryContext) Error(err error) {
func (rq *rbfQueryContext) Error(args ...interface{}) {
rq.mu.Lock()
defer rq.mu.Unlock()
rq.err = err
rq.err = errors.New(fmt.Sprint(args...))
}
func (rq *rbfQueryContext) Errorf(msg string, args ...interface{}) {
@ -464,6 +496,46 @@ type rbfQueryRead struct {
tx *rbfTxWrappers
}
func (qr *rbfQueryRead) ContainerIterator(ckey uint64) (citer roaring.ContainerIterator, found bool, err error) {
return qr.tx.tx.ContainerIterator(string(qr.fk), ckey)
}
func (qr *rbfQueryRead) ApplyFilter(ckey uint64, filter roaring.BitmapFilter) (err error) {
return qr.tx.tx.ApplyFilter(string(qr.fk), ckey, filter)
}
func (qr *rbfQueryRead) Container(ckey uint64) (*roaring.Container, error) {
return qr.tx.tx.Container(string(qr.fk), ckey)
}
func (qr *rbfQueryRead) Contains(v uint64) (exists bool, err error) {
return qr.tx.tx.Contains(string(qr.fk), v)
}
func (qr *rbfQueryRead) Count() (uint64, error) {
return qr.tx.tx.Count(string(qr.fk))
}
func (qr *rbfQueryRead) Max() (uint64, error) {
return qr.tx.tx.Max(string(qr.fk))
}
func (qr *rbfQueryRead) Min() (uint64, bool, error) {
return qr.tx.tx.Min(string(qr.fk))
}
func (qr *rbfQueryRead) CountRange(start, end uint64) (uint64, error) {
return qr.tx.tx.CountRange(string(qr.fk), start, end)
}
func (qr *rbfQueryRead) RoaringBitmap() (*roaring.Bitmap, error) {
return qr.tx.tx.RoaringBitmap(string(qr.fk))
}
func (qr *rbfQueryRead) OffsetRange(offset, start, end uint64) (*roaring.Bitmap, error) {
return qr.tx.tx.OffsetRange(string(qr.fk), offset, start, end)
}
var _ QueryRead = &rbfQueryRead{}
// rbfQueryWrite is a fragment-specific write-capable wrapper
@ -474,4 +546,30 @@ type rbfQueryWrite struct {
rbfQueryRead
}
func (qw *rbfQueryWrite) PutContainer(ckey uint64, c *roaring.Container) error {
return qw.tx.tx.PutContainer(string(qw.fk), ckey, c)
}
func (qw *rbfQueryWrite) RemoveContainer(ckey uint64) error {
return qw.tx.tx.RemoveContainer(string(qw.fk), ckey)
}
func (qw *rbfQueryWrite) Add(a ...uint64) (changeCount int, err error) {
return qw.tx.tx.Add(string(qw.fk), a...)
}
func (qw *rbfQueryWrite) Remove(a ...uint64) (changeCount int, err error) {
return qw.tx.tx.Remove(string(qw.fk), a...)
}
func (qw *rbfQueryWrite) ApplyRewriter(ckey uint64, filter roaring.BitmapRewriter) (err error) {
return qw.tx.tx.ApplyRewriter(string(qw.fk), ckey, filter)
}
func (qw *rbfQueryWrite) ImportRoaringBits(rit roaring.RoaringIterator, clear bool, rowSize uint64) (changed int, rowSet map[uint64]int, err error) {
// TODO: when we finish replacing Qcx/Tx with this, drop the unused "log"
// flag. It was only ever used by the roaring backend.
return qw.tx.tx.ImportRoaringBits(string(qw.fk), rit, clear, false, rowSize)
}
var _ QueryWrite = &rbfQueryWrite{}

49
querycontext/rbf_test.go Normal file
View file

@ -0,0 +1,49 @@
package querycontext
import (
"context"
"testing"
"github.com/molecula/featurebase/v3/roaring"
)
func TestRbfWrite(t *testing.T) {
ctx := context.Background()
txs := testTxStore(t, "foo", nil)
q, _ := txs.NewWriteQueryContext(ctx, txs.Scope().AddIndex("i"))
wr, _ := q.NewWrite("i", "f", "v", 0)
_, _ = wr.Add(23, 25)
_, _ = wr.Remove(25)
// putting nil and removing a missing container are no-ops
_ = wr.PutContainer(2, nil)
_ = wr.RemoveContainer(3)
citer, _, _ := wr.ContainerIterator(3)
if citer != nil {
citer.Close()
}
_, _ = wr.Container(7)
_, _ = wr.Count()
_, _ = wr.Max()
_, _, _ = wr.Min()
_, _ = wr.CountRange(0, 65535)
_, _ = wr.RoaringBitmap()
_, _ = wr.OffsetRange(0, 0, 0)
filter := roaring.NewBitmapColumnFilter(23)
_ = wr.ApplyFilter(0, filter)
// we're not testing ApplyRewriter and ImportRoaringBits because
// they have a lot more setup to be testable, and in practice
// they're all one-line functions anyway.
_ = q.Commit()
q, _ = txs.NewQueryContext(ctx)
rd, _ := q.NewRead("i", "f", "v", 0)
ok, _ := rd.Contains(23)
if !ok {
t.Fatalf("no 23")
}
txs.expect(1)
err := txs.Close()
if err == nil {
t.Fatalf("shouldn't close a database while still open")
}
q.Release()
}

288
querycontext/reduce_err_test.go Normal file
View file

@ -0,0 +1,288 @@
package querycontext
import (
"context"
"sync/atomic"
"testing"
"github.com/molecula/featurebase/v3/roaring"
)
type oopsieWrapper struct {
tb testing.TB
expected int64
}
func (o *oopsieWrapper) expect(n int) {
atomic.AddInt64(&o.expected, int64(n))
}
func (o *oopsieWrapper) oopsie(err error) error {
o.tb.Helper()
if err == nil {
return nil
}
remaining := atomic.AddInt64(&o.expected, -1)
if remaining < 0 {
// we use Error here, not Fatal, so we can run in arbitrary goroutines
o.tb.Error(err)
}
return err
}
// testTxStore is a wrapper which fails a test on unexpected
// errors.
type testTxStoreWrapper struct {
inner TxStore
oopsieWrapper
}
func newTestTxStoreWrapper(tb testing.TB, inner TxStore) *testTxStoreWrapper {
return &testTxStoreWrapper{inner: inner, oopsieWrapper: oopsieWrapper{tb: tb}}
}
func (t *testTxStoreWrapper) Close() error {
t.tb.Helper()
return t.oopsie(t.inner.Close())
}
func (t *testTxStoreWrapper) NewQueryContext(ctx context.Context) (*testQueryContextWrapper, error) {
t.tb.Helper()
q, err := t.inner.NewQueryContext(ctx)
return newTestQueryContextWrapper(t.tb, q), t.oopsie(err)
}
func (t *testTxStoreWrapper) NewWriteQueryContext(ctx context.Context, scope QueryScope) (*testQueryContextWrapper, error) {
t.tb.Helper()
q, err := t.inner.NewWriteQueryContext(ctx, scope)
return newTestQueryContextWrapper(t.tb, q), t.oopsie(err)
}
func (t *testTxStoreWrapper) Scope() QueryScope {
t.tb.Helper()
return t.inner.Scope()
}
func (t *testTxStoreWrapper) dbPath(dbk dbKey) (string, error) {
t.tb.Helper()
p, err := t.inner.dbPath(dbk)
return p, t.oopsie(err)
}
func (t *testTxStoreWrapper) keys(index IndexName, field FieldName, view ViewName, shard ShardID) (dbKey, fragKey) {
t.tb.Helper()
return t.inner.keys(index, field, view, shard)
}
type testQueryContextWrapper struct {
inner QueryContext
oopsieWrapper
}
func newTestQueryContextWrapper(tb testing.TB, inner QueryContext) *testQueryContextWrapper {
return &testQueryContextWrapper{inner: inner, oopsieWrapper: oopsieWrapper{tb: tb}}
}
func (t *testQueryContextWrapper) Release() {
t.tb.Helper()
t.inner.Release()
}
func (t *testQueryContextWrapper) Commit() error {
t.tb.Helper()
return t.oopsie(t.inner.Commit())
}
func (t *testQueryContextWrapper) Error(args ...interface{}) {
t.tb.Helper()
t.inner.Error(args...)
}
func (t *testQueryContextWrapper) Errorf(msg string, args ...interface{}) {
t.tb.Helper()
t.inner.Errorf(msg, args...)
}
func (t *testQueryContextWrapper) NewRead(index IndexName, field FieldName, view ViewName, shard ShardID) (*testQueryReadWrapper, error) {
t.tb.Helper()
qr, err := t.inner.NewRead(index, field, view, shard)
return newTestQueryReadWrapper(t.tb, qr), t.oopsie(err)
}
func (t *testQueryContextWrapper) NewWrite(index IndexName, field FieldName, view ViewName, shard ShardID) (*testQueryWriteWrapper, error) {
t.tb.Helper()
qw, err := t.inner.NewWrite(index, field, view, shard)
return newTestQueryWriteWrapper(t.tb, qw), t.oopsie(err)
}
type testQueryReadWrapper struct {
inner QueryRead
oopsieWrapper
}
func newTestQueryReadWrapper(tb testing.TB, inner QueryRead) *testQueryReadWrapper {
return &testQueryReadWrapper{inner: inner, oopsieWrapper: oopsieWrapper{tb: tb}}
}
func (t *testQueryReadWrapper) ContainerIterator(ckey uint64) (citer roaring.ContainerIterator, found bool, err error) {
t.tb.Helper()
citer, found, err = t.inner.ContainerIterator(ckey)
return citer, found, t.oopsie(err)
}
func (t *testQueryReadWrapper) ApplyFilter(ckey uint64, filter roaring.BitmapFilter) (err error) {
t.tb.Helper()
return t.oopsie(t.inner.ApplyFilter(ckey, filter))
}
func (t *testQueryReadWrapper) Container(ckey uint64) (*roaring.Container, error) {
t.tb.Helper()
c, err := t.inner.Container(ckey)
return c, t.oopsie(err)
}
func (t *testQueryReadWrapper) Contains(v uint64) (exists bool, err error) {
t.tb.Helper()
exists, err = t.inner.Contains(v)
return exists, t.oopsie(err)
}
func (t *testQueryReadWrapper) Count() (uint64, error) {
t.tb.Helper()
c, err := t.inner.Count()
return c, t.oopsie(err)
}
func (t *testQueryReadWrapper) Max() (uint64, error) {
t.tb.Helper()
m, err := t.inner.Max()
return m, t.oopsie(err)
}
func (t *testQueryReadWrapper) Min() (uint64, bool, error) {
t.tb.Helper()
m, ok, err := t.inner.Min()
return m, ok, t.oopsie(err)
}
func (t *testQueryReadWrapper) CountRange(start, end uint64) (uint64, error) {
t.tb.Helper()
c, err := t.inner.CountRange(start, end)
return c, t.oopsie(err)
}
func (t *testQueryReadWrapper) OffsetRange(offset, start, end uint64) (*roaring.Bitmap, error) {
t.tb.Helper()
b, err := t.inner.OffsetRange(offset, start, end)
return b, t.oopsie(err)
}
func (t *testQueryReadWrapper) RoaringBitmap() (*roaring.Bitmap, error) {
t.tb.Helper()
b, err := t.inner.RoaringBitmap()
return b, t.oopsie(err)
}
type testQueryWriteWrapper struct {
inner QueryWrite
oopsieWrapper
}
func newTestQueryWriteWrapper(tb testing.TB, inner QueryWrite) *testQueryWriteWrapper {
return &testQueryWriteWrapper{inner: inner, oopsieWrapper: oopsieWrapper{tb: tb}}
}
func (t *testQueryWriteWrapper) PutContainer(ckey uint64, c *roaring.Container) error {
t.tb.Helper()
return t.oopsie(t.inner.PutContainer(ckey, c))
}
func (t *testQueryWriteWrapper) RemoveContainer(ckey uint64) error {
t.tb.Helper()
return t.oopsie(t.inner.RemoveContainer(ckey))
}
func (t *testQueryWriteWrapper) Add(a ...uint64) (changeCount int, err error) {
c, err := t.inner.Add(a...)
t.tb.Helper()
return c, t.oopsie(err)
}
func (t *testQueryWriteWrapper) Remove(a ...uint64) (changeCount int, err error) {
c, err := t.inner.Remove(a...)
t.tb.Helper()
return c, t.oopsie(err)
}
func (t *testQueryWriteWrapper) ApplyRewriter(ckey uint64, filter roaring.BitmapRewriter) (err error) {
t.tb.Helper()
return t.oopsie(t.inner.ApplyRewriter(ckey, filter))
}
func (t *testQueryWriteWrapper) ImportRoaringBits(rit roaring.RoaringIterator, clear bool, rowSize uint64) (changed int, rowSet map[uint64]int, err error) {
c, r, err := t.inner.ImportRoaringBits(rit, clear, rowSize)
t.tb.Helper()
return c, r, t.oopsie(err)
}
// and we duplicate the QueryRead methods, because it's messy to try to embed a QueryRead wrapper
// in the QueryWrite wrapper and keep them sharing a single oopsieWrapper.
func (t *testQueryWriteWrapper) ContainerIterator(ckey uint64) (citer roaring.ContainerIterator, found bool, err error) {
citer, found, err = t.inner.ContainerIterator(ckey)
t.tb.Helper()
return citer, found, t.oopsie(err)
}
func (t *testQueryWriteWrapper) ApplyFilter(ckey uint64, filter roaring.BitmapFilter) (err error) {
t.tb.Helper()
return t.oopsie(t.inner.ApplyFilter(ckey, filter))
}
func (t *testQueryWriteWrapper) Container(ckey uint64) (*roaring.Container, error) {
c, err := t.inner.Container(ckey)
t.tb.Helper()
return c, t.oopsie(err)
}
func (t *testQueryWriteWrapper) Contains(v uint64) (exists bool, err error) {
exists, err = t.inner.Contains(v)
t.tb.Helper()
return exists, t.oopsie(err)
}
func (t *testQueryWriteWrapper) Count() (uint64, error) {
c, err := t.inner.Count()
t.tb.Helper()
return c, t.oopsie(err)
}
func (t *testQueryWriteWrapper) Max() (uint64, error) {
m, err := t.inner.Max()
t.tb.Helper()
return m, t.oopsie(err)
}
func (t *testQueryWriteWrapper) Min() (uint64, bool, error) {
m, ok, err := t.inner.Min()
t.tb.Helper()
return m, ok, t.oopsie(err)
}
func (t *testQueryWriteWrapper) CountRange(start, end uint64) (uint64, error) {
c, err := t.inner.CountRange(start, end)
t.tb.Helper()
return c, t.oopsie(err)
}
func (t *testQueryWriteWrapper) OffsetRange(offset, start, end uint64) (*roaring.Bitmap, error) {
b, err := t.inner.OffsetRange(offset, start, end)
t.tb.Helper()
return b, t.oopsie(err)
}
func (t *testQueryWriteWrapper) RoaringBitmap() (*roaring.Bitmap, error) {
t.tb.Helper()
b, err := t.inner.RoaringBitmap()
return b, t.oopsie(err)
}

64
querycontext/txstore.go
View file

@ -4,6 +4,8 @@ package querycontext
import (
"context"
"fmt"
"path/filepath"
"strings"
)
// TxStore represents a transactional database backend, mapping
@ -36,15 +38,6 @@ var _ TxStore = &rbfTxStore{}
// dbKey is an identifier which can distinguish backend databases.
type dbKey string
// Path yields a relative path corresponding to a dbKey. For example
// this could be something like `indexes/i/shards/0`. The path is
// a relative path, presumably to be combined with some root path for
// a txStore
func (d dbKey) Path() string {
// for now, let's just use the path as the dbkey
return string(d)
}
// fragKey is an identifier which can be used to tell a backend database
// which data to operate on.
type fragKey string
@ -69,6 +62,11 @@ type KeySplitter interface {
// within the database backend.
keys(IndexName, FieldName, ViewName, ShardID) (dbKey, fragKey)
// dbPath yields a filesystem-friendly string that corresponds to dbKey.
// possibly it is identical to dbKey, but you might want a terse dbKey
// like "i/0" and a longer path like "indexes/i/shards/0".
dbPath(dbKey) (string, error)
// Scope() yields a new scope which is aware of this KeySplitter
// and will give correct results for Overlap calls.
Scope() QueryScope
@ -83,7 +81,7 @@ var _ KeySplitter = &flexibleKeySplitter{}
type indexShardKeySplitter struct{}
func (*indexShardKeySplitter) keys(index IndexName, field FieldName, view ViewName, shard ShardID) (dbKey, fragKey) {
d := dbKey(fmt.Sprintf("%s/%012x", index, shard))
d := dbKey(fmt.Sprintf("%s/%08x", index, shard))
t := fragKey(fmt.Sprintf("%s:%s", field, view))
return d, t
}
@ -92,15 +90,44 @@ func (*indexShardKeySplitter) Scope() QueryScope {
return &indexShardQueryScope{}
}
func (*indexShardKeySplitter) dbPath(dbk dbKey) (string, error) {
slash := strings.IndexByte(string(dbk), '/')
if slash == -1 {
return "", fmt.Errorf("malformed dbKey %q", dbk)
}
paths := [4]string{"indexes", "", "shards", ""}
paths[1] = string(dbk)[:slash]
paths[3] = string(dbk)[slash+1:]
return filepath.Join(paths[:]...), nil
}
// fieldShardKeySplitter splits the database by field,shard pairs.
type fieldShardKeySplitter struct{}
func (*fieldShardKeySplitter) keys(index IndexName, field FieldName, view ViewName, shard ShardID) (dbKey, fragKey) {
d := dbKey(fmt.Sprintf("%s/%s/%012x", index, field, shard))
d := dbKey(fmt.Sprintf("%s/%s/%08x", index, field, shard))
t := fragKey(view)
return d, t
}
func (*fieldShardKeySplitter) dbPath(dbk dbKey) (string, error) {
slash := strings.IndexByte(string(dbk), '/')
if slash == -1 {
return "", fmt.Errorf("malformed dbKey %q", dbk)
}
paths := [6]string{"indexes", "", "fields", "", "shards", ""}
paths[1] = string(dbk)[:slash]
paths[3] = string(dbk)[slash+1:]
slash = strings.IndexByte(paths[3], '/')
if slash == -1 {
return "", fmt.Errorf("malformed dbKey %q", dbk)
}
// note order: have to grab the tail of paths[3] before cutting it off
paths[5] = paths[3][slash+1:]
paths[3] = paths[3][:slash]
return filepath.Join(paths[:]...), nil
}
func (*fieldShardKeySplitter) Scope() QueryScope {
// a flexibleQueryScope without a flexibleKeySplitter always
// splits keys by field.
@ -115,6 +142,10 @@ type flexibleKeySplitter struct {
splitIndexes map[IndexName]struct{}
}
// NewFlexibleKeySplitter creates a KeySplitter which uses index/shard
// splits by default, but splits things into fields if they're in the
// indexes provided. This design is experimental, and should be considered
// pre-deprecated for production use for now.
func NewFlexibleKeySplitter(indexes ...IndexName) *flexibleKeySplitter {
splitIndexes := make(map[IndexName]struct{}, len(indexes))
for _, index := range indexes {
@ -130,6 +161,17 @@ func (f *flexibleKeySplitter) keys(index IndexName, field FieldName, view ViewNa
return (&indexShardKeySplitter{}).keys(index, field, view, shard)
}
func (f *flexibleKeySplitter) dbPath(dbk dbKey) (string, error) {
slash := strings.IndexByte(string(dbk), '/')
if slash == -1 {
return "", fmt.Errorf("malformed dbKey %q", dbk)
}
if _, ok := f.splitIndexes[IndexName(dbk)[:slash]]; ok {
return (&fieldShardKeySplitter{}).dbPath(dbk)
}
return (&indexShardKeySplitter{}).dbPath(dbk)
}
func (f *flexibleKeySplitter) Scope() QueryScope {
return &flexibleQueryScope{splitter: f}
}

19
querycontext/visualize.go
View file

@ -3,6 +3,8 @@ package querycontext
import (
"fmt"
"io"
"os"
"strings"
)
// Rendering time!
@ -279,7 +281,13 @@ func (r *rbfTxStore) dotId() string {
}
func (r *rbfTxStore) writeNode(w io.Writer) {
fmt.Fprintf(w, `[label=<rbfTxStore<BR/>[<FONT POINT-SIZE="12" FACE="courier">%s</FONT>]>]`, r.rootPath)
// long root path is annoying to read, so
temp := os.TempDir()
path := r.rootPath
if strings.HasPrefix(r.rootPath, temp+"/") {
path = ".../" + strings.TrimPrefix(r.rootPath, temp+"/")
}
fmt.Fprintf(w, `[label=<rbfTxStore<BR/><FONT POINT-SIZE="10" FACE="courier">%s</FONT>>]`, path)
}
func (r *rbfDBQueryContexts) dotClass() string {
@ -298,13 +306,10 @@ func (r *rbfDBQueryContexts) writeNode(w io.Writer) {
} else {
locked = fmt.Sprintf("<BR/><FONT COLOR=%q>[LOCKED]</FONT>", colorWrite)
}
// We'll want something like this when we get to adding the RBF backend.
if false {
if r.db == nil {
maybeError = colorError
}
if r.db == nil {
maybeError = colorError
}
fmt.Fprintf(w, `[label=<rbfDBQueryContexts<BR/><FONT POINT-SIZE="12" FACE="courier">%s</FONT><BR/>[DB %p]%s> color=%q]`, r.key, r.db, locked, maybeError)
fmt.Fprintf(w, `[label=<rbfDBQueryContexts<BR/><FONT POINT-SIZE="12" FACE="courier">%s</FONT><BR/><FONT POINT-SIZE="12" FACE="courier">%s</FONT>%s> color=%q]`, r.key, r.dbPath, locked, maybeError)
}
func (r *rbfTxWrappers) dotClass() string {