scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-10-06 19:07:50 +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 GitHub
commit 475a9e3910
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
13 changed files with 513 additions and 287 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

20
api.go
View file

@ -1173,7 +1173,7 @@ func (api *API) ImportWithTx(ctx context.Context, tx Tx, req *ImportRequest, opt
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")
}
}
@ -1184,7 +1184,7 @@ func (api *API) ImportWithTx(ctx context.Context, tx Tx, req *ImportRequest, opt
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")
}
}
@ -1319,7 +1319,7 @@ func (api *API) ImportValueWithTx(ctx context.Context, tx Tx, req *ImportValueRe
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
@ -1330,7 +1330,7 @@ func (api *API) ImportValueWithTx(ctx context.Context, tx Tx, req *ImportValueRe
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
}
@ -1707,8 +1707,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) {
@ -1716,7 +1716,7 @@ func (api *API) TranslateIndexIDs(ctx context.Context, indexName string, ids []u
}
// TranslateKeys handles a TranslateKeyRequest.
func (api *API) TranslateKeys(ctx context.Context, r io.Reader) (_ []byte, err error) {
func (api *API) TranslateKeys(ctx context.Context, r io.Reader, writable bool) (_ []byte, err error) {
var req TranslateKeysRequest
if buf, err := ioutil.ReadAll(r); err != nil {
return nil, NewBadRequestError(errors.Wrap(err, "read translate keys request error"))
@ -1727,18 +1727,18 @@ 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 {
if ids, err = api.cluster.translateIndexKeys(ctx, req.Index, req.Keys, writable); err != nil {
return nil, err
}
} else {
if field := api.holder.Field(req.Index, req.Field); field == nil {
return nil, newNotFoundError(ErrFieldNotFound, req.Field)
} else if fi := field.ForeignIndex(); fi != "" {
ids, err = api.cluster.translateIndexKeys(ctx, fi, req.Keys)
ids, err = api.cluster.translateIndexKeys(ctx, fi, req.Keys, writable)
if err != nil {
return nil, err
}
} else if ids, err = api.cluster.translateFieldKeys(ctx, field, req.Keys...); err != nil {
} else if ids, err = api.cluster.translateFieldKeys(ctx, field, req.Keys, writable); err != nil {
return nil, errors.Wrapf(err, "translating field keys")
}
}

112
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
@ -162,64 +171,32 @@ 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 {
return 0, err
} else if id != 0 {
return id, nil
}
if s.ReadOnly() {
return 0, pilosa.ErrTranslateStoreReadOnly
}
// Find or create id under write lock.
var written bool
if err := s.db.Update(func(tx *bolt.Tx) (err error) {
bkt := tx.Bucket([]byte("keys"))
var boltKey []byte
if id, boltKey = findIDByKey(bkt, key); id != 0 {
return nil
}
id = pilosa.GenerateNextPartitionedID(s.index, maxID(tx), s.partitionID, s.partitionN)
if err := bkt.Put(boltKey, u64tob(id)); err != nil {
return err
} else if err := tx.Bucket([]byte("ids")).Put(u64tob(id), boltKey); err != nil {
return err
}
written = true
return nil
}); err != nil {
func (s *TranslateStore) TranslateKey(key string, writable bool) (uint64, error) {
ids, err := s.translateKeys([]string{key}, writable)
if 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.
// 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
}
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))
// Find ids by key under read lock.
var found int
if err := s.db.View(func(tx *bolt.Tx) error {
bkt := tx.Bucket([]byte("keys"))
func (s *TranslateStore) translateKeys(keys []string, writable bool) ([]uint64, error) {
found := 0
ids := make([]uint64, len(keys))
err := s.db.View(func(tx *bolt.Tx) error {
bkt := tx.Bucket(bucketKeys)
if bkt == nil {
return errors.Errorf(errFmtTranslateBucketNotFound, bucketKeys)
}
for i, key := range keys {
if id, _ := findIDByKey(bkt, key); id != 0 {
ids[i] = id
@ -227,34 +204,36 @@ func (s *TranslateStore) TranslateKeys(keys []string) (ids []uint64, _ error) {
}
}
return nil
}); err != nil {
})
if err != nil {
return nil, err
} else if found == len(keys) {
}
if found == len(keys) {
return ids, nil
}
if s.ReadOnly() {
return ids, pilosa.ErrTranslateStoreReadOnly
}
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) {
bkt := tx.Bucket([]byte("keys"))
bkt := tx.Bucket(bucketKeys)
for i, key := range keys {
if ids[i] != 0 {
continue
}
var boltKey []byte
if ids[i], boltKey = findIDByKey(bkt, key); ids[i] != 0 {
continue
}
ids[i] = pilosa.GenerateNextPartitionedID(s.index, maxID(tx), s.partitionID, s.partitionN)
if err := bkt.Put(boltKey, u64tob(ids[i])); err != nil {
return err
} else if err := tx.Bucket([]byte("ids")).Put(u64tob(ids[i]), boltKey); err != nil {
} else if err := tx.Bucket(bucketIDs).Put(u64tob(ids[i]), boltKey); err != nil {
return err
}
written = true
@ -263,7 +242,6 @@ func (s *TranslateStore) TranslateKeys(keys []string) (ids []uint64, _ error) {
}); err != nil {
return nil, err
}
if written {
s.notifyWrite()
}
@ -278,7 +256,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.
@ -295,7 +273,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
}
@ -303,9 +281,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
@ -398,7 +376,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
@ -436,7 +414,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])

24
cluster.go
View file

@ -2341,8 +2341,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 {
@ -2357,8 +2357,8 @@ func (c *cluster) translateFieldKey(ctx context.Context, field *Field, key strin
// 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)
func (c *cluster) translateFieldKeys(ctx context.Context, field *Field, keys []string, writable bool) ([]uint64, error) {
ids, err := field.TranslateStore().TranslateKeys(keys, writable)
// If we get a "read only" error, then forward the request
// to the coordinator.
if errors.Cause(err) == ErrTranslateStoreReadOnly {
@ -2372,21 +2372,21 @@ func (c *cluster) translateFieldKeys(ctx context.Context, field *Field, keys ...
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{}{}})
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
}
@ -2398,7 +2398,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)
@ -2423,7 +2423,7 @@ func (c *cluster) translateIndexKeySet(ctx context.Context, indexName string, ke
g.Go(func() (err error) {
var ids []uint64
if c.ownsPartition(c.Node.ID, partitionID) {
if ids, err = idx.TranslateStore(partitionID).TranslateKeys(keys); err != nil {
if ids, err = idx.TranslateStore(partitionID).TranslateKeys(keys, writable); err != nil {
return err
}
} else {
@ -2436,7 +2436,9 @@ func (c *cluster) translateIndexKeySet(ctx context.Context, indexName string, ke
mu.Lock()
defer mu.Unlock()
for i := range keys {
keyMap[keys[i]] = ids[i]
if id := ids[i]; id != 0 {
keyMap[keys[i]] = id
}
}
return nil
})

69
executor.go
View file

@ -84,6 +84,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{
@ -165,6 +183,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
@ -238,6 +264,14 @@ func (e *executor) Execute(ctx context.Context, index string, q *pql.Query, shar
if !opt.Remote {
// only translateResults if this local node is the final destination. only string/column keys.
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
@ -630,7 +664,7 @@ func (e *executor) preprocessQuery(ctx context.Context, tx Tx, index string, c *
// Translate keys to IDs.
if len(keys) > 0 {
keyIDs, err := e.Cluster.translateIndexKeys(ctx, index, keys)
keyIDs, err := e.Cluster.translateIndexKeys(ctx, index, keys, false)
if err != nil {
return nil, errors.Wrap(err, "translating column IDs in ConstRow")
}
@ -953,7 +987,7 @@ func (e *executor) executeFieldValueCall(ctx context.Context, tx Tx, index strin
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")
}
@ -3888,7 +3922,7 @@ func (e *executor) executeSetRow(ctx context.Context, tx Tx, indexName string, c
if !field.Keys() {
return false, errors.New("can't Store() a key into an unkeyed field")
}
id, err := field.TranslateStore().TranslateKey(argKey)
id, err := field.TranslateStore().TranslateKey(argKey, true)
if err != nil {
return false, errors.Wrap(err, "translating a key for Store()")
}
@ -4652,9 +4686,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
}
}
@ -4670,14 +4707,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
}
}
@ -4744,7 +4781,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
@ -4814,7 +4851,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)
}
}
@ -4837,7 +4874,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)
}
}
@ -4852,7 +4889,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
}
}
@ -4860,7 +4897,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")
}
}
@ -4905,7 +4942,7 @@ func (e *executor) translateCall(ctx context.Context, indexName string, c *pql.C
var id uint64
var err error
if fi := field.ForeignIndex(); fi != "" {
ids, err := e.Cluster.translateIndexKeys(ctx, fi, []string{prevStr})
ids, err := e.Cluster.translateIndexKeys(ctx, fi, []string{prevStr}, writable)
if err != nil {
return errors.Wrap(err, "translating foreign index key in groupby previous")
}
@ -4913,7 +4950,7 @@ func (e *executor) translateCall(ctx context.Context, indexName string, c *pql.C
id = ids[0]
}
} else {
id, err = e.Cluster.translateFieldKey(ctx, field, prevStr)
id, err = e.Cluster.translateFieldKey(ctx, field, prevStr, writable)
if err != nil {
return errors.Wrapf(err, "translating field key: %s", prevStr)
}

8
executor_internal_test.go
View file

@ -59,7 +59,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)
}
@ -123,7 +125,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)
}
@ -191,7 +193,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

