From 75a68cf60c9482cf4d5a22dc279c009ba4d91500 Mon Sep 17 00:00:00 2001 From: Seebs Date: Wed, 11 Jan 2023 11:56:26 -0600 Subject: [PATCH] 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. --- querycontext/dot_test.go | 18 +- querycontext/query_context.go | 126 +++--- querycontext/query_context_test.go | 47 ++- querycontext/rbf.go | 658 ++++++++++++++++++++++++++++- querycontext/rbf_test.go | 4 +- querycontext/reduce_err_test.go | 11 +- querycontext/txstore.go | 240 ++++++++++- 7 files changed, 980 insertions(+), 124 deletions(-) diff --git a/querycontext/dot_test.go b/querycontext/dot_test.go index 8a84349a8..53c711eed 100644 --- a/querycontext/dot_test.go +++ b/querycontext/dot_test.go @@ -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 diff --git a/querycontext/query_context.go b/querycontext/query_context.go index 6331d24e0..5f47681e5 100644 --- a/querycontext/query_context.go +++ b/querycontext/query_context.go @@ -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 { diff --git a/querycontext/query_context_test.go b/querycontext/query_context_test.go index 7ed5585bd..748cfd587 100644 --- a/querycontext/query_context_test.go +++ b/querycontext/query_context_test.go @@ -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() +} diff --git a/querycontext/rbf.go b/querycontext/rbf.go index da44596a2..4244cc478 100644 --- a/querycontext/rbf.go +++ b/querycontext/rbf.go @@ -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{} diff --git a/querycontext/rbf_test.go b/querycontext/rbf_test.go index ba39ba4a7..d03e9e110 100644 --- a/querycontext/rbf_test.go +++ b/querycontext/rbf_test.go @@ -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") diff --git a/querycontext/reduce_err_test.go b/querycontext/reduce_err_test.go index 6b71800ac..dc54d3144 100644 --- a/querycontext/reduce_err_test.go +++ b/querycontext/reduce_err_test.go @@ -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) } diff --git a/querycontext/txstore.go b/querycontext/txstore.go index 114b97cb0..772cf87fa 100644 --- a/querycontext/txstore.go +++ b/querycontext/txstore.go @@ -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 +}