From fc5eb909aee2b37e96ecc5de9594a79f9966f66c Mon Sep 17 00:00:00 2001 From: travisturner Date: Tue, 7 Jan 2014 17:43:57 -0600 Subject: [PATCH] remove some debugging lines --- dispatch/dispatch.go | 11 ++---- executor/executor.go | 89 ++++++++++++++++++++++++++++---------------- query/planner.go | 33 +++++++++++++++- 3 files changed, 92 insertions(+), 41 deletions(-) diff --git a/dispatch/dispatch.go b/dispatch/dispatch.go index ea32b3b7b..59a46473e 100644 --- a/dispatch/dispatch.go +++ b/dispatch/dispatch.go @@ -1,13 +1,10 @@ package dispatch import ( - "fmt" "log" "pilosa/core" "pilosa/db" "pilosa/query" - - "github.com/davecgh/go-spew/spew" ) type Dispatch struct { @@ -27,18 +24,18 @@ func (self *Dispatch) Run() { log.Println("Dispatch Run...") for { message := self.service.Transport.Receive() - spew.Dump("Processing ", message) + //spew.Dump("Processing ", message) switch data := message.Data.(type) { case core.PingRequest: 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) - case query.CatQueryStep, query.GetQueryStep, query.SetQueryStep: - fmt.Println("CAT/GET/SET QUERYSTEP") + case query.CatQueryStep, query.GetQueryStep, query.SetQueryStep, query.CountQueryStep: + //fmt.Println("CAT/GET/SET QUERYSTEP") go self.service.Executor.NewJob(message) default: - log.Println("Unprocessed message", data) + //log.Println("Unprocessed message", data) } /* diff --git a/executor/executor.go b/executor/executor.go index 2503887f5..435d0987e 100644 --- a/executor/executor.go +++ b/executor/executor.go @@ -8,6 +8,7 @@ import ( "pilosa/index" "pilosa/query" "pilosa/util" + "tux21b.org/v1/gocql/uuid" "github.com/davecgh/go-spew/spew" ) @@ -27,20 +28,38 @@ func (self *Executor) Close() { } func (self *Executor) NewJob(job *db.Message) { - spew.Dump("NewJob") - spew.Dump(job.Data) switch job.Data.(type) { + case query.CountQueryStep: + spew.Dump("COUNT QUERYSTEP") + + qs := job.Data.(query.CountQueryStep) + input := qs.Input + bhi, _ := self.service.Hold.Get(input, 10) + var bh index.BitmapHandle + switch bhi.(type) { + case index.BitmapHandle: + bh = bhi.(index.BitmapHandle) + case []byte: + bh, _ = self.service.Index.FromBytes(qs.Location.FragmentId, bhi.([]byte)) + } + + count, err := self.service.Index.Count(qs.Location.FragmentId, bh) + if err != nil { + 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) + case query.CatQueryStep: qs := job.Data.(query.CatQueryStep) - fmt.Println("CAT QUERYSTEP") + spew.Dump("CAT QUERYSTEP") spew.Dump(qs) var bhs []index.BitmapHandle use_sum := false - sum := 0 + var sum uint64 for _, input := range qs.Inputs { - spew.Dump(input) bhi, _ := self.service.Hold.Get(input, 10) - spew.Dump(bhi) switch bhi.(type) { case index.BitmapHandle: spew.Dump("BH") @@ -49,47 +68,38 @@ func (self *Executor) NewJob(job *db.Message) { spew.Dump("RAW") bh, _ := self.service.Index.FromBytes(qs.Location.FragmentId, bhi.([]byte)) bhs = append(bhs, bh) - case int: + case uint64: use_sum = true - sum += bhi.(int) + sum += bhi.(uint64) } } if use_sum { spew.Dump("SUM", sum) + // 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, sum, 10) } else { - - spew.Dump("BHS", bhs) - unionbh, err := self.service.Index.Union(qs.Location.FragmentId, bhs) if err != nil { spew.Dump(err) } - spew.Dump("UNION BH", unionbh) - - /* - count, err := self.service.Index.Count(qs.Location.FragmentId, unionbh) - if err != nil { - spew.Dump(err) - } - spew.Dump("FINAL COUNT", count) - */ + unionbm, err := self.service.Index.GetBytes(qs.Location.FragmentId, unionbh) + 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, unionbm, 10) } case query.GetQueryStep: qs := job.Data.(query.GetQueryStep) - fmt.Println("GET QUERYSTEP") + spew.Dump("GET QUERYSTEP") - // perform get query with index bh, err := self.service.Index.Get(qs.Location.FragmentId, qs.Bitmap.Id) - //count, err := self.service.Index.Count(qs.Location.FragmentId, bh) if err != nil { spew.Dump(err) } - //spew.Dump("COUNT", count) - // push results to the map - if qs.LocIsDest() { self.service.Hold.Set(qs.Id, bh, 10) } else { @@ -97,12 +107,13 @@ func (self *Executor) NewJob(job *db.Message) { 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) } case query.SetQueryStep: qs := job.Data.(query.SetQueryStep) - fmt.Println("SET QUERYSTEP") + spew.Dump("SET QUERYSTEP") self.service.Index.SetBit(qs.Location.FragmentId, qs.Bitmap.Id, qs.ProfileId) default: fmt.Println("unknown") @@ -111,10 +122,7 @@ func (self *Executor) NewJob(job *db.Message) { } func (self *Executor) RunQuery(database_name string, pql string) { - log.Println("RunQuery: PQL") - database := self.service.Cluster.GetOrCreateDatabase(database_name) - process, err := self.service.GetProcess() if err != nil { spew.Dump(err) @@ -124,18 +132,33 @@ 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("-------------------------------------") + //return // loop over the query steps and send to Transport + var last_id *uuid.UUID 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 + } self.service.Transport.Push(msg) } - // TODO: add an entry to my execute map[key] that is waiting for the final result - //self.service.Hold.Get(??) - + // add an entry to my execute map[key] that is waiting for the final result + if last_id != nil { + final, err := self.service.Hold.Get(last_id, 10) + if err != nil { + spew.Dump(err) + } + spew.Dump("last_id", last_id) + spew.Dump("GRAND FINAL", final) + } } func (self *Executor) Run() { diff --git a/query/planner.go b/query/planner.go index d5e6439a6..6cbfac0f3 100644 --- a/query/planner.go +++ b/query/planner.go @@ -17,6 +17,15 @@ type QueryStep struct { destination *db.Location } +type CountQueryStep struct { + Id *uuid.UUID + Operation string + //Input QueryInput + Input *uuid.UUID + Location *db.Location + Destination *db.Location +} + type CatQueryStep struct { Id *uuid.UUID Operation string @@ -66,13 +75,20 @@ type QueryTree interface { getLocation(d *db.Database) *db.Location } -// QueryTree for UNION, INTER, and CAT queries +// QueryTree for UNION and INTERSECT queries type CompositeQueryTree struct { operation string subqueries []QueryTree location *db.Location } +// QueryTree for COUNT queries +type CountQueryTree struct { + operation string + subquery QueryTree + location *db.Location +} + // Randomly select location from subqueries (so subqueries roll up into composite queries while minimizing inter-node data traffic) func (qt *CompositeQueryTree) getLocation(d *db.Database) *db.Location { if qt.location == nil { @@ -129,6 +145,11 @@ func (qt *GetQueryTree) getLocation(d *db.Database) *db.Location { return fragment.GetLocation() } +// Uses consistent hashing function to select node containing data for GET operation +func (qt *CountQueryTree) getLocation(d *db.Database) *db.Location { + return qt.subquery.getLocation(d) +} + // Uses consistent hashing function to select node containing data for GET operation func (qt *SetQueryTree) getLocation(d *db.Database) *db.Location { slice, err := d.GetSliceForProfile(qt.profile_id) @@ -165,6 +186,9 @@ func (qp *QueryPlanner) buildTree(query *Query, slice int) QueryTree { if query.Operation == "get" { tree = &GetQueryTree{query.Inputs[0].(*db.Bitmap), slice} return tree + } else if query.Operation == "count" { + subquery := qp.buildTree(query.Inputs[0].(*Query), slice) + tree = &CountQueryTree{operation: query.Operation, subquery: subquery} } else { subqueries := make([]QueryTree, len(query.Inputs)) for i, input := range query.Inputs { @@ -207,6 +231,13 @@ func (qp *QueryPlanner) flatten(qt QueryTree, id *uuid.UUID, location *db.Locati step := SetQueryStep{id, "set", set.bitmap, set.profile_id, set.getLocation(qp.Database), location} plan := QueryPlan{step} return &plan + } else if cnt, ok := qt.(*CountQueryTree); ok { + sub_id := uuid.RandomUUID() + step := CountQueryStep{id, "count", &sub_id, cnt.getLocation(qp.Database), location} + step.Input = &sub_id + subq_steps := qp.flatten(cnt.subquery, &sub_id, cnt.getLocation(qp.Database)) + plan = append(plan, *subq_steps...) + plan = append(plan, step) } return &plan }