pass #1 top filler

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)

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

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 {

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
 }

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)

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 {

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