Merge pull request #1964 from jaffee/v1.3-backport-1951

Backport #1951

Author: jaffee

field.go

@@ -58,10 +58,6 @@ const (
 	FieldTypeBool  = "bool"
 )
 
-func defaultShardValidator(shard uint64) bool {
-	return true
-}
-
 // Field represents a container for views.
 type Field struct {
 	mu    sync.RWMutex
@@ -84,7 +80,6 @@ type Field struct {
 
 	// Shards with data on any node in the cluster, according to this node.
 	remoteAvailableShards *roaring.Bitmap
-	shardValidator        func(uint64) bool
 
 	logger logger.Logger
 }
@@ -212,8 +207,7 @@ func newField(path, index, name string, opts FieldOption) (*Field, error) {
 
 		remoteAvailableShards: roaring.NewBitmap(),
 
-		shardValidator: defaultShardValidator,
-		logger:         logger.NopLogger,
+		logger: logger.NopLogger,
 	}
 	return f, nil
 }
@@ -762,7 +756,6 @@ func (f *Field) newView(path, name string) *view {
 	view.rowAttrStore = f.rowAttrStore
 	view.stats = f.Stats.WithTags(fmt.Sprintf("view:%s", name))
 	view.broadcaster = f.broadcaster
-	view.shardValidator = f.shardValidator
 	return view
 }
 

holder.go

@@ -77,8 +77,6 @@ type Holder struct {
 	// The interval at which the cached row ids are persisted to disk.
 	cacheFlushInterval time.Duration
 
-	shardValidatorFunc func(index string, shard uint64) bool
-
 	Logger logger.Logger
 }
 
@@ -125,9 +123,6 @@ func NewHolder() *Holder {
 		NewAttrStore: newNopAttrStore,
 
 		cacheFlushInterval: defaultCacheFlushInterval,
-		shardValidatorFunc: func(index string, shard uint64) bool {
-			return true //default
-		},
 
 		Logger: logger.NopLogger,
 	}
@@ -430,9 +425,6 @@ func (h *Holder) newIndex(path, name string) (*Index, error) {
 	index.broadcaster = h.broadcaster
 	index.newAttrStore = h.NewAttrStore
 	index.columnAttrs = h.NewAttrStore(filepath.Join(index.path, ".data"))
-	index.shardValidator = func(shard uint64) bool {
-		return h.shardValidatorFunc(name, shard)
-	}
 	return index, nil
 }
 

index.go

@@ -50,9 +50,8 @@ type Index struct {
 	// Column attribute storage and cache.
 	columnAttrs AttrStore
 
-	broadcaster    broadcaster
-	Stats          stats.StatsClient
-	shardValidator func(uint64) bool
+	broadcaster broadcaster
+	Stats       stats.StatsClient
 
 	logger logger.Logger
 }
@@ -75,7 +74,6 @@ func NewIndex(path, name string) (*Index, error) {
 		broadcaster:    NopBroadcaster,
 		Stats:          stats.NopStatsClient,
 		logger:         logger.NopLogger,
-		shardValidator: defaultShardValidator,
 		trackExistence: true,
 	}, nil
 }
@@ -405,7 +403,6 @@ func (i *Index) newField(path, name string) (*Field, error) {
 	f.Stats = i.Stats.WithTags(fmt.Sprintf("field:%s", name))
 	f.broadcaster = i.broadcaster
 	f.rowAttrStore = i.newAttrStore(filepath.Join(f.path, ".data"))
-	f.shardValidator = i.shardValidator
 	return f, nil
 }
 

server.go

