mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-09-07 09:05:55 +00:00
Merge pull request #1446 from jaffee/unexport-holder-stuff
Unexport holder stuff
This commit is contained in:
Matthew Jaffee
GitHub
commit
f2be368222
9 changed files with 220 additions and 180 deletions
10
api.go
10
api.go
|
|
@ -337,7 +337,7 @@ func (api *API) ExportCSV(ctx context.Context, indexName string, fieldName strin
|
|||
}
|
||||
|
||||
// Find the fragment.
|
||||
f := api.holder.Fragment(indexName, fieldName, ViewStandard, shard)
|
||||
f := api.holder.fragment(indexName, fieldName, ViewStandard, shard)
|
||||
if f == nil {
|
||||
return ErrFragmentNotFound
|
||||
}
|
||||
|
|
@ -379,7 +379,7 @@ func (api *API) MarshalFragment(ctx context.Context, indexName string, fieldName
|
|||
}
|
||||
|
||||
// Retrieve fragment from holder.
|
||||
f := api.holder.Fragment(indexName, fieldName, ViewStandard, shard)
|
||||
f := api.holder.fragment(indexName, fieldName, ViewStandard, shard)
|
||||
if f == nil {
|
||||
return nil, ErrFragmentNotFound
|
||||
}
|
||||
|
|
@ -437,7 +437,7 @@ func (api *API) FragmentBlockData(ctx context.Context, body io.Reader) ([]byte,
|
|||
}
|
||||
|
||||
// Retrieve fragment from holder.
|
||||
f := api.holder.Fragment(req.Index, req.Field, ViewStandard, req.Shard)
|
||||
f := api.holder.fragment(req.Index, req.Field, ViewStandard, req.Shard)
|
||||
if f == nil {
|
||||
return nil, ErrFragmentNotFound
|
||||
}
|
||||
|
|
@ -461,7 +461,7 @@ func (api *API) FragmentBlocks(ctx context.Context, indexName string, fieldName
|
|||
}
|
||||
|
||||
// Retrieve fragment from holder.
|
||||
f := api.holder.Fragment(indexName, fieldName, ViewStandard, shard)
|
||||
f := api.holder.fragment(indexName, fieldName, ViewStandard, shard)
|
||||
if f == nil {
|
||||
return nil, ErrFragmentNotFound
|
||||
}
|
||||
|
|
@ -697,7 +697,7 @@ func (api *API) ImportValue(ctx context.Context, req internal.ImportValueRequest
|
|||
|
||||
// MaxShards returns the maximum shard number for each index in a map.
|
||||
func (api *API) MaxShards(ctx context.Context) map[string]uint64 {
|
||||
return api.holder.MaxShards()
|
||||
return api.holder.maxShards()
|
||||
}
|
||||
|
||||
// StatsWithTags returns an instance of whatever implementation of StatsClient
|
||||
|
|
|
|||
|
|
@ -450,7 +450,7 @@ func (c *cluster) setState(state string) {
|
|||
// been removed.
|
||||
// It's safe to do a cleanup after state changes back to normal.
|
||||
if doCleanup {
|
||||
var cleaner HolderCleaner
|
||||
var cleaner holderCleaner
|
||||
cleaner.Node = c.Node
|
||||
cleaner.Holder = c.holder
|
||||
cleaner.Cluster = c
|
||||
|
|
@ -1150,7 +1150,7 @@ func (c *cluster) generateResizeJobByAction(nodeAction nodeAction) (*resizeJob,
|
|||
Node: EncodeNode(toCluster.unprotectedNodeByID(id)),
|
||||
Coordinator: EncodeNode(c.coordinatorNode()),
|
||||
Sources: sources,
|
||||
Schema: c.holder.EncodeSchema(), // Include the schema to ensure it's in sync on the receiving node.
|
||||
Schema: c.holder.encodeSchema(), // Include the schema to ensure it's in sync on the receiving node.
|
||||
ClusterStatus: c.Status(),
|
||||
}
|
||||
j.Instructions = append(j.Instructions, instr)
|
||||
|
|
@ -1205,7 +1205,7 @@ func (c *cluster) followResizeInstruction(instr *internal.ResizeInstruction) err
|
|||
|
||||
// Sync the schema received in the resize instruction.
|
||||
c.logger.Printf("Holder ApplySchema")
|
||||
if err := c.holder.ApplySchema(instr.Schema); err != nil {
|
||||
if err := c.holder.applySchema(instr.Schema); err != nil {
|
||||
return errors.Wrap(err, "applying schema")
|
||||
}
|
||||
|
||||
|
|
|
|||
18
executor.go
18
executor.go
|
|
@ -427,7 +427,7 @@ func (e *executor) executeSumCountShard(ctx context.Context, index string, c *pq
|
|||
return ValCount{}, nil
|
||||
}
|
||||
|
||||
fragment := e.Holder.Fragment(index, fieldName, viewBSIGroupPrefix+fieldName, shard)
|
||||
fragment := e.Holder.fragment(index, fieldName, viewBSIGroupPrefix+fieldName, shard)
|
||||
if fragment == nil {
|
||||
return ValCount{}, nil
|
||||
}
|
||||
|
|
@ -465,7 +465,7 @@ func (e *executor) executeMinShard(ctx context.Context, index string, c *pql.Cal
|
|||
return ValCount{}, nil
|
||||
}
|
||||
|
||||
fragment := e.Holder.Fragment(index, fieldName, viewBSIGroupPrefix+fieldName, shard)
|
||||
fragment := e.Holder.fragment(index, fieldName, viewBSIGroupPrefix+fieldName, shard)
|
||||
if fragment == nil {
|
||||
return ValCount{}, nil
|
||||
}
|
||||
|
|
@ -503,7 +503,7 @@ func (e *executor) executeMaxShard(ctx context.Context, index string, c *pql.Cal
|
|||
return ValCount{}, nil
|
||||
}
|
||||
|
||||
fragment := e.Holder.Fragment(index, fieldName, viewBSIGroupPrefix+fieldName, shard)
|
||||
fragment := e.Holder.fragment(index, fieldName, viewBSIGroupPrefix+fieldName, shard)
|
||||
if fragment == nil {
|
||||
return ValCount{}, nil
|
||||
}
|
||||
|
|
@ -623,7 +623,7 @@ func (e *executor) executeTopNShard(ctx context.Context, index string, c *pql.Ca
|
|||
field = defaultField
|
||||
}
|
||||
|
||||
f := e.Holder.Fragment(index, field, ViewStandard, shard)
|
||||
f := e.Holder.fragment(index, field, ViewStandard, shard)
|
||||
if f == nil {
|
||||
return nil, nil
|
||||
}
|
||||
|
|
@ -693,7 +693,7 @@ func (e *executor) executeBitmapShard(ctx context.Context, index string, c *pql.
|
|||
return nil, fmt.Errorf("Row() must specify %v", rowLabel)
|
||||
}
|
||||
|
||||
frag := e.Holder.Fragment(index, fieldName, ViewStandard, shard)
|
||||
frag := e.Holder.fragment(index, fieldName, ViewStandard, shard)
|
||||
if frag == nil {
|
||||
return NewRow(), nil
|
||||
}
|
||||
|
|
@ -785,7 +785,7 @@ func (e *executor) executeRangeShard(ctx context.Context, index string, c *pql.C
|
|||
// Union bitmaps across all time-based views.
|
||||
row := &Row{}
|
||||
for _, view := range viewsByTimeRange(ViewStandard, startTime, endTime, q) {
|
||||
f := e.Holder.Fragment(index, fieldName, view, shard)
|
||||
f := e.Holder.fragment(index, fieldName, view, shard)
|
||||
if f == nil {
|
||||
continue
|
||||
}
|
||||
|
|
@ -836,7 +836,7 @@ func (e *executor) executeBSIGroupRangeShard(ctx context.Context, index string,
|
|||
}
|
||||
|
||||
// Retrieve fragment.
|
||||
frag := e.Holder.Fragment(index, fieldName, viewBSIGroupPrefix+fieldName, shard)
|
||||
frag := e.Holder.fragment(index, fieldName, viewBSIGroupPrefix+fieldName, shard)
|
||||
if frag == nil {
|
||||
return NewRow(), nil
|
||||
}
|
||||
|
|
@ -871,7 +871,7 @@ func (e *executor) executeBSIGroupRangeShard(ctx context.Context, index string,
|
|||
}
|
||||
|
||||
// Retrieve fragment.
|
||||
frag := e.Holder.Fragment(index, fieldName, viewBSIGroupPrefix+fieldName, shard)
|
||||
frag := e.Holder.fragment(index, fieldName, viewBSIGroupPrefix+fieldName, shard)
|
||||
if frag == nil {
|
||||
return NewRow(), nil
|
||||
}
|
||||
|
|
@ -904,7 +904,7 @@ func (e *executor) executeBSIGroupRangeShard(ctx context.Context, index string,
|
|||
}
|
||||
|
||||
// Retrieve fragment.
|
||||
frag := e.Holder.Fragment(index, fieldName, viewBSIGroupPrefix+fieldName, shard)
|
||||
frag := e.Holder.fragment(index, fieldName, viewBSIGroupPrefix+fieldName, shard)
|
||||
if frag == nil {
|
||||
return NewRow(), nil
|
||||
}
|
||||
|
|
|
|||
78
holder.go
78
holder.go
|
|
@ -36,8 +36,8 @@ const (
|
|||
// defaultCacheFlushInterval is the default value for Fragment.CacheFlushInterval.
|
||||
defaultCacheFlushInterval = 1 * time.Minute
|
||||
|
||||
// FileLimit is the maximum open file limit (ulimit -n) to automatically set.
|
||||
FileLimit = 262144 // (512^2)
|
||||
// fileLimit is the maximum open file limit (ulimit -n) to automatically set.
|
||||
fileLimit = 262144 // (512^2)
|
||||
)
|
||||
|
||||
// Holder represents a container for indexes.
|
||||
|
|
@ -50,7 +50,7 @@ type Holder struct {
|
|||
// opened channel is closed once Open() completes.
|
||||
opened chan struct{}
|
||||
|
||||
Broadcaster broadcaster
|
||||
broadcaster broadcaster
|
||||
|
||||
NewAttrStore func(string) AttrStore
|
||||
|
||||
|
|
@ -65,7 +65,7 @@ type Holder struct {
|
|||
Path string
|
||||
|
||||
// The interval at which the cached row ids are persisted to disk.
|
||||
CacheFlushInterval time.Duration
|
||||
cacheFlushInterval time.Duration
|
||||
|
||||
Logger Logger
|
||||
}
|
||||
|
|
@ -78,12 +78,12 @@ func NewHolder() *Holder {
|
|||
|
||||
opened: make(chan struct{}),
|
||||
|
||||
Broadcaster: NopBroadcaster,
|
||||
broadcaster: NopBroadcaster,
|
||||
Stats: NopStatsClient,
|
||||
|
||||
NewAttrStore: newNopAttrStore,
|
||||
|
||||
CacheFlushInterval: defaultCacheFlushInterval,
|
||||
cacheFlushInterval: defaultCacheFlushInterval,
|
||||
|
||||
Logger: NopLogger,
|
||||
}
|
||||
|
|
@ -200,8 +200,8 @@ func (h *Holder) HasData() (bool, error) {
|
|||
return false, nil
|
||||
}
|
||||
|
||||
// MaxShards returns MaxShard map for all indexes.
|
||||
func (h *Holder) MaxShards() map[string]uint64 {
|
||||
// maxShards returns MaxShard map for all indexes.
|
||||
func (h *Holder) maxShards() map[string]uint64 {
|
||||
a := make(map[string]uint64)
|
||||
for _, index := range h.Indexes() {
|
||||
a[index.Name()] = index.MaxShard()
|
||||
|
|
@ -229,8 +229,8 @@ func (h *Holder) Schema() []*IndexInfo {
|
|||
return a
|
||||
}
|
||||
|
||||
// ApplySchema applies an internal Schema to Holder.
|
||||
func (h *Holder) ApplySchema(schema *internal.Schema) error {
|
||||
// applySchema applies an internal Schema to Holder.
|
||||
func (h *Holder) applySchema(schema *internal.Schema) error {
|
||||
// Create indexes that don't exist.
|
||||
for _, index := range schema.Indexes {
|
||||
opt := IndexOptions{}
|
||||
|
|
@ -257,15 +257,15 @@ func (h *Holder) ApplySchema(schema *internal.Schema) error {
|
|||
return nil
|
||||
}
|
||||
|
||||
// EncodeMaxShards creates and internal representation of max shards.
|
||||
func (h *Holder) EncodeMaxShards() *internal.MaxShards {
|
||||
// encodeMaxShards creates and internal representation of max shards.
|
||||
func (h *Holder) encodeMaxShards() *internal.MaxShards {
|
||||
return &internal.MaxShards{
|
||||
Standard: h.MaxShards(),
|
||||
Standard: h.maxShards(),
|
||||
}
|
||||
}
|
||||
|
||||
// EncodeSchema creates an internal representation of schema.
|
||||
func (h *Holder) EncodeSchema() *internal.Schema {
|
||||
// encodeSchema creates an internal representation of schema.
|
||||
func (h *Holder) encodeSchema() *internal.Schema {
|
||||
return &internal.Schema{
|
||||
Indexes: EncodeIndexes(h.Indexes()),
|
||||
}
|
||||
|
|
@ -360,7 +360,7 @@ func (h *Holder) newIndex(path, name string) (*Index, error) {
|
|||
}
|
||||
index.Logger = h.Logger
|
||||
index.Stats = h.Stats.WithTags(fmt.Sprintf("index:%s", index.Name()))
|
||||
index.broadcaster = h.Broadcaster
|
||||
index.broadcaster = h.broadcaster
|
||||
index.NewAttrStore = h.NewAttrStore
|
||||
index.columnAttrStore = h.NewAttrStore(filepath.Join(index.path, ".data"))
|
||||
return index, nil
|
||||
|
|
@ -402,8 +402,8 @@ func (h *Holder) Field(index, name string) *Field {
|
|||
return idx.Field(name)
|
||||
}
|
||||
|
||||
// View returns the view for an index, field, and name.
|
||||
func (h *Holder) View(index, field, name string) *View {
|
||||
// view returns the view for an index, field, and name.
|
||||
func (h *Holder) view(index, field, name string) *View {
|
||||
f := h.Field(index, field)
|
||||
if f == nil {
|
||||
return nil
|
||||
|
|
@ -411,9 +411,9 @@ func (h *Holder) View(index, field, name string) *View {
|
|||
return f.view(name)
|
||||
}
|
||||
|
||||
// Fragment returns the fragment for an index, field & shard.
|
||||
func (h *Holder) Fragment(index, field, view string, shard uint64) *Fragment {
|
||||
v := h.View(index, field, view)
|
||||
// fragment returns the fragment for an index, field & shard.
|
||||
func (h *Holder) fragment(index, field, view string, shard uint64) *Fragment {
|
||||
v := h.view(index, field, view)
|
||||
if v == nil {
|
||||
return nil
|
||||
}
|
||||
|
|
@ -423,7 +423,7 @@ func (h *Holder) Fragment(index, field, view string, shard uint64) *Fragment {
|
|||
// monitorCacheFlush periodically flushes all fragment caches sequentially.
|
||||
// This is run in a goroutine.
|
||||
func (h *Holder) monitorCacheFlush() {
|
||||
ticker := time.NewTicker(h.CacheFlushInterval)
|
||||
ticker := time.NewTicker(h.cacheFlushInterval)
|
||||
defer ticker.Stop()
|
||||
|
||||
for {
|
||||
|
|
@ -476,11 +476,11 @@ func (h *Holder) setFileLimit() {
|
|||
return
|
||||
}
|
||||
// If the soft limit is lower than the FileLimit constant, we will try to change it.
|
||||
if oldLimit.Cur < FileLimit {
|
||||
newLimit.Cur = FileLimit
|
||||
if oldLimit.Cur < fileLimit {
|
||||
newLimit.Cur = fileLimit
|
||||
// If the hard limit is not high enough, we will try to change it too.
|
||||
if oldLimit.Max < FileLimit {
|
||||
newLimit.Max = FileLimit
|
||||
if oldLimit.Max < fileLimit {
|
||||
newLimit.Max = fileLimit
|
||||
} else {
|
||||
newLimit.Max = oldLimit.Max
|
||||
}
|
||||
|
|
@ -508,8 +508,8 @@ func (h *Holder) setFileLimit() {
|
|||
if err := syscall.Getrlimit(syscall.RLIMIT_NOFILE, oldLimit); err != nil {
|
||||
h.Logger.Printf("ERROR checking open file limit: %s", err)
|
||||
} else {
|
||||
if oldLimit.Cur < FileLimit {
|
||||
h.Logger.Printf("WARNING: Tried to set open file limit to %d, but it is %d. You may consider running \"sudo ulimit -n %d\" before starting Pilosa to avoid \"too many open files\" error. See https://www.pilosa.com/docs/administration/#open-file-limits for more information.", FileLimit, oldLimit.Cur, FileLimit)
|
||||
if oldLimit.Cur < fileLimit {
|
||||
h.Logger.Printf("WARNING: Tried to set open file limit to %d, but it is %d. You may consider running \"sudo ulimit -n %d\" before starting Pilosa to avoid \"too many open files\" error. See https://www.pilosa.com/docs/administration/#open-file-limits for more information.", fileLimit, oldLimit.Cur, fileLimit)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -561,9 +561,9 @@ func (h *Holder) logStartup() error {
|
|||
return nil
|
||||
}
|
||||
|
||||
// HolderSyncer is an active anti-entropy tool that compares the local holder
|
||||
// holderSyncer is an active anti-entropy tool that compares the local holder
|
||||
// with a remote holder based on block checksums and resolves differences.
|
||||
type HolderSyncer struct {
|
||||
type holderSyncer struct {
|
||||
mu sync.Mutex
|
||||
|
||||
Holder *Holder
|
||||
|
|
@ -579,7 +579,7 @@ type HolderSyncer struct {
|
|||
}
|
||||
|
||||
// IsClosing returns true if the syncer has been marked to close.
|
||||
func (s *HolderSyncer) IsClosing() bool {
|
||||
func (s *holderSyncer) IsClosing() bool {
|
||||
select {
|
||||
case <-s.Closing:
|
||||
return true
|
||||
|
|
@ -589,7 +589,7 @@ func (s *HolderSyncer) IsClosing() bool {
|
|||
}
|
||||
|
||||
// SyncHolder compares the holder on host with the local holder and resolves differences.
|
||||
func (s *HolderSyncer) SyncHolder() error {
|
||||
func (s *holderSyncer) SyncHolder() error {
|
||||
s.mu.Lock() // only allow one instance of SyncHolder to be running at a time
|
||||
defer s.mu.Unlock()
|
||||
ti := time.Now()
|
||||
|
|
@ -651,7 +651,7 @@ func (s *HolderSyncer) SyncHolder() error {
|
|||
}
|
||||
|
||||
// syncIndex synchronizes index attributes with the rest of the cluster.
|
||||
func (s *HolderSyncer) syncIndex(index string) error {
|
||||
func (s *holderSyncer) syncIndex(index string) error {
|
||||
// Retrieve index reference.
|
||||
idx := s.Holder.Index(index)
|
||||
if idx == nil {
|
||||
|
|
@ -694,7 +694,7 @@ func (s *HolderSyncer) syncIndex(index string) error {
|
|||
}
|
||||
|
||||
// syncField synchronizes field attributes with the rest of the cluster.
|
||||
func (s *HolderSyncer) syncField(index, name string) error {
|
||||
func (s *holderSyncer) syncField(index, name string) error {
|
||||
// Retrieve field reference.
|
||||
f := s.Holder.Field(index, name)
|
||||
if f == nil {
|
||||
|
|
@ -740,7 +740,7 @@ func (s *HolderSyncer) syncField(index, name string) error {
|
|||
}
|
||||
|
||||
// syncFragment synchronizes a fragment with the rest of the cluster.
|
||||
func (s *HolderSyncer) syncFragment(index, field, view string, shard uint64) error {
|
||||
func (s *holderSyncer) syncFragment(index, field, view string, shard uint64) error {
|
||||
// Retrieve local field.
|
||||
f := s.Holder.Field(index, field)
|
||||
if f == nil {
|
||||
|
|
@ -773,8 +773,8 @@ func (s *HolderSyncer) syncFragment(index, field, view string, shard uint64) err
|
|||
return nil
|
||||
}
|
||||
|
||||
// HolderCleaner removes fragments and data files that are no longer used.
|
||||
type HolderCleaner struct {
|
||||
// holderCleaner removes fragments and data files that are no longer used.
|
||||
type holderCleaner struct {
|
||||
Node *Node
|
||||
|
||||
Holder *Holder
|
||||
|
|
@ -785,7 +785,7 @@ type HolderCleaner struct {
|
|||
}
|
||||
|
||||
// IsClosing returns true if the cleaner has been marked to close.
|
||||
func (c *HolderCleaner) IsClosing() bool {
|
||||
func (c *holderCleaner) IsClosing() bool {
|
||||
select {
|
||||
case <-c.Closing:
|
||||
return true
|
||||
|
|
@ -796,7 +796,7 @@ func (c *HolderCleaner) IsClosing() bool {
|
|||
|
||||
// CleanHolder compares the holder with the cluster state and removes
|
||||
// any unnecessary fragments and files.
|
||||
func (c *HolderCleaner) CleanHolder() error {
|
||||
func (c *holderCleaner) CleanHolder() error {
|
||||
for _, index := range c.Holder.Indexes() {
|
||||
// Verify cleaner has not closed.
|
||||
if c.IsClosing() {
|
||||
|
|
|
|||
|
|
@ -18,6 +18,7 @@ import (
|
|||
"io/ioutil"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"reflect"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
|
@ -57,6 +58,43 @@ func newHolder() *tHolder {
|
|||
return h
|
||||
}
|
||||
|
||||
// MustCreateFieldIfNotExists returns a given field. Panic on error.
|
||||
func (h *tHolder) MustCreateFieldIfNotExists(index, field string) *Field {
|
||||
f, err := h.MustCreateIndexIfNotExists(index, IndexOptions{}).CreateFieldIfNotExists(field, FieldOptions{})
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
return f
|
||||
}
|
||||
|
||||
// MustCreateIndexIfNotExists returns a given index. Panic on error.
|
||||
func (h *tHolder) MustCreateIndexIfNotExists(index string, opt IndexOptions) *Index {
|
||||
idx, err := h.Holder.CreateIndexIfNotExists(index, opt)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
return idx
|
||||
}
|
||||
|
||||
// SetBit clears a bit on the given field.
|
||||
func (h *tHolder) SetBit(index, field string, rowID, columnID uint64) {
|
||||
f := h.MustCreateFieldIfNotExists(index, field)
|
||||
_, err := f.SetBit(rowID, columnID, nil)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
}
|
||||
|
||||
// Row returns a Row for a given field.
|
||||
func (h *tHolder) Row(index, field string, rowID uint64) *Row {
|
||||
f := h.MustCreateFieldIfNotExists(index, field)
|
||||
row, err := f.Row(rowID)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
return row
|
||||
}
|
||||
|
||||
func TestHolder_Optn(t *testing.T) {
|
||||
t.Run("ErrViewPermission", func(t *testing.T) {
|
||||
if os.Geteuid() == 0 {
|
||||
|
|
@ -137,3 +175,117 @@ func TestHolder_Optn(t *testing.T) {
|
|||
})
|
||||
|
||||
}
|
||||
|
||||
// Ensure holder can clean up orphaned fragments.
|
||||
func TestHolderCleaner_CleanHolder(t *testing.T) {
|
||||
cluster := NewTestCluster(2)
|
||||
|
||||
// Create a local holder.
|
||||
hldr0 := newHolder()
|
||||
defer hldr0.Close()
|
||||
|
||||
// Mock 2-node, fully replicated cluster.
|
||||
cluster.ReplicaN = 2
|
||||
|
||||
cluster.Nodes[0].URI = NewTestURIFromHostPort("localhost", 0)
|
||||
|
||||
// Create fields on nodes.
|
||||
for _, hldr := range []*tHolder{hldr0} {
|
||||
hldr.MustCreateFieldIfNotExists("i", "f")
|
||||
hldr.MustCreateFieldIfNotExists("i", "f0")
|
||||
hldr.MustCreateFieldIfNotExists("y", "z")
|
||||
}
|
||||
|
||||
// Set data on the local holder.
|
||||
hldr0.SetBit("i", "f", 0, 10)
|
||||
hldr0.SetBit("i", "f", 0, 4000)
|
||||
hldr0.SetBit("i", "f", 2, 20)
|
||||
hldr0.SetBit("i", "f", 3, 10)
|
||||
hldr0.SetBit("i", "f", 120, 10)
|
||||
hldr0.SetBit("i", "f", 200, 4)
|
||||
|
||||
hldr0.SetBit("i", "f0", 9, ShardWidth+5)
|
||||
|
||||
hldr0.SetBit("y", "z", 10, (2*ShardWidth)+4)
|
||||
hldr0.SetBit("y", "z", 10, (2*ShardWidth)+5)
|
||||
hldr0.SetBit("y", "z", 10, (2*ShardWidth)+7)
|
||||
|
||||
// Set highest shard.
|
||||
hldr0.Index("i").SetRemoteMaxShard(1)
|
||||
hldr0.Index("y").SetRemoteMaxShard(2)
|
||||
|
||||
// Keep replication the same and ensure we get the expected results.
|
||||
cluster.ReplicaN = 2
|
||||
|
||||
// Set up cleaner for replication 2.
|
||||
cleaner2 := holderCleaner{
|
||||
Node: cluster.Nodes[0],
|
||||
Holder: hldr0.Holder,
|
||||
Cluster: cluster,
|
||||
}
|
||||
|
||||
if err := cleaner2.CleanHolder(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// Verify data is the same on both nodes.
|
||||
for i, hldr := range []*tHolder{hldr0} {
|
||||
if a := hldr.Row("i", "f", 0).Columns(); !reflect.DeepEqual(a, []uint64{10, 4000}) {
|
||||
t.Fatalf("unexpected columns(%d/0): %+v", i, a)
|
||||
} else if a := hldr.Row("i", "f", 2).Columns(); !reflect.DeepEqual(a, []uint64{20}) {
|
||||
t.Fatalf("unexpected columns(%d/2): %+v", i, a)
|
||||
} else if a := hldr.Row("i", "f", 3).Columns(); !reflect.DeepEqual(a, []uint64{10}) {
|
||||
t.Fatalf("unexpected columns(%d/3): %+v", i, a)
|
||||
} else if a := hldr.Row("i", "f", 120).Columns(); !reflect.DeepEqual(a, []uint64{10}) {
|
||||
t.Fatalf("unexpected columns(%d/120): %+v", i, a)
|
||||
} else if a := hldr.Row("i", "f", 200).Columns(); !reflect.DeepEqual(a, []uint64{4}) {
|
||||
t.Fatalf("unexpected columns(%d/200): %+v", i, a)
|
||||
}
|
||||
|
||||
if a := hldr.Row("i", "f0", 9).Columns(); !reflect.DeepEqual(a, []uint64{ShardWidth + 5}) {
|
||||
t.Fatalf("unexpected columns(%d/d/f0): %+v", i, a)
|
||||
}
|
||||
|
||||
if a := hldr.Row("y", "z", 10).Columns(); !reflect.DeepEqual(a, []uint64{(2 * ShardWidth) + 4, (2 * ShardWidth) + 5, (2 * ShardWidth) + 7}) {
|
||||
t.Fatalf("unexpected columns(%d/y/z): %+v", i, a)
|
||||
}
|
||||
}
|
||||
|
||||
// Change replication factor to ensure we have fragments to remove.
|
||||
cluster.ReplicaN = 1
|
||||
|
||||
// Set up cleaner for replication 1.
|
||||
cleaner1 := holderCleaner{
|
||||
Node: cluster.Nodes[0],
|
||||
Holder: hldr0.Holder,
|
||||
Cluster: cluster,
|
||||
}
|
||||
|
||||
if err := cleaner1.CleanHolder(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// Verify data is the same on both nodes.
|
||||
for i, hldr := range []*tHolder{hldr0} {
|
||||
if a := hldr.Row("i", "f", 0).Columns(); !reflect.DeepEqual(a, []uint64{10, 4000}) {
|
||||
t.Fatalf("unexpected columns(%d/0): %+v", i, a)
|
||||
} else if a := hldr.Row("i", "f", 2).Columns(); !reflect.DeepEqual(a, []uint64{20}) {
|
||||
t.Fatalf("unexpected columns(%d/2): %+v", i, a)
|
||||
} else if a := hldr.Row("i", "f", 3).Columns(); !reflect.DeepEqual(a, []uint64{10}) {
|
||||
t.Fatalf("unexpected columns(%d/3): %+v", i, a)
|
||||
} else if a := hldr.Row("i", "f", 120).Columns(); !reflect.DeepEqual(a, []uint64{10}) {
|
||||
t.Fatalf("unexpected columns(%d/120): %+v", i, a)
|
||||
} else if a := hldr.Row("i", "f", 200).Columns(); !reflect.DeepEqual(a, []uint64{4}) {
|
||||
t.Fatalf("unexpected columns(%d/200): %+v", i, a)
|
||||
}
|
||||
|
||||
f := hldr.fragment("i", "f0", ViewStandard, 1)
|
||||
if f != nil {
|
||||
t.Fatalf("expected fragment to be deleted: (%d/i/f0): %+v", i, f)
|
||||
}
|
||||
|
||||
if a := hldr.Row("y", "z", 10).Columns(); !reflect.DeepEqual(a, []uint64{(2 * ShardWidth) + 4, (2 * ShardWidth) + 5, (2 * ShardWidth) + 7}) {
|
||||
t.Fatalf("unexpected columns(%d/y/z): %+v", i, a)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
114
holder_test.go
114
holder_test.go
|
|
@ -362,117 +362,3 @@ func TestHolderSyncer_SyncHolder(t *testing.T) {
|
|||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Ensure holder can clean up orphaned fragments.
|
||||
func TestHolderCleaner_CleanHolder(t *testing.T) {
|
||||
cluster := pilosa.NewTestCluster(2)
|
||||
|
||||
// Create a local holder.
|
||||
hldr0 := test.MustOpenHolder()
|
||||
defer hldr0.Close()
|
||||
|
||||
// Mock 2-node, fully replicated cluster.
|
||||
cluster.ReplicaN = 2
|
||||
|
||||
cluster.Nodes[0].URI = pilosa.NewTestURIFromHostPort("localhost", 0)
|
||||
|
||||
// Create fields on nodes.
|
||||
for _, hldr := range []*test.Holder{hldr0} {
|
||||
hldr.MustCreateFieldIfNotExists("i", "f")
|
||||
hldr.MustCreateFieldIfNotExists("i", "f0")
|
||||
hldr.MustCreateFieldIfNotExists("y", "z")
|
||||
}
|
||||
|
||||
// Set data on the local holder.
|
||||
hldr0.SetBit("i", "f", 0, 10)
|
||||
hldr0.SetBit("i", "f", 0, 4000)
|
||||
hldr0.SetBit("i", "f", 2, 20)
|
||||
hldr0.SetBit("i", "f", 3, 10)
|
||||
hldr0.SetBit("i", "f", 120, 10)
|
||||
hldr0.SetBit("i", "f", 200, 4)
|
||||
|
||||
hldr0.SetBit("i", "f0", 9, ShardWidth+5)
|
||||
|
||||
hldr0.SetBit("y", "z", 10, (2*ShardWidth)+4)
|
||||
hldr0.SetBit("y", "z", 10, (2*ShardWidth)+5)
|
||||
hldr0.SetBit("y", "z", 10, (2*ShardWidth)+7)
|
||||
|
||||
// Set highest shard.
|
||||
hldr0.Index("i").SetRemoteMaxShard(1)
|
||||
hldr0.Index("y").SetRemoteMaxShard(2)
|
||||
|
||||
// Keep replication the same and ensure we get the expected results.
|
||||
cluster.ReplicaN = 2
|
||||
|
||||
// Set up cleaner for replication 2.
|
||||
cleaner2 := pilosa.HolderCleaner{
|
||||
Node: cluster.Nodes[0],
|
||||
Holder: hldr0.Holder,
|
||||
Cluster: cluster,
|
||||
}
|
||||
|
||||
if err := cleaner2.CleanHolder(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// Verify data is the same on both nodes.
|
||||
for i, hldr := range []*test.Holder{hldr0} {
|
||||
if a := hldr.Row("i", "f", 0).Columns(); !reflect.DeepEqual(a, []uint64{10, 4000}) {
|
||||
t.Fatalf("unexpected columns(%d/0): %+v", i, a)
|
||||
} else if a := hldr.Row("i", "f", 2).Columns(); !reflect.DeepEqual(a, []uint64{20}) {
|
||||
t.Fatalf("unexpected columns(%d/2): %+v", i, a)
|
||||
} else if a := hldr.Row("i", "f", 3).Columns(); !reflect.DeepEqual(a, []uint64{10}) {
|
||||
t.Fatalf("unexpected columns(%d/3): %+v", i, a)
|
||||
} else if a := hldr.Row("i", "f", 120).Columns(); !reflect.DeepEqual(a, []uint64{10}) {
|
||||
t.Fatalf("unexpected columns(%d/120): %+v", i, a)
|
||||
} else if a := hldr.Row("i", "f", 200).Columns(); !reflect.DeepEqual(a, []uint64{4}) {
|
||||
t.Fatalf("unexpected columns(%d/200): %+v", i, a)
|
||||
}
|
||||
|
||||
if a := hldr.Row("i", "f0", 9).Columns(); !reflect.DeepEqual(a, []uint64{ShardWidth + 5}) {
|
||||
t.Fatalf("unexpected columns(%d/d/f0): %+v", i, a)
|
||||
}
|
||||
|
||||
if a := hldr.Row("y", "z", 10).Columns(); !reflect.DeepEqual(a, []uint64{(2 * ShardWidth) + 4, (2 * ShardWidth) + 5, (2 * ShardWidth) + 7}) {
|
||||
t.Fatalf("unexpected columns(%d/y/z): %+v", i, a)
|
||||
}
|
||||
}
|
||||
|
||||
// Change replication factor to ensure we have fragments to remove.
|
||||
cluster.ReplicaN = 1
|
||||
|
||||
// Set up cleaner for replication 1.
|
||||
cleaner1 := pilosa.HolderCleaner{
|
||||
Node: cluster.Nodes[0],
|
||||
Holder: hldr0.Holder,
|
||||
Cluster: cluster,
|
||||
}
|
||||
|
||||
if err := cleaner1.CleanHolder(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// Verify data is the same on both nodes.
|
||||
for i, hldr := range []*test.Holder{hldr0} {
|
||||
if a := hldr.Row("i", "f", 0).Columns(); !reflect.DeepEqual(a, []uint64{10, 4000}) {
|
||||
t.Fatalf("unexpected columns(%d/0): %+v", i, a)
|
||||
} else if a := hldr.Row("i", "f", 2).Columns(); !reflect.DeepEqual(a, []uint64{20}) {
|
||||
t.Fatalf("unexpected columns(%d/2): %+v", i, a)
|
||||
} else if a := hldr.Row("i", "f", 3).Columns(); !reflect.DeepEqual(a, []uint64{10}) {
|
||||
t.Fatalf("unexpected columns(%d/3): %+v", i, a)
|
||||
} else if a := hldr.Row("i", "f", 120).Columns(); !reflect.DeepEqual(a, []uint64{10}) {
|
||||
t.Fatalf("unexpected columns(%d/120): %+v", i, a)
|
||||
} else if a := hldr.Row("i", "f", 200).Columns(); !reflect.DeepEqual(a, []uint64{4}) {
|
||||
t.Fatalf("unexpected columns(%d/200): %+v", i, a)
|
||||
}
|
||||
|
||||
f := hldr.Fragment("i", "f0", pilosa.ViewStandard, 1)
|
||||
if f != nil {
|
||||
t.Fatalf("expected fragment to be deleted: (%d/i/f0): %+v", i, f)
|
||||
}
|
||||
|
||||
if a := hldr.Row("y", "z", 10).Columns(); !reflect.DeepEqual(a, []uint64{(2 * ShardWidth) + 4, (2 * ShardWidth) + 5, (2 * ShardWidth) + 7}) {
|
||||
t.Fatalf("unexpected columns(%d/y/z): %+v", i, a)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -306,7 +306,9 @@ func TestClient_FragmentBlocks(t *testing.T) {
|
|||
}
|
||||
|
||||
// Verify data matches local blocks.
|
||||
if a := hldr.Fragment("i", "f", pilosa.ViewStandard, 0).Blocks(); !reflect.DeepEqual(a, blocks) {
|
||||
if a, err := cmd.API.FragmentBlocks(context.Background(), "i", "f", 0); err != nil {
|
||||
t.Fatal(err)
|
||||
} else if !reflect.DeepEqual(a, blocks) {
|
||||
t.Fatalf("blocks mismatch:\n\nexp=%s\n\ngot=%s\n\n", spew.Sdump(a), spew.Sdump(blocks))
|
||||
}
|
||||
}
|
||||
|
|
|
|||
12
server.go
12
server.go
|
|
@ -70,7 +70,7 @@ type Server struct {
|
|||
diagnosticInterval time.Duration
|
||||
maxWritesPerRequest int
|
||||
isCoordinator bool
|
||||
syncer HolderSyncer
|
||||
syncer holderSyncer
|
||||
|
||||
primaryTranslateStore TranslateStore
|
||||
|
||||
|
|
@ -298,7 +298,7 @@ func NewServer(opts ...ServerOption) (*Server, error) {
|
|||
s.executor.MaxWritesPerRequest = s.maxWritesPerRequest
|
||||
s.cluster.broadcaster = s
|
||||
s.cluster.maxWritesPerRequest = s.maxWritesPerRequest
|
||||
s.holder.Broadcaster = s
|
||||
s.holder.broadcaster = s
|
||||
|
||||
err = s.cluster.setup()
|
||||
if err != nil {
|
||||
|
|
@ -572,8 +572,8 @@ func (s *Server) LocalStatus() (proto.Message, error) {
|
|||
|
||||
ns := internal.NodeStatus{
|
||||
Node: EncodeNode(s.cluster.Node),
|
||||
MaxShards: s.holder.EncodeMaxShards(),
|
||||
Schema: s.holder.EncodeSchema(),
|
||||
MaxShards: s.holder.encodeMaxShards(),
|
||||
Schema: s.holder.encodeSchema(),
|
||||
}
|
||||
|
||||
return &ns, nil
|
||||
|
|
@ -606,12 +606,12 @@ func (s *Server) mergeRemoteStatus(ns *internal.NodeStatus) error {
|
|||
}
|
||||
|
||||
// Sync schema.
|
||||
if err := s.holder.ApplySchema(ns.Schema); err != nil {
|
||||
if err := s.holder.applySchema(ns.Schema); err != nil {
|
||||
return errors.Wrap(err, "applying schema")
|
||||
}
|
||||
|
||||
// Sync maxShards.
|
||||
oldmaxshards := s.holder.MaxShards()
|
||||
oldmaxshards := s.holder.maxShards()
|
||||
for index, newMax := range ns.MaxShards.Standard {
|
||||
localIndex := s.holder.Index(index)
|
||||
// if we don't know about an index locally, log an error because
|
||||
|
|
|
|||
|
|
@ -360,7 +360,7 @@ func (t *ClusterCluster) FollowResizeInstruction(instr *internal.ResizeInstructi
|
|||
destCluster := t.clusterByID(instrNode.ID)
|
||||
|
||||
// Sync the schema received in the resize instruction.
|
||||
if err := destCluster.holder.ApplySchema(instr.Schema); err != nil {
|
||||
if err := destCluster.holder.applySchema(instr.Schema); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
|
|
@ -368,8 +368,8 @@ func (t *ClusterCluster) FollowResizeInstruction(instr *internal.ResizeInstructi
|
|||
srcNode := DecodeNode(src.Node)
|
||||
srcCluster := t.clusterByID(srcNode.ID)
|
||||
|
||||
srcFragment := srcCluster.holder.Fragment(src.Index, src.Field, src.View, src.Shard)
|
||||
destFragment := destCluster.holder.Fragment(src.Index, src.Field, src.View, src.Shard)
|
||||
srcFragment := srcCluster.holder.fragment(src.Index, src.Field, src.View, src.Shard)
|
||||
destFragment := destCluster.holder.fragment(src.Index, src.Field, src.View, src.Shard)
|
||||
if destFragment == nil {
|
||||
// Create fragment on destination if it doesn't exist.
|
||||
f := destCluster.holder.Field(src.Index, src.Field)
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue