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

replace roaring.Bitmap by [][]byte

Browse source
This commit is contained in:
Kuba Podgórski 2021-05-10 18:45:23 +02:00
parent 046b98bdf9
commit 6bdca67882
3 changed files with 38 additions and 43 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

26
disco/disco.go
View file

@ -20,8 +20,6 @@ import (
"io"
"path"
"sync"
"github.com/pilosa/pilosa/v2/roaring"
)
var (
@ -173,8 +171,8 @@ type Resizer interface {
// Sharder is an interface used to maintain the set of availableShards bitmaps
// per field.
type Sharder interface {
Shards(ctx context.Context, index, field string) (*roaring.Bitmap, error)
SetShards(ctx context.Context, index, field string, shards *roaring.Bitmap) error
Shards(ctx context.Context, index, field string) ([][]byte, error)
SetShards(ctx context.Context, index, field string, shards []byte) error
}
// NopDisCo represents a DisCo that doesn't do anything.
@ -266,12 +264,12 @@ var NopSharder Sharder = &nopSharder{}
type nopSharder struct{}
// Shards is a no-op implementation of the Sharder Shards method.
func (n *nopSharder) Shards(ctx context.Context, index, field string) (*roaring.Bitmap, error) {
func (n *nopSharder) Shards(ctx context.Context, index, field string) ([][]byte, error) {
return nil, nil
}
// AddShards is a no-op implementation of the Sharder AddShards method.
func (n *nopSharder) SetShards(ctx context.Context, index, field string, shards *roaring.Bitmap) error {
func (n *nopSharder) SetShards(ctx context.Context, index, field string, shards []byte) error {
return nil
}
@ -471,30 +469,32 @@ func (s *inMemSchemator) DeleteView(ctx context.Context, index, field, view stri
}
var InMemSharder Sharder = &inMemSharder{
shards: make(map[string]*roaring.Bitmap),
shards: make(map[string][]byte),
}
type inMemSharder struct {
mu sync.RWMutex
shards map[string]*roaring.Bitmap
shards map[string][]byte
}
func (s *inMemSharder) Shards(ctx context.Context, index, field string) (*roaring.Bitmap, error) {
func (s *inMemSharder) Shards(ctx context.Context, index, field string) ([][]byte, error) {
key := path.Join("/shard/", index, field)
s.mu.RLock()
defer s.mu.RUnlock()
b := s.shards[key]
if b == nil {
return roaring.NewBitmap(), nil
return nil, nil
}
return b, nil
return [][]byte{b}, nil
}
func (s *inMemSharder) SetShards(ctx context.Context, index, field string, shards *roaring.Bitmap) error {
func (s *inMemSharder) SetShards(ctx context.Context, index, field string, shards []byte) error {
key := path.Join("/shard/", index, field)
s.mu.Lock()
defer s.mu.Unlock()
s.shards[key] = shards
s.shards[key] = make([]byte, len(shards))
copy(s.shards[key], shards)
return nil
}

27
etcd/embed.go
View file

@ -29,7 +29,6 @@ import (
"github.com/pilosa/pilosa/v2/disco"
"github.com/pilosa/pilosa/v2/logger"
"github.com/pilosa/pilosa/v2/roaring"
"github.com/pilosa/pilosa/v2/topology"
"github.com/pkg/errors"
"go.etcd.io/etcd/clientv3"
@ -1035,40 +1034,24 @@ func memberAdd(cli *clientv3.Client, peerURL string) (id uint64, name string) {
}
// Shards implements the Sharder interface.
func (e *Etcd) Shards(ctx context.Context, index, field string) (*roaring.Bitmap, error) {
func (e *Etcd) Shards(ctx context.Context, index, field string) ([][]byte, error) {
key := path.Join(shardPrefix, index, field)
_, vals, err := e.getKeyWithPrefix(ctx, key)
bm := roaring.NewBitmap()
if errors.Cause(err) == disco.ErrKeyDoesNotExist {
e.logger.Warnf("key: %s, err: %v", key, err)
return bm, nil
return nil, nil
}
for _, v := range vals {
b := roaring.NewBitmap()
if err = b.UnmarshalBinary(v); err != nil {
return nil, errors.Wrap(err, "unmarshalling shards")
}
bm.UnionInPlace(b)
}
return bm, nil
return vals, nil
}
// SetShards implements the Sharder interface.
func (e *Etcd) SetShards(ctx context.Context, index, field string, shards *roaring.Bitmap) error {
func (e *Etcd) SetShards(ctx context.Context, index, field string, shards []byte) error {
key := path.Join(shardPrefix, index, field, e.e.Server.ID().String())
// Write shards to etcd.
var buf bytes.Buffer
if _, err := shards.WriteTo(&buf); err != nil {
return errors.Wrap(err, "writing shards to bytes buffer")
}
op := clientv3.OpPut(key, "")
op.WithValueBytes(buf.Bytes())
op.WithValueBytes(shards)
return e.retryClient(func(cli *clientv3.Client) (err error) {
_, err = cli.Txn(ctx).Then(op).Commit()
return

28
field.go
View file

@ -15,6 +15,7 @@
package pilosa
import (
"bytes"
"context"
"encoding/json"
"fmt"
@ -126,7 +127,7 @@ type Field struct {
// Synchronization primitives needed for async writing of
// the remoteAvailableShards
availableShardChan chan *roaring.Bitmap
availableShardChan chan []byte
doneChan chan struct{}
wg sync.WaitGroup
}
@ -478,8 +479,16 @@ func (f *Field) loadAvailableShards() error {
return errors.Wrap(err, "loading available shards")
}
bm := roaring.NewBitmap()
for _, s := range shards {
b := roaring.NewBitmap()
if err = b.UnmarshalBinary(s); err != nil {
return errors.Wrap(err, "available shards corrupt")
}
bm.UnionInPlace(b)
}
// Merge bitmap from file into field.
f.mergeRemoteAvailableShards(shards)
f.mergeRemoteAvailableShards(bm)
return nil
}
@ -492,8 +501,11 @@ func (f *Field) saveAvailableShards() error {
}
func (f *Field) unprotectedSaveAvailableShards() error {
f.remoteAvailableShards.Optimize()
f.availableShardChan <- f.remoteAvailableShards
var buf bytes.Buffer
if _, err := f.remoteAvailableShards.WriteTo(&buf); err != nil {
return errors.Wrap(err, "rendering available shards ")
}
f.availableShardChan <- buf.Bytes()
return nil
}
@ -579,7 +591,7 @@ func (f *Field) Open() error {
}
}
f.availableShardChan = make(chan *roaring.Bitmap)
f.availableShardChan = make(chan []byte)
f.doneChan = make(chan struct{})
f.wg.Add(1)
go f.writeAvailableShards()
@ -594,14 +606,14 @@ func (f *Field) Open() error {
return nil
}
func (f *Field) blockingWriteAvailableShards(availableShards *roaring.Bitmap) {
func (f *Field) blockingWriteAvailableShards(availableShards []byte) {
err := f.holder.sharder.SetShards(context.Background(), f.index, f.name, availableShards)
if err != nil {
f.holder.Logger.Errorf("writting available shards: %v", err)
}
}
func (f *Field) nonBlockingWriteAvailableShards(availableShards *roaring.Bitmap, done chan bool) {
func (f *Field) nonBlockingWriteAvailableShards(availableShards []byte, done chan bool) {
if availableShards == nil {
return
}
@ -614,7 +626,7 @@ func (f *Field) nonBlockingWriteAvailableShards(availableShards *roaring.Bitmap,
func (f *Field) writeAvailableShards() {
defer f.wg.Done()
ticker := time.NewTicker(availableShardFileFlushDuration.Get())
var data *roaring.Bitmap
var data []byte
tracker := make(chan bool)
writing := false