diff --git a/core/ping.go b/core/ping.go index 0c0a9e1bb..5b2c25ec0 100644 --- a/core/ping.go +++ b/core/ping.go @@ -20,6 +20,9 @@ type PongRequest struct { func (self PongRequest) ResultId() *uuid.UUID { return self.Id } +func (self PongRequest) ResultData() interface{} { + return self.Id +} func init() { gob.Register(PingRequest{}) diff --git a/db/db.go b/db/db.go index 434c583c0..402cf4018 100644 --- a/db/db.go +++ b/db/db.go @@ -27,6 +27,7 @@ type Message struct { type HoldResult interface { ResultId() *uuid.UUID + ResultData() interface{} } func init() { diff --git a/dispatch/dispatch.go b/dispatch/dispatch.go index fe280cfed..96c8106bb 100644 --- a/dispatch/dispatch.go +++ b/dispatch/dispatch.go @@ -30,7 +30,7 @@ func (self *Dispatch) Run() { pong := db.Message{Data: core.PongRequest{Id: data.Id}} self.service.Transport.Send(&pong, data.Source) case db.HoldResult: - self.service.Hold.Set(data.ResultId(), data, 30) + self.service.Hold.Set(data.ResultId(), data.ResultData(), 30) case query.CatQueryStep, query.GetQueryStep, query.SetQueryStep, query.CountQueryStep: //fmt.Println("CAT/GET/SET QUERYSTEP") go self.service.Executor.NewJob(message) diff --git a/executor/executor.go b/executor/executor.go index 3af1a7752..05c3e1b51 100644 --- a/executor/executor.go +++ b/executor/executor.go @@ -34,13 +34,13 @@ func (self *Executor) NewJob(job *db.Message) { qs := job.Data.(query.CountQueryStep) input := qs.Input - bhi, _ := self.service.Hold.Get(input, 10) + value, _ := self.service.Hold.Get(input, 10) var bh index.BitmapHandle - switch bhi.(type) { + switch val := value.(type) { case index.BitmapHandle: - bh = bhi.(index.BitmapHandle) + bh = val case []byte: - bh, _ = self.service.Index.FromBytes(qs.Location.FragmentId, bhi.([]byte)) + bh, _ = self.service.Index.FromBytes(qs.Location.FragmentId, val) } count, err := self.service.Index.Count(qs.Location.FragmentId, bh) @@ -48,8 +48,9 @@ func (self *Executor) NewJob(job *db.Message) { spew.Dump(err) } spew.Dump("SLICE COUNT", count) - // TODO: instead of adding to the local hold, we need to send result to transport (which may go to a remote process's hold) - self.service.Hold.Set(qs.Id, count, 10) + + result_message := db.Message{Data: query.CountQueryResult{Id: qs.Id, Data: count}} + self.service.Transport.Send(&result_message, qs.Destination.ProcessId) case query.CatQueryStep: qs := job.Data.(query.CatQueryStep) @@ -60,11 +61,11 @@ func (self *Executor) NewJob(job *db.Message) { // either create a list of bitmap handles to cat (i.e. union), or sum the integer values for _, input := range qs.Inputs { value, _ := self.service.Hold.Get(input, 10) - switch value.(type) { + switch val := value.(type) { case index.BitmapHandle: - handles = append(handles, value.(index.BitmapHandle)) + handles = append(handles, val) case []byte: - bh, _ := self.service.Index.FromBytes(qs.Location.FragmentId, value.([]byte)) + bh, _ := self.service.Index.FromBytes(qs.Location.FragmentId, val) handles = append(handles, bh) case uint64: use_sum = true @@ -94,16 +95,19 @@ func (self *Executor) NewJob(job *db.Message) { if err != nil { spew.Dump(err) } + + var result interface{} if qs.LocIsDest() { - self.service.Hold.Set(qs.Id, bh, 10) + result = bh } else { bm, err := self.service.Index.GetBytes(qs.Location.FragmentId, bh) if err != nil { spew.Dump(err) } - // TODO: instead of adding to the local hold, we need to send result to transport (which may go to a remote process's hold) - self.service.Hold.Set(qs.Id, bm, 10) + result = bm } + result_message := db.Message{Data: query.GetQueryResult{Id: qs.Id, Data: result}} + self.service.Transport.Send(&result_message, qs.Destination.ProcessId) case query.SetQueryStep: qs := job.Data.(query.SetQueryStep) diff --git a/query/planner.go b/query/planner.go index 09941f202..6582c3623 100644 --- a/query/planner.go +++ b/query/planner.go @@ -33,6 +33,19 @@ type CountQueryStep struct { Destination *db.Location } +type CountQueryResult struct { + Id *uuid.UUID + Data interface{} +} + +func (self CountQueryResult) ResultId() *uuid.UUID { + return self.Id +} + +func (self CountQueryResult) ResultData() interface{} { + return self.Data +} + // QueryTree for COUNT queries type CountQueryTree struct { operation string @@ -65,6 +78,10 @@ func (self CatQueryResult) ResultId() *uuid.UUID { return self.Id } +func (self CatQueryResult) ResultData() interface{} { + return self.Data +} + // QueryTree for CAT queries type CatQueryTree struct { subqueries []QueryTree @@ -103,6 +120,18 @@ func (qs GetQueryStep) LocIsDest() bool { return false } +type GetQueryResult struct { + Id *uuid.UUID + Data interface{} +} + +func (self GetQueryResult) ResultId() *uuid.UUID { + return self.Id +} +func (self GetQueryResult) ResultData() interface{} { + return self.Data +} + // QueryTree for GET queries type GetQueryTree struct { bitmap *db.Bitmap @@ -151,6 +180,8 @@ func (qt *SetQueryTree) getLocation(d *db.Database) *db.Location { func init() { gob.Register(CatQueryResult{}) + gob.Register(GetQueryResult{}) + gob.Register(CountQueryResult{}) } ///////////////////////////////////////////////////////////////////////////////////////////////////