From cd6e4d07e8f3bd9df559747873e6876900de27da Mon Sep 17 00:00:00 2001 From: Todd Gruben Date: Fri, 29 Aug 2014 16:29:21 +0000 Subject: [PATCH] added stash keyword --- core/query.go | 53 +++++++++++++++++++--- deps.json | 2 +- executor/executor.go | 4 +- index/storage_test.go | 72 +++++++++++++++-------------- query/parser.go | 6 +++ query/planner.go | 102 ++++++++++++++++++++++++++++++++++++++++-- query/planner_test.go | 21 ++++++++- 7 files changed, 213 insertions(+), 47 deletions(-) diff --git a/core/query.go b/core/query.go index e075cb50b..c51d15862 100644 --- a/core/query.go +++ b/core/query.go @@ -1,6 +1,7 @@ package core import ( + "log" "pilosa/db" "pilosa/index" "pilosa/query" @@ -136,9 +137,46 @@ func (self *Service) DifferenceQueryStepHandler(msg *db.Message) { self.Transport.Send(&result_message, qs.Destination.ProcessId) } +func (self *Service) StashQueryStepHandler(msg *db.Message) { + qs := msg.Data.(query.StashQueryStep) + + part := make(chan interface{}) + num_parts := len(qs.Inputs) + + for _, input := range qs.Inputs { + go func(id *util.GUID, part chan interface{}) { + value, _ := self.Hold.Get(id, 10) + part <- value + }(input, part) + } + //just collect all the handles and return them + result := query.Stash{make([]query.CacheItem, 0)} + for i := 0; i < num_parts; i++ { + value := <-part + + switch val := value.(type) { + case index.BitmapHandle: + log.Println("STASH ADDING HANDLE", val) + //not sure what to do here.... + //result.Handles = append(result.Handles, val) + case []byte: + bh, _ := self.Index.FromBytes(qs.Location.FragmentId, val) + item := query.CacheItem{qs.Location.FragmentId, bh} + result.Stash = append(result.Stash, item) + case query.Stash: + result.Stash = append(result.Stash, val.Stash...) + default: + log.Println("UNEXCPECTED MESSAG", value) + } + } + result_message := db.Message{Data: query.StashQueryResult{&query.BaseQueryResult{Id: qs.Id, Data: result}}} + self.Transport.Send(&result_message, qs.Destination.ProcessId) + +} + func (self *Service) CatQueryStepHandler(msg *db.Message) { qs := msg.Data.(query.CatQueryStep) - //spew.Dump("CAT QUERYSTEP") + spew.Dump("CAT QUERYSTEP") var handles []index.BitmapHandle return_type := "bitmap-handles" var sum uint64 @@ -161,6 +199,7 @@ func (self *Service) CatQueryStepHandler(msg *db.Message) { } //for _, input := range qs.Inputs { + check_pair := false for i := 0; i < num_parts; i++ { value := <-part @@ -191,13 +230,15 @@ func (self *Service) CatQueryStepHandler(msg *db.Message) { process util.GUID handle index.BitmapHandle }{val.ProcessId, val.HBitmap} + check_pair = true } } - - tasks := BuildTask(merge_map, slice_map, all_slice) - FetchMissing(tasks, self) - for k, v := range GatherResults(tasks, self) { - merge_map[k] += v + if check_pair { //no point in doing this for non top-n handling + tasks := BuildTask(merge_map, slice_map, all_slice) + FetchMissing(tasks, self) + for k, v := range GatherResults(tasks, self) { + merge_map[k] += v + } } // either return the sum, or return the compressed bitmap resulting from the cat (union) diff --git a/deps.json b/deps.json index a3e7e1dc4..fd88c3567 100644 --- a/deps.json +++ b/deps.json @@ -21,7 +21,7 @@ }, "gocql": { "repo": "github.com/gocql/gocql", - "version": "f2deeb64d238e3a470f4489d734fab820cb44b4b", + "version": "7bd964ec83ae675586ba314806c02341c015c45b", "type": "git" }, "goleveldb": { diff --git a/executor/executor.go b/executor/executor.go index d383a932b..a6f839db9 100644 --- a/executor/executor.go +++ b/executor/executor.go @@ -46,6 +46,8 @@ func (self *Executor) NewJob(job *db.Message) { self.service.SetQueryStepHandler(job) case query.RangeQueryStep: self.service.RangeQueryStepHandler(job) + case query.StashQueryStep: + self.service.StashQueryStepHandler(job) // case query.MaskQueryStep: // self.service.MaskQueryStepHandler(job) default: @@ -103,7 +105,7 @@ func (self *Executor) RunPQL(database_name string, pql string) (interface{}, err database := self.service.Cluster.GetOrCreateDatabase(database_name) // see if the outer query function is a custom query - reserved_functions := stringSlice{"get", "set", "union", "intersect", "difference", "count", "top-n", "mask", "range"} + reserved_functions := stringSlice{"get", "set", "union", "intersect", "difference", "count", "top-n", "mask", "range", "stash", "recall"} tokens, err := query.Lex(pql) if err != nil { return nil, err diff --git a/index/storage_test.go b/index/storage_test.go index 69dd22b3a..34a60d17f 100644 --- a/index/storage_test.go +++ b/index/storage_test.go @@ -2,7 +2,9 @@ package index import ( "fmt" + "net" "testing" + "time" // "io/ioutil" // "time" @@ -17,42 +19,44 @@ func TestStorage(t *testing.T) { filter := 10 bitmap_id := uint64(1234) /* Convey("KV ", t, func() { - storage, _ := NewKVStorage("/tmp/", 0, db) - bm := storage.Fetch(bitmap_id, db, slice) - SetBit(bm, 0) - SetBit(bm, 1) - SetBit(bm, 2) - storage.Store(int64(bitmap_id), db, frame, slice, filter, bm.(*Bitmap)) - bm2, _ := storage.Fetch(bitmap_id, db, slice) - So(BitCount(bm), ShouldEqual, BitCount(bm2)) - So(BitCount(bm), ShouldEqual, bm.Count()) - So(BitCount(bm), ShouldEqual, 3) + storage, _ := NewKVStorage("/tmp/", 0, db) + bm := storage.Fetch(bitmap_id, db, slice) + SetBit(bm, 0) + SetBit(bm, 1) + SetBit(bm, 2) + storage.Store(int64(bitmap_id), db, frame, slice, filter, bm.(*Bitmap)) + bm2, _ := storage.Fetch(bitmap_id, db, slice) + So(BitCount(bm), ShouldEqual, BitCount(bm2)) + So(BitCount(bm), ShouldEqual, bm.Count()) + So(BitCount(bm), ShouldEqual, 3) - }) - c, err := net.DialTimeout("tcp", "127.0.0.1:9042", 100*time.Millisecond) - if err != nil { - fmt.Println("NO cassandra. Skipping test.") - } else { - c.Close() - Convey("cassandra", t, func() { - fmt.Println("GO") - storage := NewCassStorage("127.0.0.1", "hotbox") - - fmt.Println("FETCH") - bm, _ := storage.Fetch(bitmap_id, db, frame, slice) - SetBit(bm, 0) - SetBit(bm, 1) - SetBit(bm, 2) - fmt.Println("STORE") - storage.Store(int64(bitmap_id), db, frame, slice, uint64(filter), bm.(*Bitmap)) - fmt.Println("FETCH") - bm2, _ := storage.Fetch(bitmap_id, db, frame, slice) - So(BitCount(bm), ShouldEqual, BitCount(bm2)) - So(BitCount(bm), ShouldEqual, bm.Count()) - So(BitCount(bm), ShouldEqual, 3) - - }) + }) */ + c, err := net.DialTimeout("tcp", "127.0.0.1:9042", 100*time.Millisecond) + if err != nil { + fmt.Println("NO cassandra. Skipping test.") + } else { + c.Close() + Convey("cassandra", t, func() { + fmt.Println("GO") + //storage := NewCassStorage("127.0.0.1", "hotbox") + storage := NewCassStorage() + + fmt.Println("FETCH") + bm, _ := storage.Fetch(bitmap_id, db, frame, slice) + SetBit(bm, 0) + SetBit(bm, 1) + SetBit(bm, 2) + fmt.Println("STORE") + storage.Store(int64(bitmap_id), db, frame, slice, uint64(filter), bm.(*Bitmap)) + fmt.Println("FETCH") + bm2, _ := storage.Fetch(bitmap_id, db, frame, slice) + So(BitCount(bm), ShouldEqual, BitCount(bm2)) + So(BitCount(bm), ShouldEqual, bm.Count()) + So(BitCount(bm), ShouldEqual, 3) + + }) + } Convey("leveldb", t, func() { storage := NewLevelDBStorage("./basic/one") diff --git a/query/parser.go b/query/parser.go index a0a413877..fac11a8ed 100644 --- a/query/parser.go +++ b/query/parser.go @@ -275,6 +275,12 @@ ArgLoop: if query.Operation == "mask" { return nil, fmt.Errorf("No Args Given") } + if query.Operation == "stash" { + return nil, fmt.Errorf("No Args Given") + } + if query.Operation == "recall" { + return nil, fmt.Errorf("No Args Given") + } } return query, nil } diff --git a/query/planner.go b/query/planner.go index 0f9fe1f6a..9415b913f 100644 --- a/query/planner.go +++ b/query/planner.go @@ -7,6 +7,7 @@ import ( "log" "math/rand" "pilosa/db" + "pilosa/index" "pilosa/util" "time" @@ -260,6 +261,9 @@ type CatQueryStep struct { Inputs []*util.GUID N int } +type Appendable interface { + Append(subq QueryTree) +} type CatQueryResult struct { *BaseQueryResult @@ -272,6 +276,10 @@ type CatQueryTree struct { N int } +func (qt *CatQueryTree) Append(subtree QueryTree) { + qt.subqueries = append(qt.subqueries, subtree) +} + // Uses consistent hashing function to select node containing data for GET operation func (qt *CatQueryTree) getLocation(d *db.Database) (*db.Location, error) { var err error @@ -367,6 +375,9 @@ func init() { gob.Register(CountQueryResult{}) gob.Register(TopNQueryResult{}) gob.Register(FillResult{}) + gob.Register(StashQueryResult{}) + gob.Register(CacheItem{}) + gob.Register(Stash{}) gob.Register(SetQueryStep{}) gob.Register(GetQueryStep{}) @@ -377,6 +388,7 @@ func init() { gob.Register(DifferenceQueryStep{}) gob.Register(CountQueryStep{}) gob.Register(TopNQueryStep{}) + gob.Register(StashQueryStep{}) } /////////////////////////////////////////////////////////////////////////////////////////////////// @@ -425,13 +437,22 @@ func (self *QueryPlanner) buildTree(query *Query, slice int) (QueryTree, error) } // handle the remaining operations, taking slice into consideration + //I'm kinda thinking for stash it needs a differnt handler..i guess for now I'll see if i can use the cathandler if slice == -1 { var n int n_, ok := query.Args["n"] if ok { n = n_.(int) } - tree = &CatQueryTree{N: n} + var p Appendable + + if query.Operation == "stash" { + // log.Println("STASH:", n) + tree = &StashQueryTree{N: n} + } else { + tree = &CatQueryTree{N: n} + } + p = tree.(Appendable) numSlices, err := self.Database.NumSlices() if err != nil { return nil, err @@ -442,8 +463,8 @@ func (self *QueryPlanner) buildTree(query *Query, slice int) (QueryTree, error) if err != nil { return nil, err } - composite := tree.(*CatQueryTree) - composite.subqueries = append(composite.subqueries, subtree) + + p.Append(subtree) } } else { if query.Operation == "get" { @@ -520,6 +541,17 @@ func (self *QueryPlanner) buildTree(query *Query, slice int) (QueryTree, error) } } tree = &DifferenceQueryTree{subqueries: subqueries} + } else if query.Operation == "stash" { + + subqueries := make([]QueryTree, len(query.Subqueries)) + var err error + for i, query := range query.Subqueries { + subqueries[i], err = self.buildTree(&query, slice) + if err != nil { + return nil, err + } + } + tree = &StashQueryTree{subqueries: subqueries} } else { //TODO return error gracefully log.Println(spew.Sdump(query)) @@ -549,6 +581,23 @@ func (self *QueryPlanner) flatten(qt QueryTree, id *util.GUID, location *db.Loca plan = append(plan, *subq_steps...) } plan = append(plan, step) + } else if stash, ok := qt.(*StashQueryTree); ok { + inputs := make([]*util.GUID, len(stash.subqueries)) + loc, err := stash.getLocation(self.Database) + if err != nil { + return nil, err + } + step := StashQueryStep{&BaseQueryStep{id, "stash", loc, location}, inputs, stash.N} + for index, subq := range stash.subqueries { + sub_id := util.RandomUUID() + step.Inputs[index] = &sub_id + subq_steps, err := self.flatten(subq, &sub_id, loc) + if err != nil { + return nil, err + } + plan = append(plan, *subq_steps...) + } + plan = append(plan, step) } else if union, ok := qt.(*UnionQueryTree); ok { inputs := make([]*util.GUID, len(union.subqueries)) loc, err := union.getLocation(self.Database) @@ -737,3 +786,50 @@ func (qt *RangeQueryTree) getLocation(d *db.Database) (*db.Location, error) { type FillResult struct { *BaseQueryResult } + +/////////////////////////////////////////////////////////////////////////////////////////////////// +//Stash +/////////////////////////////////////////////////////////////////////////////////////////////////// + +type CacheItem struct { + FragmentId util.SUUID + Handle index.BitmapHandle +} + +type Stash struct { + Stash []CacheItem //index.BitmapHandle //probably need to make the a struct with fragment_id and handle +} +type StashQueryStep struct { + *BaseQueryStep + Inputs []*util.GUID + N int +} + +type StashQueryResult struct { + *BaseQueryResult +} + +// QueryTree for UNION queries +type StashQueryTree struct { + subqueries []QueryTree + location *db.Location + N int +} + +func (qt *StashQueryTree) Append(subtree QueryTree) { + qt.subqueries = append(qt.subqueries, subtree) +} + +// Uses consistent hashing function to select node containing data for GET operation +func (qt *StashQueryTree) getLocation(d *db.Database) (*db.Location, error) { + var err error + if qt.location == nil { + subqueryLength := len(qt.subqueries) + if subqueryLength > 0 { + locationIndex := rand.Intn(subqueryLength) + subquery := qt.subqueries[locationIndex] + qt.location, err = subquery.getLocation(d) + } + } + return qt.location, err +} diff --git a/query/planner_test.go b/query/planner_test.go index 237f456c9..b01bdcab4 100644 --- a/query/planner_test.go +++ b/query/planner_test.go @@ -6,6 +6,7 @@ import ( "pilosa/util" "testing" + "github.com/davecgh/go-spew/spew" . "github.com/smartystreets/goconvey/convey" ) @@ -181,12 +182,11 @@ func TestQueryPlanner(t *testing.T) { So(err, ShouldEqual, nil) database, fragment1 := basic_database() - qplanner := QueryPlanner{Database: database, Query: query} destination := fragment1.GetLocation() - id := util.RandomUUID() qpp, err := qplanner.Plan(query, &id, destination) + So(err, ShouldEqual, nil) qp := *qpp So(err, ShouldEqual, nil) @@ -215,4 +215,21 @@ func TestQueryPlanner(t *testing.T) { So(err, ShouldNotEqual, nil) }) + Convey("Stash including parsing", t, func() { + qp, err := QueryForPQL("stash(union(get(10,default),get(20,default)))") + So(err, ShouldEqual, nil) + + database, fragment1 := basic_database() + qplanner := QueryPlanner{Database: database, Query: qp} + destination := fragment1.GetLocation() + id := util.RandomUUID() + qpp, _ := qplanner.Plan(qp, &id, destination) + p := *qpp + + So(len(p), ShouldNotEqual, 0) + log.Println(len(p)) + spew.Dump(p[0]) + //So(p[0].(StashQueryStep).Operation, ShouldEqual, "stash") + }) + }