diff --git a/core/http.go b/core/http.go index d8171b0ef..83cb4c0b2 100644 --- a/core/http.go +++ b/core/http.go @@ -66,7 +66,13 @@ func (self *WebService) HandleQuery(w http.ResponseWriter, r *http.Request) { cluster := self.service.Cluster database := cluster.GetOrCreateDatabase("main") pql := string(body) - query.Execute(database, pql) + 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) + close(results_ch) } func (self *WebService) HandleStats(w http.ResponseWriter, r *http.Request) { diff --git a/core/service.go b/core/service.go index 2d3888778..33c47d572 100644 --- a/core/service.go +++ b/core/service.go @@ -25,6 +25,7 @@ type Service struct { ProcessMap *ProcessMap Transport interfaces.Transporter Dispatch interfaces.Dispatcher + Executor interfaces.Executorer WebService *WebService process_id *uuid.UUID Index *index.FragmentContainer @@ -79,6 +80,7 @@ func (service *Service) Run() { go service.WebService.Run() go service.Transport.Run() go service.Dispatch.Run() + go service.Executor.Run() sigterm, sighup := service.GetSignals() for { diff --git a/cruncher/cruncher.go b/cruncher/cruncher.go index 78b526eed..d11dc6159 100644 --- a/cruncher/cruncher.go +++ b/cruncher/cruncher.go @@ -3,6 +3,7 @@ package cruncher import ( "pilosa/core" "pilosa/dispatch" + "pilosa/executor" "pilosa/index" "pilosa/transport" @@ -25,6 +26,7 @@ func NewCruncher() *Cruncher { fragment_container := index.NewFragmentContainer() cruncher := Cruncher{service, make(chan bool), fragment_container} cruncher.Transport = transport.NewTcpTransport(service) - cruncher.Dispatch = dispatch.NewCruncherDispatch(service) + cruncher.Dispatch = dispatch.NewDispatch(service) + cruncher.Executor = executor.NewExecutor(service) return &cruncher } diff --git a/dispatch/dispatch.go b/dispatch/dispatch.go index c40ead3d9..74b03d747 100644 --- a/dispatch/dispatch.go +++ b/dispatch/dispatch.go @@ -7,20 +7,20 @@ import ( "github.com/davecgh/go-spew/spew" ) -type CruncherDispatch struct { +type Dispatch struct { service *core.Service } -func (self *CruncherDispatch) Init() error { +func (self *Dispatch) Init() error { log.Println("Starting Dispatcher") return nil } -func (self *CruncherDispatch) Close() { +func (self *Dispatch) Close() { log.Println("Shutting down Dispatcher") } -func (self *CruncherDispatch) Run() { +func (self *Dispatch) Run() { log.Println("Dispatch Run...") for { message := self.service.Transport.Receive() @@ -118,6 +118,6 @@ func (self *CruncherDispatch) Run() { } } -func NewCruncherDispatch(service *core.Service) *CruncherDispatch { - return &CruncherDispatch{service} +func NewDispatch(service *core.Service) *Dispatch { + return &Dispatch{service} } diff --git a/executor/executor.go b/executor/executor.go new file mode 100644 index 000000000..8e946e924 --- /dev/null +++ b/executor/executor.go @@ -0,0 +1,63 @@ +package executor + +import ( + "log" + "pilosa/core" + "pilosa/query" + + "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 +} + +func (self *Executor) Init() error { + log.Println("Starting Executor") + return nil +} + +func (self *Executor) Close() { + log.Println("Shutting down Executor") +} + +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 + for _, qs := range *query_plan { + spew.Dump(qs) + } + /* + spew.Dump("sleep...") + time.Sleep(4 * 1e9) + */ + // TODO: send the query steps out & wait for responses + qr := new(query.QueryResults) + qr.Data = 999 + job.results_ch <- qr +} + +func (self *Executor) Run() { + log.Println("Executor Run...") + for { + job := <-self.inbox + go self.executeJob(job) + } +} + +func NewExecutor(service *core.Service) *Executor { + return &Executor{service, make(chan *Job)} +} diff --git a/interfaces/core.go b/interfaces/core.go index 1143370bf..d48ad9c43 100644 --- a/interfaces/core.go +++ b/interfaces/core.go @@ -2,6 +2,7 @@ package interfaces import ( "pilosa/db" + "pilosa/query" ) type Transporter interface { @@ -16,3 +17,10 @@ type Dispatcher interface { Close() Run() } + +type Executorer interface { + Init() error + Close() + Run() + NewJob(*query.QueryPlan, chan *query.QueryResults) +} diff --git a/query/query.go b/query/query.go index 6c5d3c102..a0f276035 100644 --- a/query/query.go +++ b/query/query.go @@ -9,6 +9,10 @@ import ( type QueryInput interface{} +type QueryResults struct { + Data interface{} +} + type Query struct { Operation string Inputs []QueryInput //"strconv" @@ -17,7 +21,7 @@ type Query struct { Profile_id int } -func Execute(database *db.Database, pql string) { +func QueryPlanForPQL(database *db.Database, pql string) *QueryPlan { //spew.Dump("EXECUTE") //spew.Dump(pql) tokens := Lex(pql) @@ -40,4 +44,5 @@ func Execute(database *db.Database, pql string) { destination := db.Process{} query_plan := query_planner.Plan(query, id, &destination) spew.Dump(query_plan) + return query_plan }