@@ -321,9 +321,6 @@ func NewServer(opts ...ServerOption) (*Server, error) {
 	s.cluster.broadcaster = s
 	s.cluster.maxWritesPerRequest = s.maxWritesPerRequest
 	s.holder.broadcaster = s
-	s.holder.shardValidatorFunc = func(index string, shard uint64) bool {
-		return s.cluster.ownsShard(s.nodeID, index, shard)
-	}
 
 	err = s.cluster.setup()
 	if err != nil {

server_internal_test.go

@@ -16,12 +16,9 @@ package pilosa
 
 import (
 	"io/ioutil"
-	"os"
 	"runtime"
 	"testing"
 	"time"
-
-	"github.com/pilosa/pilosa/roaring"
 )
 
 // Ensure the file handle count is working
@@ -63,58 +60,3 @@ func TestMonitorAntiEntropyZero(t *testing.T) {
 		t.Fatalf("monitorAntiEntropy should have returned immediately with duration 0")
 	}
 }
-
-func TestOnlyOpenOwnedFiles(t *testing.T) {
-	path, err := ioutil.TempDir("", "pilosa")
-	if err != nil {
-		t.Fatalf("getting temp dir: %v", err)
-	}
-	defer func() {
-		err := os.RemoveAll(path)
-		if err != nil {
-			t.Logf("cleaning up temp dir: %v", err)
-		}
-	}()
-	bm := roaring.NewFileBitmap(1, 2, 3)
-	err = os.MkdirAll(path+"/i/f/views/standard/fragments", os.ModeDir|os.ModePerm)
-	if err != nil {
-		t.Fatalf("mkdirall: %v", err)
-	}
-	one, err := os.Create(path + "/i/f/views/standard/fragments/1")
-	if err != nil {
-		t.Fatalf("creating one: %v", err)
-	}
-	two, err := os.Create(path + "/i/f/views/standard/fragments/2")
-	if err != nil {
-		t.Fatalf("creating two: %v", err)
-	}
-	_, err = bm.WriteTo(one)
-	if err != nil {
-		t.Fatalf("writing to one: %v", err)
-	}
-	_, err = bm.WriteTo(two)
-	if err != nil {
-		t.Fatalf("writing to two: %v", err)
-	}
-
-	h := NewHolder()
-	h.Path = path
-	h.shardValidatorFunc = func(index string, shard uint64) bool {
-		return shard == 1
-	}
-
-	err = h.Open()
-	if err != nil {
-		t.Fatalf("opening holder: %v", err)
-	}
-
-	view := h.Index("i").Field("f").view("standard")
-
-	if len(view.fragments) != 1 {
-		t.Errorf("should have one fragment, but have: %d", len(view.fragments))
-	}
-
-	if _, ok := view.fragments[1]; !ok {
-		t.Errorf("should have fragment 1, but fragments: %#v", view.fragments)
-	}
-}

view.go

@@ -52,11 +52,10 @@ type view struct {
 	// Fragments by shard.
 	fragments map[uint64]*fragment
 
-	broadcaster    broadcaster
-	stats          stats.StatsClient
-	rowAttrStore   AttrStore
-	logger         logger.Logger
-	shardValidator func(uint64) bool
+	broadcaster  broadcaster
+	stats        stats.StatsClient
+	rowAttrStore AttrStore
+	logger       logger.Logger
 }
 
 // newView returns a new instance of View.
@@ -73,10 +72,9 @@ func newView(path, index, field, name string, fieldOptions FieldOptions) *view {
 
 		fragments: make(map[uint64]*fragment),
 
-		broadcaster:    NopBroadcaster,
-		stats:          stats.NopStatsClient,
-		logger:         logger.NopLogger,
-		shardValidator: defaultShardValidator,
+		broadcaster: NopBroadcaster,
+		stats:       stats.NopStatsClient,
+		logger:      logger.NopLogger,
 	}
 }
 
@@ -134,10 +132,6 @@ func (v *view) openFragments() error {
 		if err != nil {
 			continue
 		}
-		//skip shard if not owned
-		if !v.shardValidator(shard) {
-			continue
-		}
 
 		frag := v.newFragment(v.fragmentPath(shard), shard)
 		if err := frag.Open(); err != nil {