From 4b0789ebf089c8e329c78a3a08c06c692af3dfff Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Kuba=20Podg=C3=B3rski?= Date: Thu, 27 Aug 2020 16:15:49 +0200 Subject: [PATCH] Merge pull request #732 from kuba--/translatekey-writable Add writable argument to TranslateKey functions. --- api.go | 12 +- boltdb/translate.go | 39 +++-- boltdb/translate_test.go | 104 ++++++++++-- cluster.go | 14 +- executor.go | 61 +++++-- executor_internal_test.go | 8 +- executor_test.go | 13 +- mock/translator.go | 12 +- pql/ast.go | 13 ++ server/grpc.go | 349 +++++++++++++++++++++++--------------- translate.go | 31 ++-- translator_test.go | 14 +- 12 files changed, 448 insertions(+), 222 deletions(-) diff --git a/api.go b/api.go index 5c70564a2..2695ef9d2 100644 --- a/api.go +++ b/api.go @@ -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) { diff --git a/boltdb/translate.go b/boltdb/translate.go index 8ddcbd555..5dd7f6fae 100644 --- a/boltdb/translate.go +++ b/boltdb/translate.go @@ -32,11 +32,20 @@ var ( // ErrTranslateStoreClosed is returned when reading from an TranslateEntryReader // and the underlying store is closed. ErrTranslateStoreClosed = errors.New("boltdb: translate store closing") + + // ErrTranslateKeyNotFound is returned when translating key + // and the underlying store returns an empty set + ErrTranslateKeyNotFound = errors.New("boltdb: translating key returned empty set") + + bucketKeys = []byte("keys") + bucketIDs = []byte("ids") ) const ( // snapshotExt is the file extension used for an in-process snapshot. snapshotExt = ".snapshotting" + + errFmtTranslateBucketNotFound = "boltdb: translate bucket '%s' not found" ) // OpenTranslateStore opens and initializes a boltdb translation store. @@ -102,9 +111,9 @@ func (s *TranslateStore) Open() (err error) { // Initialize buckets. if err := s.db.Update(func(tx *bolt.Tx) error { - if _, err := tx.CreateBucketIfNotExists([]byte("keys")); err != nil { + if _, err := tx.CreateBucketIfNotExists(bucketKeys); err != nil { return err - } else if _, err := tx.CreateBucketIfNotExists([]byte("ids")); err != nil { + } else if _, err := tx.CreateBucketIfNotExists(bucketIDs); err != nil { return err } return nil @@ -195,12 +204,11 @@ func (s *TranslateStore) TranslateKey(key string, writable bool) (uint64, error) }); err != nil { return 0, err } - - if written { - s.notifyWrite() + if len(ids) == 0 { + // this should not happen + return 0, ErrTranslateKeyNotFound } - - return id, nil + return ids[0], nil } // TranslateKeys converts a slice of string keys to a slice of integer IDs. @@ -241,7 +249,9 @@ func (s *TranslateStore) translateKeys(keys []string, writable bool) ([]uint64, } return nil, nil } - + if !writable { + return nil, pilosa.ErrTranslatingKeyNotFound + } // Find or create ids under write lock if any keys were not found. var written bool if err := s.db.Update(func(tx *bolt.Tx) (err error) { @@ -265,7 +275,6 @@ func (s *TranslateStore) translateKeys(keys []string, writable bool) ([]uint64, }); err != nil { return nil, err } - if written { s.notifyWrite() } @@ -280,7 +289,7 @@ func (s *TranslateStore) TranslateID(id uint64) (string, error) { return "", err } defer func() { _ = tx.Rollback() }() - return findKeyByID(tx.Bucket([]byte("ids")), id), nil + return findKeyByID(tx.Bucket(bucketIDs), id), nil } // TranslateIDs converts a list of integer IDs to a list of string keys. @@ -297,7 +306,7 @@ func (s *TranslateStore) TranslateIDs(ids []uint64) ([]string, error) { keys := make([]string, len(ids)) for i, id := range ids { - keys[i] = findKeyByID(tx.Bucket([]byte("ids")), id) + keys[i] = findKeyByID(tx.Bucket(bucketIDs), id) } return keys, nil } @@ -305,9 +314,9 @@ func (s *TranslateStore) TranslateIDs(ids []uint64) ([]string, error) { // ForceSet writes the id/key pair to the store even if read only. Used by replication. func (s *TranslateStore) ForceSet(id uint64, key string) error { if err := s.db.Update(func(tx *bolt.Tx) (err error) { - if err := tx.Bucket([]byte("keys")).Put([]byte(key), u64tob(id)); err != nil { + if err := tx.Bucket(bucketKeys).Put([]byte(key), u64tob(id)); err != nil { return err - } else if err := tx.Bucket([]byte("ids")).Put(u64tob(id), []byte(key)); err != nil { + } else if err := tx.Bucket(bucketIDs).Put(u64tob(id), []byte(key)); err != nil { return err } return nil @@ -400,7 +409,7 @@ func (s *TranslateStore) ReadFrom(r io.Reader) (n int64, err error) { // MaxID returns the highest id in the store. func maxID(tx *bolt.Tx) uint64 { - if key, _ := tx.Bucket([]byte("ids")).Cursor().Last(); key != nil { + if key, _ := tx.Bucket(bucketIDs).Cursor().Last(); key != nil { return btou64(key) } return 0 @@ -438,7 +447,7 @@ func (r *TranslateEntryReader) ReadEntry(entry *pilosa.TranslateEntry) error { var found bool if err := r.store.db.View(func(tx *bolt.Tx) error { // Find ID/key lookup at offset or later. - cur := tx.Bucket([]byte("ids")).Cursor() + cur := tx.Bucket(bucketIDs).Cursor() key, value := cur.Seek(u64tob(r.offset)) if key == nil { return nil diff --git a/boltdb/translate_test.go b/boltdb/translate_test.go index ef9a726b5..73b4e5345 100644 --- a/boltdb/translate_test.go +++ b/boltdb/translate_test.go @@ -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]) diff --git a/cluster.go b/cluster.go index e4f9aaf18..3905c91d1 100644 --- a/cluster.go +++ b/cluster.go @@ -2325,8 +2325,8 @@ func (c *cluster) setStatic(hosts []string) error { } // translateFieldKey gets a single key from translateFieldKeys. -func (c *cluster) translateFieldKey(ctx context.Context, field *Field, key string) (uint64, error) { - ids, err := c.translateFieldKeys(ctx, field, key) +func (c *cluster) translateFieldKey(ctx context.Context, field *Field, key string, writable bool) (uint64, error) { + ids, err := c.translateFieldKeys(ctx, field, []string{key}, writable) if err != nil { return 0, err } else if len(ids) == 0 { @@ -2359,21 +2359,21 @@ func (c *cluster) translateFieldKeys(ctx context.Context, field *Field, keys []s return ids, nil } -func (c *cluster) translateIndexKey(ctx context.Context, indexName string, key string) (uint64, error) { - keyMap, err := c.translateIndexKeySet(ctx, indexName, map[string]struct{}{key: struct{}{}}) +func (c *cluster) translateIndexKey(ctx context.Context, indexName string, key string, writable bool) (uint64, error) { + keyMap, err := c.translateIndexKeySet(ctx, indexName, map[string]struct{}{key: struct{}{}}, writable) if err != nil { return 0, err } return keyMap[key], nil } -func (c *cluster) translateIndexKeys(ctx context.Context, indexName string, keys []string) ([]uint64, error) { +func (c *cluster) translateIndexKeys(ctx context.Context, indexName string, keys []string, writable bool) ([]uint64, error) { keySet := make(map[string]struct{}) for _, key := range keys { keySet[key] = struct{}{} } - keyMap, err := c.translateIndexKeySet(ctx, indexName, keySet) + keyMap, err := c.translateIndexKeySet(ctx, indexName, keySet, writable) if err != nil { return nil, err } @@ -2394,7 +2394,7 @@ func (c *cluster) translateIndexKeys(ctx context.Context, indexName string, keys return ids, nil } -func (c *cluster) translateIndexKeySet(ctx context.Context, indexName string, keySet map[string]struct{}) (map[string]uint64, error) { +func (c *cluster) translateIndexKeySet(ctx context.Context, indexName string, keySet map[string]struct{}, writable bool) (map[string]uint64, error) { keyMap := make(map[string]uint64) idx := c.holder.Index(indexName) diff --git a/executor.go b/executor.go index a44606f2d..b047016b6 100644 --- a/executor.go +++ b/executor.go @@ -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") } } diff --git a/executor_internal_test.go b/executor_internal_test.go index 909f48f7a..534469864 100644 --- a/executor_internal_test.go +++ b/executor_internal_test.go @@ -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) } diff --git a/executor_test.go b/executor_test.go index 0bbf3bde8..c5a26a6b0 100644 --- a/executor_test.go +++ b/executor_test.go @@ -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) + } + } } } }) diff --git a/mock/translator.go b/mock/translator.go index dc1e5a420..8a28b504c 100644 --- a/mock/translator.go +++ b/mock/translator.go @@ -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) { diff --git a/pql/ast.go b/pql/ast.go index e93a2aaf5..d3858b951 100644 --- a/pql/ast.go +++ b/pql/ast.go @@ -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 { diff --git a/server/grpc.go b/server/grpc.go index 70df01429..6b2b8c3aa 100644 --- a/server/grpc.go +++ b/server/grpc.go @@ -357,6 +357,7 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe } ci = nil // only include headers with the first row + colAdded := 0 for _, field := range fields { // TODO: handle `time` fields switch field.Type() { @@ -371,17 +372,21 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe return errors.Wrapf(err, "querying rows for set: %s", pql) } - ids, ok := resp.Results[0].(pilosa.RowIdentifiers) - if !ok { - return errors.Wrap(err, "getting row identifiers") - } + if len(resp.Results) > 0 { + ids, ok := resp.Results[0].(pilosa.RowIdentifiers) + if !ok { + return errors.Wrap(err, "getting row identifiers") + } - if len(ids.Keys) > 0 { - rowResp.Columns = append(rowResp.Columns, - &pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_StringArrayVal{StringArrayVal: &pb.StringArray{Vals: ids.Keys}}}) - } else { - rowResp.Columns = append(rowResp.Columns, - &pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Uint64ArrayVal{Uint64ArrayVal: &pb.Uint64Array{Vals: ids.Rows}}}) + if len(ids.Keys) > 0 { + rowResp.Columns = append(rowResp.Columns, + &pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_StringArrayVal{StringArrayVal: &pb.StringArray{Vals: ids.Keys}}}) + colAdded++ + } else if len(ids.Rows) > 0 { + rowResp.Columns = append(rowResp.Columns, + &pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Uint64ArrayVal{Uint64ArrayVal: &pb.Uint64Array{Vals: ids.Rows}}}) + colAdded++ + } } case "mutex": @@ -395,20 +400,24 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe return errors.Wrap(err, "querying rows for mutex") } - ids, ok := resp.Results[0].(pilosa.RowIdentifiers) - if !ok { - return errors.Wrap(err, "getting row identifiers") - } + if len(resp.Results) > 0 { + ids, ok := resp.Results[0].(pilosa.RowIdentifiers) + if !ok { + return errors.Wrap(err, "getting row identifiers") + } - if len(ids.Keys) == 1 { - rowResp.Columns = append(rowResp.Columns, - &pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_StringVal{StringVal: ids.Keys[0]}}) - } else if len(ids.Rows) == 1 { - rowResp.Columns = append(rowResp.Columns, - &pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Uint64Val{Uint64Val: ids.Rows[0]}}) - } else { - rowResp.Columns = append(rowResp.Columns, - &pb.ColumnResponse{ColumnVal: nil}) + if len(ids.Keys) == 1 { + rowResp.Columns = append(rowResp.Columns, + &pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_StringVal{StringVal: ids.Keys[0]}}) + colAdded++ + } else if len(ids.Rows) == 1 { + rowResp.Columns = append(rowResp.Columns, + &pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Uint64Val{Uint64Val: ids.Rows[0]}}) + colAdded++ + } else { + rowResp.Columns = append(rowResp.Columns, + &pb.ColumnResponse{ColumnVal: nil}) + } } case "int": @@ -428,15 +437,17 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe return errors.Wrap(err, "getting int field value for column") } - valCount, ok := resp.Results[0].(pilosa.ValCount) - if ok && valCount.Count == 1 { - vals, err := h.api.TranslateIndexIDs(stream.Context(), fi, []uint64{uint64(valCount.Val)}) - if err != nil { - return errors.Wrap(err, "getting keys for ids") - } - if len(vals) > 0 && vals[0] != "" { - value = vals[0] - exists = true + if len(resp.Results) > 0 { + valCount, ok := resp.Results[0].(pilosa.ValCount) + if ok && valCount.Count == 1 { + vals, err := h.api.TranslateIndexIDs(stream.Context(), fi, []uint64{uint64(valCount.Val)}) + if err != nil { + return errors.Wrap(err, "getting keys for ids") + } + if len(vals) > 0 && vals[0] != "" { + value = vals[0] + exists = true + } } } } else { @@ -448,6 +459,7 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe if exists { rowResp.Columns = append(rowResp.Columns, &pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_StringVal{StringVal: value}}) + colAdded++ } else { rowResp.Columns = append(rowResp.Columns, &pb.ColumnResponse{ColumnVal: nil}) @@ -463,13 +475,16 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe return errors.Wrap(err, "getting int field value for column") } - valCount, ok := resp.Results[0].(pilosa.ValCount) - if ok && valCount.Count == 1 { - rowResp.Columns = append(rowResp.Columns, - &pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Int64Val{Int64Val: valCount.Val}}) - } else { - rowResp.Columns = append(rowResp.Columns, - &pb.ColumnResponse{ColumnVal: nil}) + if len(resp.Results) > 0 { + valCount, ok := resp.Results[0].(pilosa.ValCount) + if ok && valCount.Count == 1 { + rowResp.Columns = append(rowResp.Columns, + &pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Int64Val{Int64Val: valCount.Val}}) + colAdded++ + } else { + rowResp.Columns = append(rowResp.Columns, + &pb.ColumnResponse{ColumnVal: nil}) + } } } @@ -484,13 +499,16 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe return errors.Wrap(err, "getting decimal field value for column") } - valCount, ok := resp.Results[0].(pilosa.ValCount) - if ok && valCount.Count == 1 { - rowResp.Columns = append(rowResp.Columns, - &pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_DecimalVal{DecimalVal: &pb.Decimal{Value: valCount.DecimalVal.Value, Scale: valCount.DecimalVal.Scale}}}) - } else { - rowResp.Columns = append(rowResp.Columns, - &pb.ColumnResponse{ColumnVal: nil}) + if len(resp.Results) > 0 { + valCount, ok := resp.Results[0].(pilosa.ValCount) + if ok && valCount.Count == 1 { + rowResp.Columns = append(rowResp.Columns, + &pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_DecimalVal{DecimalVal: &pb.Decimal{Value: valCount.DecimalVal.Value, Scale: valCount.DecimalVal.Scale}}}) + colAdded++ + } else { + rowResp.Columns = append(rowResp.Columns, + &pb.ColumnResponse{ColumnVal: nil}) + } } case "bool": @@ -504,21 +522,24 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe return errors.Wrap(err, "querying rows for bool") } - ids, ok := resp.Results[0].(pilosa.RowIdentifiers) - if !ok { - return errors.Wrap(err, "getting row identifiers") - } - - if len(ids.Rows) == 1 { - var bval bool - if ids.Rows[0] == 1 { - bval = true + if len(resp.Results) > 0 { + ids, ok := resp.Results[0].(pilosa.RowIdentifiers) + if !ok { + return errors.Wrap(err, "getting row identifiers") + } + + if len(ids.Rows) == 1 { + var bval bool + if ids.Rows[0] == 1 { + bval = true + } + rowResp.Columns = append(rowResp.Columns, + &pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_BoolVal{BoolVal: bval}}) + colAdded++ + } else { + rowResp.Columns = append(rowResp.Columns, + &pb.ColumnResponse{ColumnVal: nil}) } - rowResp.Columns = append(rowResp.Columns, - &pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_BoolVal{BoolVal: bval}}) - } else { - rowResp.Columns = append(rowResp.Columns, - &pb.ColumnResponse{ColumnVal: nil}) } case "time": @@ -527,8 +548,25 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe } } - if err := stream.Send(rowResp); err != nil { - return errors.Wrap(err, "sending response to stream") + // For SQL queries like: + // SELECT * FROM t WHERE _id=garbageID; + // we don't want to return any rows. + // So, check here if we added any columns. + // + // Because we don't have keys to translate + // and _id is an artificial field that's why for query: + // SELECT _id FROM t WHERE _id=existing-id; + // we return an empty result. + // + // TODO(kuba--): We need to find a way to check here if + // existing-id is not a garbage. + // + // A query which will work here is 'SELECT *' or any query with more columns + // than just _id. + if colAdded > 0 { + if err := stream.Send(rowResp); err != nil { + return errors.Wrap(err, "sending response to stream") + } } } @@ -546,6 +584,7 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe return errToStatusError(errors.New("invalid key columns")) } + forceSend := false ci := []*pb.ColumnInfo{ {Name: "_id", Datatype: "string"}, } @@ -565,6 +604,11 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe end = uint64(len(cols)) } cols = cols[offset:end] + if len(cols) == 1 { + if id, err := h.api.TranslateIndexKey(stream.Context(), index.Name(), cols[0], false); id != 0 && err == nil { + forceSend = true + } + } } else { // Prevent getting too many records by forcing a limit. pql := fmt.Sprintf("All(limit=%d, offset=%d)", limit, offset) @@ -577,18 +621,20 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe return errors.Wrapf(err, "querying for all: %s", pql) } - ids, ok := resp.Results[0].(*pilosa.Row) - if !ok { - return errors.Wrap(err, "getting results as a row") - } + if len(resp.Results) > 0 { + ids, ok := resp.Results[0].(*pilosa.Row) + if !ok { + return errors.Wrap(err, "getting results as a row") + } - limitedCols := ids.Keys - if len(limitedCols) == 0 { - // If cols is still empty after the limit/offset, then - // return with no results. - return nil + limitedCols := ids.Keys + if len(limitedCols) == 0 { + // If cols is still empty after the limit/offset, then + // return with no results. + return nil + } + cols = limitedCols } - cols = limitedCols } for _, col := range cols { @@ -600,6 +646,7 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe } ci = nil // only include headers with the first row + colAdded := 0 for _, field := range fields { // TODO: handle `time` fields switch field.Type() { @@ -614,17 +661,21 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe return errors.Wrap(err, "querying set rows(keys)") } - ids, ok := resp.Results[0].(pilosa.RowIdentifiers) - if !ok { - return errors.Wrap(err, "getting row identifiers") - } + if len(resp.Results) > 0 { + ids, ok := resp.Results[0].(pilosa.RowIdentifiers) + if !ok { + return errors.Wrap(err, "getting row identifiers") + } - if len(ids.Keys) > 0 { - rowResp.Columns = append(rowResp.Columns, - &pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_StringArrayVal{StringArrayVal: &pb.StringArray{Vals: ids.Keys}}}) - } else { - rowResp.Columns = append(rowResp.Columns, - &pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Uint64ArrayVal{Uint64ArrayVal: &pb.Uint64Array{Vals: ids.Rows}}}) + if len(ids.Keys) > 0 { + rowResp.Columns = append(rowResp.Columns, + &pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_StringArrayVal{StringArrayVal: &pb.StringArray{Vals: ids.Keys}}}) + colAdded++ + } else if len(ids.Rows) > 0 { + rowResp.Columns = append(rowResp.Columns, + &pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Uint64ArrayVal{Uint64ArrayVal: &pb.Uint64Array{Vals: ids.Rows}}}) + colAdded++ + } } case "mutex": @@ -638,25 +689,29 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe return errors.Wrap(err, "querying mutex rows(keys)") } - ids, ok := resp.Results[0].(pilosa.RowIdentifiers) - if !ok { - return errors.Wrap(err, "getting row identifiers") - } + if len(resp.Results) > 0 { + ids, ok := resp.Results[0].(pilosa.RowIdentifiers) + if !ok { + return errors.Wrap(err, "getting row identifiers") + } - if len(ids.Keys) == 1 { - rowResp.Columns = append(rowResp.Columns, - &pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_StringVal{StringVal: ids.Keys[0]}}) - } else if len(ids.Rows) == 1 { - rowResp.Columns = append(rowResp.Columns, - &pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Uint64Val{Uint64Val: ids.Rows[0]}}) - } else { - rowResp.Columns = append(rowResp.Columns, - &pb.ColumnResponse{ColumnVal: nil}) + if len(ids.Keys) == 1 { + rowResp.Columns = append(rowResp.Columns, + &pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_StringVal{StringVal: ids.Keys[0]}}) + colAdded++ + } else if len(ids.Rows) == 1 { + rowResp.Columns = append(rowResp.Columns, + &pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Uint64Val{Uint64Val: ids.Rows[0]}}) + colAdded++ + } else { + rowResp.Columns = append(rowResp.Columns, + &pb.ColumnResponse{ColumnVal: nil}) + } } case "int": // Translate column key. - id, err := h.api.TranslateIndexKey(stream.Context(), index.Name(), col) + id, err := h.api.TranslateIndexKey(stream.Context(), index.Name(), col, false) if err != nil { return errors.Wrap(err, "translating column key") } @@ -677,15 +732,17 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe return errors.Wrap(err, "getting int field value for column") } - valCount, ok := resp.Results[0].(pilosa.ValCount) - if ok && valCount.Count == 1 { - vals, err := h.api.TranslateIndexIDs(stream.Context(), fi, []uint64{uint64(valCount.Val)}) - if err != nil { - return errors.Wrap(err, "getting keys for ids") - } - if len(vals) > 0 && vals[0] != "" { - value = vals[0] - exists = true + if len(resp.Results) > 0 { + valCount, ok := resp.Results[0].(pilosa.ValCount) + if ok && valCount.Count == 1 { + vals, err := h.api.TranslateIndexIDs(stream.Context(), fi, []uint64{uint64(valCount.Val)}) + if err != nil { + return errors.Wrap(err, "getting keys for ids") + } + if len(vals) > 0 && vals[0] != "" { + value = vals[0] + exists = true + } } } } else { @@ -697,6 +754,7 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe if exists { rowResp.Columns = append(rowResp.Columns, &pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_StringVal{StringVal: value}}) + colAdded++ } else { rowResp.Columns = append(rowResp.Columns, &pb.ColumnResponse{ColumnVal: nil}) @@ -712,13 +770,16 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe return errors.Wrap(err, "getting int field value for column") } - valCount, ok := resp.Results[0].(pilosa.ValCount) - if ok && valCount.Count == 1 { - rowResp.Columns = append(rowResp.Columns, - &pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Int64Val{Int64Val: valCount.Val}}) - } else { - rowResp.Columns = append(rowResp.Columns, - &pb.ColumnResponse{ColumnVal: nil}) + if len(resp.Results) > 0 { + valCount, ok := resp.Results[0].(pilosa.ValCount) + if ok && valCount.Count == 1 { + rowResp.Columns = append(rowResp.Columns, + &pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Int64Val{Int64Val: valCount.Val}}) + colAdded++ + } else { + rowResp.Columns = append(rowResp.Columns, + &pb.ColumnResponse{ColumnVal: nil}) + } } } @@ -733,13 +794,16 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe return errors.Wrap(err, "getting decimal field value for column") } - valCount, ok := resp.Results[0].(pilosa.ValCount) - if ok && valCount.Count == 1 { - rowResp.Columns = append(rowResp.Columns, - &pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_DecimalVal{DecimalVal: &pb.Decimal{Value: valCount.DecimalVal.Value, Scale: valCount.DecimalVal.Scale}}}) - } else { - rowResp.Columns = append(rowResp.Columns, - &pb.ColumnResponse{ColumnVal: nil}) + if len(resp.Results) > 0 { + valCount, ok := resp.Results[0].(pilosa.ValCount) + if ok && valCount.Count == 1 { + rowResp.Columns = append(rowResp.Columns, + &pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_DecimalVal{DecimalVal: &pb.Decimal{Value: valCount.DecimalVal.Value, Scale: valCount.DecimalVal.Scale}}}) + colAdded++ + } else { + rowResp.Columns = append(rowResp.Columns, + &pb.ColumnResponse{ColumnVal: nil}) + } } case "bool": @@ -753,21 +817,24 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe return errors.Wrap(err, "querying bool rows(keys)") } - ids, ok := resp.Results[0].(pilosa.RowIdentifiers) - if !ok { - return errors.Wrap(err, "getting row identifiers") - } - - if len(ids.Rows) == 1 { - var bval bool - if ids.Rows[0] == 1 { - bval = true + if len(resp.Results) > 0 { + ids, ok := resp.Results[0].(pilosa.RowIdentifiers) + if !ok { + return errors.Wrap(err, "getting row identifiers") + } + + if len(ids.Rows) == 1 { + var bval bool + if ids.Rows[0] == 1 { + bval = true + } + rowResp.Columns = append(rowResp.Columns, + &pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_BoolVal{BoolVal: bval}}) + colAdded++ + } else { + rowResp.Columns = append(rowResp.Columns, + &pb.ColumnResponse{ColumnVal: nil}) } - rowResp.Columns = append(rowResp.Columns, - &pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_BoolVal{BoolVal: bval}}) - } else { - rowResp.Columns = append(rowResp.Columns, - &pb.ColumnResponse{ColumnVal: nil}) } case "time": @@ -776,8 +843,20 @@ func (h *GRPCHandler) Inspect(req *pb.InspectRequest, stream pb.Pilosa_InspectSe } } - if err := stream.Send(rowResp); err != nil { - return errors.Wrap(err, "sending response to stream") + // For SQL queries like: + // SELECT _id FROM parent WHERE _id="garbage"; + // we get here without any real columns and fields, and we did not + // translate any keys. That's why we don't want to send anything back + // and return fake response like: + // + // _id + // ------- + // + // (1 row) + if colAdded > 0 || forceSend { + if err := stream.Send(rowResp); err != nil { + return errors.Wrap(err, "sending response to stream") + } } } diff --git a/translate.go b/translate.go index 045d9c03b..4efa26129 100644 --- a/translate.go +++ b/translate.go @@ -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 { diff --git a/translator_test.go b/translator_test.go index 86c78bfc6..8d71cc729 100644 --- a/translator_test.go +++ b/translator_test.go @@ -36,21 +36,21 @@ func TestInMemTranslateStore_TranslateKey(t *testing.T) { s := pilosa.NewInMemTranslateStore("IDX", "FLD", 0, pilosa.DefaultPartitionN) // Ensure initial key translates to ID 1. - if id, err := s.TranslateKey("foo"); err != nil { + if id, err := s.TranslateKey("foo", true); err != nil { t.Fatal(err) } else if got, want := id, uint64(1); got != want { t.Fatalf("TranslateKey()=%d, want %d", got, want) } // Ensure next key autoincrements. - if id, err := s.TranslateKey("bar"); err != nil { + if id, err := s.TranslateKey("bar", true); err != nil { t.Fatal(err) } else if got, want := id, uint64(2); got != want { t.Fatalf("TranslateKey()=%d, want %d", got, want) } // Ensure retranslating existing key returns original ID. - if id, err := s.TranslateKey("foo"); err != nil { + if id, err := s.TranslateKey("foo", true); err != nil { t.Fatal(err) } else if got, want := id, uint64(1); got != want { t.Fatalf("TranslateKey()=%d, want %d", got, want) @@ -61,9 +61,9 @@ func TestInMemTranslateStore_TranslateID(t *testing.T) { s := pilosa.NewInMemTranslateStore("IDX", "FLD", 0, pilosa.DefaultPartitionN) // Setup initial keys. - if _, err := s.TranslateKey("foo"); err != nil { + if _, err := s.TranslateKey("foo", true); err != nil { t.Fatal(err) - } else if _, err := s.TranslateKey("bar"); err != nil { + } else if _, err := s.TranslateKey("bar", true); err != nil { t.Fatal(err) } @@ -289,7 +289,7 @@ func TestTranslation_Reset(t *testing.T) { if err != nil { t.Fatal(err) } - if _, err := node0.API.TranslateKeys(ctx, bytes.NewReader(reqBody)); err != nil { + if _, err := node0.API.TranslateKeys(ctx, bytes.NewReader(reqBody), true); err != nil { t.Fatal(err) } }) @@ -558,7 +558,7 @@ func TestTranslation_Coordinator(t *testing.T) { fld := "f" // Create an index without keys. - if _, err := node0.API.CreateIndex(ctx, idx, + if _, err := node1.API.CreateIndex(ctx, idx, pilosa.IndexOptions{ Keys: false, }); err != nil {