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

Merge branch 'master' into limit-cluster

Browse source
This commit is contained in:
Nia 2020-09-23 13:46:02 -04:00 • committed by GitHub
commit e0f25f1d20
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
12 changed files with 265 additions and 76 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
api.go
View file

@ -1827,7 +1827,7 @@ func (api *API) TranslateIDs(ctx context.Context, r io.Reader) (_ []byte, err er
if err != nil {
return nil, err
}
} else if keys, err = field.TranslateStore().TranslateIDs(req.IDs); err != nil {
} else if keys, err = api.cluster.translateFieldListIDs(field, req.IDs); err != nil {
return nil, err
}
}

3
badger.go
View file

@ -31,6 +31,7 @@ import (
badger "github.com/dgraph-io/badger/v2"
badgeroptions "github.com/dgraph-io/badger/v2/options"
"github.com/pilosa/pilosa/v2/hash"
"github.com/pilosa/pilosa/v2/rbf"
"github.com/pilosa/pilosa/v2/roaring"
"github.com/pilosa/pilosa/v2/testhook"
"github.com/pilosa/pilosa/v2/txkey"
@ -660,7 +661,7 @@ func (tx *BadgerTx) Type() string {
func (tx *BadgerTx) UseRowCache() bool {
//the row cache speeds up queries.
return false
return rbf.EnableRowCache
}
// overWriteOurAllocs provides detection of memory

40
cluster.go
View file

@ -2377,6 +2377,46 @@ func (c *cluster) translateFieldKeys(ctx context.Context, field *Field, keys []s
return ids, nil
}
func (c *cluster) translateFieldIDs(field *Field, ids map[uint64]struct{}) (map[uint64]string, error) {
idList := make([]uint64, len(ids))
{
i := 0
for id := range ids {
idList[i] = id
i++
}
}
keyList, err := c.translateFieldListIDs(field, idList)
if err != nil {
return nil, err
}
mapped := make(map[uint64]string, len(idList))
for i, key := range keyList {
mapped[idList[i]] = key
}
return mapped, nil
}
func (c *cluster) translateFieldListIDs(field *Field, ids []uint64) (keys []string, err error) {
coordinator := c.coordinatorNode()
if coordinator == nil {
return nil, errors.Errorf("translating field(%s/%s) ids(%v) - cannot find coordinator node", field.Index(), field.Name(), ids)
}
if c.Node.ID == coordinator.ID {
keys, err = field.TranslateStore().TranslateIDs(ids)
} else {
keys, err = c.InternalClient.TranslateIDsNode(context.Background(), &coordinator.URI, field.Index(), field.Name(), ids)
}
if err != nil {
return nil, errors.Wrapf(err, "translating field(%s/%s) ids(%v)", field.Index(), field.Name(), ids)
}
return keys, err
}
func (c *cluster) translateIndexKey(ctx context.Context, indexName string, key string, writable bool) (uint64, error) {
keyMap, err := c.translateIndexKeySet(ctx, indexName, map[string]struct{}{key: struct{}{}}, writable)
if err != nil {

28
executor.go
View file

@ -5090,26 +5090,6 @@ func (e *executor) collectResultIDs(index string, idx *Index, call *pql.Call, re
return nil
}
func (e *executor) translateFieldIDs(field *Field, ids map[uint64]struct{}) (map[uint64]string, error) {
idList := make([]uint64, len(ids))
{
i := 0
for id := range ids {
idList[i] = id
i++
}
}
keyList, err := field.TranslateStore().TranslateIDs(idList)
if err != nil {
return nil, err
}
mapped := make(map[uint64]string, len(idList))
for i, key := range keyList {
mapped[idList[i]] = key
}
return mapped, nil
}
// preTranslateMatrixSet translates the IDs of a set field in an extracted matrix.
func (e *executor) preTranslateMatrixSet(mat ExtractedIDMatrix, fieldIdx uint, field *Field) (map[uint64]string, error) {
ids := make(map[uint64]struct{}, len(mat.Columns))
@ -5119,7 +5099,7 @@ func (e *executor) preTranslateMatrixSet(mat ExtractedIDMatrix, fieldIdx uint, f
}
}
return e.translateFieldIDs(field, ids)
return e.Cluster.translateFieldIDs(field, ids)
}
func (e *executor) translateResult(ctx context.Context, index string, idx *Index, call *pql.Call, result interface{}, idSet map[uint64]string) (_ interface{}, err error) {
@ -5205,7 +5185,7 @@ func (e *executor) translateResult(ctx context.Context, index string, idx *Index
for i := range result.Pairs {
ids[i] = result.Pairs[i].ID
}
keys, err := field.TranslateStore().TranslateIDs(ids)
keys, err := e.Cluster.translateFieldListIDs(field, ids)
if err != nil {
return nil, err
}
@ -5256,7 +5236,7 @@ func (e *executor) translateResult(ctx context.Context, index string, idx *Index
fieldTranslations := make(map[string]map[uint64]string)
for field, ids := range fieldIDs {
trans, err := e.translateFieldIDs(field, ids)
trans, err := e.Cluster.translateFieldIDs(field, ids)
if err != nil {
return nil, errors.Wrapf(err, "translating IDs in field %q", field.Name())
}
@ -5308,7 +5288,7 @@ func (e *executor) translateResult(ctx context.Context, index string, idx *Index
if field := idx.Field(fieldName); field == nil {
return nil, newNotFoundError(ErrFieldNotFound, fieldName)
} else if field.Keys() {
keys, err := field.TranslateStore().TranslateIDs(result)
keys, err := e.Cluster.translateFieldListIDs(field, result)
if err != nil {
return nil, errors.Wrap(err, "translating row IDs")
}

82
lmdb.go
View file

@ -33,6 +33,7 @@ import (
"github.com/glycerine/lmdb-go/lmdb"
"github.com/pilosa/pilosa/v2/hash"
"github.com/pilosa/pilosa/v2/rbf"
"github.com/pilosa/pilosa/v2/roaring"
"github.com/pilosa/pilosa/v2/txkey"
"github.com/pkg/errors"
@ -274,10 +275,7 @@ func (w *LMDBWrapper) CleanupTx(tx Tx) {
}
func (tx *LMDBTx) IsDone() (done bool) {
tx.mu.Lock()
done = tx.unlocked
tx.mu.Unlock()
return
return atomic.LoadInt64(&tx.unlocked) == 1
}
func (w *LMDBWrapper) OpenListString() (r string) {
@ -434,9 +432,11 @@ func (w *LMDBWrapper) NewTx(write bool, initialIndexName string, o Txo) (tx Tx,
}
tx = ltx
w.muDb.Lock()
w.openTx[ltx] = true
w.muDb.Unlock()
if isDebugRun {
w.muDb.Lock()
w.openTx[ltx] = true
w.muDb.Unlock()
}
return
}
@ -445,13 +445,14 @@ func (w *LMDBWrapper) Close() (err error) {
w.muDb.Lock()
defer w.muDb.Unlock()
if !w.closed {
// complain if there are still Tx in flight, b/c otherwise we will see
// the somewhat mysterious 'panic: should not be in ReadSlot.free() with slot still owned by gid=107043; refCount=1'
if len(w.openTx) > 0 {
AlwaysPrintf("error: cannot close LMDBWrapper with Tx still in flight.")
return
if isDebugRun {
// complain if there are still Tx in flight, b/c otherwise we will see
// the somewhat mysterious 'panic: should not be in ReadSlot.free() with slot still owned by gid=107043; refCount=1'
if len(w.openTx) > 0 {
AlwaysPrintf("error: cannot close LMDBWrapper with Tx still in flight.")
return
}
}
w.reg.unregister(w)
w.closed = true
w.env.CloseDBI(w.dbi)
@ -494,7 +495,7 @@ type LMDBTx struct {
DeleteEmptyContainer bool
unlocked bool // runtime.UnlockOSThread has been done.
unlocked int64 // runtime.UnlockOSThread has been done if > 0
o Txo
@ -535,7 +536,7 @@ func (tx *LMDBTx) Type() string {
}
func (tx *LMDBTx) UseRowCache() bool {
return false
return rbf.EnableRowCache
}
// Pointer gives us a memory address for the underlying transaction for debugging.
@ -545,15 +546,21 @@ func (tx *LMDBTx) Pointer() string {
return fmt.Sprintf("%p", tx)
}
const isDebugRun = false
// Rollback rolls back the transaction.
func (tx *LMDBTx) Rollback() {
tx.sanity()
alreadyDone := atomic.CompareAndSwapInt64(&tx.unlocked, 0, 1)
if !alreadyDone {
return
}
//vv("lmdb rollback tx _sn_ %v; stack \n%v", tx.sn) // , stack())
tx.Db.muDb.Lock()
delete(tx.Db.openTx, tx)
tx.Db.muDb.Unlock()
if isDebugRun {
tx.sanity()
tx.Db.muDb.Lock()
delete(tx.Db.openTx, tx)
tx.Db.muDb.Unlock()
}
tx.mu.Lock()
defer tx.mu.Unlock()
@ -561,25 +568,26 @@ func (tx *LMDBTx) Rollback() {
//tx.debugOnlyGidcheck()
tx.tx.Abort() // must hold tx.mu mutex lock
if !tx.unlocked {
runtime.UnlockOSThread()
tx.unlocked = true
tx.o.dbs.Cleanup(tx)
}
// use CAS above instead of testing a bool unlocked.
runtime.UnlockOSThread()
tx.o.dbs.Cleanup(tx)
}
// Commit commits the transaction to permanent storage.
// Commits can handle up to 100k updates to fragments
// at once, but not more. This is a LMDBDB imposed limit.
func (tx *LMDBTx) Commit() error {
tx.sanity()
alreadyDone := atomic.CompareAndSwapInt64(&tx.unlocked, 0, 1)
if !alreadyDone {
return nil
}
//vv("lmdb commit tx _sn_ %v; stack \n%v", tx.sn, stack())
tx.Db.muDb.Lock()
delete(tx.Db.openTx, tx)
tx.Db.muDb.Unlock()
if isDebugRun {
tx.sanity()
tx.Db.muDb.Lock()
delete(tx.Db.openTx, tx)
tx.Db.muDb.Unlock()
}
tx.mu.Lock()
defer tx.mu.Unlock()
@ -587,11 +595,9 @@ func (tx *LMDBTx) Commit() error {
err := tx.tx.Commit() // must hold tx.mu mutex lock
panicOn(err)
if !tx.unlocked {
runtime.UnlockOSThread()
tx.unlocked = true
tx.o.dbs.Cleanup(tx)
}
// replace the if !tx.unlocked with the CAS on tx.unlocked above.
runtime.UnlockOSThread()
tx.o.dbs.Cleanup(tx)
return err
}

37
pprof.go
View file

@ -29,7 +29,7 @@ func CPUProfileForDur(dur time.Duration, outpath string) {
// per-query pprof output:
txsrc := os.Getenv("PILOSA_TXSRC")
if txsrc == "" {
txsrc = "roaring"
txsrc = DefaultTxsrc
}
path := outpath + "." + txsrc
f, err := os.Create(path)
@ -53,7 +53,7 @@ func MemProfileForDur(dur time.Duration, outpath string) {
// per-query pprof output:
txsrc := os.Getenv("PILOSA_TXSRC")
if txsrc == "" {
txsrc = "roaring"
txsrc = DefaultTxsrc
}
path := outpath + "." + txsrc
f, err := os.Create(path)
@ -73,3 +73,36 @@ func MemProfileForDur(dur time.Duration, outpath string) {
AlwaysPrintf("wrote memory profile after dur '%v', output: '%v'", dur, path)
}()
}
type pprofProfile struct {
fdCpu *os.File
}
var _ = newPprof
var _ = pprofProfile{}
// for manually calling Close() to stop profiling.
func newPprof() (pp *pprofProfile) {
pp = &pprofProfile{}
f, err := os.Create("cpu.manual.pprof")
panicOn(err)
pp.fdCpu = f
_ = pprof.StartCPUProfile(pp.fdCpu)
return
}
func (pp *pprofProfile) Close() {
pprof.StopCPUProfile()
pp.fdCpu.Close()
f, err := os.Create("mem.manual.pprof")
panicOn(err)
runtime.GC() // get up-to-date statistics
if err := pprof.WriteHeapProfile(f); err != nil {
panic(fmt.Sprintf("could not write memory profile: %v", err))
}
f.Close()
}

2
rbf.go
View file

@ -68,7 +68,7 @@ func (w *RbfDBWrapper) CleanupTx(tx Tx) {
r.done = true
r.mu.Unlock()
// try not to old r.mu while locking w.muDb
// try not to hold r.mu while locking w.muDb
w.muDb.Lock()
delete(w.openTx, r)

2
rbf/cursorx.go
View file

@ -26,7 +26,7 @@ import (
// if enableRowCache, then we must not return mmap-ed memory
// directly, but only a copy.
const EnableRowCache = false
const EnableRowCache = true
// DoAllocZero means we copy mmap read data and
// wipe it afterwards to catch retention of data

3
rrtx.go
View file

@ -25,6 +25,7 @@ import (
"sync"
"sync/atomic"
"github.com/pilosa/pilosa/v2/rbf"
"github.com/pilosa/pilosa/v2/roaring"
"github.com/pkg/errors"
)
@ -62,7 +63,7 @@ func (tx *RoaringTx) Dump(short bool, shard uint64) {
}
func (tx *RoaringTx) UseRowCache() bool {
return false
return rbf.EnableRowCache
}
func (tx *RoaringTx) SliceOfShards(index, field, view, optionalViewPath string) (sliceOfShards []uint64, err error) {

18
server/grpc.go
View file

@ -557,7 +557,13 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Uint64ArrayVal{Uint64ArrayVal: &pb.Uint64Array{Vals: ids.Rows}}})
colAdded++
} else {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: nil})
}
} else {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: nil})
}
case "mutex":
@ -589,6 +595,9 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: nil})
}
} else {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: nil})
}
case "int":
@ -656,6 +665,9 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: nil})
}
} else {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: nil})
}
}
@ -680,6 +692,9 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: nil})
}
} else {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: nil})
}
case "bool":
@ -711,6 +726,9 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: nil})
}
} else {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: nil})
}
case "time":

