mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-08-28 10:54:59 +00:00
querycontext updates/changes to support the big changeover
This is a unification of a number of bug fixes, feature additions, and so on. Features include: * Dropping the "New" from NewWrite/NewRead. * IndexName->keys.Index, etc. * Add a new "Flush" operation which is necessary to get the intended behavior of Delete, which allows us to commit/flush changes without letting go of a write lock. * Some additional wrapping and locking in rbfTxWrappers to support that. rbfQueryRead/Write now forward their calls to the parent rbfTxWrappers, so it can lock around the reference to its underlying tx, so the flush operation can replace that tx safely. * AddIndexShards now treats no shards as "all shards", to simplify call sites. * Added parameters to NewRBFTxStore to let it interact with executor's logger and worker pool. * Internally, support explicit closes of parts of the database which can also check for errors and fail if it's in use. * Add ability to request a map of fields and views in use for a given index/shard pair. This is probably deprecated but we need it for the way backup/restore work. * Add ability to request a complete map of the database showing which shards exist for which index/field/view tuples. This is backwards from how we store things on disk, but we need it to allow creating the right in-memory data structures on database open. * Add "Backend()" method to let us distinguish backends in case we some day have them again. * Support deleting indexes, fields, or fragments. * Support Backup (returning a ReadCloser that dumps the RBF file, implicitly merging any current WAL) and Restore (create a new RBF file). * Change directory structure and fragment keys to match existing databases, so we should in theory be able to open an existing data directory. * Fragment delete probably doesn't lock correctly and this should be reviewed. * Export the DOT-format Dump so we can hook it up to a debug endpoint. This wants to be explored more; ideally the front-end UI should be able to display this. * Create a NopTxStore which can be used like a TxStore but everything that can error errors out. This is then used to let a holder that hasn't had a txstore initialized work anyway. There's at least a couple of open issues that need to be revisited here.
This commit is contained in:
Seebs
parent
672ebb6306
commit
75a68cf60c
7 changed files with 980 additions and 124 deletions
|
|
@ -5,12 +5,14 @@ import (
|
|||
"context"
|
||||
"os"
|
||||
"testing"
|
||||
|
||||
"github.com/molecula/featurebase/v3/keys"
|
||||
)
|
||||
|
||||
func TestDot(t *testing.T) {
|
||||
// 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} {
|
||||
for _, splitter := range []KeySplitter{&flexibleKeySplitter{splitIndexes: map[keys.Index]struct{}{"i": {}}}, nil} {
|
||||
txs := testTxStore(t, "foo", splitter)
|
||||
file, err := os.Create("test.dot")
|
||||
if err != nil {
|
||||
|
|
@ -19,16 +21,16 @@ func TestDot(t *testing.T) {
|
|||
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)
|
||||
q.Write("i", "f", "v", 0)
|
||||
q.Write("i", "g", "v", 0)
|
||||
q.Write("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)
|
||||
q.Read("i", "f", "v", 2)
|
||||
q.Read("j", "f", "v", 0)
|
||||
q2, _ := txs.NewQueryContext(context.Background())
|
||||
defer q2.Release()
|
||||
q2.NewRead("i", "f", "v", 0)
|
||||
q2.NewRead("i", "g", "v", 0)
|
||||
q2.Read("i", "f", "v", 0)
|
||||
q2.Read("i", "g", "v", 0)
|
||||
var dg dotGraph
|
||||
// we know what we actually have here...
|
||||
inner := txs.inner
|
||||
|
|
|
|||
|
|
@ -5,6 +5,7 @@ import (
|
|||
"sort"
|
||||
"strings"
|
||||
|
||||
"github.com/molecula/featurebase/v3/keys"
|
||||
"github.com/molecula/featurebase/v3/roaring"
|
||||
)
|
||||
|
||||
|
|
@ -28,10 +29,10 @@ import (
|
|||
// fail, and refuse to commit, if that context is canceled before you try
|
||||
// to commit.
|
||||
type QueryContext interface {
|
||||
// NewRead requests a new QueryRead object for the indicated fragment.
|
||||
NewRead(IndexName, FieldName, ViewName, ShardID) (QueryRead, error)
|
||||
// NewWrite requests a new QueryWrite object for the indicated fragment.
|
||||
NewWrite(IndexName, FieldName, ViewName, ShardID) (QueryWrite, error)
|
||||
// Read requests a new QueryRead object for the indicated fragment.
|
||||
Read(keys.Index, keys.Field, keys.View, keys.Shard) (QueryRead, error)
|
||||
// Write requests a new QueryWrite object for the indicated fragment.
|
||||
Write(keys.Index, keys.Field, keys.View, keys.Shard) (QueryWrite, error)
|
||||
// Error sets a persistent error state and indicates that this QueryContext
|
||||
// must not commit its writes.
|
||||
Error(...interface{})
|
||||
|
|
@ -48,6 +49,11 @@ type QueryContext interface {
|
|||
// It is an error to try to commit twice or use the QueryContext after a
|
||||
// commit.
|
||||
Commit() error
|
||||
// Flush tries to flush everything related to the given index and shard.
|
||||
// This is NOT portable across backends and is a temporary workaround
|
||||
// needed by our delete flow. Flush() is basically like a commit followed
|
||||
// immediately be reopening, without removing our locks.
|
||||
Flush(keys.Index, keys.Shard) error
|
||||
}
|
||||
|
||||
// QueryRead represents read access to a fragment. When functions in
|
||||
|
|
@ -161,12 +167,14 @@ type QueryWrite interface {
|
|||
// 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
|
||||
type FieldName string
|
||||
type ViewName string
|
||||
type ShardID uint64
|
||||
// Flush is an Inadvisable Workaround for the problem that sometimes we want
|
||||
// to do a thing that has the impact of committing writes so far without releasing
|
||||
// our broader write lock. It may not work with all backends. If it does work, and
|
||||
// doesn't return an error, writes so far to this QueryWrite's backend are flushed.
|
||||
// This may also affect the backend used by other QueryWrites.
|
||||
Flush() error
|
||||
}
|
||||
|
||||
// QueryScope represents a possible set of things that can be written
|
||||
// to. A QueryScope can in principle represent arbitrary patterns with
|
||||
|
|
@ -190,7 +198,7 @@ type ShardID uint64
|
|||
type QueryScope interface {
|
||||
// Allowed determines whether a specific fragment
|
||||
// is covered by this QueryScope.
|
||||
Allowed(IndexName, FieldName, ViewName, ShardID) bool
|
||||
Allowed(keys.Index, keys.Field, keys.View, keys.Shard) bool
|
||||
|
||||
// Overlap reports whether there are any overlaps between this
|
||||
// QueryScope object and another. An overlap exists wherever
|
||||
|
|
@ -199,10 +207,17 @@ type QueryScope interface {
|
|||
Overlap(QueryScope) bool
|
||||
|
||||
AddAll() QueryScope
|
||||
AddIndex(IndexName) QueryScope
|
||||
AddField(IndexName, FieldName) QueryScope
|
||||
AddIndexShards(IndexName, ...ShardID) QueryScope
|
||||
AddFieldShards(IndexName, FieldName, ...ShardID) QueryScope
|
||||
|
||||
// AddIndex adds the whole index, across all shards.
|
||||
AddIndex(keys.Index) QueryScope
|
||||
// AddIndex adds the whole field, across all shards.
|
||||
AddField(keys.Index, keys.Field) QueryScope
|
||||
// AddIndexShards adds the index only for the given shards, but if
|
||||
// there's no shards, it is equivalent to AddIndex.
|
||||
AddIndexShards(keys.Index, ...keys.Shard) QueryScope
|
||||
// AddFieldShards adds the field only for the given shards, but if
|
||||
// there's no shards, it is equivalent to AddField.
|
||||
AddFieldShards(keys.Index, keys.Field, ...keys.Shard) QueryScope
|
||||
|
||||
String() string
|
||||
}
|
||||
|
|
@ -212,7 +227,7 @@ type QueryScope interface {
|
|||
// within those indexes. An empty shard list indicates all shards,
|
||||
// an absent key indicates no shards. Shard lists are stored sorted.
|
||||
type indexShardQueryScope struct {
|
||||
shards map[IndexName]shardList
|
||||
shards map[keys.Index]shardList
|
||||
all bool
|
||||
}
|
||||
|
||||
|
|
@ -239,22 +254,23 @@ func (i *indexShardQueryScope) AddAll() QueryScope {
|
|||
}
|
||||
|
||||
// AddIndex adds the given index, with all shards writable.
|
||||
func (i *indexShardQueryScope) AddIndex(index IndexName) QueryScope {
|
||||
func (i *indexShardQueryScope) AddIndex(index keys.Index) QueryScope {
|
||||
if i.shards == nil {
|
||||
i.shards = map[IndexName]shardList{index: {all: true}}
|
||||
i.shards = map[keys.Index]shardList{index: {all: true}}
|
||||
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) QueryScope {
|
||||
// AddIndexShards adds the given index for the given shards. If there's no shards,
|
||||
// it's equivalent to AddIndex.
|
||||
func (i *indexShardQueryScope) AddIndexShards(index keys.Index, shards ...keys.Shard) QueryScope {
|
||||
if i.all {
|
||||
return i
|
||||
}
|
||||
if i.shards == nil {
|
||||
i.shards = map[IndexName]shardList{}
|
||||
i.shards = map[keys.Index]shardList{}
|
||||
}
|
||||
existing := i.shards[index]
|
||||
// We could at this point check whether anything previously existed, and
|
||||
|
|
@ -263,22 +279,26 @@ func (i *indexShardQueryScope) AddIndexShards(index IndexName, shards ...ShardID
|
|||
if existing.all {
|
||||
return i
|
||||
}
|
||||
for _, shard := range shards {
|
||||
existing.Add(shard)
|
||||
if len(shards) == 0 {
|
||||
existing.all = true
|
||||
} else {
|
||||
for _, shard := range shards {
|
||||
existing.Add(shard)
|
||||
}
|
||||
}
|
||||
i.shards[index] = existing
|
||||
return i
|
||||
}
|
||||
|
||||
func (i *indexShardQueryScope) AddField(index IndexName, _ FieldName) QueryScope {
|
||||
func (i *indexShardQueryScope) AddField(index keys.Index, _ keys.Field) QueryScope {
|
||||
return i.AddIndex(index)
|
||||
}
|
||||
|
||||
func (i *indexShardQueryScope) AddFieldShards(index IndexName, field FieldName, shards ...ShardID) QueryScope {
|
||||
func (i *indexShardQueryScope) AddFieldShards(index keys.Index, field keys.Field, shards ...keys.Shard) QueryScope {
|
||||
return i.AddIndexShards(index, shards...)
|
||||
}
|
||||
|
||||
func (i *indexShardQueryScope) Allowed(index IndexName, _ FieldName, _ ViewName, shard ShardID) bool {
|
||||
func (i *indexShardQueryScope) Allowed(index keys.Index, _ keys.Field, _ keys.View, shard keys.Shard) bool {
|
||||
shards, ok := i.shards[index]
|
||||
if !ok {
|
||||
return false
|
||||
|
|
@ -306,7 +326,7 @@ func (i *indexShardQueryScope) Overlap(qw QueryScope) bool {
|
|||
// reservations.
|
||||
type indexScope struct {
|
||||
all shardList
|
||||
any map[FieldName]shardList
|
||||
any map[keys.Field]shardList
|
||||
}
|
||||
|
||||
// Overlap determines whether two index scopes overlap. This
|
||||
|
|
@ -337,25 +357,25 @@ func (i *indexScope) Overlap(other *indexScope) bool {
|
|||
return false
|
||||
}
|
||||
|
||||
func (scope *indexScope) AddField(field FieldName) {
|
||||
func (scope *indexScope) AddField(field keys.Field) {
|
||||
if scope.all.all {
|
||||
// We already cover everything.
|
||||
return
|
||||
}
|
||||
if scope.any == nil {
|
||||
scope.any = map[FieldName]shardList{field: {all: true}}
|
||||
scope.any = map[keys.Field]shardList{field: {all: true}}
|
||||
return
|
||||
}
|
||||
scope.any[field] = shardList{all: true}
|
||||
}
|
||||
|
||||
func (scope *indexScope) AddFieldShards(field FieldName, shards ...ShardID) {
|
||||
func (scope *indexScope) AddFieldShards(field keys.Field, shards ...keys.Shard) {
|
||||
if scope.all.all {
|
||||
// We already cover everything.
|
||||
return
|
||||
}
|
||||
if scope.any == nil {
|
||||
scope.any = map[FieldName]shardList{}
|
||||
scope.any = map[keys.Field]shardList{}
|
||||
}
|
||||
existing, ok := scope.any[field]
|
||||
if !ok {
|
||||
|
|
@ -364,8 +384,12 @@ func (scope *indexScope) AddFieldShards(field FieldName, shards ...ShardID) {
|
|||
if existing.all {
|
||||
return
|
||||
}
|
||||
for _, shard := range shards {
|
||||
existing.Add(shard)
|
||||
if len(shards) == 0 {
|
||||
existing.all = true
|
||||
} else {
|
||||
for _, shard := range shards {
|
||||
existing.Add(shard)
|
||||
}
|
||||
}
|
||||
scope.any[field] = existing
|
||||
}
|
||||
|
|
@ -420,7 +444,7 @@ func (scope *indexScope) Complexity() string {
|
|||
type flexibleQueryScope struct {
|
||||
all bool
|
||||
splitter *flexibleKeySplitter
|
||||
indexes map[IndexName]*indexScope
|
||||
indexes map[keys.Index]*indexScope
|
||||
}
|
||||
|
||||
var _ QueryScope = &flexibleQueryScope{}
|
||||
|
|
@ -447,12 +471,12 @@ func (i *flexibleQueryScope) AddAll() QueryScope {
|
|||
}
|
||||
|
||||
// AddIndex adds the given index, with all shards writable.
|
||||
func (i *flexibleQueryScope) AddIndex(index IndexName) QueryScope {
|
||||
func (i *flexibleQueryScope) AddIndex(index keys.Index) QueryScope {
|
||||
if i.all {
|
||||
return i
|
||||
}
|
||||
if i.indexes == nil {
|
||||
i.indexes = map[IndexName]*indexScope{index: {all: shardList{all: true}}}
|
||||
i.indexes = map[keys.Index]*indexScope{index: {all: shardList{all: true}}}
|
||||
return i
|
||||
}
|
||||
i.indexes[index] = &indexScope{all: shardList{all: true}}
|
||||
|
|
@ -460,12 +484,12 @@ func (i *flexibleQueryScope) AddIndex(index IndexName) QueryScope {
|
|||
}
|
||||
|
||||
// AddIndexShards adds the given index for the given shards.
|
||||
func (i *flexibleQueryScope) AddIndexShards(index IndexName, shards ...ShardID) QueryScope {
|
||||
func (i *flexibleQueryScope) AddIndexShards(index keys.Index, shards ...keys.Shard) QueryScope {
|
||||
if i.all {
|
||||
return i
|
||||
}
|
||||
if i.indexes == nil {
|
||||
i.indexes = map[IndexName]*indexScope{}
|
||||
i.indexes = map[keys.Index]*indexScope{}
|
||||
}
|
||||
// We could at this point check whether anything previously existed, and
|
||||
// if not, just use a new {any: shards} shardlist, but we want to verify
|
||||
|
|
@ -478,15 +502,19 @@ func (i *flexibleQueryScope) AddIndexShards(index IndexName, shards ...ShardID)
|
|||
if scope.all.all {
|
||||
return i
|
||||
}
|
||||
for _, shard := range shards {
|
||||
scope.all.Add(shard)
|
||||
if len(shards) == 0 {
|
||||
scope.all.all = true
|
||||
} else {
|
||||
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) QueryScope {
|
||||
func (i *flexibleQueryScope) AddField(index keys.Index, field keys.Field) QueryScope {
|
||||
if i.all {
|
||||
return i
|
||||
}
|
||||
|
|
@ -497,7 +525,7 @@ func (i *flexibleQueryScope) AddField(index IndexName, field FieldName) QuerySco
|
|||
}
|
||||
}
|
||||
if i.indexes == nil {
|
||||
i.indexes = make(map[IndexName]*indexScope)
|
||||
i.indexes = make(map[keys.Index]*indexScope)
|
||||
}
|
||||
scope := i.indexes[index]
|
||||
if scope == nil {
|
||||
|
|
@ -509,7 +537,7 @@ func (i *flexibleQueryScope) AddField(index IndexName, field FieldName) QuerySco
|
|||
}
|
||||
|
||||
// AddFieldShards adds the given index for the given shards.
|
||||
func (i *flexibleQueryScope) AddFieldShards(index IndexName, field FieldName, shards ...ShardID) QueryScope {
|
||||
func (i *flexibleQueryScope) AddFieldShards(index keys.Index, field keys.Field, shards ...keys.Shard) QueryScope {
|
||||
if i.all {
|
||||
return i
|
||||
}
|
||||
|
|
@ -520,7 +548,7 @@ func (i *flexibleQueryScope) AddFieldShards(index IndexName, field FieldName, sh
|
|||
}
|
||||
}
|
||||
if i.indexes == nil {
|
||||
i.indexes = make(map[IndexName]*indexScope)
|
||||
i.indexes = make(map[keys.Index]*indexScope)
|
||||
}
|
||||
scope := i.indexes[index]
|
||||
if scope == nil {
|
||||
|
|
@ -531,7 +559,7 @@ func (i *flexibleQueryScope) AddFieldShards(index IndexName, field FieldName, sh
|
|||
return i
|
||||
}
|
||||
|
||||
func (i *flexibleQueryScope) Allowed(index IndexName, field FieldName, _ ViewName, shard ShardID) bool {
|
||||
func (i *flexibleQueryScope) Allowed(index keys.Index, field keys.Field, _ keys.View, shard keys.Shard) bool {
|
||||
if i.all {
|
||||
return true
|
||||
}
|
||||
|
|
@ -587,13 +615,13 @@ func (i *flexibleQueryScope) Overlap(qw QueryScope) (out bool) {
|
|||
// not a good fit.
|
||||
type shardList struct {
|
||||
all bool
|
||||
any []ShardID
|
||||
any []keys.Shard
|
||||
}
|
||||
|
||||
// findShard returns the positive index at which shard was found
|
||||
// in the shard list, or the negative index at which it would have
|
||||
// been (and thus the insertion point for an add).
|
||||
func (s *shardList) findShard(shard ShardID) int {
|
||||
func (s *shardList) findShard(shard keys.Shard) int {
|
||||
l, h := 0, len(s.any)
|
||||
for h > l {
|
||||
m := (h + l) / 2
|
||||
|
|
@ -617,7 +645,7 @@ func (s *shardList) findShard(shard ShardID) int {
|
|||
|
||||
// Allowed indicates whether the given shard is currently included
|
||||
// in the set.
|
||||
func (s *shardList) Allowed(shard ShardID) bool {
|
||||
func (s *shardList) Allowed(shard keys.Shard) bool {
|
||||
if s.all {
|
||||
return true
|
||||
}
|
||||
|
|
@ -653,7 +681,7 @@ func (s *shardList) Overlap(other shardList) bool {
|
|||
|
||||
// Add adds the given shard to the shardlist, maintaining
|
||||
// sorted order.
|
||||
func (s *shardList) Add(shard ShardID) {
|
||||
func (s *shardList) Add(shard keys.Shard) {
|
||||
if s.all {
|
||||
return
|
||||
}
|
||||
|
|
@ -661,7 +689,7 @@ func (s *shardList) Add(shard ShardID) {
|
|||
// 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}
|
||||
s.any = []keys.Shard{shard}
|
||||
return
|
||||
}
|
||||
if s.any[len(s.any)-1] < shard {
|
||||
|
|
|
|||
|
|
@ -11,6 +11,7 @@ import (
|
|||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/molecula/featurebase/v3/keys"
|
||||
rbfcfg "github.com/molecula/featurebase/v3/rbf/cfg"
|
||||
"golang.org/x/sync/errgroup"
|
||||
)
|
||||
|
|
@ -29,7 +30,7 @@ var rbfTestConfig = func() *rbfcfg.Config {
|
|||
// 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)
|
||||
txs, err := NewRBFTxStore(dir, rbfTestConfig, nil, nil, ks)
|
||||
if err != nil {
|
||||
tb.Fatalf("opening TxStore: %v", err)
|
||||
}
|
||||
|
|
@ -45,7 +46,7 @@ func testTxStore(tb testing.TB, path string, ks KeySplitter) *testTxStoreWrapper
|
|||
func TestTxStoreClose(t *testing.T) {
|
||||
dir := filepath.Join(t.TempDir(), "foo")
|
||||
ctx := context.Background()
|
||||
txs, err := NewRBFTxStore(dir, rbfTestConfig, nil)
|
||||
txs, err := NewRBFTxStore(dir, rbfTestConfig, nil, nil, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("opening TxStore: %v", err)
|
||||
}
|
||||
|
|
@ -53,7 +54,7 @@ func TestTxStoreClose(t *testing.T) {
|
|||
if err != nil {
|
||||
t.Fatalf("creating initial query context: %v", err)
|
||||
}
|
||||
_, err = qcx.NewWrite("i", "f", "v", 0)
|
||||
_, err = qcx.Write("i", "f", "v", 0)
|
||||
if err == nil {
|
||||
t.Fatalf("should get error requesting a write from a read qcx")
|
||||
}
|
||||
|
|
@ -85,12 +86,12 @@ func TestTxStoreClose(t *testing.T) {
|
|||
func TestShardList(t *testing.T) {
|
||||
rng := rand.New(rand.NewSource(3))
|
||||
prev := &shardList{}
|
||||
prevSeen := map[ShardID]struct{}{}
|
||||
prevSeen := map[keys.Shard]struct{}{}
|
||||
for i := 0; i < 20; i++ {
|
||||
sl := &shardList{}
|
||||
seen := make(map[ShardID]struct{})
|
||||
seen := make(map[keys.Shard]struct{})
|
||||
for j := 0; j < 20; j++ {
|
||||
add := ShardID(rng.Intn(20))
|
||||
add := keys.Shard(rng.Intn(20))
|
||||
sl.Add(add)
|
||||
seen[add] = struct{}{}
|
||||
}
|
||||
|
|
@ -104,7 +105,7 @@ func TestShardList(t *testing.T) {
|
|||
}
|
||||
}
|
||||
// verify that we neither allow things not included, nor disallow things included
|
||||
for j := ShardID(0); j < 20; j++ {
|
||||
for j := keys.Shard(0); j < 20; j++ {
|
||||
_, ok1 := seen[j]
|
||||
ok2 := sl.Allowed(j)
|
||||
if ok1 != ok2 {
|
||||
|
|
@ -134,7 +135,7 @@ func TestTxStore(t *testing.T) {
|
|||
ctx := context.Background()
|
||||
txs := testTxStore(t, "foo", nil)
|
||||
q, _ := txs.NewQueryContext(ctx)
|
||||
_, _ = q.NewRead("i", "f", "v", 0)
|
||||
_, _ = q.Read("i", "f", "v", 0)
|
||||
txs.expect(1)
|
||||
err := txs.Close()
|
||||
if err == nil {
|
||||
|
|
@ -166,9 +167,9 @@ func TestTxStore(t *testing.T) {
|
|||
// 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 keys.Index
|
||||
field keys.Field
|
||||
shard keys.Shard
|
||||
justRead bool
|
||||
}
|
||||
|
||||
|
|
@ -182,8 +183,8 @@ func satisfyWriteRequest(t *testing.T, txs *testTxStoreWrapper, requests []write
|
|||
scope := txs.Scope()
|
||||
prevIndex := requests[0].index
|
||||
prevField := requests[0].field
|
||||
shards := []ShardID{requests[0].shard}
|
||||
add := func(index IndexName, field FieldName, shards ...ShardID) {
|
||||
shards := []keys.Shard{requests[0].shard}
|
||||
add := func(index keys.Index, field keys.Field, shards ...keys.Shard) {
|
||||
if field == "" {
|
||||
if len(shards) == 1 && shards[0] == 0 {
|
||||
scope.AddIndex(index)
|
||||
|
|
@ -224,20 +225,20 @@ func satisfyWriteRequest(t *testing.T, txs *testTxStoreWrapper, requests []write
|
|||
for _, i := range rand.Perm(len(requests)) {
|
||||
req := requests[i]
|
||||
if req.justRead {
|
||||
qr, err := qcx.NewRead(req.index, req.field, "v", req.shard)
|
||||
qr, err := qcx.Read(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)
|
||||
_, err := qcx.Write(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)
|
||||
qw, err := qcx.Write(req.index, req.field, "v", req.shard)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
|
@ -270,7 +271,7 @@ func delayWriteRequest(t *testing.T, txs *testTxStoreWrapper, writeList []writeR
|
|||
}
|
||||
for _, i := range rand.Perm(len(writeList)) {
|
||||
write := writeList[i]
|
||||
qw, err := qcx.NewWrite(write.index, write.field, "v", write.shard)
|
||||
qw, err := qcx.Write(write.index, write.field, "v", write.shard)
|
||||
if err != nil {
|
||||
qcx.Release()
|
||||
return nil, err
|
||||
|
|
@ -288,7 +289,7 @@ func testSomeWriteRequests(t *testing.T, writeRequests [][]writeReq) {
|
|||
// try some of these with a splitIndexes splitter, some with an indexShard splitter,
|
||||
// so they both get tested
|
||||
if len(writeRequests)%3 == 1 {
|
||||
splitter = &flexibleKeySplitter{splitIndexes: map[IndexName]struct{}{"a": {}}}
|
||||
splitter = &flexibleKeySplitter{splitIndexes: map[keys.Index]struct{}{"a": {}}}
|
||||
} else {
|
||||
splitter = &indexShardKeySplitter{}
|
||||
}
|
||||
|
|
@ -494,15 +495,15 @@ func buildRequestsFromBytes(data []byte) [][]writeReq {
|
|||
total := 0
|
||||
for len(data) >= 2 {
|
||||
idx := data[0] % 5
|
||||
index := IndexName(rune(idx) + 'a')
|
||||
index := keys.Index(rune(idx) + 'a')
|
||||
total += int(idx)
|
||||
fld := (data[0] / 5) % 3
|
||||
field := FieldName(rune(fld) + 'a')
|
||||
field := keys.Field(rune(fld) + 'a')
|
||||
// generate some index-only requests
|
||||
if fld == 0 {
|
||||
field = ""
|
||||
}
|
||||
shard := ShardID(data[1] % 8)
|
||||
shard := keys.Shard(data[1] % 8)
|
||||
total += int(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
|
||||
|
|
@ -538,3 +539,7 @@ func FuzzWriteRequests(f *testing.F) {
|
|||
testSomeWriteRequests(t, writeRequests)
|
||||
})
|
||||
}
|
||||
|
||||
func TestNopTxStore(t *testing.T) {
|
||||
NopTxStore.Scope()
|
||||
}
|
||||
|
|
|
|||
|
|
@ -3,16 +3,23 @@ package querycontext
|
|||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"runtime"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
|
||||
"github.com/molecula/featurebase/v3/keys"
|
||||
"github.com/molecula/featurebase/v3/logger"
|
||||
"github.com/molecula/featurebase/v3/rbf"
|
||||
rbfcfg "github.com/molecula/featurebase/v3/rbf/cfg"
|
||||
"github.com/molecula/featurebase/v3/roaring"
|
||||
"github.com/molecula/featurebase/v3/task"
|
||||
"github.com/pkg/errors"
|
||||
"golang.org/x/sync/errgroup"
|
||||
)
|
||||
|
||||
// rbfDBQueryContexts represents an actual backend DB, and a map of the
|
||||
|
|
@ -30,6 +37,18 @@ type rbfDBQueryContexts struct {
|
|||
queryContexts map[*rbfTxWrappers]*rbfQueryContext
|
||||
}
|
||||
|
||||
func (r *rbfDBQueryContexts) close() error {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
if len(r.queryContexts) > 0 {
|
||||
return fmt.Errorf("db %q: %d transaction(s) still open", r.dbPath, len(r.queryContexts))
|
||||
}
|
||||
if r.db != nil {
|
||||
return r.db.Close()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// writeTx obtains a write Tx, and associates it with the given query context. only
|
||||
// one thing should ever have a write context at once, and because the QueryContext is
|
||||
// supposed to be protecting us here, it's actually just plain an error for us
|
||||
|
|
@ -63,6 +82,22 @@ func (r *rbfDBQueryContexts) readTx(rq *rbfQueryContext) (*rbfTxWrappers, error)
|
|||
return q, nil
|
||||
}
|
||||
|
||||
// contents reports what fragment keys are in this database.
|
||||
// This should really return []fragKey, but we're not there yet.
|
||||
func (r *rbfDBQueryContexts) contents() ([]string, error) {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
// scan database for fields and views...
|
||||
tx, err := r.db.Begin(false)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("scanning database at %q: %v", r.dbPath, err)
|
||||
}
|
||||
defer tx.Rollback()
|
||||
fvs := tx.FieldViews()
|
||||
tx.Rollback()
|
||||
return fvs, nil
|
||||
}
|
||||
|
||||
// release marks a given rbfQueryContext as no longer using the db
|
||||
func (r *rbfDBQueryContexts) release(rt *rbfTxWrappers) {
|
||||
r.mu.Lock()
|
||||
|
|
@ -111,16 +146,28 @@ type rbfTxStore struct {
|
|||
queries map[*rbfQueryContext]struct{}
|
||||
cfg *rbfcfg.Config
|
||||
closed bool
|
||||
workerPool *task.Pool
|
||||
logger logger.Logger
|
||||
}
|
||||
|
||||
func (*rbfTxStore) Backend() string {
|
||||
return "rbf"
|
||||
}
|
||||
|
||||
// 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.
|
||||
//
|
||||
// If present, the logger is used for diagnostic messages, otherwise they're
|
||||
// written to os.Stderr. If a task.Pool is provided, blocking operations
|
||||
// may report themselves to it as blocked or unblocked. (For instance, a
|
||||
// request for a WriteQueryContext will mark itself as blocked while waiting
|
||||
// on previous contexts.)
|
||||
//
|
||||
// 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) {
|
||||
func NewRBFTxStore(path string, cfg *rbfcfg.Config, logger logger.Logger, workers *task.Pool, splitter KeySplitter) (*rbfTxStore, error) {
|
||||
if cfg == nil {
|
||||
cfg = rbfcfg.NewDefaultConfig()
|
||||
}
|
||||
|
|
@ -134,11 +181,38 @@ func NewRBFTxStore(path string, cfg *rbfcfg.Config, splitter KeySplitter) (*rbfT
|
|||
writeScopes: make(map[*rbfQueryContext]QueryScope),
|
||||
queries: make(map[*rbfQueryContext]struct{}),
|
||||
cfg: cfg,
|
||||
workerPool: workers,
|
||||
logger: logger,
|
||||
}
|
||||
r.writeQueue = sync.NewCond(&r.mu)
|
||||
return r, nil
|
||||
}
|
||||
|
||||
func (r *rbfTxStore) DumpDot(w io.Writer) error {
|
||||
var dg dotGraph
|
||||
dg.enqueue(r)
|
||||
dg.build(5) // 5 is the right depth to see the whole tree, roughly
|
||||
return dg.Write(w)
|
||||
}
|
||||
|
||||
// createDBQueryContext creates a new rbfDBQueryContexts associating
|
||||
// the dbKey with the provided path, and opening the underlying file.
|
||||
func (r *rbfTxStore) createDBQueryContext(dbk dbKey, dbPath string) (*rbfDBQueryContexts, error) {
|
||||
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
|
||||
}
|
||||
|
||||
// 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) {
|
||||
|
|
@ -150,19 +224,29 @@ func (r *rbfTxStore) getDB(dbk dbKey) (*rbfDBQueryContexts, error) {
|
|||
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()
|
||||
return r.createDBQueryContext(dbk, dbPath)
|
||||
}
|
||||
|
||||
// getExistingDB gets an existing db for the given key. call only when you hold
|
||||
// the rbfTxStore's lock. this will open a database that already exists, even
|
||||
// if there isn't an existing rbfDBQueryContexts for it, but will not create
|
||||
// a new file. If the file doesn't exist, this function returns a nil
|
||||
// *rbfDBQueryContexts, and a nil error. You have to check the returned db
|
||||
// before using it. The returned DB is stashed in r.dbs as usual.
|
||||
func (r *rbfTxStore) getExistingDB(dbk dbKey) (*rbfDBQueryContexts, error) {
|
||||
db, ok := r.dbs[dbk]
|
||||
if ok && db != nil {
|
||||
return db, nil
|
||||
}
|
||||
dbPath, err := r.dbPath(dbk)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
r.dbs[dbk] = db
|
||||
return db, nil
|
||||
_, err = os.Stat(filepath.Join(r.rootPath, dbPath))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return r.createDBQueryContext(dbk, dbPath)
|
||||
}
|
||||
|
||||
// newReadTx creates a read Tx in the backend, and returns an rbfTxWrappers
|
||||
|
|
@ -245,7 +329,13 @@ func (r *rbfTxStore) NewQueryContext(ctx context.Context) (QueryContext, error)
|
|||
// for any {index, field, view, shard} that's Allowed() by writes, and can also create
|
||||
// read requests for others.
|
||||
func (r *rbfTxStore) NewWriteQueryContext(ctx context.Context, scope QueryScope) (QueryContext, error) {
|
||||
// we assign random names so that we can debug which query context is which more easily
|
||||
if r.workerPool != nil {
|
||||
// if we have been given a worker pool, we mark this worker as
|
||||
// blocked until we make it through creating the context, because
|
||||
// creating a write query context can block for quite a while.
|
||||
r.workerPool.Block()
|
||||
defer r.workerPool.Unblock()
|
||||
}
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
if r.closed {
|
||||
|
|
@ -260,6 +350,7 @@ func (r *rbfTxStore) NewWriteQueryContext(ctx context.Context, scope QueryScope)
|
|||
scope: scope,
|
||||
queries: make(map[dbKey]*rbfTxWrappers),
|
||||
}
|
||||
|
||||
rq.name = fmt.Sprintf("wqcx-%p", rq)
|
||||
r.writeScopes[rq] = scope
|
||||
r.queries[rq] = struct{}{}
|
||||
|
|
@ -267,6 +358,322 @@ func (r *rbfTxStore) NewWriteQueryContext(ctx context.Context, scope QueryScope)
|
|||
return rq, nil
|
||||
}
|
||||
|
||||
func (r *rbfTxStore) DeleteIndex(index keys.Index) error {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
dirPath := filepath.Join(r.rootPath, string(index), "backends", "rbf")
|
||||
dirInfo, err := os.Stat(dirPath)
|
||||
if err != nil {
|
||||
// The directory doesn't exist, so presumably we don't have anything
|
||||
// to do here.
|
||||
if os.IsNotExist(err) {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
}
|
||||
if !dirInfo.IsDir() {
|
||||
return fmt.Errorf("%q exists, but is not a directory", dirPath)
|
||||
}
|
||||
prefix := string(index) + "/"
|
||||
for k, db := range r.dbs {
|
||||
if !strings.HasPrefix(string(k), prefix) {
|
||||
continue
|
||||
}
|
||||
// Close every matching db.
|
||||
// close will fail if there are any open querycontexts, so we can't
|
||||
// close anything that's in use. If we close a database, we
|
||||
// immediately remove it from our table of databases (which we can
|
||||
// do because we're holding the lock on the TxStore).
|
||||
// It's harmless to close a database no one is using, it can just
|
||||
// be reopened later.
|
||||
if err := db.close(); err != nil {
|
||||
return err
|
||||
}
|
||||
delete(r.dbs, k)
|
||||
}
|
||||
// If we got here, every such database is closed. Let's remove
|
||||
// the directory tree for the rbf backend, in its entirety.
|
||||
return os.RemoveAll(dirPath)
|
||||
}
|
||||
|
||||
func (r *rbfTxStore) DeleteField(index keys.Index, field keys.Field) error {
|
||||
// We don't have to worry about existing reads, because existing
|
||||
// reads will be against their own snapshot of the RBF file and
|
||||
// won't be affected by changes we make.
|
||||
//
|
||||
// But something could already have a write lock on one of these files,
|
||||
// and we can't safely grab the TxStore lock and then wait for a lock on
|
||||
// the file. The simplest solution: Create a WriteQueryContext for each
|
||||
// shard as we get to it.
|
||||
indexPath := filepath.Join(r.rootPath, string(index))
|
||||
shards, err := filepath.Glob(filepath.Join(indexPath, "backends", "rbf", "shard.*"))
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "finding shards")
|
||||
}
|
||||
// the field prefix will be the same in all of these
|
||||
fieldPrefix := fmt.Sprintf("~%s;", field)
|
||||
for _, shardPath := range shards {
|
||||
shardName := filepath.Base(shardPath)
|
||||
shardNum, err := strconv.ParseInt(shardName[6:], 10, 64)
|
||||
if err != nil {
|
||||
return fmt.Errorf("malformed shard path %q: %v", shardName, err)
|
||||
}
|
||||
shard := keys.Shard(shardNum)
|
||||
// Do the fancy bits in their own function so we can use defer.
|
||||
err = func() error {
|
||||
// create a query context, giving us for free all the fancy lock
|
||||
// interactions we already figured out once, get a QueryWrite for it
|
||||
// because that's the easiest way to get a working rbf.Tx with the
|
||||
// right scope, then peek under the hood briefly.
|
||||
qcx, err := r.NewWriteQueryContext(context.TODO(), r.Scope().AddIndexShards(index, shard))
|
||||
if err != nil {
|
||||
return fmt.Errorf("getting write context for %q/%d: %w", index, shard, err)
|
||||
}
|
||||
defer qcx.Release()
|
||||
// This is theoretically more narrow than what we're doing, but we are the
|
||||
// implementation internals and are allowed to cheat like this.
|
||||
qw, err := qcx.Write(index, field, "", shard)
|
||||
if err != nil {
|
||||
return fmt.Errorf("getting write access for %q/%d: %w", index, shard, err)
|
||||
}
|
||||
rqw, ok := qw.(*rbfQueryWrite)
|
||||
if !ok {
|
||||
return errors.New("internal error: wrong tx type for query write")
|
||||
}
|
||||
err = rqw.tx.tx.DeleteBitmapsWithPrefix(fieldPrefix)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "deleting bitmap")
|
||||
}
|
||||
return qcx.Commit()
|
||||
}()
|
||||
if err != nil {
|
||||
return fmt.Errorf("deleting field %q from shard %d of index %q: %w", field, shard, index, err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
type rbfSnapshotReadCloser struct {
|
||||
qcx QueryContext
|
||||
tx *rbf.Tx
|
||||
io.Reader
|
||||
}
|
||||
|
||||
func (r *rbfSnapshotReadCloser) Close() error {
|
||||
r.tx.Rollback()
|
||||
r.qcx.Release()
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *rbfTxStore) Backup(qcx QueryContext, index keys.Index, shard keys.Shard) (_ io.ReadCloser, failed error) {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
// We're responsible for ensuring the querycontext is released, either
|
||||
// when we're done or when our returned ReadCloser is closed.
|
||||
defer func() {
|
||||
if failed != nil {
|
||||
qcx.Release()
|
||||
}
|
||||
}()
|
||||
dbk, _ := r.keys(index, "", "", shard)
|
||||
db, err := r.getExistingDB(dbk)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if db == nil {
|
||||
return nil, fmt.Errorf("no data for %q/%d", index, shard)
|
||||
}
|
||||
tx, err := db.db.Begin(false)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
reader, err := tx.SnapshotReader()
|
||||
if err != nil {
|
||||
tx.Rollback()
|
||||
return nil, err
|
||||
}
|
||||
return &rbfSnapshotReadCloser{qcx: qcx, tx: tx, Reader: reader}, nil
|
||||
}
|
||||
|
||||
// Restore takes a reader containing data compatible with the backend,
|
||||
// such as an RBF file, and replaces all the existing data for this index
|
||||
// and shard.
|
||||
func (r *rbfTxStore) Restore(qcx QueryContext, index keys.Index, shard keys.Shard, src io.Reader) (err error) {
|
||||
defer func() {
|
||||
// clarify the error once so we don't have to do this everywhere
|
||||
if err != nil {
|
||||
err = fmt.Errorf("can't restore %q/%d: %w", index, shard, err)
|
||||
}
|
||||
}()
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
dbk, _ := r.keys(index, "", "", shard)
|
||||
if db := r.dbs[dbk]; db != nil {
|
||||
// uh-oh... we already have this?
|
||||
// let's try to close it.
|
||||
err := db.close()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
delete(r.dbs, dbk)
|
||||
}
|
||||
dbPath, err := r.dbPath(dbk)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
dirPath := filepath.Join(r.rootPath, dbPath)
|
||||
parent := filepath.Dir(dirPath)
|
||||
if err := os.MkdirAll(parent, 0o750); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := os.Mkdir(dirPath, 0o750); err != nil {
|
||||
if os.IsExist(err) {
|
||||
// remove existing directory and try again
|
||||
err = os.RemoveAll(dirPath)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
err = os.Mkdir(dirPath, 0o750)
|
||||
}
|
||||
// err could now be nil because it was previously an Exist
|
||||
// error, and we replaced it with the result of Mkdir
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
// the directory now exists, so.
|
||||
dataPath := filepath.Join(dirPath, "data")
|
||||
tmpPath := dataPath + ".tmp"
|
||||
file, err := os.OpenFile(tmpPath, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0o600)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer func() {
|
||||
if file != nil {
|
||||
// can't really do anything about an error here. it's okay to
|
||||
// double-close files, it just produces an error.
|
||||
_ = file.Close()
|
||||
// remove the temp file, if it still exists, which it might in a failed
|
||||
// path.
|
||||
_ = os.Remove(tmpPath)
|
||||
}
|
||||
}()
|
||||
_, err = io.Copy(file, src)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "copying to file")
|
||||
}
|
||||
if err = file.Sync(); err != nil {
|
||||
return err
|
||||
}
|
||||
if err = file.Close(); err != nil {
|
||||
return err
|
||||
}
|
||||
if err = os.Rename(tmpPath, dataPath); err != nil {
|
||||
return err
|
||||
}
|
||||
// reopen the file, now that it exists, and report an error if that doesn't work
|
||||
_, err = r.getExistingDB(dbk)
|
||||
return err
|
||||
}
|
||||
|
||||
func (r *rbfTxStore) ListFieldViews(index keys.Index, shard keys.Shard) (map[keys.Field][]keys.View, error) {
|
||||
fvs, err := r.dbContents(index, shard)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("scanning database for existing contents: %v", err)
|
||||
}
|
||||
// Once we've done a couple of shards, fields and views are likely
|
||||
// to already exist.
|
||||
result := make(map[keys.Field][]keys.View)
|
||||
for _, fv := range fvs {
|
||||
_, field, view, _, err := r.parseFragKey(fragKey(fv))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("invalid fragment key %q for %q/%d: %w", fv, index, shard, err)
|
||||
}
|
||||
result[field] = append(result[field], view)
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
// DeleteFragment tries to delete the given fragment. It's not an error for
|
||||
// it to already not exist. This is untested and unverified as of this
|
||||
// writing; I wrote it just to try to verify the API.
|
||||
func (r *rbfTxStore) DeleteFragments(index keys.Index, field keys.Field, views []keys.View, shards []keys.Shard) error {
|
||||
if _, ok := r.KeySplitter.(*indexShardKeySplitter); !ok {
|
||||
return fmt.Errorf("fragment batch delete only supported for indexShardKeySplitter, have %T", r.KeySplitter)
|
||||
}
|
||||
// nothing to do
|
||||
if len(views) == 0 || len(shards) == 0 {
|
||||
return nil
|
||||
}
|
||||
dbKeys := make([]dbKey, len(shards))
|
||||
fragKeys := make([]fragKey, len(views))
|
||||
|
||||
for i, view := range views {
|
||||
fragKeys[i] = r.fragKey(index, field, view, shards[0])
|
||||
}
|
||||
for i, shard := range shards {
|
||||
dbKeys[i] = r.dbKey(index, field, views[0], shard)
|
||||
}
|
||||
// Now, for every shard, we want to delete all those views. WARNING:
|
||||
// WE DO NOT RESPECT THE QUERYCONTEXT WRITE GATES. We are just going
|
||||
// straight into the RBF database and requesting a Tx. This should be
|
||||
// safe because we do not block on *anything external to the individual
|
||||
// databases, and each of these operations will separately release its
|
||||
// lock the moment it's done.
|
||||
ctx := context.Background()
|
||||
// we use a WithContext errgroup because this provides the convenient
|
||||
// trait that if one of the deletes fails, the others can bail prematurely.
|
||||
|
||||
eg, ctx := errgroup.WithContext(ctx)
|
||||
// Completely arbitrary value: allow up to 8 running at once.
|
||||
sema := make(chan struct{}, 8)
|
||||
var err error
|
||||
for _, dbKey := range dbKeys {
|
||||
if ctx.Err() != nil {
|
||||
break
|
||||
}
|
||||
var db *rbfDBQueryContexts
|
||||
db, err = r.getExistingDB(dbKey)
|
||||
if err != nil {
|
||||
// if we encounter an error, we stop. note that err has
|
||||
// scope outside this loop, so this error is available to
|
||||
// the surrounding code.
|
||||
break
|
||||
}
|
||||
// getExistingDB is allowed to return a nil DB with no error,
|
||||
// indicating that the DB didn't exist.
|
||||
if db == nil {
|
||||
continue
|
||||
}
|
||||
// use channel as a cheap semaphore to cap our simultaneous
|
||||
// deletes
|
||||
sema <- struct{}{}
|
||||
eg.Go(func() error {
|
||||
defer func() {
|
||||
<-sema
|
||||
}()
|
||||
tx, err := db.db.Begin(true)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer tx.Rollback()
|
||||
for _, fragKey := range fragKeys {
|
||||
err := tx.DeleteBitmapsWithPrefix(string(fragKey))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return tx.Commit()
|
||||
})
|
||||
}
|
||||
err2 := eg.Wait()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return err2
|
||||
}
|
||||
|
||||
func (r *rbfTxStore) Close() error {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
|
|
@ -279,17 +686,15 @@ func (r *rbfTxStore) Close() error {
|
|||
return fmt.Errorf("can't close TxStore while query contexts are outstanding")
|
||||
}
|
||||
// even if we fail to close databases, *we're* closed and will
|
||||
// no longer allow new QueryContexts
|
||||
// no longer allow new QueryContexts. because of this, we try to close
|
||||
// the whole database.
|
||||
r.closed = true
|
||||
for key, db := range r.dbs {
|
||||
if len(db.queryContexts) > 0 {
|
||||
firstErr = fmt.Errorf("db %q: %d transaction(s) still open", db.dbPath, len(db.queryContexts))
|
||||
continue
|
||||
}
|
||||
if db.db != nil {
|
||||
if err := db.db.Close(); err != nil {
|
||||
if err := db.close(); err != nil {
|
||||
if firstErr == nil {
|
||||
firstErr = err
|
||||
}
|
||||
continue
|
||||
}
|
||||
// we remove this from our list of databases anyway, as long as it wasn't
|
||||
// still in use. we can't retry the close itself.
|
||||
|
|
@ -298,6 +703,72 @@ func (r *rbfTxStore) Close() error {
|
|||
return firstErr
|
||||
}
|
||||
|
||||
func (r *rbfTxStore) dbContents(index keys.Index, shard keys.Shard) ([]string, error) {
|
||||
dbk, _ := r.keys(keys.Index(index), "", "", shard)
|
||||
db, err := r.getExistingDB(dbk)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("opening database for %q/%d: %v", index, shard, err)
|
||||
}
|
||||
if db == nil {
|
||||
return nil, fmt.Errorf("opening existing database for %q/%d: no error, but also no database", index, shard)
|
||||
}
|
||||
return db.contents()
|
||||
}
|
||||
|
||||
func (r *rbfTxStore) Contents() (keys.DBContents, error) {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
indexes, err := filepath.Glob(filepath.Join(r.rootPath, "*"))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
contents := make(keys.DBContents, len(indexes))
|
||||
// For each index...
|
||||
for _, indexPath := range indexes {
|
||||
index := filepath.Base(indexPath)
|
||||
shards, err := filepath.Glob(filepath.Join(indexPath, "backends", "rbf", "shard.*"))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
// Create an IndexContents. We can do this unconditionally because we know
|
||||
// we haven't seen this index before.
|
||||
indexContents := make(keys.IndexContents)
|
||||
contents[keys.Index(index)] = indexContents
|
||||
for _, shardPath := range shards {
|
||||
shardName := filepath.Base(shardPath)
|
||||
shardNum, err := strconv.ParseInt(shardName[6:], 10, 64)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("malformed shard path %q: %v", shardName, err)
|
||||
}
|
||||
shard := keys.Shard(shardNum)
|
||||
fvs, err := r.dbContents(keys.Index(index), shard)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("scanning database for existing contents: %v", err)
|
||||
}
|
||||
// Once we've done a couple of shards, fields and views are likely
|
||||
// to already exist.
|
||||
for _, fv := range fvs {
|
||||
_, field, view, _, err := r.parseFragKey(fragKey(fv))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("invalid fragment key %q in %q: %w", fv, shardPath, err)
|
||||
}
|
||||
fieldContents := indexContents[field]
|
||||
if fieldContents == nil {
|
||||
fieldContents = make(keys.FieldContents)
|
||||
indexContents[field] = fieldContents
|
||||
}
|
||||
viewContents := fieldContents[view]
|
||||
if viewContents == nil {
|
||||
viewContents = make(keys.ViewContents)
|
||||
fieldContents[view] = viewContents
|
||||
}
|
||||
viewContents[shard] = struct{}{}
|
||||
}
|
||||
}
|
||||
}
|
||||
return contents, nil
|
||||
}
|
||||
|
||||
// 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 QueryWrite. (We
|
||||
|
|
@ -314,9 +785,12 @@ type rbfTxWrappers struct {
|
|||
tx *rbf.Tx
|
||||
queries map[fragKey]QueryRead
|
||||
writeTx bool
|
||||
mu sync.Mutex // ops are not concurrency-safe
|
||||
}
|
||||
|
||||
func (txw *rbfTxWrappers) readKey(fk fragKey) (QueryRead, error) {
|
||||
txw.mu.Lock()
|
||||
defer txw.mu.Unlock()
|
||||
qr, ok := txw.queries[fk]
|
||||
if !ok {
|
||||
read := rbfQueryRead{tx: txw, fk: fk}
|
||||
|
|
@ -337,6 +811,8 @@ func (txw *rbfTxWrappers) readKey(fk fragKey) (QueryRead, error) {
|
|||
var errWriteRequestOnNonWrite = errors.New("write request tried to use read-only transaction")
|
||||
|
||||
func (txw *rbfTxWrappers) writeKey(fk fragKey) (QueryWrite, error) {
|
||||
txw.mu.Lock()
|
||||
defer txw.mu.Unlock()
|
||||
if qr, ok := txw.queries[fk]; ok {
|
||||
if qw, ok := qr.(*rbfQueryWrite); ok {
|
||||
return qw, nil
|
||||
|
|
@ -348,6 +824,127 @@ func (txw *rbfTxWrappers) writeKey(fk fragKey) (QueryWrite, error) {
|
|||
return write, nil
|
||||
}
|
||||
|
||||
// To allow for flushes, which can replace our tx, we provide wrappers for all
|
||||
// the QueryRead/QueryWrite functions which use a shared lock. Note that rbf is
|
||||
// also, now-redundantly, doing locking on the operations that go through it,
|
||||
// but also we're controlling access to txw.tx.
|
||||
|
||||
func (txw *rbfTxWrappers) ContainerIterator(fk fragKey, ckey uint64) (citer roaring.ContainerIterator, found bool, err error) {
|
||||
txw.mu.Lock()
|
||||
defer txw.mu.Unlock()
|
||||
return txw.tx.ContainerIterator(string(fk), ckey)
|
||||
}
|
||||
|
||||
func (txw *rbfTxWrappers) ApplyFilter(fk fragKey, ckey uint64, filter roaring.BitmapFilter) (err error) {
|
||||
txw.mu.Lock()
|
||||
defer txw.mu.Unlock()
|
||||
return txw.tx.ApplyFilter(string(fk), ckey, filter)
|
||||
}
|
||||
|
||||
func (txw *rbfTxWrappers) Container(fk fragKey, ckey uint64) (*roaring.Container, error) {
|
||||
txw.mu.Lock()
|
||||
defer txw.mu.Unlock()
|
||||
return txw.tx.Container(string(fk), ckey)
|
||||
}
|
||||
|
||||
func (txw *rbfTxWrappers) Contains(fk fragKey, v uint64) (exists bool, err error) {
|
||||
txw.mu.Lock()
|
||||
defer txw.mu.Unlock()
|
||||
return txw.tx.Contains(string(fk), v)
|
||||
}
|
||||
|
||||
func (txw *rbfTxWrappers) Count(fk fragKey) (uint64, error) {
|
||||
txw.mu.Lock()
|
||||
defer txw.mu.Unlock()
|
||||
return txw.tx.Count(string(fk))
|
||||
}
|
||||
|
||||
func (txw *rbfTxWrappers) Max(fk fragKey) (uint64, error) {
|
||||
txw.mu.Lock()
|
||||
defer txw.mu.Unlock()
|
||||
return txw.tx.Max(string(fk))
|
||||
}
|
||||
|
||||
func (txw *rbfTxWrappers) Min(fk fragKey) (uint64, bool, error) {
|
||||
txw.mu.Lock()
|
||||
defer txw.mu.Unlock()
|
||||
return txw.tx.Min(string(fk))
|
||||
}
|
||||
|
||||
func (txw *rbfTxWrappers) CountRange(fk fragKey, start, end uint64) (uint64, error) {
|
||||
txw.mu.Lock()
|
||||
defer txw.mu.Unlock()
|
||||
return txw.tx.CountRange(string(fk), start, end)
|
||||
}
|
||||
|
||||
func (txw *rbfTxWrappers) RoaringBitmap(fk fragKey) (*roaring.Bitmap, error) {
|
||||
txw.mu.Lock()
|
||||
defer txw.mu.Unlock()
|
||||
return txw.tx.RoaringBitmap(string(fk))
|
||||
}
|
||||
|
||||
func (txw *rbfTxWrappers) OffsetRange(fk fragKey, offset, start, end uint64) (*roaring.Bitmap, error) {
|
||||
txw.mu.Lock()
|
||||
defer txw.mu.Unlock()
|
||||
return txw.tx.OffsetRange(string(fk), offset, start, end)
|
||||
}
|
||||
|
||||
func (txw *rbfTxWrappers) PutContainer(fk fragKey, ckey uint64, c *roaring.Container) error {
|
||||
txw.mu.Lock()
|
||||
defer txw.mu.Unlock()
|
||||
return txw.tx.PutContainer(string(fk), ckey, c)
|
||||
}
|
||||
|
||||
func (txw *rbfTxWrappers) RemoveContainer(fk fragKey, ckey uint64) error {
|
||||
txw.mu.Lock()
|
||||
defer txw.mu.Unlock()
|
||||
return txw.tx.RemoveContainer(string(fk), ckey)
|
||||
}
|
||||
|
||||
func (txw *rbfTxWrappers) Add(fk fragKey, a ...uint64) (changeCount int, err error) {
|
||||
txw.mu.Lock()
|
||||
defer txw.mu.Unlock()
|
||||
return txw.tx.Add(string(fk), a...)
|
||||
}
|
||||
|
||||
func (txw *rbfTxWrappers) Remove(fk fragKey, a ...uint64) (changeCount int, err error) {
|
||||
txw.mu.Lock()
|
||||
defer txw.mu.Unlock()
|
||||
return txw.tx.Remove(string(fk), a...)
|
||||
}
|
||||
|
||||
func (txw *rbfTxWrappers) ApplyRewriter(fk fragKey, ckey uint64, filter roaring.BitmapRewriter) (err error) {
|
||||
txw.mu.Lock()
|
||||
defer txw.mu.Unlock()
|
||||
return txw.tx.ApplyRewriter(string(fk), ckey, filter)
|
||||
}
|
||||
|
||||
func (txw *rbfTxWrappers) ImportRoaringBits(fk fragKey, 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.
|
||||
txw.mu.Lock()
|
||||
defer txw.mu.Unlock()
|
||||
return txw.tx.ImportRoaringBits(string(fk), rit, clear, false, rowSize)
|
||||
}
|
||||
|
||||
func (txw *rbfTxWrappers) Flush() error {
|
||||
if !txw.writeTx {
|
||||
return errors.New("attempt to flush non-write transaction")
|
||||
}
|
||||
txw.mu.Lock()
|
||||
defer txw.mu.Unlock()
|
||||
err := txw.tx.Commit()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
tx, err := txw.db.db.Begin(true)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "trying to obtain fresh transaction")
|
||||
}
|
||||
txw.tx = tx
|
||||
return nil
|
||||
}
|
||||
|
||||
// rbfQueryContext represents a query context backed by an rbfTxStore. It tracks
|
||||
// its current access to backend resources with a map[dbKey]*rbfTxWrappers.
|
||||
// For each dbKey it uses, it may have a Tx, which is always a single shared Tx
|
||||
|
|
@ -370,7 +967,20 @@ func (rq *rbfQueryContext) String() string {
|
|||
return fmt.Sprintf("rbf-Qcx<%s>%s", rq.name, rq.scope)
|
||||
}
|
||||
|
||||
func (rq *rbfQueryContext) NewRead(index IndexName, field FieldName, view ViewName, shard ShardID) (QueryRead, error) {
|
||||
// Flush only works correctly when we're using an index/shard split.
|
||||
func (rq *rbfQueryContext) Flush(index keys.Index, shard keys.Shard) error {
|
||||
rq.mu.Lock()
|
||||
defer rq.mu.Unlock()
|
||||
dbk, _ := rq.txStore.keys(index, "", "", shard)
|
||||
txw, ok := rq.queries[dbk]
|
||||
if ok {
|
||||
return txw.Flush()
|
||||
}
|
||||
// it's okay if we didn't have one.
|
||||
return nil
|
||||
}
|
||||
|
||||
func (rq *rbfQueryContext) Read(index keys.Index, field keys.Field, view keys.View, shard keys.Shard) (QueryRead, error) {
|
||||
rq.mu.Lock()
|
||||
defer rq.mu.Unlock()
|
||||
dbk, fk := rq.txStore.keys(index, field, view, shard)
|
||||
|
|
@ -394,7 +1004,7 @@ func (rq *rbfQueryContext) NewRead(index IndexName, field FieldName, view ViewNa
|
|||
return queries.readKey(fk)
|
||||
}
|
||||
|
||||
func (rq *rbfQueryContext) NewWrite(index IndexName, field FieldName, view ViewName, shard ShardID) (QueryWrite, error) {
|
||||
func (rq *rbfQueryContext) Write(index keys.Index, field keys.Field, view keys.View, shard keys.Shard) (QueryWrite, error) {
|
||||
if rq.scope == nil {
|
||||
return nil, errors.New("read-only query context can't write")
|
||||
}
|
||||
|
|
@ -572,4 +1182,8 @@ func (qw *rbfQueryWrite) ImportRoaringBits(rit roaring.RoaringIterator, clear bo
|
|||
return qw.tx.tx.ImportRoaringBits(string(qw.fk), rit, clear, false, rowSize)
|
||||
}
|
||||
|
||||
func (qw *rbfQueryWrite) Flush() (err error) {
|
||||
return qw.tx.Flush()
|
||||
}
|
||||
|
||||
var _ QueryWrite = &rbfQueryWrite{}
|
||||
|
|
|
|||
|
|
@ -11,7 +11,7 @@ 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, _ := q.Write("i", "f", "v", 0)
|
||||
_, _ = wr.Add(23, 25)
|
||||
_, _ = wr.Remove(25)
|
||||
// putting nil and removing a missing container are no-ops
|
||||
|
|
@ -35,7 +35,7 @@ func TestRbfWrite(t *testing.T) {
|
|||
// they're all one-line functions anyway.
|
||||
_ = q.Commit()
|
||||
q, _ = txs.NewQueryContext(ctx)
|
||||
rd, _ := q.NewRead("i", "f", "v", 0)
|
||||
rd, _ := q.Read("i", "f", "v", 0)
|
||||
ok, _ := rd.Contains(23)
|
||||
if !ok {
|
||||
t.Fatalf("no 23")
|
||||
|
|
|
|||
|
|
@ -5,6 +5,7 @@ import (
|
|||
"sync/atomic"
|
||||
"testing"
|
||||
|
||||
"github.com/molecula/featurebase/v3/keys"
|
||||
"github.com/molecula/featurebase/v3/roaring"
|
||||
)
|
||||
|
||||
|
|
@ -69,7 +70,7 @@ func (t *testTxStoreWrapper) dbPath(dbk dbKey) (string, error) {
|
|||
return p, t.oopsie(err)
|
||||
}
|
||||
|
||||
func (t *testTxStoreWrapper) keys(index IndexName, field FieldName, view ViewName, shard ShardID) (dbKey, fragKey) {
|
||||
func (t *testTxStoreWrapper) keys(index keys.Index, field keys.Field, view keys.View, shard keys.Shard) (dbKey, fragKey) {
|
||||
t.tb.Helper()
|
||||
return t.inner.keys(index, field, view, shard)
|
||||
}
|
||||
|
|
@ -103,15 +104,15 @@ func (t *testQueryContextWrapper) Errorf(msg string, args ...interface{}) {
|
|||
t.inner.Errorf(msg, args...)
|
||||
}
|
||||
|
||||
func (t *testQueryContextWrapper) NewRead(index IndexName, field FieldName, view ViewName, shard ShardID) (*testQueryReadWrapper, error) {
|
||||
func (t *testQueryContextWrapper) Read(index keys.Index, field keys.Field, view keys.View, shard keys.Shard) (*testQueryReadWrapper, error) {
|
||||
t.tb.Helper()
|
||||
qr, err := t.inner.NewRead(index, field, view, shard)
|
||||
qr, err := t.inner.Read(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) {
|
||||
func (t *testQueryContextWrapper) Write(index keys.Index, field keys.Field, view keys.View, shard keys.Shard) (*testQueryWriteWrapper, error) {
|
||||
t.tb.Helper()
|
||||
qw, err := t.inner.NewWrite(index, field, view, shard)
|
||||
qw, err := t.inner.Write(index, field, view, shard)
|
||||
return newTestQueryWriteWrapper(t.tb, qw), t.oopsie(err)
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -4,8 +4,12 @@ package querycontext
|
|||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"io"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
"github.com/molecula/featurebase/v3/keys"
|
||||
)
|
||||
|
||||
// TxStore represents a transactional database backend, mapping
|
||||
|
|
@ -30,10 +34,60 @@ type TxStore interface {
|
|||
// Close attempts to shut down. It can fail if there are still
|
||||
// open transactions.
|
||||
Close() error
|
||||
|
||||
// DeleteIndex deletes the specified index. This may imply closing and
|
||||
// deleting files. The close requires this to block until outstanding
|
||||
// reads against this index complete. Reads that refer to an index that's
|
||||
// being deleted may get partial results.
|
||||
DeleteIndex(keys.Index) error
|
||||
|
||||
// DeleteField deletes the specified field. As with DeleteIndex, behavior
|
||||
// when there are outstdanding reads may result in some reads seeing partial
|
||||
// data.
|
||||
DeleteField(keys.Index, keys.Field) error
|
||||
|
||||
// DeleteFragments deletes the specified views from the specified shards.
|
||||
// The implementation is encouraged to bundle write operations, for instance,
|
||||
// deleting all the views from each shard at once.
|
||||
DeleteFragments(keys.Index, keys.Field, []keys.View, []keys.Shard) error
|
||||
|
||||
// ListFieldViews requests a list of fields/view pairs represented in a
|
||||
// given index/shard. It is only sensical when using indexShardKeySplitter.
|
||||
ListFieldViews(keys.Index, keys.Shard) (map[keys.Field][]keys.View, error)
|
||||
|
||||
// Backend gives a textual identifier for the backend, such as "rbf".
|
||||
// Currently we only support one, but we have an API for exposing the
|
||||
// information and some day we may want another again.
|
||||
Backend() string
|
||||
|
||||
// DumpDot dumps a representation of the TxStore's current state in
|
||||
// graphviz dot format to the provided writer.
|
||||
DumpDot(w io.Writer) error
|
||||
|
||||
// Contents provides a hierarchical map of the TxStore's contents,
|
||||
// according to the logical hierarchy of the database rather than
|
||||
// whatever internal structure the TxStore uses.
|
||||
Contents() (keys.DBContents, error)
|
||||
}
|
||||
|
||||
// TxBackupRestore represents a TxStore that supports converting hunks of its
|
||||
// database to and from raw byte streams. Not every TxStore is a TxBackupRestore.
|
||||
type TxBackupRestore interface {
|
||||
// Backup provides an io.ReadCloser which contains the data in some format
|
||||
// compatible with Restore. For RBF, that would be "just an RBF file."
|
||||
// The backup reflects the state of the provided query context, and Backup
|
||||
// either closes that QueryContext on error, or returns a readcloser which
|
||||
// releases it on Close.
|
||||
Backup(QueryContext, keys.Index, keys.Shard) (io.ReadCloser, error)
|
||||
// Restore takes a reader containing data compatible with the backend,
|
||||
// such as an RBF file, and replaces all the existing data for this index
|
||||
// and shard. It does not release or close the QueryContext.
|
||||
Restore(QueryContext, keys.Index, keys.Shard, io.Reader) error
|
||||
}
|
||||
|
||||
// verify that rbfTxStore implements this interface
|
||||
var _ TxStore = &rbfTxStore{}
|
||||
var _ TxBackupRestore = &rbfTxStore{}
|
||||
|
||||
// dbKey is an identifier which can distinguish backend databases.
|
||||
type dbKey string
|
||||
|
|
@ -60,7 +114,19 @@ type KeySplitter interface {
|
|||
// backing database, and thus that they must share a backing
|
||||
// database transaction. The fragKey result is used by operations
|
||||
// within the database backend.
|
||||
keys(IndexName, FieldName, ViewName, ShardID) (dbKey, fragKey)
|
||||
keys(keys.Index, keys.Field, keys.View, keys.Shard) (dbKey, fragKey)
|
||||
|
||||
// dbKey provides just the dbKey half. exercise caution; it's
|
||||
// up to you to be sure you're varying this the right ways.
|
||||
dbKey(keys.Index, keys.Field, keys.View, keys.Shard) dbKey
|
||||
// fragKey provides just the fragKey half. exercise caution; it's
|
||||
// up to you to be sure you're varying this the right ways.
|
||||
fragKey(keys.Index, keys.Field, keys.View, keys.Shard) fragKey
|
||||
|
||||
// This is used essentially once, in the path where we have an existing
|
||||
// RBF file and want to know what it contains. We only support the case
|
||||
// where it's field/view right now.
|
||||
parseFragKey(fragKey) (keys.Index, keys.Field, keys.View, keys.Shard, error)
|
||||
|
||||
// dbPath yields a filesystem-friendly string that corresponds to dbKey.
|
||||
// possibly it is identical to dbKey, but you might want a terse dbKey
|
||||
|
|
@ -80,10 +146,34 @@ var _ KeySplitter = &flexibleKeySplitter{}
|
|||
// (the default for RBF).
|
||||
type indexShardKeySplitter struct{}
|
||||
|
||||
func (*indexShardKeySplitter) keys(index IndexName, field FieldName, view ViewName, shard ShardID) (dbKey, fragKey) {
|
||||
d := dbKey(fmt.Sprintf("%s/%08x", index, shard))
|
||||
t := fragKey(fmt.Sprintf("%s:%s", field, view))
|
||||
return d, t
|
||||
// Compatibility note: The existing database format uses this as its proposed name for the directory for a database:
|
||||
// path := dbs.HolderPath + sep + dbs.Index + sep + backendsDir + sep + ty.DirectoryName() + sep + fmt.Sprintf("shard.%04v", dbs.Shard)
|
||||
// and the field/view keys used in the database are:
|
||||
// ~field;view<
|
||||
// we're using these for compatibility.
|
||||
|
||||
func (i *indexShardKeySplitter) keys(index keys.Index, field keys.Field, view keys.View, shard keys.Shard) (dbKey, fragKey) {
|
||||
return i.dbKey(index, field, view, shard), i.fragKey(index, field, view, shard)
|
||||
}
|
||||
|
||||
func (*indexShardKeySplitter) dbKey(index keys.Index, field keys.Field, view keys.View, shard keys.Shard) dbKey {
|
||||
return dbKey(fmt.Sprintf("%s/%08x", index, shard))
|
||||
}
|
||||
|
||||
func (*indexShardKeySplitter) fragKey(index keys.Index, field keys.Field, view keys.View, shard keys.Shard) fragKey {
|
||||
return fragKey(fmt.Sprintf("~%s;%s<", field, view))
|
||||
}
|
||||
|
||||
func (*indexShardKeySplitter) parseFragKey(fk fragKey) (keys.Index, keys.Field, keys.View, keys.Shard, error) {
|
||||
l := len(fk)
|
||||
if l < 3 || fk[0] != '~' || fk[l-1] != '<' {
|
||||
return "", "", "", 0, fmt.Errorf("malformed frag key %q", fk)
|
||||
}
|
||||
semi := strings.IndexByte(string(fk), ';')
|
||||
if semi == -1 {
|
||||
return "", "", "", 0, fmt.Errorf("malformed frag key %q", fk)
|
||||
}
|
||||
return "", keys.Field(fk[1:semi]), keys.View(fk[semi+1 : l-1]), 0, nil
|
||||
}
|
||||
|
||||
func (*indexShardKeySplitter) Scope() QueryScope {
|
||||
|
|
@ -95,19 +185,35 @@ func (*indexShardKeySplitter) dbPath(dbk dbKey) (string, error) {
|
|||
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:]
|
||||
index := string(dbk)[:slash]
|
||||
shardHex := string(dbk)[slash+1:]
|
||||
shardNum, err := strconv.ParseInt(shardHex, 16, 64)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("invalid hex shard number in dbKey %q", dbk)
|
||||
}
|
||||
// the %04d here is a compatibility thing with old short_txkey and should
|
||||
// probably be fixed.
|
||||
paths := [5]string{index, "backends", "rbf", "shard." + fmt.Sprintf("%04d", shardNum)}
|
||||
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/%08x", index, field, shard))
|
||||
t := fragKey(view)
|
||||
return d, t
|
||||
func (f *fieldShardKeySplitter) keys(index keys.Index, field keys.Field, view keys.View, shard keys.Shard) (dbKey, fragKey) {
|
||||
return f.dbKey(index, field, view, shard), f.fragKey(index, field, view, shard)
|
||||
}
|
||||
|
||||
func (*fieldShardKeySplitter) dbKey(index keys.Index, field keys.Field, view keys.View, shard keys.Shard) dbKey {
|
||||
return dbKey(fmt.Sprintf("%s/%s/%08x", index, field, shard))
|
||||
}
|
||||
|
||||
func (*fieldShardKeySplitter) fragKey(index keys.Index, field keys.Field, view keys.View, shard keys.Shard) fragKey {
|
||||
return fragKey(view)
|
||||
}
|
||||
|
||||
func (*fieldShardKeySplitter) parseFragKey(fk fragKey) (keys.Index, keys.Field, keys.View, keys.Shard, error) {
|
||||
return "", "", "", 0, errUnimplemented
|
||||
}
|
||||
|
||||
func (*fieldShardKeySplitter) dbPath(dbk dbKey) (string, error) {
|
||||
|
|
@ -139,34 +245,54 @@ func (*fieldShardKeySplitter) Scope() QueryScope {
|
|||
// exists primarily to explore the API space. Do not alter the splitIndexes
|
||||
// map after initial creation, it will produce inconsistent results.
|
||||
type flexibleKeySplitter struct {
|
||||
splitIndexes map[IndexName]struct{}
|
||||
splitIndexes map[keys.Index]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))
|
||||
func NewFlexibleKeySplitter(indexes ...keys.Index) *flexibleKeySplitter {
|
||||
splitIndexes := make(map[keys.Index]struct{}, len(indexes))
|
||||
for _, index := range indexes {
|
||||
splitIndexes[index] = struct{}{}
|
||||
}
|
||||
return &flexibleKeySplitter{splitIndexes: splitIndexes}
|
||||
}
|
||||
|
||||
func (f *flexibleKeySplitter) keys(index IndexName, field FieldName, view ViewName, shard ShardID) (dbKey, fragKey) {
|
||||
func (f *flexibleKeySplitter) keys(index keys.Index, field keys.Field, view keys.View, shard keys.Shard) (dbKey, fragKey) {
|
||||
if _, ok := f.splitIndexes[index]; ok {
|
||||
return (&fieldShardKeySplitter{}).keys(index, field, view, shard)
|
||||
}
|
||||
return (&indexShardKeySplitter{}).keys(index, field, view, shard)
|
||||
}
|
||||
|
||||
func (f *flexibleKeySplitter) dbKey(index keys.Index, field keys.Field, view keys.View, shard keys.Shard) dbKey {
|
||||
if _, ok := f.splitIndexes[index]; ok {
|
||||
return (&fieldShardKeySplitter{}).dbKey(index, field, view, shard)
|
||||
}
|
||||
return (&indexShardKeySplitter{}).dbKey(index, field, view, shard)
|
||||
}
|
||||
|
||||
func (f *flexibleKeySplitter) fragKey(index keys.Index, field keys.Field, view keys.View, shard keys.Shard) fragKey {
|
||||
if _, ok := f.splitIndexes[index]; ok {
|
||||
return (&fieldShardKeySplitter{}).fragKey(index, field, view, shard)
|
||||
}
|
||||
return (&indexShardKeySplitter{}).fragKey(index, field, view, shard)
|
||||
}
|
||||
|
||||
// unimplemented because we don't want to figure out what the fragkey implies, and
|
||||
// in any event we only ever get called with the index style ones
|
||||
func (*flexibleKeySplitter) parseFragKey(fk fragKey) (keys.Index, keys.Field, keys.View, keys.Shard, error) {
|
||||
return "", "", "", 0, errUnimplemented
|
||||
}
|
||||
|
||||
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 {
|
||||
if _, ok := f.splitIndexes[keys.Index(dbk)[:slash]]; ok {
|
||||
return (&fieldShardKeySplitter{}).dbPath(dbk)
|
||||
}
|
||||
return (&indexShardKeySplitter{}).dbPath(dbk)
|
||||
|
|
@ -175,3 +301,83 @@ func (f *flexibleKeySplitter) dbPath(dbk dbKey) (string, error) {
|
|||
func (f *flexibleKeySplitter) Scope() QueryScope {
|
||||
return &flexibleQueryScope{splitter: f}
|
||||
}
|
||||
|
||||
type unimplementedError struct{}
|
||||
|
||||
func (unimplementedError) Error() string {
|
||||
return "unimplemented"
|
||||
}
|
||||
|
||||
var errUnimplemented unimplementedError
|
||||
|
||||
type noValidStoreError struct{}
|
||||
|
||||
func (noValidStoreError) Error() string {
|
||||
return "no valid TxStore selected"
|
||||
}
|
||||
|
||||
var errNoValidStore noValidStoreError
|
||||
|
||||
var _ TxStore = &nopTxStore{}
|
||||
|
||||
// nopTxStore is a TxStore that always fails and won't let you do anything.
|
||||
type nopTxStore struct {
|
||||
indexShardKeySplitter
|
||||
}
|
||||
|
||||
var NopTxStore = &nopTxStore{}
|
||||
|
||||
// NewQueryContext yields a new query context which is read-only.
|
||||
func (*nopTxStore) NewQueryContext(context.Context) (QueryContext, error) {
|
||||
return nil, errNoValidStore
|
||||
}
|
||||
|
||||
// NewWriteQueryContext yields a new query context which can
|
||||
// write to the things in the given QueryScope
|
||||
func (*nopTxStore) NewWriteQueryContext(context.Context, QueryScope) (QueryContext, error) {
|
||||
return nil, errNoValidStore
|
||||
}
|
||||
|
||||
func (*nopTxStore) DumpDot(w io.Writer) error {
|
||||
return errNoValidStore
|
||||
}
|
||||
|
||||
func (*nopTxStore) Backend() string {
|
||||
return "none"
|
||||
}
|
||||
|
||||
// Close attempts to shut down. It can fail if there are still
|
||||
// open transactions.
|
||||
func (*nopTxStore) Close() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// DeleteIndex deletes the specified index. This may imply closing and
|
||||
// deleting files. The close requires this to block until outstanding
|
||||
// reads against this index complete. Reads that refer to an index that's
|
||||
// being deleted may get partial results.
|
||||
func (*nopTxStore) DeleteIndex(keys.Index) error {
|
||||
return errNoValidStore
|
||||
}
|
||||
|
||||
// DeleteField deletes the specified field. As with DeleteIndex, behavior
|
||||
// when there are outstdanding reads may result in some reads seeing partial
|
||||
// data.
|
||||
func (*nopTxStore) DeleteField(keys.Index, keys.Field) error {
|
||||
return errNoValidStore
|
||||
}
|
||||
|
||||
// DeleteFragments deletes the specified views from the specified shards.
|
||||
// The implementation is encouraged to bundle write operations, for instance,
|
||||
// deleting all the views from each shard at once.
|
||||
func (*nopTxStore) DeleteFragments(keys.Index, keys.Field, []keys.View, []keys.Shard) error {
|
||||
return errNoValidStore
|
||||
}
|
||||
|
||||
func (*nopTxStore) ListFieldViews(keys.Index, keys.Shard) (map[keys.Field][]keys.View, error) {
|
||||
return nil, errNoValidStore
|
||||
}
|
||||
|
||||
func (*nopTxStore) Contents() (keys.DBContents, error) {
|
||||
return nil, errNoValidStore
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue