scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-08-28 10:54:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #847 from kuba--/fix-writable

Browse source 
Fix TranslateStore writable
This commit is contained in:
Kuba Podgórski 2020-09-15 20:36:27 +02:00 • committed by Jason Aten
parent 817d6f1498
commit 6348956850
12 changed files with 2568 additions and 103 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

32
api.go
View file

@ -1588,9 +1588,11 @@ func (api *API) TranslateIndexIDs(ctx context.Context, indexName string, ids []u
}
// TranslateKeys handles a TranslateKeyRequest.
// ErrTranslatingKeyNotFound error will be swallowed here, so the empty response will be returned.
func (api *API) TranslateKeys(ctx context.Context, r io.Reader) (_ []byte, err error) {
var req TranslateKeysRequest
if buf, err := ioutil.ReadAll(r); err != nil {
buf, err := ioutil.ReadAll(r)
if err != nil {
return nil, NewBadRequestError(errors.Wrap(err, "read translate keys request error"))
} else if err := api.Serializer.Unmarshal(buf, &req); err != nil {
return nil, NewBadRequestError(errors.Wrap(err, "unmarshal translate keys request error"))
@ -1599,25 +1601,25 @@ func (api *API) TranslateKeys(ctx context.Context, r io.Reader) (_ []byte, err e
// Lookup store for either index or field and translate keys.
var ids []uint64
if req.Field == "" {
if ids, err = api.cluster.translateIndexKeys(ctx, req.Index, req.Keys); err != nil {
return nil, err
}
ids, err = api.cluster.translateIndexKeys(ctx, req.Index, req.Keys, !req.NotWritable)
} else {
if field := api.holder.Field(req.Index, req.Field); field == nil {
return nil, ErrFieldNotFound
} else if fi := field.ForeignIndex(); fi != "" {
ids, err = api.cluster.translateIndexKeys(ctx, fi, req.Keys)
if 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")
field := api.holder.Field(req.Index, req.Field)
if field == nil {
return nil, newNotFoundError(ErrFieldNotFound, req.Field)
}
if fi := field.ForeignIndex(); fi != "" {
ids, err = api.cluster.translateIndexKeys(ctx, fi, req.Keys, !req.NotWritable)
} else {
ids, err = api.cluster.translateFieldKeys(ctx, field, req.Keys, !req.NotWritable)
}
}
if err != nil && errors.Cause(err) != ErrTranslatingKeyNotFound {
return nil, errors.WithMessage(err, "translating keys")
}
// Encode response.
buf, err := api.Serializer.Marshal(&TranslateKeysResponse{IDs: ids})
if err != nil {
if buf, err = api.Serializer.Marshal(&TranslateKeysResponse{IDs: ids}); err != nil {
return nil, errors.Wrap(err, "translate keys response encoding error")
}
return buf, nil

27
boltdb/translate.go
View file

@ -161,20 +161,17 @@ func (s *TranslateStore) Size() int64 {
}
// TranslateKey converts a string key to an integer ID.
// If key does not have an associated id then one is created.
func (s *TranslateStore) TranslateKey(key string) (id uint64, _ error) {
// Find id by key under read lock.
if err := s.db.View(func(tx *bolt.Tx) error {
id, _ = findIDByKey(tx.Bucket([]byte("keys")), key)
return nil
}); err != nil {
// If key does not have an associated id then one is created, unless writable is false,
// then the function will return the error pilosa.ErrTranslatingKeyNotFound.
func (s *TranslateStore) TranslateKey(key string, writable bool) (uint64, error) {
ids, err := s.translateKeys([]string{key}, writable)
if err != nil {
return 0, err
} else if id != 0 {
return id, nil
}
if s.ReadOnly() {
return 0, pilosa.ErrTranslateStoreReadOnly
if len(ids) == 0 {
return 0, ErrTranslateKeyNotFound
}
// Find or create id under write lock.
@ -207,11 +204,11 @@ func (s *TranslateStore) TranslateKey(key string) (id uint64, _ error) {
}
// TranslateKeys converts a slice of string keys to a slice of integer IDs.
// If a key does not have an associated id then one is created.
func (s *TranslateStore) TranslateKeys(keys []string) (ids []uint64, _ error) {
if len(keys) == 0 {
return nil, nil
}
// If a key does not have an associated id then one is created, unless writable is false,
// then the function will return the error pilosa.ErrTranslatingKeyNotFound.
func (s *TranslateStore) TranslateKeys(keys []string, writable bool) ([]uint64, error) {
return s.translateKeys(keys, writable)
}
// Allocate slice for ID mapping.
ids = make([]uint64, len(keys))

8
client.go
View file

@ -88,7 +88,9 @@ type InternalClient interface {
// InternalQueryClient is the internal interface for querying a node.
type InternalQueryClient interface {
QueryNode(ctx context.Context, uri *URI, index string, queryRequest *QueryRequest) (*QueryResponse, error)
TranslateKeysNode(ctx context.Context, uri *URI, index, field string, keys []string) ([]uint64, error)
// Trasnlate keys on the particular node. The parameter writable informs TranslateStore if we can generate a new ID if any of keys does not exist.
TranslateKeysNode(ctx context.Context, uri *URI, index, field string, keys []string, writable bool) ([]uint64, error)
TranslateIDsNode(ctx context.Context, uri *URI, index, field string, id []uint64) ([]string, error)
}
@ -98,7 +100,7 @@ func (n *nopInternalQueryClient) QueryNode(ctx context.Context, uri *URI, index
return nil, nil
}
func (n nopInternalQueryClient) TranslateKeysNode(ctx context.Context, uri *URI, index, field string, keys []string) ([]uint64, error) {
func (n nopInternalQueryClient) TranslateKeysNode(ctx context.Context, uri *URI, index, field string, keys []string, writable bool) ([]uint64, error) {
return nil, nil
}
@ -145,7 +147,7 @@ func (n nopInternalClient) Query(ctx context.Context, index string, queryRequest
func (n nopInternalClient) QueryNode(ctx context.Context, uri *URI, index string, queryRequest *QueryRequest) (*QueryResponse, error) {
return nil, nil
}
func (n nopInternalClient) TranslateKeysNode(ctx context.Context, uri *URI, index, field string, keys []string) ([]uint64, error) {
func (n nopInternalClient) TranslateKeysNode(ctx context.Context, uri *URI, index, field string, keys []string, writable bool) ([]uint64, error) {
return nil, nil
}
func (n nopInternalClient) TranslateIDsNode(ctx context.Context, uri *URI, index, field string, ids []uint64) ([]string, error) {

11
cluster.go
View file

@ -2328,7 +2328,7 @@ func (c *cluster) translateFieldKey(ctx context.Context, field *Field, key strin
if err != nil {
return 0, err
} else if len(ids) == 0 {
return 0, errors.New("translating key on coordinator returned empty set")
return 0, nil
}
return ids[0], nil
}
@ -2345,11 +2345,12 @@ func (c *cluster) translateFieldKeys(ctx context.Context, field *Field, keys ...
// 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 {
ids, err := c.InternalClient.TranslateKeysNode(ctx, &coordinatorNode.URI, field.Index(), field.Name(), keys, writable)
if err == nil {
return ids, nil
}
return ids, errors.Wrap(err, "translating keys on coordinator")
}
return ids, err
}
@ -2410,7 +2411,7 @@ func (c *cluster) translateIndexKeySet(ctx context.Context, indexName string, ke
}
} else {
nodes := c.partitionNodes(partitionID)
if ids, err = c.InternalClient.TranslateKeysNode(ctx, &nodes[0].URI, indexName, "", keys); err != nil {
if ids, err = c.InternalClient.TranslateKeysNode(ctx, &nodes[0].URI, indexName, "", keys, writable); err != nil {
return err
}
}

8
encoding/proto/proto.go
View file

@ -834,9 +834,10 @@ func (s Serializer) encodeRecalculateCaches(*pilosa.RecalculateCaches) *internal
func (s Serializer) encodeTranslateKeysRequest(request *pilosa.TranslateKeysRequest) *internal.TranslateKeysRequest {
return &internal.TranslateKeysRequest{
Index: request.Index,
Field: request.Field,
Keys: request.Keys,
Index: request.Index,
Field: request.Field,
Keys: request.Keys,
NotWritable: request.NotWritable,
}
}
@ -1262,6 +1263,7 @@ func (s Serializer) decodeTranslateKeysRequest(pb *internal.TranslateKeysRequest
m.Index = pb.Index
m.Field = pb.Field
m.Keys = pb.Keys
m.NotWritable = pb.NotWritable
}
func (s Serializer) decodeTranslateKeysResponse(pb *internal.TranslateKeysResponse, m *pilosa.TranslateKeysResponse) {

3
handler.go
View file

@ -266,6 +266,9 @@ type TranslateKeysRequest struct {
Index string
Field string
Keys []string
// it's a awkward name, just to keep backward compatibility with go-pilosa and idk.
NotWritable bool
}
// TranslateKeysResponse is the structured response of a key

15
http/client.go
View file

@ -1140,8 +1140,9 @@ func (c *InternalClient) SendMessage(ctx context.Context, uri *pilosa.URI, msg [
return errors.Wrap(err, "draining SendMessage response body")
}
// TranslateKeysNode sends a key translation request to a specific node.
func (c *InternalClient) TranslateKeysNode(ctx context.Context, uri *pilosa.URI, index, field string, keys []string) ([]uint64, error) {
// TranslateKeysNode function is mainly called to translate keys from coordinator node.
// If coordinator node returns 404 error the function wraps it with pilosa.ErrTranslatingKeyNotFound.
func (c *InternalClient) TranslateKeysNode(ctx context.Context, uri *pilosa.URI, index, field string, keys []string, writable bool) ([]uint64, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "TranslateKeysNode")
defer span.Finish()
@ -1150,9 +1151,10 @@ func (c *InternalClient) TranslateKeysNode(ctx context.Context, uri *pilosa.URI,
}
buf, err := c.serializer.Marshal(&pilosa.TranslateKeysRequest{
Index: index,
Field: field,
Keys: keys,
Index: index,
Field: field,
Keys: keys,
NotWritable: !writable,
})
if err != nil {
return nil, errors.Wrap(err, "marshaling TranslateKeysRequest")
@ -1174,6 +1176,9 @@ func (c *InternalClient) TranslateKeysNode(ctx context.Context, uri *pilosa.URI,
// Execute request against the host.
resp, err := c.executeRequest(req.WithContext(ctx))
if err != nil {
if resp != nil && resp.StatusCode == http.StatusNotFound {
return nil, errors.Wrap(pilosa.ErrTranslatingKeyNotFound, err.Error())
}
return nil, err
}
defer resp.Body.Close()

23
http/handler.go
View file

@ -2154,16 +2154,21 @@ func (h *Handler) handlePostTranslateKeys(w http.ResponseWriter, r *http.Request
}
buf, err := h.api.TranslateKeys(r.Context(), r.Body)
if err != nil {
http.Error(w, fmt.Sprintf("translate keys: %v", err), http.StatusInternalServerError)
return
}
switch errors.Cause(err) {
case nil:
// Write response.
if _, err = w.Write(buf); err != nil {
h.logger.Printf("writing translate keys response: %v", err)
}
// Write response.
_, err = w.Write(buf)
if err != nil {
h.logger.Printf("writing translate keys response: %v", err)
return
case pilosa.ErrTranslatingKeyNotFound:
http.Error(w, fmt.Sprintf("translate keys: %v", err), http.StatusNotFound)
case pilosa.ErrTranslateStoreReadOnly:
http.Error(w, fmt.Sprintf("translate keys: %v", err), http.StatusPreconditionFailed)
default:
http.Error(w, fmt.Sprintf("translate keys: %v", err), http.StatusInternalServerError)
}
}

2379
internal/public.pb.go
View file

File diff suppressed because it is too large Load diff

2
internal/public.proto
View file

@ -126,6 +126,7 @@ message ImportRequest {
repeated int64 Timestamps = 6;
int64 IndexCreatedAt = 9;
int64 FieldCreatedAt = 10;
bool Clear = 11;
}
message ImportValueRequest {
@ -145,6 +146,7 @@ message TranslateKeysRequest {
string Index = 1;
string Field = 2;
repeated string Keys = 3;
bool NotWritable = 4;
}
message TranslateKeysResponse {

7
server/config_test.go
View file

@ -48,12 +48,7 @@ func TestDuration(t *testing.T) {
t.Fatalf("Unexpected marshalled value %v", v)
}
err := d.UnmarshalText([]byte("5"))
if err.Error() != "time: missing unit in duration 5" {
t.Fatalf("expected time: missing unit in duration: %s", err)
}
err = d.UnmarshalText([]byte("3m2s"))
err := d.UnmarshalText([]byte("3m2s"))
if err != nil {
t.Fatalf("unexpected error: %v", err)
}

156
translator_test.go
View file

@ -289,13 +289,167 @@ func TestTranslation_Reset(t *testing.T) {
if err != nil {
t.Fatal(err)
}
if _, err := node0.API.TranslateKeys(ctx, bytes.NewReader(reqBody)); err != nil {
t.Fatal(err)
}
})
}
func TestTranslation_KeyNotFound(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)
node1 := c.GetNode(1)
node2 := c.GetNode(2)
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)
}
// write a new key and get id
req, err := node0.API.Serializer.Marshal(&pilosa.TranslateKeysRequest{
Index: idx,
Field: fld,
Keys: []string{"k1"},
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 resp pilosa.TranslateKeysResponse
if err = node0.API.Serializer.Unmarshal(buf, &resp); err != nil {
t.Fatal(err)
}
id0 := resp.IDs[0]
// read non-existing key
req, err = node3.API.Serializer.Marshal(&pilosa.TranslateKeysRequest{
Index: idx,
Field: fld,
Keys: []string{"k2"},
NotWritable: true,
})
if err != nil {
t.Fatal(err)
}
if buf, err = node3.API.TranslateKeys(ctx, bytes.NewReader(req)); err != nil {
t.Fatal(err)
}
if err = node3.API.Serializer.Unmarshal(buf, &resp); err != nil {
t.Fatal(err)
} else if resp.IDs != nil {
t.Fatalf("TranslateKeys(%+v): expected: nil, got: %d", req, resp)
}
req, err = node2.API.Serializer.Marshal(&pilosa.TranslateKeysRequest{
Index: idx,
Keys: []string{"k2"},
NotWritable: true,
})
if err != nil {
t.Fatal(err)
}
if buf, err = node2.API.TranslateKeys(ctx, bytes.NewReader(req)); err != nil {
t.Fatal(err)
}
if err = node2.API.Serializer.Unmarshal(buf, &resp); err != nil {
t.Fatal(err)
} else if resp.IDs != nil {
t.Fatalf("TranslateKeys(%+v): expected: nil, got: %d", req, resp)
}
req, err = node1.API.Serializer.Marshal(&pilosa.TranslateKeysRequest{
Index: idx,
Field: fld,
Keys: []string{"k2"},
NotWritable: false,
})
if err != nil {
t.Fatal(err)
}
if buf, err = node0.API.TranslateKeys(ctx, bytes.NewReader(req)); err != nil {
t.Fatal(err)
}
if err = node0.API.Serializer.Unmarshal(buf, &resp); err != nil {
t.Fatal(err)
}
if resp.IDs[0] != id0+1 {
t.Fatalf("TranslateKeys(%+v): expected: %d, got: %d", req, id0+1, resp.IDs[0])
}
}
}
func TestInMemTranslateStore_ReadKey(t *testing.T) {
s := pilosa.NewInMemTranslateStore("IDX", "FLD", 0, pilosa.DefaultPartitionN)
id, err := s.TranslateKey("foo", false)
if err != pilosa.ErrTranslatingKeyNotFound {
t.Fatal(err)
}
if got, want := id, uint64(0); got != want {
t.Fatalf("TranslateKey()=%d, want %d", got, want)
}
// Ensure next key autoincrements.
if id, err = s.TranslateKey("foo", true); err != nil {
t.Fatal(err)
}
if got, want := id, uint64(1); got != want {
t.Fatalf("TranslateKey()=%d, want %d", got, want)
}
id1, err := s.TranslateKey("foo", false)
if err != nil {
t.Fatal(err)
}
if got, want := id1, id; got != want || id == 0 {
t.Fatalf("TranslateKey()=%d, want %d", got, want)
}
}
// Test index key translation replication under node failure.
func TestTranslation_Replication(t *testing.T) {
t.Run("Replication", func(t *testing.T) {