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

Merge pull request #890 from molecula/v2.1.4

Browse source 
V2.1.4
This commit is contained in:
Matthew Jaffee 2020-09-23 13:25:50 -05:00 • committed by GitHub
commit 75d857ead2
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
25 changed files with 841 additions and 451 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

5
apimethod_string.go
View file

@ -38,11 +38,12 @@ func _() {
_ = x[apiFinishTransaction-27]
_ = x[apiTransactions-28]
_ = x[apiGetTransaction-29]
_ = x[apiActiveQueries-30]
}
const _apiMethod_name = "apiClusterMessageapiCreateFieldapiCreateIndexapiDeleteFieldapiDeleteAvailableShardapiDeleteIndexapiDeleteViewapiExportCSVapiFragmentBlockDataapiFragmentBlocksapiFragmentDataapiTranslateDataapiFieldapiFieldAttrDiffapiImportapiImportValueapiIndexapiIndexAttrDiffapiQueryapiRecalculateCachesapiRemoveNodeapiResizeAbortapiSetCoordinatorapiShardNodesapiViewsapiApplySchemaapiStartTransactionapiFinishTransactionapiTransactionsapiGetTransaction"
const _apiMethod_name = "apiClusterMessageapiCreateFieldapiCreateIndexapiDeleteFieldapiDeleteAvailableShardapiDeleteIndexapiDeleteViewapiExportCSVapiFragmentBlockDataapiFragmentBlocksapiFragmentDataapiTranslateDataapiFieldapiFieldAttrDiffapiImportapiImportValueapiIndexapiIndexAttrDiffapiQueryapiRecalculateCachesapiRemoveNodeapiResizeAbortapiSetCoordinatorapiShardNodesapiViewsapiApplySchemaapiStartTransactionapiFinishTransactionapiTransactionsapiGetTransactionapiActiveQueries"
var _apiMethod_index = [...]uint16{0, 17, 31, 45, 59, 82, 96, 109, 121, 141, 158, 173, 189, 197, 213, 222, 236, 244, 260, 268, 288, 301, 315, 332, 345, 353, 367, 386, 406, 421, 438}
var _apiMethod_index = [...]uint16{0, 17, 31, 45, 59, 82, 96, 109, 121, 141, 158, 173, 189, 197, 213, 222, 236, 244, 260, 268, 288, 301, 315, 332, 345, 353, 367, 386, 406, 421, 438, 454}
func (i apiMethod) String() string {
if i < 0 || i >= apiMethod(len(_apiMethod_index)-1) {

153
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
@ -161,109 +170,81 @@ 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 {
// If key does not have an associated id then one is created, unless writable is false,
// then the function will return the error pilosa.ErrTranslatingKeyNotFound.
func (s *TranslateStore) TranslateKey(key string, writable bool) (uint64, error) {
ids, err := s.translateKeys([]string{key}, writable)
if err != nil {
return 0, err
}
if written {
s.notifyWrite()
if len(ids) == 0 {
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
}
// If a key does not have an associated id then one is created, unless writable is false,
// then the function will return the error pilosa.ErrTranslatingKeyNotFound.
func (s *TranslateStore) TranslateKeys(keys []string, writable bool) ([]uint64, error) {
return s.translateKeys(keys, writable)
}
// Allocate slice for ID mapping.
ids = make([]uint64, len(keys))
func (s *TranslateStore) translateKeys(keys []string, writable bool) ([]uint64, error) {
ids := make([]uint64, 0, 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"))
for i, key := range keys {
if id, _ := findIDByKey(bkt, key); id != 0 {
ids[i] = id
found++
if s.ReadOnly() || !writable {
found := 0
if err := s.db.View(func(tx *bolt.Tx) error {
bkt := tx.Bucket(bucketKeys)
if bkt == nil {
return errors.Errorf(errFmtTranslateBucketNotFound, bucketKeys)
}
for _, key := range keys {
if id, _ := findIDByKey(bkt, key); id != 0 {
ids = append(ids, id)
found++
}
}
return nil
}); err != nil {
return nil, err
}
return nil
}); err != nil {
return nil, err
} else if found == len(keys) {
return ids, nil
}
if s.ReadOnly() {
return ids, pilosa.ErrTranslateStoreReadOnly
if found == len(keys) {
return ids, nil
}
if s.ReadOnly() {
return ids, pilosa.ErrTranslateStoreReadOnly
}
if !writable {
return nil, pilosa.ErrTranslatingKeyNotFound
}
return nil, nil
}
// 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"))
for i, key := range keys {
if ids[i] != 0 {
bkt := tx.Bucket(bucketKeys)
for _, key := range keys {
id, boltKey := findIDByKey(bkt, key)
if id != 0 {
ids = append(ids, id)
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 {
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(ids[i]), boltKey); err != nil {
} else if err := tx.Bucket(bucketIDs).Put(u64tob(id), boltKey); err != nil {
return err
}
ids = append(ids, id)
written = true
}
return nil
}); err != nil {
return nil, err
}
if written {
s.notifyWrite()
}
@ -278,7 +259,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.
@ -293,9 +274,11 @@ func (s *TranslateStore) TranslateIDs(ids []uint64) ([]string, error) {
}
defer func() { _ = tx.Rollback() }()
bucket := tx.Bucket(bucketIDs)
keys := make([]string, len(ids))
for i, id := range ids {
keys[i] = findKeyByID(tx.Bucket([]byte("ids")), id)
keys[i] = findKeyByID(bucket, id)
}
return keys, nil
}
@ -303,9 +286,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 +381,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 +419,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])

8
client.go
View file

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

133
cluster.go
View file

@ -1085,16 +1085,18 @@ func (c *cluster) partitionNodes(partitionID int) []*Node {
return nodes
}
// ownsPartition returns true if a host owns a partition.
func (c *cluster) ownsPartition(nodeID string, partition int) bool {
func (c *cluster) primaryPartitionNode(partition int) *Node {
c.mu.RLock()
defer c.mu.RUnlock()
return c.unprotectedOwnsPartition(nodeID, partition)
return c.unprotectedPrimaryPartitionNode(partition)
}
// unprotectedOwnsPartition returns true if a host owns a partition.
func (c *cluster) unprotectedOwnsPartition(nodeID string, partition int) bool {
return Nodes(c.partitionNodes(partition)).ContainsID(nodeID)
// unprotectedPrimaryPartition returns tprimary node of partition.
func (c *cluster) unprotectedPrimaryPartitionNode(partition int) *Node {
if nodes := c.partitionNodes(partition); len(nodes) > 0 {
return nodes[0]
}
return nil
}
// containsShards is like OwnsShards, but it includes replicas.
@ -2323,64 +2325,76 @@ 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 {
return 0, errors.New("translating key on coordinator returned empty set")
return 0, nil
}
return ids[0], nil
}
// translateFieldKeys is basically a wrapper around
// field.TranslateStore().TranslateKey(key), but in
// the case where the local node's translate store
// is read-only (i.e. it's not the primary translate
// store), then this method will forward the translation
// the case where the local node is not coordinator, then this method will forward the translation
// request to the coordinator.
func (c *cluster) translateFieldKeys(ctx context.Context, field *Field, keys ...string) ([]uint64, error) {
ids, err := field.TranslateStore().TranslateKeys(keys)
// If we get a "read only" error, then forward the request
// to the coordinator.
if errors.Cause(err) == ErrTranslateStoreReadOnly {
coordinatorNode := c.coordinatorNode()
if ids, err := c.InternalClient.TranslateKeysNode(ctx, &coordinatorNode.URI, field.Index(), field.Name(), keys); err != nil {
return ids, errors.Wrap(err, "translating keys on coordinator")
} else {
return ids, nil
}
func (c *cluster) translateFieldKeys(ctx context.Context, field *Field, keys []string, writable bool) (ids []uint64, err error) {
coordinator := c.coordinatorNode()
if coordinator == nil {
return nil, errors.Errorf("translating field(%s/%s) keys(%v) - cannot find coordinator node", field.Index(), field.Name(), keys)
}
return ids, err
if c.Node.ID == coordinator.ID {
ids, err = field.TranslateStore().TranslateKeys(keys, writable)
} else {
// If it's writable, then forward the request to the coordinator.
ids, err = c.InternalClient.TranslateKeysNode(ctx, &coordinator.URI, field.Index(), field.Name(), keys, writable)
}
if err != nil {
return nil, errors.Wrapf(err, "translating field(%s/%s) keys(%v)", field.Index(), field.Name(), keys)
}
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
}
// make sure that ids line up with keys, but
// not appending, but assigning directly 1:1 into the slice.
ids := make([]uint64, len(keys))
for i := range keys {
ids[i] = keyMap[keys[i]]
for i, k := range keys {
id, ok := keyMap[k]
if !writable {
if !ok || id == 0 {
c.holder.Logger.Debugf("internal translateIndexKeys error: keyMap had no entry for k='%v', and was not writable", k)
return nil, ErrTranslatingKeyNotFound
}
}
ids[i] = id
}
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)
@ -2404,22 +2418,29 @@ 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 {
return err
}
primary := c.primaryPartitionNode(partitionID)
if primary == nil {
return errors.Errorf("translating index(%s) keys(%v) on partition(%d) - cannot find primary node", indexName, keys, partitionID)
}
if c.Node.ID == primary.ID {
ids, err = idx.TranslateStore(partitionID).TranslateKeys(keys, writable)
} else {
nodes := c.partitionNodes(partitionID)
if ids, err = c.InternalClient.TranslateKeysNode(ctx, &nodes[0].URI, indexName, "", keys); err != nil {
return err
}
ids, err = c.InternalClient.TranslateKeysNode(ctx, &primary.URI, indexName, "", keys, writable)
}
if err != nil {
return errors.Wrapf(err, "translating index(%s) keys(%v) on partition(%d)", indexName, keys, partitionID)
}
mu.Lock()
defer mu.Unlock()
for i := range keys {
keyMap[keys[i]] = ids[i]
for i, id := range ids {
if id != 0 {
keyMap[keys[i]] = id
}
}
mu.Unlock()
return nil
})
}
@ -2471,22 +2492,28 @@ func (c *cluster) translateIndexIDSet(ctx context.Context, indexName string, idS
g.Go(func() (err error) {
var keys []string
if c.ownsPartition(c.Node.ID, partitionID) {
if keys, err = index.TranslateStore(partitionID).TranslateIDs(ids); err != nil {
return err
}
primary := c.primaryPartitionNode(partitionID)
if primary == nil {
return errors.Errorf("translating index(%s) ids(%v) on partition(%d) - cannot find primary node", indexName, ids, partitionID)
}
if c.Node.ID == primary.ID {
keys, err = index.TranslateStore(partitionID).TranslateIDs(ids)
} else {
nodes := c.partitionNodes(partitionID)
if keys, err = c.InternalClient.TranslateIDsNode(ctx, &nodes[0].URI, indexName, "", ids); err != nil {
return err
}
keys, err = c.InternalClient.TranslateIDsNode(ctx, &primary.URI, indexName, "", ids)
}
if err != nil {
return errors.Wrapf(err, "translating index(%s) ids(%v) on partition(%d)", indexName, ids, partitionID)
}
mu.Lock()
defer mu.Unlock()
for i := range ids {
idMap[ids[i]] = keys[i]
for i, id := range ids {
idMap[id] = keys[i]
}
mu.Unlock()
return nil
})
}

8
encoding/proto/proto.go
View file

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

63
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")
}
}
@ -4164,7 +4201,7 @@ func (e *executor) translateCall(ctx context.Context, indexName string, c *pql.C
return errors.New("prev value must be a string when field 'keys' option enabled")
}
// TODO: does this need to take field.ForeignIndex() into consideration?
id, err := 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

@ -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)
}
}
}
}
})

