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 #1027 from niaow/translate-maybe-v2.1

Browse source 
Allow querying without creating keys (v2.1.x backport)
This commit is contained in:
Nia 2020-10-23 17:06:24 -04:00 • committed by GitHub
commit 59cb045285
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
16 changed files with 2070 additions and 532 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

32
api.go
View file

@ -1661,6 +1661,38 @@ func (api *API) TranslateIDs(ctx context.Context, r io.Reader) (_ []byte, err er
return buf, nil
}
// FindIndexKeys looks up column keys in the index, mapping them to IDs.
// If a key does not exist, it will be absent from the resulting map.
func (api *API) FindIndexKeys(ctx context.Context, index string, keys ...string) (map[string]uint64, error) {
return api.cluster.findIndexKeys(ctx, index, keys...)
}
// FindFieldKeys looks up keys in a field, mapping them to IDs.
// If a key does not exist, it will be absent from the resulting map.
func (api *API) FindFieldKeys(ctx context.Context, index, field string, keys ...string) (map[string]uint64, error) {
f := api.holder.Field(index, field)
if f == nil {
return nil, errors.Wrapf(ErrFieldNotFound, "finding keys for field %q", field)
}
return api.cluster.findFieldKeys(ctx, f, keys...)
}
// CreateIndexKeys looks up column keys in the index, mapping them to IDs.
// If a key does not exist, it will be created.
func (api *API) CreateIndexKeys(ctx context.Context, index string, keys ...string) (map[string]uint64, error) {
return api.cluster.createIndexKeys(ctx, index, keys...)
}
// CreateFieldKeys looks up keys in a field, mapping them to IDs.
// If a key does not exist, it will be created.
func (api *API) CreateFieldKeys(ctx context.Context, index, field string, keys ...string) (map[string]uint64, error) {
f := api.holder.Field(index, field)
if f == nil {
return nil, errors.Wrapf(ErrFieldNotFound, "finding keys for field %q", field)
}
return api.cluster.createFieldKeys(ctx, f, keys...)
}
// PrimaryReplicaNodeURL returns the URL of the cluster's primary replica.
func (api *API) PrimaryReplicaNodeURL() url.URL {
node := api.cluster.PrimaryReplicaNode()

73
boltdb/translate.go
View file

@ -190,6 +190,79 @@ func (s *TranslateStore) TranslateKeys(keys []string, writable bool) ([]uint64,
return s.translateKeys(keys, writable)
}
// FindKeys looks up the ID for each key.
// Keys are not created if they do not exist.
// Missing keys are not considered errors, so the length of the result may be less than that of the input.
func (s *TranslateStore) FindKeys(keys ...string) (map[string]uint64, error) {
result := make(map[string]uint64, len(keys))
err := s.db.View(func(tx *bolt.Tx) error {
bkt := tx.Bucket(bucketKeys)
if bkt == nil {
return errors.Errorf(errFmtTranslateBucketNotFound, bucketKeys)
}
for _, key := range keys {
id, _ := findIDByKey(bkt, key)
if id == 0 {
// The key does not exist.
continue
}
result[key] = id
}
return nil
})
if err != nil {
return nil, err
}
return result, nil
}
// CreateKeys maps all keys to IDs, creating the IDs if they do not exist.
// If the translator is read-only, this will return an error.
func (s *TranslateStore) CreateKeys(keys ...string) (map[string]uint64, error) {
if s.ReadOnly() {
return nil, pilosa.ErrTranslateStoreReadOnly
}
written := false
result := make(map[string]uint64, len(keys))
err := s.db.Update(func(tx *bolt.Tx) error {
keyBucket := tx.Bucket(bucketKeys)
if keyBucket == nil {
return errors.Errorf(errFmtTranslateBucketNotFound, bucketKeys)
}
idBucket := tx.Bucket(bucketIDs)
if keyBucket == nil {
return errors.Errorf(errFmtTranslateBucketNotFound, bucketIDs)
}
for _, key := range keys {
id, boltKey := findIDByKey(keyBucket, key)
if id == 0 {
// The key does not exist.
id = pilosa.GenerateNextPartitionedID(s.index, maxID(tx), s.partitionID, s.partitionN)
if err := keyBucket.Put(boltKey, u64tob(id)); err != nil {
return err
} else if err := idBucket.Put(u64tob(id), boltKey); err != nil {
return err
}
written = true
}
result[key] = id
}
return nil
})
if err != nil {
return nil, err
}
if written {
s.notifyWrite()
}
return result, nil
}
func (s *TranslateStore) translateKeys(keys []string, writable bool) ([]uint64, error) {
ids := make([]uint64, 0, len(keys))

118
boltdb/translate_test.go
View file

@ -92,6 +92,46 @@ func TestTranslateStore_TranslateKeys(t *testing.T) {
}
}
func TestTranslateStore_CreateKeys(t *testing.T) {
s := MustOpenNewTranslateStore()
defer MustCloseTranslateStore(s)
ids, err := s.CreateKeys("abc", "abc")
if err != nil {
t.Fatal(err)
} else if _, ok := ids["abc"]; !ok {
t.Fatalf(`missing "abc"; got %v`, ids)
} else if len(ids) > 1 {
t.Fatalf("expected one key, got %d in %v", len(ids), ids)
}
// Ensure different keys translate to different IDs.
ids1, err := s.CreateKeys("foo", "bar")
if err != nil {
t.Fatal(err)
} else if foo, bar := ids1["foo"], ids1["bar"]; foo == bar {
t.Fatalf(`"foo" and "bar" map back to the same ID %d`, foo)
}
// Ensure retranslation returns original IDs.
if ids, err := s.CreateKeys("bar", "foo"); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(ids, ids1) {
t.Fatalf("retranslation produced result %v which is different from original translation %v", ids, ids1)
}
// Ensure retranslating with existing and non-existing keys returns correctly.
if ids, err := s.CreateKeys("foo", "baz", "bar"); err != nil {
t.Fatal(err)
} else if got, want := ids["foo"], ids1["foo"]; got != want {
t.Fatalf(`mismatched ID %d for "foo" (previously %d)`, got, want)
} else if _, ok := ids["baz"]; !ok {
t.Fatalf(`missing translation for "baz"; got %v`, ids)
} else if got, want := ids["bar"], ids1["bar"]; got != want {
t.Fatalf(`mismatched ID %d for "bar" (previously %d)`, got, want)
}
}
func TestTranslateStore_ReadKey(t *testing.T) {
s := MustOpenNewTranslateStore()
defer MustCloseTranslateStore(s)
@ -216,6 +256,84 @@ func TestTranslateStore_TranslateIDs(t *testing.T) {
}
}
func TestTranslateStore_FindKeys(t *testing.T) {
cases := []struct {
name string
data []string
lookup []string
}{
{
name: "All",
data: []string{"plugh", "xyzzy", "h"},
lookup: []string{"plugh", "xyzzy", "h"},
},
{
name: "Extra",
data: []string{"plugh", "xyzzy", "h"},
lookup: []string{"plugh", "xyzzy", "h", "65"},
},
{
name: "None",
data: []string{"a", "b", "c"},
lookup: []string{"d", "e"},
},
{
name: "Empty",
lookup: []string{"h"},
},
{
name: "LookupNothing",
},
}
for _, c := range cases {
c := c
t.Run(c.name, func(t *testing.T) {
s := MustOpenNewTranslateStore()
defer MustCloseTranslateStore(s)
var naiveMap map[string]uint64
if c.data != nil {
// Load in key data.
keys := c.data
ids, err := s.TranslateKeys(keys, true)
if err != nil {
t.Errorf("failed to import keys: %v", err)
return
}
if len(ids) != len(keys) {
t.Errorf("mapped %d keys to %d ids", len(keys), len(ids))
return
}
naiveMap = make(map[string]uint64, len(keys))
for i, key := range keys {
naiveMap[key] = ids[i]
}
}
// Compute expected lookup result.
result := map[string]uint64{}
for _, key := range c.lookup {
id, ok := naiveMap[key]
if !ok {
// The key is expected to be missing.
continue
}
result[key] = id
}
// Find the keys.
found, err := s.FindKeys(c.lookup...)
if err != nil {
t.Errorf("failed to find keys: %v", err)
} else if !reflect.DeepEqual(result, found) {
t.Errorf("expected %v but found %v", result, found)
}
})
}
}
func TestTranslateStore_EntryReader(t *testing.T) {
t.Run("OK", func(t *testing.T) {
s := MustOpenNewTranslateStore()

39
client.go
View file

@ -92,11 +92,17 @@ type InternalQueryClient interface {
// 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)
FindIndexKeysNode(ctx context.Context, uri *URI, index string, keys ...string) (map[string]uint64, error)
FindFieldKeysNode(ctx context.Context, uri *URI, index string, field string, keys ...string) (map[string]uint64, error)
CreateIndexKeysNode(ctx context.Context, uri *URI, index string, keys ...string) (map[string]uint64, error)
CreateFieldKeysNode(ctx context.Context, uri *URI, index string, field string, keys ...string) (map[string]uint64, error)
}
type nopInternalQueryClient struct{}
func (n *nopInternalQueryClient) QueryNode(ctx context.Context, uri *URI, index string, queryRequest *QueryRequest) (*QueryResponse, error) {
func (n nopInternalQueryClient) QueryNode(ctx context.Context, uri *URI, index string, queryRequest *QueryRequest) (*QueryResponse, error) {
return nil, nil
}
@ -108,15 +114,31 @@ func (n nopInternalQueryClient) TranslateIDsNode(ctx context.Context, uri *URI,
return nil, nil
}
func newNopInternalQueryClient() *nopInternalQueryClient {
return &nopInternalQueryClient{}
func (n nopInternalQueryClient) FindIndexKeysNode(ctx context.Context, uri *URI, index string, keys ...string) (map[string]uint64, error) {
return nil, nil
}
func (n nopInternalQueryClient) FindFieldKeysNode(ctx context.Context, uri *URI, index string, field string, keys ...string) (map[string]uint64, error) {
return nil, nil
}
func (n nopInternalQueryClient) CreateIndexKeysNode(ctx context.Context, uri *URI, index string, keys ...string) (map[string]uint64, error) {
return nil, nil
}
func (n nopInternalQueryClient) CreateFieldKeysNode(ctx context.Context, uri *URI, index string, field string, keys ...string) (map[string]uint64, error) {
return nil, nil
}
func newNopInternalQueryClient() nopInternalQueryClient {
return nopInternalQueryClient{}
}
var _ InternalQueryClient = newNopInternalQueryClient()
//===============
type nopInternalClient struct{}
type nopInternalClient struct{ nopInternalQueryClient }
func newNopInternalClient() nopInternalClient {
return nopInternalClient{}
@ -144,15 +166,6 @@ func (n nopInternalClient) Nodes(ctx context.Context) ([]*Node, error) {
func (n nopInternalClient) Query(ctx context.Context, index string, queryRequest *QueryRequest) (*QueryResponse, error) {
return nil, nil
}
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, writable bool) ([]uint64, error) {
return nil, nil
}
func (n nopInternalClient) TranslateIDsNode(ctx context.Context, uri *URI, index, field string, ids []uint64) ([]string, error) {
return nil, nil
}
func (n nopInternalClient) Import(ctx context.Context, index, field string, shard uint64, bits []Bit, opts ...ImportOption) error {
return nil
}

340
cluster.go
View file

@ -2324,17 +2324,6 @@ func (c *cluster) setStatic(hosts []string) error {
return nil
}
// translateFieldKey gets a single key from translateFieldKeys.
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, nil
}
return ids[0], nil
}
// translateFieldKeys is basically a wrapper around
// field.TranslateStore().TranslateKey(key), but in
// the case where the local node is not coordinator, then this method will forward the translation
@ -2359,6 +2348,133 @@ func (c *cluster) translateFieldKeys(ctx context.Context, field *Field, keys []s
return ids, nil
}
func (c *cluster) findFieldKeys(ctx context.Context, field *Field, keys ...string) (map[string]uint64, error) {
if idx := field.ForeignIndex(); idx != "" {
// The field uses foreign index keys.
// Therefore, the field keys are actually column keys on a different index.
return c.findIndexKeys(ctx, idx, keys...)
}
if !field.Keys() {
return nil, errors.Wrap(ErrTranslatingKeyNotFound, "field is not keyed")
}
// Attempt to find the keys locally.
localTranslations, err := field.TranslateStore().FindKeys(keys...)
if err != nil {
return nil, errors.Wrapf(err, "translating field(%s/%s) keys(%v) locally", field.Index(), field.Name(), keys)
}
// Check for missing keys.
var missing []string
if len(keys) > len(localTranslations) {
// There are either duplicate keys or missing keys.
// This should work either way.
missing = make([]string, 0, len(keys)-len(localTranslations))
for _, k := range keys {
_, found := localTranslations[k]
if !found {
missing = append(missing, k)
}
}
} else if len(localTranslations) > len(keys) {
panic(fmt.Sprintf("more translations than keys! translation count=%v, key count=%v", len(localTranslations), len(keys)))
}
if len(missing) == 0 {
// All keys were available locally.
return localTranslations, nil
}
// It is possible that the missing keys exist, but have not been synced to the local replica.
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)
}
if c.Node.ID == coordinator.ID {
// The local copy is the authoritative copy.
return localTranslations, nil
}
// Forward the missing keys to the coordinator.
// The coordinator has the authoritative copy.
remoteTranslations, err := c.InternalClient.FindFieldKeysNode(ctx, &coordinator.URI, field.Index(), field.Name(), missing...)
if err != nil {
return nil, errors.Wrapf(err, "translating field(%s/%s) keys(%v) remotely", field.Index(), field.Name(), keys)
}
// Merge the remote translations into the local translations.
translations := localTranslations
for key, id := range remoteTranslations {
translations[key] = id
}
return translations, nil
}
func (c *cluster) createFieldKeys(ctx context.Context, field *Field, keys ...string) (map[string]uint64, error) {
if idx := field.ForeignIndex(); idx != "" {
// The field uses foreign index keys.
// Therefore, the field keys are actually column keys on a different index.
return c.createIndexKeys(ctx, idx, keys...)
}
if !field.Keys() {
return nil, errors.Wrap(ErrTranslatingKeyNotFound, "field is not keyed")
}
// The coordinator is the only node that can create field keys, since it owns the authoritative copy.
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)
}
if c.Node.ID == coordinator.ID {
// The local copy is the authoritative copy.
return field.TranslateStore().CreateKeys(keys...)
}
// Attempt to find the keys locally.
// They cannot be created locally, but skipping keys that exist can reduce network usage.
localTranslations, err := field.TranslateStore().FindKeys(keys...)
if err != nil {
return nil, errors.Wrapf(err, "translating field(%s/%s) keys(%v) locally", field.Index(), field.Name(), keys)
}
// Check for missing keys.
var missing []string
if len(keys) > len(localTranslations) {
// There are either duplicate keys or missing keys.
// This should work either way.
missing = make([]string, 0, len(keys)-len(localTranslations))
for _, k := range keys {
_, found := localTranslations[k]
if !found {
missing = append(missing, k)
}
}
} else if len(localTranslations) > len(keys) {
panic(fmt.Sprintf("more translations than keys! translation count=%v, key count=%v", len(localTranslations), len(keys)))
}
if len(missing) == 0 {
// All keys exist locally.
// There is no need to create anything.
return localTranslations, nil
}
// Forward the missing keys to the coordinator to be created.
remoteTranslations, err := c.InternalClient.CreateFieldKeysNode(ctx, &coordinator.URI, field.Index(), field.Name(), missing...)
if err != nil {
return nil, errors.Wrapf(err, "translating field(%s/%s) keys(%v) remotely", field.Index(), field.Name(), keys)
}
// Merge the remote translations into the local translations.
translations := localTranslations
for key, id := range remoteTranslations {
translations[key] = id
}
return translations, nil
}
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 {
@ -2450,6 +2566,208 @@ func (c *cluster) translateIndexKeySet(ctx context.Context, indexName string, ke
return keyMap, nil
}
func (c *cluster) findIndexKeys(ctx context.Context, indexName string, keys ...string) (map[string]uint64, error) {
done := ctx.Done()
idx := c.holder.Index(indexName)
if idx == nil {
return nil, ErrIndexNotFound
}
// Split keys by partition.
keysByPartition := make(map[int][]string, c.partitionN)
for _, key := range keys {
partitionID := c.keyPartition(indexName, key)
keysByPartition[partitionID] = append(keysByPartition[partitionID], key)
}
// TODO: use local replicas to short-circuit network traffic
// Group keys by node.
keysByNode := make(map[*Node][]string)
for partitionID, keys := range keysByPartition {
// Find the primary node for this partition.
primary := c.primaryPartitionNode(partitionID)
if primary == nil {
return nil, errors.Errorf("translating index(%s) keys(%v) on partition(%d) - cannot find primary node", indexName, keys, partitionID)
}
if c.Node.ID == primary.ID {
// The partition is local.
continue
}
// Group the partition to be processed remotely.
keysByNode[primary] = append(keysByNode[primary], keys...)
// Delete remote keys from the by-partition map so that it can be used for local translation.
delete(keysByPartition, partitionID)
}
// Start translating keys remotely.
// On child calls, there are no remote results since we were only sent the keys that we own.
remoteResults := make(chan map[string]uint64, len(keysByNode))
var g errgroup.Group
defer g.Wait() //nolint:errcheck
for node, keys := range keysByNode {
node, keys := node, keys
g.Go(func() error {
translations, err := c.InternalClient.FindIndexKeysNode(ctx, &node.URI, indexName, keys...)
if err != nil {
return errors.Wrapf(err, "translating index(%s) keys(%v) on node %s", indexName, keys, node.ID)
}
remoteResults <- translations
return nil
})
}
// Translate local keys.
translations := make(map[string]uint64)
for partitionID, keys := range keysByPartition {
// Handle cancellation.
select {
case <-done:
return nil, ctx.Err()
default:
}
// Find the keys within the partition.
t, err := idx.TranslateStore(partitionID).FindKeys(keys...)
if err != nil {
return nil, errors.Wrapf(err, "translating index(%s) keys(%v) on partition(%d)", idx.Name(), keys, partitionID)
}
// Merge the translations from this partition.
for key, id := range t {
translations[key] = id
}
}
// Wait for remote key sets.
if err := g.Wait(); err != nil {
return nil, err
}
// Merge the translations.
// All data should have been written to here while we waited.
// Closing the channel prevents the range from blocking.
close(remoteResults)
for t := range remoteResults {
for key, id := range t {
translations[key] = id
}
}
return translations, nil
}
func (c *cluster) createIndexKeys(ctx context.Context, indexName string, keys ...string) (map[string]uint64, error) {
// Check for early cancellation.
done := ctx.Done()
select {
case <-done:
return nil, ctx.Err()
default:
}
idx := c.holder.Index(indexName)
if idx == nil {
return nil, ErrIndexNotFound
}
// Split keys by partition.
keysByPartition := make(map[int][]string, c.partitionN)
for _, key := range keys {
partitionID := c.keyPartition(indexName, key)
keysByPartition[partitionID] = append(keysByPartition[partitionID], key)
}
// TODO: use local replicas to short-circuit network traffic
// Group keys by node.
// Delete remote keys from the by-partition map so that it can be used for local translation.
keysByNode := make(map[*Node][]string)
for partitionID, keys := range keysByPartition {
// Find the primary node for this partition.
primary := c.primaryPartitionNode(partitionID)
if primary == nil {
return nil, errors.Errorf("translating index(%s) keys(%v) on partition(%d) - cannot find primary node", indexName, keys, partitionID)
}
if c.Node.ID == primary.ID {
// The partition is local.
continue
}
// Group the partition to be processed remotely.
keysByNode[primary] = append(keysByNode[primary], keys...)
delete(keysByPartition, partitionID)
}
translateResults := make(chan map[string]uint64, len(keysByNode)+len(keysByPartition))
var g errgroup.Group
defer g.Wait() //nolint:errcheck
// Start translating keys remotely.
// On child calls, there are no remote results since we were only sent the keys that we own.
for node, keys := range keysByNode {
node, keys := node, keys
g.Go(func() error {
translations, err := c.InternalClient.CreateIndexKeysNode(ctx, &node.URI, indexName, keys...)
if err != nil {
return errors.Wrapf(err, "translating index(%s) keys(%v) on node %s", indexName, keys, node.ID)
}
translateResults <- translations
return nil
})
}
// Translate local keys.
// TODO: make this less horrible (why fsync why?????)
// This is kinda terrible because each goroutine does an fsync, thus locking up an entire OS thread.
// AHHHHHHHHHHHHHHHHHH
for partitionID, keys := range keysByPartition {
partitionID, keys := partitionID, keys
g.Go(func() error {
// Handle cancellation.
select {
case <-done:
return ctx.Err()
default:
}
translations, err := idx.TranslateStore(partitionID).CreateKeys(keys...)
if err != nil {
return errors.Wrapf(err, "translating index(%s) keys(%v) on partition(%d)", idx.Name(), keys, partitionID)
}
translateResults <- translations
return nil
})
}
// Wait for remote key sets.
if err := g.Wait(); err != nil {
return nil, err
}
// Merge the translations.
// All data should have been written to here while we waited.
// Closing the channel prevents the range from blocking.
translations := make(map[string]uint64, len(keys))
close(translateResults)
for t := range translateResults {
for key, id := range t {
translations[key] = id
}
}
return translations, nil
}
func (c *cluster) translateIndexIDs(ctx context.Context, indexName string, ids []uint64) ([]string, error) {
idSet := make(map[uint64]struct{})
for _, id := range ids {

1063
executor.go
View file

File diff suppressed because it is too large Load diff

125
executor_internal_test.go
View file

@ -20,121 +20,11 @@ import (
"fmt"
"io/ioutil"
"strconv"
"strings"
"testing"
"github.com/pilosa/pilosa/v2/pql"
)
func TestExecutor_TranslateGroupByCall(t *testing.T) {
holder := NewHolder(DefaultPartitionN)
cluster := NewTestCluster(1)
e := &executor{
Holder: holder,
Cluster: cluster,
}
e.Holder.Path, _ = ioutil.TempDir(*TempDir, "")
err := e.Holder.Open()
if err != nil {
t.Fatalf("opening holder: %v", err)
}
idx, err := e.Holder.CreateIndex("i", IndexOptions{})
if err != nil {
t.Fatalf("creating index: %v", err)
}
_, erra := idx.CreateField("ak", OptFieldKeys())
_, errb := idx.CreateField("b")
_, errc := idx.CreateField("ck", OptFieldKeys())
if erra != nil || errb != nil || errc != nil {
t.Fatalf("creating fields %v, %v, %v", erra, errb, errc)
}
query, err := pql.ParseString(`GroupBy(Rows(ak), Rows(b), Rows(ck), previous=["la", 0, "ha"], having=Condition(count > 10))`)
if err != nil {
t.Fatalf("parsing query: %v", err)
}
c := query.Calls[0]
// 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)
}
if len(c.Args["previous"].([]interface{})) != 3 {
t.Fatalf("unexpected length for 'previous' arg %v", c.Args["previous"])
}
for i, v := range c.Args["previous"].([]interface{}) {
if !isInt(v) {
t.Fatalf("expected all items in previous to be ints, but '%v' at index %d is %[1]T", v, i)
}
}
if having, hok := c.Args["having"].(*pql.Call); !hok {
t.Fatal("expected having to be a call")
} else if cond, cok := having.Args["count"].(*pql.Condition); !cok {
t.Fatal("expected condition to be a count")
} else if cond.Op != pql.GT {
t.Fatal("expected condition op to be >")
} else {
val, ok := cond.Uint64Value()
if !ok || val != uint64(10) {
t.Fatal("expected condition val to be uint64(10)")
}
}
errTests := []struct {
pql string
err string
}{
{
pql: `GroupBy(Rows(notfound), previous=1)`,
err: "'previous' argument must be list",
},
{
pql: `GroupBy(Rows(ak), previous=["la", 0])`,
err: "mismatched lengths",
},
{
pql: `GroupBy(Rows(ak), previous=[1])`,
err: "prev value must be a string",
},
{
pql: `GroupBy(Rows(notfound), previous=[1])`,
err: ErrFieldNotFound.Error(),
},
// TODO: an unknown key will actually allocate an id. this is probably bad.
// {
// pql: `GroupBy(Rows(ak), previous=["zoop"])`,
// err: "translating row key '",
// },
{
pql: `GroupBy(Rows(b), previous=["la"])`,
err: "which doesn't use string keys",
},
}
for i, test := range errTests {
t.Run(fmt.Sprintf("#%d_%s", i, test.err), func(t *testing.T) {
query, err := pql.ParseString(test.pql)
if err != nil {
t.Fatalf("parsing query: %v", err)
}
c := query.Calls[0]
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)
}
if !strings.Contains(err.Error(), test.err) {
t.Fatalf("expected '%s', got '%v'", test.err, err)
}
})
}
}
func TestExecutor_TranslateRowsOnBool(t *testing.T) {
holder := NewHolder(DefaultPartitionN)
defer holder.Close()
@ -183,7 +73,11 @@ func TestExecutor_TranslateRowsOnBool(t *testing.T) {
}
c := query.Calls[0]
err = e.translateCall(context.Background(), "i", c, make(map[string]map[string]uint64), true)
colTranslations, rowTranslations, err := e.preTranslate(context.Background(), "i", c)
if err != nil {
t.Fatalf("pre-translating call: %v", err)
}
_, err = e.translateCall(c, "i", colTranslations, rowTranslations)
if err != nil {
t.Fatalf("translating call: %v", err)
}
@ -191,15 +85,6 @@ func TestExecutor_TranslateRowsOnBool(t *testing.T) {
}
}
func isInt(a interface{}) bool {
switch a.(type) {
case int, int64, uint, uint64:
return true
default:
return false
}
}
func TestFilterWithLimit(t *testing.T) {
f := filterWithLimit(5)

121
executor_test.go
View file

@ -543,13 +543,13 @@ func TestExecutor_Execute_Set(t *testing.T) {
})
t.Run("ErrInvalidColValueType", func(t *testing.T) {
if _, err := cmd.API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `Set("foo", f=1)`}); err == nil || errors.Cause(err).Error() != `string 'col' value not allowed unless index 'keys' option enabled` {
if _, err := cmd.API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `Set("foo", f=1)`}); err == nil || !hasCause(err, pilosa.ErrTranslatingKeyNotFound) || !strings.Contains(err.Error(), "unkeyed index") {
t.Fatalf("The error is: '%v'", err)
}
})
t.Run("ErrInvalidRowValueType", func(t *testing.T) {
if _, err := cmd.API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `Set(2, f="bar")`}); err == nil || errors.Cause(err).Error() != `string 'row' value not allowed unless field 'keys' option enabled` {
if _, err := cmd.API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `Set(2, f="bar")`}); err == nil || !hasCause(err, pilosa.ErrTranslatingKeyNotFound) || !strings.Contains(err.Error(), "field is not keyed") {
t.Fatal(err)
}
})
@ -639,7 +639,7 @@ func TestExecutor_Execute_Set(t *testing.T) {
if _, err := index.CreateField("f", pilosa.OptFieldTypeDefault(), pilosa.OptFieldKeys()); err != nil {
t.Fatal(err)
}
if _, err := cmd.API.Query(context.Background(), &pilosa.QueryRequest{Index: "inokey", Query: `Set(2, f=1.2)`}); err == nil || !strings.Contains(err.Error(), "row value must be a string or non-negative integer") {
if _, err := cmd.API.Query(context.Background(), &pilosa.QueryRequest{Index: "inokey", Query: `Set(2, f=1.2)`}); err == nil || !strings.Contains(err.Error(), "invalid value") {
t.Fatal(err)
}
@ -921,7 +921,7 @@ func TestExecutor_Execute_SetValue(t *testing.T) {
}
})
t.Run("", func(t *testing.T) {
t.Run("Err", func(t *testing.T) {
c := test.MustRunCluster(t, 1)
defer c.Close()
hldr := test.Holder{Holder: c[0].Server.Holder()}
@ -931,26 +931,39 @@ func TestExecutor_Execute_SetValue(t *testing.T) {
t.Fatal(err)
}
t.Run("ErrColumnBSIGroupRequired", func(t *testing.T) {
if _, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `Set(invalid_column_name=10, f=100)`}); err == nil || errors.Cause(err).Error() != `Set() column argument 'col' required` {
t.Run("ColumnBSIGroupRequired", func(t *testing.T) {
if _, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `Set(f=100)`}); err == nil || errors.Cause(err).Error() != `Set() column argument 'col' required` {
t.Fatalf("unexpected error: %s", err)
}
})
t.Run("ErrColumnBSIGroupValue", func(t *testing.T) {
if _, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `Set("bad_column", f=100)`}); err == nil || errors.Cause(err).Error() != `string 'col' value not allowed unless index 'keys' option enabled` {
t.Run("ColumnBSIGroupValue", func(t *testing.T) {
if _, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `Set("bad_column", f=100)`}); err == nil || !hasCause(err, pilosa.ErrTranslatingKeyNotFound) || !strings.Contains(err.Error(), "unkeyed index") {
t.Fatalf("unexpected error: %s", err)
}
})
t.Run("ErrInvalidBSIGroupValueType", func(t *testing.T) {
if _, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `Set(10, f="hello")`}); err == nil || errors.Cause(err).Error() != `string 'row' value not allowed unless field 'keys' option enabled` {
t.Run("InvalidBSIGroupValueType", func(t *testing.T) {
if _, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `Set(10, f="hello")`}); err == nil || !hasCause(err, pilosa.ErrTranslatingKeyNotFound) || !strings.Contains(err.Error(), "field is not keyed") {
t.Fatalf("unexpected error: %s", err)
}
})
})
}
func hasCause(err, cause error) bool {
for err != cause {
innerErr := errors.Cause(err)
if innerErr == err {
// This is the innermost accessible error, and it does not have that cause.
return false
}
err = innerErr
}
return true
}
// Ensure a SetRowAttrs() query can be executed.
func TestExecutor_Execute_SetRowAttrs(t *testing.T) {
c := test.MustRunCluster(t, 1)
@ -5907,3 +5920,91 @@ func TestTimelessClearRegression(t *testing.T) {
t.Fatal("clear supposedly failed")
}
}
func TestMissingKeyRegression(t *testing.T) {
c := test.MustRunCluster(t, 1)
defer c.Close()
c.CreateField(t, "i", pilosa.IndexOptions{Keys: true, TrackExistence: true}, "f", pilosa.OptFieldKeys())
tests := []struct {
name string
query string
expected []interface{}
}{
{
name: "RowGarbage",
query: `Row(f="garbage")`,
expected: []interface{}{[]string(nil)},
},
{
name: "Set",
query: `Set("a", f="example")`,
expected: []interface{}{true},
},
{
name: "Count",
query: `Count(Row(f="example"))`,
expected: []interface{}{uint64(1)},
},
{
name: "NotGarbage",
query: `Not(Row(f="garbage"))`,
expected: []interface{}{[]string{"a"}},
},
{
name: "DifferenceGarbage",
query: `Difference(All(), Row(f="garbage"))`,
expected: []interface{}{[]string{"a"}},
},
/*{
// Key translation works here, but it seems the actual count query is processing stale data.
// Uncomment it when the bug has been fixed.
name: "SetAndCount",
query: `Set("a", f="example")` + "\n" +
`Count(Row(f="example"))`,
expected: []interface{}{true, uint64(1)},
},*/
{
name: "CountNothing",
query: `Count(Row(f="garbage"))`,
expected: []interface{}{uint64(0)},
},
{
name: "StoreInvertSelf",
query: `Store(Not(Row(f="xyzzy")), f="xyzzy")`,
expected: []interface{}{true},
},
{
name: "SetClear",
query: `Set("b", f="plugh")` + "\n" +
`Clear("b", f="plugh")`,
expected: []interface{}{true, true},
},
{
name: "ClearMix",
query: `Clear("a", f="garbage")` + "\n" +
`Clear("a", f="example")`,
expected: []interface{}{false, true},
},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
resp := c.Query(t, "i", tc.query)
if len(resp.Results) != len(tc.expected) {
t.Errorf("expected %d results but got %d", len(resp.Results), len(tc.expected))
return
}
for i, r := range resp.Results {
if row, ok := r.(*pilosa.Row); ok {
r = row.Keys
}
expect := tc.expected[i]
if !reflect.DeepEqual(r, expect) {
t.Errorf("result %d differs: expected %v but got %v", i, expect, r)
}
}
})
}
}

