From 7e797efdc23f231352e9645452bbdd2f792ab4c2 Mon Sep 17 00:00:00 2001 From: Travis Turner Date: Mon, 6 Nov 2017 14:32:20 -0600 Subject: [PATCH] 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{