mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-10-10 21:07:53 +00:00
standardize key translation on the find and create methods
This migrates existing code from the old TranslateKey(s) endpoints to the newer CreateKeys and FindKeys endpoints. The CreateKeys and FindKeys endpoints were created previously as the TranslateKeys endpoint had no way to behave sanely when the looked-up key did not exist (the parallel-arrays representation did not have a good way to represent a missing key). This change also removes the old TranslateKey(s) functions from the translation stores. It leaves a wrapper emulating the TranslateKey(s) endpoints so that old idk still works for now.
This commit is contained in:
Nia Weiss
parent
b886d1e571
commit
c0ef297b11
8 changed files with 454 additions and 773 deletions
|
|
@ -184,27 +184,6 @@ func (s *TranslateStore) Size() int64 {
|
|||
return tx.Size()
|
||||
}
|
||||
|
||||
// TranslateKey converts a string key to an integer ID.
|
||||
// If key does not have an associated id then one is created, unless writable is false,
|
||||
// then the function will return the error pilosa.ErrTranslatingKeyNotFound.
|
||||
func (s *TranslateStore) TranslateKey(key string, writable bool) (uint64, error) {
|
||||
ids, err := s.translateKeys([]string{key}, writable)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
if len(ids) == 0 {
|
||||
return 0, ErrTranslateKeyNotFound
|
||||
}
|
||||
return ids[0], nil
|
||||
}
|
||||
|
||||
// TranslateKeys converts a slice of string keys to a slice of integer IDs.
|
||||
// If a key does not have an associated id then one is created, unless writable is false,
|
||||
// then the function will return the error pilosa.ErrTranslatingKeyNotFound.
|
||||
func (s *TranslateStore) TranslateKeys(keys []string, writable bool) ([]uint64, error) {
|
||||
return s.translateKeys(keys, writable)
|
||||
}
|
||||
|
||||
// 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.
|
||||
|
|
@ -328,90 +307,6 @@ func (s *TranslateStore) Match(filter func([]byte) bool) ([]uint64, error) {
|
|||
return matches, nil
|
||||
}
|
||||
|
||||
func (s *TranslateStore) translateKeys(keys []string, writable bool) ([]uint64, error) {
|
||||
ids := make([]uint64, 0, len(keys))
|
||||
|
||||
if s.ReadOnly() || !writable {
|
||||
found := 0
|
||||
if err := s.db.View(func(tx *bolt.Tx) error {
|
||||
bkt := tx.Bucket(bucketKeys)
|
||||
if bkt == nil {
|
||||
return errors.Errorf(errFmtTranslateBucketNotFound, bucketKeys)
|
||||
}
|
||||
for _, key := range keys {
|
||||
if id, _ := findIDByKey(bkt, key); id != 0 {
|
||||
ids = append(ids, id)
|
||||
found++
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if found == len(keys) {
|
||||
return ids, nil
|
||||
}
|
||||
if s.ReadOnly() {
|
||||
return ids, pilosa.ErrTranslateStoreReadOnly
|
||||
}
|
||||
if !writable {
|
||||
return nil, pilosa.ErrTranslatingKeyNotFound
|
||||
}
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
// Find or create ids under write lock if any keys were not found.
|
||||
written := false
|
||||
idScratch := make([]byte, translateTransactionSize*8)
|
||||
for len(keys) > 0 {
|
||||
// boltdb performs badly if you write really large numbers of
|
||||
// keys all at once...
|
||||
if err := s.db.Update(func(tx *bolt.Tx) (err error) {
|
||||
keyBucket := tx.Bucket(bucketKeys)
|
||||
if keyBucket == nil {
|
||||
return errors.Errorf(errFmtTranslateBucketNotFound, bucketKeys)
|
||||
}
|
||||
idBucket := tx.Bucket(bucketIDs)
|
||||
if idBucket == nil {
|
||||
return errors.Errorf(errFmtTranslateBucketNotFound, bucketIDs)
|
||||
}
|
||||
puts := 0
|
||||
for idx, key := range keys {
|
||||
id, boltKey := findIDByKey(keyBucket, key)
|
||||
if id != 0 {
|
||||
ids = append(ids, id)
|
||||
continue
|
||||
}
|
||||
id = pilosa.GenerateNextPartitionedID(s.index, maxID(tx), s.partitionID, s.partitionN)
|
||||
idBytes := idScratch[puts*8 : puts*8+8]
|
||||
binary.BigEndian.PutUint64(idBytes, id)
|
||||
puts++
|
||||
if err := keyBucket.Put(boltKey, idBytes); err != nil {
|
||||
return err
|
||||
} else if err := idBucket.Put(idBytes, boltKey); err != nil {
|
||||
return err
|
||||
}
|
||||
ids = append(ids, id)
|
||||
written = true
|
||||
if puts == translateTransactionSize {
|
||||
keys = keys[idx+1:]
|
||||
return nil
|
||||
}
|
||||
}
|
||||
// this marks that we've processed the whole list.
|
||||
keys = keys[len(keys):]
|
||||
return nil
|
||||
}); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
if written {
|
||||
s.notifyWrite()
|
||||
}
|
||||
return ids, nil
|
||||
}
|
||||
|
||||
// TranslateID converts an integer ID to a string key.
|
||||
// Returns a blank string if ID does not exist.
|
||||
func (s *TranslateStore) TranslateID(id uint64) (string, error) {
|
||||
|
|
|
|||
|
|
@ -30,71 +30,6 @@ import (
|
|||
|
||||
//var vv = pilosa.VV
|
||||
|
||||
func TestTranslateStore_TranslateKey(t *testing.T) {
|
||||
s := MustOpenNewTranslateStore(t)
|
||||
defer MustCloseTranslateStore(s)
|
||||
|
||||
// Ensure initial key translates to first ID for shard
|
||||
id1, err := s.TranslateKey("foo", true)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// Ensure next key autoincrements.
|
||||
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", true); err != nil {
|
||||
t.Fatal(err)
|
||||
} else if got, want := id, id1; got != want {
|
||||
t.Fatalf("TranslateKey()=%d, want %d", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTranslateStore_TranslateKeys(t *testing.T) {
|
||||
s := MustOpenNewTranslateStore(t)
|
||||
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"}, true)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
} else if got, want := ids1[1], ids1[0]+1; got != want {
|
||||
t.Fatalf("TranslateKeys()[1]=%d, want %d", got, want)
|
||||
}
|
||||
|
||||
// Ensure retranslation returns original IDs.
|
||||
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)
|
||||
} else if got, want := ids[1], ids1[1]; got != want {
|
||||
t.Fatalf("TranslateKeys()[1]=%d, want %d", got, want)
|
||||
}
|
||||
|
||||
// Ensure retranslating with existing and non-existing keys returns correctly.
|
||||
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)
|
||||
} else if got, want := ids[1], ids1[0]+2; got != want {
|
||||
t.Fatalf("TranslateKeys()[1]=%d, want %d", got, want)
|
||||
} else if got, want := ids[2], ids1[1]; got != want {
|
||||
t.Fatalf("TranslateKeys()[2]=%d, want %d", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTranslateStore_CreateKeys(t *testing.T) {
|
||||
s := MustOpenNewTranslateStore(t)
|
||||
defer MustCloseTranslateStore(s)
|
||||
|
|
@ -135,106 +70,26 @@ func TestTranslateStore_CreateKeys(t *testing.T) {
|
|||
}
|
||||
}
|
||||
|
||||
func TestTranslateStore_ReadKey(t *testing.T) {
|
||||
s := MustOpenNewTranslateStore(t)
|
||||
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(t)
|
||||
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(t)
|
||||
defer MustCloseTranslateStore(s)
|
||||
|
||||
// Setup initial keys.
|
||||
id1, err := s.TranslateKey("foo", true)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
id2, err := s.TranslateKey("bar", true)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
id3, err := s.TranslateKey("", true)
|
||||
ids, err := s.CreateKeys("foo", "bar", "")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// Ensure IDs can be translated back to keys.
|
||||
if key, err := s.TranslateID(id1); err != nil {
|
||||
t.Fatal(err)
|
||||
} else if got, want := key, "foo"; got != want {
|
||||
t.Fatalf("TranslateID()=%s, want %s", got, want)
|
||||
for key, id := range ids {
|
||||
k, err := s.TranslateID(id)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if k != key {
|
||||
t.Fatalf("TranslateID()=%s, want %s", k, key)
|
||||
}
|
||||
}
|
||||
|
||||
if key, err := s.TranslateID(id2); err != nil {
|
||||
t.Fatal(err)
|
||||
} else if got, want := key, "bar"; got != want {
|
||||
t.Fatalf("TranslateID()=%s, want %s", got, want)
|
||||
}
|
||||
|
||||
if key, err := s.TranslateID(id3); err != nil {
|
||||
t.Fatal(err)
|
||||
} else if got, want := key, ""; got != want {
|
||||
t.Fatalf("TranslateID()=%s, want %s", got, want)
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
func TestTranslateStore_TranslateIDs(t *testing.T) {
|
||||
|
|
@ -242,13 +97,13 @@ func TestTranslateStore_TranslateIDs(t *testing.T) {
|
|||
defer MustCloseTranslateStore(s)
|
||||
|
||||
// Setup initial keys.
|
||||
ids, err := s.TranslateKeys([]string{"foo", "bar"}, true)
|
||||
ids, err := s.CreateKeys("foo", "bar")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// Ensure IDs can be translated back to keys.
|
||||
if keys, err := s.TranslateIDs([]uint64{ids[0], ids[1], 1}); err != nil {
|
||||
if keys, err := s.TranslateIDs([]uint64{ids["foo"], ids["bar"], 1}); err != nil {
|
||||
t.Fatal(err)
|
||||
} else if got, want := keys[0], "foo"; got != want {
|
||||
t.Fatalf("TranslateIDs()[0]=%s, want %s", got, want)
|
||||
|
|
@ -299,7 +154,7 @@ func TestTranslateStore_FindKeys(t *testing.T) {
|
|||
if c.data != nil {
|
||||
// Load in key data.
|
||||
keys := c.data
|
||||
ids, err := s.TranslateKeys(keys, true)
|
||||
ids, err := s.CreateKeys(keys...)
|
||||
if err != nil {
|
||||
t.Errorf("failed to import keys: %v", err)
|
||||
return
|
||||
|
|
@ -308,10 +163,7 @@ func TestTranslateStore_FindKeys(t *testing.T) {
|
|||
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]
|
||||
}
|
||||
naiveMap = ids
|
||||
}
|
||||
|
||||
// Compute expected lookup result.
|
||||
|
|
@ -344,11 +196,12 @@ func TestTranslateStore_MaxID(t *testing.T) {
|
|||
// Generate a bunch of keys.
|
||||
var lastk uint64
|
||||
for i := 0; i < 1026; i++ {
|
||||
k, err := s.TranslateKey(strconv.Itoa(i), true)
|
||||
key := strconv.Itoa(i)
|
||||
ids, err := s.CreateKeys(key)
|
||||
if err != nil {
|
||||
t.Fatalf("translating %d: %v", i, err)
|
||||
}
|
||||
lastk = k
|
||||
lastk = ids[key]
|
||||
}
|
||||
|
||||
// Verify the max ID.
|
||||
|
|
@ -367,7 +220,7 @@ func TestTranslateStore_EntryReader(t *testing.T) {
|
|||
defer MustCloseTranslateStore(s)
|
||||
|
||||
// Create multiple new keys.
|
||||
ids1, err := s.TranslateKeys([]string{"foo", "bar"}, true)
|
||||
ids1, err := s.CreateKeys("foo", "bar")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
|
@ -383,7 +236,7 @@ func TestTranslateStore_EntryReader(t *testing.T) {
|
|||
// Read first entry.
|
||||
if err := r.ReadEntry(&entry); err != nil {
|
||||
t.Fatal(err)
|
||||
} else if got, want := entry.ID, ids1[0]; got != want {
|
||||
} else if got, want := entry.ID, ids1["foo"]; got != want {
|
||||
t.Fatalf("ReadEntry() ID=%d, want %d", got, want)
|
||||
} else if got, want := entry.Key, "foo"; got != want {
|
||||
t.Fatalf("ReadEntry() Key=%s, want %s", got, want)
|
||||
|
|
@ -392,14 +245,14 @@ func TestTranslateStore_EntryReader(t *testing.T) {
|
|||
// Read next entry.
|
||||
if err := r.ReadEntry(&entry); err != nil {
|
||||
t.Fatal(err)
|
||||
} else if got, want := entry.ID, ids1[1]; got != want {
|
||||
} else if got, want := entry.ID, ids1["bar"]; got != want {
|
||||
t.Fatalf("ReadEntry() ID=%d, want %d", got, want)
|
||||
} else if got, want := entry.Key, "bar"; got != want {
|
||||
t.Fatalf("ReadEntry() Key=%s, want %s", got, want)
|
||||
}
|
||||
|
||||
// Insert next key while reader is open.
|
||||
id2, err := s.TranslateKey("baz", true)
|
||||
ids2, err := s.CreateKeys("baz")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
|
@ -407,7 +260,7 @@ func TestTranslateStore_EntryReader(t *testing.T) {
|
|||
// Read newly created entry.
|
||||
if err := r.ReadEntry(&entry); err != nil {
|
||||
t.Fatal(err)
|
||||
} else if got, want := entry.ID, id2; got != want {
|
||||
} else if got, want := entry.ID, ids2["baz"]; got != want {
|
||||
t.Fatalf("ReadEntry() ID=%d, want %d", got, want)
|
||||
} else if got, want := entry.Key, "baz"; got != want {
|
||||
t.Fatalf("ReadEntry() Key=%s, want %s", got, want)
|
||||
|
|
@ -439,11 +292,11 @@ func TestTranslateStore_EntryReader(t *testing.T) {
|
|||
translateErr := make(chan error)
|
||||
go func() {
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
id, err := s.TranslateKey("foo", true)
|
||||
ids, err := s.CreateKeys("foo")
|
||||
if err != nil {
|
||||
translateErr <- err
|
||||
}
|
||||
cache <- id
|
||||
cache <- ids["foo"]
|
||||
}()
|
||||
|
||||
var entry pilosa.TranslateEntry
|
||||
|
|
@ -560,7 +413,7 @@ func TestTranslateStore_ReadWrite(t *testing.T) {
|
|||
}
|
||||
|
||||
// Populate the store with the keys in batch0.
|
||||
batch0IDs, err := s.TranslateKeys(batch0, true)
|
||||
batch0IDs, err := s.CreateKeys(batch0...)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
|
@ -577,15 +430,18 @@ func TestTranslateStore_ReadWrite(t *testing.T) {
|
|||
}
|
||||
|
||||
// Populate the store with the keys in batch1.
|
||||
batch1IDs, err := s.TranslateKeys(batch1, true)
|
||||
batch1IDs, err := s.CreateKeys(batch1...)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
expIDs := []uint64{batch0IDs[50], batch1IDs[50]}
|
||||
expIDs := map[string]uint64{
|
||||
"key50": batch0IDs["key50"],
|
||||
"key150": batch1IDs["key150"],
|
||||
}
|
||||
|
||||
// Check the IDs for a key from each batch.
|
||||
if ids, err := s.TranslateKeys([]string{"key50", "key150"}, false); err != nil {
|
||||
if ids, err := s.FindKeys("key50", "key150"); err != nil {
|
||||
t.Fatal(err)
|
||||
} else if !reflect.DeepEqual(expIDs, ids) {
|
||||
t.Fatalf("first expected ids: %v, but got: %v", expIDs, ids)
|
||||
|
|
@ -600,12 +456,12 @@ 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"}, true); err != nil {
|
||||
if ids, err := s.CreateKeys("key50", "key150"); 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])
|
||||
} else if ids[1] == expIDs[1] {
|
||||
t.Fatalf("last expected different ids[1]: %d, but got: %d", expIDs[1], ids[1])
|
||||
} else if ids["key50"] != expIDs["key50"] {
|
||||
t.Fatalf("last expected ids[key50]: %d, but got: %d", expIDs["key50"], ids["key50"])
|
||||
} else if ids["key150"] == expIDs["key150"] {
|
||||
t.Fatalf("last expected different ids[key150]: %d, but got: %d", expIDs["key150"], ids["key150"])
|
||||
}
|
||||
})
|
||||
}
|
||||
|
|
|
|||
368
client/batch.go
368
client/batch.go
|
|
@ -401,29 +401,6 @@ func (b *Batch) getRowTranslation(field, key string) (uint64, bool) {
|
|||
return trans.id, ok
|
||||
}
|
||||
|
||||
func (b *Batch) addRowTranslations(fieldName string, keys []string, ids []uint64) {
|
||||
rowCache := b.rowTranslations[fieldName]
|
||||
if rowCache == nil {
|
||||
rowCache = make(map[string]agedTranslation)
|
||||
b.rowTranslations[fieldName] = rowCache
|
||||
}
|
||||
for i, k := range keys {
|
||||
rowCache[k] = agedTranslation{
|
||||
id: ids[i],
|
||||
lastUsed: b.cycle,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (b *Batch) addColTranslations(keys []string, ids []uint64) {
|
||||
for i, k := range keys {
|
||||
b.colTranslations[k] = agedTranslation{
|
||||
id: ids[i],
|
||||
lastUsed: b.cycle,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Add adds a record to the batch. Performance will be best if record
|
||||
// IDs are shard-sorted. That is, all records which belong to the same
|
||||
// Pilosa shard are added adjacent to each other. If the records are
|
||||
|
|
@ -773,196 +750,273 @@ func (b *Batch) Flush() error {
|
|||
}
|
||||
|
||||
func (b *Batch) doTranslation() error {
|
||||
keys := make([]string, 0)
|
||||
eg := egpool.Group{PoolSize: 20}
|
||||
|
||||
// translate column keys if there are any
|
||||
|
||||
// TODO test. Also this implementation (using a set to de-dup
|
||||
// keys) will likely have much worse performance than the previous
|
||||
// one (two slices, one of keys one of ids) in the case that most
|
||||
// of the keys are unique.
|
||||
keySet := make(map[string]uint64)
|
||||
for _, key := range b.toTranslateID {
|
||||
if key != "" {
|
||||
if _, ok := keySet[key]; ok {
|
||||
// Translate the column keys.
|
||||
eg.Go(func() error {
|
||||
// Dedupliucate keys to translate.
|
||||
dedup := make(map[string]struct{})
|
||||
var keys []string
|
||||
for _, key := range b.toTranslateID {
|
||||
if key == "" {
|
||||
continue
|
||||
}
|
||||
keys = append(keys, key)
|
||||
keySet[key] = 0
|
||||
}
|
||||
}
|
||||
|
||||
if len(keys) > 0 {
|
||||
if _, ok := dedup[key]; ok {
|
||||
continue
|
||||
}
|
||||
dedup[key] = struct{}{}
|
||||
|
||||
keys = append(keys, key)
|
||||
}
|
||||
if len(keys) == 0 {
|
||||
// There are no column keys to translate.
|
||||
return nil
|
||||
}
|
||||
|
||||
// Create the keys.
|
||||
start := time.Now()
|
||||
ids, err := b.translateColumnKeys(b.index, keys)
|
||||
trans, err := b.createIndexKeys(b.index, keys...)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "translating col keys")
|
||||
}
|
||||
if len(ids) != len(keys) {
|
||||
return errors.Errorf("requested IDs for %d column keys but got %d back", len(keys), len(ids))
|
||||
if len(trans) != len(keys) {
|
||||
return errors.Errorf("requested IDs for %d column keys but got %d back", len(keys), len(trans))
|
||||
}
|
||||
b.log.Debugf("translating %d column keys took %v", len(keys), time.Since(start))
|
||||
b.addColTranslations(keys, ids)
|
||||
for j, id := range ids {
|
||||
keySet[keys[j]] = id
|
||||
}
|
||||
for index, ttkey := range b.toTranslateID {
|
||||
if ttkey != "" {
|
||||
b.ids[index] = keySet[ttkey]
|
||||
|
||||
// Apply keys to translation cache.
|
||||
for key, id := range trans {
|
||||
b.colTranslations[key] = agedTranslation{
|
||||
id: id,
|
||||
lastUsed: b.cycle,
|
||||
}
|
||||
}
|
||||
}
|
||||
// translate row keys
|
||||
|
||||
// Translate remaining keys in batch.
|
||||
for index, ttkey := range b.toTranslateID {
|
||||
if ttkey == "" {
|
||||
continue
|
||||
}
|
||||
|
||||
b.ids[index] = trans[ttkey]
|
||||
}
|
||||
|
||||
return nil
|
||||
})
|
||||
|
||||
// Translate the row keys.
|
||||
for i, tt := range b.toTranslate {
|
||||
fieldName := b.header[i].Name()
|
||||
keys = keys[:0]
|
||||
|
||||
// make a slice of keys
|
||||
for k := range tt {
|
||||
keys = append(keys, k)
|
||||
}
|
||||
// append keys to clear so we can translate them all in one
|
||||
// request. ttEnd is the index where clearing starts which we
|
||||
// use later on.
|
||||
ttEnd := len(keys)
|
||||
// Skip this if there are no keys to translate.
|
||||
ttc := b.toTranslateClear[i]
|
||||
for k := range ttc {
|
||||
keys = append(keys, k)
|
||||
}
|
||||
|
||||
if len(keys) == 0 {
|
||||
if len(tt) == 0 && len(ttc) == 0 {
|
||||
continue
|
||||
}
|
||||
|
||||
// translate keys from Pilosa
|
||||
start := time.Now()
|
||||
ids, err := b.translateRowKeys(b.headerMap[fieldName], keys)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "translating row keys")
|
||||
}
|
||||
if len(ids) != len(keys) {
|
||||
return errors.Errorf("requested IDs for %d row keys but got %d back", len(keys), len(ids))
|
||||
}
|
||||
b.log.Debugf("translating %d row keys for %s took %v", len(keys), fieldName, time.Since(start))
|
||||
b.addRowTranslations(fieldName, keys, ids)
|
||||
// Look up the associated field.
|
||||
field := b.header[i]
|
||||
fieldName := field.Name()
|
||||
|
||||
switch b.header[i].Opts().Type() {
|
||||
case FieldTypeInt:
|
||||
// handle foreign key int fields — fill out b.values instead of b.rows
|
||||
for j := 0; j < ttEnd; j++ {
|
||||
key := keys[j]
|
||||
id := ids[j]
|
||||
for _, recordIdx := range tt[key] {
|
||||
b.values[fieldName][recordIdx] = int64(id)
|
||||
}
|
||||
}
|
||||
case FieldTypeDecimal:
|
||||
return errors.Errorf("unexpected field type for translation: decimal")
|
||||
default:
|
||||
// fill out missing IDs in local batch records with translated IDs
|
||||
rows := b.rowIDs[i]
|
||||
for j := 0; j < ttEnd; j++ {
|
||||
key := keys[j]
|
||||
id := ids[j]
|
||||
for _, recordIdx := range tt[key] {
|
||||
rows[recordIdx] = id
|
||||
}
|
||||
}
|
||||
// fill out missing IDs in clear lists.
|
||||
clearRows := b.clearRowIDs[i]
|
||||
for j := ttEnd; j < len(keys); j++ {
|
||||
key := keys[j]
|
||||
id := ids[j]
|
||||
for _, recordIdx := range ttc[key] {
|
||||
clearRows[recordIdx] = id
|
||||
}
|
||||
}
|
||||
// Fetch the translation cache.
|
||||
rowCache := b.rowTranslations[fieldName]
|
||||
if rowCache == nil {
|
||||
rowCache = make(map[string]agedTranslation)
|
||||
b.rowTranslations[fieldName] = rowCache
|
||||
}
|
||||
|
||||
i, tt := i, tt
|
||||
eg.Go(func() error {
|
||||
// Collect the keys to translate.
|
||||
keys := make([]string, 0, len(tt)+len(ttc))
|
||||
for k := range tt {
|
||||
keys = append(keys, k)
|
||||
}
|
||||
for k := range ttc {
|
||||
keys = append(keys, k)
|
||||
}
|
||||
|
||||
// Create the keys.
|
||||
start := time.Now()
|
||||
trans, err := b.createFieldKeys(field, keys...)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "translating field keys")
|
||||
}
|
||||
b.log.Debugf("translating %d field keys for %s took %v", len(trans), fieldName, time.Since(start))
|
||||
|
||||
// Apply keys to translation cache.
|
||||
for key, id := range trans {
|
||||
rowCache[key] = agedTranslation{
|
||||
id: id,
|
||||
lastUsed: b.cycle,
|
||||
}
|
||||
}
|
||||
|
||||
switch ftype := field.Opts().Type(); ftype {
|
||||
case FieldTypeSet, FieldTypeMutex:
|
||||
// Fill out missing IDs in local batch records with translated IDs.
|
||||
rows := b.rowIDs[i]
|
||||
for key, idxs := range tt {
|
||||
id, ok := trans[key]
|
||||
if !ok {
|
||||
return errors.Errorf("key translation missing: %q in field %q", key, fieldName)
|
||||
}
|
||||
|
||||
for _, i := range idxs {
|
||||
rows[i] = id
|
||||
}
|
||||
}
|
||||
|
||||
// Fill out missing IDs in clear lists.
|
||||
clearRows := b.clearRowIDs[i]
|
||||
for key, idxs := range ttc {
|
||||
id, ok := trans[key]
|
||||
if !ok {
|
||||
return errors.Errorf("key translation missing: %q in field %q", key, fieldName)
|
||||
}
|
||||
|
||||
for _, i := range idxs {
|
||||
clearRows[i] = id
|
||||
}
|
||||
}
|
||||
|
||||
case FieldTypeInt:
|
||||
// Handle foreign key int fields — fill out b.values instead of b.rows.
|
||||
vals := b.values[fieldName]
|
||||
for key, idxs := range tt {
|
||||
id, ok := trans[key]
|
||||
if !ok {
|
||||
return errors.Errorf("key translation missing: %q in field %q", key, fieldName)
|
||||
}
|
||||
|
||||
for _, i := range idxs {
|
||||
vals[i] = int64(id)
|
||||
}
|
||||
}
|
||||
|
||||
default:
|
||||
return errors.Errorf("unexpected field type for translation: %q", ftype)
|
||||
}
|
||||
|
||||
return nil
|
||||
})
|
||||
}
|
||||
|
||||
for fieldName, tt := range b.toTranslateSets {
|
||||
keys = keys[:0]
|
||||
|
||||
for k := range tt {
|
||||
keys = append(keys, k)
|
||||
}
|
||||
|
||||
if len(keys) == 0 {
|
||||
// Skip this if there are no keys to translate.
|
||||
if len(tt) == 0 {
|
||||
continue
|
||||
}
|
||||
// translate keys from Pilosa
|
||||
start := time.Now()
|
||||
ids, err := b.translateRowKeys(b.headerMap[fieldName], keys)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "translating row keys (sets)")
|
||||
|
||||
// Look up the associated field.
|
||||
field := b.headerMap[fieldName]
|
||||
|
||||
// Fetch the translation cache.
|
||||
rowCache := b.rowTranslations[fieldName]
|
||||
if rowCache == nil {
|
||||
rowCache = make(map[string]agedTranslation)
|
||||
b.rowTranslations[fieldName] = rowCache
|
||||
}
|
||||
if len(ids) != len(keys) {
|
||||
return errors.Errorf("requested IDs for %d row (set) keys but got %d back", len(keys), len(ids))
|
||||
}
|
||||
b.log.Debugf("translating %d row keys(sets) for %s took %v", len(keys), fieldName, time.Since(start))
|
||||
b.addRowTranslations(fieldName, keys, ids)
|
||||
rowIDSets := b.rowIDSets[fieldName]
|
||||
rowIDSets = rowIDSets[:cap(b.ids)]
|
||||
b.rowIDSets[fieldName] = rowIDSets
|
||||
for j, key := range keys {
|
||||
rowID := ids[j]
|
||||
for _, recordIdx := range tt[key] {
|
||||
rowIDSets[recordIdx] = append(rowIDSets[recordIdx], rowID)
|
||||
|
||||
fieldName, tt := fieldName, tt
|
||||
eg.Go(func() error {
|
||||
// Collect the keys to translate.
|
||||
keys := make([]string, 0, len(tt))
|
||||
for k := range tt {
|
||||
keys = append(keys, k)
|
||||
}
|
||||
}
|
||||
|
||||
// Create the keys.
|
||||
start := time.Now()
|
||||
trans, err := b.createFieldKeys(field, keys...)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "translating field keys")
|
||||
}
|
||||
b.log.Debugf("translating %d field keys for %s took %v", len(trans), fieldName, time.Since(start))
|
||||
|
||||
// Apply keys to translation cache.
|
||||
for key, id := range trans {
|
||||
rowCache[key] = agedTranslation{
|
||||
id: id,
|
||||
lastUsed: b.cycle,
|
||||
}
|
||||
}
|
||||
|
||||
// Fill out missing IDs in local batch records with translated IDs.
|
||||
rowIDSets := b.rowIDSets[fieldName]
|
||||
for key, idxs := range tt {
|
||||
id, ok := trans[key]
|
||||
if !ok {
|
||||
return errors.Errorf("key translation missing: %q in field %q", key, fieldName)
|
||||
}
|
||||
|
||||
for _, i := range idxs {
|
||||
rowIDSets[i] = append(rowIDSets[i], id)
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
})
|
||||
}
|
||||
|
||||
return nil
|
||||
return eg.Wait()
|
||||
}
|
||||
|
||||
func (b *Batch) translateColumnKeys(index *Index, keys []string) ([]uint64, error) {
|
||||
func (b *Batch) createIndexKeys(index *Index, keys ...string) (map[string]uint64, error) {
|
||||
batchSize := b.keyTranslateBatchSize
|
||||
if batchSize <= 0 {
|
||||
batchSize = len(keys)
|
||||
if batchSize <= 0 || len(keys) <= batchSize {
|
||||
return b.client.CreateIndexKeys(index, keys...)
|
||||
}
|
||||
|
||||
ids := make([]uint64, 0, len(keys))
|
||||
for i := 0; i < len(keys); i += batchSize {
|
||||
keySlice := keys[i:]
|
||||
results := make(map[string]uint64, len(keys))
|
||||
for len(keys) > 0 {
|
||||
keySlice := keys
|
||||
if len(keySlice) > batchSize {
|
||||
keySlice = keySlice[:batchSize]
|
||||
}
|
||||
|
||||
idSlice, err := b.client.TranslateColumnKeys(b.index, keySlice)
|
||||
trans, err := b.client.CreateIndexKeys(index, keySlice...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
} else if len(idSlice) != len(keySlice) {
|
||||
return nil, errors.Errorf("requested IDs slice for %d column keys but got %d back", len(keySlice), len(idSlice))
|
||||
} else if len(trans) != len(keySlice) {
|
||||
return nil, errors.Errorf("requested IDs for %d column keys but got %d back", len(keySlice), len(trans))
|
||||
}
|
||||
ids = append(ids, idSlice...)
|
||||
for key, id := range trans {
|
||||
results[key] = id
|
||||
}
|
||||
|
||||
keys = keys[len(keySlice):]
|
||||
}
|
||||
|
||||
return ids, nil
|
||||
return results, nil
|
||||
}
|
||||
|
||||
func (b *Batch) translateRowKeys(field *Field, keys []string) ([]uint64, error) {
|
||||
func (b *Batch) createFieldKeys(field *Field, keys ...string) (map[string]uint64, error) {
|
||||
batchSize := b.keyTranslateBatchSize
|
||||
if batchSize <= 0 {
|
||||
batchSize = len(keys)
|
||||
if batchSize <= 0 || len(keys) <= batchSize {
|
||||
return b.client.CreateFieldKeys(field, keys...)
|
||||
}
|
||||
|
||||
ids := make([]uint64, 0, len(keys))
|
||||
for i := 0; i < len(keys); i += batchSize {
|
||||
keySlice := keys[i:]
|
||||
results := make(map[string]uint64, len(keys))
|
||||
for len(keys) > 0 {
|
||||
keySlice := keys
|
||||
if len(keySlice) > batchSize {
|
||||
keySlice = keySlice[:batchSize]
|
||||
}
|
||||
|
||||
idSlice, err := b.client.TranslateRowKeys(field, keySlice)
|
||||
trans, err := b.client.CreateFieldKeys(field, keySlice...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
} else if len(idSlice) != len(keySlice) {
|
||||
return nil, errors.Errorf("requested IDs slice for %d row keys but got %d back", len(keySlice), len(idSlice))
|
||||
} else if len(trans) != len(keySlice) {
|
||||
return nil, errors.Errorf("requested IDs for %d row keys but got %d back", len(keySlice), len(trans))
|
||||
}
|
||||
ids = append(ids, idSlice...)
|
||||
for key, id := range trans {
|
||||
results[key] = id
|
||||
}
|
||||
|
||||
keys = keys[len(keySlice):]
|
||||
}
|
||||
|
||||
return ids, nil
|
||||
return results, nil
|
||||
}
|
||||
|
||||
func (b *Batch) doImport(frags, clearFrags fragments) error {
|
||||
|
|
|
|||
257
client/client.go
257
client/client.go
|
|
@ -21,7 +21,6 @@ import (
|
|||
"crypto/tls"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"hash/fnv"
|
||||
"io"
|
||||
"math/rand"
|
||||
"net"
|
||||
|
|
@ -75,9 +74,6 @@ type Client struct {
|
|||
shardNodes shardNodes
|
||||
tick *time.Ticker
|
||||
done chan struct{}
|
||||
|
||||
// TODO — remove this! Pilosa should have API which gives us the node->partition mapping
|
||||
Hasher Hasher
|
||||
}
|
||||
|
||||
func (c *Client) getURIsForShard(index string, shard uint64) ([]*pnet.URI, error) {
|
||||
|
|
@ -209,9 +205,6 @@ func newClientWithOptions(options *ClientOptions) *Client {
|
|||
done: make(chan struct{}),
|
||||
|
||||
nat: options.nat,
|
||||
|
||||
// TODO get rid of this. Pilosa should have api to expose node->partition mapping.
|
||||
Hasher: &jmphasher{},
|
||||
}
|
||||
|
||||
if options.tracer == nil {
|
||||
|
|
@ -1017,150 +1010,132 @@ func (c *Client) augmentHeaders(headers map[string]string) map[string]string {
|
|||
return headers
|
||||
}
|
||||
|
||||
func (c *Client) TranslateRowKeys(field *Field, keys []string) ([]uint64, error) {
|
||||
req := &pb.TranslateKeysRequest{
|
||||
Index: field.index.name,
|
||||
Field: field.name,
|
||||
Keys: keys,
|
||||
}
|
||||
return c.translateKeys(req)
|
||||
}
|
||||
// FindFieldKeys looks up the IDs associated with specified keys in a field.
|
||||
// If a key does not exist, the result will not include it.
|
||||
func (c *Client) FindFieldKeys(field *Field, keys ...string) (map[string]uint64, error) {
|
||||
path := fmt.Sprintf("/internal/translate/field/%s/%s/keys/find", field.index.name, field.name)
|
||||
|
||||
func (c *Client) TranslateColumnKeys(index *Index, keys []string) ([]uint64, error) {
|
||||
// If a manual server URI override has been provided, there's no
|
||||
// point in partitioning the translation request on the client
|
||||
// because every request is going to be sent to the manual URI.
|
||||
if c.manualServerURI != nil {
|
||||
req := &pb.TranslateKeysRequest{
|
||||
Index: index.name,
|
||||
Keys: keys,
|
||||
}
|
||||
return c.translateKeys(req)
|
||||
}
|
||||
|
||||
// Get the list of hosts from the server.
|
||||
// TODO: it's not ideal to request the list of nodes from the server
|
||||
// on every call to TranslateColumnKeys(), but if we cache that list
|
||||
// on the client, we risk calculating the partition distribution based
|
||||
// on a stale node list. This TODO is here to indicate that we may,
|
||||
// in the future, want to remove the overhead of this status request.
|
||||
status, err := c.Status()
|
||||
reqData, err := json.Marshal(keys)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "getting cluster status")
|
||||
return nil, errors.Wrap(err, "marshalling request")
|
||||
}
|
||||
|
||||
hosts := make([]pnet.URI, len(status.Nodes))
|
||||
for i, node := range status.Nodes {
|
||||
hosts[i] = node.URI.URI()
|
||||
}
|
||||
headers := c.augmentHeaders(map[string]string{
|
||||
"Content-Type": "application/json",
|
||||
"Accept": "application/json",
|
||||
})
|
||||
|
||||
keysByNode := make(map[pnet.URI][]string, len(hosts))
|
||||
for _, key := range keys {
|
||||
// TODO 256 is DefaultPartitionN in Pilosa. Eventually this will likely be an index configuration parameter.
|
||||
partitionID := keyPartition(index.Name(), key, 256)
|
||||
uri := c.partitionOwner(partitionID, hosts)
|
||||
keysByNode[uri] = append(keysByNode[uri], key)
|
||||
}
|
||||
|
||||
eg := errgroup.Group{}
|
||||
idsByNode := make(map[pnet.URI][]uint64, len(keysByNode))
|
||||
ibnLock := &sync.Mutex{}
|
||||
for uri, keys := range keysByNode {
|
||||
uri := uri
|
||||
keys := keys
|
||||
eg.Go(func() error {
|
||||
req := &pb.TranslateKeysRequest{
|
||||
Index: index.name,
|
||||
Keys: keys,
|
||||
}
|
||||
ids, err := c.translateKeys(req, uri)
|
||||
if err != nil {
|
||||
return errors.Wrapf(err, "translating column keys at %v", uri)
|
||||
}
|
||||
ibnLock.Lock()
|
||||
idsByNode[uri] = ids
|
||||
ibnLock.Unlock()
|
||||
return nil
|
||||
})
|
||||
}
|
||||
if err := eg.Wait(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
finalIDs := make([]uint64, len(keys))
|
||||
// put the ids back together into one slice
|
||||
for uri, uriIds := range idsByNode {
|
||||
uriKeys := keysByNode[uri]
|
||||
kidx := 0
|
||||
for i, key := range uriKeys {
|
||||
for ; keys[kidx] != key; kidx++ {
|
||||
}
|
||||
finalIDs[kidx] = uriIds[i]
|
||||
kidx++
|
||||
}
|
||||
}
|
||||
return finalIDs, nil
|
||||
}
|
||||
|
||||
func (c *Client) partitionOwner(partitionID int, hosts []pnet.URI) pnet.URI {
|
||||
nodeIndex := c.Hasher.Hash(uint64(partitionID), len(hosts))
|
||||
return hosts[nodeIndex]
|
||||
}
|
||||
|
||||
// Hasher represents an interface to hash integers into buckets.
|
||||
type Hasher interface {
|
||||
// Hashes the key into a number between [0,N).
|
||||
Hash(key uint64, n int) int
|
||||
}
|
||||
|
||||
// jmphasher represents an implementation of jmphash. Implements Hasher.
|
||||
type jmphasher struct{}
|
||||
|
||||
// Hash returns the integer hash for the given key.
|
||||
func (h *jmphasher) Hash(key uint64, n int) int {
|
||||
b, j := int64(-1), int64(0)
|
||||
for j < int64(n) {
|
||||
b = j
|
||||
key = key*uint64(2862933555777941757) + 1
|
||||
j = int64(float64(b+1) * (float64(int64(1)<<31) / float64((key>>33)+1)))
|
||||
}
|
||||
return int(b)
|
||||
}
|
||||
|
||||
func keyPartition(index, key string, partitionN int) int {
|
||||
// Hash the bytes and mod by partition count.
|
||||
h := fnv.New64a()
|
||||
_, _ = h.Write([]byte(index))
|
||||
_, _ = h.Write([]byte(key))
|
||||
return int(h.Sum64() % uint64(partitionN))
|
||||
}
|
||||
|
||||
func (c *Client) translateKeys(req *pb.TranslateKeysRequest, uris ...pnet.URI) ([]uint64, error) {
|
||||
if len(req.Keys) == 0 {
|
||||
return []uint64{}, nil
|
||||
}
|
||||
reqData, err := proto.Marshal(req)
|
||||
status, body, err := c.HTTPRequest(http.MethodPost, path, reqData, headers)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "marshalling translate keys request")
|
||||
return nil, errors.Wrap(err, "executing request")
|
||||
}
|
||||
if status != http.StatusOK {
|
||||
return nil, errors.Errorf("find field keys request failed (status: %d): %q", status, string(body))
|
||||
}
|
||||
|
||||
var respData []byte
|
||||
if len(uris) == 0 {
|
||||
if _, respData, err = c.httpRequest("POST", "/internal/translate/keys", reqData, defaultProtobufHeaders(), true); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
} else {
|
||||
if _, respData, err = c.doRequest(&uris[0], "POST", "/internal/translate/keys", defaultProtobufHeaders(), reqData); err != nil {
|
||||
return nil, errors.Wrapf(err, "reading response body of /internal/translate/keys request to %v", uris[0])
|
||||
}
|
||||
}
|
||||
|
||||
idsResp := &pb.TranslateKeysResponse{}
|
||||
err = proto.Unmarshal(respData, idsResp)
|
||||
var result map[string]uint64
|
||||
err = json.Unmarshal(body, &result)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "unmarshalling traslate keys response")
|
||||
return nil, errors.Wrap(err, "unmarshalling response")
|
||||
}
|
||||
return idsResp.IDs, nil
|
||||
|
||||
return result, nil
|
||||
}
|
||||
|
||||
// CreateFieldKeys looks up the IDs associated with specified keys in a field.
|
||||
// If a key does not exist, it will be created.
|
||||
func (c *Client) CreateFieldKeys(field *Field, keys ...string) (map[string]uint64, error) {
|
||||
path := fmt.Sprintf("/internal/translate/field/%s/%s/keys/create", field.index.name, field.name)
|
||||
|
||||
reqData, err := json.Marshal(keys)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "marshalling request")
|
||||
}
|
||||
|
||||
headers := c.augmentHeaders(map[string]string{
|
||||
"Content-Type": "application/json",
|
||||
"Accept": "application/json",
|
||||
})
|
||||
|
||||
status, body, err := c.httpRequest(http.MethodPost, path, reqData, headers, field.options.foreignIndex == "")
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "executing request")
|
||||
}
|
||||
if status != http.StatusOK {
|
||||
return nil, errors.Errorf("find field keys request failed (status: %d): %q", status, string(body))
|
||||
}
|
||||
|
||||
var result map[string]uint64
|
||||
err = json.Unmarshal(body, &result)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "unmarshalling response")
|
||||
}
|
||||
|
||||
return result, nil
|
||||
}
|
||||
|
||||
// FindIndexKeys looks up the IDs associated with specified column keys in an index.
|
||||
// If a key does not exist, the result will not include it.
|
||||
func (c *Client) FindIndexKeys(idx *Index, keys ...string) (map[string]uint64, error) {
|
||||
path := fmt.Sprintf("/internal/translate/index/%s/keys/find", idx.name)
|
||||
|
||||
reqData, err := json.Marshal(keys)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "marshalling request")
|
||||
}
|
||||
|
||||
headers := c.augmentHeaders(map[string]string{
|
||||
"Content-Type": "application/json",
|
||||
"Accept": "application/json",
|
||||
})
|
||||
|
||||
status, body, err := c.HTTPRequest(http.MethodPost, path, reqData, headers)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "executing request")
|
||||
}
|
||||
if status != http.StatusOK {
|
||||
return nil, errors.Errorf("find field keys request failed (status: %d): %q", status, string(body))
|
||||
}
|
||||
|
||||
var result map[string]uint64
|
||||
err = json.Unmarshal(body, &result)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "unmarshalling response")
|
||||
}
|
||||
|
||||
return result, nil
|
||||
}
|
||||
|
||||
// CreateIndexKeys looks up the IDs associated with specified column keys in an index.
|
||||
// If a key does not exist, it will be created.
|
||||
func (c *Client) CreateIndexKeys(idx *Index, keys ...string) (map[string]uint64, error) {
|
||||
path := fmt.Sprintf("/internal/translate/index/%s/keys/create", idx.name)
|
||||
|
||||
reqData, err := json.Marshal(keys)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "marshalling request")
|
||||
}
|
||||
|
||||
headers := c.augmentHeaders(map[string]string{
|
||||
"Content-Type": "application/json",
|
||||
"Accept": "application/json",
|
||||
})
|
||||
|
||||
status, body, err := c.HTTPRequest(http.MethodPost, path, reqData, headers)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "executing request")
|
||||
}
|
||||
if status != http.StatusOK {
|
||||
return nil, errors.Errorf("find field keys request failed (status: %d): %q", status, string(body))
|
||||
}
|
||||
|
||||
var result map[string]uint64
|
||||
err = json.Unmarshal(body, &result)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "unmarshalling response")
|
||||
}
|
||||
|
||||
return result, nil
|
||||
}
|
||||
|
||||
type TransactionResponse struct {
|
||||
|
|
|
|||
|
|
@ -33,6 +33,7 @@ var (
|
|||
testIndexWithKeys *Index
|
||||
testIndexWithKeysNoTrack *Index
|
||||
testIndexAtomicRecord *Index
|
||||
testIndexKeyTranslation *Index
|
||||
|
||||
testField *Field
|
||||
testFieldTimeQuantum *Field
|
||||
|
|
@ -52,6 +53,7 @@ func setup(t *testing.T, r *require.Assertions, cli *Client) {
|
|||
)
|
||||
testField = testIndex.Field("test-field")
|
||||
testFieldTimeQuantum = testIndex.Field("test-field-timequantum", OptFieldTypeTime(TimeQuantumYear))
|
||||
testIndexKeyTranslation = testSchema.Index("test-index-key-translation", OptIndexKeys(true))
|
||||
|
||||
testIndexAtomicRecord = testSchema.Index("test-index-atomic-record")
|
||||
testFieldInt0 = testIndexAtomicRecord.Field("test-field-int0", OptFieldTypeInt(-1000, 1000))
|
||||
|
|
@ -693,24 +695,39 @@ func TestClientAgainstCluster(t *testing.T) {
|
|||
setup(t, require, cli)
|
||||
defer tearDown(t, require, cli)
|
||||
|
||||
testFieldTranslate := testIndex.Field("test-field-translate-rowkeys", OptFieldKeys(true))
|
||||
testFieldTranslate := testIndexKeyTranslation.Field("test-field-translate", OptFieldKeys(true))
|
||||
err := cli.EnsureField(testFieldTranslate)
|
||||
require.NoError(err)
|
||||
|
||||
_, err = cli.Query(testIndex.BatchQuery(
|
||||
testFieldTranslate.Set("key1", 10),
|
||||
testFieldTranslate.Set("key2", 1000),
|
||||
))
|
||||
require.NoErrorf(err, "Set(key1, 10) Set(key2, 1000)")
|
||||
trans, err := cli.CreateFieldKeys(testFieldTranslate, "key1", "key2")
|
||||
require.NoErrorf(err, "CreateFieldKeys")
|
||||
|
||||
rowIDs, err := cli.TranslateRowKeys(testFieldTranslate, []string{"key1", "key2"})
|
||||
require.NoErrorf(err, "TranslateRowKeys")
|
||||
target := map[string]uint64{"key1": 1, "key2": 2}
|
||||
require.Equalf(target, trans, "CreateFieldKeys")
|
||||
|
||||
target := []uint64{1, 2}
|
||||
require.Equalf(target, rowIDs, "TranslateRowKeys")
|
||||
trans, err = cli.FindFieldKeys(testFieldTranslate, "key1", "key2", "key3")
|
||||
require.NoErrorf(err, "FindFieldKeys")
|
||||
|
||||
require.Equalf(target, trans, "FindFieldKeys")
|
||||
})
|
||||
|
||||
t.Run("TranslateColKeys", func(t *testing.T) {
|
||||
setup(t, require, cli)
|
||||
defer tearDown(t, require, cli)
|
||||
|
||||
trans, err := cli.CreateIndexKeys(testIndexKeyTranslation, "key1", "key2")
|
||||
require.NoErrorf(err, "CreateIndexKeys")
|
||||
|
||||
target := map[string]uint64{"key1": 65011713, "key2": 63963137}
|
||||
require.Equalf(target, trans, "CreateIndexKeys")
|
||||
|
||||
trans, err = cli.FindIndexKeys(testIndexKeyTranslation, "key1", "key2", "key3")
|
||||
require.NoErrorf(err, "FindIndexKeys")
|
||||
|
||||
require.Equalf(target, trans, "FindIndexKeys")
|
||||
})
|
||||
|
||||
t.Run("Transactions", func(t *testing.T) {
|
||||
trns, err := cli.StartTransaction("blah", time.Minute, false, time.Minute)
|
||||
require.NoErrorf(err, "StartTransaction(blah)")
|
||||
require.Equalf("blah", trns.ID, "TranslateColumnKeys ID")
|
||||
|
|
|
|||
128
cluster.go
128
cluster.go
|
|
@ -1313,28 +1313,27 @@ func (c *cluster) unprotectedPrimaryReplicaNode() *topology.Node {
|
|||
return cNodes[pos-1]
|
||||
}
|
||||
|
||||
// translateFieldKeys is basically a wrapper around
|
||||
// field.TranslateStore().TranslateKey(key), but in
|
||||
// the case where the local node is not primary, then this method will forward the translation
|
||||
// request to the primary.
|
||||
func (c *cluster) translateFieldKeys(ctx context.Context, field *Field, keys []string, writable bool) (ids []uint64, err error) {
|
||||
// Create a snapshot of the cluster to use for node/partition calculations.
|
||||
snap := topology.NewClusterSnapshot(c.noder, c.Hasher, c.ReplicaN)
|
||||
|
||||
primary := snap.PrimaryFieldTranslationNode()
|
||||
if primary == nil {
|
||||
return nil, errors.Errorf("translating field(%s/%s) keys(%v) - cannot find primary node", field.Index(), field.Name(), keys)
|
||||
}
|
||||
|
||||
if c.Node.ID == primary.ID {
|
||||
ids, err = field.TranslateStore().TranslateKeys(keys, writable)
|
||||
// TODO: remove this when it is no longer used
|
||||
func (c *cluster) translateFieldKeys(ctx context.Context, field *Field, keys []string, writable bool) ([]uint64, error) {
|
||||
var trans map[string]uint64
|
||||
var err error
|
||||
if writable {
|
||||
trans, err = c.createFieldKeys(ctx, field, keys...)
|
||||
} else {
|
||||
// If it's writable, then forward the request to the primary.
|
||||
ids, err = c.InternalClient.TranslateKeysNode(ctx, &primary.URI, field.Index(), field.Name(), keys, writable)
|
||||
trans, err = c.findFieldKeys(ctx, field, keys...)
|
||||
}
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
return nil, errors.Wrapf(err, "translating field(%s/%s) keys(%v)", field.Index(), field.Name(), keys)
|
||||
ids := make([]uint64, len(keys))
|
||||
for i, key := range keys {
|
||||
id, ok := trans[key]
|
||||
if !ok {
|
||||
return nil, ErrTranslatingKeyNotFound
|
||||
}
|
||||
|
||||
ids[i] = id
|
||||
}
|
||||
|
||||
return ids, nil
|
||||
|
|
@ -1527,6 +1526,7 @@ func (c *cluster) translateFieldListIDs(field *Field, ids []uint64) (keys []stri
|
|||
return keys, err
|
||||
}
|
||||
|
||||
// TODO: remove this when it is no longer used
|
||||
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 {
|
||||
|
|
@ -1535,90 +1535,52 @@ func (c *cluster) translateIndexKey(ctx context.Context, indexName string, key s
|
|||
return keyMap[key], nil
|
||||
}
|
||||
|
||||
// TODO: remove this when it is no longer used
|
||||
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{}{}
|
||||
var trans map[string]uint64
|
||||
var err error
|
||||
if writable {
|
||||
trans, err = c.createIndexKeys(ctx, indexName, keys...)
|
||||
} else {
|
||||
trans, err = c.findIndexKeys(ctx, indexName, keys...)
|
||||
}
|
||||
|
||||
keyMap, err := c.translateIndexKeySet(ctx, indexName, keySet, writable)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// make sure that ids line up with keys, but
|
||||
// not appending, but assigning directly 1:1 into the slice.
|
||||
ids := make([]uint64, len(keys))
|
||||
for i, k := range keys {
|
||||
id, ok := keyMap[k]
|
||||
if !writable {
|
||||
if !ok || id == 0 {
|
||||
c.holder.Logger.Debugf("internal translateIndexKeys error: keyMap had no entry for k='%v', and was not writable", k)
|
||||
return nil, ErrTranslatingKeyNotFound
|
||||
}
|
||||
for i, key := range keys {
|
||||
id, ok := trans[key]
|
||||
if !ok {
|
||||
return nil, ErrTranslatingKeyNotFound
|
||||
}
|
||||
|
||||
ids[i] = id
|
||||
}
|
||||
|
||||
return ids, nil
|
||||
}
|
||||
|
||||
// TODO: remove this when it is no longer used
|
||||
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)
|
||||
if idx == nil {
|
||||
return nil, ErrIndexNotFound
|
||||
}
|
||||
|
||||
// Create a snapshot of the cluster to use for node/partition calculations.
|
||||
snap := topology.NewClusterSnapshot(c.noder, c.Hasher, c.ReplicaN)
|
||||
|
||||
// Split keys by partition.
|
||||
keysByPartition := make(map[int][]string, c.partitionN)
|
||||
keys := make([]string, 0, len(keySet))
|
||||
for key := range keySet {
|
||||
partitionID := snap.KeyToKeyPartition(indexName, key)
|
||||
keysByPartition[partitionID] = append(keysByPartition[partitionID], key)
|
||||
keys = append(keys, key)
|
||||
}
|
||||
|
||||
// Translate keys by partition.
|
||||
var g errgroup.Group
|
||||
var mu sync.Mutex
|
||||
for partitionID := range keysByPartition {
|
||||
partitionID := partitionID
|
||||
keys := keysByPartition[partitionID]
|
||||
|
||||
g.Go(func() (err error) {
|
||||
var ids []uint64
|
||||
|
||||
primary := snap.PrimaryPartitionNode(partitionID)
|
||||
if primary == nil {
|
||||
return errors.Errorf("translating index(%s) keys(%v) on partition(%d) - cannot find primary node", indexName, keys, partitionID)
|
||||
}
|
||||
|
||||
if c.Node.ID == primary.ID {
|
||||
ids, err = idx.TranslateStore(partitionID).TranslateKeys(keys, writable)
|
||||
} else {
|
||||
ids, err = c.InternalClient.TranslateKeysNode(ctx, &primary.URI, indexName, "", keys, writable)
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
return errors.Wrapf(err, "translating index(%s) keys(%v) on partition(%d)", indexName, keys, partitionID)
|
||||
}
|
||||
|
||||
mu.Lock()
|
||||
for i, id := range ids {
|
||||
if id != 0 {
|
||||
keyMap[keys[i]] = id
|
||||
}
|
||||
}
|
||||
mu.Unlock()
|
||||
return nil
|
||||
})
|
||||
if writable {
|
||||
return c.createIndexKeys(ctx, indexName, keys...)
|
||||
}
|
||||
if err := g.Wait(); err != nil {
|
||||
|
||||
trans, err := c.findIndexKeys(ctx, indexName, keys...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return keyMap, nil
|
||||
if len(trans) != len(keys) {
|
||||
return nil, ErrTranslatingKeyNotFound
|
||||
}
|
||||
|
||||
return trans, nil
|
||||
}
|
||||
|
||||
func (c *cluster) findIndexKeys(ctx context.Context, indexName string, keys ...string) (map[string]uint64, error) {
|
||||
|
|
|
|||
53
translate.go
53
translate.go
|
|
@ -66,13 +66,6 @@ type TranslateStore interface { // TODO: refactor this interface; readonly shoul
|
|||
ReadOnly() bool
|
||||
SetReadOnly(v bool)
|
||||
|
||||
// Converts a string key to its autoincrementing integer ID value.
|
||||
//
|
||||
// Translated id must be associated with a shard in the store's partition
|
||||
// unless partition is set to -1.
|
||||
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.
|
||||
|
|
@ -407,29 +400,6 @@ func (s *InMemTranslateStore) SetReadOnly(v bool) {
|
|||
s.readOnly = v
|
||||
}
|
||||
|
||||
// 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, writable bool) (uint64, error) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
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, 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], writable); err != nil {
|
||||
return ids, err
|
||||
}
|
||||
}
|
||||
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.
|
||||
|
|
@ -498,29 +468,6 @@ func (s *InMemTranslateStore) Match(filter func([]byte) bool) ([]uint64, error)
|
|||
return matches, nil
|
||||
}
|
||||
|
||||
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
|
||||
}
|
||||
if s.readOnly {
|
||||
return 0, ErrTranslatingKeyNotFound
|
||||
}
|
||||
if !writable {
|
||||
return 0, ErrTranslatingKeyNotFound
|
||||
}
|
||||
|
||||
// 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)
|
||||
return id, nil
|
||||
}
|
||||
|
||||
// TranslateID converts an integer ID to a string key.
|
||||
// Returns a blank string if ID does not exist.
|
||||
func (s *InMemTranslateStore) TranslateID(id uint64) (string, error) {
|
||||
|
|
|
|||
|
|
@ -35,38 +35,13 @@ import (
|
|||
"golang.org/x/sync/errgroup"
|
||||
)
|
||||
|
||||
func TestInMemTranslateStore_TranslateKey(t *testing.T) {
|
||||
s := pilosa.NewInMemTranslateStore("IDX", "FLD", 0, topology.DefaultPartitionN)
|
||||
|
||||
// Ensure initial key translates to ID 1.
|
||||
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", 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", true); err != nil {
|
||||
t.Fatal(err)
|
||||
} else if got, want := id, uint64(1); got != want {
|
||||
t.Fatalf("TranslateKey()=%d, want %d", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
func TestInMemTranslateStore_TranslateID(t *testing.T) {
|
||||
s := pilosa.NewInMemTranslateStore("IDX", "FLD", 0, topology.DefaultPartitionN)
|
||||
|
||||
// Setup initial keys.
|
||||
if _, err := s.TranslateKey("foo", true); err != nil {
|
||||
if _, err := s.CreateKeys("foo"); err != nil {
|
||||
t.Fatal(err)
|
||||
} else if _, err := s.TranslateKey("bar", true); err != nil {
|
||||
} else if _, err := s.CreateKeys("bar"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
|
|
@ -313,30 +288,30 @@ func TestTranslation_KeyNotFound(t *testing.T) {
|
|||
func TestInMemTranslateStore_ReadKey(t *testing.T) {
|
||||
s := pilosa.NewInMemTranslateStore("IDX", "FLD", 0, topology.DefaultPartitionN)
|
||||
|
||||
id, err := s.TranslateKey("foo", false)
|
||||
if err != pilosa.ErrTranslatingKeyNotFound {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got, want := id, uint64(0); got != want {
|
||||
t.Fatalf("TranslateKey()=%d, want %d", got, want)
|
||||
}
|
||||
|
||||
// Ensure next key autoincrements.
|
||||
if id, err = s.TranslateKey("foo", true); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got, want := id, uint64(1); got != want {
|
||||
t.Fatalf("TranslateKey()=%d, want %d", got, want)
|
||||
}
|
||||
|
||||
id1, err := s.TranslateKey("foo", false)
|
||||
ids, err := s.FindKeys("foo")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got, want := id1, id; got != want || id == 0 {
|
||||
if len(ids) != 0 {
|
||||
t.Errorf("unexpected IDs: %v", ids)
|
||||
}
|
||||
|
||||
// Ensure next key autoincrements.
|
||||
ids, err = s.CreateKeys("foo")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got, want := ids["foo"], uint64(1); got != want {
|
||||
t.Fatalf("TranslateKey()=%d, want %d", got, want)
|
||||
}
|
||||
|
||||
ids, err = s.FindKeys("foo")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got, want := ids["foo"], uint64(1); got != want {
|
||||
t.Fatalf("TranslateKey()=%d, want %d", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
// Test key translation with multiple nodes.
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue