From 85f8034d9fc08bca3e0abb87fbf3d56512b6e5d1 Mon Sep 17 00:00:00 2001 From: travisturner Date: Mon, 3 Feb 2014 23:11:26 -0600 Subject: [PATCH] add in Top-N logic --- core/query.go | 80 ++++++++++++++++++++++++++++++------- executor/executor.go | 5 ++- index/brand.go | 2 +- index/fragment_container.go | 2 + query/planner.go | 35 ++++++++++++++++ query/query.go | 2 - 6 files changed, 106 insertions(+), 20 deletions(-) diff --git a/core/query.go b/core/query.go index 412242fda..13e922ccf 100644 --- a/core/query.go +++ b/core/query.go @@ -4,12 +4,13 @@ import ( "pilosa/db" "pilosa/index" "pilosa/query" + "sort" "github.com/davecgh/go-spew/spew" ) func (self *Service) CountQueryStepHandler(msg *db.Message) { - spew.Dump("COUNT QUERYSTEP") + //spew.Dump("COUNT QUERYSTEP") qs := msg.Data.(query.CountQueryStep) input := qs.Input value, _ := self.Hold.Get(input, 10) @@ -24,13 +25,33 @@ func (self *Service) CountQueryStepHandler(msg *db.Message) { if err != nil { spew.Dump(err) } - spew.Dump("SLICE COUNT", count) + //spew.Dump("SLICE COUNT", count) result_message := db.Message{Data: query.CountQueryResult{&query.BaseQueryResult{Id: qs.Id, Data: count}}} self.Transport.Send(&result_message, qs.Destination.ProcessId) } +func (self *Service) TopNQueryStepHandler(msg *db.Message) { + //spew.Dump("TOP-N QUERYSTEP") + qs := msg.Data.(query.TopNQueryStep) + input := qs.Input + value, _ := self.Hold.Get(input, 10) + var bh index.BitmapHandle + switch val := value.(type) { + case index.BitmapHandle: + bh = val + case []byte: + bh, _ = self.Index.FromBytes(qs.Location.FragmentId, val) + } + topn, err := self.Index.TopN(qs.Location.FragmentId, bh, 8) //TODO: get the N from the query (default is 8) + if err != nil { + spew.Dump(err) + } + result_message := db.Message{Data: query.TopNQueryResult{&query.BaseQueryResult{Id: qs.Id, Data: topn}}} + self.Transport.Send(&result_message, qs.Destination.ProcessId) +} + func (self *Service) UnionQueryStepHandler(msg *db.Message) { - spew.Dump("UNION QUERYSTEP") + //spew.Dump("UNION QUERYSTEP") qs := msg.Data.(query.UnionQueryStep) var handles []index.BitmapHandle // create a list of bitmap handles @@ -65,7 +86,7 @@ func (self *Service) UnionQueryStepHandler(msg *db.Message) { } func (self *Service) IntersectQueryStepHandler(msg *db.Message) { - spew.Dump("INTERSECT QUERYSTEP") + //spew.Dump("INTERSECT QUERYSTEP") qs := msg.Data.(query.IntersectQueryStep) var handles []index.BitmapHandle // create a list of bitmap handles @@ -101,10 +122,11 @@ func (self *Service) IntersectQueryStepHandler(msg *db.Message) { func (self *Service) CatQueryStepHandler(msg *db.Message) { qs := msg.Data.(query.CatQueryStep) - spew.Dump("CAT QUERYSTEP") + //spew.Dump("CAT QUERYSTEP") var handles []index.BitmapHandle - use_sum := false + return_type := "bitmap-handles" var sum uint64 + merge_map := map[uint64]uint64{} // either create a list of bitmap handles to cat (i.e. union), or sum the integer values for _, input := range qs.Inputs { value, _ := self.Hold.Get(input, 10) @@ -115,20 +137,47 @@ func (self *Service) CatQueryStepHandler(msg *db.Message) { bh, _ := self.Index.FromBytes(qs.Location.FragmentId, val) handles = append(handles, bh) case uint64: - use_sum = true - sum += value.(uint64) + return_type = "sum" + sum += val + case []index.Pair: + spew.Dump(val) + return_type = "pair-list" + for _, pair := range val { + _, ok := merge_map[pair.Key] + if ok { + merge_map[pair.Key] += pair.Count + } else { + merge_map[pair.Key] = pair.Count + } + } } } + // either return the sum, or return the compressed bitmap resulting from the cat (union) var result interface{} - if use_sum { + if return_type == "sum" { result = sum - } else { + } else if return_type == "bitmap-handles" { bh, err := self.Index.Union(qs.Location.FragmentId, handles) result, err = self.Index.GetBytes(qs.Location.FragmentId, bh) if err != nil { spew.Dump(err) } + } else if return_type == "pair-list" { + rank_list := make(index.RankList, 0, len(merge_map)) + for k, v := range merge_map { + rank := new(index.Rank) + rank.Pair = &index.Pair{k, v} + rank_list = append(rank_list, rank) + } + sort.Sort(rank_list) + pair_list := make([]index.Pair, 0, len(merge_map)) + for _, r := range rank_list { + pair_list = append(pair_list, *r.Pair) + } + result = pair_list[:8] //TODO: get the N from the query (default is 8) + } else { + result = "NONE" } result_message := db.Message{Data: query.CatQueryResult{&query.BaseQueryResult{Id: qs.Id, Data: result}}} self.Transport.Send(&result_message, qs.Destination.ProcessId) @@ -136,7 +185,7 @@ func (self *Service) CatQueryStepHandler(msg *db.Message) { func (self *Service) GetQueryStepHandler(msg *db.Message) { qs := msg.Data.(query.GetQueryStep) - spew.Dump("GET QUERYSTEP") + //spew.Dump("GET QUERYSTEP") bh, err := self.Index.Get(qs.Location.FragmentId, qs.Bitmap.Id) if err != nil { @@ -158,11 +207,12 @@ func (self *Service) GetQueryStepHandler(msg *db.Message) { } func (self *Service) SetQueryStepHandler(msg *db.Message) { - spew.Dump("SET QUERYSTEP") + //spew.Dump("SET QUERYSTEP") qs := msg.Data.(query.SetQueryStep) - result, err := self.Index.SetBit(qs.Location.FragmentId, qs.Bitmap.Id, qs.ProfileId) - spew.Dump("result:", result) - spew.Dump("err:", err) + //result, err := self.Index.SetBit(qs.Location.FragmentId, qs.Bitmap.Id, qs.ProfileId) + result, _ := self.Index.SetBit(qs.Location.FragmentId, qs.Bitmap.Id, qs.ProfileId) + //spew.Dump("result:", result) + //spew.Dump("err:", err) result_message := db.Message{Data: query.SetQueryResult{&query.BaseQueryResult{Id: qs.Id, Data: result}}} self.Transport.Send(&result_message, qs.Destination.ProcessId) diff --git a/executor/executor.go b/executor/executor.go index 9e07b9261..92f0651b0 100644 --- a/executor/executor.go +++ b/executor/executor.go @@ -30,6 +30,8 @@ func (self *Executor) NewJob(job *db.Message) { switch job.Data.(type) { case query.CountQueryStep: self.service.CountQueryStepHandler(job) + case query.TopNQueryStep: + self.service.TopNQueryStepHandler(job) case query.UnionQueryStep: self.service.UnionQueryStepHandler(job) case query.IntersectQueryStep: @@ -69,7 +71,6 @@ func (self *Executor) runQuery(database *db.Database, qry *query.Query) { fragment_id := util.SUUID(0) destination := db.Location{&process_id, fragment_id} - spew.Dump("QUERY.ID:", qry.Id) query_plan := query.QueryPlanForQuery(database, qry, &destination) // loop over the query steps and send to Transport for _, qs := range *query_plan { @@ -86,7 +87,7 @@ func (self *Executor) RunPQL(database_name string, pql string) interface{} { database := self.service.Cluster.GetOrCreateDatabase(database_name) // see if the outer query function is a custom query - reserved_functions := stringSlice{"get", "set", "union", "intersect", "count"} + reserved_functions := stringSlice{"get", "set", "union", "intersect", "count", "top-n"} tokens := query.Lex(pql) outer_token := tokens[0].Text diff --git a/index/brand.go b/index/brand.go index d8f8ed8f2..fc64c6256 100644 --- a/index/brand.go +++ b/index/brand.go @@ -175,7 +175,7 @@ func (self *Brand) Store(bitmap_id uint64, bm IBitmap) { func (self *Brand) TopN(src_bitmap IBitmap, n int) []Pair { self.rank_counter = 0 - self.Rank() //TERRIBLE REMOVE TIS ASAP + self.Rank() // TODO: TERRIBLE REMOVE THIS ASAP is := new(IntSet) return self.TopNCat(src_bitmap, n, is) } diff --git a/index/fragment_container.go b/index/fragment_container.go index ad3337f11..9f3aa43ac 100644 --- a/index/fragment_container.go +++ b/index/fragment_container.go @@ -26,6 +26,8 @@ type BitmapHandle uint64 func init() { var vh BitmapHandle gob.Register(vh) + var lp []Pair + gob.Register(lp) } func (self *FragmentContainer) Shutdown() { diff --git a/query/planner.go b/query/planner.go index b0ee427ff..ac36a12b5 100644 --- a/query/planner.go +++ b/query/planner.go @@ -88,6 +88,29 @@ func (qt *CountQueryTree) getLocation(d *db.Database) *db.Location { return qt.subquery.getLocation(d) } +/////////////////////////////////////////////////////////////////////////////////////////////////// +// TOP-N +/////////////////////////////////////////////////////////////////////////////////////////////////// +type TopNQueryStep struct { + *BaseQueryStep + Input *uuid.UUID +} + +type TopNQueryResult struct { + *BaseQueryResult +} + +// QueryTree for TOP-N queries +type TopNQueryTree struct { + subquery QueryTree + location *db.Location +} + +// Uses consistent hashing function to select node containing data for GET operation +func (qt *TopNQueryTree) getLocation(d *db.Database) *db.Location { + return qt.subquery.getLocation(d) +} + /////////////////////////////////////////////////////////////////////////////////////////////////// // UNION /////////////////////////////////////////////////////////////////////////////////////////////////// @@ -248,6 +271,7 @@ func init() { gob.Register(UnionQueryResult{}) gob.Register(IntersectQueryResult{}) gob.Register(CountQueryResult{}) + gob.Register(TopNQueryResult{}) gob.Register(SetQueryStep{}) gob.Register(GetQueryStep{}) @@ -255,6 +279,7 @@ func init() { gob.Register(UnionQueryStep{}) gob.Register(IntersectQueryStep{}) gob.Register(CountQueryStep{}) + gob.Register(TopNQueryStep{}) } /////////////////////////////////////////////////////////////////////////////////////////////////// @@ -308,6 +333,7 @@ func (qp *QueryPlanner) buildTree(query *Query, slice int) QueryTree { panic(err) } for slice := 0; slice < numSlices; slice++ { + //for slice := 0; slice < 3; slice++ { subtree := qp.buildTree(query, slice) composite := tree.(*CatQueryTree) composite.subqueries = append(composite.subqueries, subtree) @@ -319,6 +345,9 @@ func (qp *QueryPlanner) buildTree(query *Query, slice int) QueryTree { } else if query.Operation == "count" { subquery := qp.buildTree(query.Inputs[0].(*Query), slice) tree = &CountQueryTree{subquery: subquery} + } else if query.Operation == "top-n" { + subquery := qp.buildTree(query.Inputs[0].(*Query), slice) + tree = &TopNQueryTree{subquery: subquery} } else if query.Operation == "union" { subqueries := make([]QueryTree, len(query.Inputs)) for i, input := range query.Inputs { @@ -399,6 +428,12 @@ func (qp *QueryPlanner) flatten(qt QueryTree, id *uuid.UUID, location *db.Locati subq_steps := qp.flatten(cnt.subquery, &sub_id, cnt.getLocation(qp.Database)) plan = append(plan, *subq_steps...) plan = append(plan, step) + } else if topn, ok := qt.(*TopNQueryTree); ok { + sub_id := uuid.RandomUUID() + step := &TopNQueryStep{&BaseQueryStep{id, "top-n", topn.getLocation(qp.Database), location}, &sub_id} + subq_steps := qp.flatten(topn.subquery, &sub_id, topn.getLocation(qp.Database)) + plan = append(plan, *subq_steps...) + plan = append(plan, step) } return &plan } diff --git a/query/query.go b/query/query.go index 3fd1b4a89..e865e430b 100644 --- a/query/query.go +++ b/query/query.go @@ -3,7 +3,6 @@ package query import ( "pilosa/db" "strings" - "github.com/davecgh/go-spew/spew" "tux21b.org/v1/gocql/uuid" ) @@ -65,7 +64,6 @@ func QueryPlanForQuery(database *db.Database, query *Query, destination *db.Loca func TokensToString(tokens []Token) string { var str []string for i, _ := range tokens { - spew.Dump(tokens[i].Text) str = append(str, tokens[i].Text) } // for now, we're just using this function to pull the filter out of the outer function "outerfunc(filter)"