scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-08-28 10:54:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

make profile_id uint64. handle Get/Set/Cat QuerySteps

Browse source
This commit is contained in:
travisturner 2014-01-06 20:18:41 -06:00
parent 0679ed17a2
commit 763a649f1a
12 changed files with 120 additions and 158 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
config/config.go
View file

@ -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
}

16
core/http.go
View file

@ -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) {

10
core/service.go
View file

@ -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 {

4
db/topology.go
View file

@ -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 {

4
dispatch/dispatch.go
View file

@ -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")

130
executor/executor.go
View file

@ -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)}
}

13
hold/hold.go
View file

@ -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()
}
*/

4
hold/hold_test.go
View file

@ -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")

7
interfaces/core.go
View file

@ -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)
}

6
query/parser.go
View file

@ -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--

62
query/planner.go
View file

@ -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
}

10
query/query.go
View file

@ -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
}