From d06ffd207f7481d55abf162bbbca04fc829f09e2 Mon Sep 17 00:00:00 2001 From: Travis Date: Thu, 5 Mar 2020 14:40:36 -0600 Subject: [PATCH] forward field translation request to coordinator --- api.go | 8 ++--- cluster.go | 34 ++++++++++++++++-- executor.go | 20 +++++------ executor_internal_test.go | 5 +-- translator_test.go | 74 +++++++++++++++++++++++++++++++++++++++ 5 files changed, 123 insertions(+), 18 deletions(-) diff --git a/api.go b/api.go index c97c5c2a6..1fbafb36f 100644 --- a/api.go +++ b/api.go @@ -966,8 +966,8 @@ func (api *API) Import(ctx context.Context, req *ImportRequest, opts ...ImportOp if len(req.RowIDs) != 0 { return errors.New("row ids cannot be used because field uses string keys") } - if req.RowIDs, err = field.TranslateStore().TranslateKeys(req.RowKeys); err != nil { - return errors.Wrap(err, "translating rows") + if req.RowIDs, err = api.cluster.translateFieldKeys(ctx, field, req.RowKeys); err != nil { + return errors.Wrapf(err, "translating field keys") } } @@ -1479,8 +1479,8 @@ func (api *API) TranslateKeys(ctx context.Context, r io.Reader) (_ []byte, err e if err != nil { return nil, err } - } else if ids, err = field.TranslateStore().TranslateKeys(req.Keys); err != nil { - return nil, err + } else if ids, err = api.cluster.translateFieldKeys(ctx, field, req.Keys); err != nil { + return nil, errors.Wrapf(err, "translating field keys") } } diff --git a/cluster.go b/cluster.go index cc36de430..86b8fc6aa 100644 --- a/cluster.go +++ b/cluster.go @@ -237,8 +237,6 @@ type cluster struct { // nolint: maligned logger logger.Logger InternalClient InternalClient - - // OpenTranslateReader OpenTranslateReaderFunc } // newCluster returns a new instance of Cluster with defaults. @@ -2087,6 +2085,38 @@ func (c *cluster) setStatic(hosts []string) error { return nil } +// translateFieldKey gets a single key from translateFieldKeys. +func (c *cluster) translateFieldKey(ctx context.Context, field *Field, key string) (uint64, error) { + ids, err := c.translateFieldKeys(ctx, field, []string{key}) + if err != nil { + return 0, err + } else if len(ids) == 0 { + return 0, errors.New("translating key on coordinator returned empty set") + } + return ids[0], nil +} + +// translateFieldKeys is basically a wrapper around +// field.TranslateStore().TranslateKey(key), but in +// the case where the local node's translate store +// is read-only (i.e. it's not the primary translate +// store), then this method will forward the translation +// request to the coordinator. +func (c *cluster) translateFieldKeys(ctx context.Context, field *Field, keys []string) ([]uint64, error) { + ids, err := field.TranslateStore().TranslateKeys(keys) + // If we get a "read only" error, then forward the request + // to the coordinator. + if errors.Cause(err) == ErrTranslateStoreReadOnly { + coordinatorNode := c.coordinatorNode() + if ids, err := c.InternalClient.TranslateKeysNode(ctx, &coordinatorNode.URI, field.Index(), field.Name(), keys); err != nil { + return ids, errors.Wrap(err, "translating keys on coordinator") + } else { + return ids, nil + } + } + return ids, err +} + func (c *cluster) translateIndexKey(ctx context.Context, indexName string, key string) (uint64, error) { keyMap, err := c.translateIndexKeySet(ctx, indexName, map[string]struct{}{key: struct{}{}}) if err != nil { diff --git a/executor.go b/executor.go index ce0707be0..79790e6dd 100644 --- a/executor.go +++ b/executor.go @@ -3535,7 +3535,7 @@ func (e *executor) translateCalls(ctx context.Context, defaultIndexName string, // Translate calls. for i := range calls { - if err := e.translateCall(defaultIndexName, calls[i], keyMaps); err != nil { + if err := e.translateCall(ctx, defaultIndexName, calls[i], keyMaps); err != nil { return err } } @@ -3602,7 +3602,7 @@ func (e *executor) collectCallKeySets(ctx context.Context, indexName string, c * return nil } -func (e *executor) translateCall(indexName string, c *pql.Call, keyMaps map[string]map[string]uint64) (err error) { +func (e *executor) translateCall(ctx context.Context, indexName string, c *pql.Call, keyMaps map[string]map[string]uint64) (err error) { // Specifying an 'index' arg applies to all nested calls. if s := c.CallIndex(); s != "" { indexName = s @@ -3672,8 +3672,8 @@ func (e *executor) translateCall(indexName string, c *pql.Call, keyMaps map[stri if foreignIndexName != "" { id = keyMaps[foreignIndexName][cond.Value.(string)] } else { - if id, err = field.TranslateStore().TranslateKey(cond.Value.(string)); err != nil { - return errors.Wrap(err, "translating key") + if id, err = e.Cluster.translateFieldKey(ctx, field, cond.Value.(string)); err != nil { + return errors.Wrapf(err, "translating field key: %s", cond.Value) } } @@ -3695,8 +3695,8 @@ func (e *executor) translateCall(indexName string, c *pql.Call, keyMaps map[stri if foreignIndexName != "" { id = keyMaps[foreignIndexName][value] } else { - if id, err = field.TranslateStore().TranslateKey(value); err != nil { - return err + if id, err = e.Cluster.translateFieldKey(ctx, field, value); err != nil { + return errors.Wrapf(err, "translating field key: %s", value) } } c.Args[rowKey] = id @@ -3710,7 +3710,7 @@ func (e *executor) translateCall(indexName string, c *pql.Call, keyMaps map[stri // Translate child calls. for _, child := range c.Children { - if err := e.translateCall(indexName, child, keyMaps); err != nil { + if err := e.translateCall(ctx, indexName, child, keyMaps); err != nil { return err } } @@ -3718,7 +3718,7 @@ func (e *executor) translateCall(indexName string, c *pql.Call, keyMaps map[stri // Translate call args. for _, arg := range c.Args { if arg, ok := arg.(*pql.Call); ok { - if err := e.translateCall(indexName, arg, keyMaps); err != nil { + if err := e.translateCall(ctx, indexName, arg, keyMaps); err != nil { return errors.Wrap(err, "translating arg") } } @@ -3756,9 +3756,9 @@ func (e *executor) translateCall(indexName string, c *pql.Call, keyMaps map[stri return errors.New("prev value must be a string when field 'keys' option enabled") } // TODO: does this need to take field.ForeignIndex() into consideration? - id, err := field.TranslateStore().TranslateKey(prevStr) + id, err := e.Cluster.translateFieldKey(ctx, field, prevStr) if err != nil { - return errors.Wrapf(err, "translating row key '%s'", prevStr) + return errors.Wrapf(err, "translating field key: %s", prevStr) } previous[i] = id } else { diff --git a/executor_internal_test.go b/executor_internal_test.go index 395ecdcfc..86ff6cdde 100644 --- a/executor_internal_test.go +++ b/executor_internal_test.go @@ -15,6 +15,7 @@ package pilosa import ( + "context" "encoding/json" "fmt" "io/ioutil" @@ -57,7 +58,7 @@ func TestExecutor_TranslateGroupByCall(t *testing.T) { t.Fatalf("parsing query: %v", err) } c := query.Calls[0] - err = e.translateCall("i", c, make(map[string]map[string]uint64)) + err = e.translateCall(context.Background(), "i", c, make(map[string]map[string]uint64)) if err != nil { t.Fatalf("translating call: %v", err) } @@ -121,7 +122,7 @@ func TestExecutor_TranslateGroupByCall(t *testing.T) { t.Fatalf("parsing query: %v", err) } c := query.Calls[0] - err = e.translateCall("i", c, make(map[string]map[string]uint64)) + err = e.translateCall(context.Background(), "i", c, make(map[string]map[string]uint64)) if err == nil { t.Fatalf("expected error, but translated call is '%s", c) } diff --git a/translator_test.go b/translator_test.go index d8957e8bc..32a3da00d 100644 --- a/translator_test.go +++ b/translator_test.go @@ -18,7 +18,9 @@ import ( "bytes" "context" "errors" + "fmt" "io" + "reflect" "testing" "github.com/google/go-cmp/cmp" @@ -293,3 +295,75 @@ func TestTranslation_Reset(t *testing.T) { } }) } + +// Test key translation with multiple nodes. +func TestTranslation_Coordinator(t *testing.T) { + // Ensure that field key translations requests sent to + // non-coordinator nodes are forwarded to the coordinator. + t.Run("ForwardFieldKey", func(t *testing.T) { + // Start a 2-node cluster. + c := test.MustRunCluster(t, 2, + []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)), + )}, + ) + defer c.Close() + + node0 := c[0] + node1 := c[1] + + ctx := context.Background() + idx := "i" + fld := "f" + + // Create an index without keys. + if _, err := node0.API.CreateIndex(ctx, idx, + pilosa.IndexOptions{ + Keys: false, + }); err != nil { + t.Fatal(err) + } + + // Create a field with keys. + if _, err := node0.API.CreateField(ctx, idx, fld, + pilosa.OptFieldKeys(), + ); err != nil { + t.Fatal(err) + } + + key := "abc" + pql := fmt.Sprintf(`Set(1, %s="%s")`, fld, key) + + // Send a translation request to node1 (non-coordinator). + _, err := node1.API.Query(ctx, + &pilosa.QueryRequest{Index: idx, Query: pql}, + ) + if err != nil { + t.Fatal(err) + } + + // Read the row and ensure the key was set. + qry := fmt.Sprintf(`Row(%s="%s")`, fld, key) + resp, err := node0.API.Query(ctx, + &pilosa.QueryRequest{Index: idx, Query: qry}, + ) + if err != nil { + t.Fatal(err) + } + row := resp.Results[0].(*pilosa.Row) + if cols := row.Columns(); !reflect.DeepEqual(cols, []uint64{1}) { + t.Fatalf("unexpected columns: %+v", cols) + } + }) +}