Merge pull request #935 from travisturner/cluster-resize-test-setvalue

add SetFieldValue() method to TestCluster and use in tests

cluster_test.go

@@ -413,13 +413,38 @@ func TestCluster_ResizeStates(t *testing.T) {
 			t.Fatal(err)
 		}
 
-		// Add Data to node0.
-		tc.CreateFrame("i", "f", pilosa.FrameOptions{})
+		// Add Bit Data to node0.
+		if err := tc.CreateFrame("i", "f", pilosa.FrameOptions{}); err != nil {
+			t.Fatal(err)
+		}
 		tc.SetBit("i", "f", "standard", 1, 101, nil)
 		tc.SetBit("i", "f", "standard", 1, 1300000, nil)
 
+		// Add Field Data to node0.
+		if err := tc.CreateFrame("i", "fields", pilosa.FrameOptions{
+			InverseEnabled: false,
+			RangeEnabled:   true,
+			//CacheType:      pilosa.CacheTypeNone,
+			Fields: []*pilosa.Field{
+				{
+					Name: "fld0",
+					Type: pilosa.FieldTypeInt,
+					Min:  -100,
+					Max:  100,
+				},
+			},
+		}); err != nil {
+			t.Fatal(err)
+		}
+		tc.SetFieldValue("i", "fields", 1, "fld0", -10)
+		tc.SetFieldValue("i", "fields", 1, "fld0", 10)
+		tc.SetFieldValue("i", "fields", 1300000, "fld0", -99)
+		tc.SetFieldValue("i", "fields", 1300000, "fld0", 99)
+
 		// AddNode needs to block until the resize process has completed.
-		tc.AddNode(false)
+		if err := tc.AddNode(false); err != nil {
+			t.Fatal(err)
+		}
 
 		node0 := tc.Clusters[0]
 		node1 := tc.Clusters[1]
@@ -442,6 +467,7 @@ func TestCluster_ResizeStates(t *testing.T) {
 			t.Errorf("expected node1 topology: %v, but got: %v", expectedTop.NodeSet, node1.Topology.NodeSet)
 		}
 
+		// Bits
 		// Verify that node-1 contains the fragment (i/f/standard/1) transferred from node-0.
 		node0Frame := node0.Holder.Frame("i", "f")
 		node0View := node0Frame.View("standard")
