create Get/Count QueryResult as a db.HoldResult to return results. added ResultData() method to HoldResult interface

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

db/db.go

@@ -27,6 +27,7 @@ type Message struct {
 
 type HoldResult interface {
 	ResultId() *uuid.UUID
+	ResultData() interface{}
 }
 
 func init() {

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)

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)

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