From b54a289cd8ad46e67923dfc79bb579de2aad31c5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Kuba=20Podg=C3=B3rski?= Date: Tue, 22 Sep 2020 12:02:26 +0200 Subject: [PATCH 1/3] Fix holes in grpc response for inspect --- server/grpc.go | 18 ++++++++++++++++++ server/grpc_test.go | 34 +++++++++++++++++++++++++++------- 2 files changed, 45 insertions(+), 7 deletions(-) diff --git a/server/grpc.go b/server/grpc.go index 38d58393e..ec99f0685 100644 --- a/server/grpc.go +++ b/server/grpc.go @@ -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": diff --git a/server/grpc_test.go b/server/grpc_test.go index eed974073..6a7776259 100644 --- a/server/grpc_test.go +++ b/server/grpc_test.go @@ -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", From 55d4c29933adf75a56beb143d24aad9980fb06fb Mon Sep 17 00:00:00 2001 From: Jason Aten Date: Sat, 19 Sep 2020 20:12:03 -0500 Subject: [PATCH 2/3] turn off debug machinery on tx backends - enable row cache again. Was off for tx perf measurement. - centralize UseRowCache choice to just one point, in rbf.EnableRowCache --- badger.go | 3 +- lmdb.go | 82 +++++++++++++++++++++++++++----------------------- pprof.go | 37 +++++++++++++++++++++-- rbf.go | 2 +- rbf/cursorx.go | 2 +- rrtx.go | 3 +- 6 files changed, 85 insertions(+), 44 deletions(-) diff --git a/badger.go b/badger.go index f4d16facc..f667da77c 100644 --- a/badger.go +++ b/badger.go @@ -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 diff --git a/lmdb.go b/lmdb.go index 4758c4b26..73167e4ac 100644 --- a/lmdb.go +++ b/lmdb.go @@ -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 } diff --git a/pprof.go b/pprof.go index 7e9f721f0..5c9b0b339 100644 --- a/pprof.go +++ b/pprof.go @@ -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() +} diff --git a/rbf.go b/rbf.go index 79284de3c..36eebf64d 100644 --- a/rbf.go +++ b/rbf.go @@ -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) diff --git a/rbf/cursorx.go b/rbf/cursorx.go index 5d57bdfd9..26d0c002d 100644 --- a/rbf/cursorx.go +++ b/rbf/cursorx.go @@ -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 diff --git a/rrtx.go b/rrtx.go index 7ea06f659..4868e443e 100644 --- a/rrtx.go +++ b/rrtx.go @@ -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) { From ba5438e02b0a808e876958834b21f940fb4f5b65 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Kuba=20Podg=C3=B3rski?= Date: Wed, 23 Sep 2020 15:21:33 +0200 Subject: [PATCH 3/3] Trabnslate field IDs on coordinator --- api.go | 2 +- cluster.go | 40 +++++++++++++++++++++ executor.go | 28 +++------------ translator_test.go | 90 ++++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 135 insertions(+), 25 deletions(-) diff --git a/api.go b/api.go index 28860e894..878744e12 100644 --- a/api.go +++ b/api.go @@ -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 } } diff --git a/cluster.go b/cluster.go index 3946d1024..6e5450cf4 100644 --- a/cluster.go +++ b/cluster.go @@ -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 { diff --git a/executor.go b/executor.go index b3360aae5..0a5b207ae 100644 --- a/executor.go +++ b/executor.go @@ -5130,26 +5130,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)) @@ -5159,7 +5139,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) { @@ -5245,7 +5225,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 } @@ -5296,7 +5276,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()) } @@ -5348,7 +5328,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") } diff --git a/translator_test.go b/translator_test.go index 9a7607973..3de856772 100644 --- a/translator_test.go +++ b/translator_test.go @@ -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) + } + } +}