mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-09-07 00:55:55 +00:00
rebase & fix test const
This commit is contained in:
Ben Johnson
parent
1409ab5664
commit
a189477ba3
3 changed files with 4 additions and 280 deletions
274
cluster.go
274
cluster.go
|
|
@ -2215,280 +2215,6 @@ func (c *cluster) translateIndexIDSet(ctx context.Context, indexName string, idS
|
|||
return idMap, nil
|
||||
}
|
||||
|
||||
/*
|
||||
// updateTranslateStores starts stores for partitions & fields owned by this
|
||||
// node and stops stores for ones this node does not own.
|
||||
func (c *cluster) updateTranslateStores() error {
|
||||
if err := c.closeUnownedTranslateStores(); err != nil {
|
||||
return err
|
||||
} else if err := c.openOwnedTranslateStores(); err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *cluster) closeUnownedTranslateStores() error {
|
||||
for indexName, m := range c.indexTranslateStoreMap {
|
||||
idx := c.holder.Index(indexName)
|
||||
|
||||
// Close unowned partition stores.
|
||||
for partitionID, store := range m {
|
||||
if idx != nil && c.ownsShard(c.Node.ID, indexName, uint64(partitionID)) {
|
||||
continue
|
||||
}
|
||||
if err := store.Close(); err != nil {
|
||||
return err
|
||||
}
|
||||
delete(m, partitionID)
|
||||
}
|
||||
if len(c.indexTranslateStoreMap) == 0 {
|
||||
delete(c.indexTranslateStoreMap, indexName)
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// openOwnedTranslateStores ensures that all owned partition & field stores are open.
|
||||
func (c *cluster) openOwnedTranslateStores() error {
|
||||
// Open partition stores.
|
||||
for _, index := range c.holder.Indexes() {
|
||||
m := c.indexTranslateStoreMap[index.Name()]
|
||||
if m == nil {
|
||||
m = make(map[int]TranslateStore)
|
||||
c.indexTranslateStoreMap[index.Name()] = m
|
||||
}
|
||||
|
||||
for partitionID := 0; partitionID < c.partitionN; partitionID++ {
|
||||
if m[partitionID] != nil {
|
||||
continue
|
||||
}
|
||||
store, err := c.OpenTranslateStore(index.TranslateStorePath(partitionID), index.Name(), "", partitionID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
m[partitionID] = store
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *cluster) indexPartitionTranslateStore(index string, partitionID int) TranslateStore {
|
||||
m := c.indexTranslateStoreMap[index]
|
||||
if m == nil {
|
||||
return nil
|
||||
}
|
||||
return m[partitionID]
|
||||
}
|
||||
|
||||
func (c *cluster) fieldTranslateStore(index, field string) TranslateStore {
|
||||
m := c.fieldTranslateStoreMap[index]
|
||||
if m == nil {
|
||||
return nil
|
||||
}
|
||||
return m[field]
|
||||
}
|
||||
|
||||
|
||||
|
||||
func (c *cluster) translateIndexPartitionKeys(ctx context.Context, index string, partitionID int, keys []string) ([]uint64, error) {
|
||||
s := c.indexPartitionTranslateStore(index, partitionID)
|
||||
if s == nil {
|
||||
return nil, ErrTranslateStoreNotFound
|
||||
}
|
||||
return s.TranslateKeys(keys)
|
||||
}
|
||||
|
||||
func (c *cluster) translateIndexPartitionIDs(ctx context.Context, index string, partitionID int, ids []uint64) ([]string, error) {
|
||||
s := c.indexPartitionTranslateStore(index, partitionID)
|
||||
if s == nil {
|
||||
return nil, ErrTranslateStoreNotFound
|
||||
}
|
||||
return s.TranslateIDs(ids)
|
||||
}
|
||||
|
||||
func (c *cluster) translateIndexPartitionID(ctx context.Context, index string, partitionID int, id uint64) (string, error) {
|
||||
s := c.indexPartitionTranslateStore(index, partitionID)
|
||||
if s == nil {
|
||||
return "", ErrTranslateStoreNotFound
|
||||
}
|
||||
return s.TranslateID(id)
|
||||
}
|
||||
|
||||
func (c *cluster) translateFieldKey(index, field string, key string) (uint64, error) {
|
||||
s := c.fieldTranslateStore(index, field)
|
||||
if s == nil {
|
||||
return 0, ErrTranslateStoreNotFound
|
||||
}
|
||||
return s.TranslateKey(key)
|
||||
}
|
||||
|
||||
func (c *cluster) translateFieldKeys(index, field string, keys []string) ([]uint64, error) {
|
||||
s := c.fieldTranslateStore(index, field)
|
||||
if s == nil {
|
||||
return nil, ErrTranslateStoreNotFound
|
||||
}
|
||||
return s.TranslateKeys(keys)
|
||||
}
|
||||
|
||||
func (c *cluster) translateFieldID(index, field string, id uint64) (string, error) {
|
||||
s := c.fieldTranslateStore(index, field)
|
||||
if s == nil {
|
||||
return "", ErrTranslateStoreNotFound
|
||||
}
|
||||
return s.TranslateID(id)
|
||||
}
|
||||
|
||||
func (c *cluster) translateFieldIDs(index, field string, ids []uint64) ([]string, error) {
|
||||
s := c.fieldTranslateStore(index, field)
|
||||
if s == nil {
|
||||
return nil, ErrTranslateStoreNotFound
|
||||
}
|
||||
return s.TranslateIDs(ids)
|
||||
}
|
||||
|
||||
// TranslateOffsetMap returns a map of offsets for all indexes & fields.
|
||||
func (c *cluster) TranslateOffsetMap() (TranslateOffsetMap, error) {
|
||||
m := make(TranslateOffsetMap)
|
||||
|
||||
for index, partitionMap := range c.indexTranslateStoreMap {
|
||||
for partitionID, store := range partitionMap {
|
||||
id, err := store.MaxID()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
m.SetIndexPartitionOffset(index, partitionID, id+1)
|
||||
}
|
||||
}
|
||||
|
||||
for index, fieldMap := range c.fieldTranslateStoreMap {
|
||||
for field, store := range fieldMap {
|
||||
id, err := store.MaxID()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
m.SetFieldOffset(index, field, id+1)
|
||||
}
|
||||
}
|
||||
|
||||
return m, nil
|
||||
}
|
||||
|
||||
func (c *cluster) setTranslateStoreReadOnly(v bool) {
|
||||
for _, partitionMap := range c.indexTranslateStoreMap {
|
||||
for _, store := range partitionMap {
|
||||
store.SetReadOnly(v)
|
||||
}
|
||||
}
|
||||
|
||||
for _, fieldMap := range c.fieldTranslateStoreMap {
|
||||
for _, store := range fieldMap {
|
||||
store.SetReadOnly(v)
|
||||
}
|
||||
}
|
||||
}
|
||||
*/
|
||||
|
||||
/*
|
||||
// holderTranslateStoreReplicator manages the replication of translation store
|
||||
// data from a primary store to the local replica. Continually tries to
|
||||
// reconnect on disconnect.
|
||||
type holderTranslateStoreReplicator struct {
|
||||
ctx context.Context
|
||||
cancel func()
|
||||
wg sync.WaitGroup
|
||||
|
||||
holder *Holder
|
||||
nodeURL string
|
||||
|
||||
logger logger.Logger
|
||||
}
|
||||
|
||||
func newHolderTranslateStoreReplicator(h *Holder, nodeURL string) *holderTranslateStoreReplicator {
|
||||
r := &holderTranslateStoreReplicator{
|
||||
holder: h,
|
||||
nodeURL: nodeURL,
|
||||
logger: logger.NopLogger,
|
||||
}
|
||||
r.ctx, r.cancel = context.WithCancel(context.Background())
|
||||
return r
|
||||
}
|
||||
|
||||
// Open starts the background monitoring goroutine.
|
||||
func (r *holderTranslateStoreReplicator) Open() error {
|
||||
r.wg.Add(1)
|
||||
go func() { defer r.wg.Done(); r.monitor() }()
|
||||
return nil
|
||||
}
|
||||
|
||||
// Close stops the replicator.
|
||||
func (r *holderTranslateStoreReplicator) Close() error {
|
||||
r.cancel()
|
||||
return nil
|
||||
}
|
||||
|
||||
// monitor runs in a background goroutine and continually tries to connect and
|
||||
// stream translate changes from the primary store.
|
||||
func (r *holderTranslateStoreReplicator) monitor() {
|
||||
for {
|
||||
select {
|
||||
case <-r.ctx.Done():
|
||||
return
|
||||
default:
|
||||
if err := r.replicate(); err != nil {
|
||||
r.logger.Printf("cannot replicate: nodeURL=%s err=%s", r.nodeURL, err)
|
||||
}
|
||||
time.Sleep(1 * time.Second)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (r *holderTranslateStoreReplicator) replicate() error {
|
||||
// Determine the offsets of every index & field store.
|
||||
offsets, err := r.holder.TranslateOffsetMap()
|
||||
if err != nil {
|
||||
return err
|
||||
} else if len(offsets) == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Begin streaming from remote primary.
|
||||
rd, err := r.holder.OpenTranslateReader(r.ctx, r.nodeURL, offsets)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer rd.Close()
|
||||
|
||||
for {
|
||||
var entry TranslateEntry
|
||||
if err := rd.ReadEntry(&entry); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Find appropriate store.
|
||||
var store TranslateStore
|
||||
if entry.Field == "" {
|
||||
idx := r.holder.Index(entry.Index)
|
||||
if idx == nil {
|
||||
return ErrIndexNotFound
|
||||
}
|
||||
store = idx.TranslateStore()
|
||||
} else {
|
||||
f := r.holder.Field(entry.Index, entry.Field)
|
||||
if f == nil {
|
||||
return ErrFieldNotFound
|
||||
}
|
||||
store = f.TranslateStore()
|
||||
}
|
||||
|
||||
// Apply replication to store.
|
||||
if err := store.ForceSet(entry.ID, entry.Key); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
*/
|
||||
|
||||
// ClusterStatus describes the status of the cluster including its
|
||||
// state and node topology.
|
||||
|
|
|
|||
|
|
@ -23,8 +23,6 @@ import (
|
|||
"github.com/pilosa/pilosa/v2/boltdb"
|
||||
)
|
||||
|
||||
const PartitionN = 256
|
||||
|
||||
// Holder is a test wrapper for pilosa.Holder.
|
||||
type Holder struct {
|
||||
*pilosa.Holder
|
||||
|
|
@ -37,7 +35,7 @@ func NewHolder() *Holder {
|
|||
panic(err)
|
||||
}
|
||||
|
||||
h := &Holder{Holder: pilosa.NewHolder(PartitionN)}
|
||||
h := &Holder{Holder: pilosa.NewHolder(pilosa.DefaultPartitionN)}
|
||||
h.Path = path
|
||||
h.Holder.NewAttrStore = boltdb.NewAttrStore
|
||||
|
||||
|
|
@ -63,7 +61,7 @@ func (h *Holder) Close() error {
|
|||
// Note that the holder must be Closed first.
|
||||
func (h *Holder) Reopen() error {
|
||||
path, logger := h.Path, h.Holder.Logger
|
||||
h.Holder = pilosa.NewHolder(PartitionN)
|
||||
h.Holder = pilosa.NewHolder(pilosa.DefaultPartitionN)
|
||||
h.Holder.Path = path
|
||||
h.Holder.Logger = logger
|
||||
h.Holder.NewAttrStore = boltdb.NewAttrStore
|
||||
|
|
|
|||
|
|
@ -32,7 +32,7 @@ func newIndex() *Index {
|
|||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
index, err := pilosa.NewIndex(path, "i", PartitionN)
|
||||
index, err := pilosa.NewIndex(path, "i", pilosa.DefaultPartitionN)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
|
|
@ -62,7 +62,7 @@ func (i *Index) Reopen() error {
|
|||
}
|
||||
|
||||
path, name := i.Path(), i.Name()
|
||||
i.Index, err = pilosa.NewIndex(path, name, PartitionN)
|
||||
i.Index, err = pilosa.NewIndex(path, name, pilosa.DefaultPartitionN)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue