Merge branch 'master' of ops:nuevo-pilosa

core/http.go

@@ -68,7 +68,14 @@ func (self *WebService) HandleQuery(w http.ResponseWriter, r *http.Request) {
 	// TODO: we need to get the database name from the query string (for now, hard-coded)
 	database_name := "main"
 	pql := string(body)
-	self.service.Executor.RunQuery(database_name, pql)
+	results := self.service.Executor.RunPQL(database_name, pql)
+
+	encoder := json.NewEncoder(w)
+	err = encoder.Encode(results)
+	if err != nil {
+		log.Fatal("Error encoding stats")
+	}
+
 }
 
 func (self *WebService) HandleStats(w http.ResponseWriter, r *http.Request) {

core/query.go

@@ -160,5 +160,10 @@ func (self *Service) GetQueryStepHandler(msg *db.Message) {
 func (self *Service) SetQueryStepHandler(msg *db.Message) {
 	spew.Dump("SET QUERYSTEP")
 	qs := msg.Data.(query.SetQueryStep)
-	self.Index.SetBit(qs.Location.FragmentId, qs.Bitmap.Id, qs.ProfileId)
+	result, err := self.Index.SetBit(qs.Location.FragmentId, qs.Bitmap.Id, qs.ProfileId)
+	spew.Dump("result:", result)
+	spew.Dump("err:", err)
+
+	result_message := db.Message{Data: query.SetQueryResult{&query.BaseQueryResult{Id: qs.Id, Data: result}}}
+	self.Transport.Send(&result_message, qs.Destination.ProcessId)
 }

dispatch/dispatch.go

@@ -5,6 +5,8 @@ import (
 	"pilosa/core"
 	"pilosa/db"
 	"pilosa/query"
+
+	"github.com/davecgh/go-spew/spew"
 )
 
 type Dispatch struct {
@@ -29,6 +31,7 @@ func (self *Dispatch) Run() {
 			pong := db.Message{Data: core.PongRequest{Id: data.Id}}
 			self.service.Transport.Send(&pong, data.Source)
 		case db.HoldResult:
+			spew.Dump("HOLD-SET", data.ResultId())
 			self.service.Hold.Set(data.ResultId(), data.ResultData(), 30)
 		case query.PortableQueryStep:
 			go self.service.Executor.NewJob(message)

executor/executor.go

@@ -3,12 +3,11 @@ package executor
 import (
 	"fmt"
 	"log"
+	"pilosa/config"
 	"pilosa/core"
 	"pilosa/db"
 	"pilosa/query"
 	"pilosa/util"
-	"reflect"
-	"tux21b.org/v1/gocql/uuid"
 
 	"github.com/davecgh/go-spew/spew"
 )
@@ -46,8 +45,22 @@ func (self *Executor) NewJob(job *db.Message) {
 	}
 }
 
-func (self *Executor) RunQuery(database_name string, pql string) {
-	database := self.service.Cluster.GetOrCreateDatabase(database_name)
+type stringSlice []string
+
+func (slice stringSlice) pos(value string) int {
+	for p, v := range slice {
+		if v == value {
+			return p
+		}
+	}
+	return -1
+}
+
+func (self *Executor) RunQueryTest(database_name string, pql string) string {
+	return pql
+}
+
+func (self *Executor) runQuery(database *db.Database, qry *query.Query) {
 	process, err := self.service.GetProcess()
 	if err != nil {
 		spew.Dump(err)
@@ -56,33 +69,65 @@ func (self *Executor) RunQuery(database_name string, pql string) {
 	fragment_id := util.SUUID(0)
 	destination := db.Location{&process_id, fragment_id}
 
-	query_plan := query.QueryPlanForPQL(database, pql, &destination)
-	//spew.Dump(query_plan)
-
+	spew.Dump("QUERY.ID:", qry.Id)
+	query_plan := query.QueryPlanForQuery(database, qry, &destination)
 	// loop over the query steps and send to Transport
-	var last_id *uuid.UUID
 	for _, qs := range *query_plan {
 		msg := new(db.Message)
 		msg.Data = qs
 		switch step := qs.(type) {
 		case query.PortableQueryStep:
 			self.service.Transport.Send(msg, step.GetLocation().ProcessId)
-			if reflect.TypeOf(step) != reflect.TypeOf(query.SetQueryStep{}) {
-				last_id = step.GetId()
-			}
 		}
 	}
+}
 
-	// add an entry to my execute map[key] that is waiting for the final result
-	if last_id != nil {
-		final, err := self.service.Hold.Get(last_id, 10)
+func (self *Executor) RunPQL(database_name string, pql string) interface{} {
+	database := self.service.Cluster.GetOrCreateDatabase(database_name)
+
+	// see if the outer query function is a custom query
+	reserved_functions := stringSlice{"get", "set", "union", "intersect", "count"}
+	tokens := query.Lex(pql)
+	outer_token := tokens[0].Text
+
+	if reserved_functions.pos(outer_token) != -1 {
+
+		qry := query.QueryForTokens(tokens)
+		go self.runQuery(database, qry)
+
+		var final interface{}
+		final, err := self.service.Hold.Get(qry.Id, 10)
 		if err != nil {
 			spew.Dump(err)
 		}
-		spew.Dump("*******************************************************")
-		spew.Dump("FINAL", final)
-		spew.Dump("*******************************************************")
+		return final
+
+	} else {
+		macros_dir := config.GetString("macros")
+		macros_file := macros_dir + "/" + outer_token + ".js"
+		filter := query.TokensToString(tokens)
+		query_list := GetMacro(macros_file, filter).(query.PqlList)
+
+		for i, _ := range query_list {
+			qry := query.QueryForPQL(query_list[i].PQL)
+			go self.runQuery(database, qry)
+			query_list[i].Id = qry.Id
+		}
+
+		// technically, this is blocking on the hold for each query, but it may be ok, since we need them all for the final anyway.
+		final_result := make(map[string]interface{})
+		for i, _ := range query_list {
+			final, err := self.service.Hold.Get(query_list[i].Id, 10)
+			if err != nil {
+				spew.Dump(err)
+			}
+			spew.Dump(final)
+			final_result[query_list[i].Label] = final
+		}
+
+		return final_result
 	}
+
 }
 
 func (self *Executor) Run() {

executor/utils.go

@@ -0,0 +1,50 @@
+package executor
+
+import (
+	"io/ioutil"
+	"pilosa/query"
+
+	"github.com/davecgh/go-spew/spew"
+	"github.com/robertkrimen/otto"
+)
+
+func GetMacro(file_name string, filter string) interface{} {
+
+	file_data, err := ioutil.ReadFile(file_name)
+	if err != nil {
+		spew.Dump(err)
+	}
+	s := string(file_data[:])
+
+	js := "query_list = (function (filter){" + s + "})('" + filter + "');"
+
+	Otto := otto.New()
+	Otto.Run(js)
+	query_objects, err := Otto.Get("query_list")
+
+	query_list_interface, err := query_objects.Export()
+	if err != nil {
+		spew.Dump(err)
+	}
+
+	var query_list query.PqlList
+
+	// ql is []interface{}
+	switch ql := query_list_interface.(type) {
+	case []interface{}:
+		spew.Dump("INTERFACE", ql)
+		// q is map[string]interface{}
+		for i, _ := range ql {
+			q := ql[i].(map[string]interface{})
+			spew.Dump(ql[i])
+			spew.Dump(q)
+			spew.Dump(q["label"].(string))
+			spew.Dump(q["pql"].(string))
+			query_list = append(query_list, query.PqlListItem{Label: q["label"].(string), PQL: q["pql"].(string)})
+		}
+	default:
+		spew.Dump("DEFAULT")
+	}
+
+	return query_list
+}

interfaces/core.go

@@ -25,5 +25,5 @@ type Executorer interface {
 	Close()
 	Run()
 	NewJob(*db.Message)
-	RunQuery(string, string)
+	RunPQL(string, string) interface{}
 }

query/parser.go

@@ -4,6 +4,8 @@ import (
 	"errors"
 	"pilosa/db"
 	"strconv"
+
+	"tux21b.org/v1/gocql/uuid"
 )
 
 var InvalidQueryError = errors.New("Invalid query format.")
@@ -75,6 +77,8 @@ func (qp *QueryParser) walk(tokens []Token) (*Query, error) {
 	}
 
 	q := new(Query)
+	id := uuid.RandomUUID()
+	q.Id = &id
 	q.Operation = tokens[0].Text
 
 	// scan from open to close paren

query/planner.go

@@ -219,6 +219,10 @@ type SetQueryStep struct {
 	ProfileId uint64
 }
 
+type SetQueryResult struct {
+	*BaseQueryResult
+}
+
 // QueryTree for SET queries
 type SetQueryTree struct {
 	bitmap     *db.Bitmap
@@ -238,6 +242,7 @@ func (qt *SetQueryTree) getLocation(d *db.Database) *db.Location {
 ///////////////////////////////////////////////////////////////////////////////////////////////////
 
 func init() {
+	gob.Register(SetQueryResult{})
 	gob.Register(GetQueryResult{})
 	gob.Register(CatQueryResult{})
 	gob.Register(UnionQueryResult{})
@@ -401,5 +406,6 @@ func (qp *QueryPlanner) flatten(qt QueryTree, id *uuid.UUID, location *db.Locati
 // Transforms Query into QueryTree and flattens to QueryPlan object
 func (qp *QueryPlanner) Plan(query *Query, id *uuid.UUID, destination *db.Location) *QueryPlan {
 	queryTree := qp.buildTree(query, -1)
-	return qp.flatten(queryTree, id, destination)
+	//return qp.flatten(queryTree, id, destination) // TODO: remove the "id" parameter, since we are using the query.Id as the value
+	return qp.flatten(queryTree, query.Id, destination)
 }

query/planner_test.go

@@ -13,16 +13,19 @@ import (
 func TestQueryPlanner(t *testing.T) {
 	Convey("Basic query plan", t, func() {
 
+		id1 := uuid.RandomUUID()
 		bm1 := db.Bitmap{10, "general"}
 		inputs1 := []QueryInput{&bm1}
-		query1 := Query{"get", inputs1, 0}
+		query1 := Query{&id1, "get", inputs1, 0}
 
+		id2 := uuid.RandomUUID()
 		bm2 := db.Bitmap{20, "general"}
 		inputs2 := []QueryInput{&bm2}
-		query2 := Query{"get", inputs2, 0}
+		query2 := Query{&id2, "get", inputs2, 0}
 
+		id3 := uuid.RandomUUID()
 		inputs := []QueryInput{&query1, &query2}
-		query := Query{"union", inputs, 0}
+		query := Query{&id3, "union", inputs, 0}
 		/*
 
 			bm1 := db.Bitmap{10, "general"}

query/query.go

@@ -2,6 +2,9 @@ package query
 
 import (
 	"pilosa/db"
+	"strings"
+	"github.com/davecgh/go-spew/spew"
+
 	"tux21b.org/v1/gocql/uuid"
 )
 
@@ -11,7 +14,16 @@ type QueryResults struct {
 	Data interface{}
 }
 
+type PqlList []PqlListItem
+
+type PqlListItem struct {
+	Id    *uuid.UUID
+	Label string
+	PQL   string
+}
+
 type Query struct {
+	Id        *uuid.UUID
 	Operation string
 	Inputs    []QueryInput //"strconv"
 	// Represents a parsed query. Inputs can be Query or Bitmap objects
@@ -21,13 +33,42 @@ type Query struct {
 
 func QueryPlanForPQL(database *db.Database, pql string, destination *db.Location) *QueryPlan {
 	tokens := Lex(pql)
+	return QueryPlanForTokens(database, tokens, destination)
+}
+
+func QueryForPQL(pql string) *Query {
+	tokens := Lex(pql)
+	return QueryForTokens(tokens)
+}
+
+func QueryForTokens(tokens []Token) *Query {
 	query_parser := QueryParser{}
 	query, err := query_parser.Parse(tokens)
 	if err != nil {
 		panic(err)
 	}
+	return query
+}
+
+func QueryPlanForTokens(database *db.Database, tokens []Token, destination *db.Location) *QueryPlan {
+	query := QueryForTokens(tokens)
+	return QueryPlanForQuery(database, query, destination)
+}
+
+func QueryPlanForQuery(database *db.Database, query *Query, destination *db.Location) *QueryPlan {
 	query_planner := QueryPlanner{Database: database}
 	id := uuid.RandomUUID()
 	query_plan := query_planner.Plan(query, &id, destination)
 	return query_plan
 }
+
+func TokensToString(tokens []Token) string {
+	var str []string
+	for i, _ := range tokens {
+		spew.Dump(tokens[i].Text)
+		str = append(str, tokens[i].Text)
+	}
+	// for now, we're just using this function to pull the filter out of the outer function "outerfunc(filter)"
+	str = str[2 : len(str)-1]
+	return strings.Join(str, "")
+}