202
http/client.go
View file

@ -1247,6 +1247,208 @@ func (c *InternalClient) TranslateIDsNode(ctx context.Context, uri *pilosa.URI,
return tkresp.Keys, nil
}
func (c *InternalClient) FindIndexKeysNode(ctx context.Context, uri *pilosa.URI, index string, keys ...string) (transMap map[string]uint64, err error) {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.FindIndexKeysNode")
defer span.Finish()
// Create HTTP request.
u := uriPathToURL(uri, fmt.Sprintf("/internal/translate/index/%s/keys/find", index))
reqData, err := json.Marshal(keys)
if err != nil {
return nil, errors.Wrap(err, "marshalling request")
}
req, err := http.NewRequest("POST", u.String(), bytes.NewReader(reqData))
if err != nil {
return nil, errors.Wrap(err, "creating request")
}
// Apply headers.
req.Header.Set("Content-Length", strconv.Itoa(len(reqData)))
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Accept", "application/json")
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
// Send the request.
resp, err := c.executeRequest(req.WithContext(ctx))
if err != nil {
return nil, errors.Wrap(err, "executing request")
}
defer func() {
cerr := resp.Body.Close()
if cerr != nil && err == nil {
err = errors.Wrap(cerr, "closing response body")
}
}()
// Read the response body.
result, err := ioutil.ReadAll(resp.Body)
if err != nil {
return nil, errors.Wrap(err, "reading response")
}
// Decode the translations.
transMap = make(map[string]uint64, len(keys))
err = json.Unmarshal(result, &transMap)
if err != nil {
return nil, errors.Wrap(err, "json decoding")
}
return transMap, nil
}
func (c *InternalClient) FindFieldKeysNode(ctx context.Context, uri *pilosa.URI, index string, field string, keys ...string) (transMap map[string]uint64, err error) {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.FindFieldKeysNode")
defer span.Finish()
// Create HTTP request.
u := uriPathToURL(uri, fmt.Sprintf("/internal/translate/field/%s/%s/keys/find", index, field))
q := u.Query()
q.Add("remote", "true")
u.RawQuery = q.Encode()
reqData, err := json.Marshal(keys)
if err != nil {
return nil, errors.Wrap(err, "marshalling request")
}
req, err := http.NewRequest("POST", u.String(), bytes.NewReader(reqData))
if err != nil {
return nil, errors.Wrap(err, "creating request")
}
// Apply headers.
req.Header.Set("Content-Length", strconv.Itoa(len(reqData)))
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Accept", "application/json")
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
// Send the request.
resp, err := c.executeRequest(req.WithContext(ctx))
if err != nil {
return nil, errors.Wrap(err, "executing request")
}
defer func() {
cerr := resp.Body.Close()
if cerr != nil && err == nil {
err = errors.Wrap(cerr, "closing response body")
}
}()
// Read the response body.
result, err := ioutil.ReadAll(resp.Body)
if err != nil {
return nil, errors.Wrap(err, "reading response")
}
// Decode the translations.
transMap = make(map[string]uint64, len(keys))
err = json.Unmarshal(result, &transMap)
if err != nil {
return nil, errors.Wrap(err, "json decoding")
}
return transMap, nil
}
func (c *InternalClient) CreateIndexKeysNode(ctx context.Context, uri *pilosa.URI, index string, keys ...string) (transMap map[string]uint64, err error) {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.CreateIndexKeysNode")
defer span.Finish()
// Create HTTP request.
u := uriPathToURL(uri, fmt.Sprintf("/internal/translate/index/%s/keys/create", index))
reqData, err := json.Marshal(keys)
if err != nil {
return nil, errors.Wrap(err, "marshalling request")
}
req, err := http.NewRequest("POST", u.String(), bytes.NewReader(reqData))
if err != nil {
return nil, errors.Wrap(err, "creating request")
}
// Apply headers.
req.Header.Set("Content-Length", strconv.Itoa(len(reqData)))
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Accept", "application/json")
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
// Send the request.
resp, err := c.executeRequest(req.WithContext(ctx))
if err != nil {
return nil, errors.Wrap(err, "executing request")
}
defer func() {
cerr := resp.Body.Close()
if cerr != nil && err == nil {
err = errors.Wrap(cerr, "closing response body")
}
}()
// Read the response body.
result, err := ioutil.ReadAll(resp.Body)
if err != nil {
return nil, errors.Wrap(err, "reading response")
}
// Decode the translations.
transMap = make(map[string]uint64, len(keys))
err = json.Unmarshal(result, &transMap)
if err != nil {
return nil, errors.Wrap(err, "json decoding")
}
return transMap, nil
}
func (c *InternalClient) CreateFieldKeysNode(ctx context.Context, uri *pilosa.URI, index string, field string, keys ...string) (transMap map[string]uint64, err error) {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.CreateFieldKeysNode")
defer span.Finish()
// Create HTTP request.
u := uriPathToURL(uri, fmt.Sprintf("/internal/translate/field/%s/%s/keys/create", index, field))
q := u.Query()
q.Add("remote", "true")
u.RawQuery = q.Encode()
reqData, err := json.Marshal(keys)
if err != nil {
return nil, errors.Wrap(err, "marshalling request")
}
req, err := http.NewRequest("POST", u.String(), bytes.NewReader(reqData))
if err != nil {
return nil, errors.Wrap(err, "creating request")
}
// Apply headers.
req.Header.Set("Content-Length", strconv.Itoa(len(reqData)))
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Accept", "application/json")
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
// Send the request.
resp, err := c.executeRequest(req.WithContext(ctx))
if err != nil {
return nil, errors.Wrap(err, "executing request")
}
defer func() {
cerr := resp.Body.Close()
if cerr != nil && err == nil {
err = errors.Wrap(cerr, "closing response body")
}
}()
// Read the response body.
result, err := ioutil.ReadAll(resp.Body)
if err != nil {
return nil, errors.Wrap(err, "reading response")
}
// Decode the translations.
transMap = make(map[string]uint64, len(keys))
err = json.Unmarshal(result, &transMap)
if err != nil {
return nil, errors.Wrap(err, "json decoding")
}
return transMap, nil
}
func (c *InternalClient) Transactions(ctx context.Context) (map[string]*pilosa.Transaction, error) {
span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.Transactions")
defer span.Finish()

185
http/handler.go
View file

@ -385,6 +385,11 @@ func newRouter(handler *Handler) *mux.Router {
router.HandleFunc("/internal/nodes", handler.handleGetNodes).Methods("GET").Name("GetNodes")
router.HandleFunc("/internal/shards/max", handler.handleGetShardsMax).Methods("GET").Name("GetShardsMax") // TODO: deprecate, but it's being used by the client
router.HandleFunc("/internal/translate/index/{index}/keys/find", handler.handleFindIndexKeys).Methods("POST").Name("FindIndexKeys")
router.HandleFunc("/internal/translate/index/{index}/keys/create", handler.handleCreateIndexKeys).Methods("POST").Name("CreateIndexKeys")
router.HandleFunc("/internal/translate/field/{index}/{field}/keys/find", handler.handleFindFieldKeys).Methods("POST").Name("FindFieldKeys")
router.HandleFunc("/internal/translate/field/{index}/{field}/keys/create", handler.handleCreateFieldKeys).Methods("POST").Name("CreateFieldKeys")
router.Use(handler.queryArgValidator)
router.Use(handler.addQueryContext)
router.Use(handler.extractTracing)
@ -2208,3 +2213,183 @@ func readBody(r *http.Request) ([]byte, error) {
return buf.Bytes(), nil
}
func (h *Handler) handleFindIndexKeys(w http.ResponseWriter, r *http.Request) {
// Verify input and output types
if r.Header.Get("Content-Type") != "application/json" {
http.Error(w, "Unsupported media type", http.StatusUnsupportedMediaType)
return
} else if !validHeaderAcceptJSON(r.Header) {
http.Error(w, "Not acceptable", http.StatusNotAcceptable)
return
}
indexName, ok := mux.Vars(r)["index"]
if !ok {
http.Error(w, "index name is required", http.StatusBadRequest)
return
}
bd, err := readBody(r)
if err != nil {
http.Error(w, "failed to read body", http.StatusBadRequest)
return
}
var keys []string
err = json.Unmarshal(bd, &keys)
if err != nil {
http.Error(w, "failed to decode request", http.StatusBadRequest)
return
}
translations, err := h.api.FindIndexKeys(r.Context(), indexName, keys...)
if err != nil {
http.Error(w, "translating keys", http.StatusBadRequest)
return
}
err = json.NewEncoder(w).Encode(translations)
if err != nil {
http.Error(w, "encoding result", http.StatusBadRequest)
return
}
}
func (h *Handler) handleFindFieldKeys(w http.ResponseWriter, r *http.Request) {
// Verify input and output types
if r.Header.Get("Content-Type") != "application/json" {
http.Error(w, "Unsupported media type", http.StatusUnsupportedMediaType)
return
} else if !validHeaderAcceptJSON(r.Header) {
http.Error(w, "Not acceptable", http.StatusNotAcceptable)
return
}
indexName, ok := mux.Vars(r)["index"]
if !ok {
http.Error(w, "index name is required", http.StatusBadRequest)
return
}
fieldName, ok := mux.Vars(r)["field"]
if !ok {
http.Error(w, "field name is required", http.StatusBadRequest)
return
}
bd, err := readBody(r)
if err != nil {
http.Error(w, "failed to read body", http.StatusBadRequest)
return
}
var keys []string
err = json.Unmarshal(bd, &keys)
if err != nil {
http.Error(w, "failed to decode request", http.StatusBadRequest)
return
}
translations, err := h.api.FindFieldKeys(r.Context(), indexName, fieldName, keys...)
if err != nil {
http.Error(w, "translating keys", http.StatusBadRequest)
return
}
err = json.NewEncoder(w).Encode(translations)
if err != nil {
http.Error(w, "encoding result", http.StatusBadRequest)
return
}
}
func (h *Handler) handleCreateIndexKeys(w http.ResponseWriter, r *http.Request) {
// Verify input and output types
if r.Header.Get("Content-Type") != "application/json" {
http.Error(w, "Unsupported media type", http.StatusUnsupportedMediaType)
return
} else if !validHeaderAcceptJSON(r.Header) {
http.Error(w, "Not acceptable", http.StatusNotAcceptable)
return
}
indexName, ok := mux.Vars(r)["index"]
if !ok {
http.Error(w, "index name is required", http.StatusBadRequest)
return
}
bd, err := readBody(r)
if err != nil {
http.Error(w, "failed to read body", http.StatusBadRequest)
return
}
var keys []string
err = json.Unmarshal(bd, &keys)
if err != nil {
http.Error(w, "failed to decode request", http.StatusBadRequest)
return
}
translations, err := h.api.CreateIndexKeys(r.Context(), indexName, keys...)
if err != nil {
http.Error(w, "translating keys", http.StatusBadRequest)
return
}
err = json.NewEncoder(w).Encode(translations)
if err != nil {
http.Error(w, "encoding result", http.StatusBadRequest)
return
}
}
func (h *Handler) handleCreateFieldKeys(w http.ResponseWriter, r *http.Request) {
// Verify input and output types
if r.Header.Get("Content-Type") != "application/json" {
http.Error(w, "Unsupported media type", http.StatusUnsupportedMediaType)
return
} else if !validHeaderAcceptJSON(r.Header) {
http.Error(w, "Not acceptable", http.StatusNotAcceptable)
return
}
indexName, ok := mux.Vars(r)["index"]
if !ok {
http.Error(w, "index name is required", http.StatusBadRequest)
return
}
fieldName, ok := mux.Vars(r)["field"]
if !ok {
http.Error(w, "field name is required", http.StatusBadRequest)
return
}
bd, err := readBody(r)
if err != nil {
http.Error(w, "failed to read body", http.StatusBadRequest)
return
}
var keys []string
err = json.Unmarshal(bd, &keys)
if err != nil {
http.Error(w, "failed to decode request", http.StatusBadRequest)
return
}
translations, err := h.api.CreateFieldKeys(r.Context(), indexName, fieldName, keys...)
if err != nil {
http.Error(w, "translating keys", http.StatusBadRequest)
return
}
err = json.NewEncoder(w).Encode(translations)
if err != nil {
http.Error(w, "encoding result", http.StatusBadRequest)
return
}
}

12
mock/translator.go
View file

@ -21,8 +21,6 @@ import (
"github.com/pilosa/pilosa/v2"
)
var _ pilosa.TranslateStore = (*TranslateStore)(nil)
type TranslateStore struct {
CloseFunc func() error
MaxIDFunc func() (uint64, error)
@ -33,6 +31,8 @@ type TranslateStore struct {
TranslateKeysFunc func(keys []string, writable bool) ([]uint64, error)
TranslateIDFunc func(id uint64) (string, error)
TranslateIDsFunc func(ids []uint64) ([]string, error)
FindKeysFunc func(keys ...string) (map[string]uint64, error)
CreateKeysFunc func(keys ...string) (map[string]uint64, error)
ForceSetFunc func(id uint64, key string) error
EntryReaderFunc func(ctx context.Context, offset uint64) (pilosa.TranslateEntryReader, error)
}
@ -73,6 +73,14 @@ func (s *TranslateStore) TranslateIDs(ids []uint64) ([]string, error) {
return s.TranslateIDsFunc(ids)
}
func (s *TranslateStore) FindKeys(keys ...string) (map[string]uint64, error) {
return s.FindKeysFunc(keys...)
}
func (s *TranslateStore) CreateKeys(keys ...string) (map[string]uint64, error) {
return s.CreateKeysFunc(keys...)
}
func (s *TranslateStore) ForceSet(id uint64, key string) error {
return s.ForceSetFunc(id, key)
}

13
pql/ast.go
View file

@ -662,6 +662,19 @@ func (c *Call) UintSliceArg(key string) ([]uint64, bool, error) {
}
}
func (c *Call) StringArg(key string) (string, bool, error) {
val, ok := c.Args[key]
if !ok {
return "", false, nil
}
switch tval := val.(type) {
case string:
return tval, true, nil
default:
return "", true, fmt.Errorf("unexpected type %T in StringArg, val %v", tval, tval)
}
}
// CallArg is for reading the value at key from call.Args as a Call. If the
// key is not in Call.Args, the value of the returned value will be nil, and
// the error will be nil. An error is returned if the value is not a Call.

4
server/handler_test.go
View file

@ -603,7 +603,7 @@ func TestHandler_Endpoints(t *testing.T) {
h.ServeHTTP(w, test.MustNewHTTPRequest("POST", "/index/i0/query", strings.NewReader(`Row(row=30)`)))
if w.Code != gohttp.StatusBadRequest {
t.Fatalf("unexpected status code: %d", w.Code)
} else if body := w.Body.String(); body != `{"error":"executing: map reduce: field not found"}`+"\n" {
} else if body := w.Body.String(); body != `{"error":"executing: translating call: validating value for field \"row\": field not found"}`+"\n" {
t.Fatalf("unexpected body: %q", body)
}
})
@ -620,7 +620,7 @@ func TestHandler_Endpoints(t *testing.T) {
var resp pilosa.QueryResponse
if err := cmd.API.Serializer.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatal(err)
} else if s := resp.Err.Error(); s != `executing: map reduce: field not found` {
} else if s := resp.Err.Error(); s != `executing: translating call: validating value for field "row": field not found` {
t.Fatalf("unexpected error: %s", s)
}
})

