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