From 512ccaf3b9de7d0cca9c491ba2cef358a79eaf3a Mon Sep 17 00:00:00 2001 From: Todd Gruben Date: Mon, 11 Aug 2014 13:49:29 -0500 Subject: [PATCH] pass #1 top filler --- core/topn.go | 27 +++++++++++++++++---------- dispatch/dispatch.go | 5 +++++ index/brand.go | 4 ++++ index/commands.go | 15 ++++++++++----- index/fragment_container.go | 9 ++++++++- index/general.go | 5 +++++ query/planner.go | 1 + 7 files changed, 50 insertions(+), 16 deletions(-) diff --git a/core/topn.go b/core/topn.go index 08ef2b047..2ef0c0cc9 100644 --- a/core/topn.go +++ b/core/topn.go @@ -64,7 +64,6 @@ func newtask(p util.GUID) *Task { } func (t *Task) Add(frag util.SUUID, bitmap_id uint64, handle index.BitmapHandle) { - spew.Dump(t) fa, ok := t.f[frag] if !ok { fa = index.FillArgs{frag, handle, make([]uint64, 0, 0)} @@ -105,9 +104,11 @@ func (self *Service) TopFillHandler(msg *db.Message) { //in order for this to ge if err != nil { log.Println("TopFileHandler:", err) } - sendmsg := new(db.Message) - sendmsg.Data = query.BaseQueryResult{Id: &topfill.QueryId, Data: topn} - self.Transport.Send(sendmsg, &topfill.ReturnProcessId) + // sendmsg := new(db.Message) + // sendmsg.Data = query.BaseQueryResult{Id: &topfill.QueryId, Data: topn} + //sendmsg := db.Message{Data: &query.BaseQueryResult{Id: &topfill.QueryId, Data: topn}} + //self.Transport.Send(&sendmsg, &topfill.ReturnProcessId) + self.Hold.Set(&topfill.QueryId, topn, 30) } func SendRequest(process_id util.GUID, t *Task, service *Service) { @@ -132,8 +133,15 @@ func GatherResults(tasks map[util.GUID]*Task, service *Service) map[uint64]uint6 answers := make(chan []index.Pair) for _, task := range tasks { go func(id util.GUID) { - value, _ := service.Hold.Get(&id, 10) //eiher need to be the frame process or the handler process? - answers <- value.([]index.Pair) + value, err := service.Hold.Get(&id, 10) //eiher need to be the frame process or the handler process? + if value == nil { + log.Println("Bad TopN Result:", err) + empty := make([]index.Pair, 0, 0) + answers <- empty + + } else { + answers <- value.([]index.Pair) + } }(task.hold_id) } for i := 0; i < len(tasks); i++ { @@ -155,13 +163,11 @@ type TopNPackage struct { func init() { gob.Register(TopNPackage{}) + gob.Register(TopFill{}) } func (self *Service) TopNQueryStepHandler(msg *db.Message) { - //spew.Dump("TOP-N QUERYSTEP") qs := msg.Data.(query.TopNQueryStep) - //spew.Dump(qs) - //need categories in qs I just added so it would compile var categoryleaves []uint64 input := qs.Input value, _ := self.Hold.Get(input, 10) @@ -185,7 +191,8 @@ func (self *Service) TopNQueryStepHandler(msg *db.Message) { topnPackage := TopNPackage{*qs.Location.ProcessId, qs.Location.FragmentId, topn, bh} if err != nil { - spew.Dump(err) + log.Println(spew.Sdump(err)) + } result_message := db.Message{Data: query.TopNQueryResult{&query.BaseQueryResult{Id: qs.Id, Data: topnPackage}}} self.Transport.Send(&result_message, qs.Destination.ProcessId) diff --git a/dispatch/dispatch.go b/dispatch/dispatch.go index 72aadccdc..a010b7cf9 100644 --- a/dispatch/dispatch.go +++ b/dispatch/dispatch.go @@ -5,6 +5,8 @@ import ( "pilosa/core" "pilosa/db" "pilosa/query" + + "github.com/davecgh/go-spew/spew" ) type Dispatch struct { @@ -36,7 +38,10 @@ func (self *Dispatch) Run() { self.service.Hold.Set(data.ResultId(), data.ResultData(), 30) case query.PortableQueryStep: go self.service.Executor.NewJob(message) + case core.TopFill: + go self.service.TopFillHandler(message) default: + spew.Dump(data) log.Println("Unprocessed message", data) } } diff --git a/index/brand.go b/index/brand.go index 82945d1d2..ba9954fde 100644 --- a/index/brand.go +++ b/index/brand.go @@ -64,6 +64,10 @@ func (self *Brand) Clear() bool { self.bitmap_cache = make(map[uint64]*Rank) return true } +func (self *Brand) Exists(bitmap_id uint64) bool { + _, ok := self.bitmap_cache[bitmap_id] + return ok +} func (self *Brand) Get(bitmap_id uint64) IBitmap { bm, ok := self.bitmap_cache[bitmap_id] if ok { diff --git a/index/commands.go b/index/commands.go index 7169c8d7c..a999c3ff8 100644 --- a/index/commands.go +++ b/index/commands.go @@ -302,12 +302,17 @@ func NewTopFill(a FillArgs) *CmdTopFill { } func (self *CmdTopFill) Execute(f *Fragment) Calculation { - result := make([]Pair, len(self.args.Bitmaps)) + result := make([]Pair, 0) for _, v := range self.args.Bitmaps { - a := f.NewHandle(v) - res := f.intersect([]BitmapHandle{self.args.Handle, a}) - bm, _ := f.getBitmap(res) - result = append(result, Pair{v, BitCount(bm)}) + if f.exists(v) { + a := f.NewHandle(v) + res := f.intersect([]BitmapHandle{self.args.Handle, a}) + bm, _ := f.getBitmap(res) + bc := BitCount(bm) + if bc > 0 { + result = append(result, Pair{v, bc}) + } + } } return result } diff --git a/index/fragment_container.go b/index/fragment_container.go index 27dc3b040..6acb5e9e9 100644 --- a/index/fragment_container.go +++ b/index/fragment_container.go @@ -187,7 +187,9 @@ func (self *FragmentContainer) TopFillBatch(args []FillArgs) ([]Pair, error) { } ret_val := make([]Pair, len(results)) for k, v := range results { - ret_val = append(ret_val, Pair{k, v}) + if v > 0 { //don't include 0 size items + ret_val = append(ret_val, Pair{k, v}) + } } return ret_val, nil @@ -288,6 +290,7 @@ type Pilosa interface { Stats() interface{} Persist() error Load(requestChan chan Command, fragment *Fragment) + Exists(id uint64) bool } type Fragment struct { @@ -341,6 +344,10 @@ func (self *Fragment) getBitmap(bitmap BitmapHandle) (IBitmap, bool) { return bm.(IBitmap), ok } +func (self *Fragment) exists(bitmap_id uint64) bool { + return self.impl.Exists(bitmap_id) +} + func (self *Fragment) TopN(bitmap BitmapHandle, n int, categories []uint64) []Pair { bm, ok := self.cache.Get(bitmap) diff --git a/index/general.go b/index/general.go index 4842b29fa..65b1dd1cb 100644 --- a/index/general.go +++ b/index/general.go @@ -37,6 +37,11 @@ func (self *General) Clear() bool { return true } +func (self *General) Exists(bitmap_id uint64) bool { + _, ok := self.bitmap_cache.Get(bitmap_id) + return ok + +} func (self *General) Get(bitmap_id uint64) IBitmap { bm, ok := self.bitmap_cache.Get(bitmap_id) if ok { diff --git a/query/planner.go b/query/planner.go index 5d2584f4d..df441f48d 100644 --- a/query/planner.go +++ b/query/planner.go @@ -356,6 +356,7 @@ func (qt *SetQueryTree) getLocation(d *db.Database) (*db.Location, error) { /////////////////////////////////////////////////////////////////////////////////////////////////// func init() { + gob.Register(BaseQueryResult{}) gob.Register(SetQueryResult{}) gob.Register(GetQueryResult{}) gob.Register(CatQueryResult{})