From c0ef297b11a425e26714c8c1d875c52a408a500b Mon Sep 17 00:00:00 2001 From: Nia Weiss Date: Tue, 1 Jun 2021 16:25:23 -0400 Subject: [PATCH 1/3] 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. --- boltdb/translate.go | 105 ----------- boltdb/translate_test.go | 214 ++++------------------- client/batch.go | 368 ++++++++++++++++++++++----------------- client/client.go | 257 ++++++++++++--------------- client/client_it_test.go | 37 ++-- cluster.go | 128 +++++--------- translate.go | 53 ------ translator_test.go | 65 +++---- 8 files changed, 454 insertions(+), 773 deletions(-) diff --git a/boltdb/translate.go b/boltdb/translate.go index 68941fe5c..1ab18cd57 100644 --- a/boltdb/translate.go +++ b/boltdb/translate.go @@ -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) { diff --git a/boltdb/translate_test.go b/boltdb/translate_test.go index c016b0e63..41eb1c6d5 100644 --- a/boltdb/translate_test.go +++ b/boltdb/translate_test.go @@ -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"]) } }) } diff --git a/client/batch.go b/client/batch.go index dc4ba4b42..56fb1d0c2 100644 --- a/client/batch.go +++ b/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 { diff --git a/client/client.go b/client/client.go index c8ad44e3b..337535563 100644 --- a/client/client.go +++ b/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 { diff --git a/client/client_it_test.go b/client/client_it_test.go index 7c65c1b14..fb3008cda 100644 --- a/client/client_it_test.go +++ b/client/client_it_test.go @@ -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") diff --git a/cluster.go b/cluster.go index 6a139cad4..9771ca41c 100644 --- a/cluster.go +++ b/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) { diff --git a/translate.go b/translate.go index 5983bd0a1..5d7a20db3 100644 --- a/translate.go +++ b/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) { diff --git a/translator_test.go b/translator_test.go index 179bff536..17c731b20 100644 --- a/translator_test.go +++ b/translator_test.go @@ -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. From 39337fa41b594691576256e265954aba3b86aa4d Mon Sep 17 00:00:00 2001 From: Nia Weiss Date: Mon, 7 Jun 2021 11:47:56 -0400 Subject: [PATCH 2/3] fix shadowing of `require` in client internal tests The `require` tool was being shadowed with the outer test's T, causing the testing package to explode sometimes. --- client/client_it_test.go | 396 +++++++++++++++++++-------------------- 1 file changed, 197 insertions(+), 199 deletions(-) diff --git a/client/client_it_test.go b/client/client_it_test.go index fb3008cda..d0c2b9fac 100644 --- a/client/client_it_test.go +++ b/client/client_it_test.go @@ -41,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() @@ -59,20 +59,18 @@ func setup(t *testing.T, r *require.Assertions, cli *Client) { 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) { @@ -83,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 { @@ -93,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), @@ -148,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), @@ -186,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), @@ -212,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), @@ -245,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) @@ -462,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{ @@ -476,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) { @@ -484,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") @@ -545,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) { @@ -601,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( @@ -613,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++ { @@ -638,129 +636,129 @@ 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 := testIndexKeyTranslation.Field("test-field-translate", OptFieldKeys(true)) err := cli.EnsureField(testFieldTranslate) - require.NoError(err) + require.NoError(t, err) trans, err := cli.CreateFieldKeys(testFieldTranslate, "key1", "key2") - require.NoErrorf(err, "CreateFieldKeys") + require.NoErrorf(t, err, "CreateFieldKeys") target := map[string]uint64{"key1": 1, "key2": 2} - require.Equalf(target, trans, "CreateFieldKeys") + require.Equalf(t, target, trans, "CreateFieldKeys") trans, err = cli.FindFieldKeys(testFieldTranslate, "key1", "key2", "key3") - require.NoErrorf(err, "FindFieldKeys") + require.NoErrorf(t, err, "FindFieldKeys") - require.Equalf(target, trans, "FindFieldKeys") + require.Equalf(t, target, trans, "FindFieldKeys") }) t.Run("TranslateColKeys", func(t *testing.T) { - setup(t, require, cli) - defer tearDown(t, require, cli) + setup(t, cli) + defer tearDown(t, cli) trans, err := cli.CreateIndexKeys(testIndexKeyTranslation, "key1", "key2") - require.NoErrorf(err, "CreateIndexKeys") + require.NoErrorf(t, err, "CreateIndexKeys") target := map[string]uint64{"key1": 65011713, "key2": 63963137} - require.Equalf(target, trans, "CreateIndexKeys") + require.Equalf(t, target, trans, "CreateIndexKeys") trans, err = cli.FindIndexKeys(testIndexKeyTranslation, "key1", "key2", "key3") - require.NoErrorf(err, "FindIndexKeys") + require.NoErrorf(t, err, "FindIndexKeys") - require.Equalf(target, trans, "FindIndexKeys") + require.Equalf(t, 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") - 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) } } From c2f58f7af0767ca1c207f6db7e024533dbc8c2b3 Mon Sep 17 00:00:00 2001 From: Nia Weiss Date: Mon, 7 Jun 2021 12:35:07 -0400 Subject: [PATCH 3/3] fix column key translation test It previously checked hardcoded values. This does not work due to shard width differences. --- client/client_it_test.go | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/client/client_it_test.go b/client/client_it_test.go index d0c2b9fac..e6a88af5b 100644 --- a/client/client_it_test.go +++ b/client/client_it_test.go @@ -713,16 +713,19 @@ func TestClientAgainstCluster(t *testing.T) { setup(t, cli) defer tearDown(t, cli) - trans, err := cli.CreateIndexKeys(testIndexKeyTranslation, "key1", "key2") + 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") + } - target := map[string]uint64{"key1": 65011713, "key2": 63963137} - require.Equalf(t, target, trans, "CreateIndexKeys") - - trans, err = cli.FindIndexKeys(testIndexKeyTranslation, "key1", "key2", "key3") + found, err := cli.FindIndexKeys(testIndexKeyTranslation, "key1", "key2", "key3") require.NoErrorf(t, err, "FindIndexKeys") - require.Equalf(t, target, trans, "FindIndexKeys") + require.Equalf(t, created, found, "IndexKeys") }) t.Run("Transactions", func(t *testing.T) {