scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-15 08:41:02 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #1626 from niaow/fix-keys

Browse source 
[CORE-579] Standardize key translation on the find and create methods
This commit is contained in:
Nia 2021-06-07 16:17:41 -04:00 • committed by GitHub
commit e6644d25d2
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
8 changed files with 644 additions and 962 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

105
boltdb/translate.go
View file

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

214
boltdb/translate_test.go
View file

@ -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
View file

@ -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
View file

@ -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 {

416
client/client_it_test.go
View file

@ -33,6 +33,7 @@ var (
testIndexWithKeys *Index
testIndexWithKeysNoTrack *Index
testIndexAtomicRecord *Index
testIndexKeyTranslation *Index
testField *Field
testFieldTimeQuantum *Field
@ -40,7 +41,7 @@ var (
testFieldInt1 *Field
)
func setup(t *testing.T, r *require.Assertions, cli *Client) {
func setup(t *testing.T, cli *Client) {
t.Helper()
testSchema := NewSchema()
@ -52,25 +53,24 @@ 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))
testFieldInt1 = testIndexAtomicRecord.Field("test-field-int1", OptFieldTypeInt(-1000, 1000))
r.NoErrorf(cli.SyncSchema(testSchema), "SyncSchema")
require.NoErrorf(t, cli.SyncSchema(testSchema), "SyncSchema")
}
func tearDown(t *testing.T, r *require.Assertions, cli *Client) {
func tearDown(t *testing.T, cli *Client) {
t.Helper()
for _, i := range []*Index{testIndex, testIndexWithKeys, testIndexWithKeysNoTrack, testIndexAtomicRecord} {
r.NoErrorf(cli.DeleteIndex(i), "DeleteIndex(%s)", i.name)
for _, i := range []*Index{testIndex, testIndexWithKeys, testIndexWithKeysNoTrack, testIndexAtomicRecord, testIndexKeyTranslation} {
require.NoErrorf(t, cli.DeleteIndex(i), "DeleteIndex(%s)", i.name)
}
}
func TestClientAgainstCluster(t *testing.T) {
require := require.New(t)
for size, replicaN := 3, 1; replicaN <= 2; replicaN++ {
testName := fmt.Sprintf("%d.%d", size, replicaN)
t.Run(testName, func(t *testing.T) {
@ -81,7 +81,7 @@ func TestClientAgainstCluster(t *testing.T) {
n.Config.Cluster.ReplicaN = replicaN
}
err := c.Start()
require.NoError(err, "Start cluster "+testName)
require.NoError(t, err, "Start cluster "+testName)
urls := make([]string, len(c.Nodes))
for i, n := range c.Nodes {
@ -91,54 +91,54 @@ func TestClientAgainstCluster(t *testing.T) {
// Create a new client for the cluster
cli, err := newClientFromAddresses(urls, &ClientOptions{})
require.NoErrorf(err, "newClientFromAddresses(%v): %v", urls, err)
require.NoErrorf(t, err, "newClientFromAddresses(%v): %v", urls, err)
defer cli.Close()
t.Run("GetStatus", func(t *testing.T) {
status, err := cli.Status()
require.NoErrorf(err, "GET /status")
require.NoErrorf(t, err, "GET /status")
require.Equalf(disco.ClusterStateNormal, disco.ClusterState(status.State), "GET /status")
require.Equalf(t, disco.ClusterStateNormal, disco.ClusterState(status.State), "GET /status")
})
t.Run("QueryRow", func(t *testing.T) {
setup(t, require, cli)
defer tearDown(t, require, cli)
setup(t, cli)
defer tearDown(t, cli)
resp, err := cli.Query(testField.Row(1))
require.NoErrorf(err, "Query Row")
require.NotNil(resp, "Response should not be nil")
require.NoErrorf(t, err, "Query Row")
require.NotNil(t, resp, "Response should not be nil")
})
t.Run("QueryWithShards", func(t *testing.T) {
setup(t, require, cli)
defer tearDown(t, require, cli)
setup(t, cli)
defer tearDown(t, cli)
shardWidth := uint64(1 << shardwidth.Exponent)
_, err := cli.Query(testField.Set(1, 1))
require.NoErrorf(err, "Set(1, %d)", 1)
require.NoErrorf(t, err, "Set(1, %d)", 1)
_, err = cli.Query(testField.Set(1, shardWidth))
require.NoErrorf(err, "Set(1, %d)", shardWidth)
require.NoErrorf(t, err, "Set(1, %d)", shardWidth)
_, err = cli.Query(testField.Set(1, shardWidth*3))
require.NoErrorf(err, "Set(1, %d)", shardWidth*3)
require.NoErrorf(t, err, "Set(1, %d)", shardWidth*3)
resp, err := cli.Query(testField.Row(1), OptQueryShards(0, 3))
require.NoErrorf(err, "Row(1) OptQueryShards(0, 3)")
require.NoErrorf(t, err, "Row(1) OptQueryShards(0, 3)")
cols := resp.Result().Row().Columns
require.Equalf([]uint64{1, shardWidth * 3}, cols, "Unexpected results: %#v", cols)
require.Equalf(t, []uint64{1, shardWidth * 3}, cols, "Unexpected results: %#v", cols)
})
t.Run("OrmCount", func(t *testing.T) {
setup(t, require, cli)
defer tearDown(t, require, cli)
setup(t, cli)
defer tearDown(t, cli)
testFieldCount := testIndex.Field("test-field-count")
err := cli.EnsureField(testFieldCount)
require.NoError(err)
require.NoError(t, err)
qry := testIndex.BatchQuery(
testFieldCount.Set(10, 20),
@ -146,36 +146,36 @@ func TestClientAgainstCluster(t *testing.T) {
testFieldCount.Set(15, 25),
)
_, err = cli.Query(qry)
require.NoErrorf(err, "BatchQuery")
require.NoErrorf(t, err, "BatchQuery")
resp, err := cli.Query(testIndex.Count(testFieldCount.Row(10)))
require.NoErrorf(err, "Count")
require.Equalf(int64(2), resp.Result().Count(), "Count")
require.NoErrorf(t, err, "Count")
require.Equalf(t, int64(2), resp.Result().Count(), "Count")
})
t.Run("DecimalField", func(t *testing.T) {
setup(t, require, cli)
defer tearDown(t, require, cli)
setup(t, cli)
defer tearDown(t, cli)
testFieldDec := testIndex.Field("test-field-dec", OptFieldTypeDecimal(3))
err := cli.EnsureField(testFieldDec)
require.NoError(err)
require.NoError(t, err)
sch, err := cli.Schema()
require.NoErrorf(err, "Schema")
require.NoErrorf(t, err, "Schema")
idx := sch.indexes[testIndex.name]
opts := idx.Field(testFieldDec.name).Options()
require.Equalf(int64(3), opts.scale, "%s scale", testFieldDec.name)
require.Equalf(t, int64(3), opts.scale, "%s scale", testFieldDec.name)
})
t.Run("IntersectReturns", func(t *testing.T) {
setup(t, require, cli)
defer tearDown(t, require, cli)
setup(t, cli)
defer tearDown(t, cli)
testFieldSegments := testIndex.Field("test-field-segments")
err := cli.EnsureField(testFieldSegments)
require.NoError(err)
require.NoError(t, err)
qry1 := testIndex.BatchQuery(
testFieldSegments.Set(2, 10),
@ -184,23 +184,23 @@ func TestClientAgainstCluster(t *testing.T) {
testFieldSegments.Set(3, 20),
)
_, err = cli.Query(qry1)
require.NoErrorf(err, "BatchQuery")
require.NoErrorf(t, err, "BatchQuery")
qry2 := testIndex.Intersect(testFieldSegments.Row(2), testFieldSegments.Row(3))
resp, err := cli.Query(qry2)
require.NoErrorf(err, "Intersect")
require.NoErrorf(t, err, "Intersect")
require.Equalf(1, len(resp.Results()), "Intersect number of results")
require.Equalf([]uint64{10}, resp.Result().Row().Columns, "Intersect columns results")
require.Equalf(t, 1, len(resp.Results()), "Intersect number of results")
require.Equalf(t, []uint64{10}, resp.Result().Row().Columns, "Intersect columns results")
})
t.Run("TopNReturns", func(t *testing.T) {
setup(t, require, cli)
defer tearDown(t, require, cli)
setup(t, cli)
defer tearDown(t, cli)
testFieldTopN := testIndex.Field("test-field-topn")
err := cli.EnsureField(testFieldTopN)
require.NoError(err)
require.NoError(t, err)
qry := testIndex.BatchQuery(
testFieldTopN.Set(10, 5),
@ -210,30 +210,30 @@ func TestClientAgainstCluster(t *testing.T) {
testFieldTopN.Set(30, 5),
)
_, err = cli.Query(qry)
require.NoErrorf(err, "BatchQuery")
require.NoErrorf(t, err, "BatchQuery")
// XXX: The following is required to make this test pass. See: https://github.com/pilosa/pilosa/issues/625
_, _, err = cli.HTTPRequest("POST", "/recalculate-caches", nil, nil)
require.NoErrorf(err, "POST /recalculate-caches")
require.NoErrorf(t, err, "POST /recalculate-caches")
resp, err := cli.Query(testFieldTopN.TopN(2))
require.NoErrorf(err, "TopN(2)")
require.NoErrorf(t, err, "TopN(2)")
items := resp.Result().CountItems()
require.Equalf(2, len(items), "TopN result CountItems")
require.Equalf(t, 2, len(items), "TopN result CountItems")
item := items[0]
require.Equalf(uint64(10), item.ID, "TopN result item[0].ID")
require.Equalf(uint64(3), item.Count, "TopN result item[0].Count")
require.Equalf(t, uint64(10), item.ID, "TopN result item[0].ID")
require.Equalf(t, uint64(3), item.Count, "TopN result item[0].Count")
})
t.Run("MinMaxRow", func(t *testing.T) {
setup(t, require, cli)
defer tearDown(t, require, cli)
setup(t, cli)
defer tearDown(t, cli)
testFieldMinMax := testIndex.Field("test-field-minmax")
err := cli.EnsureField(testFieldMinMax)
require.NoError(err)
require.NoError(t, err)
qry := testIndex.BatchQuery(
testFieldMinMax.Set(10, 5),
@ -243,207 +243,207 @@ func TestClientAgainstCluster(t *testing.T) {
testFieldMinMax.Set(30, 5),
)
_, err = cli.Query(qry)
require.NoErrorf(err, "Setting bits")
require.NoErrorf(t, err, "Setting bits")
resp, err := cli.Query(testFieldMinMax.MinRow())
require.NoErrorf(err, "MinRow")
require.NoErrorf(t, err, "MinRow")
min := resp.Result().CountItem().ID
require.Equalf(uint64(10), min, "Min")
require.Equalf(t, uint64(10), min, "Min")
resp, err = cli.Query(testFieldMinMax.MaxRow())
require.NoErrorf(err, "MaxRow")
require.NoErrorf(t, err, "MaxRow")
max := resp.Result().CountItem().ID
require.Equalf(uint64(30), max, "Max")
require.Equalf(t, uint64(30), max, "Max")
})
t.Run("SetMutexField", func(t *testing.T) {
setup(t, require, cli)
defer tearDown(t, require, cli)
setup(t, cli)
defer tearDown(t, cli)
testFieldMutex := testIndex.Field("test-field-mutex", OptFieldTypeMutex(CacheTypeDefault, 0))
err := cli.EnsureField(testFieldMutex)
require.NoError(err)
require.NoError(t, err)
// can set mutex
_, err = cli.Query(testFieldMutex.Set(1, 100))
require.NoErrorf(err, "Set(1, 100)")
require.NoErrorf(t, err, "Set(1, 100)")
resp, err := cli.Query(testFieldMutex.Row(1))
require.NoErrorf(err, "Row(1)")
require.NoErrorf(t, err, "Row(1)")
target := []uint64{100}
require.Equalf(target, resp.Result().Row().Columns, "Row Result Columns")
require.Equalf(t, target, resp.Result().Row().Columns, "Row Result Columns")
// setting another row removes the previous
_, err = cli.Query(testFieldMutex.Set(42, 100))
require.NoErrorf(err, "Set(42, 100)")
require.NoErrorf(t, err, "Set(42, 100)")
resp, err = cli.Query(testIndex.BatchQuery(
testFieldMutex.Row(1),
testFieldMutex.Row(42),
))
require.NoErrorf(err, "BatchQuery")
require.NoErrorf(t, err, "BatchQuery")
target1 := []uint64(nil)
target42 := []uint64{100}
require.Equalf(target1, resp.Results()[0].Row().Columns, "Row Results[0] Columns")
require.Equalf(target42, resp.Results()[1].Row().Columns, "Row Results[1] Columns")
require.Equalf(t, target1, resp.Results()[0].Row().Columns, "Row Results[0] Columns")
require.Equalf(t, target42, resp.Results()[1].Row().Columns, "Row Results[1] Columns")
})
t.Run("SetBoolField", func(t *testing.T) {
setup(t, require, cli)
defer tearDown(t, require, cli)
setup(t, cli)
defer tearDown(t, cli)
testFieldBool := testIndex.Field("test-field-bool", OptFieldTypeBool())
err := cli.EnsureField(testFieldBool)
require.NoError(err)
require.NoError(t, err)
// can set bool
_, err = cli.Query(testFieldBool.Set(true, 100))
require.NoErrorf(err, "Set(true, 100)")
require.NoErrorf(t, err, "Set(true, 100)")
resp, err := cli.Query(testFieldBool.Row(true))
require.NoErrorf(err, "Row(true)")
require.NoErrorf(t, err, "Row(true)")
target := []uint64{100}
require.Equalf(target, resp.Result().Row().Columns, "Row Result Columns")
require.Equalf(t, target, resp.Result().Row().Columns, "Row Result Columns")
})
t.Run("ClearRowQuery", func(t *testing.T) {
setup(t, require, cli)
defer tearDown(t, require, cli)
setup(t, cli)
defer tearDown(t, cli)
testFieldClear := testIndex.Field("test-field-clear")
err := cli.EnsureField(testFieldClear)
require.NoError(err)
require.NoError(t, err)
_, err = cli.Query(testIndex.BatchQuery(
testFieldClear.Set(1, 100),
testFieldClear.Set(1, 200),
))
require.NoErrorf(err, "Set(1, 100) Set(1, 200)")
require.NoErrorf(t, err, "Set(1, 100) Set(1, 200)")
resp, err := cli.Query(testFieldClear.Row(1))
require.NoErrorf(err, "Row(1)")
require.NoErrorf(t, err, "Row(1)")
target := []uint64{100, 200}
require.Equalf(target, resp.Result().Row().Columns, "Row Result Columns")
require.Equalf(t, target, resp.Result().Row().Columns, "Row Result Columns")
_, err = cli.Query(testFieldClear.ClearRow(1))
require.NoErrorf(err, "ClearRow(1)")
require.NoErrorf(t, err, "ClearRow(1)")
resp, err = cli.Query(testFieldClear.Row(1))
require.NoErrorf(err, "Row(1)")
require.NoErrorf(t, err, "Row(1)")
target = []uint64(nil)
require.Equalf(target, resp.Result().Row().Columns, "Row Result Columns")
require.Equalf(t, target, resp.Result().Row().Columns, "Row Result Columns")
})
t.Run("RowsQuery", func(t *testing.T) {
setup(t, require, cli)
defer tearDown(t, require, cli)
setup(t, cli)
defer tearDown(t, cli)
testFieldRows := testIndex.Field("test-field-rows")
err := cli.EnsureField(testFieldRows)
require.NoError(err)
require.NoError(t, err)
_, err = cli.Query(testIndex.BatchQuery(
testFieldRows.Set(1, 100),
testFieldRows.Set(1, 200),
testFieldRows.Set(2, 200),
))
require.NoErrorf(err, "Set(1, 100) Set(1, 200) Set(2, 200)")
require.NoErrorf(t, err, "Set(1, 100) Set(1, 200) Set(2, 200)")
resp, err := cli.Query(testFieldRows.Rows())
require.NoErrorf(err, "Rows")
require.NoErrorf(t, err, "Rows")
target := RowIdentifiersResult{
IDs: []uint64{1, 2},
}
require.Equalf(target, resp.Result().RowIdentifiers(), "RowIdentifiers Result")
require.Equalf(t, target, resp.Result().RowIdentifiers(), "RowIdentifiers Result")
})
t.Run("UnionRowsQuery", func(t *testing.T) {
setup(t, require, cli)
defer tearDown(t, require, cli)
setup(t, cli)
defer tearDown(t, cli)
testFieldRows := testIndex.Field("test-field-rows")
err := cli.EnsureField(testFieldRows)
require.NoError(err)
require.NoError(t, err)
_, err = cli.Query(testIndex.BatchQuery(
testFieldRows.Set(1, 100),
testFieldRows.Set(1, 200),
testFieldRows.Set(2, 200),
))
require.NoErrorf(err, "Set(1, 100) Set(1, 200) Set(2, 200)")
require.NoErrorf(t, err, "Set(1, 100) Set(1, 200) Set(2, 200)")
resp, err := cli.Query(testFieldRows.Rows().Union())
require.NoErrorf(err, "Rows Union")
require.NoErrorf(t, err, "Rows Union")
target := []uint64{100, 200}
require.Equalf(target, resp.Result().Row().Columns, "Row Result Columns")
require.Equalf(t, target, resp.Result().Row().Columns, "Row Result Columns")
})
t.Run("LikeQuery", func(t *testing.T) {
setup(t, require, cli)
defer tearDown(t, require, cli)
setup(t, cli)
defer tearDown(t, cli)
testFieldLike := testIndex.Field("test-field-like", OptFieldKeys(true))
err := cli.EnsureField(testFieldLike)
require.NoError(err)
require.NoError(t, err)
_, err = cli.Query(testIndex.BatchQuery(
testFieldLike.Set("a", 100),
testFieldLike.Set("b", 200),
testFieldLike.Set("bc", 200),
))
require.NoErrorf(err, "Set(a, 100) Set(b, 200) Set(bc, 200)")
require.NoErrorf(t, err, "Set(a, 100) Set(b, 200) Set(bc, 200)")
resp, err := cli.Query(testFieldLike.Like("b%"))
require.NoErrorf(err, `Like(b%)`)
require.NoErrorf(t, err, `Like(b%)`)
target := RowIdentifiersResult{
Keys: []string{"b", "bc"},
}
require.Equalf(target, resp.Result().RowIdentifiers(), "RowIdentifiers Result")
require.Equalf(t, target, resp.Result().RowIdentifiers(), "RowIdentifiers Result")
})
t.Run("GroupByQuery", func(t *testing.T) {
setup(t, require, cli)
defer tearDown(t, require, cli)
setup(t, cli)
defer tearDown(t, cli)
testFieldGroupBy := testIndex.Field("test-field-group-by")
err := cli.EnsureField(testFieldGroupBy)
require.NoError(err)
require.NoError(t, err)
_, err = cli.Query(testIndex.BatchQuery(
testFieldGroupBy.Set(1, 100),
testFieldGroupBy.Set(1, 200),
testFieldGroupBy.Set(2, 200),
))
require.NoErrorf(err, "Set(1, 100) Set(1, 200) Set(2, 200)")
require.NoErrorf(t, err, "Set(1, 100) Set(1, 200) Set(2, 200)")
resp, err := cli.Query(testIndex.GroupBy(testFieldGroupBy.Rows()))
require.NoErrorf(err, `Like(b%)`)
require.NoErrorf(t, err, `Like(b%)`)
target := []GroupCount{
{Groups: []FieldRow{{FieldName: "test-field-group-by", RowID: 1}}, Count: 2},
{Groups: []FieldRow{{FieldName: "test-field-group-by", RowID: 2}}, Count: 1},
}
assertGroupBy(t, require, target, resp.Result().GroupCounts())
assertGroupBy(t, target, resp.Result().GroupCounts())
})
t.Run("GroupByQuery", func(t *testing.T) {
setup(t, require, cli)
defer tearDown(t, require, cli)
setup(t, cli)
defer tearDown(t, cli)
testFieldGroupBy := testIndex.Field("test-field-group-by-int", OptFieldTypeInt(-10, 10))
err := cli.EnsureField(testFieldGroupBy)
require.NoError(err)
require.NoError(t, err)
_, err = cli.Query(testIndex.RawQuery(`
Set(0, test-field-group-by-int=1)
@ -460,10 +460,10 @@ func TestClientAgainstCluster(t *testing.T) {
Set(10000, test-field-group-by-int=0)
Set(100000, test-field-group-by-int=0)
`))
require.NoError(err, "Set(0..100000)")
require.NoError(t, err, "Set(0..100000)")
resp, err := cli.Query(testIndex.GroupBy(testFieldGroupBy.Rows()))
require.NoErrorf(err, `GroupBy(Rows)`)
require.NoErrorf(t, err, `GroupBy(Rows)`)
var a, b, c, d, e, f int64 = -2, -1, 0, 1, 2, 4
target := []GroupCount{
@ -474,7 +474,7 @@ func TestClientAgainstCluster(t *testing.T) {
{Groups: []FieldRow{{FieldName: "test-field-group-by-int", Value: &e}}, Count: 1},
{Groups: []FieldRow{{FieldName: "test-field-group-by-int", Value: &f}}, Count: 1},
}
assertGroupBy(t, require, target, resp.Result().GroupCounts())
assertGroupBy(t, target, resp.Result().GroupCounts())
})
t.Run("CreateDeleteIndexField", func(t *testing.T) {
@ -482,56 +482,56 @@ func TestClientAgainstCluster(t *testing.T) {
tmpField := tmpIndex.Field("tmp-field")
err := cli.CreateIndex(tmpIndex)
require.NoError(err)
require.NoError(t, err)
err = cli.CreateField(tmpField)
require.NoError(err)
require.NoError(t, err)
err = cli.DeleteField(tmpField)
require.NoError(err)
require.NoError(t, err)
err = cli.DeleteIndex(tmpIndex)
require.NoError(err)
require.NoError(t, err)
})
t.Run("ErrorCreatingIndexField", func(t *testing.T) {
setup(t, require, cli)
defer tearDown(t, require, cli)
setup(t, cli)
defer tearDown(t, cli)
require.ErrorIs(cli.CreateIndex(testIndex), ErrIndexExists)
require.ErrorIs(cli.CreateField(testField), ErrFieldExists)
require.ErrorIs(t, cli.CreateIndex(testIndex), ErrIndexExists)
require.ErrorIs(t, cli.CreateField(testField), ErrFieldExists)
})
t.Run("Failover", func(t *testing.T) {
setup(t, require, cli)
defer tearDown(t, require, cli)
setup(t, cli)
defer tearDown(t, cli)
uri, _ := pnet.NewURIFromAddress("does-not-resolve.foo.bar")
tmpcli, _ := NewClient(NewClusterWithHost(uri, uri, uri, uri), OptClientRetries(0))
_, err := tmpcli.Query(testIndex.All())
require.Error(err, ErrTriedMaxHosts)
require.Error(t, err, ErrTriedMaxHosts)
})
t.Run("InvalidQuery", func(t *testing.T) {
setup(t, require, cli)
defer tearDown(t, require, cli)
setup(t, cli)
defer tearDown(t, cli)
_, _, err := cli.HTTPRequest("INVALID METHOD", "/foo", nil, nil)
require.Error(err)
require.Error(t, err)
_, err = cli.Query(testIndex.RawQuery("Invalid query"))
require.Error(err)
require.Error(t, err)
})
t.Run("Sync", func(t *testing.T) {
testIndexRemote := NewIndex("test-index-remote")
err := cli.EnsureIndex(testIndexRemote)
require.NoError(err)
require.NoError(t, err)
testFieldRemote := testIndexRemote.Field("test-field-remote")
err = cli.EnsureField(testFieldRemote)
require.NoError(err)
require.NoError(t, err)
schema := NewSchema()
idx1 := schema.Index("index-1")
@ -543,54 +543,54 @@ func TestClientAgainstCluster(t *testing.T) {
schema.Index(testIndexRemote.Name())
err = cli.SyncSchema(schema)
require.NoError(err)
require.NoError(t, err)
err = cli.DeleteIndex(testIndexRemote)
require.NoError(err)
require.NoError(t, err)
err = cli.DeleteIndex(idx1)
require.NoError(err)
require.NoError(t, err)
err = cli.DeleteIndex(idx2)
require.NoError(err)
require.NoError(t, err)
})
t.Run("FetchFragmentNodes", func(t *testing.T) {
setup(t, require, cli)
defer tearDown(t, require, cli)
setup(t, cli)
defer tearDown(t, cli)
nodes, err := cli.fetchFragmentNodes(testIndex.Name(), 0)
require.NoErrorf(err, "fetchFragmentNodes(%s, 0)", testIndex.name)
require.Equalf(replicaN, len(nodes), "len(nodes)")
require.NoErrorf(t, err, "fetchFragmentNodes(%s, 0)", testIndex.name)
require.Equalf(t, replicaN, len(nodes), "len(nodes)")
// running the same for coverage
nodes, err = cli.fetchFragmentNodes(testIndex.Name(), 0)
require.NoErrorf(err, "fetchFragmentNodes(%s, 0)", testIndex.name)
require.Equalf(replicaN, len(nodes), "len(nodes)")
require.NoErrorf(t, err, "fetchFragmentNodes(%s, 0)", testIndex.name)
require.Equalf(t, replicaN, len(nodes), "len(nodes)")
})
t.Run("RowRangeQuery", func(t *testing.T) {
setup(t, require, cli)
defer tearDown(t, require, cli)
setup(t, cli)
defer tearDown(t, cli)
testFieldRange := testIndex.Field("test-field-range", OptFieldTypeTime(TimeQuantumMonthDayHour))
err := cli.EnsureField(testFieldRange)
require.NoError(err)
require.NoError(t, err)
_, err = cli.Query(testIndex.BatchQuery(
testFieldRange.SetTimestamp(10, 100, time.Date(2017, time.January, 1, 0, 0, 0, 0, time.UTC)),
testFieldRange.SetTimestamp(10, 100, time.Date(2018, time.January, 1, 0, 0, 0, 0, time.UTC)),
testFieldRange.SetTimestamp(10, 100, time.Date(2019, time.January, 1, 0, 0, 0, 0, time.UTC)),
))
require.NoErrorf(err, "BatchQuery SetTimestamp")
require.NoErrorf(t, err, "BatchQuery SetTimestamp")
start := time.Date(2017, time.January, 5, 0, 0, 0, 0, time.UTC)
end := time.Date(2018, time.January, 5, 0, 0, 0, 0, time.UTC)
resp, err := cli.Query(testFieldRange.RowRange(10, start, end))
require.NoErrorf(err, "RowRange(10, %v, %v)", start, end)
require.NoErrorf(t, err, "RowRange(10, %v, %v)", start, end)
target := []uint64{100}
require.Equalf(target, resp.Result().Row().Columns, "Row Result Columns")
require.Equalf(t, target, resp.Result().Row().Columns, "Row Result Columns")
})
t.Run("StoreQuery", func(t *testing.T) {
@ -599,11 +599,11 @@ func TestClientAgainstCluster(t *testing.T) {
testFieldFrom := testIndexStore.Field("test-field-from")
testFieldTo := testIndexStore.Field("test-field-to")
err := cli.SyncSchema(schema)
require.NoError(err)
require.NoError(t, err)
defer func() {
cerr := cli.DeleteIndex(testIndexStore)
require.NoErrorf(cerr, "failed to delete index: %v", testIndexStore.name)
require.NoErrorf(t, cerr, "failed to delete index: %v", testIndexStore.name)
}()
_, err = cli.Query(testIndexStore.BatchQuery(
@ -611,22 +611,22 @@ func TestClientAgainstCluster(t *testing.T) {
testFieldFrom.Set(10, 200),
testFieldTo.Store(testFieldFrom.Row(10), 1),
))
require.NoErrorf(err, "Set(10, 100) Set(10, 200) Store(Row(10), 1)")
require.NoErrorf(t, err, "Set(10, 100) Set(10, 200) Store(Row(10), 1)")
resp, err := cli.Query(testFieldTo.Row(1))
require.NoErrorf(err, "Row(1)")
require.NoErrorf(t, err, "Row(1)")
target := []uint64{100, 200}
require.Equalf(target, resp.Result().Row().Columns, "Row Result Columns")
require.Equalf(t, target, resp.Result().Row().Columns, "Row Result Columns")
})
t.Run("MultipleClientKeyQuery", func(t *testing.T) {
setup(t, require, cli)
defer tearDown(t, require, cli)
setup(t, cli)
defer tearDown(t, cli)
testFieldMultiClient := testIndexWithKeys.Field("test-field-multiclient")
err := cli.EnsureField(testFieldMultiClient)
require.NoError(err)
require.NoError(t, err)
eg := &errgroup.Group{}
for i := 0; i < 10; i++ {
@ -636,114 +636,132 @@ func TestClientAgainstCluster(t *testing.T) {
return e
})
}
require.NoError(eg.Wait())
require.NoError(t, eg.Wait())
})
t.Run("ExportRowIDColumnID", func(t *testing.T) {
setup(t, require, cli)
defer tearDown(t, require, cli)
setup(t, cli)
defer tearDown(t, cli)
testFieldExport := testIndex.Field("test-field-export")
err := cli.EnsureField(testFieldExport)
require.NoError(err)
require.NoError(t, err)
_, err = cli.Query(testIndex.BatchQuery(
testFieldExport.Set(1, 1),
testFieldExport.Set(1, 10),
testFieldExport.Set(2, 1048577),
), nil)
require.NoErrorf(err, "Set(1, 1) Set(1, 10) Set(2, 1048577)")
require.NoErrorf(t, err, "Set(1, 1) Set(1, 10) Set(2, 1048577)")
r, err := cli.ExportField(testFieldExport)
require.NoErrorf(err, "ExportField")
require.NoErrorf(t, err, "ExportField")
b, err := ioutil.ReadAll(r)
require.NoError(err)
require.NoError(t, err)
target := "1,1\n1,10\n2,1048577\n"
require.Equalf(target, string(b), "Export Field Response")
require.Equalf(t, target, string(b), "Export Field Response")
})
t.Run("ExportRowIDColumnKey", func(t *testing.T) {
setup(t, require, cli)
defer tearDown(t, require, cli)
setup(t, cli)
defer tearDown(t, cli)
testFieldExport := testIndexWithKeys.Field("test-field-export")
err := cli.EnsureField(testFieldExport)
require.NoError(err)
require.NoError(t, err)
_, err = cli.Query(testIndexWithKeys.BatchQuery(
testFieldExport.Set(1, "one"),
testFieldExport.Set(1, "ten"),
testFieldExport.Set(2, "big-number"),
), nil)
require.NoErrorf(err, "Set(1, one) Set(1, ten) Set(2, big-number)")
require.NoErrorf(t, err, "Set(1, one) Set(1, ten) Set(2, big-number)")
r, err := cli.ExportField(testFieldExport)
require.NoErrorf(err, "ExportField")
require.NoErrorf(t, err, "ExportField")
b, err := ioutil.ReadAll(r)
require.NoError(err)
require.NoError(t, err)
target := "1,one\n1,ten\n2,big-number\n"
require.Equalf(target, string(b), "Export Field Response")
require.Equalf(t, target, string(b), "Export Field Response")
})
t.Run("TranslateRowKeys", func(t *testing.T) {
setup(t, require, cli)
defer tearDown(t, require, cli)
setup(t, cli)
defer tearDown(t, 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)
require.NoError(t, 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(t, err, "CreateFieldKeys")
rowIDs, err := cli.TranslateRowKeys(testFieldTranslate, []string{"key1", "key2"})
require.NoErrorf(err, "TranslateRowKeys")
target := map[string]uint64{"key1": 1, "key2": 2}
require.Equalf(t, target, trans, "CreateFieldKeys")
target := []uint64{1, 2}
require.Equalf(target, rowIDs, "TranslateRowKeys")
trans, err = cli.FindFieldKeys(testFieldTranslate, "key1", "key2", "key3")
require.NoErrorf(t, err, "FindFieldKeys")
require.Equalf(t, target, trans, "FindFieldKeys")
})
t.Run("TranslateColKeys", func(t *testing.T) {
setup(t, cli)
defer tearDown(t, cli)
created, err := cli.CreateIndexKeys(testIndexKeyTranslation, "key1", "key2")
require.NoErrorf(t, err, "CreateIndexKeys")
if _, ok := created["key1"]; !ok {
t.Error("key1 missing")
}
if _, ok := created["key2"]; !ok {
t.Error("key2 missing")
}
found, err := cli.FindIndexKeys(testIndexKeyTranslation, "key1", "key2", "key3")
require.NoErrorf(t, err, "FindIndexKeys")
require.Equalf(t, created, found, "IndexKeys")
})
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")
require.Equalf(time.Minute, trns.Timeout, "TranslateColumnKeys Timeout")
require.Truef(trns.Active, "TranslateColumnKeys Active")
require.NoErrorf(t, err, "StartTransaction(blah)")
require.Equalf(t, "blah", trns.ID, "TranslateColumnKeys ID")
require.Equalf(t, time.Minute, trns.Timeout, "TranslateColumnKeys Timeout")
require.Truef(t, trns.Active, "TranslateColumnKeys Active")
trnsMap, err := cli.Transactions()
require.NoErrorf(err, "Transactions")
require.Equalf(1, len(trnsMap), "Transactions len")
require.Truef(trnsMap["blah"].Active, "Transactions Active")
require.NoErrorf(t, err, "Transactions")
require.Equalf(t, 1, len(trnsMap), "Transactions len")
require.Truef(t, trnsMap["blah"].Active, "Transactions Active")
trns, err = cli.GetTransaction("blah")
require.NoErrorf(err, "GetTransaction(blah)")
require.Equalf("blah", trns.ID, "TranslateColumnKeys ID")
require.Equalf(time.Minute, trns.Timeout, "TranslateColumnKeys Timeout")
require.Truef(trns.Active, "TranslateColumnKeys Active")
require.NoErrorf(t, err, "GetTransaction(blah)")
require.Equalf(t, "blah", trns.ID, "TranslateColumnKeys ID")
require.Equalf(t, time.Minute, trns.Timeout, "TranslateColumnKeys Timeout")
require.Truef(t, trns.Active, "TranslateColumnKeys Active")
trns, err = cli.FinishTransaction("blah")
require.NoErrorf(err, "FinishTransaction(blah)")
require.Equalf("blah", trns.ID, "TranslateColumnKeys ID")
require.Equalf(time.Minute, trns.Timeout, "TranslateColumnKeys Timeout")
require.Truef(trns.Active, "TranslateColumnKeys Active")
require.NoErrorf(t, err, "FinishTransaction(blah)")
require.Equalf(t, "blah", trns.ID, "TranslateColumnKeys ID")
require.Equalf(t, time.Minute, trns.Timeout, "TranslateColumnKeys Timeout")
require.Truef(t, trns.Active, "TranslateColumnKeys Active")
})
})
}
}
func assertGroupBy(t *testing.T, r *require.Assertions, expected, results []GroupCount) {
func assertGroupBy(t *testing.T, expected, results []GroupCount) {
t.Helper()
r.Equalf(len(expected), len(results), "number of groupings mismatch")
require.Equalf(t, len(expected), len(results), "number of groupings mismatch")
for i, result := range results {
r.Equalf(expected[i], result, "unexpected result at %d", i)
require.Equalf(t, expected[i], result, "unexpected result at %d", i)
}
}

128
cluster.go
View file

@ -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
View file

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

65
translator_test.go
View file

@ -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.