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

remove in mem translate store

Browse source 
going to try making a ramdisk in CI to run tests
This commit is contained in:
Matthew Jaffee 2022-12-16 11:46:31 -06:00 • committed by Matthew Jaffee
parent a58299def4
commit 7ab117f5d1
12 changed files with 107 additions and 541 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
.gitlab/.gitlab-ci.yml
View file

@ -249,6 +249,8 @@ run go tests race:
stage: nonblocking # don't let this job block any other jobs because it takes much longer than the other tests.
image: golang:$GOVERSION
extends: .go-cache
before_script:
- if mount | awk '{if ($3 == "/mnt/ramfs") { exit -1}} ENDFILE{exit 0}'; then mount -o size=2G -t tmpfs none /mnt/ramfs; fi
rules:
- if: '$CI_PIPELINE_SOURCE == "push" || $CI_PIPELINE_SOURCE == "schedule" || $CI_PIPELINE_SOURCE == "web"'
retry: 1
@ -256,7 +258,7 @@ run go tests race:
script:
- echo "Running featurebase race tests..."
- PKG_LIST=$(go list ./... | grep -Ev 'internal/clustertests|simulacraData|batch|idk|v3/dax/test/dax' | paste -s -d, -)
- go test -race -v -timeout=10m ${PKG_LIST//,/ }
- RAMDISK=/mnt/ramfs go test -race -v -timeout=10m ${PKG_LIST//,/ }
tags:
- aws
@ -270,10 +272,12 @@ run go tests:
rules:
- if: '$CI_PIPELINE_SOURCE == "push" || $CI_PIPELINE_SOURCE == "schedule" || $CI_PIPELINE_SOURCE == "web"'
retry: 1
before_script:
- if mount | awk '{if ($3 == "/mnt/ramfs") { exit -1}} ENDFILE{exit 0}'; then mount -o size=2G -t tmpfs none /mnt/ramfs; fi
script:
- echo "Running featurebase unit tests..."
- PKG_LIST=$(go list ./... | grep -Ev 'internal/clustertests|simulacraData|batch|idk|v3/dax/test/dax' | paste -s -d, -)
- go test -tags=shardwidth22 -timeout=10m -coverprofile=coverage.out -covermode=atomic -coverpkg=${PKG_LIST} ${PKG_LIST//,/ }
- RAMDISK=/mnt/ramfs go test -tags=shardwidth22 -timeout=10m -coverprofile=coverage.out -covermode=atomic -coverpkg=${PKG_LIST} ${PKG_LIST//,/ }
artifacts:
paths:
- coverage.out

10
Makefile
View file

@ -75,6 +75,16 @@ testvsub:
echo; echo "999 done testing subpkg $$pkg"; \
done
# make a 2GB RAMDisk. Speed up tests by running them with RAMDISK=/mnt/ramfs
ramdisk-linux:
mount -o size=2G -t tmpfs none /mnt/ramfs
# make a 2GB RAMDisk. Speed up tests by running them with RAMDISK=/Volumes/RAMDisk
ramdisk-osx:
diskutil erasevolume HFS+ 'RAMDisk' `hdiutil attach -nobrowse -nomount ram://4194304`
detach-ramdisk-osx:
hdiutil detach /Volumes/RAMDisk
testvsub-race:
@set -e; for pkg in $(GOPACKAGES); do \

3
executor_test.go
View file

@ -34,6 +34,7 @@ import (
"github.com/molecula/featurebase/v3/testhook"
. "github.com/molecula/featurebase/v3/vprint" // nolint:staticcheck
"github.com/pkg/errors"
"github.com/stretchr/testify/assert"
)
var (
@ -5781,7 +5782,7 @@ func TestExecutor_Execute_Rows_Keys(t *testing.T) {
t.Fatalf("got success, expected error similar to: %+v", test.expErr)
}
rows := res.Results[0].(pilosa.RowIdentifiers)
if !reflect.DeepEqual(rows.Keys, test.exp) {
if !assert.ElementsMatch(t, rows.Keys, test.exp) {
t.Fatalf("\ngot: %+v\nexp: %+v", rows.Keys, test.exp)
} else if rows.Rows != nil {
if test.exp == nil {

2
idk/ingest.go
View file

@ -98,7 +98,7 @@ type Main struct {
AllowTimestampOutOfRange bool `help:"Allow ingest to continue when it encounters out of range timestamps in TimestampFields. (default false)"`
SkipBadRows int `help:"If you fail to process the first n rows without processing one successfully, fail."`
UseShardTransactionalEndpoint bool `flag:"use-shard-transactional-endpoint" help:"Use alternate import endpoint. Currently unstable/testing"`
UseShardTransactionalEndpoint bool `flag:"use-shard-transactional-endpoint" help:"Use alternate import endpoint that ingests data for all fields in a shard in a single atomic request. No negative performance impact and better consistency. Recommended."`
MDSAddress string `short:"" help:"MDS address."`
OrganizationID dax.OrganizationID `short:"" help:"auto-assigned organization ID"`

3
internal/clustertests/pause_node_test.go
View file

@ -16,7 +16,6 @@ import (
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/authn"
boltdb "github.com/molecula/featurebase/v3/boltdb"
"github.com/molecula/featurebase/v3/disco"
"github.com/molecula/featurebase/v3/encoding/proto"
"github.com/molecula/featurebase/v3/net"
@ -161,7 +160,7 @@ func openTranslateStores(dirPath, index string) (map[int]pilosa.TranslateStore,
return nil, err
}
// open bolt db
ts, err := boltdb.OpenTranslateStore(filePath, index, "", partition, disco.DefaultPartitionN, false)
ts, err := pilosa.OpenTranslateStore(filePath, index, "", partition, disco.DefaultPartitionN, false)
ts.SetReadOnly(true)
if err != nil {
return nil, err

5
server/handler_test.go
View file

@ -18,7 +18,6 @@ import (
"time"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/boltdb"
"github.com/molecula/featurebase/v3/encoding/proto"
"github.com/molecula/featurebase/v3/pql"
pb "github.com/molecula/featurebase/v3/proto"
@ -1394,7 +1393,7 @@ func TestHandler_Endpoints(t *testing.T) {
func TestCluster_TranslateStore(t *testing.T) {
cluster := test.MustRunUnsharedCluster(t, 1, []server.CommandOption{
server.OptCommandServerOptions(
pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore),
pilosa.OptServerOpenTranslateStore(pilosa.OpenTranslateStore),
),
})
defer cluster.Close() // nolint: errcheck
@ -1406,7 +1405,7 @@ func TestClusterTranslator(t *testing.T) {
cluster := test.MustRunUnsharedCluster(t, 3,
[]server.CommandOption{
server.OptCommandServerOptions(
pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore),
pilosa.OptServerOpenTranslateStore(pilosa.OpenTranslateStore),
)},
)
defer cluster.Close()

3
server/server.go
View file

@ -29,7 +29,6 @@ import (
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/authn"
"github.com/molecula/featurebase/v3/authz"
"github.com/molecula/featurebase/v3/boltdb"
"github.com/molecula/featurebase/v3/dax"
"github.com/molecula/featurebase/v3/dax/computer"
"github.com/molecula/featurebase/v3/dax/storage"
@ -574,7 +573,7 @@ func (m *Command) setupServer() error {
pilosa.OptServerMetricInterval(time.Duration(m.Config.Metric.PollInterval)),
pilosa.OptServerDiagnosticsInterval(diagnosticsInterval),
pilosa.OptServerExecutorPoolSize(m.Config.WorkerPoolSize),
pilosa.OptServerOpenTranslateStore(boltdb.OpenTranslateStore),
pilosa.OptServerOpenTranslateStore(pilosa.OpenTranslateStore),
pilosa.OptServerOpenTranslateReader(pilosa.GetOpenTranslateReaderWithLockerFunc(c, &sync.Mutex{})),
pilosa.OptServerOpenIDAllocator(pilosa.OpenIDAllocator),
pilosa.OptServerLogger(m.logger),

398
translate.go
View file

@ -3,10 +3,9 @@ package pilosa
import (
"context"
"encoding/json"
"fmt"
"io"
"sort"
"os"
"sync"
"github.com/molecula/featurebase/v3/disco"
@ -88,70 +87,6 @@ type TranslateStore interface { // TODO: refactor this interface; readonly shoul
Delete(records *roaring.Bitmap) (Commitor, error)
}
// TranslatorSummary is returned, for example from the boltdb string key translators,
// by calling ComputeTranslatorSummary(). Non-boltdb mocks, etc no-op that method.
type TranslatorSummary struct {
Index string
// ParitionID is filled for column keys
PartitionID int
NodeID string
StorePath string
IsPrimary bool
IsReplica bool
// PrimaryNodeIndex indexes into the cluster []node array to find the primary
PrimaryNodeIndex int
// Field is filled for row keys
Field string
// Checksum has a blake3 crypto hash of all the keys->ID and all the ID->key mappings
Checksum string
// KeyCount has the number of Key->ID mappings
KeyCount int
// IDCount has the number of ID->Key mappings
IDCount int
// false for RowIDs, true for string-Key column IDs.
IsColKey bool
}
func (s *TranslatorSummary) String() string {
return fmt.Sprintf(`
TranslatorSummary{
Index : %v
PartitionID: %v
NodeID : %v
StorePath : %v
IsPrimary : %v
IsReplica : %v
PrimaryNodeIndex: %v
Field : %v
Checksum: %v
KeyCount: %v
IDCount : %v
IsColKey: %v
}
`,
s.Index,
s.PartitionID,
s.NodeID,
s.StorePath,
s.IsPrimary,
s.IsReplica,
s.PrimaryNodeIndex,
s.Field,
s.Checksum,
s.KeyCount,
s.IDCount,
s.IsColKey,
)
}
// OpenTranslateStoreFunc represents a function for instantiating and opening a TranslateStore.
type OpenTranslateStoreFunc func(path, index, field string, partitionID, partitionN int, fsyncEnabled bool) (TranslateStore, error)
@ -328,307 +263,44 @@ func NewIndexTranslateOffsetMap() *IndexTranslateOffsetMap {
}
}
// Ensure type implements interface.
var _ TranslateStore = &InMemTranslateStore{}
// InMemTranslateStore is an in-memory storage engine for mapping keys to int values.
type InMemTranslateStore struct {
mu sync.RWMutex
index string
field string
partitionID int
partitionN int
readOnly bool
keysByID map[uint64]string
idsByKey map[string]uint64
maxID uint64
writeNotify chan struct{}
}
// NewInMemTranslateStore returns a new instance of InMemTranslateStore.
func NewInMemTranslateStore(index, field string, partitionID, partitionN int) *InMemTranslateStore {
return &InMemTranslateStore{
index: index,
field: field,
partitionID: partitionID,
partitionN: partitionN,
keysByID: make(map[uint64]string),
idsByKey: make(map[string]uint64),
writeNotify: make(chan struct{}),
}
}
var _ OpenTranslateStoreFunc = OpenInMemTranslateStore
// OpenInMemTranslateStore returns a new instance of InMemTranslateStore.
// Implements OpenTranslateStoreFunc.
// OpenInMemTranslateStore returns a new instance of a BoltDB based
// TranslateStore which removes all its files when it's closed, and
// tries to operate off a RAM disk if one is configured and set in the
// environment. Implements OpenTranslateStoreFunc.
func OpenInMemTranslateStore(rawurl, index, field string, partitionID, partitionN int, fsyncEnabled bool) (TranslateStore, error) {
return NewInMemTranslateStore(index, field, partitionID, partitionN), nil
bt := NewBoltTranslateStore(index, field, partitionID, partitionN, false)
iname := index
if len(iname) > 10 {
iname = iname[:10]
}
fname := field
if len(fname) > 10 {
fname = fname[:10]
}
tf, err := os.CreateTemp(os.Getenv("RAMDISK"), fmt.Sprintf("bolt-i%s-f%s-%d-%d-", iname, fname, partitionID, partitionN))
if err != nil {
return nil, errors.Wrap(err, "making temp file for boltdb key translation")
}
bt.Path = tf.Name()
err = bt.Open()
if err != nil {
return nil, errors.Wrap(err, "opening in mem boltdb")
}
return &BoltInMemTranslateStore{bt}, err
}
func (s *InMemTranslateStore) Close() error {
type BoltInMemTranslateStore struct {
*BoltTranslateStore
}
func (b *BoltInMemTranslateStore) Close() error {
defer os.RemoveAll(b.BoltTranslateStore.Path)
err := b.BoltTranslateStore.Close()
if err != nil {
return errors.Wrap(err, "closing in mem bolt translate store")
}
return nil
}
// PartitionID returns the partition id the store was initialized with.
func (s *InMemTranslateStore) PartitionID() int {
return s.partitionID
}
// ReadOnly returns true if the store is in read-only mode.
func (s *InMemTranslateStore) ReadOnly() bool {
s.mu.Lock()
defer s.mu.Unlock()
return s.readOnly
}
// SetReadOnly toggles the read-only mode of the store.
func (s *InMemTranslateStore) SetReadOnly(v bool) {
s.mu.Lock()
defer s.mu.Unlock()
s.readOnly = v
}
func (s *InMemTranslateStore) Delete(records *roaring.Bitmap) (Commitor, error) {
s.mu.Lock()
defer s.mu.Unlock()
for _, id := range records.Slice() {
key := s.keysByID[id]
delete(s.keysByID, id)
delete(s.idsByKey, key)
}
return &NopCommitor{}, nil
}
// FindKeys looks up the ID for each key.
// Keys are not created if they do not exist.
// Missing keys are not considered errors, so the length of the result may be less than that of the input.
func (s *InMemTranslateStore) FindKeys(keys ...string) (map[string]uint64, error) {
s.mu.RLock()
defer s.mu.RUnlock()
result := make(map[string]uint64, len(keys))
for _, key := range keys {
id, ok := s.idsByKey[key]
if !ok {
// The key does not exist.
continue
}
result[key] = id
}
return result, nil
}
// CreateKeys maps all keys to IDs, creating the IDs if they do not exist.
// If the translator is read-only, this will return an error.
func (s *InMemTranslateStore) CreateKeys(keys ...string) (map[string]uint64, error) {
s.mu.Lock()
defer s.mu.Unlock()
if s.readOnly {
return nil, ErrTranslateStoreReadOnly
}
result := make(map[string]uint64, len(keys))
for _, key := range keys {
id, ok := s.idsByKey[key]
if !ok {
// The key does not exist.
// Generate a new id and update db.
if s.field == "" {
id = GenerateNextPartitionedID(s.index, s.maxID, s.partitionID, s.partitionN)
} else {
id = s.maxID + 1
}
s.set(id, key)
}
result[key] = id
}
return result, nil
}
func (s *InMemTranslateStore) Match(filter func([]byte) bool) ([]uint64, error) {
s.mu.RLock()
defer s.mu.RUnlock()
var matches []uint64
for key, id := range s.idsByKey {
if filter([]byte(key)) {
matches = append(matches, id)
}
}
sort.Slice(matches, func(i, j int) bool {
return matches[i] < matches[j]
})
return matches, 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) {
s.mu.Lock()
defer s.mu.Unlock()
return s.translateID(id), nil
}
// TranslateIDs converts a list of integer IDs to a list of string keys.
func (s *InMemTranslateStore) TranslateIDs(ids []uint64) ([]string, error) {
s.mu.Lock()
defer s.mu.Unlock()
keys := make([]string, len(ids))
for i := range ids {
keys[i] = s.translateID(ids[i])
}
return keys, nil
}
func (s *InMemTranslateStore) translateID(id uint64) string {
return s.keysByID[id]
}
// ForceSet writes the id/key pair to the db. Used by replication.
func (s *InMemTranslateStore) ForceSet(id uint64, key string) error {
s.mu.Lock()
defer s.mu.Unlock()
s.set(id, key)
return nil
}
// set assigns the id/key pair to the store.
func (s *InMemTranslateStore) set(id uint64, key string) {
s.keysByID[id] = key
s.idsByKey[key] = id
if id > s.maxID {
s.maxID = id
}
s.notifyWrite()
}
// WriteNotify returns a channel that is closed when a new entry is written.
func (s *InMemTranslateStore) WriteNotify() <-chan struct{} {
s.mu.RLock()
ch := s.writeNotify
s.mu.RUnlock()
return ch
}
// notifyWrite sends a write notification under write lock.
func (s *InMemTranslateStore) notifyWrite() {
close(s.writeNotify)
s.writeNotify = make(chan struct{})
}
// EntryReader returns an error. Replication is not supported.
func (s *InMemTranslateStore) EntryReader(ctx context.Context, offset uint64) (TranslateEntryReader, error) {
s.mu.Lock()
defer s.mu.Unlock()
return newInMemTranslateEntryReader(ctx, s, offset), nil
}
// WriteTo implements io.WriterTo. It's not efficient or careful, but we
// don't expect to use InMemTranslateStore much, it's mostly there to
// avoid disk load during testing.
func (s *InMemTranslateStore) WriteTo(w io.Writer) (int64, error) {
bytes, err := json.Marshal(s.keysByID)
if err != nil {
return 0, err
}
n, err := w.Write(bytes)
return int64(n), err
}
// ReadFrom implements io.ReaderFrom. It's not efficient or careful, but we
// don't expect to use InMemTranslateStore much, it's mostly there to
// avoid disk load during testing.
func (s *InMemTranslateStore) ReadFrom(r io.Reader) (count int64, err error) {
s.mu.Lock()
defer s.mu.Unlock()
var bytes []byte
bytes, err = io.ReadAll(r)
count = int64(len(bytes))
if err != nil {
return count, err
}
var keysByID map[uint64]string
err = json.Unmarshal(bytes, &keysByID)
if err != nil {
return count, err
}
s.maxID = 0
s.keysByID = keysByID
s.idsByKey = make(map[string]uint64, len(s.keysByID))
for k, v := range s.keysByID {
s.idsByKey[v] = k
if k > s.maxID {
s.maxID = k
}
}
return count, nil
}
// MaxID returns the highest identifier in the store.
func (s *InMemTranslateStore) MaxID() (uint64, error) {
s.mu.RLock()
defer s.mu.RUnlock()
return s.maxID, nil
}
// inMemEntryReader represents a stream of translation entries for an inmem translation store.
type inMemTranslateEntryReader struct {
ctx context.Context
cancel func()
store *InMemTranslateStore
offset uint64
}
func newInMemTranslateEntryReader(ctx context.Context, store *InMemTranslateStore, offset uint64) *inMemTranslateEntryReader {
r := &inMemTranslateEntryReader{
store: store,
offset: offset,
}
r.ctx, r.cancel = context.WithCancel(ctx)
return r
}
// Close stops the reader.
func (r *inMemTranslateEntryReader) Close() error {
r.cancel()
return nil
}
// ReadEntry reads the next available entry.
func (r *inMemTranslateEntryReader) ReadEntry(entry *TranslateEntry) error {
for {
// Wait until our offset is less than the max id.
notify := r.store.WriteNotify()
if maxID, err := r.store.MaxID(); err != nil {
return err
} else if r.offset > maxID {
select {
case <-r.ctx.Done():
return io.EOF
case <-notify:
continue // restart loop
}
}
// Translate key for offset.
key, err := r.store.TranslateID(r.offset)
if err != nil {
return err
}
// Copy id/key pair to entry argument and increment offset for next read.
entry.Index, entry.Field = r.store.index, r.store.field
entry.ID, entry.Key = r.offset, key
r.offset++
return nil
}
}

83
boltdb/translate.go → translate_boltdb.go
View file

@ -1,5 +1,5 @@
// Copyright 2021 Molecula Corp. All rights reserved.
package boltdb
package pilosa
import (
"bytes"
@ -12,7 +12,6 @@ import (
"sync"
"time"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/roaring"
"github.com/pkg/errors"
bolt "go.etcd.io/bbolt"
@ -25,7 +24,7 @@ var _ = pprof.StartCPUProfile
var (
// ErrTranslateStoreClosed is returned when reading from an TranslateEntryReader
// and the underlying store is closed.
ErrTranslateStoreClosed = errors.New("boltdb: translate store closing")
ErrBoltTranslateStoreClosed = errors.New("boltdb: translate store closing")
// ErrTranslateKeyNotFound is returned when translating key
// and the underlying store returns an empty set
@ -45,8 +44,8 @@ const (
)
// OpenTranslateStore opens and initializes a boltdb translation store.
func OpenTranslateStore(path, index, field string, partitionID, partitionN int, fsyncEnabled bool) (pilosa.TranslateStore, error) {
s := NewTranslateStore(index, field, partitionID, partitionN, fsyncEnabled)
func OpenTranslateStore(path, index, field string, partitionID, partitionN int, fsyncEnabled bool) (TranslateStore, error) {
s := NewBoltTranslateStore(index, field, partitionID, partitionN, fsyncEnabled)
s.Path = path
if err := s.Open(); err != nil {
return nil, err
@ -55,9 +54,9 @@ func OpenTranslateStore(path, index, field string, partitionID, partitionN int,
}
// Ensure type implements interface.
var _ pilosa.TranslateStore = &TranslateStore{}
var _ TranslateStore = &BoltTranslateStore{}
// TranslateStore is an on-disk storage engine for translating string-to-uint64 values.
// BoltTranslateStore is an on-disk storage engine for translating string-to-uint64 values.
// An empty string will be converted into the sentinel byte slice:
//
// var emptyKey = []byte{
@ -67,7 +66,7 @@ var _ pilosa.TranslateStore = &TranslateStore{}
// 0xc2, 0xa0, // NO-BREAK SPACE
// 0x00,
// }
type TranslateStore struct {
type BoltTranslateStore struct {
mu sync.RWMutex
db *bolt.DB
@ -87,9 +86,9 @@ type TranslateStore struct {
Path string
}
// NewTranslateStore returns a new instance of TranslateStore.
func NewTranslateStore(index, field string, partitionID, partitionN int, fsyncEnabled bool) *TranslateStore {
return &TranslateStore{
// NewBoltTranslateStore returns a new instance of TranslateStore.
func NewBoltTranslateStore(index, field string, partitionID, partitionN int, fsyncEnabled bool) *BoltTranslateStore {
return &BoltTranslateStore{
index: index,
field: field,
partitionID: partitionID,
@ -101,7 +100,7 @@ func NewTranslateStore(index, field string, partitionID, partitionN int, fsyncEn
}
// Open opens the translate file.
func (s *TranslateStore) Open() (err error) {
func (s *BoltTranslateStore) Open() (err error) {
// add the path to the problem database if we panic handling it.
defer func() {
@ -113,7 +112,7 @@ func (s *TranslateStore) Open() (err error) {
if err := os.MkdirAll(filepath.Dir(s.Path), 0750); err != nil {
return errors.Wrapf(err, "mkdir %s", filepath.Dir(s.Path))
} else if s.db, err = bolt.Open(s.Path, 0600, &bolt.Options{Timeout: 1 * time.Second, NoSync: !s.fsyncEnabled}); err != nil {
} else if s.db, err = bolt.Open(s.Path, 0600, &bolt.Options{Timeout: 1 * time.Second, NoSync: !s.fsyncEnabled, InitialMmapSize: 0}); err != nil {
return errors.Wrapf(err, "open file: %s", err)
}
@ -136,7 +135,7 @@ func (s *TranslateStore) Open() (err error) {
}
// Close closes the underlying database.
func (s *TranslateStore) Close() (err error) {
func (s *BoltTranslateStore) Close() (err error) {
s.once.Do(func() { close(s.closing) })
if s.db != nil {
@ -148,26 +147,26 @@ func (s *TranslateStore) Close() (err error) {
}
// PartitionID returns the partition id the store was initialized with.
func (s *TranslateStore) PartitionID() int {
func (s *BoltTranslateStore) PartitionID() int {
return s.partitionID
}
// ReadOnly returns true if the store is in read-only mode.
func (s *TranslateStore) ReadOnly() bool {
func (s *BoltTranslateStore) ReadOnly() bool {
s.mu.RLock()
defer s.mu.RUnlock()
return s.readOnly
}
// SetReadOnly toggles whether store is in read-only mode.
func (s *TranslateStore) SetReadOnly(v bool) {
func (s *BoltTranslateStore) SetReadOnly(v bool) {
s.mu.Lock()
defer s.mu.Unlock()
s.readOnly = v
}
// Size returns the number of bytes in the data file.
func (s *TranslateStore) Size() int64 {
func (s *BoltTranslateStore) Size() int64 {
if s.db == nil {
return 0
}
@ -182,7 +181,7 @@ func (s *TranslateStore) Size() int64 {
// FindKeys looks up the ID for each key.
// Keys are not created if they do not exist.
// Missing keys are not considered errors, so the length of the result may be less than that of the input.
func (s *TranslateStore) FindKeys(keys ...string) (map[string]uint64, error) {
func (s *BoltTranslateStore) FindKeys(keys ...string) (map[string]uint64, error) {
result := make(map[string]uint64, len(keys))
err := s.db.View(func(tx *bolt.Tx) error {
bkt := tx.Bucket(bucketKeys)
@ -216,9 +215,9 @@ const translateTransactionSize = 16384
// CreateKeys maps all keys to IDs, creating the IDs if they do not exist.
// If the translator is read-only, this will return an error.
func (s *TranslateStore) CreateKeys(keys ...string) (map[string]uint64, error) {
func (s *BoltTranslateStore) CreateKeys(keys ...string) (map[string]uint64, error) {
if s.ReadOnly() {
return nil, pilosa.ErrTranslateStoreReadOnly
return nil, ErrTranslateStoreReadOnly
}
written := false
@ -254,7 +253,7 @@ func (s *TranslateStore) CreateKeys(keys ...string) (map[string]uint64, error) {
}
// see if we can re-use any IDs first
if id = getter.GetFreeID(); id == 0 {
id = pilosa.GenerateNextPartitionedID(s.index, maxID(tx), s.partitionID, s.partitionN)
id = GenerateNextPartitionedID(s.index, maxID(tx), s.partitionID, s.partitionN)
}
idBytes := idScratch[puts*8 : puts*8+8]
binary.BigEndian.PutUint64(idBytes, id)
@ -286,7 +285,7 @@ func (s *TranslateStore) CreateKeys(keys ...string) (map[string]uint64, error) {
}
// Match finds the IDs of all keys matching a filter.
func (s *TranslateStore) Match(filter func([]byte) bool) ([]uint64, error) {
func (s *BoltTranslateStore) Match(filter func([]byte) bool) ([]uint64, error) {
var matches []uint64
err := s.db.View(func(tx *bolt.Tx) error {
// This uses the id bucket instead of the key bucket so that matches are produced in sorted order.
@ -316,7 +315,7 @@ func (s *TranslateStore) Match(filter func([]byte) bool) ([]uint64, error) {
// 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) {
func (s *BoltTranslateStore) TranslateID(id uint64) (string, error) {
tx, err := s.db.Begin(false)
if err != nil {
return "", err
@ -326,7 +325,7 @@ func (s *TranslateStore) TranslateID(id uint64) (string, error) {
}
// TranslateIDs converts a list of integer IDs to a list of string keys.
func (s *TranslateStore) TranslateIDs(ids []uint64) ([]string, error) {
func (s *BoltTranslateStore) TranslateIDs(ids []uint64) ([]string, error) {
if len(ids) == 0 {
return nil, nil
}
@ -347,7 +346,7 @@ func (s *TranslateStore) TranslateIDs(ids []uint64) ([]string, error) {
}
// ForceSet writes the id/key pair to the store even if read only. Used by replication.
func (s *TranslateStore) ForceSet(id uint64, key string) error {
func (s *BoltTranslateStore) ForceSet(id uint64, key string) error {
if err := s.db.Update(func(tx *bolt.Tx) (err error) {
if err := tx.Bucket(bucketKeys).Put([]byte(key), u64tob(id)); err != nil {
return err
@ -363,13 +362,13 @@ func (s *TranslateStore) ForceSet(id uint64, key string) error {
}
// EntryReader returns a reader that streams the underlying data file.
func (s *TranslateStore) EntryReader(ctx context.Context, offset uint64) (pilosa.TranslateEntryReader, error) {
func (s *BoltTranslateStore) EntryReader(ctx context.Context, offset uint64) (TranslateEntryReader, error) {
ctx, cancel := context.WithCancel(ctx)
return &TranslateEntryReader{ctx: ctx, cancel: cancel, store: s, offset: offset}, nil
return &BoltTranslateEntryReader{ctx: ctx, cancel: cancel, store: s, offset: offset}, nil
}
// WriteNotify returns a channel that is closed when a new entry is written.
func (s *TranslateStore) WriteNotify() <-chan struct{} {
func (s *BoltTranslateStore) WriteNotify() <-chan struct{} {
s.mu.RLock()
ch := s.writeNotify
s.mu.RUnlock()
@ -377,7 +376,7 @@ func (s *TranslateStore) WriteNotify() <-chan struct{} {
}
// notifyWrite sends a write notification under write lock.
func (s *TranslateStore) notifyWrite() {
func (s *BoltTranslateStore) notifyWrite() {
s.mu.Lock()
defer s.mu.Unlock()
close(s.writeNotify)
@ -385,7 +384,7 @@ func (s *TranslateStore) notifyWrite() {
}
// MaxID returns the highest id in the store.
func (s *TranslateStore) MaxID() (max uint64, err error) {
func (s *BoltTranslateStore) MaxID() (max uint64, err error) {
if err := s.db.View(func(tx *bolt.Tx) error {
max = maxID(tx)
return nil
@ -396,7 +395,7 @@ func (s *TranslateStore) MaxID() (max uint64, err error) {
}
// WriteTo writes the contents of the store to the writer.
func (s *TranslateStore) WriteTo(w io.Writer) (int64, error) {
func (s *BoltTranslateStore) WriteTo(w io.Writer) (int64, error) {
tx, err := s.db.Begin(false)
if err != nil {
return 0, err
@ -406,7 +405,7 @@ func (s *TranslateStore) WriteTo(w io.Writer) (int64, error) {
}
// ReadFrom reads the content and overwrites the existing store.
func (s *TranslateStore) ReadFrom(r io.Reader) (n int64, err error) {
func (s *BoltTranslateStore) ReadFrom(r io.Reader) (n int64, err error) {
// Close store.
if err := s.Close(); err != nil {
return 0, errors.Wrap(err, "closing store")
@ -450,27 +449,27 @@ func maxID(tx *bolt.Tx) uint64 {
return 0
}
type TranslateEntryReader struct {
type BoltTranslateEntryReader struct {
ctx context.Context
store *TranslateStore
store *BoltTranslateStore
offset uint64
cancel func()
}
// Close closes the reader.
func (r *TranslateEntryReader) Close() error {
func (r *BoltTranslateEntryReader) Close() error {
r.cancel()
return nil
}
// ReadEntry reads the next entry from the underlying translate store.
func (r *TranslateEntryReader) ReadEntry(entry *pilosa.TranslateEntry) error {
func (r *BoltTranslateEntryReader) ReadEntry(entry *TranslateEntry) error {
// Ensure reader has not been closed before read.
select {
case <-r.ctx.Done():
return r.ctx.Err()
case <-r.store.closing:
return ErrTranslateStoreClosed
return ErrBoltTranslateStoreClosed
default:
}
@ -510,7 +509,7 @@ func (r *TranslateEntryReader) ReadEntry(entry *pilosa.TranslateEntry) error {
case <-r.ctx.Done():
return r.ctx.Err()
case <-r.store.closing:
return ErrTranslateStoreClosed
return ErrBoltTranslateStoreClosed
case <-writeNotify:
}
}
@ -533,7 +532,7 @@ func (w *boltWrapper) Rollback() {
w.tx.Rollback()
}
}
func (s *TranslateStore) FreeIDs() (*roaring.Bitmap, error) {
func (s *BoltTranslateStore) FreeIDs() (*roaring.Bitmap, error) {
result := roaring.NewBitmap()
err := s.db.View(func(tx *bolt.Tx) error {
bkt := tx.Bucket(bucketFree)
@ -549,7 +548,7 @@ func (s *TranslateStore) FreeIDs() (*roaring.Bitmap, error) {
})
return result, err
}
func (s *TranslateStore) MergeFree(tx *bolt.Tx, newIDs *roaring.Bitmap) error {
func (s *BoltTranslateStore) MergeFree(tx *bolt.Tx, newIDs *roaring.Bitmap) error {
bkt := tx.Bucket(bucketFree)
b := bkt.Get(freeKey)
buf := new(bytes.Buffer)
@ -572,7 +571,7 @@ func (s *TranslateStore) MergeFree(tx *bolt.Tx, newIDs *roaring.Bitmap) error {
// Delete removes the lookeup pairs in order to make avialble for reuse but doesn't commit the
// transaction for that is tied to the associated rbf transaction being successful
func (s *TranslateStore) Delete(records *roaring.Bitmap) (pilosa.Commitor, error) {
func (s *BoltTranslateStore) Delete(records *roaring.Bitmap) (Commitor, error) {
tx, err := s.db.Begin(true)
if err != nil {
return nil, err

2
boltdb/translate_internal_test.go → translate_boltdb_internal_test.go
View file

@ -1,4 +1,4 @@
package boltdb
package pilosa
import (
"path/filepath"

15
boltdb/translate_test.go → translate_boltdb_test.go
View file

@ -1,5 +1,5 @@
// Copyright 2021 Molecula Corp. All rights reserved.
package boltdb_test
package pilosa_test
import (
"bytes"
@ -11,7 +11,6 @@ import (
"time"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/boltdb"
"github.com/molecula/featurebase/v3/disco"
"github.com/molecula/featurebase/v3/roaring"
"github.com/molecula/featurebase/v3/testhook"
@ -203,7 +202,7 @@ func TestTranslateStore_MaxID(t *testing.T) {
}
}
func TestTranslateStore_EntryReader(t *testing.T) {
func TestBoltTranslateStore_EntryReader(t *testing.T) {
t.Run("OK", func(t *testing.T) {
s := MustOpenNewTranslateStore(t)
defer MustCloseTranslateStore(s)
@ -361,7 +360,7 @@ func TestTranslateStore_EntryReader(t *testing.T) {
}()
var entry pilosa.TranslateEntry
if err := r.ReadEntry(&entry); err != boltdb.ErrTranslateStoreClosed {
if err := r.ReadEntry(&entry); err != pilosa.ErrBoltTranslateStoreClosed {
t.Fatalf("unexpected error: %#v", err)
}
@ -374,7 +373,7 @@ func TestTranslateStore_EntryReader(t *testing.T) {
}
// MustNewTranslateStore returns a new TranslateStore with a temporary path.
func MustNewTranslateStore(tb testing.TB) *boltdb.TranslateStore {
func MustNewTranslateStore(tb testing.TB) *pilosa.BoltTranslateStore {
f, err := testhook.TempFile(tb, "translate-store")
if err != nil {
panic(err)
@ -382,7 +381,7 @@ func MustNewTranslateStore(tb testing.TB) *boltdb.TranslateStore {
panic(err)
}
s := boltdb.NewTranslateStore("I", "F", 0, disco.DefaultPartitionN, false)
s := pilosa.NewBoltTranslateStore("I", "F", 0, disco.DefaultPartitionN, false)
s.Path = f.Name()
return s
}
@ -502,7 +501,7 @@ func TestTranslateStore_ReadWrite(t *testing.T) {
}
// MustOpenNewTranslateStore returns a new, opened TranslateStore.
func MustOpenNewTranslateStore(tb testing.TB) *boltdb.TranslateStore {
func MustOpenNewTranslateStore(tb testing.TB) *pilosa.BoltTranslateStore {
s := MustNewTranslateStore(tb)
if err := s.Open(); err != nil {
tb.Fatalf("opening s: %v", err)
@ -511,7 +510,7 @@ func MustOpenNewTranslateStore(tb testing.TB) *boltdb.TranslateStore {
}
// MustCloseTranslateStore closes s and removes the underlying data file.
func MustCloseTranslateStore(s *boltdb.TranslateStore) {
func MustCloseTranslateStore(s *pilosa.BoltTranslateStore) {
if err := s.Close(); err != nil {
panic(err)
}

116
translator_test.go
View file

@ -12,37 +12,12 @@ import (
"github.com/google/go-cmp/cmp"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/featurebase/v3/disco"
"github.com/molecula/featurebase/v3/mock"
"github.com/molecula/featurebase/v3/test"
"github.com/pkg/errors"
"golang.org/x/sync/errgroup"
)
func TestInMemTranslateStore_TranslateID(t *testing.T) {
s := pilosa.NewInMemTranslateStore("IDX", "FLD", 0, disco.DefaultPartitionN)
// Setup initial keys.
if _, err := s.CreateKeys("foo"); err != nil {
t.Fatal(err)
} else if _, err := s.CreateKeys("bar"); err != nil {
t.Fatal(err)
}
// Ensure IDs can be translated back to keys.
if key, err := s.TranslateID(1); err != nil {
t.Fatal(err)
} else if got, want := key, "foo"; got != want {
t.Fatalf("TranslateID()=%s, want %s", got, want)
}
if key, err := s.TranslateID(2); err != nil {
t.Fatal(err)
} else if got, want := key, "bar"; got != want {
t.Fatalf("TranslateID()=%s, want %s", got, want)
}
}
func TestMultiTranslateEntryReader(t *testing.T) {
t.Run("None", func(t *testing.T) {
r := pilosa.NewMultiTranslateEntryReader(context.Background(), nil)
@ -244,97 +219,6 @@ func TestTranslation_KeyNotFound(t *testing.T) {
}
}
func TestInMemTranslateStore_ReadKey(t *testing.T) {
s := pilosa.NewInMemTranslateStore("IDX", "FLD", 0, disco.DefaultPartitionN)
ids, err := s.FindKeys("foo")
if err != nil {
t.Fatal(err)
}
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.
func TestTranslation_Primary(t *testing.T) {
// Ensure that field key translations requests sent to
// non-primary nodes are forwarded to the primary.
t.Run("ForwardFieldKey", func(t *testing.T) {
// Start a 2-node cluster.
c := test.MustRunCluster(t, 3)
defer c.Close()
node0 := c.GetPrimary()
node1 := c.GetNonPrimary()
ctx := context.Background()
index := c.Idx()
fld := "f"
// Create an index without keys.
if _, err := node1.API.CreateIndex(ctx, index,
pilosa.IndexOptions{
Keys: false,
}); err != nil {
t.Fatal(err)
}
// Create a field with keys.
if _, err := node1.API.CreateField(ctx, index, fld,
pilosa.OptFieldKeys(),
); err != nil {
t.Fatal(err)
}
keys := []string{"one", "two", "three"}
for i := range keys {
pql := fmt.Sprintf(`Set(%d, %s="%s")`, i+1, fld, keys[i])
// Send a translation request to node1 (non-primary).
_, err := node1.API.Query(ctx,
&pilosa.QueryRequest{Index: index, Query: pql},
)
if err != nil {
t.Fatal(err)
}
}
for i := len(keys) - 1; i >= 0; i-- {
// Read the row and ensure the key was set.
qry := fmt.Sprintf(`Row(%s="%s")`, fld, keys[i])
resp, err := node0.API.Query(ctx,
&pilosa.QueryRequest{Index: index, Query: qry},
)
if err != nil {
t.Fatal(err)
}
row := resp.Results[0].(*pilosa.Row)
val := uint64(i + 1)
if cols := row.Columns(); !reflect.DeepEqual(cols, []uint64{val}) {
t.Fatalf("unexpected columns: %+v", cols)
}
}
})
}
func TestTranslation_TranslateIDsOnCluster(t *testing.T) {
c := test.MustRunCluster(t, 4)
defer c.Close()