scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-10 15:01:03 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Fix RBF/pilosa hanging tests

Browse source
This commit is contained in:
Ben Johnson 2020-08-06 10:33:37 -06:00
parent 5306426b19
commit 78f9fbe332
11 changed files with 37 additions and 49 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

27
api.go
View file

@ -1066,10 +1066,6 @@ func (api *API) Import(ctx context.Context, req *ImportRequest, opts ...ImportOp
return errors.Wrap(err, "getting index and field")
}
// Obtain transaction.
tx := index.Txf.NewTx(Txo{Write: true, Index: index})
defer tx.Rollback()
if err := req.ValidateWithTimestamp(index.CreatedAt(), field.CreatedAt()); err != nil {
return errors.Wrap(err, "validating import value request")
}
@ -1162,12 +1158,23 @@ func (api *API) Import(ctx context.Context, req *ImportRequest, opts ...ImportOp
// Import columnIDs into existence field.
if !options.Clear {
if err := importExistenceColumns(tx, index, req.ColumnIDs); err != nil {
api.server.logger.Printf("import existence error: index=%s, field=%s, shard=%d, columns=%d, err=%s", req.Index, req.Field, req.Shard, len(req.ColumnIDs), err)
if err := func() error {
tx := index.Txf.NewTx(Txo{Write: true, Index: index})
defer tx.Rollback()
if err := importExistenceColumns(tx, index, req.ColumnIDs); err != nil {
api.server.logger.Printf("import existence error: index=%s, field=%s, shard=%d, columns=%d, err=%s", req.Index, req.Field, req.Shard, len(req.ColumnIDs), err)
return err
}
return tx.Commit()
}(); err != nil {
return errors.Wrap(err, "importing existence columns")
}
}
tx := index.Txf.NewTx(Txo{Write: true, Index: index})
defer tx.Rollback()
// Import into fragment.
err = field.Import(tx, req.RowIDs, req.ColumnIDs, timestamps, opts...)
if err != nil {
@ -1198,10 +1205,6 @@ func (api *API) ImportValue(ctx context.Context, req *ImportValueRequest, opts .
return errors.Wrap(err, "validating import value request")
}
// Obtain transaction.
tx := index.Txf.NewTx(Txo{Write: true, Index: index})
defer tx.Rollback()
// Set up import options.
options, err := setUpImportOptions(opts...)
if err != nil {
@ -1257,6 +1260,10 @@ func (api *API) ImportValue(ctx context.Context, req *ImportValueRequest, opts .
// if we're importing into a specific shard
if req.Shard != math.MaxUint64 {
// Obtain transaction.
tx := index.Txf.NewTx(Txo{Write: true, Index: index})
defer tx.Rollback()
// Check that column IDs match the stated shard.
if s1, s2 := req.ColumnIDs[0]/ShardWidth, req.ColumnIDs[len(req.ColumnIDs)-1]/ShardWidth; s1 != s2 && s2 != req.Shard {
return errors.Errorf("shard %d specified, but import spans shards %d to %d", req.Shard, s1, s2)

4
api_test.go
View file

@ -165,8 +165,6 @@ func TestAPI_ImportColumnAttrs(t *testing.T) {
}
func TestAPI_Import(t *testing.T) {
skipForRBF(t)
c := test.MustRunCluster(t, 2,
[]server.CommandOption{
server.OptCommandServerOptions(
@ -278,8 +276,6 @@ func TestAPI_Import(t *testing.T) {
}
func TestAPI_ImportValue(t *testing.T) {
skipForRBF(t)
c := test.MustRunCluster(t, 2,
[]server.CommandOption{
server.OptCommandServerOptions(

7
cluster.go
View file

@ -1590,8 +1590,11 @@ func (c *cluster) followResizeInstruction(instr *ResizeInstruction) error {
}
// Create view.
v, err := f.createViewIfNotExists(src.View)
if err != nil {
var v *view
if err := func() (err error) {
v, err = f.createViewIfNotExists(src.View)
return err
}(); err != nil {
return errors.Wrap(err, "creating view")
}

6
executor_test.go
View file

@ -3633,8 +3633,6 @@ func TestExecutor_Execute_FieldValue(t *testing.T) {
// Ensure an all query can be executed.
func TestExecutor_Execute_All(t *testing.T) {
skipForRBF(t)
t.Run("ColumnID", func(t *testing.T) {
c := test.MustRunCluster(t, 1)
defer c.Close()
@ -4395,8 +4393,6 @@ func TestExecutor_Execute_Query_Error(t *testing.T) {
}
func TestExecutor_GroupByStrings(t *testing.T) {
skipForRBF(t)
c := test.MustRunCluster(t, 1)
defer c.Close()
c.CreateField(t, "istring", pilosa.IndexOptions{Keys: true}, "generals", pilosa.OptFieldKeys())
@ -4853,8 +4849,6 @@ func sameStringSlice(x, y []string) bool {
}
func TestExecutor_Execute_GroupBy(t *testing.T) {
skipForRBF(t)
groupByTest := func(t *testing.T, clusterSize int) {
c := test.MustRunCluster(t, 1)
defer c.Close()

1
field.go
View file

@ -748,6 +748,7 @@ fileLoop:
}()
name := filepath.Base(fi.Name())
f.holder.Logger.Debugf("open index/field/view: %s/%s/%s", f.index, f.name, fi.Name())
view := f.newView(f.viewPath(name), name)
if err := view.open(); err != nil {
return fmt.Errorf("opening view: view=%s, err=%s", view.name, err)

2
fragment.go
View file

@ -258,7 +258,7 @@ func (f *fragment) Open() error {
f.checksums = make(map[int][]byte)
// Read last bit to determine max row.
tx := f.idx.Txf.NewTx(Txo{Write: !writable, Index: f.idx, Fragment: f})
tx := f.idx.Txf.NewTx(Txo{Write: false, Index: f.idx, Fragment: f})
defer tx.Rollback()
return f.calculateMaxRowID(tx)
}(); err != nil {

10
fragment_internal_test.go
View file

@ -1604,10 +1604,8 @@ func TestFragment_TopN_CacheSize(t *testing.T) {
defer f.Clean(t)
// Obtain transaction.
tx := index.Txf.NewTx(Txo{Write: writable, Index: index, Fragment: f})
f.txTestingOnly = tx
defer tx.Rollback() // okay to call 2x. f.Clean(t) will do Rollback() too.
tx := index.Txf.NewTx(Txo{Write: writable, Index: index})
defer tx.Rollback()
// Set bits on various rows.
f.mustSetBits(tx, 100, 1, 2, 3)
@ -1812,7 +1810,7 @@ func TestFragment_RankCache_Persistence(t *testing.T) {
}
// Obtain transaction.
tx := index.Txf.NewTx(Txo{Write: writable, Index: index, Fragment: f})
tx := index.Txf.NewTx(Txo{Write: writable, Index: index})
defer tx.Rollback()
// Set bits on the fragment.
@ -5324,8 +5322,6 @@ func check(t *testing.T, tx Tx, f *fragment, exp map[uint64]map[uint64]struct{})
}
func TestImportValueConcurrent(t *testing.T) {
skipForRBF(t)
f, idx := mustOpenBSIFragment("i", "f", viewBSIGroupPrefix+"foo", 0)
switch idx.Txf.TxType() {
case blueGreenBadgerRoaring, blueGreenRoaringBadger:

7
pilosa_test.go
View file

@ -15,7 +15,6 @@
package pilosa_test
import (
"os"
"strings"
"testing"
@ -56,9 +55,3 @@ func TestAddressWithDefaults(t *testing.T) {
}
}
}
func skipForRBF(tb testing.TB) {
if os.Getenv("PILOSA_TXSRC") == "rbf" {
tb.Skip("skip for RBF")
}
}

16
rbf/db.go
View file

@ -424,13 +424,13 @@ func (db *DB) addWALSegment() error {
func (db *DB) Close() (err error) {
// TODO(bbj): Add wait group to hang until last Tx is complete.
db.mu.Lock()
defer db.mu.Unlock()
// Wait for writer lock.
db.rwmu.Lock()
defer db.rwmu.Unlock()
db.mu.Lock()
defer db.mu.Unlock()
db.opened = false
// Close mmap handle.
@ -546,6 +546,11 @@ func (db *DB) initFreelistPage() error {
func (db *DB) Begin(writable bool) (_ *Tx, err error) {
// TODO(BBJ): Acquire write lock if writable.
// Ensure only one writable transaction at a time.
if writable {
db.rwmu.Lock()
}
db.mu.Lock()
defer db.mu.Unlock()
@ -555,11 +560,6 @@ func (db *DB) Begin(writable bool) (_ *Tx, err error) {
tx := &Tx{db: db, pageMap: db.pageMap, writable: writable}
// Ensure only one writable transaction at a time.
if tx.writable {
db.rwmu.Lock()
}
// Copy meta page into transaction's buffer.
// This page is only written at the end of a dirty transaction.
page, err := db.readPage(db.pageMap, 0)

3
server.go
View file

@ -720,8 +720,7 @@ func (s *Server) receiveMessage(m Message) error {
if f == nil {
return fmt.Errorf("local field not found: %s", obj.Field)
}
_, _, err := f.createViewIfNotExistsBase(obj.View)
if err != nil {
if _, _, err := f.createViewIfNotExistsBase(obj.View); err != nil {
return err
}
case *DeleteViewMessage:

3
view.go
View file

@ -166,8 +166,7 @@ var workQueue = make(chan struct{}, runtime.NumCPU()*2)
// replaces v.openFragments() with Tx generic code.
func (v *view) openFragmentsInTx() error {
tx := v.idx.Txf.NewTx(Txo{Write: !writable, Index: v.idx})
tx := v.idx.Txf.NewTx(Txo{Write: false, Index: v.idx})
defer tx.Rollback()
shards, err := tx.SliceOfShards(v.index, v.field, v.name, v.path)