From bc62bc47e9b10f5af84c09274696c6b7da0a3b19 Mon Sep 17 00:00:00 2001 From: travisturner Date: Wed, 8 Jan 2014 15:08:54 -0600 Subject: [PATCH] PortableQueryStep interface to help transfer query-steps across processes --- dispatch/dispatch.go | 2 +- executor/executor.go | 20 ++++++++-------- query/planner.go | 56 +++++++++++++++++++++++++++++++++++++++++++- 3 files changed, 66 insertions(+), 12 deletions(-) diff --git a/dispatch/dispatch.go b/dispatch/dispatch.go index 0e0a3cb2e..3b8754419 100644 --- a/dispatch/dispatch.go +++ b/dispatch/dispatch.go @@ -30,7 +30,7 @@ func (self *Dispatch) Run() { self.service.Transport.Send(&pong, data.Source) case db.HoldResult: self.service.Hold.Set(data.ResultId(), data.ResultData(), 30) - case query.CatQueryStep, query.GetQueryStep, query.SetQueryStep, query.CountQueryStep, query.UnionQueryStep, query.IntersectQueryStep: + case query.PortableQueryStep: go self.service.Executor.NewJob(message) default: log.Println("Unprocessed message", data) diff --git a/executor/executor.go b/executor/executor.go index 1e46a328d..efd7ec943 100644 --- a/executor/executor.go +++ b/executor/executor.go @@ -8,6 +8,7 @@ import ( "pilosa/index" "pilosa/query" "pilosa/util" + "reflect" "tux21b.org/v1/gocql/uuid" "github.com/davecgh/go-spew/spew" @@ -31,7 +32,6 @@ func (self *Executor) NewJob(job *db.Message) { switch job.Data.(type) { case query.CountQueryStep: spew.Dump("COUNT QUERYSTEP") - qs := job.Data.(query.CountQueryStep) input := qs.Input value, _ := self.service.Hold.Get(input, 10) @@ -198,9 +198,9 @@ func (self *Executor) RunQuery(database_name string, pql string) { destination := db.Location{&process_id, fragment_id} query_plan := query.QueryPlanForPQL(database, pql, &destination) - spew.Dump("--------query_plan-------------------") - spew.Dump(query_plan) - spew.Dump("-------------------------------------") + //spew.Dump("--------query_plan-------------------") + //spew.Dump(query_plan) + //spew.Dump("-------------------------------------") //return // loop over the query steps and send to Transport @@ -208,12 +208,13 @@ func (self *Executor) RunQuery(database_name string, pql string) { for _, qs := range *query_plan { msg := new(db.Message) msg.Data = qs - spew.Dump("qs", qs) switch step := qs.(type) { - case query.CatQueryStep: - last_id = step.Id + case query.PortableQueryStep: + self.service.Transport.Send(msg, step.GetLocation().ProcessId) + if reflect.TypeOf(step) != reflect.TypeOf(query.SetQueryStep{}) { + last_id = step.GetId() + } } - self.service.Transport.Push(msg) } // add an entry to my execute map[key] that is waiting for the final result @@ -222,9 +223,8 @@ func (self *Executor) RunQuery(database_name string, pql string) { if err != nil { spew.Dump(err) } - spew.Dump("last_id", last_id) spew.Dump("*******************************************************") - spew.Dump("GRAND FINAL", final) + spew.Dump("FINAL", final) spew.Dump("*******************************************************") } } diff --git a/query/planner.go b/query/planner.go index 333fa03a3..018770070 100644 --- a/query/planner.go +++ b/query/planner.go @@ -22,6 +22,11 @@ func (q QueryStep) StringHOLD() string { return fmt.Sprintf("%s %s %s, LOC: %s, DEST: %s", q.operation, q.id.String(), q.inputs, q.location, q.destination) } +type PortableQueryStep interface { + GetId() *uuid.UUID + GetLocation() *db.Location +} + /////////////////////////////////////////////////////////////////////////////////////////////////// // COUNT /////////////////////////////////////////////////////////////////////////////////////////////////// @@ -33,6 +38,13 @@ type CountQueryStep struct { Destination *db.Location } +func (self CountQueryStep) GetId() *uuid.UUID { + return self.Id +} +func (self CountQueryStep) GetLocation() *db.Location { + return self.Location +} + type CountQueryResult struct { Id *uuid.UUID Data interface{} @@ -68,6 +80,13 @@ type UnionQueryStep struct { Destination *db.Location } +func (self UnionQueryStep) GetId() *uuid.UUID { + return self.Id +} +func (self UnionQueryStep) GetLocation() *db.Location { + return self.Location +} + func (qs UnionQueryStep) LocIsDest() bool { if qs.Location.ProcessId == qs.Destination.ProcessId && qs.Location.FragmentId == qs.Destination.FragmentId { return true @@ -118,6 +137,13 @@ type IntersectQueryStep struct { Destination *db.Location } +func (self IntersectQueryStep) GetId() *uuid.UUID { + return self.Id +} +func (self IntersectQueryStep) GetLocation() *db.Location { + return self.Location +} + func (qs IntersectQueryStep) LocIsDest() bool { if qs.Location.ProcessId == qs.Destination.ProcessId && qs.Location.FragmentId == qs.Destination.FragmentId { return true @@ -168,6 +194,13 @@ type CatQueryStep struct { Destination *db.Location } +func (self CatQueryStep) GetId() *uuid.UUID { + return self.Id +} +func (self CatQueryStep) GetLocation() *db.Location { + return self.Location +} + type CatQueryResult struct { Id *uuid.UUID Data interface{} @@ -212,6 +245,13 @@ type GetQueryStep struct { Destination *db.Location } +func (self GetQueryStep) GetId() *uuid.UUID { + return self.Id +} +func (self GetQueryStep) GetLocation() *db.Location { + return self.Location +} + func (qs GetQueryStep) LocIsDest() bool { if qs.Location.ProcessId == qs.Destination.ProcessId && qs.Location.FragmentId == qs.Destination.FragmentId { return true @@ -259,6 +299,13 @@ type SetQueryStep struct { Destination *db.Location } +func (self SetQueryStep) GetId() *uuid.UUID { + return self.Id +} +func (self SetQueryStep) GetLocation() *db.Location { + return self.Location +} + // QueryTree for SET queries type SetQueryTree struct { bitmap *db.Bitmap @@ -278,11 +325,18 @@ func (qt *SetQueryTree) getLocation(d *db.Database) *db.Location { /////////////////////////////////////////////////////////////////////////////////////////////////// func init() { + gob.Register(GetQueryResult{}) gob.Register(CatQueryResult{}) gob.Register(UnionQueryResult{}) gob.Register(IntersectQueryResult{}) - gob.Register(GetQueryResult{}) gob.Register(CountQueryResult{}) + + gob.Register(SetQueryStep{}) + gob.Register(GetQueryStep{}) + gob.Register(CatQueryStep{}) + gob.Register(UnionQueryStep{}) + gob.Register(IntersectQueryStep{}) + gob.Register(CountQueryStep{}) } ///////////////////////////////////////////////////////////////////////////////////////////////////