scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-08-28 10:54:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

address review comments

Browse source
This commit is contained in:
Nia Weiss 2020-10-15 17:22:18 -04:00
parent 22d6011d05
commit 8a130c150e
No known key found for this signature in database
GPG key ID: 895E83409BFDA1BB
3 changed files with 47 additions and 14 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
boltdb/translate.go
View file

@ -250,18 +250,22 @@ func (s *TranslateStore) CreateKeys(keys ...string) (map[string]uint64, error) {
written := false
result := make(map[string]uint64, len(keys))
err := s.db.Update(func(tx *bolt.Tx) error {
bkt := tx.Bucket(bucketKeys)
if bkt == nil {
keyBucket := tx.Bucket(bucketKeys)
if keyBucket == nil {
return errors.Errorf(errFmtTranslateBucketNotFound, bucketKeys)
}
idBucket := tx.Bucket(bucketIDs)
if keyBucket == nil {
return errors.Errorf(errFmtTranslateBucketNotFound, bucketIDs)
}
for _, key := range keys {
id, boltKey := findIDByKey(bkt, key)
id, boltKey := findIDByKey(keyBucket, key)
if id == 0 {
// The key does not exist.
id = pilosa.GenerateNextPartitionedID(s.index, maxID(tx), s.partitionID, s.partitionN)
if err := bkt.Put(boltKey, u64tob(id)); err != nil {
if err := keyBucket.Put(boltKey, u64tob(id)); err != nil {
return err
} else if err := tx.Bucket(bucketIDs).Put(u64tob(id), boltKey); err != nil {
} else if err := idBucket.Put(u64tob(id), boltKey); err != nil {
return err
}
written = true

8
boltdb/translate_test.go
View file

@ -295,7 +295,7 @@ func TestTranslateStore_FindKeys(t *testing.T) {
s := MustOpenNewTranslateStore()
defer MustCloseTranslateStore(s)
var naieveMap map[string]uint64
var naiveMap map[string]uint64
if c.data != nil {
// Load in key data.
keys := c.data
@ -308,16 +308,16 @@ func TestTranslateStore_FindKeys(t *testing.T) {
t.Errorf("mapped %d keys to %d ids", len(keys), len(ids))
return
}
naieveMap = make(map[string]uint64, len(keys))
naiveMap = make(map[string]uint64, len(keys))
for i, key := range keys {
naieveMap[key] = ids[i]
naiveMap[key] = ids[i]
}
}
// Compute expected lookup result.
result := map[string]uint64{}
for _, key := range c.lookup {
id, ok := naieveMap[key]
id, ok := naiveMap[key]
if !ok {
// The key is expected to be missing.
continue

39
cluster.go
View file

@ -2559,6 +2559,8 @@ func (c *cluster) findFieldKeys(ctx context.Context, field *Field, keys ...strin
missing = append(missing, k)
}
}
} else if len(localTranslations) > len(keys) {
panic(fmt.Sprintf("more translations than keys! translation count=%v, key count=%v", len(localTranslations), len(keys)))
}
if len(missing) == 0 {
// All keys were available locally.
@ -2577,7 +2579,7 @@ func (c *cluster) findFieldKeys(ctx context.Context, field *Field, keys ...strin
// Forward the missing keys to the coordinator.
// The coordinator has the authoritative copy.
remoteTranslations, err := c.InternalClient.FindFieldKeysNode(ctx, &coordinator.URI, field.Index(), field.Name(), keys...)
remoteTranslations, err := c.InternalClient.FindFieldKeysNode(ctx, &coordinator.URI, field.Index(), field.Name(), missing...)
if err != nil {
return nil, errors.Wrapf(err, "translating field(%s/%s) keys(%v) remotely", field.Index(), field.Name(), keys)
}
@ -2621,6 +2623,8 @@ func (c *cluster) createFieldKeys(ctx context.Context, field *Field, keys ...str
missing = append(missing, k)
}
}
} else if len(localTranslations) > len(keys) {
panic(fmt.Sprintf("more translations than keys! translation count=%v, key count=%v", len(localTranslations), len(keys)))
}
if len(missing) == 0 {
// All keys exist locally.
@ -2629,7 +2633,7 @@ func (c *cluster) createFieldKeys(ctx context.Context, field *Field, keys ...str
}
// Forward the missing keys to the coordinator to be created.
remoteTranslations, err := c.InternalClient.CreateFieldKeysNode(ctx, &coordinator.URI, field.Index(), field.Name(), keys...)
remoteTranslations, err := c.InternalClient.CreateFieldKeysNode(ctx, &coordinator.URI, field.Index(), field.Name(), missing...)
if err != nil {
return nil, errors.Wrapf(err, "translating field(%s/%s) keys(%v) remotely", field.Index(), field.Name(), keys)
}
@ -2796,6 +2800,8 @@ func (c *cluster) translateIndexKeySet(ctx context.Context, indexName string, ke
}
func (c *cluster) findIndexKeys(ctx context.Context, indexName string, keys ...string) (map[string]uint64, error) {
done := ctx.Done()
idx := c.holder.Index(indexName)
if idx == nil {
return nil, ErrIndexNotFound
@ -2811,7 +2817,6 @@ func (c *cluster) findIndexKeys(ctx context.Context, indexName string, keys ...s
// TODO: use local replicas to short-circuit network traffic
// Group keys by node.
// Delete remote keys from the by-partition map so that it can be used for local translation.
keysByNode := make(map[*Node][]string)
for partitionID, keys := range keysByPartition {
// Find the primary node for this partition.
@ -2827,11 +2832,13 @@ func (c *cluster) findIndexKeys(ctx context.Context, indexName string, keys ...s
// Group the partition to be processed remotely.
keysByNode[primary] = append(keysByNode[primary], keys...)
// Delete remote keys from the by-partition map so that it can be used for local translation.
delete(keysByPartition, partitionID)
}
// Start translating keys remotely.
// On child calls, there are no remote results.
// On child calls, there are no remote results since we were only sent the keys that we own.
remoteResults := make(chan map[string]uint64, len(keysByNode))
var g errgroup.Group
defer g.Wait()
@ -2852,6 +2859,13 @@ func (c *cluster) findIndexKeys(ctx context.Context, indexName string, keys ...s
// Translate local keys.
translations := make(map[string]uint64)
for partitionID, keys := range keysByPartition {
// Handle cancellation.
select {
case <-done:
return nil, ctx.Err()
default:
}
// Find the keys within the partition.
t, err := idx.TranslateStore(partitionID).FindKeys(keys...)
if err != nil {
@ -2880,6 +2894,14 @@ func (c *cluster) findIndexKeys(ctx context.Context, indexName string, keys ...s
}
func (c *cluster) createIndexKeys(ctx context.Context, indexName string, keys ...string) (map[string]uint64, error) {
// Check for early cancellation.
done := ctx.Done()
select {
case <-done:
return nil, ctx.Err()
default:
}
idx := c.holder.Index(indexName)
if idx == nil {
return nil, ErrIndexNotFound
@ -2919,7 +2941,7 @@ func (c *cluster) createIndexKeys(ctx context.Context, indexName string, keys ..
defer g.Wait()
// Start translating keys remotely.
// On child calls, there are no remote results.
// On child calls, there are no remote results since we were only sent the keys that we own.
for node, keys := range keysByNode {
node, keys := node, keys
@ -2942,6 +2964,13 @@ func (c *cluster) createIndexKeys(ctx context.Context, indexName string, keys ..
partitionID, keys := partitionID, keys
g.Go(func() error {
// Handle cancellation.
select {
case <-done:
return ctx.Err()
default:
}
translations, err := idx.TranslateStore(partitionID).CreateKeys(keys...)
if err != nil {
return errors.Wrapf(err, "translating index(%s) keys(%v) on partition(%d)", idx.Name(), keys, partitionID)