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

Merge pull request #732 from kuba--/translatekey-writable

Browse source 
Add writable argument to TranslateKey functions.
This commit is contained in:
Kuba Podgórski 2020-08-27 16:15:49 +02:00 • committed by Jason Aten
parent 450ae490a5
commit 4b0789ebf0
12 changed files with 448 additions and 222 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
api.go
View file

@ -1077,7 +1077,7 @@ 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 = api.cluster.translateFieldKeys(ctx, field, req.RowKeys...); err != nil {
if req.RowIDs, err = api.cluster.translateFieldKeys(ctx, field, req.RowKeys, true); err != nil {
return errors.Wrapf(err, "translating field keys")
}
}
@ -1088,7 +1088,7 @@ func (api *API) Import(ctx context.Context, req *ImportRequest, opts ...ImportOp
if len(req.ColumnIDs) != 0 {
return errors.New("column ids cannot be used because index uses string keys")
}
if req.ColumnIDs, err = api.cluster.translateIndexKeys(ctx, req.Index, req.ColumnKeys); err != nil {
if req.ColumnIDs, err = api.cluster.translateIndexKeys(ctx, req.Index, req.ColumnKeys, true); err != nil {
return errors.Wrap(err, "translating columns")
}
}
@ -1201,7 +1201,7 @@ func (api *API) ImportValue(ctx context.Context, req *ImportValueRequest, opts .
if len(req.ColumnIDs) != 0 {
return errors.New("column ids cannot be used because index uses string keys")
}
if req.ColumnIDs, err = api.cluster.translateIndexKeys(ctx, req.Index, req.ColumnKeys); err != nil {
if req.ColumnIDs, err = api.cluster.translateIndexKeys(ctx, req.Index, req.ColumnKeys, true); err != nil {
return errors.Wrap(err, "translating columns")
}
req.Shard = math.MaxUint64
@ -1212,7 +1212,7 @@ func (api *API) ImportValue(ctx context.Context, req *ImportValueRequest, opts .
if field.Keys() {
// Perform translation.
span.LogKV("rowKeys", true)
uints, err := api.cluster.translateIndexKeys(ctx, field.ForeignIndex(), req.StringValues)
uints, err := api.cluster.translateIndexKeys(ctx, field.ForeignIndex(), req.StringValues, true)
if err != nil {
return err
}
@ -1579,8 +1579,8 @@ func (api *API) GetTranslateEntryReader(ctx context.Context, offsets TranslateOf
return NewMultiTranslateEntryReader(ctx, a), nil
}
func (api *API) TranslateIndexKey(ctx context.Context, indexName string, key string) (uint64, error) {
return api.cluster.translateIndexKey(ctx, indexName, key)
func (api *API) TranslateIndexKey(ctx context.Context, indexName string, key string, writable bool) (uint64, error) {
return api.cluster.translateIndexKey(ctx, indexName, key, writable)
}
func (api *API) TranslateIndexIDs(ctx context.Context, indexName string, ids []uint64) ([]string, error) {

39
boltdb/translate.go
View file

@ -32,11 +32,20 @@ var (
// ErrTranslateStoreClosed is returned when reading from an TranslateEntryReader
// and the underlying store is closed.
ErrTranslateStoreClosed = errors.New("boltdb: translate store closing")
// ErrTranslateKeyNotFound is returned when translating key
// and the underlying store returns an empty set
ErrTranslateKeyNotFound = errors.New("boltdb: translating key returned empty set")
bucketKeys = []byte("keys")
bucketIDs = []byte("ids")
)
const (
// snapshotExt is the file extension used for an in-process snapshot.
snapshotExt = ".snapshotting"
errFmtTranslateBucketNotFound = "boltdb: translate bucket '%s' not found"
)
// OpenTranslateStore opens and initializes a boltdb translation store.
@ -102,9 +111,9 @@ func (s *TranslateStore) Open() (err error) {
// Initialize buckets.
if err := s.db.Update(func(tx *bolt.Tx) error {
if _, err := tx.CreateBucketIfNotExists([]byte("keys")); err != nil {
if _, err := tx.CreateBucketIfNotExists(bucketKeys); err != nil {
return err
} else if _, err := tx.CreateBucketIfNotExists([]byte("ids")); err != nil {
} else if _, err := tx.CreateBucketIfNotExists(bucketIDs); err != nil {
return err
}
return nil
@ -195,12 +204,11 @@ func (s *TranslateStore) TranslateKey(key string, writable bool) (uint64, error)
}); err != nil {
return 0, err
}
if written {
s.notifyWrite()
if len(ids) == 0 {
// this should not happen
return 0, ErrTranslateKeyNotFound
}
return id, nil
return ids[0], nil
}
// TranslateKeys converts a slice of string keys to a slice of integer IDs.
@ -241,7 +249,9 @@ func (s *TranslateStore) translateKeys(keys []string, writable bool) ([]uint64,
}
return nil, nil
}
if !writable {
return nil, pilosa.ErrTranslatingKeyNotFound
}
// Find or create ids under write lock if any keys were not found.
var written bool
if err := s.db.Update(func(tx *bolt.Tx) (err error) {
@ -265,7 +275,6 @@ func (s *TranslateStore) translateKeys(keys []string, writable bool) ([]uint64,
}); err != nil {
return nil, err
}
if written {
s.notifyWrite()
}
@ -280,7 +289,7 @@ func (s *TranslateStore) TranslateID(id uint64) (string, error) {
return "", err
}
defer func() { _ = tx.Rollback() }()
return findKeyByID(tx.Bucket([]byte("ids")), id), nil
return findKeyByID(tx.Bucket(bucketIDs), id), nil
}
// TranslateIDs converts a list of integer IDs to a list of string keys.
@ -297,7 +306,7 @@ func (s *TranslateStore) TranslateIDs(ids []uint64) ([]string, error) {
keys := make([]string, len(ids))
for i, id := range ids {
keys[i] = findKeyByID(tx.Bucket([]byte("ids")), id)
keys[i] = findKeyByID(tx.Bucket(bucketIDs), id)
}
return keys, nil
}
@ -305,9 +314,9 @@ func (s *TranslateStore) TranslateIDs(ids []uint64) ([]string, error) {
// ForceSet writes the id/key pair to the store even if read only. Used by replication.
func (s *TranslateStore) ForceSet(id uint64, key string) error {
if err := s.db.Update(func(tx *bolt.Tx) (err error) {
if err := tx.Bucket([]byte("keys")).Put([]byte(key), u64tob(id)); err != nil {
if err := tx.Bucket(bucketKeys).Put([]byte(key), u64tob(id)); err != nil {
return err
} else if err := tx.Bucket([]byte("ids")).Put(u64tob(id), []byte(key)); err != nil {
} else if err := tx.Bucket(bucketIDs).Put(u64tob(id), []byte(key)); err != nil {
return err
}
return nil
@ -400,7 +409,7 @@ func (s *TranslateStore) ReadFrom(r io.Reader) (n int64, err error) {
// MaxID returns the highest id in the store.
func maxID(tx *bolt.Tx) uint64 {
if key, _ := tx.Bucket([]byte("ids")).Cursor().Last(); key != nil {
if key, _ := tx.Bucket(bucketIDs).Cursor().Last(); key != nil {
return btou64(key)
}
return 0
@ -438,7 +447,7 @@ func (r *TranslateEntryReader) ReadEntry(entry *pilosa.TranslateEntry) error {
var found bool
if err := r.store.db.View(func(tx *bolt.Tx) error {
// Find ID/key lookup at offset or later.
cur := tx.Bucket([]byte("ids")).Cursor()
cur := tx.Bucket(bucketIDs).Cursor()
key, value := cur.Seek(u64tob(r.offset))
if key == nil {
return nil

104
boltdb/translate_test.go
View file

@ -32,20 +32,20 @@ func TestTranslateStore_TranslateKey(t *testing.T) {
defer MustCloseTranslateStore(s)
// Ensure initial key translates to first ID for shard
id1, err := s.TranslateKey("foo")
id1, err := s.TranslateKey("foo", true)
if err != nil {
t.Fatal(err)
}
// Ensure next key autoincrements.
if id, err := s.TranslateKey("bar"); err != nil {
if id, err := s.TranslateKey("bar", true); err != nil {
t.Fatal(err)
} else if got, want := id, id1+1; got != want {
t.Fatalf("TranslateKey()=%d, want %d", got, want)
}
// Ensure retranslating existing key returns original ID.
if id, err := s.TranslateKey("foo"); err != nil {
if id, err := s.TranslateKey("foo", true); err != nil {
t.Fatal(err)
} else if got, want := id, id1; got != want {
t.Fatalf("TranslateKey()=%d, want %d", got, want)
@ -56,8 +56,15 @@ func TestTranslateStore_TranslateKeys(t *testing.T) {
s := MustOpenNewTranslateStore()
defer MustCloseTranslateStore(s)
ids, err := s.TranslateKeys([]string{"abc", "abc"}, true)
if err != nil {
t.Fatal(err)
} else if got, want := ids[1], ids[0]; got != want {
t.Fatalf("TranslateKeys()[1]=%d, want %d", got, want)
}
// Ensure initial keys translate to incrementing IDs.
ids1, err := s.TranslateKeys([]string{"foo", "bar"})
ids1, err := s.TranslateKeys([]string{"foo", "bar"}, true)
if err != nil {
t.Fatal(err)
} else if got, want := ids1[1], ids1[0]+1; got != want {
@ -65,7 +72,7 @@ func TestTranslateStore_TranslateKeys(t *testing.T) {
}
// Ensure retranslation returns original IDs.
if ids, err := s.TranslateKeys([]string{"foo", "bar"}); err != nil {
if ids, err := s.TranslateKeys([]string{"foo", "bar"}, true); err != nil {
t.Fatal(err)
} else if got, want := ids[0], ids1[0]; got != want {
t.Fatalf("TranslateKeys()[0]=%d, want %d", got, want)
@ -74,7 +81,7 @@ func TestTranslateStore_TranslateKeys(t *testing.T) {
}
// Ensure retranslating with existing and non-existing keys returns correctly.
if ids, err := s.TranslateKeys([]string{"foo", "baz", "bar"}); err != nil {
if ids, err := s.TranslateKeys([]string{"foo", "baz", "bar"}, true); err != nil {
t.Fatal(err)
} else if got, want := ids[0], ids1[0]; got != want {
t.Fatalf("TranslateKeys()[0]=%d, want %d", got, want)
@ -85,20 +92,83 @@ func TestTranslateStore_TranslateKeys(t *testing.T) {
}
}
func TestTranslateStore_ReadKey(t *testing.T) {
s := MustOpenNewTranslateStore()
defer MustCloseTranslateStore(s)
id, err := s.TranslateKey("foo", false)
if err != pilosa.ErrTranslatingKeyNotFound {
t.Fatal(err)
}
if id != 0 {
t.Fatalf("TranslateKey()=%d, want %d", id, 0)
}
s.SetReadOnly(true)
id, err = s.TranslateKey("foo", true)
if err == nil {
t.Fatalf("got error: %+v, want: 'translate store read only'", err)
}
if id != 0 {
t.Fatalf("TranslateKey()=%d, want %d", id, 0)
}
s.SetReadOnly(false)
// Ensure next key autoincrements.
if id, err = s.TranslateKey("foo", true); err != nil {
t.Fatal(err)
}
id1, err := s.TranslateKey("foo", false)
if err != nil {
t.Fatal(err)
}
if id1 != id {
t.Fatalf("TranslateKey()=%d, want %d", id1, id)
}
}
func TestTranslateStore_ReadKeys(t *testing.T) {
s := MustOpenNewTranslateStore()
defer MustCloseTranslateStore(s)
ids, err := s.TranslateKeys([]string{"foo", "bar", "baz", "baz", "bar", "foo"}, false)
if err != pilosa.ErrTranslatingKeyNotFound {
t.Fatal(err)
}
for _, id := range ids {
if id != 0 {
t.Fatalf("TranslateKeys()=%d, want %d", id, 0)
}
}
// Ensure next key autoincrements.
if ids, err = s.TranslateKeys([]string{"foo", "bar", "baz", "baz", "bar", "foo"}, true); err != nil {
t.Fatal(err)
}
ids1, err := s.TranslateKeys([]string{"foo", "bar", "baz", "baz", "bar", "foo"}, false)
if err != nil {
t.Fatal(err)
}
for i := range ids1 {
if ids1[i] != ids[i] {
t.Fatalf("TranslateKeys()=%d, want %d", ids1[i], ids[i])
}
}
}
func TestTranslateStore_TranslateID(t *testing.T) {
s := MustOpenNewTranslateStore()
defer MustCloseTranslateStore(s)
// Setup initial keys.
id1, err := s.TranslateKey("foo")
id1, err := s.TranslateKey("foo", true)
if err != nil {
t.Fatal(err)
}
id2, err := s.TranslateKey("bar")
id2, err := s.TranslateKey("bar", true)
if err != nil {
t.Fatal(err)
}
id3, err := s.TranslateKey("")
id3, err := s.TranslateKey("", true)
if err != nil {
t.Fatal(err)
}
@ -129,7 +199,7 @@ func TestTranslateStore_TranslateIDs(t *testing.T) {
defer MustCloseTranslateStore(s)
// Setup initial keys.
ids, err := s.TranslateKeys([]string{"foo", "bar"})
ids, err := s.TranslateKeys([]string{"foo", "bar"}, true)
if err != nil {
t.Fatal(err)
}
@ -152,7 +222,7 @@ func TestTranslateStore_EntryReader(t *testing.T) {
defer MustCloseTranslateStore(s)
// Create multiple new keys.
ids1, err := s.TranslateKeys([]string{"foo", "bar"})
ids1, err := s.TranslateKeys([]string{"foo", "bar"}, true)
if err != nil {
t.Fatal(err)
}
@ -184,7 +254,7 @@ func TestTranslateStore_EntryReader(t *testing.T) {
}
// Insert next key while reader is open.
id2, err := s.TranslateKey("baz")
id2, err := s.TranslateKey("baz", true)
if err != nil {
t.Fatal(err)
}
@ -224,7 +294,7 @@ func TestTranslateStore_EntryReader(t *testing.T) {
translateErr := make(chan error)
go func() {
time.Sleep(100 * time.Millisecond)
id, err := s.TranslateKey("foo")
id, err := s.TranslateKey("foo", true)
if err != nil {
translateErr <- err
}
@ -345,7 +415,7 @@ func TestTranslateStore_ReadWrite(t *testing.T) {
}
// Populate the store with the keys in batch0.
batch0IDs, err := s.TranslateKeys(batch0)
batch0IDs, err := s.TranslateKeys(batch0, true)
if err != nil {
t.Fatal(err)
}
@ -362,7 +432,7 @@ func TestTranslateStore_ReadWrite(t *testing.T) {
}
// Populate the store with the keys in batch1.
batch1IDs, err := s.TranslateKeys(batch1)
batch1IDs, err := s.TranslateKeys(batch1, true)
if err != nil {
t.Fatal(err)
}
@ -370,7 +440,7 @@ func TestTranslateStore_ReadWrite(t *testing.T) {
expIDs := []uint64{batch0IDs[50], batch1IDs[50]}
// Check the IDs for a key from each batch.
if ids, err := s.TranslateKeys([]string{"key50", "key150"}); err != nil {
if ids, err := s.TranslateKeys([]string{"key50", "key150"}, false); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(expIDs, ids) {
t.Fatalf("first expected ids: %v, but got: %v", expIDs, ids)
@ -385,7 +455,7 @@ func TestTranslateStore_ReadWrite(t *testing.T) {
// This time, we expect the second key to be different because
// we overwrote the store, and then just set that key.
if ids, err := s.TranslateKeys([]string{"key50", "key150"}); err != nil {
if ids, err := s.TranslateKeys([]string{"key50", "key150"}, true); err != nil {
t.Fatal(err)
} else if ids[0] != expIDs[0] {
t.Fatalf("last expected ids[0]: %d, but got: %d", expIDs[0], ids[0])

14
cluster.go
View file

@ -2325,8 +2325,8 @@ func (c *cluster) setStatic(hosts []string) error {
}
// 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, key)
func (c *cluster) translateFieldKey(ctx context.Context, field *Field, key string, writable bool) (uint64, error) {
ids, err := c.translateFieldKeys(ctx, field, []string{key}, writable)
if err != nil {
return 0, err
} else if len(ids) == 0 {
@ -2359,21 +2359,21 @@ func (c *cluster) translateFieldKeys(ctx context.Context, field *Field, keys []s
return ids, nil
}
func (c *cluster) translateIndexKey(ctx context.Context, indexName string, key string) (uint64, error) {
keyMap, err := c.translateIndexKeySet(ctx, indexName, map[string]struct{}{key: struct{}{}})
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 {
return 0, err
}
return keyMap[key], nil
}
func (c *cluster) translateIndexKeys(ctx context.Context, indexName string, keys []string) ([]uint64, error) {
func (c *cluster) translateIndexKeys(ctx context.Context, indexName string, keys []string, writable bool) ([]uint64, error) {
keySet := make(map[string]struct{})
for _, key := range keys {
keySet[key] = struct{}{}
}
keyMap, err := c.translateIndexKeySet(ctx, indexName, keySet)
keyMap, err := c.translateIndexKeySet(ctx, indexName, keySet, writable)
if err != nil {
return nil, err
}
@ -2394,7 +2394,7 @@ func (c *cluster) translateIndexKeys(ctx context.Context, indexName string, keys
return ids, nil
}
func (c *cluster) translateIndexKeySet(ctx context.Context, indexName string, keySet map[string]struct{}) (map[string]uint64, error) {
func (c *cluster) translateIndexKeySet(ctx context.Context, indexName string, keySet map[string]struct{}, writable bool) (map[string]uint64, error) {
keyMap := make(map[string]uint64)
idx := c.holder.Index(indexName)

61
executor.go
View file

@ -89,6 +89,24 @@ func optExecutorWorkerPoolSize(size int) executorOption {
}
}
func emptyResult(c *pql.Call) interface{} {
switch c.Name {
case "Clear", "ClearRow":
return false
case "Row":
return Row{Keys: []string{}}
case "Rows":
return RowIdentifiers{Keys: []string{}}
case "IncludesColumn":
return false
}
return nil
}
// newExecutor returns a new instance of Executor.
func newExecutor(opts ...executorOption) *executor {
e := &executor{
@ -194,6 +212,14 @@ func (e *executor) Execute(ctx context.Context, index string, q *pql.Query, shar
// No need to translate a remote call.
if !opt.Remote {
if err := e.translateCalls(ctx, index, q.Calls); err != nil {
if errors.Cause(err) == ErrTranslatingKeyNotFound {
// No error - return empty result
resp.Results = make([]interface{}, len(q.Calls))
for i, c := range q.Calls {
resp.Results[i] = emptyResult(c)
}
return resp, nil
}
return resp, err
} else if err := validateQueryContext(ctx); err != nil {
return resp, err
@ -260,6 +286,14 @@ func (e *executor) Execute(ctx context.Context, index string, q *pql.Query, shar
// No need to translate a remote call.
if !opt.Remote {
if err := e.translateResults(ctx, index, idx, q.Calls, results); err != nil {
if errors.Cause(err) == ErrTranslatingKeyNotFound {
// No error - return empty result
resp.Results = make([]interface{}, len(q.Calls))
for i, c := range q.Calls {
resp.Results[i] = emptyResult(c)
}
return resp, nil
}
return resp, err
} else if err := validateQueryContext(ctx); err != nil {
return resp, err
@ -721,7 +755,7 @@ func (e *executor) executeFieldValueCall(ctx context.Context, index string, c *p
var colID uint64
if key, ok := colKey.(string); ok && idx.Keys() {
id, err := e.Cluster.translateIndexKey(ctx, index, key)
id, err := e.Cluster.translateIndexKey(ctx, index, key, false)
if err != nil {
return ValCount{}, errors.Wrap(err, "getting column id")
}
@ -3918,9 +3952,12 @@ func (e *executor) translateCalls(ctx context.Context, defaultIndexName string,
// Generate a list of all used
keySets := make(map[string]map[string]struct{})
keySets[defaultIndexName] = make(map[string]struct{})
for i := range calls {
if err := e.collectCallKeySets(ctx, defaultIndexName, calls[i], keySets); err != nil {
writable := false
for _, c := range calls {
if c.Writable() {
writable = true
}
if err := e.collectCallKeySets(ctx, defaultIndexName, c, keySets); err != nil {
return err
}
}
@ -3936,14 +3973,14 @@ func (e *executor) translateCalls(ctx context.Context, defaultIndexName string,
if !idx.Keys() || len(keySets) == 0 {
continue
}
if keyMaps[indexName], err = e.Cluster.translateIndexKeySet(ctx, indexName, keySet); err != nil {
if keyMaps[indexName], err = e.Cluster.translateIndexKeySet(ctx, indexName, keySet, writable); err != nil {
return err
}
}
// Translate calls.
for i := range calls {
if err := e.translateCall(ctx, defaultIndexName, calls[i], keyMaps); err != nil {
for _, c := range calls {
if err := e.translateCall(ctx, defaultIndexName, c, keyMaps, c.Writable()); err != nil {
return err
}
}
@ -4010,7 +4047,7 @@ func (e *executor) collectCallKeySets(ctx context.Context, indexName string, c *
return nil
}
func (e *executor) translateCall(ctx context.Context, 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, writable bool) (err error) {
// Specifying an 'index' arg applies to all nested calls.
if s := c.CallIndex(); s != "" {
indexName = s
@ -4080,7 +4117,7 @@ func (e *executor) translateCall(ctx context.Context, indexName string, c *pql.C
if foreignIndexName != "" {
id = keyMaps[foreignIndexName][cond.Value.(string)]
} else {
if id, err = e.Cluster.translateFieldKey(ctx, field, cond.Value.(string)); err != nil {
if id, err = e.Cluster.translateFieldKey(ctx, field, cond.Value.(string), writable); err != nil {
return errors.Wrapf(err, "translating field key: %s", cond.Value)
}
}
@ -4103,7 +4140,7 @@ func (e *executor) translateCall(ctx context.Context, indexName string, c *pql.C
if foreignIndexName != "" {
id = keyMaps[foreignIndexName][value]
} else {
if id, err = e.Cluster.translateFieldKey(ctx, field, value); err != nil {
if id, err = e.Cluster.translateFieldKey(ctx, field, value, writable); err != nil {
return errors.Wrapf(err, "translating field key: %s", value)
}
}
@ -4118,7 +4155,7 @@ func (e *executor) translateCall(ctx context.Context, indexName string, c *pql.C
// Translate child calls.
for _, child := range c.Children {
if err := e.translateCall(ctx, indexName, child, keyMaps); err != nil {
if err := e.translateCall(ctx, indexName, child, keyMaps, writable); err != nil {
return err
}
}
@ -4126,7 +4163,7 @@ func (e *executor) translateCall(ctx context.Context, indexName string, c *pql.C
// Translate call args.
for _, arg := range c.Args {
if arg, ok := arg.(*pql.Call); ok {
if err := e.translateCall(ctx, indexName, arg, keyMaps); err != nil {
if err := e.translateCall(ctx, indexName, arg, keyMaps, writable); err != nil {
return errors.Wrap(err, "translating arg")
}
}

8
executor_internal_test.go
View file

@ -58,7 +58,9 @@ func TestExecutor_TranslateGroupByCall(t *testing.T) {
t.Fatalf("parsing query: %v", err)
}
c := query.Calls[0]
err = e.translateCall(context.Background(), "i", c, make(map[string]map[string]uint64))
// this is writable call just for testing purpose - to test previous argument
// generally GroupBy calls are not writable and keys should already exist
err = e.translateCall(context.Background(), "i", c, make(map[string]map[string]uint64), true)
if err != nil {
t.Fatalf("translating call: %v", err)
}
@ -122,7 +124,7 @@ func TestExecutor_TranslateGroupByCall(t *testing.T) {
t.Fatalf("parsing query: %v", err)
}
c := query.Calls[0]
err = e.translateCall(context.Background(), "i", c, make(map[string]map[string]uint64))
err = e.translateCall(context.Background(), "i", c, make(map[string]map[string]uint64), false)
if err == nil {
t.Fatalf("expected error, but translated call is '%s", c)
}
@ -181,7 +183,7 @@ func TestExecutor_TranslateRowsOnBool(t *testing.T) {
}
c := query.Calls[0]
err = e.translateCall(context.Background(), "i", c, make(map[string]map[string]uint64))
err = e.translateCall(context.Background(), "i", c, make(map[string]map[string]uint64), true)
if err != nil {
t.Fatalf("translating call: %v", err)
}

13
executor_test.go
View file

@ -4746,7 +4746,18 @@ func TestExecutor_Execute_Rows_Keys(t *testing.T) {
if !reflect.DeepEqual(rows.Keys, test.exp) {
t.Fatalf("\ngot: %+v\nexp: %+v", rows.Keys, test.exp)
} else if rows.Rows != nil {
t.Fatalf("\ngot: %+v\nexp: nil", rows.Rows)
if test.exp == nil {
if res.Results != nil {
t.Fatalf("\ngot: %+v\nexp: nil, %[1]T, %#[1]v", res.Results)
}
} else {
rows := res.Results[0].(pilosa.RowIdentifiers)
if !reflect.DeepEqual(rows.Keys, test.exp) {
t.Fatalf("\ngot: %+v %[1]T\nexp: %+v %[2]T", rows.Keys, test.exp)
} else if rows.Rows != nil {
t.Fatalf("\ngot: %+v %[1]T\nexp: nil", rows.Rows)
}
}
}
}
})

12
mock/translator.go
View file

@ -29,8 +29,8 @@ type TranslateStore struct {
PartitionIDFunc func() int
ReadOnlyFunc func() bool
SetReadOnlyFunc func(v bool)
TranslateKeyFunc func(key string) (uint64, error)
TranslateKeysFunc func(keys []string) ([]uint64, error)
TranslateKeyFunc func(key string, writable bool) (uint64, error)
TranslateKeysFunc func(keys []string, writable bool) ([]uint64, error)
TranslateIDFunc func(id uint64) (string, error)
TranslateIDsFunc func(ids []uint64) ([]string, error)
ForceSetFunc func(id uint64, key string) error
@ -57,12 +57,12 @@ func (s *TranslateStore) SetReadOnly(v bool) {
s.SetReadOnlyFunc(v)
}
func (s *TranslateStore) TranslateKey(key string) (uint64, error) {
return s.TranslateKeyFunc(key)
func (s *TranslateStore) TranslateKey(key string, writable bool) (uint64, error) {
return s.TranslateKeyFunc(key, writable)
}
func (s *TranslateStore) TranslateKeys(keys []string) ([]uint64, error) {
return s.TranslateKeysFunc(keys)
func (s *TranslateStore) TranslateKeys(keys []string, writable bool) ([]uint64, error) {
return s.TranslateKeysFunc(keys, writable)
}
func (s *TranslateStore) TranslateID(id uint64) (string, error) {

13
pql/ast.go
View file

@ -791,6 +791,19 @@ func (c *Call) TranslateInfo(columnLabel, rowLabel string) (colKey, rowKey, fiel
}
}
// Writable returns true if call is mutable (e.g. can write new translation keys)
func (c *Call) Writable() bool {
switch c.Name {
case "Set", "SetRowAttrs", "SetColumnAttrs", "SetBit":
return true
case "Not":
// to support queries like Not(Row(f="garbage"))
return true
default:
return false
}
}
func (c *Call) ArgString(key string) string {
value, ok := c.Args[key]
if !ok {

349
server/grpc.go
View file

@ -357,6 +357,7 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe
}
ci = nil // only include headers with the first row
colAdded := 0
for _, field := range fields {
// TODO: handle `time` fields
switch field.Type() {
@ -371,17 +372,21 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe
return errors.Wrapf(err, "querying rows for set: %s", pql)
}
ids, ok := resp.Results[0].(pilosa.RowIdentifiers)
if !ok {
return errors.Wrap(err, "getting row identifiers")
}
if len(resp.Results) > 0 {
ids, ok := resp.Results[0].(pilosa.RowIdentifiers)
if !ok {
return errors.Wrap(err, "getting row identifiers")
}
if len(ids.Keys) > 0 {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_StringArrayVal{StringArrayVal: &pb.StringArray{Vals: ids.Keys}}})
} else {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Uint64ArrayVal{Uint64ArrayVal: &pb.Uint64Array{Vals: ids.Rows}}})
if len(ids.Keys) > 0 {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_StringArrayVal{StringArrayVal: &pb.StringArray{Vals: ids.Keys}}})
colAdded++
} else if len(ids.Rows) > 0 {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Uint64ArrayVal{Uint64ArrayVal: &pb.Uint64Array{Vals: ids.Rows}}})
colAdded++
}
}
case "mutex":
@ -395,20 +400,24 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe
return errors.Wrap(err, "querying rows for mutex")
}
ids, ok := resp.Results[0].(pilosa.RowIdentifiers)
if !ok {
return errors.Wrap(err, "getting row identifiers")
}
if len(resp.Results) > 0 {
ids, ok := resp.Results[0].(pilosa.RowIdentifiers)
if !ok {
return errors.Wrap(err, "getting row identifiers")
}
if len(ids.Keys) == 1 {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_StringVal{StringVal: ids.Keys[0]}})
} else if len(ids.Rows) == 1 {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Uint64Val{Uint64Val: ids.Rows[0]}})
} else {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: nil})
if len(ids.Keys) == 1 {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_StringVal{StringVal: ids.Keys[0]}})
colAdded++
} else if len(ids.Rows) == 1 {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Uint64Val{Uint64Val: ids.Rows[0]}})
colAdded++
} else {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: nil})
}
}
case "int":
@ -428,15 +437,17 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe
return errors.Wrap(err, "getting int field value for column")
}
valCount, ok := resp.Results[0].(pilosa.ValCount)
if ok && valCount.Count == 1 {
vals, err := h.api.TranslateIndexIDs(stream.Context(), fi, []uint64{uint64(valCount.Val)})
if err != nil {
return errors.Wrap(err, "getting keys for ids")
}
if len(vals) > 0 && vals[0] != "" {
value = vals[0]
exists = true
if len(resp.Results) > 0 {
valCount, ok := resp.Results[0].(pilosa.ValCount)
if ok && valCount.Count == 1 {
vals, err := h.api.TranslateIndexIDs(stream.Context(), fi, []uint64{uint64(valCount.Val)})
if err != nil {
return errors.Wrap(err, "getting keys for ids")
}
if len(vals) > 0 && vals[0] != "" {
value = vals[0]
exists = true
}
}
}
} else {
@ -448,6 +459,7 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe
if exists {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_StringVal{StringVal: value}})
colAdded++
} else {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: nil})
@ -463,13 +475,16 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe
return errors.Wrap(err, "getting int field value for column")
}
valCount, ok := resp.Results[0].(pilosa.ValCount)
if ok && valCount.Count == 1 {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Int64Val{Int64Val: valCount.Val}})
} else {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: nil})
if len(resp.Results) > 0 {
valCount, ok := resp.Results[0].(pilosa.ValCount)
if ok && valCount.Count == 1 {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Int64Val{Int64Val: valCount.Val}})
colAdded++
} else {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: nil})
}
}
}
@ -484,13 +499,16 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe
return errors.Wrap(err, "getting decimal field value for column")
}
valCount, ok := resp.Results[0].(pilosa.ValCount)
if ok && valCount.Count == 1 {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_DecimalVal{DecimalVal: &pb.Decimal{Value: valCount.DecimalVal.Value, Scale: valCount.DecimalVal.Scale}}})
} else {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: nil})
if len(resp.Results) > 0 {
valCount, ok := resp.Results[0].(pilosa.ValCount)
if ok && valCount.Count == 1 {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_DecimalVal{DecimalVal: &pb.Decimal{Value: valCount.DecimalVal.Value, Scale: valCount.DecimalVal.Scale}}})
colAdded++
} else {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: nil})
}
}
case "bool":
@ -504,21 +522,24 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe
return errors.Wrap(err, "querying rows for bool")
}
ids, ok := resp.Results[0].(pilosa.RowIdentifiers)
if !ok {
return errors.Wrap(err, "getting row identifiers")
}
if len(ids.Rows) == 1 {
var bval bool
if ids.Rows[0] == 1 {
bval = true
if len(resp.Results) > 0 {
ids, ok := resp.Results[0].(pilosa.RowIdentifiers)
if !ok {
return errors.Wrap(err, "getting row identifiers")
}
if len(ids.Rows) == 1 {
var bval bool
if ids.Rows[0] == 1 {
bval = true
}
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_BoolVal{BoolVal: bval}})
colAdded++
} else {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: nil})
}
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_BoolVal{BoolVal: bval}})
} else {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: nil})
}
case "time":
@ -527,8 +548,25 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe
}
}
if err := stream.Send(rowResp); err != nil {
return errors.Wrap(err, "sending response to stream")
// For SQL queries like:
// SELECT * FROM t WHERE _id=garbageID;
// we don't want to return any rows.
// So, check here if we added any columns.
//
// Because we don't have keys to translate
// and _id is an artificial field that's why for query:
// SELECT _id FROM t WHERE _id=existing-id;
// we return an empty result.
//
// TODO(kuba--): We need to find a way to check here if
// existing-id is not a garbage.
//
// A query which will work here is 'SELECT *' or any query with more columns
// than just _id.
if colAdded > 0 {
if err := stream.Send(rowResp); err != nil {
return errors.Wrap(err, "sending response to stream")
}
}
}
@ -546,6 +584,7 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe
return errToStatusError(errors.New("invalid key columns"))
}
forceSend := false
ci := []*pb.ColumnInfo{
{Name: "_id", Datatype: "string"},
}
@ -565,6 +604,11 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe
end = uint64(len(cols))
}
cols = cols[offset:end]
if len(cols) == 1 {
if id, err := h.api.TranslateIndexKey(stream.Context(), index.Name(), cols[0], false); id != 0 && err == nil {
forceSend = true
}
}
} else {
// Prevent getting too many records by forcing a limit.
pql := fmt.Sprintf("All(limit=%d, offset=%d)", limit, offset)
@ -577,18 +621,20 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe
return errors.Wrapf(err, "querying for all: %s", pql)
}
ids, ok := resp.Results[0].(*pilosa.Row)
if !ok {
return errors.Wrap(err, "getting results as a row")
}
if len(resp.Results) > 0 {
ids, ok := resp.Results[0].(*pilosa.Row)
if !ok {
return errors.Wrap(err, "getting results as a row")
}
limitedCols := ids.Keys
if len(limitedCols) == 0 {
// If cols is still empty after the limit/offset, then
// return with no results.
return nil
limitedCols := ids.Keys
if len(limitedCols) == 0 {
// If cols is still empty after the limit/offset, then
// return with no results.
return nil
}
cols = limitedCols
}
cols = limitedCols
}
for _, col := range cols {
@ -600,6 +646,7 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe
}
ci = nil // only include headers with the first row
colAdded := 0
for _, field := range fields {
// TODO: handle `time` fields
switch field.Type() {
@ -614,17 +661,21 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe
return errors.Wrap(err, "querying set rows(keys)")
}
ids, ok := resp.Results[0].(pilosa.RowIdentifiers)
if !ok {
return errors.Wrap(err, "getting row identifiers")
}
if len(resp.Results) > 0 {
ids, ok := resp.Results[0].(pilosa.RowIdentifiers)
if !ok {
return errors.Wrap(err, "getting row identifiers")
}
if len(ids.Keys) > 0 {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_StringArrayVal{StringArrayVal: &pb.StringArray{Vals: ids.Keys}}})
} else {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Uint64ArrayVal{Uint64ArrayVal: &pb.Uint64Array{Vals: ids.Rows}}})
if len(ids.Keys) > 0 {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_StringArrayVal{StringArrayVal: &pb.StringArray{Vals: ids.Keys}}})
colAdded++
} else if len(ids.Rows) > 0 {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Uint64ArrayVal{Uint64ArrayVal: &pb.Uint64Array{Vals: ids.Rows}}})
colAdded++
}
}
case "mutex":
@ -638,25 +689,29 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe
return errors.Wrap(err, "querying mutex rows(keys)")
}
ids, ok := resp.Results[0].(pilosa.RowIdentifiers)
if !ok {
return errors.Wrap(err, "getting row identifiers")
}
if len(resp.Results) > 0 {
ids, ok := resp.Results[0].(pilosa.RowIdentifiers)
if !ok {
return errors.Wrap(err, "getting row identifiers")
}
if len(ids.Keys) == 1 {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_StringVal{StringVal: ids.Keys[0]}})
} else if len(ids.Rows) == 1 {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Uint64Val{Uint64Val: ids.Rows[0]}})
} else {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: nil})
if len(ids.Keys) == 1 {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_StringVal{StringVal: ids.Keys[0]}})
colAdded++
} else if len(ids.Rows) == 1 {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Uint64Val{Uint64Val: ids.Rows[0]}})
colAdded++
} else {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: nil})
}
}
case "int":
// Translate column key.
id, err := h.api.TranslateIndexKey(stream.Context(), index.Name(), col)
id, err := h.api.TranslateIndexKey(stream.Context(), index.Name(), col, false)
if err != nil {
return errors.Wrap(err, "translating column key")
}
@ -677,15 +732,17 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe
return errors.Wrap(err, "getting int field value for column")
}
valCount, ok := resp.Results[0].(pilosa.ValCount)
if ok && valCount.Count == 1 {
vals, err := h.api.TranslateIndexIDs(stream.Context(), fi, []uint64{uint64(valCount.Val)})
if err != nil {
return errors.Wrap(err, "getting keys for ids")
}
if len(vals) > 0 && vals[0] != "" {
value = vals[0]
exists = true
if len(resp.Results) > 0 {
valCount, ok := resp.Results[0].(pilosa.ValCount)
if ok && valCount.Count == 1 {
vals, err := h.api.TranslateIndexIDs(stream.Context(), fi, []uint64{uint64(valCount.Val)})
if err != nil {
return errors.Wrap(err, "getting keys for ids")
}
if len(vals) > 0 && vals[0] != "" {
value = vals[0]
exists = true
}
}
}
} else {
@ -697,6 +754,7 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe
if exists {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_StringVal{StringVal: value}})
colAdded++
} else {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: nil})
@ -712,13 +770,16 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe
return errors.Wrap(err, "getting int field value for column")
}
valCount, ok := resp.Results[0].(pilosa.ValCount)
if ok && valCount.Count == 1 {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Int64Val{Int64Val: valCount.Val}})
} else {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: nil})
if len(resp.Results) > 0 {
valCount, ok := resp.Results[0].(pilosa.ValCount)
if ok && valCount.Count == 1 {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Int64Val{Int64Val: valCount.Val}})
colAdded++
} else {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: nil})
}
}
}
@ -733,13 +794,16 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe
return errors.Wrap(err, "getting decimal field value for column")
}
valCount, ok := resp.Results[0].(pilosa.ValCount)
if ok && valCount.Count == 1 {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_DecimalVal{DecimalVal: &pb.Decimal{Value: valCount.DecimalVal.Value, Scale: valCount.DecimalVal.Scale}}})
} else {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: nil})
if len(resp.Results) > 0 {
valCount, ok := resp.Results[0].(pilosa.ValCount)
if ok && valCount.Count == 1 {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_DecimalVal{DecimalVal: &pb.Decimal{Value: valCount.DecimalVal.Value, Scale: valCount.DecimalVal.Scale}}})
colAdded++
} else {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: nil})
}
}
case "bool":
@ -753,21 +817,24 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe
return errors.Wrap(err, "querying bool rows(keys)")
}
ids, ok := resp.Results[0].(pilosa.RowIdentifiers)
if !ok {
return errors.Wrap(err, "getting row identifiers")
}
if len(ids.Rows) == 1 {
var bval bool
if ids.Rows[0] == 1 {
bval = true
if len(resp.Results) > 0 {
ids, ok := resp.Results[0].(pilosa.RowIdentifiers)
if !ok {
return errors.Wrap(err, "getting row identifiers")
}
if len(ids.Rows) == 1 {
var bval bool
if ids.Rows[0] == 1 {
bval = true
}
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_BoolVal{BoolVal: bval}})
colAdded++
} else {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: nil})
}
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_BoolVal{BoolVal: bval}})
} else {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: nil})
}
case "time":
@ -776,8 +843,20 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe
}
}
if err := stream.Send(rowResp); err != nil {
return errors.Wrap(err, "sending response to stream")
// For SQL queries like:
// SELECT _id FROM parent WHERE _id="garbage";
// we get here without any real columns and fields, and we did not
// translate any keys. That's why we don't want to send anything back
// and return fake response like:
//
// _id
// -------
// <nil>
// (1 row)
if colAdded > 0 || forceSend {
if err := stream.Send(rowResp); err != nil {
return errors.Wrap(err, "sending response to stream")
}
}
}