2
stats/stats_test.go
View file

@ -197,7 +197,7 @@ func TestStatsCount_SetColumnAttrs(t *testing.T) {
called := false
idx := hldr.Holder.Index("d")
if idx == nil {
t.Fatal("idex not found")
t.Fatal("index not found")
}
hldr.Holder.Stats = &MockStats{

62
translate.go
View file

@ -50,7 +50,7 @@ var (
// 0xc2, 0xa0, // NO-BREAK SPACE
// 0x00,
// }
type TranslateStore interface {
type TranslateStore interface { // TODO: refactor this interface; readonly should be part of the type and replication should be an impl detail
io.Closer
// Returns the maximum ID set on the store.
@ -71,6 +71,15 @@ type TranslateStore interface {
TranslateKey(key string, writable bool) (uint64, error)
TranslateKeys(key []string, writable bool) ([]uint64, error)
// FindKeys looks up the ID for each key.
// Keys are not created if they do not exist.
// Missing keys are not considered errors, so the length of the result may be less than that of the input.
FindKeys(keys ...string) (map[string]uint64, error)
// CreateKeys maps all keys to IDs, creating the IDs if they do not exist.
// If the translator is read-only, this will return an error.
CreateKeys(keys ...string) (map[string]uint64, error)
// Converts an integer ID to its associated string key.
TranslateID(id uint64) (string, error)
TranslateIDs(id []uint64) ([]string, error)
@ -335,6 +344,57 @@ func (s *InMemTranslateStore) TranslateKeys(keys []string, writable bool) (_ []u
return ids, nil
}
// FindKeys looks up the ID for each key.
// Keys are not created if they do not exist.
// Missing keys are not considered errors, so the length of the result may be less than that of the input.
func (s *InMemTranslateStore) FindKeys(keys ...string) (map[string]uint64, error) {
s.mu.RLock()
defer s.mu.RUnlock()
result := make(map[string]uint64, len(keys))
for _, key := range keys {
id, ok := s.idsByKey[key]
if !ok {
// The key does not exist.
continue
}
result[key] = id
}
return result, nil
}
// CreateKeys maps all keys to IDs, creating the IDs if they do not exist.
// If the translator is read-only, this will return an error.
func (s *InMemTranslateStore) CreateKeys(keys ...string) (map[string]uint64, error) {
s.mu.Lock()
defer s.mu.Unlock()
if s.readOnly {
return nil, ErrTranslateStoreReadOnly
}
result := make(map[string]uint64, len(keys))
for _, key := range keys {
id, ok := s.idsByKey[key]
if !ok {
// The key does not exist.
// Generate a new id and update db.
if s.field == "" {
id = GenerateNextPartitionedID(s.index, s.maxID, s.partitionID, s.partitionN)
} else {
id = s.maxID + 1
}
s.set(id, key)
}
result[key] = id
}
return result, nil
}
func (s *InMemTranslateStore) translateKey(key string, writable bool) (_ uint64, err error) {
id := s.idsByKey[key]
if id != 0 {

211
translator_test.go
View file

@ -17,11 +17,11 @@ package pilosa_test
import (
"bytes"
"context"
"errors"
"fmt"
"io"
"reflect"
"testing"
"time"
"github.com/google/go-cmp/cmp"
"github.com/pilosa/pilosa/v2"
@ -30,6 +30,8 @@ import (
"github.com/pilosa/pilosa/v2/mock"
"github.com/pilosa/pilosa/v2/server"
"github.com/pilosa/pilosa/v2/test"
"github.com/pkg/errors"
"golang.org/x/sync/errgroup"
)
func TestInMemTranslateStore_TranslateKey(t *testing.T) {
@ -471,3 +473,210 @@ func TestTranslation_Coordinator(t *testing.T) {
}
})
}
func TestTranslation_Cluster_CreateFind(t *testing.T) {
c := test.MustRunCluster(t, 3)
defer c.Close()
c.CreateField(t, "i", pilosa.IndexOptions{Keys: true}, "f", pilosa.OptFieldKeys())
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
// Use the alphabet to test keys.
testKeys := make(map[string]struct{})
for i := 'a'; i <= 'z'; i++ {
testKeys[string(i)] = struct{}{}
}
t.Run("Index", func(t *testing.T) {
// Create all index keys, split across nodes.
{
parts := make([][]string, len(c))
{
// Randomly partition the keys.
i := 0
for k := range testKeys {
parts[i%len(c)] = append(parts[i%len(c)], k)
i++
}
}
// Create some keys on each node.
var g errgroup.Group
defer g.Wait() //nolint:errcheck
for i, keys := range parts {
i, keys := i, keys
g.Go(func() error {
_, err := c[i].API.CreateIndexKeys(ctx, "i", keys...)
return err
})
}
if err := g.Wait(); err != nil {
t.Errorf("creating keys: %v", err)
return
}
}
// Check that all index keys exist, and consistently map to the same IDs.
{
// Convert the keys to a list.
keyList := make([]string, 0, len(testKeys))
for k := range testKeys {
keyList = append(keyList, k)
}
// Obtain authoritative translations for the keys.
translations, err := c[0].API.FindIndexKeys(ctx, "i", keyList...)
if err != nil {
t.Errorf("obtaining authoritative translations: %v", err)
return
}
for _, k := range keyList {
if _, ok := translations[k]; !ok {
t.Errorf("key %q is missing", k)
}
}
// Check that all nodes agree on these translations.
var g errgroup.Group
defer g.Wait() //nolint:errcheck
for i, n := range c {
i, api := i, n.API
g.Go(func() (err error) {
defer func() { err = errors.Wrapf(err, "translating on node %d", i) }()
localTranslations, err := api.FindIndexKeys(ctx, "i", keyList...)
if err != nil {
return errors.Wrap(err, "finding translations")
}
return compareTranslations(translations, localTranslations)
})
}
if err := g.Wait(); err != nil {
t.Errorf("finding keys: %v", err)
return
}
// Check that re-invoking create returns the original translations.
for i, n := range c {
i, api := i, n.API
g.Go(func() (err error) {
defer func() { err = errors.Wrapf(err, "translating on node %d", i) }()
localTranslations, err := api.CreateIndexKeys(ctx, "i", keyList...)
if err != nil {
return errors.Wrap(err, "finding translations")
}
return compareTranslations(translations, localTranslations)
})
}
if err := g.Wait(); err != nil {
t.Errorf("checking re-create of keys: %v", err)
return
}
}
})
t.Run("Field", func(t *testing.T) {
// Create all field keys, split across nodes.
{
parts := make([][]string, len(c))
{
// Randomly partition the keys.
i := 0
for k := range testKeys {
parts[i%len(c)] = append(parts[i%len(c)], k)
i++
}
}
// Create some keys on each node.
var g errgroup.Group
defer g.Wait() //nolint:errcheck
for i, keys := range parts {
i, keys := i, keys
g.Go(func() error {
_, err := c[i].API.CreateFieldKeys(ctx, "i", "f", keys...)
return err
})
}
if err := g.Wait(); err != nil {
t.Errorf("creating keys: %v", err)
return
}
}
// Check that all field keys exist, and consistently map to the same IDs.
{
// Convert the keys to a list.
keyList := make([]string, 0, len(testKeys))
for k := range testKeys {
keyList = append(keyList, k)
}
// Obtain authoritative translations for the keys.
translations, err := c[0].API.FindFieldKeys(ctx, "i", "f", keyList...)
if err != nil {
t.Errorf("obtaining authoritative translations: %v", err)
return
}
for _, k := range keyList {
if _, ok := translations[k]; !ok {
t.Errorf("key %q is missing", k)
}
}
// Check that all nodes agree on these translations.
var g errgroup.Group
defer g.Wait() //nolint:errcheck
for i, n := range c {
i, api := i, n.API
g.Go(func() (err error) {
defer func() { err = errors.Wrapf(err, "translating on node %d", i) }()
localTranslations, err := api.FindFieldKeys(ctx, "i", "f", keyList...)
if err != nil {
return errors.Wrap(err, "finding translations")
}
return compareTranslations(translations, localTranslations)
})
}
if err := g.Wait(); err != nil {
t.Errorf("finding keys: %v", err)
return
}
// Check that re-invoking create returns the original translations.
for i, n := range c {
i, api := i, n.API
g.Go(func() (err error) {
defer func() { err = errors.Wrapf(err, "translating on node %d", i) }()
localTranslations, err := api.CreateFieldKeys(ctx, "i", "f", keyList...)
if err != nil {
return errors.Wrap(err, "finding translations")
}
return compareTranslations(translations, localTranslations)
})
}
if err := g.Wait(); err != nil {
t.Errorf("checking re-create of keys: %v", err)
return
}
}
})
}
func compareTranslations(expected, got map[string]uint64) error {
for key, id := range got {
if realID, ok := expected[key]; !ok {
return errors.Errorf("unexpected key %q mapped to ID %d", key, id)
} else if id != realID {
return errors.Errorf("mismatched translation: expected %q:%d but got %q:%d", key, realID, key, id)
}
}
for key, realID := range expected {
if id, ok := got[key]; !ok {
return errors.Errorf("missing translation of key %q", key)
} else if id != realID {
return errors.Errorf("mismatched translation: expected %q:%d but got %q:%d", key, realID, key, id)
}
}
return nil
}