diff --git a/core/http.go b/core/http.go index 4a7e2bfdf..533931558 100644 --- a/core/http.go +++ b/core/http.go @@ -7,7 +7,6 @@ import ( "net/http" "pilosa/config" "pilosa/db" - "pilosa/query" "strconv" "github.com/davecgh/go-spew/spew" @@ -31,6 +30,7 @@ func (self *WebService) Run() { mux.HandleFunc("/info", self.HandleInfo) mux.HandleFunc("/processes", self.HandleProcesses) mux.HandleFunc("/listen", self.HandleListen) + mux.HandleFunc("/test", self.HandleTest) s := &http.Server{ Addr: ":" + port_string, Handler: mux, @@ -63,17 +63,26 @@ func (self *WebService) HandleQuery(w http.ResponseWriter, r *http.Request) { http.Error(w, "Error reading POST data", http.StatusBadRequest) return } - cluster := self.service.Cluster - database := cluster.GetOrCreateDatabase("main") - pql := string(body) - query_plan := query.QueryPlanForPQL(database, pql) - 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) + // 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) { @@ -98,6 +107,22 @@ func (self *WebService) HandleInfo(w http.ResponseWriter, r *http.Request) { spew.Fdump(w, self.service.Cluster) } +func (self *WebService) HandleTest(w http.ResponseWriter, r *http.Request) { + if r.Method != "GET" { + http.Error(w, "Only GET allowed", http.StatusMethodNotAllowed) + return + } + spew.Dump("TEST!") + + msg := new(db.Message) + msg.Data = "mystring" + self.service.Transport.Push(msg) + + msg2 := new(db.Message) + msg2.Data = 789 + self.service.Transport.Push(msg2) +} + func (self *WebService) HandleProcesses(w http.ResponseWriter, r *http.Request) { if r.Method != "GET" { http.Error(w, "Only GET allowed", http.StatusMethodNotAllowed) diff --git a/core/service.go b/core/service.go index 33c47d572..b6c182046 100644 --- a/core/service.go +++ b/core/service.go @@ -95,3 +95,7 @@ func (service *Service) Run() { } } } + +type Message interface { + Handle(*Service) +} diff --git a/db/db.go b/db/db.go index f49216985..0433ace2e 100644 --- a/db/db.go +++ b/db/db.go @@ -1,7 +1,20 @@ package db +type Message struct { + Data interface{} `json:data` +} + +/* +import "pilosa/core" + +type Message interface { + Handle(*core.Service) +} + + type Message struct { Key string `json:key` Data interface{} `json:data` Destination Location } +*/ diff --git a/db/topology.go b/db/topology.go index 7ccd0a5e4..52c99edb7 100644 --- a/db/topology.go +++ b/db/topology.go @@ -5,8 +5,6 @@ import ( "fmt" "log" "pilosa/util" - "strconv" - "strings" "sync" "github.com/nu7hatch/gouuid" @@ -18,10 +16,18 @@ var SliceDoesNotExistError = errors.New("Slice does not exist.") var FragmentDoesNotExistError = errors.New("Fragment does not exist.") var FrameSliceIntersectDoesNotExistError = errors.New("FrameSliceIntersect does not exist.") +/* type Location struct { Ip string Port int } +*/ +type Location struct { + ProcessId *uuid.UUID + FragmentId util.SUUID +} + +type ProcessId *uuid.UUID type Process struct { id *uuid.UUID @@ -83,6 +89,7 @@ func (self *Process) SetPortHttp(port int) { self.port_http = port } +/* // Create a Location struct given a string in form "0.0.0.0:0" func NewLocation(location_string string) (*Location, error) { splitstring := strings.Split(location_string, ":") @@ -103,6 +110,7 @@ func (location *Location) ToString() string { // Map of node location to their router type NodeMap map[Location]Location +*/ /////////// CLUSTERS //////////////////////////////////////////////////////////////////// @@ -291,10 +299,22 @@ type Fragment struct { process *Process } +func (f *Fragment) GetId() util.SUUID { + return f.id +} + func (f *Fragment) GetProcess() *Process { return f.process } +func (f *Fragment) GetProcessId() *uuid.UUID { + return f.process.id +} + +func (f *Fragment) GetLocation() *Location { + return &Location{f.process.id, f.id} +} + // rename this one func (d *Database) GetFragmentForBitmap(slice *Slice, bitmap *Bitmap) (*Fragment, error) { //d.mutex.Lock() @@ -388,6 +408,6 @@ func (d *Database) GetSliceForProfile(profile_id int) (*Slice, error) { } type Bitmap struct { - Id int + Id uint64 FrameType string } diff --git a/dispatch/dispatch.go b/dispatch/dispatch.go index 74b03d747..cf55b0ead 100644 --- a/dispatch/dispatch.go +++ b/dispatch/dispatch.go @@ -1,8 +1,10 @@ package dispatch import ( + "fmt" "log" "pilosa/core" + "pilosa/query" "github.com/davecgh/go-spew/spew" ) @@ -25,9 +27,16 @@ func (self *Dispatch) Run() { for { message := self.service.Transport.Receive() log.Println("Processing ", message) - spew.Dump(message.Key) spew.Dump(message.Data) + switch message.Data.(type) { + case query.GetQueryStep, query.SetQueryStep: + fmt.Println("GET/SET QUERYSTEP") + self.service.Executor.NewJob(message) + default: + fmt.Println("unknown") + } + /* path := message.Data.(string) diff --git a/executor/executor.go b/executor/executor.go index 8e946e924..9bf992480 100644 --- a/executor/executor.go +++ b/executor/executor.go @@ -1,21 +1,29 @@ package executor import ( + "fmt" "log" "pilosa/core" + "pilosa/db" "pilosa/query" + "time" + "tux21b.org/v1/gocql/uuid" "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 *Job + inbox chan *db.Message + qs_chan chan *query.QueryStep + hold map[*uuid.UUID]chan *query.QueryResults } func (self *Executor) Init() error { @@ -27,37 +35,138 @@ func (self *Executor) Close() { log.Println("Shutting down Executor") } +func (self *Executor) NewJob(job *db.Message) { + //j := Job{qp, results} + //self.inbox <- &j + spew.Dump("NewJob") + spew.Dump(job.Data) + // TODO: switch on job.Data type + switch job.Data.(type) { + 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) + 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) + + //self.Set(job.Data.Id, query_results) + case query.SetQueryStep: + fmt.Println("SET QUERYSTEP") + //self.Get() + default: + fmt.Println("unknown") + + } +} + +/* func (self *Executor) NewJob(qp *query.QueryPlan, results chan *query.QueryResults) { j := Job{qp, results} self.inbox <- &j } +*/ -//func (self *Executor) ExecuteJob(qp *query.QueryPlan) { -func (self *Executor) executeJob(job *Job) { - log.Println("Executor: ExecuteJob") - //spew.Dump(job) - query_plan := job.query_plan +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) + + // 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() + + time.Sleep(4 * 1e9) /* - spew.Dump("sleep...") - 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) + */ - // TODO: send the query steps out & wait for responses - qr := new(query.QueryResults) - qr.Data = 999 - job.results_ch <- qr + +} + +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 { - job := <-self.inbox - go self.executeJob(job) - } + /* + 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 *Job)} + return &Executor{service, make(chan *db.Message), make(chan *query.QueryStep), make(map[*uuid.UUID]chan *query.QueryResults)} } diff --git a/interfaces/core.go b/interfaces/core.go index d48ad9c43..040dbf482 100644 --- a/interfaces/core.go +++ b/interfaces/core.go @@ -10,6 +10,7 @@ type Transporter interface { Close() Send(*db.Message) Receive() *db.Message + Push(*db.Message) } type Dispatcher interface { @@ -22,5 +23,8 @@ type Executorer interface { Init() error Close() Run() - NewJob(*query.QueryPlan, chan *query.QueryResults) + //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 9a3ade22d..cf3ea4fa8 100644 --- a/query/parser.go +++ b/query/parser.go @@ -14,23 +14,29 @@ func (qp *QueryParser) walkInputs(tokens []Token) ([]QueryInput, int) { // BITMAP if tokens[0].Type == TYPE_ID { // TODO: look for frame type in the tokens list - bitmap_id, err := strconv.Atoi(tokens[0].Text) + b, err := strconv.Atoi(tokens[0].Text) + bitmap_id := uint64(b) if err != nil { panic(err) } // if the next 2 tokens are comma-frame, then we have a frame, else set to a default frame_type := "general" - if len(tokens) > 2 && tokens[2].Type == TYPE_FRAME { - frame_type = tokens[2].Text - } - bm := db.Bitmap{bitmap_id, frame_type} profile_id := 0 - if len(tokens) > 4 && tokens[4].Type == TYPE_PROFILE { + if len(tokens) > 4 && tokens[2].Type == TYPE_FRAME && tokens[4].Type == TYPE_PROFILE { + frame_type = tokens[2].Text profile_id, err = strconv.Atoi(tokens[4].Text) if err != nil { panic(err) } + } else if len(tokens) > 2 && tokens[2].Type == TYPE_FRAME { + frame_type = tokens[2].Text + } else if len(tokens) > 2 && tokens[2].Type == TYPE_PROFILE { + profile_id, err = strconv.Atoi(tokens[2].Text) + if err != nil { + panic(err) + } } + bm := db.Bitmap{bitmap_id, frame_type} return []QueryInput{&bm}, profile_id } diff --git a/query/planner.go b/query/planner.go index a73e71221..9c4bceaf9 100644 --- a/query/planner.go +++ b/query/planner.go @@ -10,37 +10,56 @@ import ( // A single step in the query plan. type QueryStep struct { - id uuid.UUID - operation string - inputs []QueryInput - location *db.Process - return_process *db.Process + id uuid.UUID + operation string + inputs []QueryInput + location *db.Location + destination *db.Location +} + +type GetQueryStep struct { + Id uuid.UUID + Operation string + Bitmap *db.Bitmap + Slice int + Location *db.Location + Destination *db.Location +} + +type SetQueryStep struct { + Id uuid.UUID + Operation string + Bitmap *db.Bitmap + ProfileId int + Location *db.Location + Destination *db.Location } 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.return_process) + return fmt.Sprintf("%s %s %s, LOC: %s, DEST: %s", q.operation, q.id.String(), q.inputs, q.location, q.destination) } // This is the output of the query planner. Contains a list of steps which can be performed in parallel -type QueryPlan []QueryStep +//type QueryPlan []QueryStep +type QueryPlan []interface{} type QueryPlanner struct { Database *db.Database } type QueryTree interface { - getLocation(d *db.Database) *db.Process + getLocation(d *db.Database) *db.Location } // QueryTree for UNION, INTER, and CAT queries type CompositeQueryTree struct { operation string subqueries []QueryTree - location *db.Process + 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.Process { +func (qt *CompositeQueryTree) getLocation(d *db.Database) *db.Location { if qt.location == nil { subqueryLength := len(qt.subqueries) if subqueryLength > 0 { @@ -65,23 +84,23 @@ type SetQueryTree struct { } // Uses consistent hashing function to select node containing data for GET operation -func (qt *GetQueryTree) getLocation(d *db.Database) *db.Process { +func (qt *GetQueryTree) getLocation(d *db.Database) *db.Location { slice := d.GetOrCreateSlice(qt.slice) // TODO: this should probably be just GetSlice (no create) fragment, err := d.GetFragmentForBitmap(slice, qt.bitmap) if err != nil { panic(err) } - return fragment.GetProcess() + return fragment.GetLocation() } // Uses consistent hashing function to select node containing data for GET operation -func (qt *SetQueryTree) getLocation(d *db.Database) *db.Process { +func (qt *SetQueryTree) getLocation(d *db.Database) *db.Location { slice, err := d.GetSliceForProfile(qt.profile_id) fragment, err := d.GetFragmentForBitmap(slice, qt.bitmap) if err != nil { panic(err) } - return fragment.GetProcess() + return fragment.GetLocation() } // Builds QueryTree object from Query. Pass slice=-1 to perform operation on all slices @@ -122,7 +141,7 @@ func (qp *QueryPlanner) buildTree(query *Query, slice int) QueryTree { } // Produces flattened QueryPlan from QueryTree input -func (qp *QueryPlanner) flatten(qt QueryTree, id *uuid.UUID, location *db.Process) *QueryPlan { +func (qp *QueryPlanner) flatten(qt QueryTree, id *uuid.UUID, location *db.Location) *QueryPlan { plan := QueryPlan{} if composite, ok := qt.(*CompositeQueryTree); ok { inputs := make([]QueryInput, len(composite.subqueries)) @@ -136,11 +155,11 @@ func (qp *QueryPlanner) flatten(qt QueryTree, id *uuid.UUID, location *db.Proces } plan = append(plan, step) } else if get, ok := qt.(*GetQueryTree); ok { - step := QueryStep{*id, "get", []QueryInput{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 := QueryStep{*id, "set", []QueryInput{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 } @@ -148,7 +167,7 @@ func (qp *QueryPlanner) flatten(qt QueryTree, id *uuid.UUID, location *db.Proces } // Transforms Query into QueryTree and flattens to QueryPlan object -func (qp *QueryPlanner) Plan(query *Query, id *uuid.UUID, destination *db.Process) *QueryPlan { +func (qp *QueryPlanner) Plan(query *Query, id *uuid.UUID, destination *db.Location) *QueryPlan { queryTree := qp.buildTree(query, -1) return qp.flatten(queryTree, id, destination) } diff --git a/query/planner_test.go b/query/planner_test.go index e173ae0ed..89481b56b 100644 --- a/query/planner_test.go +++ b/query/planner_test.go @@ -53,13 +53,14 @@ func TestQueryPlanner(t *testing.T) { fragment2.SetProcess(process2) qplanner := QueryPlanner{Database: database} - destination := db.Process{} + destination := fragment1.GetLocation() id, _ := uuid.NewV4() - qp := qplanner.Plan(&query, id, &destination) + qp := qplanner.Plan(&query, id, destination) for i, qs := range *qp { - spew.Dump(i, qs, qs.inputs) + //spew.Dump(i, qs, qs.inputs) + spew.Dump(i, qs) spew.Dump("**************************************************************") } }) diff --git a/query/query.go b/query/query.go index 20f4e6fd3..96f4ff38d 100644 --- a/query/query.go +++ b/query/query.go @@ -2,6 +2,9 @@ package query import ( "pilosa/db" + "pilosa/util" + + "github.com/davecgh/go-spew/spew" "github.com/nu7hatch/gouuid" ) @@ -20,26 +23,20 @@ type Query struct { } func QueryPlanForPQL(database *db.Database, pql string) *QueryPlan { - //spew.Dump("EXECUTE") - //spew.Dump(pql) tokens := Lex(pql) - //spew.Dump(tokens) - query_parser := QueryParser{} query, err := query_parser.Parse(tokens) + spew.Dump("-------------------------------------") + spew.Dump(query) + spew.Dump("-------------------------------------") if err != nil { panic(err) } - //spew.Dump(query) - - // switch on different query types: - //if query.Operation == "set" { - //spew.Dump("SET!!") - //} - query_planner := QueryPlanner{Database: database} id, _ := uuid.NewV4() - destination := db.Process{} + process_id, _ := uuid.NewV4() + fragment_id := util.SUUID(1) + destination := db.Location{process_id, fragment_id} query_plan := query_planner.Plan(query, id, &destination) return query_plan } diff --git a/transport/tcp.go b/transport/tcp.go index 34e73d5ad..e29707937 100644 --- a/transport/tcp.go +++ b/transport/tcp.go @@ -29,6 +29,10 @@ func (self *TcpTransport) Receive() *db.Message { return <-self.inbox } +func (self *TcpTransport) Push(message *db.Message) { + self.inbox <- message +} + func NewTcpTransport(service *core.Service) *TcpTransport { return &TcpTransport{config.GetInt("port_tcp"), make(chan *db.Message), nil} }