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

added stash keyword

Browse source
This commit is contained in:
Todd Gruben 2014-08-29 16:29:21 +00:00
parent cc0078c0ad
commit cd6e4d07e8
7 changed files with 213 additions and 47 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

53
core/query.go
View file

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

2
deps.json
View file

@ -21,7 +21,7 @@
},
"gocql": {
"repo": "github.com/gocql/gocql",
"version": "f2deeb64d238e3a470f4489d734fab820cb44b4b",
"version": "7bd964ec83ae675586ba314806c02341c015c45b",
"type": "git"
},
"goleveldb": {

4
executor/executor.go
View file

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

72
index/storage_test.go
View file

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

6
query/parser.go
View file

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

102
query/planner.go
View file

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

21
query/planner_test.go
View file

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