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

query parser and planner (PQL string > lex > tokens > parser > query > planner > query plan)

Browse source
This commit is contained in:
travisturner 2013-12-27 11:13:37 -06:00
parent 6d7c683a8a
commit d7a3ecd70a
9 changed files with 248 additions and 106 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
core/etcd.go
View file

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

6
core/http.go
View file

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

11
db/topology.go
View file

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

23
query/lexer.go
View file

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

54
query/lexer_test.go
View file

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

112
query/parser.go
View file

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

21
query/parser_test.go Normal file
View file

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

62
query/planner.go
View file

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

59
query/planner_test.go
View file

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