2
go.mod
View file

@ -35,11 +35,11 @@ require (
github.com/uber-go/atomic v1.4.0 // indirect
github.com/uber/jaeger-client-go v2.16.0+incompatible
github.com/uber/jaeger-lib v2.2.0+incompatible // indirect
github.com/zeebo/blake3 v0.0.4
go.uber.org/atomic v1.4.0 // indirect
golang.org/x/crypto v0.0.0-20190426145343-a29dc8fdc734 // indirect
golang.org/x/net v0.0.0-20190424112056-4829fb13d2c6 // indirect
golang.org/x/sync v0.0.0-20190423024810-112230192c58
golang.org/x/sys v0.0.0-20190429190828-d89cdac9e872 // indirect
golang.org/x/text v0.3.2 // indirect
google.golang.org/grpc v1.28.0
modernc.org/mathutil v1.0.0

7
go.sum
View file

@ -168,6 +168,11 @@ github.com/uber/jaeger-lib v2.2.0+incompatible h1:MxZXOiR2JuoANZ3J6DE/U0kSFv/eJ/
github.com/uber/jaeger-lib v2.2.0+incompatible/go.mod h1:ComeNDZlWwrWnDv8aPp0Ba6+uUTzImX/AauajbLI56U=
github.com/ugorji/go/codec v0.0.0-20181204163529-d75b2dcb6bc8/go.mod h1:VFNgLljTbGfSG7qAOspJ7OScBnGdDN/yBr0sguwnwf0=
github.com/xordataexchange/crypt v0.0.3-0.20170626215501-b2862e3d0a77/go.mod h1:aYKd//L2LvnjZzWKhF00oedf4jCCReLcmhLdhm1A27Q=
github.com/zeebo/assert v0.0.0-20181109011804-10f827ce2ed6/go.mod h1:yssERNPivllc1yU3BvpjYI5BUW+zglcz6QWqeVRL5t0=
github.com/zeebo/assert v1.1.0/go.mod h1:Pq9JiuJQpG8JLJdtkwrJESF0Foym2/D9XMU5ciN/wJ0=
github.com/zeebo/blake3 v0.0.4 h1:vtZ4X8B2lKXZFg2Xyg6Wo36mvmnJvc2VQYTtA4RDCkI=
github.com/zeebo/blake3 v0.0.4/go.mod h1:YOZo8A49yNqM0X/Y+JmDUZshJWLt1laHsNSn5ny2i34=
github.com/zeebo/pcg v0.0.0-20181207190024-3cdc6b625a05/go.mod h1:Gr+78ptB0MwXxm//LBaEvBiaXY7hXJ6KGe2V32X2F6E=
go.uber.org/atomic v1.4.0 h1:cxzIVoETapQEqDhQu3QfnvXAV4AlzcvUCxkVUFw3+EU=
go.uber.org/atomic v1.4.0/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
@ -210,6 +215,8 @@ golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5h
golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20190429190828-d89cdac9e872 h1:cGjJzUd8RgBw428LXP65YXni0aiGNA4Bl+ls8SmLOm8=
golang.org/x/sys v0.0.0-20190429190828-d89cdac9e872/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20200202164722-d101bd2416d5 h1:LfCXLvNmTYH9kEmVgqbnsWfruoXZIrh4YBgqVHtDvw0=
golang.org/x/sys v0.0.0-20200202164722-d101bd2416d5/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/text v0.3.0 h1:g61tztE5qeGQ89tm6NTjjM9VPIm088od1l6aSorWRWg=
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
golang.org/x/text v0.3.2 h1:tW2bmiBqwgJj/UpqtC8EpXEZVYOwU0yG4iWbprSVAcs=

3
handler.go
View file

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

6
holder.go
View file

@ -1425,9 +1425,11 @@ func (s *holderSyncer) setTranslateReadOnlyFlags() {
// done using it.
index.mu.RLock()
for partitionID := 0; partitionID < s.Cluster.partitionN; partitionID++ {
ownsPartition := s.Cluster.unprotectedOwnsPartition(s.Node.ID, partitionID)
primary := s.Cluster.unprotectedPrimaryPartitionNode(partitionID)
isPrimary := primary != nil && s.Node.ID == primary.ID
if ts := index.TranslateStore(partitionID); ts != nil {
ts.SetReadOnly(!ownsPartition)
ts.SetReadOnly(!isPrimary)
}
}
index.mu.RUnlock()

15
http/client.go
View file

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

23
http/handler.go
View file

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

244
internal/public.pb.go
View file

@ -1185,6 +1185,7 @@ type ImportRequest struct {
Timestamps []int64 `protobuf:"varint,6,rep,packed,name=Timestamps,proto3" json:"Timestamps,omitempty"`
IndexCreatedAt int64 `protobuf:"varint,9,opt,name=IndexCreatedAt,proto3" json:"IndexCreatedAt,omitempty"`
FieldCreatedAt int64 `protobuf:"varint,10,opt,name=FieldCreatedAt,proto3" json:"FieldCreatedAt,omitempty"`
Clear bool `protobuf:"varint,11,opt,name=Clear,proto3" json:"Clear,omitempty"`
XXX_NoUnkeyedLiteral struct{} `json:"-"`
XXX_unrecognized []byte `json:"-"`
XXX_sizecache int32 `json:"-"`
@ -1293,6 +1294,13 @@ func (m *ImportRequest) GetFieldCreatedAt() int64 {
return 0
}
func (m *ImportRequest) GetClear() bool {
if m != nil {
return m.Clear
}
return false
}
type ImportValueRequest struct {
Index string `protobuf:"bytes,1,opt,name=Index,proto3" json:"Index,omitempty"`
Field string `protobuf:"bytes,2,opt,name=Field,proto3" json:"Field,omitempty"`
@ -1416,6 +1424,7 @@ type TranslateKeysRequest struct {
Index string `protobuf:"bytes,1,opt,name=Index,proto3" json:"Index,omitempty"`
Field string `protobuf:"bytes,2,opt,name=Field,proto3" json:"Field,omitempty"`
Keys []string `protobuf:"bytes,3,rep,name=Keys,proto3" json:"Keys,omitempty"`
NotWritable bool `protobuf:"varint,4,opt,name=NotWritable,proto3" json:"NotWritable,omitempty"`
XXX_NoUnkeyedLiteral struct{} `json:"-"`
XXX_unrecognized []byte `json:"-"`
XXX_sizecache int32 `json:"-"`
@ -1475,6 +1484,13 @@ func (m *TranslateKeysRequest) GetKeys() []string {
return nil
}
func (m *TranslateKeysRequest) GetNotWritable() bool {
if m != nil {
return m.NotWritable
}
return false
}
type TranslateKeysResponse struct {
IDs []uint64 `protobuf:"varint,3,rep,packed,name=IDs,proto3" json:"IDs,omitempty"`
XXX_NoUnkeyedLiteral struct{} `json:"-"`
@ -1893,86 +1909,88 @@ func init() {
func init() { proto.RegisterFile("public.proto", fileDescriptor_413a91106d7bcce8) }
var fileDescriptor_413a91106d7bcce8 = []byte{
// 1258 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xac, 0x57, 0xcd, 0x8e, 0x1b, 0x45,
0x10, 0x66, 0x3c, 0xe3, 0xbf, 0xb2, 0x77, 0x13, 0x3a, 0x4e, 0x18, 0xa1, 0xb0, 0xb1, 0x46, 0x01,
0x19, 0x0e, 0x1b, 0x6d, 0x08, 0x51, 0x4e, 0x40, 0x36, 0xde, 0x80, 0x15, 0x65, 0x15, 0xca, 0x2b,
0x73, 0x43, 0x9a, 0xb5, 0x9b, 0xcd, 0x88, 0xf1, 0x8c, 0x99, 0x1f, 0x9c, 0x3d, 0xf2, 0x0c, 0x5c,
0x78, 0x04, 0xae, 0xbc, 0x02, 0x27, 0x8e, 0x3c, 0x02, 0x5a, 0x38, 0xf3, 0x02, 0x5c, 0x50, 0x55,
0x4f, 0xbb, 0xc7, 0xde, 0xd9, 0xcd, 0x2a, 0xe2, 0xd6, 0x5f, 0x55, 0x4d, 0x75, 0xd5, 0xd7, 0xd5,
0x55, 0x3d, 0xd0, 0x5d, 0xe4, 0xc7, 0x61, 0x30, 0xdd, 0x5d, 0x24, 0x71, 0x16, 0x8b, 0x56, 0x10,
0x65, 0x32, 0x89, 0xfc, 0xd0, 0x4b, 0xc1, 0xc6, 0x78, 0x29, 0x5c, 0x68, 0x3e, 0x89, 0xc3, 0x7c,
0x1e, 0xa5, 0xae, 0xd5, 0xb7, 0x07, 0x0e, 0x6a, 0x28, 0x04, 0x38, 0xcf, 0xe4, 0x69, 0xea, 0xda,
0x7d, 0x7b, 0xd0, 0x46, 0x5e, 0x8b, 0xbb, 0x50, 0x7f, 0x9c, 0x65, 0x49, 0xea, 0xd6, 0xfa, 0xf6,
0xa0, 0x73, 0x7f, 0x7b, 0x57, 0xbb, 0xdb, 0x25, 0x31, 0x2a, 0x25, 0xf9, 0xc4, 0xd8, 0x4f, 0x82,
0xe8, 0xc4, 0x75, 0xfa, 0xd6, 0xa0, 0x8b, 0x1a, 0x7a, 0xcf, 0xa1, 0x3d, 0x0e, 0x4e, 0x22, 0x39,
0xa3, 0xad, 0xef, 0x80, 0xfd, 0x22, 0xa6, 0x6d, 0xad, 0x41, 0xe7, 0xfe, 0x96, 0x71, 0x85, 0xf1,
0x12, 0x49, 0x43, 0x06, 0x87, 0xf2, 0xc4, 0xad, 0x55, 0x1a, 0x1c, 0xca, 0x13, 0xef, 0x11, 0x6c,
0x63, 0xbc, 0x1c, 0xcd, 0x64, 0x94, 0x05, 0xdf, 0x06, 0x32, 0xe1, 0xa0, 0x31, 0x5e, 0xea, 0x5c,
0x78, 0xbd, 0x4a, 0xa4, 0x66, 0x12, 0xf1, 0x3e, 0x05, 0xe7, 0x85, 0x1f, 0x24, 0x62, 0x1b, 0x6a,
0xa3, 0x21, 0x87, 0xe0, 0x60, 0x6d, 0x34, 0x14, 0xd7, 0xc1, 0x7e, 0x26, 0x4f, 0x5d, 0xbb, 0x6f,
0x0d, 0xda, 0x48, 0x4b, 0xd1, 0x83, 0xfa, 0x93, 0x38, 0x8f, 0x32, 0x0e, 0xc3, 0x41, 0x05, 0xbc,
0x03, 0x68, 0xd3, 0xf7, 0x4f, 0x03, 0x19, 0xce, 0x84, 0xa7, 0x9c, 0x15, 0x99, 0x94, 0x48, 0x21,
0x29, 0xaa, 0x8d, 0x7a, 0x50, 0x67, 0x63, 0x76, 0xd3, 0x46, 0x05, 0xbc, 0x2f, 0x01, 0x48, 0x9b,
0x2a, 0x3f, 0x77, 0xa1, 0xce, 0x88, 0xa3, 0x3f, 0xef, 0x48, 0x29, 0x2f, 0xf0, 0xf4, 0x1e, 0xd4,
0x47, 0x51, 0xf6, 0xf0, 0x01, 0xa9, 0x27, 0x7e, 0x98, 0x4b, 0x8e, 0xc6, 0x46, 0x05, 0xbc, 0x1c,
0x5a, 0x6c, 0x47, 0xbc, 0xaf, 0x1c, 0x58, 0x25, 0x07, 0x24, 0x25, 0x2e, 0x87, 0x3a, 0x4f, 0x06,
0xe2, 0x16, 0x34, 0x30, 0x5e, 0x1a, 0x4a, 0x0a, 0x24, 0xde, 0xd7, 0xbb, 0x38, 0x9c, 0xf3, 0x35,
0x13, 0x2a, 0x47, 0xa1, 0xb7, 0xfd, 0x06, 0xe0, 0x8b, 0x24, 0xce, 0x17, 0x4c, 0x9a, 0x18, 0x40,
0x9d, 0x51, 0x91, 0x9f, 0x30, 0x1f, 0xe9, 0xd8, 0x50, 0x19, 0x54, 0x93, 0x4e, 0x87, 0x33, 0xce,
0xe7, 0x1c, 0x89, 0x8d, 0xb4, 0xf4, 0x7e, 0xb4, 0xa0, 0x35, 0xf1, 0xc3, 0x95, 0x7a, 0xe2, 0x87,
0x45, 0xde, 0xb4, 0x5c, 0x77, 0x63, 0x6b, 0x37, 0xef, 0x42, 0xeb, 0x69, 0x18, 0xfb, 0x19, 0x19,
0x93, 0x2f, 0x0b, 0x57, 0x58, 0xec, 0x01, 0x0c, 0xe5, 0x34, 0x98, 0xfb, 0x21, 0x69, 0x55, 0x72,
0x6f, 0x9b, 0x38, 0x0b, 0x1d, 0x96, 0x8c, 0xbc, 0x4f, 0xa0, 0x59, 0xa0, 0x6a, 0xee, 0x49, 0x3a,
0x9e, 0xfa, 0xa1, 0xd4, 0x51, 0x30, 0xf0, 0xbe, 0x86, 0x2d, 0x75, 0xd3, 0xe8, 0xce, 0x8c, 0x65,
0x76, 0x85, 0x52, 0xbc, 0xd2, 0xed, 0xf3, 0x7e, 0xb1, 0xc0, 0xa1, 0x95, 0x76, 0x60, 0x19, 0x07,
0x02, 0x9c, 0xa3, 0xd3, 0x85, 0x2c, 0x58, 0xe5, 0xb5, 0xe8, 0x43, 0x67, 0x9c, 0xd1, 0xe5, 0x54,
0x91, 0xab, 0xed, 0xca, 0x22, 0xe2, 0x6b, 0x14, 0x65, 0xe6, 0xb8, 0x6d, 0x5c, 0x61, 0x71, 0x1b,
0xda, 0xfb, 0x71, 0x1c, 0x2a, 0x65, 0xbd, 0x6f, 0x0d, 0x5a, 0x68, 0x04, 0x62, 0x07, 0x40, 0x33,
0x9b, 0x4b, 0xb7, 0xc1, 0x5c, 0x97, 0x24, 0xde, 0x3d, 0x68, 0x52, 0xa4, 0xcf, 0xfd, 0x85, 0xc9,
0xcd, 0xba, 0x2c, 0xb7, 0x7f, 0x2d, 0xe8, 0x7e, 0x95, 0xcb, 0xe4, 0x14, 0xe5, 0xf7, 0xb9, 0x4c,
0x33, 0xe2, 0x96, 0xb1, 0xae, 0x65, 0x06, 0x54, 0xb5, 0xe3, 0x97, 0x7e, 0x32, 0x53, 0x4c, 0x39,
0x58, 0x20, 0xca, 0xd5, 0x70, 0x9e, 0x72, 0xae, 0x2d, 0x2c, 0x8b, 0xb8, 0xde, 0xe5, 0x3c, 0xce,
0x74, 0x32, 0x05, 0x12, 0x03, 0xb8, 0x76, 0xf0, 0x6a, 0x1a, 0xe6, 0x33, 0x89, 0xf1, 0x52, 0x7d,
0xdd, 0x60, 0x83, 0x4d, 0xb1, 0xf8, 0x00, 0xb6, 0x0b, 0x91, 0xee, 0xab, 0x4d, 0x36, 0xdc, 0x90,
0x8a, 0x3d, 0xe8, 0x1e, 0xcc, 0x8f, 0xe5, 0x6c, 0x26, 0x67, 0x43, 0x3f, 0xf3, 0xdd, 0x16, 0xe7,
0xbd, 0xd1, 0xe5, 0xd6, 0x4c, 0xbc, 0x9f, 0x2c, 0xd8, 0x2a, 0xb2, 0x4f, 0x17, 0x71, 0x94, 0x4a,
0x3a, 0xe2, 0x83, 0x24, 0xd1, 0x47, 0x7c, 0x90, 0x24, 0xe2, 0x1e, 0x34, 0x51, 0xa6, 0x79, 0x98,
0xe9, 0x2a, 0xb9, 0x69, 0x3c, 0xea, 0x6f, 0xf3, 0x30, 0x43, 0x6d, 0x25, 0x3e, 0x83, 0xed, 0xb5,
0x3a, 0x54, 0x0d, 0xbf, 0x73, 0xff, 0x1d, 0xf3, 0xdd, 0x9a, 0x1e, 0x37, 0xcc, 0xbd, 0x7f, 0x6c,
0xe8, 0x94, 0x3c, 0xaf, 0x8a, 0x8c, 0xf8, 0xd9, 0x2a, 0x8a, 0xec, 0x0e, 0x0f, 0x9b, 0x0b, 0x5a,
0x3d, 0xf5, 0xa4, 0x2e, 0x58, 0x87, 0x45, 0x59, 0x5a, 0x87, 0xa6, 0x11, 0xda, 0x97, 0x35, 0x42,
0x1a, 0x5d, 0x2f, 0xfd, 0xe8, 0x44, 0xce, 0xb8, 0x2c, 0x5b, 0xa8, 0xa1, 0xd8, 0x35, 0x5d, 0x81,
0xcf, 0x71, 0xad, 0xd7, 0x68, 0x0d, 0x9a, 0xce, 0xa1, 0xba, 0xdc, 0x68, 0x48, 0x67, 0xc5, 0xf5,
0xa2, 0x90, 0x78, 0x08, 0x1d, 0xd3, 0xbe, 0xd2, 0xe2, 0x88, 0x7a, 0xc6, 0x95, 0x51, 0x62, 0xd9,
0x50, 0x7c, 0xbe, 0x39, 0x97, 0xdc, 0x36, 0x47, 0xe1, 0xae, 0x65, 0x5e, 0xd2, 0xe3, 0xe6, 0x1c,
0xdb, 0x2b, 0x0d, 0x4a, 0x17, 0xf8, 0xe3, 0x1b, 0xe6, 0xe3, 0x95, 0x0a, 0x4b, 0xe3, 0xf4, 0x41,
0x79, 0x96, 0xb8, 0x1d, 0xfe, 0xa6, 0xb7, 0xce, 0x9c, 0xd2, 0x61, 0x79, 0xe6, 0xec, 0x95, 0x06,
0x99, 0xdb, 0xdd, 0xdc, 0x68, 0xa5, 0x42, 0x63, 0xe5, 0xfd, 0x5a, 0x83, 0xad, 0xd1, 0x7c, 0x11,
0x27, 0x59, 0xe9, 0x16, 0x8e, 0xa2, 0x99, 0x7c, 0xa5, 0x6f, 0x21, 0x83, 0xea, 0x41, 0xc5, 0xdd,
0x90, 0x6e, 0x23, 0xdf, 0x3e, 0x07, 0x15, 0x28, 0x9d, 0x80, 0xb3, 0x76, 0x02, 0xb7, 0xa1, 0xad,
0xca, 0x8d, 0x54, 0x75, 0x56, 0x19, 0x81, 0x7a, 0x68, 0x2c, 0x79, 0xb8, 0x37, 0x79, 0xb8, 0x6b,
0x48, 0x9d, 0x47, 0x99, 0xb1, 0xb2, 0xc5, 0xca, 0x92, 0x84, 0xf4, 0x47, 0xc1, 0x5c, 0xa6, 0x99,
0x3f, 0x5f, 0xd0, 0x55, 0xb6, 0x07, 0x36, 0x96, 0x24, 0x74, 0x8b, 0x39, 0x89, 0x27, 0x89, 0xf4,
0x33, 0x39, 0x7b, 0x9c, 0xf1, 0x09, 0xda, 0xb8, 0x21, 0x25, 0x3b, 0x4e, 0xcb, 0xd8, 0x81, 0xb2,
0x5b, 0x97, 0x7a, 0xbf, 0xd5, 0x40, 0x28, 0xce, 0xb8, 0xf3, 0xfd, 0x7f, 0xc4, 0x5d, 0x4e, 0xd0,
0x3a, 0x0d, 0xcd, 0x73, 0x34, 0xdc, 0x82, 0x06, 0xc7, 0xa3, 0x29, 0x28, 0x10, 0x35, 0x4a, 0xd3,
0xa6, 0x15, 0x7f, 0x16, 0x96, 0x45, 0xc2, 0x83, 0x6e, 0x69, 0x46, 0x50, 0x81, 0x93, 0xef, 0x35,
0x59, 0x05, 0x89, 0x70, 0x45, 0x12, 0x3b, 0x95, 0x24, 0x4e, 0xa0, 0x77, 0x94, 0xf8, 0x51, 0x1a,
0xfa, 0x99, 0xa4, 0xf0, 0xdf, 0x84, 0xc5, 0x8a, 0x57, 0xad, 0xf7, 0x21, 0xdc, 0xdc, 0xf0, 0x6b,
0xda, 0x2b, 0xd1, 0x6a, 0x33, 0xad, 0xb4, 0xf4, 0xc6, 0x70, 0x63, 0x65, 0x3a, 0x1a, 0xbe, 0x51,
0x04, 0xe7, 0x9d, 0x7e, 0x54, 0xca, 0x8b, 0x9d, 0x16, 0xdb, 0x57, 0xc5, 0xba, 0x0f, 0x6e, 0x71,
0xf7, 0xd4, 0x93, 0xba, 0x88, 0x60, 0x12, 0xc8, 0x25, 0xd9, 0x1f, 0xfa, 0x73, 0x59, 0x04, 0xc1,
0x6b, 0x92, 0xf1, 0x78, 0xa9, 0xf1, 0x43, 0x9c, 0xd7, 0xde, 0xdf, 0x16, 0xf4, 0xaa, 0x9c, 0xf0,
0x7b, 0x29, 0x94, 0xbe, 0x1a, 0x28, 0x2d, 0x54, 0x40, 0x3c, 0x82, 0xfa, 0x0f, 0x81, 0x5c, 0xea,
0x81, 0xe2, 0x95, 0xde, 0x7a, 0x17, 0x44, 0x82, 0xea, 0x03, 0x2a, 0xaf, 0xc7, 0xd3, 0x2c, 0x88,
0x23, 0xfd, 0x7a, 0x54, 0x88, 0xf6, 0xd9, 0x0f, 0xe3, 0xe9, 0x77, 0xdc, 0xb7, 0x1d, 0x54, 0xa0,
0xa2, 0x5c, 0xea, 0x57, 0x2c, 0x97, 0x46, 0xf5, 0x9d, 0xb3, 0x34, 0x57, 0xa5, 0x09, 0xff, 0xda,
0x13, 0x53, 0x77, 0x4c, 0x3f, 0xd5, 0xf8, 0x8e, 0xb9, 0xea, 0x99, 0x62, 0x5e, 0x63, 0x1a, 0xd2,
0xd3, 0x88, 0x96, 0x13, 0x3f, 0x54, 0x8d, 0xab, 0x8d, 0x2b, 0xfc, 0x9a, 0x9b, 0x79, 0x3e, 0xd9,
0x46, 0x55, 0xb2, 0xfb, 0xd7, 0x7f, 0x3f, 0xdb, 0xb1, 0xfe, 0x38, 0xdb, 0xb1, 0xfe, 0x3c, 0xdb,
0xb1, 0x7e, 0xfe, 0x6b, 0xe7, 0xad, 0xe3, 0x06, 0xff, 0xc9, 0x7d, 0xfc, 0x5f, 0x00, 0x00, 0x00,
0xff, 0xff, 0xb8, 0x93, 0x5b, 0x24, 0xd9, 0x0d, 0x00, 0x00,
// 1281 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xac, 0x57, 0xcd, 0x6e, 0x1c, 0x45,
0x10, 0x66, 0x76, 0x66, 0xff, 0x6a, 0xd7, 0x4e, 0xe8, 0x38, 0x61, 0x84, 0x82, 0xb3, 0x6a, 0x05,
0xb4, 0x70, 0x70, 0xe4, 0x10, 0xa2, 0x9c, 0x80, 0x38, 0xeb, 0xc0, 0x2a, 0x8a, 0x15, 0x7a, 0x23,
0xe7, 0x86, 0x34, 0xf6, 0x36, 0xce, 0x88, 0xd9, 0x99, 0x65, 0xa6, 0x87, 0x8d, 0x8f, 0x3c, 0x03,
0x17, 0x1e, 0x81, 0xe7, 0xc8, 0x05, 0x8e, 0x3c, 0x02, 0x0a, 0x9c, 0x79, 0x01, 0x2e, 0xa8, 0xaa,
0xa7, 0xb7, 0x7b, 0xd7, 0x13, 0xc7, 0x8a, 0xb8, 0xf5, 0x57, 0x55, 0x53, 0x5d, 0xf5, 0x75, 0x75,
0x55, 0x0f, 0xf4, 0xe7, 0xe5, 0x51, 0x12, 0x1f, 0xef, 0xcc, 0xf3, 0x4c, 0x65, 0xac, 0x13, 0xa7,
0x4a, 0xe6, 0x69, 0x94, 0xf0, 0x02, 0x7c, 0x91, 0x2d, 0x58, 0x08, 0xed, 0x07, 0x59, 0x52, 0xce,
0xd2, 0x22, 0xf4, 0x06, 0xfe, 0x30, 0x10, 0x06, 0x32, 0x06, 0xc1, 0x23, 0x79, 0x5a, 0x84, 0xfe,
0xc0, 0x1f, 0x76, 0x05, 0xad, 0xd9, 0x4d, 0x68, 0xde, 0x57, 0x2a, 0x2f, 0xc2, 0xc6, 0xc0, 0x1f,
0xf6, 0x6e, 0x6f, 0xee, 0x18, 0x77, 0x3b, 0x28, 0x16, 0x5a, 0x89, 0x3e, 0x45, 0x16, 0xe5, 0x71,
0x7a, 0x12, 0x06, 0x03, 0x6f, 0xd8, 0x17, 0x06, 0xf2, 0xc7, 0xd0, 0x9d, 0xc4, 0x27, 0xa9, 0x9c,
0xe2, 0xd6, 0x37, 0xc0, 0x7f, 0x92, 0xe1, 0xb6, 0xde, 0xb0, 0x77, 0x7b, 0xc3, 0xba, 0x12, 0xd9,
0x42, 0xa0, 0x06, 0x0d, 0x0e, 0xe4, 0x49, 0xd8, 0xa8, 0x35, 0x38, 0x90, 0x27, 0xfc, 0x1e, 0x6c,
0x8a, 0x6c, 0x31, 0x9e, 0xca, 0x54, 0xc5, 0xdf, 0xc5, 0x32, 0xa7, 0xa0, 0x45, 0xb6, 0x30, 0xb9,
0xd0, 0x7a, 0x99, 0x48, 0xc3, 0x26, 0xc2, 0x3f, 0x87, 0xe0, 0x49, 0x14, 0xe7, 0x6c, 0x13, 0x1a,
0xe3, 0x11, 0x85, 0x10, 0x88, 0xc6, 0x78, 0xc4, 0x2e, 0x83, 0xff, 0x48, 0x9e, 0x86, 0xfe, 0xc0,
0x1b, 0x76, 0x05, 0x2e, 0xd9, 0x16, 0x34, 0x1f, 0x64, 0x65, 0xaa, 0x28, 0x8c, 0x40, 0x68, 0xc0,
0xf7, 0xa1, 0x8b, 0xdf, 0x3f, 0x8c, 0x65, 0x32, 0x65, 0x5c, 0x3b, 0xab, 0x32, 0x71, 0x48, 0x41,
0xa9, 0xd0, 0x1b, 0x6d, 0x41, 0x93, 0x8c, 0xc9, 0x4d, 0x57, 0x68, 0xc0, 0xbf, 0x06, 0x40, 0x6d,
0xa1, 0xfd, 0xdc, 0x84, 0x26, 0x21, 0x8a, 0xfe, 0xac, 0x23, 0xad, 0x7c, 0x8d, 0xa7, 0x0f, 0xa0,
0x39, 0x4e, 0xd5, 0xdd, 0x3b, 0xa8, 0x3e, 0x8c, 0x92, 0x52, 0x52, 0x34, 0xbe, 0xd0, 0x80, 0x97,
0xd0, 0x21, 0x3b, 0xe4, 0x7d, 0xe9, 0xc0, 0x73, 0x1c, 0xa0, 0x14, 0xb9, 0x1c, 0x99, 0x3c, 0x09,
0xb0, 0x6b, 0xd0, 0x12, 0xd9, 0xc2, 0x52, 0x52, 0x21, 0xf6, 0xa1, 0xd9, 0x25, 0xa0, 0x9c, 0x2f,
0xd9, 0x50, 0x29, 0x0a, 0xb3, 0xed, 0xb7, 0x00, 0x5f, 0xe5, 0x59, 0x39, 0x27, 0xd2, 0xd8, 0x10,
0x9a, 0x84, 0xaa, 0xfc, 0x98, 0xfd, 0xc8, 0xc4, 0x26, 0xb4, 0x41, 0x3d, 0xe9, 0x78, 0x38, 0x93,
0x72, 0x46, 0x91, 0xf8, 0x02, 0x97, 0xfc, 0x27, 0x0f, 0x3a, 0x87, 0x51, 0xb2, 0x54, 0x1f, 0x46,
0x49, 0x95, 0x37, 0x2e, 0x57, 0xdd, 0xf8, 0xc6, 0xcd, 0xfb, 0xd0, 0x79, 0x98, 0x64, 0x91, 0x42,
0x63, 0xf4, 0xe5, 0x89, 0x25, 0x66, 0xbb, 0x00, 0x23, 0x79, 0x1c, 0xcf, 0xa2, 0x04, 0xb5, 0x3a,
0xb9, 0x77, 0x6d, 0x9c, 0x95, 0x4e, 0x38, 0x46, 0xfc, 0x33, 0x68, 0x57, 0xa8, 0x9e, 0x7b, 0x94,
0x4e, 0x8e, 0xa3, 0x44, 0x9a, 0x28, 0x08, 0xf0, 0x67, 0xb0, 0xa1, 0x6f, 0x1a, 0xde, 0x99, 0x89,
0x54, 0x17, 0x28, 0xc5, 0x0b, 0xdd, 0x3e, 0xfe, 0xab, 0x07, 0x01, 0xae, 0x8c, 0x03, 0xcf, 0x3a,
0x60, 0x10, 0x3c, 0x3d, 0x9d, 0xcb, 0x8a, 0x55, 0x5a, 0xb3, 0x01, 0xf4, 0x26, 0x0a, 0x2f, 0xa7,
0x8e, 0x5c, 0x6f, 0xe7, 0x8a, 0x90, 0xaf, 0x71, 0xaa, 0xec, 0x71, 0xfb, 0x62, 0x89, 0xd9, 0x75,
0xe8, 0xee, 0x65, 0x59, 0xa2, 0x95, 0xcd, 0x81, 0x37, 0xec, 0x08, 0x2b, 0x60, 0xdb, 0x00, 0x86,
0xd9, 0x52, 0x86, 0x2d, 0xe2, 0xda, 0x91, 0xf0, 0x5b, 0xd0, 0xc6, 0x48, 0x1f, 0x47, 0x73, 0x9b,
0x9b, 0x77, 0x5e, 0x6e, 0xff, 0x7a, 0xd0, 0xff, 0xa6, 0x94, 0xf9, 0xa9, 0x90, 0x3f, 0x94, 0xb2,
0x50, 0xc8, 0x2d, 0x61, 0x53, 0xcb, 0x04, 0xb0, 0x6a, 0x27, 0xcf, 0xa3, 0x7c, 0xaa, 0x99, 0x0a,
0x44, 0x85, 0x30, 0x57, 0xcb, 0x79, 0x41, 0xb9, 0x76, 0x84, 0x2b, 0xa2, 0x7a, 0x97, 0xb3, 0x4c,
0x99, 0x64, 0x2a, 0xc4, 0x86, 0x70, 0x69, 0xff, 0xc5, 0x71, 0x52, 0x4e, 0xa5, 0xc8, 0x16, 0xfa,
0xeb, 0x16, 0x19, 0xac, 0x8b, 0xd9, 0x47, 0xb0, 0x59, 0x89, 0x4c, 0x5f, 0x6d, 0x93, 0xe1, 0x9a,
0x94, 0xed, 0x42, 0x7f, 0x7f, 0x76, 0x24, 0xa7, 0x53, 0x39, 0x1d, 0x45, 0x2a, 0x0a, 0x3b, 0x94,
0xf7, 0x5a, 0x97, 0x5b, 0x31, 0xe1, 0x3f, 0x7b, 0xb0, 0x51, 0x65, 0x5f, 0xcc, 0xb3, 0xb4, 0x90,
0x78, 0xc4, 0xfb, 0x79, 0x6e, 0x8e, 0x78, 0x3f, 0xcf, 0xd9, 0x2d, 0x68, 0x0b, 0x59, 0x94, 0x89,
0x32, 0x55, 0x72, 0xd5, 0x7a, 0x34, 0xdf, 0x96, 0x89, 0x12, 0xc6, 0x8a, 0x7d, 0x01, 0x9b, 0x2b,
0x75, 0xa8, 0x1b, 0x7e, 0xef, 0xf6, 0x7b, 0xf6, 0xbb, 0x15, 0xbd, 0x58, 0x33, 0xe7, 0xff, 0xf8,
0xd0, 0x73, 0x3c, 0x2f, 0x8b, 0x0c, 0xf9, 0xd9, 0xa8, 0x8a, 0xec, 0x06, 0x0d, 0x9b, 0xd7, 0xb4,
0x7a, 0xec, 0x49, 0x7d, 0xf0, 0x0e, 0xaa, 0xb2, 0xf4, 0x0e, 0x6c, 0x23, 0xf4, 0xcf, 0x6b, 0x84,
0x38, 0xba, 0x9e, 0x47, 0xe9, 0x89, 0x9c, 0x52, 0x59, 0x76, 0x84, 0x81, 0x6c, 0xc7, 0x76, 0x05,
0x3a, 0xc7, 0x95, 0x5e, 0x63, 0x34, 0xc2, 0x76, 0x0e, 0xdd, 0xe5, 0xc6, 0x23, 0x3c, 0x2b, 0xaa,
0x17, 0x8d, 0xd8, 0x5d, 0xe8, 0xd9, 0xf6, 0x55, 0x54, 0x47, 0xb4, 0x65, 0x5d, 0x59, 0xa5, 0x70,
0x0d, 0xd9, 0x97, 0xeb, 0x73, 0x29, 0xec, 0x52, 0x14, 0xe1, 0x4a, 0xe6, 0x8e, 0x5e, 0xac, 0xcf,
0xb1, 0x5d, 0x67, 0x50, 0x86, 0x40, 0x1f, 0x5f, 0xb1, 0x1f, 0x2f, 0x55, 0xc2, 0x19, 0xa7, 0x77,
0xdc, 0x59, 0x12, 0xf6, 0xe8, 0x9b, 0xad, 0x55, 0xe6, 0xb4, 0x4e, 0xb8, 0x33, 0x67, 0xd7, 0x19,
0x64, 0x61, 0x7f, 0x7d, 0xa3, 0xa5, 0x4a, 0x58, 0x2b, 0xfe, 0x5b, 0x03, 0x36, 0xc6, 0xb3, 0x79,
0x96, 0x2b, 0xe7, 0x16, 0x8e, 0xd3, 0xa9, 0x7c, 0x61, 0x6e, 0x21, 0x81, 0xfa, 0x41, 0x45, 0xdd,
0x10, 0x6f, 0x23, 0xdd, 0xbe, 0x40, 0x68, 0xe0, 0x9c, 0x40, 0xb0, 0x72, 0x02, 0xd7, 0xa1, 0xab,
0xcb, 0x0d, 0x55, 0x4d, 0x52, 0x59, 0x81, 0x7e, 0x68, 0x2c, 0x68, 0xb8, 0xb7, 0x69, 0xb8, 0x1b,
0x88, 0x9d, 0x47, 0x9b, 0x91, 0xb2, 0x43, 0x4a, 0x47, 0x82, 0xfa, 0xa7, 0xf1, 0x4c, 0x16, 0x2a,
0x9a, 0xcd, 0xf1, 0x2a, 0xfb, 0x43, 0x5f, 0x38, 0x12, 0xbc, 0xc5, 0x94, 0xc4, 0x83, 0x5c, 0x46,
0x4a, 0x4e, 0xef, 0x2b, 0x3a, 0x41, 0x5f, 0xac, 0x49, 0xd1, 0x8e, 0xd2, 0xb2, 0x76, 0xa0, 0xed,
0x56, 0xa5, 0x34, 0x89, 0x12, 0x19, 0xe5, 0x74, 0x2e, 0x1d, 0xa1, 0x01, 0x7f, 0xd9, 0x00, 0xa6,
0x99, 0xa4, 0x7e, 0xf8, 0xff, 0xd1, 0x79, 0x3e, 0x6d, 0xab, 0xe4, 0xb4, 0xcf, 0x90, 0x73, 0x0d,
0x5a, 0x14, 0x8f, 0x21, 0xa6, 0x42, 0xd8, 0x3e, 0x6d, 0xf3, 0xd6, 0xac, 0x7a, 0xc2, 0x15, 0x31,
0x0e, 0x7d, 0x67, 0x72, 0x60, 0xd9, 0xa3, 0xef, 0x15, 0x59, 0x0d, 0xb5, 0x70, 0x41, 0x6a, 0x7b,
0x75, 0xd4, 0xf2, 0x17, 0xb0, 0xf5, 0x34, 0x8f, 0xd2, 0x22, 0x89, 0x94, 0xc4, 0xf0, 0xdf, 0x86,
0xc5, 0xba, 0xb7, 0xee, 0x00, 0x7a, 0x07, 0x99, 0x7a, 0x96, 0xc7, 0x2a, 0x3a, 0x4a, 0x64, 0xd5,
0x62, 0x5c, 0x11, 0xff, 0x18, 0xae, 0xae, 0xed, 0x6c, 0xdb, 0x32, 0x12, 0xef, 0x13, 0xf1, 0xb8,
0xe4, 0x13, 0xb8, 0xb2, 0x34, 0x1d, 0x8f, 0xde, 0x2a, 0xc6, 0xb3, 0x4e, 0x3f, 0x71, 0x32, 0x27,
0xa7, 0xd5, 0xf6, 0x35, 0xd9, 0xf0, 0x3d, 0x08, 0xab, 0x3b, 0xab, 0x9f, 0xe2, 0x55, 0x04, 0x87,
0xb1, 0x5c, 0xa0, 0xfd, 0x41, 0x34, 0x93, 0x55, 0x10, 0xb4, 0x46, 0x19, 0x8d, 0xa5, 0x06, 0x3d,
0xe0, 0x69, 0xcd, 0xff, 0xf6, 0x60, 0xab, 0xce, 0x89, 0xad, 0x6e, 0xcf, 0xa9, 0x6e, 0x76, 0x0f,
0x9a, 0x3f, 0xc6, 0x72, 0x61, 0x06, 0x11, 0x77, 0xde, 0x88, 0xaf, 0x89, 0x44, 0xe8, 0x0f, 0xb0,
0x00, 0xef, 0x1f, 0xab, 0x38, 0x4b, 0xcd, 0xab, 0x53, 0x23, 0xdc, 0x67, 0x2f, 0xc9, 0x8e, 0xbf,
0xa7, 0xc3, 0x08, 0x84, 0x06, 0x35, 0x05, 0xd5, 0xbc, 0x60, 0x41, 0xb5, 0x6a, 0x0b, 0xea, 0xa5,
0x67, 0xb8, 0x72, 0x5e, 0x06, 0x6f, 0x3c, 0x31, 0x7d, 0x0b, 0xcd, 0x13, 0x8f, 0x6e, 0x61, 0xa8,
0x9f, 0x37, 0xf6, 0x15, 0x67, 0x20, 0x3e, 0xa9, 0x70, 0x79, 0x18, 0x25, 0xba, 0xe1, 0x75, 0xc5,
0x12, 0xbf, 0xe1, 0xee, 0x9e, 0x4d, 0xb6, 0x55, 0x97, 0xec, 0xde, 0xe5, 0xdf, 0x5f, 0x6d, 0x7b,
0x7f, 0xbc, 0xda, 0xf6, 0xfe, 0x7c, 0xb5, 0xed, 0xfd, 0xf2, 0xd7, 0xf6, 0x3b, 0x47, 0x2d, 0xfa,
0x03, 0xfc, 0xf4, 0xbf, 0x00, 0x00, 0x00, 0xff, 0xff, 0xb6, 0xd2, 0x7d, 0x80, 0x11, 0x0e, 0x00,
0x00,
}
func (m *Row) Marshal() (dAtA []byte, err error) {
@ -3044,6 +3062,16 @@ func (m *ImportRequest) MarshalToSizedBuffer(dAtA []byte) (int, error) {
i -= len(m.XXX_unrecognized)
copy(dAtA[i:], m.XXX_unrecognized)
}
if m.Clear {
i--
if m.Clear {
dAtA[i] = 1
} else {
dAtA[i] = 0
}
i--
dAtA[i] = 0x58
}
if m.FieldCreatedAt != 0 {
i = encodeVarintPublic(dAtA, i, uint64(m.FieldCreatedAt))
i--
@ -3294,6 +3322,16 @@ func (m *TranslateKeysRequest) MarshalToSizedBuffer(dAtA []byte) (int, error) {
i -= len(m.XXX_unrecognized)
copy(dAtA[i:], m.XXX_unrecognized)
}
if m.NotWritable {
i--
if m.NotWritable {
dAtA[i] = 1
} else {
dAtA[i] = 0
}
i--
dAtA[i] = 0x20
}
if len(m.Keys) > 0 {
for iNdEx := len(m.Keys) - 1; iNdEx >= 0; iNdEx-- {
i -= len(m.Keys[iNdEx])
@ -4180,6 +4218,9 @@ func (m *ImportRequest) Size() (n int) {
if m.FieldCreatedAt != 0 {
n += 1 + sovPublic(uint64(m.FieldCreatedAt))
}
if m.Clear {
n += 2
}
if m.XXX_unrecognized != nil {
n += len(m.XXX_unrecognized)
}
@ -4264,6 +4305,9 @@ func (m *TranslateKeysRequest) Size() (n int) {
n += 1 + l + sovPublic(uint64(l))
}
}
if m.NotWritable {
n += 2
}
if m.XXX_unrecognized != nil {
n += len(m.XXX_unrecognized)
}
@ -7678,6 +7722,26 @@ func (m *ImportRequest) Unmarshal(dAtA []byte) error {
break
}
}
case 11:
if wireType != 0 {
return fmt.Errorf("proto: wrong wireType = %d for field Clear", wireType)
}
var v int
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowPublic
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
v |= int(b&0x7F) << shift
if b < 0x80 {
break
}
}
m.Clear = bool(v != 0)
default:
iNdEx = preIndex
skippy, err := skipPublic(dAtA[iNdEx:])
@ -8273,6 +8337,26 @@ func (m *TranslateKeysRequest) Unmarshal(dAtA []byte) error {
}
m.Keys = append(m.Keys, string(dAtA[iNdEx:postIndex]))
iNdEx = postIndex
case 4:
if wireType != 0 {
return fmt.Errorf("proto: wrong wireType = %d for field NotWritable", wireType)
}
var v int
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowPublic
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
v |= int(b&0x7F) << shift
if b < 0x80 {
break
}
}
m.NotWritable = bool(v != 0)
default:
iNdEx = preIndex
skippy, err := skipPublic(dAtA[iNdEx:])

2
internal/public.proto
View file

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

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 {

7
pql/pql.peg.go
View file

@ -8,6 +8,7 @@ import (
"os"
"sort"
"strconv"
"strings"
)
const endSymbol rune = 1114112
@ -432,6 +433,12 @@ func (p *PQL) WriteSyntaxTree(w io.Writer) {
p.tokens32.WriteSyntaxTree(w, p.Buffer)
}
func (p *PQL) SprintSyntaxTree() string {
var bldr strings.Builder
p.WriteSyntaxTree(&bldr)
return bldr.String()
}
func (p *PQL) Execute() {
buffer, _buffer, text, begin, end := p.Buffer, p.buffer, "", 0, 0
for _, token := range p.Tokens() {

7
server/config_test.go
View file

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

339
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,27 @@ 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}}})
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++
} else {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: nil})
}
} else {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Uint64ArrayVal{Uint64ArrayVal: &pb.Uint64Array{Vals: ids.Rows}}})
&pb.ColumnResponse{ColumnVal: nil})
}
case "mutex":
@ -395,17 +406,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]}})
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})
}
} else {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: nil})
@ -428,15 +446,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 +468,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,10 +484,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}})
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})
}
} else {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: nil})
@ -484,10 +511,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}}})
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})
}
} else {
rowResp.Columns = append(rowResp.Columns,
&pb.ColumnResponse{ColumnVal: nil})
@ -504,18 +537,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})
@ -527,8 +566,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 +602,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 +622,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 +639,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 +664,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 +679,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 +707,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 +750,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 +772,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 +788,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 +812,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 +835,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 +861,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 {

42
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,13 +289,41 @@ func TestTranslation_Reset(t *testing.T) {
if err != nil {
t.Fatal(err)
}
if _, err := node0.API.TranslateKeys(ctx, bytes.NewReader(reqBody)); err != nil {
t.Fatal(err)
}
})
}
func 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) {
@ -404,7 +432,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 {