diff --git a/querycontext/dot_test.go b/querycontext/dot_test.go
index ad6da2ab1..8a84349a8 100644
--- a/querycontext/dot_test.go
+++ b/querycontext/dot_test.go
@@ -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)
+ }
}
}
diff --git a/querycontext/query_context.go b/querycontext/query_context.go
index 84a59461f..6331d24e0 100644
--- a/querycontext/query_context.go
+++ b/querycontext/query_context.go
@@ -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
diff --git a/querycontext/query_context_test.go b/querycontext/query_context_test.go
index bc96c07fb..1001b2d1b 100644
--- a/querycontext/query_context_test.go
+++ b/querycontext/query_context_test.go
@@ -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)
})
}
diff --git a/querycontext/rbf.go b/querycontext/rbf.go
index f0c82c106..da44596a2 100644
--- a/querycontext/rbf.go
+++ b/querycontext/rbf.go
@@ -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{}
diff --git a/querycontext/rbf_test.go b/querycontext/rbf_test.go
new file mode 100644
index 000000000..ba39ba4a7
--- /dev/null
+++ b/querycontext/rbf_test.go
@@ -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()
+}
diff --git a/querycontext/reduce_err_test.go b/querycontext/reduce_err_test.go
new file mode 100644
index 000000000..6b71800ac
--- /dev/null
+++ b/querycontext/reduce_err_test.go
@@ -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)
+}
diff --git a/querycontext/txstore.go b/querycontext/txstore.go
index 5083f30bd..114b97cb0 100644
--- a/querycontext/txstore.go
+++ b/querycontext/txstore.go
@@ -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}
}
diff --git a/querycontext/visualize.go b/querycontext/visualize.go
index 19a110607..c2c6e6153 100644
--- a/querycontext/visualize.go
+++ b/querycontext/visualize.go
@@ -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=[%s]>]`, 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=%s>]`, path)
}
func (r *rbfDBQueryContexts) dotClass() string {
@@ -298,13 +306,10 @@ func (r *rbfDBQueryContexts) writeNode(w io.Writer) {
} else {
locked = fmt.Sprintf("
[LOCKED]", 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=%s
[DB %p]%s> color=%q]`, r.key, r.db, locked, maybeError)
+ fmt.Fprintf(w, `[label=%s
%s%s> color=%q]`, r.key, r.dbPath, locked, maybeError)
}
func (r *rbfTxWrappers) dotClass() string {