featurebase/executor/executor.go

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 *db.Message
	qs_chan chan *query.QueryStep
	hold    map[*uuid.UUID]chan *query.QueryResults
}

func (self *Executor) Init() error {
	log.Println("Starting Executor")
	return nil
}

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) 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)
	/*
		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), make(map[*uuid.UUID]chan *query.QueryResults)}
}