From 0d5a2e7bfd4470d5cf4ab7c4a3bcccda05266dce Mon Sep 17 00:00:00 2001 From: Travis Turner Date: Mon, 6 Nov 2017 10:43:55 -0600 Subject: [PATCH 1/3] add SetFieldValue() method to TestCluster and use in tests --- cluster_test.go | 40 ++++++++++++++++++++++++++++++++++++++-- test/cluster.go | 24 ++++++++++++++++++++++++ 2 files changed, 62 insertions(+), 2 deletions(-) diff --git a/cluster_test.go b/cluster_test.go index a51823993..7667aa24d 100644 --- a/cluster_test.go +++ b/cluster_test.go @@ -499,11 +499,30 @@ func TestCluster_ResizeStates(t *testing.T) { t.Fatal(err) } - // Add Data to node0. + // Add Bit Data to node0. tc.CreateFrame("i", "f", pilosa.FrameOptions{}) tc.SetBit("i", "f", "standard", 1, 101, nil) tc.SetBit("i", "f", "standard", 1, 1300000, nil) + // Add Field Data to node0. + 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, + }, + }, + }) + 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) @@ -528,6 +547,7 @@ func TestCluster_ResizeStates(t *testing.T) { t.Errorf("expected node1 topology: %v, but got: %v", expectedTop, node1.Topology) } + // 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") @@ -540,7 +560,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. diff --git a/test/cluster.go b/test/cluster.go index e73289827..248b61325 100644 --- a/test/cluster.go +++ b/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 { From 7e797efdc23f231352e9645452bbdd2f792ab4c2 Mon Sep 17 00:00:00 2001 From: Travis Turner Date: Mon, 6 Nov 2017 14:32:20 -0600 Subject: [PATCH 2/3] Allow CacheType to be set for a RangeEnabled frame (to apply to the standard frame) --- cluster_test.go | 16 ++++++--- fragment.go | 5 +++ frame.go | 7 ++++ index.go | 2 -- index_test.go | 12 ------- test/cluster.go | 88 +++++++++++++++++++++++++++---------------------- 6 files changed, 72 insertions(+), 58 deletions(-) diff --git a/cluster_test.go b/cluster_test.go index 7667aa24d..911742a27 100644 --- a/cluster_test.go +++ b/cluster_test.go @@ -500,15 +500,17 @@ func TestCluster_ResizeStates(t *testing.T) { } // Add Bit Data to node0. - tc.CreateFrame("i", "f", pilosa.FrameOptions{}) + 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. - tc.CreateFrame("i", "fields", pilosa.FrameOptions{ + if err := tc.CreateFrame("i", "fields", pilosa.FrameOptions{ InverseEnabled: false, RangeEnabled: true, - CacheType: pilosa.CacheTypeNone, + //CacheType: pilosa.CacheTypeNone, Fields: []*pilosa.Field{ { Name: "fld0", @@ -517,14 +519,18 @@ func TestCluster_ResizeStates(t *testing.T) { 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] diff --git a/fragment.go b/fragment.go index 172fc22a3..745d1d613 100644 --- a/fragment.go +++ b/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() diff --git a/frame.go b/frame.go index c57edf38e..58920ccc2 100644 --- a/frame.go +++ b/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 } diff --git a/index.go b/index.go index 734ea7ed5..be6ae031c 100644 --- a/index.go +++ b/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 { diff --git a/index_test.go b/index_test.go index 077b2ded6..7dfec3fb9 100644 --- a/index_test.go +++ b/index_test.go @@ -136,18 +136,6 @@ func TestIndex_CreateFrame(t *testing.T) { } }) - t.Run("ErrRangeCacheNotAllowed", 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 { - t.Fatal(err) - } - }) - t.Run("RangeEnabledWithCacheTypeNone", func(t *testing.T) { index := test.MustOpenIndex() defer index.Close() diff --git a/test/cluster.go b/test/cluster.go index 248b61325..ce0563eba 100644 --- a/test/cluster.go +++ b/test/cluster.go @@ -179,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 } @@ -320,7 +319,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) @@ -338,50 +340,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{ From fee1b5583888263d2a990b1ba84a947c4b1c879a Mon Sep 17 00:00:00 2001 From: Travis Turner Date: Mon, 6 Nov 2017 16:03:23 -0600 Subject: [PATCH 3/3] replace test ErrRangeCacheNotAllowed with ErrRangeCacheAllowed --- index_test.go | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/index_test.go b/index_test.go index 7dfec3fb9..5e5dce218 100644 --- a/index_test.go +++ b/index_test.go @@ -136,6 +136,18 @@ func TestIndex_CreateFrame(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 != nil { + t.Fatal(err) + } + }) + t.Run("RangeEnabledWithCacheTypeNone", func(t *testing.T) { index := test.MustOpenIndex() defer index.Close()