@ -5230,7 +5230,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)
}
}
}
}
})

2
http/handler.go
View file

@ -2331,7 +2331,7 @@ func (h *Handler) handlePostTranslateKeys(w http.ResponseWriter, r *http.Request
return
}
buf, err := h.api.TranslateKeys(r.Context(), r.Body)
buf, err := h.api.TranslateKeys(r.Context(), r.Body, true)
if err != nil {
http.Error(w, fmt.Sprintf("translate keys: %v", err), http.StatusInternalServerError)
return

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

@ -801,6 +801,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

@ -476,6 +476,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() {
@ -490,17 +491,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":
@ -514,20 +519,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":
@ -547,15 +556,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 {
@ -567,6 +578,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})
@ -582,13 +594,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})
}
}
}
@ -603,13 +618,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":
@ -623,21 +641,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":
@ -646,8 +667,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")
}
}
}
@ -665,6 +703,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"},
}
@ -688,6 +727,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)
@ -700,18 +744,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 {
@ -723,6 +769,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() {
@ -737,17 +784,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":
@ -761,25 +812,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")
}
@ -800,15 +855,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 {
@ -820,6 +877,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})
@ -835,13 +893,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})
}
}
}
@ -856,13 +917,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":
@ -876,21 +940,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":
@ -899,8 +966,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 {

43
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)
}
@ -290,12 +290,41 @@ 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)
}
})
}
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) {
@ -406,7 +435,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 {