scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-07 09:05:55 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

add query.Execute(pql) to http/query handler

Browse source
This commit is contained in:
travisturner 2013-12-27 13:46:07 -06:00
parent d7a3ecd70a
commit 0ea1b17ea1
7 changed files with 125 additions and 110 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
core/http.go
View file

@ -7,6 +7,7 @@ import (
"net/http"
"pilosa/config"
"pilosa/db"
"pilosa/query"
"strconv"
"github.com/davecgh/go-spew/spew"
@ -62,9 +63,12 @@ func (self *WebService) HandleQuery(w http.ResponseWriter, r *http.Request) {
http.Error(w, "Error reading POST data", http.StatusBadRequest)
return
}
log.Println(body)
//q := query.QueryParser{string(body)}
//log.Println(q) // TODO: parse and perform
cluster := self.service.Cluster
database := cluster.GetOrCreateDatabase("main")
pql := string(body)
query.Execute(database, pql)
}
func (self *WebService) HandleStats(w http.ResponseWriter, r *http.Request) {

7
cruncher/cruncher.go
View file

@ -17,13 +17,6 @@ type Cruncher struct {
func (cruncher *Cruncher) Run() {
spew.Dump("Cruncher.Run")
/*
bh = api.Get(frag,tileid)
api.SetBit(frag,bh,1)
api.Count(frag,bh)
api.Union(frag,[bh1,bh2])
api.Intersect(frag,[bh1,bh2])
*/
cruncher.Service.Run()
}

150
dispatch/dispatch.go
View file

@ -3,10 +3,6 @@ package dispatch
import (
"log"
"pilosa/core"
"pilosa/index"
"pilosa/util"
"strconv"
"strings"
"github.com/davecgh/go-spew/spew"
)
@ -32,91 +28,93 @@ func (self *CruncherDispatch) Run() {
spew.Dump(message.Key)
spew.Dump(message.Data)
path := message.Data.(string)
/*
path := message.Data.(string)
bits := strings.Split(path, "/")
bits := strings.Split(path, "/")
var fragment_id util.SUUID
var bitmaps []uint64
var profile_id uint64
var s uint64
var fragment_id util.SUUID
var bitmaps []uint64
var profile_id uint64
var s uint64
command := bits[0]
if len(bits) > 1 {
fragment_id = util.Hex_to_SUUID(bits[1])
}
if len(bits) > 2 {
bitmap_ids := strings.Split(bits[2], ",")
spew.Dump(bitmap_ids)
for i := range bitmap_ids {
spew.Dump(i, bitmap_ids[i])
s, _ = strconv.ParseUint(bitmap_ids[i], 10, 64)
bitmaps = append(bitmaps, s)
command := bits[0]
if len(bits) > 1 {
fragment_id = util.Hex_to_SUUID(bits[1])
}
if len(bits) > 2 {
bitmap_ids := strings.Split(bits[2], ",")
spew.Dump(bitmap_ids)
for i := range bitmap_ids {
spew.Dump(i, bitmap_ids[i])
s, _ = strconv.ParseUint(bitmap_ids[i], 10, 64)
bitmaps = append(bitmaps, s)
}
}
if len(bits) > 3 {
profile_id, _ = strconv.ParseUint(bits[3], 10, 64)
}
}
if len(bits) > 3 {
profile_id, _ = strconv.ParseUint(bits[3], 10, 64)
}
spew.Dump("COMMAND:", command)
spew.Dump("FRAGID:", fragment_id)
spew.Dump("BITMAPS:", bitmaps)
spew.Dump("PROFILEID:", profile_id)
spew.Dump("COMMAND:", command)
spew.Dump("FRAGID:", fragment_id)
spew.Dump("BITMAPS:", bitmaps)
spew.Dump("PROFILEID:", profile_id)
if command == "set" {
res, err := self.service.Process.SetBit(fragment_id, bitmaps[0], profile_id)
spew.Dump("SET")
spew.Dump(res)
spew.Dump(err)
}
if command == "count" {
spew.Dump("COUNT")
bh, err := self.service.Process.Get(fragment_id, bitmaps[0])
if err != nil {
if command == "set" {
res, err := self.service.Process.SetBit(fragment_id, bitmaps[0], profile_id)
spew.Dump("SET")
spew.Dump(res)
spew.Dump(err)
}
count, err := self.service.Process.Count(fragment_id, bh)
if err != nil {
spew.Dump(err)
}
spew.Dump(count)
}
if command == "intersect" {
spew.Dump("INTERSECT")
var bhs []index.BitmapHandle
for i := range bitmaps {
bh, _ := self.service.Process.Get(fragment_id, bitmaps[i])
bhs = append(bhs, bh)
}
bhi, err := self.service.Process.Intersect(fragment_id, bhs)
if err != nil {
spew.Dump(err)
if command == "count" {
spew.Dump("COUNT")
bh, err := self.service.Process.Get(fragment_id, bitmaps[0])
if err != nil {
spew.Dump(err)
}
count, err := self.service.Process.Count(fragment_id, bh)
if err != nil {
spew.Dump(err)
}
spew.Dump(count)
}
if command == "intersect" {
spew.Dump("INTERSECT")
var bhs []index.BitmapHandle
for i := range bitmaps {
bh, _ := self.service.Process.Get(fragment_id, bitmaps[i])
bhs = append(bhs, bh)
}
bhi, err := self.service.Process.Intersect(fragment_id, bhs)
if err != nil {
spew.Dump(err)
}
count, err := self.service.Process.Count(fragment_id, bhi)
if err != nil {
spew.Dump(err)
}
spew.Dump(count)
}
if command == "union" {
spew.Dump("UNION")
var bhs []index.BitmapHandle
for i := range bitmaps {
bh, _ := self.service.Process.Get(fragment_id, bitmaps[i])
bhs = append(bhs, bh)
}
bhi, err := self.service.Process.Union(fragment_id, bhs)
if err != nil {
spew.Dump(err)
count, err := self.service.Process.Count(fragment_id, bhi)
if err != nil {
spew.Dump(err)
}
spew.Dump(count)
}
if command == "union" {
spew.Dump("UNION")
var bhs []index.BitmapHandle
for i := range bitmaps {
bh, _ := self.service.Process.Get(fragment_id, bitmaps[i])
bhs = append(bhs, bh)
}
bhi, err := self.service.Process.Union(fragment_id, bhs)
if err != nil {
spew.Dump(err)
}
count, err := self.service.Process.Count(fragment_id, bhi)
if err != nil {
spew.Dump(err)
count, err := self.service.Process.Count(fragment_id, bhi)
if err != nil {
spew.Dump(err)
}
spew.Dump(count)
}
spew.Dump(count)
}
*/
}
}

20
query/planner.go
View file

@ -1,12 +1,11 @@
package query
import (
"github.com/davecgh/go-spew/spew"
"github.com/nu7hatch/gouuid"
//"strconv"
"fmt"
"math/rand"
"pilosa/db"
"github.com/nu7hatch/gouuid"
)
// A single step in the query plan.
@ -22,14 +21,6 @@ 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.return_process)
}
type QueryInput interface{}
// Represents a parsed query. Inputs can be Query or Bitmap objects
type Query struct {
Operation string
Inputs []QueryInput // Maybe Bitmap and Query objects should have different fields to avoid using interface{}
}
// This is the output of the query planner. Contains a list of steps which can be performed in parallel
type QueryPlan []QueryStep
@ -129,12 +120,7 @@ func (qp *QueryPlanner) flatten(qt QueryTree, id *uuid.UUID, location *db.Proces
}
// Transforms Query into QueryTree and flattens to QueryPlan object
func (qp *QueryPlanner) Plan(query *Query, id *uuid.UUID, destination *db.Process, slice int) *QueryPlan {
func (qp *QueryPlanner) Plan(query *Query, id *uuid.UUID, destination *db.Process) *QueryPlan {
queryTree := qp.buildTree(query, -1)
spew.Dump("--------------------------------------------")
spew.Dump(queryTree)
spew.Dump("--------------------------------------------")
return qp.flatten(queryTree, id, destination)
//return &QueryPlan{}
}

2
query/planner_test.go
View file

@ -56,7 +56,7 @@ func TestQueryPlanner(t *testing.T) {
destination := db.Process{}
id, _ := uuid.NewV4()
qp := qplanner.Plan(&query, id, &destination, 0)
qp := qplanner.Plan(&query, id, &destination)
for i, qs := range *qp {
spew.Dump(i, qs, qs.inputs)

37
query/query.go Normal file
View file

@ -0,0 +1,37 @@
package query
import (
"pilosa/db"
"github.com/davecgh/go-spew/spew"
"github.com/nu7hatch/gouuid"
)
type QueryInput interface{}
type Query struct {
Operation string
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{}
}
func Execute(database *db.Database, pql string) {
//spew.Dump("EXECUTE")
//spew.Dump(pql)
tokens := Lex(pql)
//spew.Dump(tokens)
query_parser := QueryParser{}
query, err := query_parser.Parse(tokens)
if err != nil {
panic(err)
}
//spew.Dump(query)
query_planner := QueryPlanner{Database: database}
id, _ := uuid.NewV4()
destination := db.Process{}
query_plan := query_planner.Plan(query, id, &destination)
spew.Dump(query_plan)
}

9
transport/tcp.go
View file

@ -1,14 +1,10 @@
package transport
import (
"fmt"
"log"
"net"
"net/http"
"pilosa/config"
"pilosa/core"
"pilosa/db"
"time"
)
type TcpTransport struct {
@ -17,6 +13,7 @@ type TcpTransport struct {
stop chan bool
}
/*
func (self *TcpTransport) RunServer(porti int) {
http.Handle("/", self)
port := fmt.Sprintf(":%d", porti)
@ -54,11 +51,11 @@ func (self *TcpTransport) ServeHTTP(w http.ResponseWriter, r *http.Request) {
msg.Data = path
self.inbox <- msg
}
*/
func (self *TcpTransport) Run() {
log.Println("Initializing TCP transport")
self.RunServer(self.port)
//self.RunServer(self.port)
}
func (self *TcpTransport) Close() {