scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-07 00:55:55 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

add in Top-N logic

Browse source
This commit is contained in:
travisturner 2014-02-03 23:11:26 -06:00
parent 391e707649
commit 85f8034d9f
6 changed files with 106 additions and 20 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

80
core/query.go
View file

@ -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)

5
executor/executor.go
View file

@ -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

2
index/brand.go
View file

@ -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)
}

2
index/fragment_container.go
View file

@ -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() {

35
query/planner.go
View file

@ -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
}

2
query/query.go
View file

@ -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)"