From d7a3ecd70af20f4de563da2d8bddcf12e0e86e29 Mon Sep 17 00:00:00 2001 From: travisturner Date: Fri, 27 Dec 2013 11:13:37 -0600 Subject: [PATCH] query parser and planner (PQL string > lex > tokens > parser > query > planner > query plan) --- core/etcd.go | 6 +-- core/http.go | 6 +-- db/topology.go | 11 +++-- query/lexer.go | 23 ++++++++- query/lexer_test.go | 54 ++++++++++++++------ query/parser.go | 112 ++++++++++++++++++++++++++---------------- query/parser_test.go | 21 ++++++++ query/planner.go | 62 +++++++++++------------ query/planner_test.go | 59 ++++++++++++++++++++-- 9 files changed, 248 insertions(+), 106 deletions(-) create mode 100644 query/parser_test.go diff --git a/core/etcd.go b/core/etcd.go index 2ebc91188..4d321f6ff 100644 --- a/core/etcd.go +++ b/core/etcd.go @@ -100,9 +100,6 @@ func (self *TopologyMapper) handlenode(node *etcd.Node) error { } if len(bits) > 7 { fragment_id = util.Hex_to_SUUID(bits[7]) - if err != nil { - return err - } fragment = database.GetOrCreateFragment(frame, slice, fragment_id) } @@ -117,7 +114,7 @@ func (self *TopologyMapper) handlenode(node *etcd.Node) error { process = db.NewProcess(process_uuid) fragment.SetProcess(process) - if self.service.process_id.String() == process_uuid.String() { + if self.service.Id.String() == process_uuid.String() { self.service.Process.AddFragment(bits[1], bits[3], slice_int, fragment_id) } @@ -294,6 +291,7 @@ func (self *ProcessMapper) Run() { self_path := path + "/" + id_string log.Println("Writing configuration to etcd...") + log.Println(self_path) var err error _, err = self.service.Etcd.Set(self_path+"/port_tcp", strconv.Itoa(config.GetInt("port_tcp")), 0) diff --git a/core/http.go b/core/http.go index 2e02fbb2a..d58298430 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" @@ -63,8 +62,9 @@ func (self *WebService) HandleQuery(w http.ResponseWriter, r *http.Request) { http.Error(w, "Error reading POST data", http.StatusBadRequest) return } - q := query.QueryParser{string(body)} - log.Println(q) // TODO: parse and perform + log.Println(body) + //q := query.QueryParser{string(body)} + //log.Println(q) // TODO: parse and perform } func (self *WebService) HandleStats(w http.ResponseWriter, r *http.Request) { diff --git a/db/topology.go b/db/topology.go index 46efc941e..7ccd0a5e4 100644 --- a/db/topology.go +++ b/db/topology.go @@ -291,11 +291,14 @@ type Fragment struct { process *Process } +func (f *Fragment) GetProcess() *Process { + return f.process +} + // rename this one -func (d *Database) OldGetFragment(bitmap Bitmap, profile_id int) (*Fragment, error) { - d.mutex.Lock() - defer d.mutex.Unlock() - slice, _ := d.GetSliceForProfile(profile_id) +func (d *Database) GetFragmentForBitmap(slice *Slice, bitmap *Bitmap) (*Fragment, error) { + //d.mutex.Lock() + //defer d.mutex.Unlock() frame, _ := d.getFrame(bitmap.FrameType) fsi, err := d.GetFrameSliceIntersect(frame, slice) frag_id_s, err := fsi.hashring.Get(fmt.Sprintf("%d", bitmap.Id)) diff --git a/query/lexer.go b/query/lexer.go index aefc7c5cb..64cad2b25 100644 --- a/query/lexer.go +++ b/query/lexer.go @@ -6,7 +6,6 @@ import ( "strings" "unicode" "unicode/utf8" - //"github.com/davecgh/go-spew/spew" ) const ( @@ -14,6 +13,7 @@ const ( TYPE_LP = iota TYPE_RP = iota TYPE_ID = iota + TYPE_FRAME = iota TYPE_COMMA = iota ) @@ -129,7 +129,7 @@ func stateID(lexer *Lexer) statefn { // if next is comma peeked := lexer.peek() if peeked == rune(',') { - return stateComma + return stateFrameComma } else if peeked == rune(')') { return stateRP } else { @@ -151,6 +151,25 @@ func stateRP(lexer *Lexer) statefn { } } +func stateFrameComma(lexer *Lexer) statefn { + lexer.pos += 1 + lexer.emit(TYPE_COMMA) + return stateFrame +} + +func stateFrame(lexer *Lexer) statefn { + lexer.peek() + digits := "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_." // TODO: make this more flexible. accept anything up to a space or RP: ")" + lexer.acceptRun(digits) + lexer.emit(TYPE_FRAME) + peeked := lexer.peek() + if peeked == rune(')') { + return stateRP + } + // should never get here + return stateRP +} + func stateComma(lexer *Lexer) statefn { lexer.pos += 1 lexer.emit(TYPE_COMMA) diff --git a/query/lexer_test.go b/query/lexer_test.go index 890d9a3d5..21f454ca5 100644 --- a/query/lexer_test.go +++ b/query/lexer_test.go @@ -1,8 +1,8 @@ package query import ( - . "github.com/smartystreets/goconvey/convey" "testing" + . "github.com/smartystreets/goconvey/convey" ) func TestLexer(t *testing.T) { @@ -55,8 +55,8 @@ func TestLexer(t *testing.T) { So(tokens2[16].Text, ShouldEqual, ")") So(tokens2[16].Type, ShouldEqual, TYPE_RP) - tokens3 := Lex("intersect(get(10), get(11), concat(12,14))") - So(len(tokens3), ShouldEqual, 19) + tokens3 := Lex("intersect(get(10), get(11), concat(get(12),get(14)))") + So(len(tokens3), ShouldEqual, 25) So(tokens3[0].Text, ShouldEqual, "intersect") So(tokens3[0].Type, ShouldEqual, TYPE_FUNC) So(tokens3[1].Text, ShouldEqual, "(") @@ -85,20 +85,46 @@ func TestLexer(t *testing.T) { So(tokens3[12].Type, ShouldEqual, TYPE_FUNC) So(tokens3[13].Text, ShouldEqual, "(") So(tokens3[13].Type, ShouldEqual, TYPE_LP) - So(tokens3[14].Text, ShouldEqual, "12") - So(tokens3[14].Type, ShouldEqual, TYPE_ID) - So(tokens3[15].Text, ShouldEqual, ",") - So(tokens3[15].Type, ShouldEqual, TYPE_COMMA) - So(tokens3[16].Text, ShouldEqual, "14") + So(tokens3[14].Text, ShouldEqual, "get") + So(tokens3[14].Type, ShouldEqual, TYPE_FUNC) + So(tokens3[15].Text, ShouldEqual, "(") + So(tokens3[15].Type, ShouldEqual, TYPE_LP) + So(tokens3[16].Text, ShouldEqual, "12") So(tokens3[16].Type, ShouldEqual, TYPE_ID) So(tokens3[17].Text, ShouldEqual, ")") So(tokens3[17].Type, ShouldEqual, TYPE_RP) - So(tokens3[18].Text, ShouldEqual, ")") - So(tokens3[18].Type, ShouldEqual, TYPE_RP) + So(tokens3[18].Text, ShouldEqual, ",") + So(tokens3[18].Type, ShouldEqual, TYPE_COMMA) - tokens4 := Lex("concat(1,2,345,890)") - So(len(tokens4), ShouldEqual, 10) - So(tokens4[6].Text, ShouldEqual, "345") - So(tokens4[6].Type, ShouldEqual, TYPE_ID) + tokens4 := Lex("concat(get(1, brand),get(2))") + So(len(tokens4), ShouldEqual, 14) + So(tokens4[0].Text, ShouldEqual, "concat") + So(tokens4[0].Type, ShouldEqual, TYPE_FUNC) + So(tokens4[1].Text, ShouldEqual, "(") + So(tokens4[1].Type, ShouldEqual, TYPE_LP) + So(tokens4[2].Text, ShouldEqual, "get") + So(tokens4[2].Type, ShouldEqual, TYPE_FUNC) + So(tokens4[3].Text, ShouldEqual, "(") + So(tokens4[3].Type, ShouldEqual, TYPE_LP) + So(tokens4[4].Text, ShouldEqual, "1") + So(tokens4[4].Type, ShouldEqual, TYPE_ID) + So(tokens4[5].Text, ShouldEqual, ",") + So(tokens4[5].Type, ShouldEqual, TYPE_COMMA) + So(tokens4[6].Text, ShouldEqual, "brand") + So(tokens4[6].Type, ShouldEqual, TYPE_FRAME) + So(tokens4[7].Text, ShouldEqual, ")") + So(tokens4[7].Type, ShouldEqual, TYPE_RP) + So(tokens4[8].Text, ShouldEqual, ",") + So(tokens4[8].Type, ShouldEqual, TYPE_COMMA) + So(tokens4[9].Text, ShouldEqual, "get") + So(tokens4[9].Type, ShouldEqual, TYPE_FUNC) + So(tokens4[10].Text, ShouldEqual, "(") + So(tokens4[10].Type, ShouldEqual, TYPE_LP) + So(tokens4[11].Text, ShouldEqual, "2") + So(tokens4[11].Type, ShouldEqual, TYPE_ID) + So(tokens4[12].Text, ShouldEqual, ")") + So(tokens4[12].Type, ShouldEqual, TYPE_RP) + So(tokens4[13].Text, ShouldEqual, ")") + So(tokens4[13].Type, ShouldEqual, TYPE_RP) }) } diff --git a/query/parser.go b/query/parser.go index a24b01ba9..7a4686c7d 100644 --- a/query/parser.go +++ b/query/parser.go @@ -1,62 +1,88 @@ package query import ( - "encoding/json" "errors" "pilosa/db" - //"github.com/davecgh/go-spew/spew" + "strconv" ) var InvalidQueryError = errors.New("Invalid query format.") -type QueryParser struct { - QueryString string -} +type QueryParser struct{} -func (q *QueryParser) Walk(data interface{}) (*Query, error) { - query := new(Query) - - slice, ok := data.([]interface{}) - if !ok { - return nil, InvalidQueryError +func (qp *QueryParser) walkInputs(tokens []Token) []QueryInput { + // BITMAP + if tokens[0].Type == TYPE_ID { + // TODO: look for frame type in the tokens list + bitmap_id, err := strconv.Atoi(tokens[0].Text) + 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) == 3 && tokens[2].Type == TYPE_FRAME { + frame_type = tokens[2].Text + } + bm := db.Bitmap{bitmap_id, frame_type} + return []QueryInput{&bm} } - operation, ok := slice[0].(string) - if !ok { - return nil, InvalidQueryError - } - if operation == "union" || operation == "intersect" { - query.Operation = operation - inputs := slice[1:] - query.Inputs = make([]QueryInput, len(inputs)) - for idx, input := range inputs { - subquery, err := q.Walk(input) - if err != nil { - return nil, err + // LIST OF QUERIES + qi := []QueryInput{} + open_parens := -1 // >=0 means i'm inside the search for end paren + start := 0 + for i := 0; i < len(tokens); i++ { + if tokens[i].Type == TYPE_FUNC && open_parens == -1 { + start = i + } else if tokens[i].Type == TYPE_LP { + open_parens++ + } else if tokens[i].Type == TYPE_RP { + if open_parens == 0 { + q, err := qp.walk(tokens[start : i+1]) + if err != nil { + panic(err) + } + qi = append(qi, q) + open_parens = -1 + } else { + open_parens-- } - query.Inputs[idx] = subquery } - } else if operation == "bitmap" { - query.Operation = "get" - frame, ok := slice[1].(string) - if !ok { - return nil, InvalidQueryError - } - id, ok := slice[2].(float64) - if !ok { - return nil, InvalidQueryError - } - id_int := int(id) - query.Inputs = []QueryInput{db.Bitmap{id_int, frame}} } - - return query, nil + return qi } -func (q *QueryParser) Parse() (*Query, error) { - var data interface{} - if err := json.Unmarshal([]byte(q.QueryString), &data); err != nil { - return nil, err +func (qp *QueryParser) walk(tokens []Token) (*Query, error) { + + if tokens[0].Type != TYPE_FUNC { + panic("BAD!") } - return q.Walk(data) + if tokens[1].Type != TYPE_LP { + panic("BAD!") + } + + q := new(Query) + q.Operation = tokens[0].Text + + // scan from open to close paren + open_parens := 0 + for i := 2; i < len(tokens); i++ { + // 1 must be "(" + if tokens[i].Type == TYPE_LP { + open_parens++ + } else if tokens[i].Type == TYPE_RP { + if open_parens == 0 { + if i == len(tokens)-1 { + q.Inputs = qp.walkInputs(tokens[2:i]) + } + } else { + open_parens-- + } + } + } + return q, nil +} + +func (qp *QueryParser) Parse(tokens []Token) (*Query, error) { + return qp.walk(tokens) } diff --git a/query/parser_test.go b/query/parser_test.go new file mode 100644 index 000000000..bed194513 --- /dev/null +++ b/query/parser_test.go @@ -0,0 +1,21 @@ +package query + +import ( + "testing" + + "github.com/davecgh/go-spew/spew" + . "github.com/smartystreets/goconvey/convey" +) + +func TestQueryParser(t *testing.T) { + Convey("Basic query parse", t, func() { + + tokens := Lex("union(get(10,general), get(11,brand), get(12))") + qp := QueryParser{} + q, err := qp.Parse(tokens) + if err != nil { + panic(err) + } + spew.Dump(q) + }) +} diff --git a/query/planner.go b/query/planner.go index 397864b14..d5eb7a736 100644 --- a/query/planner.go +++ b/query/planner.go @@ -1,6 +1,7 @@ package query import ( + "github.com/davecgh/go-spew/spew" "github.com/nu7hatch/gouuid" //"strconv" "fmt" @@ -10,15 +11,15 @@ import ( // A single step in the query plan. type QueryStep struct { - id uuid.UUID - operation string - inputs []QueryInput - location string - destination string + id uuid.UUID + operation string + inputs []QueryInput + location *db.Process + return_process *db.Process } func (q QueryStep) String() string { - return fmt.Sprintf("%s %s %s, LOC: %s, DEST: %s", q.operation, q.id.String(), q.inputs, q.location, q.destination) + return fmt.Sprintf("%s %s %s, LOC: %s, DEST: %s", q.operation, q.id.String(), q.inputs, q.location, q.return_process) } type QueryInput interface{} @@ -33,24 +34,23 @@ type Query struct { type QueryPlan []QueryStep type QueryPlanner struct { - Cluster *db.Cluster Database *db.Database } type QueryTree interface { - getLocation(d *db.Database) string + getLocation(d *db.Database) *db.Process } // QueryTree for UNION, INTER, and CAT queries type CompositeQueryTree struct { operation string subqueries []QueryTree - location string + location *db.Process } // 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) string { - if qt.location == "" { +func (qt *CompositeQueryTree) getLocation(d *db.Database) *db.Process { + if qt.location == nil { subqueryLength := len(qt.subqueries) if subqueryLength > 1 { locationIndex := rand.Intn(subqueryLength) @@ -63,28 +63,18 @@ func (qt *CompositeQueryTree) getLocation(d *db.Database) string { // QueryTree for GET queries type GetQueryTree struct { - bitmap db.Bitmap + bitmap *db.Bitmap slice int } // Uses consistent hashing function to select node containing data for GET operation -func (qt *GetQueryTree) getLocation(d *db.Database) string { - /* - frame, err := d.GetFrame(qt.bitmap.FrameType) - if err != nil { - panic(err) - } - slice := frame.Slices[qt.slice] - hashString, err := slice.Hashring.Get(strconv.Itoa(qt.bitmap.Id)) - - var sliceIndex int - var fragIndex int - fmt.Sscan(hashString, &fragIndex, &sliceIndex) - fragment := slice.Fragments[fragIndex] - - return fmt.Sprintf(fragment.Node) - */ - return "Nothing yet" +func (qt *GetQueryTree) getLocation(d *db.Database) *db.Process { + 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() } // Builds QueryTree object from Query. Pass slice=-1 to perform operation on all slices @@ -103,7 +93,7 @@ func (qp *QueryPlanner) buildTree(query *Query, slice int) QueryTree { } } else { if query.Operation == "get" { - tree = &GetQueryTree{query.Inputs[0].(db.Bitmap), slice} + tree = &GetQueryTree{query.Inputs[0].(*db.Bitmap), slice} return tree } else { subqueries := make([]QueryTree, len(query.Inputs)) @@ -117,14 +107,15 @@ 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 string) *QueryPlan { +func (qp *QueryPlanner) flatten(qt QueryTree, id *uuid.UUID, location *db.Process) *QueryPlan { 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} for index, subq := range composite.subqueries { sub_id, _ := uuid.NewV4() - step.inputs[index] = []QueryInput{"wait", sub_id} + // 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...) } @@ -138,7 +129,12 @@ func (qp *QueryPlanner) flatten(qt QueryTree, id *uuid.UUID, location string) *Q } // Transforms Query into QueryTree and flattens to QueryPlan object -func (qp *QueryPlanner) Plan(query *Query, id *uuid.UUID, destination string, slice int) *QueryPlan { +func (qp *QueryPlanner) Plan(query *Query, id *uuid.UUID, destination *db.Process, slice int) *QueryPlan { + queryTree := qp.buildTree(query, -1) + spew.Dump("--------------------------------------------") + spew.Dump(queryTree) + spew.Dump("--------------------------------------------") return qp.flatten(queryTree, id, destination) + //return &QueryPlan{} } diff --git a/query/planner_test.go b/query/planner_test.go index 6cef3873b..e3509e4a3 100644 --- a/query/planner_test.go +++ b/query/planner_test.go @@ -1,13 +1,66 @@ package query import ( - . "github.com/smartystreets/goconvey/convey" - "log" + "pilosa/db" + "pilosa/util" "testing" + + "github.com/davecgh/go-spew/spew" + "github.com/nu7hatch/gouuid" + . "github.com/smartystreets/goconvey/convey" ) func TestQueryPlanner(t *testing.T) { Convey("Basic query plan", t, func() { - log.Println("query planner test") + + bm1 := db.Bitmap{10, "general"} + inputs1 := []QueryInput{&bm1} + query1 := Query{"get", inputs1} + + bm2 := db.Bitmap{20, "general"} + inputs2 := []QueryInput{&bm2} + query2 := Query{"get", inputs2} + + inputs := []QueryInput{&query1, &query2} + query := Query{"union", inputs} + /* + + bm1 := db.Bitmap{10, "general"} + inputs1 := []QueryInput{&bm1} + query1 := Query{"get", inputs1} + query := query1 + */ + + // create an empty database + cluster := db.NewCluster() + database := cluster.GetOrCreateDatabase("main") + frame := database.GetOrCreateFrame("general") + + slice1 := database.GetOrCreateSlice(0) + fragment_id1 := util.Id() + fragment1 := database.GetOrCreateFragment(frame, slice1, fragment_id1) + process_id1, _ := uuid.NewV4() + process1 := db.NewProcess(process_id1) + process1.SetHost("----192.1.1.0----") + fragment1.SetProcess(process1) + + slice2 := database.GetOrCreateSlice(1) + fragment_id2 := util.Id() + fragment2 := database.GetOrCreateFragment(frame, slice2, fragment_id2) + process_id2, _ := uuid.NewV4() + process2 := db.NewProcess(process_id2) + process2.SetHost("----192.1.1.1----") + fragment2.SetProcess(process2) + + qplanner := QueryPlanner{Database: database} + destination := db.Process{} + + id, _ := uuid.NewV4() + qp := qplanner.Plan(&query, id, &destination, 0) + + for i, qs := range *qp { + spew.Dump(i, qs, qs.inputs) + spew.Dump("**************************************************************") + } }) }