34
server/grpc_test.go
View file

@ -432,6 +432,32 @@ func TestQuerySQLUnary(t *testing.T) {
},
eq: equal,
},
{
// Extract(Limit(ConstRow(columns=[2]), limit=100, offset=0),Rows(age),Rows(color),Rows(height),Rows(score))
sql: "select * from grouper",
exp: tableResponse{
headers: []columnInfo{
{"_id", "uint64"},
{"age", "int64"},
{"color", "[]string"},
{"height", "int64"},
{"score", "int64"},
},
rows: []row{
{[]columnResponse{uint64(1), int64(27), []string{"blue"}, int64(20), int64(-10)}},
{[]columnResponse{uint64(2), int64(16), []string{"blue"}, int64(30), int64(-8)}},
{[]columnResponse{uint64(3), int64(19), []string{"red"}, int64(40), int64(6)}},
{[]columnResponse{uint64(4), int64(27), []string{"green"}, int64(50), int64(0)}},
{[]columnResponse{uint64(5), int64(16), []string{"blue"}, int64(60), int64(-2)}},
{[]columnResponse{uint64(6), int64(34), []string{"blue"}, int64(70), int64(100)}},
{[]columnResponse{uint64(7), int64(27), []string{"blue"}, int64(80), int64(0)}},
{[]columnResponse{uint64(8), int64(16), []string{}, int64(90), int64(-13)}},
{[]columnResponse{uint64(9), int64(16), []string{"red"}, int64(100), int64(80)}},
{[]columnResponse{uint64(10), int64(31), []string{"red"}, int64(110), int64(-2)}},
},
},
eq: equal,
},
// join
{
// Count(Intersect(All(),Distinct(Row(grouperid!=null),index='joiner',field='grouperid')))
@ -471,7 +497,6 @@ func TestQuerySQLUnary(t *testing.T) {
headers: []columnInfo{{"_id", "uint64"}},
rows: []row{
{[]columnResponse{uint64(3)}},
{[]columnResponse{uint64(8)}},
{[]columnResponse{uint64(9)}},
},
},
@ -484,7 +509,6 @@ func TestQuerySQLUnary(t *testing.T) {
headers: []columnInfo{{"_id", "uint64"}},
rows: []row{
{[]columnResponse{uint64(3)}},
{[]columnResponse{uint64(8)}},
{[]columnResponse{uint64(9)}},
},
},
@ -630,7 +654,7 @@ func TestQuerySQLUnary(t *testing.T) {
},
rows: []row{
{[]columnResponse{int64(16), "blue", uint64(2)}},
{[]columnResponse{int64(16), "red", uint64(2)}},
{[]columnResponse{int64(16), "red", uint64(1)}},
{[]columnResponse{int64(19), "red", uint64(1)}},
{[]columnResponse{int64(27), "blue", uint64(2)}},
{[]columnResponse{int64(27), "green", uint64(1)}},
@ -764,7 +788,6 @@ func TestQuerySQLUnary(t *testing.T) {
exp: tableResponse{
headers: []columnInfo{{"_id", "uint64"}},
rows: []row{
{[]columnResponse{uint64(8)}},
{[]columnResponse{uint64(9)}},
},
},
@ -777,7 +800,6 @@ func TestQuerySQLUnary(t *testing.T) {
exp: tableResponse{
headers: []columnInfo{{"_id", "uint64"}},
rows: []row{
{[]columnResponse{uint64(8)}},
{[]columnResponse{uint64(9)}},
},
},
@ -789,7 +811,6 @@ func TestQuerySQLUnary(t *testing.T) {
exp: tableResponse{
headers: []columnInfo{{"_id", "uint64"}},
rows: []row{
{[]columnResponse{uint64(8)}},
{[]columnResponse{uint64(9)}},
},
},
@ -872,7 +893,6 @@ func setUpTestQuerySQLUnary(ctx context.Context, t *testing.T) (gh *server.GRPCH
6: "blue",
7: "blue",
3: "red",
8: "red",
9: "red",
10: "red",
4: "green",

90
translator_test.go
View file

@ -605,3 +605,93 @@ func TestTranslation_Coordinator(t *testing.T) {
}
})
}
func TestTranslation_TranslateIDsOnCluster(t *testing.T) {
c := test.MustRunCluster(t, 4,
[]server.CommandOption{
server.OptCommandServerOptions(
pilosa.OptServerIsCoordinator(true),
pilosa.OptServerNodeID("node0"),
pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore),
pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)),
)},
[]server.CommandOption{
server.OptCommandServerOptions(
pilosa.OptServerIsCoordinator(false),
pilosa.OptServerNodeID("node1"),
pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore),
pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)),
)},
[]server.CommandOption{
server.OptCommandServerOptions(
pilosa.OptServerIsCoordinator(false),
pilosa.OptServerNodeID("node2"),
pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore),
pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)),
)},
[]server.CommandOption{
server.OptCommandServerOptions(
pilosa.OptServerIsCoordinator(false),
pilosa.OptServerNodeID("node3"),
pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore),
pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)),
)},
)
defer c.Close()
node0 := c.GetNode(0)
node3 := c.GetNode(3)
ctx := context.Background()
idx, fld := "i", "f"
// Create an index with keys.
if _, err := node0.API.CreateIndex(ctx, idx, pilosa.IndexOptions{Keys: true}); err != nil {
t.Fatal(err)
}
// Create an index with keys.
if _, err := node0.API.CreateField(ctx, idx, fld, pilosa.OptFieldKeys()); err != nil {
t.Fatal(err)
}
keys := []string{"k0", "k1", "k2", "k3", "k4", "k5", "k6", "k7", "k8", "k9"}
// write a new key and get id
req, err := node0.API.Serializer.Marshal(&pilosa.TranslateKeysRequest{
Index: idx,
Field: fld,
Keys: keys,
NotWritable: false,
})
if err != nil {
t.Fatal(err)
}
if buf, err := node0.API.TranslateKeys(ctx, bytes.NewReader(req)); err != nil {
t.Fatal(err)
} else {
var (
respKeys pilosa.TranslateKeysResponse
respIDs pilosa.TranslateIDsResponse
)
if err = node0.API.Serializer.Unmarshal(buf, &respKeys); err != nil {
t.Fatal(err)
}
ids := respKeys.IDs
// translate ids
req, err = node3.API.Serializer.Marshal(&pilosa.TranslateIDsRequest{
Index: idx,
Field: fld,
IDs: ids,
})
if err != nil {
t.Fatal(err)
}
if buf, err = node3.API.TranslateIDs(ctx, bytes.NewReader(req)); err != nil {
t.Fatal(err)
}
if err = node3.API.Serializer.Unmarshal(buf, &respIDs); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(respIDs.Keys, keys) {
t.Fatalf("TranslateIDs(%+v): expected: %+v, got: %+v", ids, keys, respIDs.Keys)
}
}
}