@@ -454,7 +480,23 @@ func TestCluster_ResizeStates(t *testing.T) {
 		// Ensure checksums are the same.
 		orig := node0Fragment.Checksum()
 		if chksum := node1Fragment.Checksum(); !bytes.Equal(chksum, orig) {
-			t.Fatalf("expected checksum to match: %x - %x", chksum, orig)
+			t.Fatalf("expected standard view checksum to match: %x - %x", chksum, orig)
+		}
+
+		// Values
+		// Verify that node-1 contains the fragment (i/fields/field_fld0/1) transferred from node-0.
+		node0Frame = node0.Holder.Frame("i", "fields")
+		node0View = node0Frame.View("field_fld0")
+		node0Fragment = node0View.Fragment(1)
+
+		node1Frame = node1.Holder.Frame("i", "fields")
+		node1View = node1Frame.View("field_fld0")
+		node1Fragment = node1View.Fragment(1)
+
+		// Ensure checksums are the same.
+		orig = node0Fragment.Checksum()
+		if chksum := node1Fragment.Checksum(); !bytes.Equal(chksum, orig) {
+			t.Fatalf("expected field view checksum to match: %x - %x", chksum, orig)
 		}
 
 		// Close TestCluster.

fragment.go

@@ -254,6 +254,7 @@ func (f *Fragment) openCache() error {
 		f.cache = NewLRUCache(f.CacheSize)
 	case CacheTypeNone:
 		f.cache = NewNopCache()
+		return nil
 	default:
 		return ErrInvalidCacheType
 	}
@@ -1451,6 +1452,10 @@ func (f *Fragment) flushCache() error {
 		return nil
 	}
 
+	if f.CacheType == CacheTypeNone {
+		return nil
+	}
+
 	// Retrieve a list of row ids from the cache.
 	ids := f.cache.IDs()
 

frame.go

@@ -22,6 +22,7 @@ import (
 	"os"
 	"path/filepath"
 	"sort"
+	"strings"
 	"sync"
 	"time"
 
@@ -584,6 +585,12 @@ func (f *Frame) CreateViewIfNotExists(name string) (*View, error) {
 	}
 
 	view := f.newView(f.ViewPath(name), name)
+
+	// Never keep a cache for field views.
+	if strings.HasPrefix(name, ViewFieldPrefix) {
+		view.cacheType = CacheTypeNone
+	}
+
 	if err := view.Open(); err != nil {
 		return nil, err
 	}

index.go

@@ -454,8 +454,6 @@ func (i *Index) createFrame(name string, opt FrameOptions) (*Frame, error) {
 	if opt.RangeEnabled {
 		if opt.InverseEnabled {
 			return nil, ErrInverseRangeNotAllowed
-		} else if opt.CacheType != "" && opt.CacheType != CacheTypeNone {
-			return nil, ErrRangeCacheNotAllowed
 		}
 	} else {
 		if len(opt.Fields) > 0 {

index_test.go

@@ -136,14 +136,14 @@ func TestIndex_CreateFrame(t *testing.T) {
 			}
 		})
 
-		t.Run("ErrRangeCacheNotAllowed", func(t *testing.T) {
+		t.Run("ErrRangeCacheAllowed", func(t *testing.T) {
 			index := test.MustOpenIndex()
 			defer index.Close()
 
 			if _, err := index.CreateFrame("f", pilosa.FrameOptions{
 				RangeEnabled: true,
 				CacheType:    pilosa.CacheTypeRanked,
-			}); err != pilosa.ErrRangeCacheNotAllowed {
+			}); err != nil {
 				t.Fatal(err)
 			}
 		})

test/cluster.go

@@ -129,6 +129,30 @@ func (t *TestCluster) SetBit(index, frame, view string, rowID, colID uint64, x *
 	return nil
 }
 
+func (t *TestCluster) SetFieldValue(index, frame string, columnID uint64, name string, value int64) error {
+	// Determine which node should receive the SetFieldValue.
+	c0 := t.Clusters[0] // use the first node's cluster to determine slice location.
+	slice := columnID / pilosa.SliceWidth
+	nodes := c0.FragmentNodes(index, slice)
+
+	for _, node := range nodes {
+		c := t.clusterByURI(node.URI)
+		if c == nil {
+			continue
+		}
+		f := c.Holder.Frame(index, frame)
+		if f == nil {
+			return fmt.Errorf("index/frame does not exist: %s/%s", index, frame)
+		}
+		_, err := f.SetFieldValue(columnID, name, value)
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
 func (t *TestCluster) clusterByURI(uri pilosa.URI) *pilosa.Cluster {
 	for _, c := range t.Clusters {
 		if c.URI == uri {
@@ -155,7 +179,6 @@ func (t *TestCluster) AddNode(saveTopology bool) error {
 			URI:   c.URI,
 		}
 
-		//go coord.ReceiveEvent(ev)
 		if err := coord.ReceiveEvent(ev); err != nil {
 			return err
 		}
@@ -308,7 +331,10 @@ func (t *TestCluster) SendAsync(pb proto.Message) error {
 func (t *TestCluster) SendTo(to *pilosa.Node, pb proto.Message) error {
 	switch obj := pb.(type) {
 	case *internal.ResizeInstruction:
-		t.FollowResizeInstruction(obj)
+		err := t.FollowResizeInstruction(obj)
+		if err != nil {
+			return err
+		}
 	case *internal.ResizeInstructionComplete:
 		coord := t.clusterByURI(to.URI)
 		go coord.MarkResizeInstructionComplete(obj)
@@ -326,50 +352,58 @@ func (t *TestCluster) FollowResizeInstruction(instr *internal.ResizeInstruction)
 		Error: "",
 	}
 
-	// figure out which node it was meant for, then call the operation on that cluster
-	// basically need to mimic this: client.RetrieveSliceFromURI(context.Background(), src.Index, src.Frame, src.View, src.Slice, srcURI)
-	instrURI := pilosa.DecodeURI(instr.URI)
-	destCluster := t.clusterByURI(instrURI)
+	// Stop processing on any error.
+	if err := func() error {
 
-	// Sync the schema received in the resize instruction.
-	if err := destCluster.Holder.ApplySchema(instr.Schema); err != nil {
-		return err
-	}
+		// figure out which node it was meant for, then call the operation on that cluster
+		// basically need to mimic this: client.RetrieveSliceFromURI(context.Background(), src.Index, src.Frame, src.View, src.Slice, srcURI)
+		instrURI := pilosa.DecodeURI(instr.URI)
+		destCluster := t.clusterByURI(instrURI)
 
-	for _, src := range instr.Sources {
-		srcURI := pilosa.DecodeURI(src.URI)
-		srcCluster := t.clusterByURI(srcURI)
+		// Sync the schema received in the resize instruction.
+		if err := destCluster.Holder.ApplySchema(instr.Schema); err != nil {
+			return err
+		}
 
-		srcFragment := srcCluster.Holder.Fragment(src.Index, src.Frame, src.View, src.Slice)
-		destFragment := destCluster.Holder.Fragment(src.Index, src.Frame, src.View, src.Slice)
-		if destFragment == nil {
-			// Create fragment on destination if it doesn't exist.
-			f := destCluster.Holder.Frame(src.Index, src.Frame)
-			v := f.View(src.View)
-			var err error
-			destFragment, err = v.CreateFragmentIfNotExists(src.Slice)
-			if err != nil {
+		for _, src := range instr.Sources {
+			srcURI := pilosa.DecodeURI(src.URI)
+			srcCluster := t.clusterByURI(srcURI)
+
+			srcFragment := srcCluster.Holder.Fragment(src.Index, src.Frame, src.View, src.Slice)
+			destFragment := destCluster.Holder.Fragment(src.Index, src.Frame, src.View, src.Slice)
+			if destFragment == nil {
+				// Create fragment on destination if it doesn't exist.
+				f := destCluster.Holder.Frame(src.Index, src.Frame)
+				v := f.View(src.View)
+				var err error
+				destFragment, err = v.CreateFragmentIfNotExists(src.Slice)
+				if err != nil {
+					return err
+				}
+			}
+
+			buf := bytes.NewBuffer(nil)
+
+			bw := bufio.NewWriter(buf)
+			br := bufio.NewReader(buf)
+
+			// Get the fragment from source.
+			if _, err := srcFragment.WriteTo(bw); err != nil {
+				return err
+			}
+
+			// Flush the bufio.buf to the io.Writer (buf).
+			bw.Flush()
+
+			// Write data to destination.
+			if _, err := destFragment.ReadFrom(br); err != nil {
 				return err
 			}
 		}
 
-		buf := bytes.NewBuffer(nil)
-
-		bw := bufio.NewWriter(buf)
-		br := bufio.NewReader(buf)
-
-		// Get the fragment from source.
-		if _, err := srcFragment.WriteTo(bw); err != nil {
-			return err
-		}
-
-		// Flush the bufio.buf to the io.Writer (buf).
-		bw.Flush()
-
-		// Write data to destination.
-		if _, err := destFragment.ReadFrom(br); err != nil {
-			return err
-		}
+		return nil
+	}(); err != nil {
+		complete.Error = err.Error()
 	}
 
 	node := &pilosa.Node{