From 763a649f1aaff4701f421d9e4c06277225f0ee6c Mon Sep 17 00:00:00 2001 From: travisturner Date: Mon, 6 Jan 2014 20:18:41 -0600 Subject: [PATCH] make profile_id uint64. handle Get/Set/Cat QuerySteps --- config/config.go | 12 ---- core/http.go | 16 ------ core/service.go | 10 +++- db/topology.go | 4 +- dispatch/dispatch.go | 4 +- executor/executor.go | 130 ++++++++++++------------------------------- hold/hold.go | 13 ++++- hold/hold_test.go | 4 ++ interfaces/core.go | 7 +-- query/parser.go | 6 +- query/planner.go | 62 +++++++++++++++++---- query/query.go | 10 +--- 12 files changed, 120 insertions(+), 158 deletions(-) diff --git a/config/config.go b/config/config.go index daccb2b02..9cd3419e3 100644 --- a/config/config.go +++ b/config/config.go @@ -8,7 +8,6 @@ import ( "sync" "launchpad.net/goyaml" - "tux21b.org/v1/gocql/uuid" ) type Config struct { @@ -106,14 +105,3 @@ func (self *Config) GetString(key string) string { } return "" } - -func GetUUID(key string) *uuid.UUID { - value, ok := GetSafe(key) - if ok { - value_uuid, err := uuid.ParseUUID(value.(string)) - if err == nil { - return &value_uuid - } - } - return nil -} diff --git a/core/http.go b/core/http.go index 533931558..bcacb6c20 100644 --- a/core/http.go +++ b/core/http.go @@ -63,26 +63,10 @@ func (self *WebService) HandleQuery(w http.ResponseWriter, r *http.Request) { http.Error(w, "Error reading POST data", http.StatusBadRequest) return } - // TODO: we need to get the database name from the query string (for now, hard-coded) database_name := "main" - //cluster := self.service.Cluster - //database := cluster.GetOrCreateDatabase(database_name) - pql := string(body) self.service.Executor.RunQuery(database_name, pql) - - /* - query_plan := query.QueryPlanForPQL(database, pql) - spew.Dump(query_plan) - - results_ch := make(chan *query.QueryResults) - self.service.Executor.NewJob(query_plan, results_ch) - results := <-results_ch - spew.Dump("Results") - spew.Dump(results) - close(results_ch) - */ } func (self *WebService) HandleStats(w http.ResponseWriter, r *http.Request) { diff --git a/core/service.go b/core/service.go index 133df3e1b..e34a5dd4d 100644 --- a/core/service.go +++ b/core/service.go @@ -6,6 +6,7 @@ import ( "os/signal" "pilosa/config" "pilosa/db" + "pilosa/hold" "pilosa/index" "pilosa/interfaces" "syscall" @@ -27,8 +28,8 @@ type Service struct { Dispatch interfaces.Dispatcher Executor interfaces.Executorer WebService *WebService - process_id *uuid.UUID Index *index.FragmentContainer + Hold *hold.Holder } func NewService() *Service { @@ -41,8 +42,8 @@ func NewService() *Service { service.ProcessMapper = NewProcessMapper(service, "/pilosa/0") service.ProcessMap = NewProcessMap() service.WebService = NewWebService(service) - service.process_id = config.GetUUID("process_id") service.Index = index.NewFragmentContainer() + service.Hold = hold.NewHolder() return service } @@ -65,6 +66,10 @@ func (service *Service) init_id() { service.Id = &id } +func (self *Service) GetProcess() (*db.Process, error) { + return self.ProcessMap.GetProcess(self.Id) +} + func (service *Service) GetSignals() (chan os.Signal, chan os.Signal) { hupChan := make(chan os.Signal, 1) termChan := make(chan os.Signal, 1) @@ -81,6 +86,7 @@ func (service *Service) Run() { go service.Transport.Run() go service.Dispatch.Run() go service.Executor.Run() + go service.Hold.Run() sigterm, sighup := service.GetSignals() for { diff --git a/db/topology.go b/db/topology.go index 10bf97273..32ed4ab86 100644 --- a/db/topology.go +++ b/db/topology.go @@ -394,9 +394,9 @@ func (f *Fragment) SetProcess(process *Process) { /////////////////////////////////////////////////////////////////////////////////////////////// // Get a slice from a database -func (d *Database) GetSliceForProfile(profile_id int) (*Slice, error) { +func (d *Database) GetSliceForProfile(profile_id uint64) (*Slice, error) { slice_id := profile_id / SLICE_WIDTH - return d.getSlice(slice_id) + return d.getSlice(int(slice_id)) } type Bitmap struct { diff --git a/dispatch/dispatch.go b/dispatch/dispatch.go index df48a07a3..12eba7f04 100644 --- a/dispatch/dispatch.go +++ b/dispatch/dispatch.go @@ -30,8 +30,8 @@ func (self *Dispatch) Run() { spew.Dump(message.Data) switch message.Data.(type) { - case query.GetQueryStep, query.SetQueryStep: - fmt.Println("GET/SET QUERYSTEP") + case query.CatQueryStep, query.GetQueryStep, query.SetQueryStep: + fmt.Println("CAT/GET/SET QUERYSTEP") go self.service.Executor.NewJob(message) default: fmt.Println("unknown") diff --git a/executor/executor.go b/executor/executor.go index 9e026ace2..9ded6e7a2 100644 --- a/executor/executor.go +++ b/executor/executor.go @@ -5,24 +5,16 @@ import ( "log" "pilosa/core" "pilosa/db" + "pilosa/index" "pilosa/query" - "time" - "tux21b.org/v1/gocql/uuid" + "pilosa/util" "github.com/davecgh/go-spew/spew" ) -/* -type Job struct { - query_plan *query.QueryPlan - results_ch chan *query.QueryResults -} -*/ - type Executor struct { service *core.Service inbox chan *db.Message - qs_chan chan *query.QueryStep } func (self *Executor) Init() error { @@ -38,125 +30,77 @@ func (self *Executor) NewJob(job *db.Message) { spew.Dump("NewJob") spew.Dump(job.Data) switch job.Data.(type) { + case query.CatQueryStep: + qs := job.Data.(query.CatQueryStep) + fmt.Println("CAT QUERYSTEP") + spew.Dump(qs) + for _, input := range qs.Inputs { + spew.Dump(input) + bh := self.service.Hold.Get(input).(index.BitmapHandle) + // TODO: git rid of this count, need the cat to do a sum() or a true cat() + count, err := self.service.Index.Count(qs.Location.FragmentId, bh) + if err != nil { + spew.Dump(err) + } + spew.Dump("COUNT", count) + } case query.GetQueryStep: qs := job.Data.(query.GetQueryStep) fmt.Println("GET QUERYSTEP") + // perform get query with index - // push results to the map - //query_results = ??? - - // TEMP - profile_id := uint64(7899) - self.service.Index.SetBit(qs.Location.FragmentId, qs.Bitmap.Id, profile_id) - // END TEMP - - spew.Dump("COUNT") 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) } - count, err := self.service.Index.Count(qs.Location.FragmentId, bh) - if err != nil { - spew.Dump(err) - } - spew.Dump(count) - spew.Dump(qs.Id) + //spew.Dump("COUNT", count) + // push results to the map + self.service.Hold.Set(qs.Id, bh) - //self.Set(qs.Id, count) case query.SetQueryStep: + qs := job.Data.(query.SetQueryStep) fmt.Println("SET QUERYSTEP") - //self.Get() + self.service.Index.SetBit(qs.Location.FragmentId, qs.Bitmap.Id, qs.ProfileId) default: fmt.Println("unknown") } } -func (self *Executor) NewQS(qs *query.QueryStep) { - self.qs_chan <- qs -} - func (self *Executor) RunQuery(database_name string, pql string) { log.Println("RunQuery: PQL") - spew.Dump(pql) database := self.service.Cluster.GetOrCreateDatabase(database_name) - query_plan := query.QueryPlanForPQL(database, pql) - //spew.Dump(query_plan) + + process, err := self.service.GetProcess() + if err != nil { + spew.Dump(err) + } + process_id := process.Id() + fragment_id := util.SUUID(0) + destination := db.Location{&process_id, fragment_id} + + query_plan := query.QueryPlanForPQL(database, pql, &destination) + spew.Dump(query_plan) // loop over the query steps and send to Transport for _, qs := range *query_plan { - spew.Dump(qs) msg := new(db.Message) msg.Data = qs self.service.Transport.Push(msg) } + // TODO: add an entry to my execute map[key] that is waiting for the final result - self.Get() + //self.service.Hold.Get(??) - time.Sleep(4 * 1e9) - /* - results_ch := make(chan *query.QueryResults) - self.service.Executor.NewJob(query_plan, results_ch) - results := <-results_ch - spew.Dump("Results") - spew.Dump(results) - close(results_ch) - - */ - -} - -func (self *Executor) Get() { - log.Println("Executor: Get") -} - -func (self *Executor) Set(id *uuid.UUID, results *query.QueryResults) { - log.Println("Executor: Set") -} - -func (self *Executor) executeQueryPlan(job *db.Message) { - log.Println("Executor: ExecuteJob") - - /* - //spew.Dump(job) - query_plan := job.query_plan - for _, qs := range *query_plan { - spew.Dump(qs) - //query.HandleQueryStep(&qs) - //res, err := self.service.Process.SetBit(fragment_id, bitmaps[0], profile_id) - - msg := new(db.Message) - msg.Data = qs - self.service.Transport.Push(msg) - - } - // TODO: send the query steps out & wait for responses - qr := new(query.QueryResults) - qr.Data = 999 - job.results_ch <- qr - */ -} - -func (self *Executor) executeQS(qs *query.QueryStep) { - spew.Dump("EXEC QS") } func (self *Executor) Run() { log.Println("Executor Run...") - /* - for { - select { - case job := <-self.inbox: - go self.executeQueryPlan(job) - case qs := <-self.qs_chan: - go self.executeQS(qs) - } - } - */ } func NewExecutor(service *core.Service) *Executor { - return &Executor{service, make(chan *db.Message), make(chan *query.QueryStep)} + return &Executor{service, make(chan *db.Message)} } diff --git a/hold/hold.go b/hold/hold.go index 87586f1aa..5b0dbe55f 100644 --- a/hold/hold.go +++ b/hold/hold.go @@ -17,7 +17,7 @@ type Holder struct { delchan chan delhold } -var Hold Holder +//var Hold Holder func (self *Holder) DelChan(id *uuid.UUID) { req := delhold{id} @@ -44,7 +44,7 @@ func (self *Holder) Set(id *uuid.UUID, value interface{}) { }() } -func (self *Holder) run() { +func (self *Holder) Run() { var greq gethold var dreq delhold for { @@ -62,7 +62,14 @@ func (self *Holder) run() { } } +func NewHolder() *Holder { + h := Holder{make(map[uuid.UUID]holdchan), make(chan gethold), make(chan delhold)} + return &h +} + +/* func init() { Hold = Holder{make(map[uuid.UUID]holdchan), make(chan gethold), make(chan delhold)} - go Hold.run() + go Hold.Run() } +*/ diff --git a/hold/hold_test.go b/hold/hold_test.go index e8bfd4214..e3348604b 100644 --- a/hold/hold_test.go +++ b/hold/hold_test.go @@ -9,6 +9,10 @@ import ( ) func TestHoldChan(t *testing.T) { + + Hold := Holder{make(map[uuid.UUID]holdchan), make(chan gethold), make(chan delhold)} + go Hold.Run() + Convey("set then get", t, func() { id := uuid.RandomUUID() Hold.Set(&id, "derp") diff --git a/interfaces/core.go b/interfaces/core.go index 040dbf482..e916f1c53 100644 --- a/interfaces/core.go +++ b/interfaces/core.go @@ -1,9 +1,6 @@ package interfaces -import ( - "pilosa/db" - "pilosa/query" -) +import "pilosa/db" type Transporter interface { Run() @@ -23,8 +20,6 @@ type Executorer interface { Init() error Close() Run() - //NewJob(*query.QueryPlan, chan *query.QueryResults) NewJob(*db.Message) - NewQS(*query.QueryStep) RunQuery(string, string) } diff --git a/query/parser.go b/query/parser.go index cf3ea4fa8..43f63b673 100644 --- a/query/parser.go +++ b/query/parser.go @@ -10,7 +10,7 @@ var InvalidQueryError = errors.New("Invalid query format.") type QueryParser struct{} -func (qp *QueryParser) walkInputs(tokens []Token) ([]QueryInput, int) { +func (qp *QueryParser) walkInputs(tokens []Token) ([]QueryInput, uint64) { // BITMAP if tokens[0].Type == TYPE_ID { // TODO: look for frame type in the tokens list @@ -37,7 +37,7 @@ func (qp *QueryParser) walkInputs(tokens []Token) ([]QueryInput, int) { } } bm := db.Bitmap{bitmap_id, frame_type} - return []QueryInput{&bm}, profile_id + return []QueryInput{&bm}, uint64(profile_id) } // LIST OF QUERIES @@ -86,7 +86,7 @@ func (qp *QueryParser) walk(tokens []Token) (*Query, error) { } else if tokens[i].Type == TYPE_RP { if open_parens == 0 { if i == len(tokens)-1 { - q.Inputs, q.Profile_id = qp.walkInputs(tokens[2:i]) + q.Inputs, q.ProfileId = qp.walkInputs(tokens[2:i]) } } else { open_parens-- diff --git a/query/planner.go b/query/planner.go index 43177c3a0..3c13560d0 100644 --- a/query/planner.go +++ b/query/planner.go @@ -10,15 +10,23 @@ import ( // A single step in the query plan. type QueryStep struct { - id uuid.UUID + id *uuid.UUID operation string inputs []QueryInput location *db.Location destination *db.Location } +type CatQueryStep struct { + Id *uuid.UUID + Operation string + Inputs []*uuid.UUID + Location *db.Location + Destination *db.Location +} + type GetQueryStep struct { - Id uuid.UUID + Id *uuid.UUID Operation string Bitmap *db.Bitmap Slice int @@ -27,10 +35,10 @@ type GetQueryStep struct { } type SetQueryStep struct { - Id uuid.UUID + Id *uuid.UUID Operation string Bitmap *db.Bitmap - ProfileId int + ProfileId uint64 Location *db.Location Destination *db.Location } @@ -71,6 +79,12 @@ func (qt *CompositeQueryTree) getLocation(d *db.Database) *db.Location { return qt.location } +// QueryTree for CAT queries +type CatQueryTree struct { + subqueries []QueryTree + location *db.Location +} + // QueryTree for GET queries type GetQueryTree struct { bitmap *db.Bitmap @@ -80,7 +94,22 @@ type GetQueryTree struct { // QueryTree for SET queries type SetQueryTree struct { bitmap *db.Bitmap - profile_id int + profile_id uint64 +} + +// Uses consistent hashing function to select node containing data for GET operation +func (qt *CatQueryTree) getLocation(d *db.Database) *db.Location { + //loc := new(db.Location) + //return loc + if qt.location == nil { + subqueryLength := len(qt.subqueries) + if subqueryLength > 0 { + locationIndex := rand.Intn(subqueryLength) + subquery := qt.subqueries[locationIndex] + qt.location = subquery.getLocation(d) + } + } + return qt.location } // Uses consistent hashing function to select node containing data for GET operation @@ -109,20 +138,20 @@ func (qp *QueryPlanner) buildTree(query *Query, slice int) QueryTree { // handle SET operation regardless of the slice if query.Operation == "set" { - tree = &SetQueryTree{query.Inputs[0].(*db.Bitmap), query.Profile_id} + tree = &SetQueryTree{query.Inputs[0].(*db.Bitmap), query.ProfileId} return tree } // handle the remaining operations, taking slice into consideration if slice == -1 { - tree = &CompositeQueryTree{operation: "cat"} + tree = &CatQueryTree{} numSlices, err := qp.Database.NumSlices() if err != nil { panic(err) } for slice := 0; slice < numSlices; slice++ { subtree := qp.buildTree(query, slice) - composite := tree.(*CompositeQueryTree) + composite := tree.(*CatQueryTree) composite.subqueries = append(composite.subqueries, subtree) } } else { @@ -145,21 +174,30 @@ func (qp *QueryPlanner) flatten(qt QueryTree, id *uuid.UUID, location *db.Locati plan := QueryPlan{} if composite, ok := qt.(*CompositeQueryTree); ok { inputs := make([]QueryInput, len(composite.subqueries)) - step := QueryStep{*id, composite.operation, inputs, composite.getLocation(qp.Database), location} + step := QueryStep{id, composite.operation, inputs, composite.getLocation(qp.Database), location} for index, subq := range composite.subqueries { sub_id := uuid.RandomUUID() - // this is the "wait" step step.inputs[index] = &sub_id subq_steps := qp.flatten(subq, &sub_id, composite.getLocation(qp.Database)) plan = append(plan, *subq_steps...) } plan = append(plan, step) + } else if cat, ok := qt.(*CatQueryTree); ok { + inputs := make([]*uuid.UUID, len(cat.subqueries)) + step := CatQueryStep{id, "cat", inputs, cat.getLocation(qp.Database), location} + for index, subq := range cat.subqueries { + sub_id := uuid.RandomUUID() + step.Inputs[index] = &sub_id + subq_steps := qp.flatten(subq, &sub_id, cat.getLocation(qp.Database)) + plan = append(plan, *subq_steps...) + } + plan = append(plan, step) } else if get, ok := qt.(*GetQueryTree); ok { - step := GetQueryStep{*id, "get", get.bitmap, get.slice, get.getLocation(qp.Database), location} + step := GetQueryStep{id, "get", get.bitmap, get.slice, get.getLocation(qp.Database), location} plan := QueryPlan{step} return &plan } else if set, ok := qt.(*SetQueryTree); ok { - step := SetQueryStep{*id, "set", set.bitmap, set.profile_id, set.getLocation(qp.Database), location} + step := SetQueryStep{id, "set", set.bitmap, set.profile_id, set.getLocation(qp.Database), location} plan := QueryPlan{step} return &plan } diff --git a/query/query.go b/query/query.go index b9cb72aa6..695cb0969 100644 --- a/query/query.go +++ b/query/query.go @@ -2,7 +2,6 @@ package query import ( "pilosa/db" - "pilosa/util" "github.com/davecgh/go-spew/spew" "tux21b.org/v1/gocql/uuid" @@ -19,10 +18,10 @@ type Query struct { Inputs []QueryInput //"strconv" // Represents a parsed query. Inputs can be Query or Bitmap objects // Maybe Bitmap and Query objects should have different fields to avoid using interface{} - Profile_id int + ProfileId uint64 } -func QueryPlanForPQL(database *db.Database, pql string) *QueryPlan { +func QueryPlanForPQL(database *db.Database, pql string, destination *db.Location) *QueryPlan { tokens := Lex(pql) query_parser := QueryParser{} query, err := query_parser.Parse(tokens) @@ -34,9 +33,6 @@ func QueryPlanForPQL(database *db.Database, pql string) *QueryPlan { } query_planner := QueryPlanner{Database: database} id := uuid.RandomUUID() - process_id := uuid.RandomUUID() - fragment_id := util.SUUID(1) - destination := db.Location{&process_id, fragment_id} - query_plan := query_planner.Plan(query, &id, &destination) + query_plan := query_planner.Plan(query, &id, destination) return query_plan }