31
translate.go
View file

@ -37,6 +37,7 @@ var (
ErrReplicationNotSupported = errors.New("replication not supported")
ErrTranslateStoreReadOnly = errors.New("translate store could not find or create key, translate store read only")
ErrTranslateStoreNotFound = errors.New("translate store not found")
ErrTranslatingKeyNotFound = errors.New("translating key not found")
ErrCannotOpenV1TranslateFile = errors.New("cannot open v1 translate .keys file")
)
@ -67,8 +68,8 @@ type TranslateStore interface {
//
// Translated id must be associated with a shard in the store's partition
// unless partition is set to -1.
TranslateKey(key string) (uint64, error)
TranslateKeys(key []string) ([]uint64, error)
TranslateKey(key string, writable bool) (uint64, error)
TranslateKeys(key []string, writable bool) ([]uint64, error)
// Converts an integer ID to its associated string key.
TranslateID(id uint64) (string, error)
@ -311,39 +312,43 @@ func (s *InMemTranslateStore) SetReadOnly(v bool) {
s.readOnly = v
}
// TranslateKeys converts a string key to an integer ID.
// TranslateKey converts a string key to an integer ID.
// If key does not have an associated id then one is created.
func (s *InMemTranslateStore) TranslateKey(key string) (uint64, error) {
func (s *InMemTranslateStore) TranslateKey(key string, writable bool) (uint64, error) {
s.mu.Lock()
defer s.mu.Unlock()
return s.translateKey(key)
return s.translateKey(key, writable)
}
// TranslateKeys converts a string key to an integer ID.
// If key does not have an associated id then one is created.
func (s *InMemTranslateStore) TranslateKeys(keys []string) (_ []uint64, err error) {
func (s *InMemTranslateStore) TranslateKeys(keys []string, writable bool) (_ []uint64, err error) {
s.mu.Lock()
defer s.mu.Unlock()
ids := make([]uint64, len(keys))
for i := range keys {
if ids[i], err = s.translateKey(keys[i]); err != nil {
if ids[i], err = s.translateKey(keys[i], writable); err != nil {
return ids, err
}
}
return ids, nil
}
func (s *InMemTranslateStore) translateKey(key string) (_ uint64, err error) {
// Return id if it has been added.
if id, ok := s.idsByKey[key]; ok {
func (s *InMemTranslateStore) translateKey(key string, writable bool) (_ uint64, err error) {
id := s.idsByKey[key]
if id != 0 {
// Return id if it has been added.
return id, nil
} else if s.readOnly {
return 0, nil
}
if s.readOnly {
return 0, ErrTranslatingKeyNotFound
}
if !writable {
return 0, ErrTranslatingKeyNotFound
}
// Generate a new id and update db.
var id uint64
if s.field == "" {
id = GenerateNextPartitionedID(s.index, s.maxID, s.partitionID, s.partitionN)
} else {

14
translator_test.go
View file

@ -36,21 +36,21 @@ func TestInMemTranslateStore_TranslateKey(t *testing.T) {
s := pilosa.NewInMemTranslateStore("IDX", "FLD", 0, pilosa.DefaultPartitionN)
// Ensure initial key translates to ID 1.
if id, err := s.TranslateKey("foo"); err != nil {
if id, err := s.TranslateKey("foo", true); err != nil {
t.Fatal(err)
} else if got, want := id, uint64(1); got != want {
t.Fatalf("TranslateKey()=%d, want %d", got, want)
}
// Ensure next key autoincrements.
if id, err := s.TranslateKey("bar"); err != nil {
if id, err := s.TranslateKey("bar", true); err != nil {
t.Fatal(err)
} else if got, want := id, uint64(2); got != want {
t.Fatalf("TranslateKey()=%d, want %d", got, want)
}
// Ensure retranslating existing key returns original ID.
if id, err := s.TranslateKey("foo"); err != nil {
if id, err := s.TranslateKey("foo", true); err != nil {
t.Fatal(err)
} else if got, want := id, uint64(1); got != want {
t.Fatalf("TranslateKey()=%d, want %d", got, want)
@ -61,9 +61,9 @@ func TestInMemTranslateStore_TranslateID(t *testing.T) {
s := pilosa.NewInMemTranslateStore("IDX", "FLD", 0, pilosa.DefaultPartitionN)
// Setup initial keys.
if _, err := s.TranslateKey("foo"); err != nil {
if _, err := s.TranslateKey("foo", true); err != nil {
t.Fatal(err)
} else if _, err := s.TranslateKey("bar"); err != nil {
} else if _, err := s.TranslateKey("bar", true); err != nil {
t.Fatal(err)
}
@ -289,7 +289,7 @@ func TestTranslation_Reset(t *testing.T) {
if err != nil {
t.Fatal(err)
}
if _, err := node0.API.TranslateKeys(ctx, bytes.NewReader(reqBody)); err != nil {
if _, err := node0.API.TranslateKeys(ctx, bytes.NewReader(reqBody), true); err != nil {
t.Fatal(err)
}
})
@ -558,7 +558,7 @@ func TestTranslation_Coordinator(t *testing.T) {
fld := "f"
// Create an index without keys.
if _, err := node0.API.CreateIndex(ctx, idx,
if _, err := node1.API.CreateIndex(ctx, idx,
pilosa.IndexOptions{
Keys: false,
}); err != nil {