From ef56ef6af38924b808dddd2f1cc3109de4331000 Mon Sep 17 00:00:00 2001 From: Todd Gruben Date: Tue, 1 Apr 2014 15:42:35 -0500 Subject: [PATCH] added filter(category) to data model --- core/batch.go | 5 +- core/http.go | 9 ++- core/query.go | 2 +- db/topology.go | 1 + dispatch/dispatch.go | 2 +- index/brand.go | 32 ++++---- index/commands.go | 14 ++-- index/fragment_container.go | 85 ++++---------------- index/fragment_container_test.go | 10 +-- index/general.go | 18 +++-- index/storage.go | 6 +- index/storage_cass.go | 23 +++--- index/storage_kvlite.go | 133 ------------------------------- index/storage_mem.go | 10 +-- index/storage_test.go | 37 +++++---- query/parser.go | 4 +- remote/remote_test.go | 26 ++---- 17 files changed, 117 insertions(+), 300 deletions(-) delete mode 100644 index/storage_kvlite.go diff --git a/core/batch.go b/core/batch.go index ed30a39e4..e5d1a1571 100644 --- a/core/batch.go +++ b/core/batch.go @@ -14,6 +14,7 @@ type BatchRequest struct { Fragment_id SUUID Bitmap_id uint64 Compressed_bitmap string + Filter int } type BatchResponse struct { @@ -32,13 +33,13 @@ func init() { gob.Register(BatchResponse{}) } -func (self *Service) Batch(database_name, frame, compressed_bitmap string, bitmap_id uint64, slice int) error { +func (self *Service) Batch(database_name, frame, compressed_bitmap string, bitmap_id uint64, slice int, filter int) error { //determine the fragment_id from database/frame/slice database := self.Cluster.GetOrCreateDatabase(database_name) oslice := database.GetOrCreateSlice(slice) //need to find processid and fragment id for that slice - fragment, err := database.GetFragmentForBitmap(oslice, &db.Bitmap{bitmap_id, frame}) + fragment, err := database.GetFragmentForBitmap(oslice, &db.Bitmap{bitmap_id, frame, filter}) if err == nil { id := uuid.RandomUUID() batch := db.Message{Data: BatchRequest{Id: &id, Source: self.Id, Fragment_id: fragment.GetId(), Bitmap_id: bitmap_id, Compressed_bitmap: compressed_bitmap}} diff --git a/core/http.go b/core/http.go index 41c4cd547..aa3117d75 100644 --- a/core/http.go +++ b/core/http.go @@ -113,7 +113,14 @@ func (self *WebService) HandleBatch(w http.ResponseWriter, r *http.Request) { return } - results := self.service.Batch(database_name, frame, compressed_bitmap, bitmap_id, int(slice)) + filter_string := r.Form.Get("filter") + filter, _ := strconv.ParseInt(filter_string, 10, 32) + if bitmap_id_string == "" { + http.Error(w, "Provide a filter for categories", http.StatusNotFound) + return + } + + results := self.service.Batch(database_name, frame, compressed_bitmap, bitmap_id, int(slice), int(filter)) encoder := json.NewEncoder(w) err = encoder.Encode(results) diff --git a/core/query.go b/core/query.go index 55a461778..735b1ec2f 100644 --- a/core/query.go +++ b/core/query.go @@ -213,7 +213,7 @@ func (self *Service) GetQueryStepHandler(msg *db.Message) { func (self *Service) SetQueryStepHandler(msg *db.Message) { //spew.Dump("SET QUERYSTEP") qs := msg.Data.(query.SetQueryStep) - result, _ := self.Index.SetBit(qs.Location.FragmentId, qs.Bitmap.Id, qs.ProfileId) + result, _ := self.Index.SetBit(qs.Location.FragmentId, qs.Bitmap.Id, qs.ProfileId, qs.Bitmap.Filter) result_message := db.Message{Data: query.SetQueryResult{&query.BaseQueryResult{Id: qs.Id, Data: result}}} self.Transport.Send(&result_message, qs.Destination.ProcessId) diff --git a/db/topology.go b/db/topology.go index 267ef4124..62e3ceec6 100644 --- a/db/topology.go +++ b/db/topology.go @@ -413,4 +413,5 @@ func (d *Database) GetSliceForProfile(profile_id uint64) (*Slice, error) { type Bitmap struct { Id uint64 FrameType string + Filter int } diff --git a/dispatch/dispatch.go b/dispatch/dispatch.go index 9c2bd1a44..72aadccdc 100644 --- a/dispatch/dispatch.go +++ b/dispatch/dispatch.go @@ -27,7 +27,7 @@ func (self *Dispatch) Run() { switch data := message.Data.(type) { case core.BatchRequest: response := db.Message{Data: core.BatchResponse{Id: data.Id}} - self.service.Index.LoadBitmap(data.Fragment_id, data.Bitmap_id, data.Compressed_bitmap) + self.service.Index.LoadBitmap(data.Fragment_id, data.Bitmap_id, data.Compressed_bitmap, data.Filter) self.service.Transport.Send(&response, data.Source) case core.PingRequest: pong := db.Message{Data: core.PongRequest{Id: data.Id}} diff --git a/index/brand.go b/index/brand.go index 3bf4fea5b..f9ace80be 100644 --- a/index/brand.go +++ b/index/brand.go @@ -29,6 +29,7 @@ func (p RankList) Less(i, j int) bool { return p[i].Count > p[j].Count } type Brand struct { bitmap_cache map[uint64]*Rank db string + frame string slice int storage Storage rankings RankList @@ -37,12 +38,12 @@ type Brand struct { threshold_length int threshold_idx int skip int - finder ICategoryFinder } -func NewBrand(db string, slice int, s Storage, threshold_len int, threshold int, skipp int, c ICategoryFinder) *Brand { +func NewBrand(db string, frame string, slice int, s Storage, threshold_len int, threshold int, skipp int) *Brand { f := new(Brand) f.storage = s + f.frame = frame f.slice = slice f.db = db f.rank_counter = 0 @@ -50,7 +51,6 @@ func NewBrand(db string, slice int, s Storage, threshold_len int, threshold int, f.threshold_length = threshold_len f.threshold_idx = threshold f.skip = skipp - f.finder = c f.Clear() //alloc the cache return f @@ -65,9 +65,9 @@ func (self *Brand) Get(bitmap_id uint64) IBitmap { return bm.bitmap } //I should fetch the category here..need to come up with a good source - b := self.storage.Fetch(bitmap_id, self.db, self.slice) + b, filter := self.storage.Fetch(bitmap_id, self.db, self.frame, self.slice) - self.cache_it(b, bitmap_id, self.finder.GetCategory(bitmap_id)) + self.cache_it(b, bitmap_id, filter) return b } @@ -89,13 +89,13 @@ func (self *Brand) trim() { } -func (self *Brand) SetBit(bitmap_id uint64, bit_pos uint64) bool { +func (self *Brand) SetBit(bitmap_id uint64, bit_pos uint64, filter int) bool { bm := self.Get(bitmap_id) change, chunk, address := SetBit(bm, bit_pos) if change { val := chunk.Value.Block[address.BlockIndex] - self.storage.StoreBlock(int64(bitmap_id), self.db, self.slice, int64(address.ChunkKey), int32(address.BlockIndex), int64(val)) - self.storage.StoreBlock(int64(bitmap_id), self.db, self.slice, COUNTER_KEY, 0, int64(bm.Count())) + self.storage.StoreBlock(int64(bitmap_id), self.db, self.frame, self.slice, filter, int64(address.ChunkKey), int32(address.BlockIndex), int64(val)) + self.storage.StoreBlock(int64(bitmap_id), self.db, self.frame, self.slice, filter, COUNTER_KEY, 0, int64(bm.Count())) if bm.Count() >= self.threshold_value { self.Rank() //need to optimize this @@ -114,10 +114,8 @@ func (self *Brand) Rank() { var list RankList for k, item := range self.bitmap_cache { - if k < 9223372036854775808 { //all demographics is greater than 2^63 - if item.bitmap.Count() > 50 { - list = append(list, &Rank{&Pair{k, item.bitmap.Count()}, item.bitmap, item.category}) - } + if item.bitmap.Count() > 50 { + list = append(list, &Rank{&Pair{k, item.bitmap.Count()}, item.bitmap, item.category}) } } sort.Sort(list) @@ -167,11 +165,9 @@ func (self *Brand) Stats() interface{} { "skip": self.skip} return stats } -func (self *Brand) Store(bitmap_id uint64, bm IBitmap) { - //oldbm:=self.Get(bitmap_id) - //nbm = Union(oldbm, bm) - self.storage.Store(int64(bitmap_id), self.db, self.slice, bm.(*Bitmap)) - self.cache_it(bm, bitmap_id, self.finder.GetCategory(bitmap_id)) +func (self *Brand) Store(bitmap_id uint64, bm IBitmap, filter int) { + self.storage.Store(int64(bitmap_id), self.db, self.frame, self.slice, filter, bm.(*Bitmap)) + self.cache_it(bm, bitmap_id, filter) } func (self *Brand) TopN(src_bitmap IBitmap, n int, categories []int) []Pair { @@ -279,7 +275,7 @@ func (self *Brand) getFileName() string { base = "." } - return fmt.Sprintf("%s/Brand.%s.%d.json", base, self.db, self.slice) + return fmt.Sprintf("%s/%s.%s.%d.json", base, self.db, self.frame, self.slice) } func (self *Brand) Persist() error { diff --git a/index/commands.go b/index/commands.go index 3401fcc68..cbea5d43c 100644 --- a/index/commands.go +++ b/index/commands.go @@ -99,14 +99,15 @@ type CmdSetBit struct { *Responder bitmap_id uint64 bit_pos uint64 + filter int } -func NewSetBit(bitmap_id uint64, bit_pos uint64) *CmdSetBit { - result := &CmdSetBit{NewResponder("SetBit"), bitmap_id, bit_pos} +func NewSetBit(bitmap_id uint64, bit_pos uint64, filter int) *CmdSetBit { + result := &CmdSetBit{NewResponder("SetBit"), bitmap_id, bit_pos, filter} return result } func (self *CmdSetBit) Execute(f *Fragment) Calculation { - return f.impl.SetBit(self.bitmap_id, self.bit_pos) + return f.impl.SetBit(self.bitmap_id, self.bit_pos, self.filter) } type CmdGetBytes struct { @@ -208,15 +209,16 @@ type CmdLoader struct { *Responder bitmap_id uint64 compressed_bitmap string + filter int } -func NewLoader(bitmap_id uint64, compressed_bitmap string) *CmdLoader { - return &CmdLoader{NewResponder("Loader"), bitmap_id, compressed_bitmap} +func NewLoader(bitmap_id uint64, compressed_bitmap string, filter int) *CmdLoader { + return &CmdLoader{NewResponder("Loader"), bitmap_id, compressed_bitmap, filter} } func (self *CmdLoader) Execute(f *Fragment) Calculation { nbm := NewBitmap() nbm.FromCompressString(self.compressed_bitmap) - f.impl.Store(self.bitmap_id, nbm) + f.impl.Store(self.bitmap_id, nbm, self.filter) return "ok" } diff --git a/index/fragment_container.go b/index/fragment_container.go index 9d709a753..ef886bf7e 100644 --- a/index/fragment_container.go +++ b/index/fragment_container.go @@ -9,6 +9,7 @@ import ( "log" "pilosa/config" . "pilosa/util" + "strings" "sync" "time" @@ -18,7 +19,6 @@ import ( type FragmentContainer struct { fragments map[SUUID]*Fragment - finder *CategoryFinder } func lookup(stmt *sql.Stmt, tile_id uint64) int { @@ -32,58 +32,11 @@ func lookup(stmt *sql.Stmt, tile_id uint64) int { } -type ICategoryFinder interface { - GetCategory(in uint64) int -} - -type CategoryFinder struct { - in chan uint64 - out chan int -} - -func NewCategoryFinder() *CategoryFinder { - ptr := new(CategoryFinder) - ptr.in = make(chan uint64) - ptr.out = make(chan int) - return ptr -} - -func (self *CategoryFinder) Start() { - connection := config.GetString("category_db_uri") - db, err := sql.Open("mysql", connection) - stmt, err := db.Prepare("select b.category_id from audience_brand b, accounts_tile t where b.id = t.object_id and t.id=?") - cache := lru.New(1000) - defer stmt.Close() - - for { - select { - case id := <-self.in: - category := 0 - if err == nil { - category, found := cache.Get(id) - if !found { - category = lookup(stmt, id) - cache.Add(id, category) - - } - } - self.out <- category - } - } -} - func NewFragmentContainer() *FragmentContainer { f := new(FragmentContainer) f.fragments = make(map[SUUID]*Fragment) - f.finder = NewCategoryFinder() - go f.finder.Start() - return f } -func (self *FragmentContainer) GetCategory(in uint64) int { - self.finder.in <- in - return <-self.finder.out -} type BitmapHandle uint64 @@ -106,9 +59,9 @@ func (self *FragmentContainer) Shutdown() { log.Println("Container Shutdown Complete") } -func (self *FragmentContainer) LoadBitmap(frag_id SUUID, bitmap_id uint64, compressed_bitmap string) { +func (self *FragmentContainer) LoadBitmap(frag_id SUUID, bitmap_id uint64, compressed_bitmap string, filter int) { if fragment, found := self.GetFragment(frag_id); found { - request := NewLoader(bitmap_id, compressed_bitmap) + request := NewLoader(bitmap_id, compressed_bitmap, filter) fragment.requestChan <- request request.Response() return @@ -212,9 +165,9 @@ func (self *FragmentContainer) FromBytes(frag_id SUUID, bytes []byte) (BitmapHan return 0, errors.New("Invalid Bitmap Handle") } -func (self *FragmentContainer) SetBit(frag_id SUUID, bitmap_id uint64, pos uint64) (bool, error) { +func (self *FragmentContainer) SetBit(frag_id SUUID, bitmap_id uint64, pos uint64, category int) (bool, error) { if fragment, found := self.GetFragment(frag_id); found { - request := NewSetBit(bitmap_id, pos) + request := NewSetBit(bitmap_id, pos, category) fragment.requestChan <- request return request.Response().answer.(bool), nil } @@ -232,7 +185,7 @@ func (self *FragmentContainer) Clear(frag_id SUUID) (bool, error) { func (self *FragmentContainer) AddFragment(db string, frame string, slice int, id SUUID) { log.Println("ADD FRAGMENT", frame) - f := NewFragment(id, db, slice, frame, self) + f := NewFragment(id, db, slice, frame) self.fragments[id] = f go f.ServeFragment() @@ -241,10 +194,10 @@ func (self *FragmentContainer) AddFragment(db string, frame string, slice int, i type Pilosa interface { Get(id uint64) IBitmap - SetBit(id uint64, bit_pos uint64) bool + SetBit(id uint64, bit_pos uint64, filter int) bool TopN(b IBitmap, n int, categories []int) []Pair Clear() bool - Store(bitmap_id uint64, bm IBitmap) + Store(bitmap_id uint64, bm IBitmap, filter int) Stats() interface{} Persist() error Load(requestChan chan Command, fragment *Fragment) @@ -268,13 +221,6 @@ func getStorage(db string, slice int, frame string) Storage { switch storage_method { default: return NewMemoryStorage() - case "localfile": - storage_path := config.GetString("kv_base_path") - if storage_path == "" { - storage_path = "/tmp/pilosa" - } - s, _ := NewKVStorage(storage_path, slice, db) - return s case "cassandra": host := config.GetString("cassandra_host") if host == "" { @@ -289,16 +235,15 @@ func getStorage(db string, slice int, frame string) Storage { return nil } -func NewFragment(frag_id SUUID, db string, slice int, frame string, p ICategoryFinder) *Fragment { +func NewFragment(frag_id SUUID, db string, slice int, frame string) *Fragment { var impl Pilosa log.Println(fmt.Sprintf("XXXXXXXXXXXXXXXXXXXXXXXXXXX(%s)", frame)) - switch frame { - case "brand": - log.Println("Brand") - impl = NewBrand(db, slice, getStorage(db, slice, frame), 50000, 45000, 100, p) - default: - log.Println("General") - impl = NewGeneral(db, slice, getStorage(db, slice, frame)) + if strings.HasSuffix(frame, ".") { + log.Println(frame + "TOP") + impl = NewBrand(db, frame, slice, getStorage(db, slice, frame), 50000, 45000, 100) + } else { + log.Println(frame) + impl = NewGeneral(db, frame, slice, getStorage(db, slice, frame)) } f := new(Fragment) diff --git a/index/fragment_container_test.go b/index/fragment_container_test.go index e769b1ff0..0f76b825b 100644 --- a/index/fragment_container_test.go +++ b/index/fragment_container_test.go @@ -26,9 +26,9 @@ func TestFragment(t *testing.T) { Convey("SetBit/Count 1 1", t, func() { // bh, _ := dummy.Get(id, 1234) bi1 := uint64(1234) - changed, _ := dummy.SetBit(general, bi1, 1) + changed, _ := dummy.SetBit(general, bi1, 1, 0) So(changed, ShouldEqual, true) - changed, _ = dummy.SetBit(general, bi1, 1) + changed, _ = dummy.SetBit(general, bi1, 1, 0) So(changed, ShouldEqual, false) bh, _ := dummy.Get(general, bi1) num, _ := dummy.Count(general, bh) @@ -39,7 +39,7 @@ func TestFragment(t *testing.T) { bi1 := uint64(1234) bi2 := uint64(4321) - dummy.SetBit(general, bi2, 65537) //set_bit creates the bitmap + dummy.SetBit(general, bi2, 65537, 0) //set_bit creates the bitmap bh1, _ := dummy.Get(general, bi1) bh2, _ := dummy.Get(general, bi2) @@ -94,7 +94,7 @@ func TestFragment(t *testing.T) { Convey("Brand SetBit Small", t, func() { bi1 := uint64(1029) for x := uint64(0); x < 1000; x++ { - dummy.SetBit(brand, bi1, x) + dummy.SetBit(brand, bi1, x, 0) } So(1, ShouldEqual, 1) }) @@ -153,7 +153,7 @@ func TestFragment(t *testing.T) { compressed := "H4sIAAAJbogA/2JmYRBQ+9/IzMjI6pxRmpfN+L+JgZGJkdk7tZKRjYGRNSwxpzSV8X8LAwOD8v9moDIup5z85GzHoqLESpAwI1AjWITxfxtQjdT/VqAIV7SxUWxpZl6JmQlImJGN0YGB4R+j+v8mJkaFH/8h4B+M8X+UgcwAhZTm/yZgMCLCarC4bbAxYGHFNBpWBBmwsGIeDSuCDFhYsYyGFUEGLKxYR8OKIAMWVmyjYUWQwcDwfyYwqNgHLKjk4Q7BBAAAAAD//wEAAP//QNipzzcJAAA=" // compressed := "H4sIAAAJbogA/2JmYWBS+9/IzMjI6pxRmpfN+L+JgZGJkdk7tZKRjYGRNSwxpzSV8X8LAwOD8v9moDIup5z85GzHoqLESpAwI1AjWITxfxtQjdT/VqAIV7SxUWxpZl6JmQlImJGN0YGBweN/E+M/RgdGRoUf/6EAk/GbGUQy4AUAAAAA//8BAAD//2vjG9ezAAAA" - dummy.LoadBitmap(brand, b, compressed) + dummy.LoadBitmap(brand, b, compressed, 0) bh1, _ := dummy.Get(brand, b) before, _ := dummy.Count(brand, bh1) So(15228, ShouldEqual, before) diff --git a/index/general.go b/index/general.go index 91f96dcf7..e1892a9cf 100644 --- a/index/general.go +++ b/index/general.go @@ -14,13 +14,15 @@ type General struct { bitmap_cache *lru.Cache keys map[uint64]interface{} db string + frame string slice int storage Storage } -func NewGeneral(db string, slice int, s Storage) *General { +func NewGeneral(db string, frame string, slice int, s Storage) *General { f := new(General) f.storage = s + f.frame = frame f.slice = slice f.db = db f.Clear() @@ -40,18 +42,18 @@ func (self *General) Get(bitmap_id uint64) IBitmap { if ok { return bm.(*Bitmap) } - bm = self.storage.Fetch(bitmap_id, self.db, self.slice) + bm, _ = self.storage.Fetch(bitmap_id, self.db, self.frame, self.slice) self.bitmap_cache.Add(bitmap_id, bm) self.keys[bitmap_id] = 0 return bm.(*Bitmap) } -func (self *General) SetBit(bitmap_id uint64, bit_pos uint64) bool { +func (self *General) SetBit(bitmap_id uint64, bit_pos uint64, filter int) bool { bm := self.Get(bitmap_id) change, chunk, address := SetBit(bm, bit_pos) if change { val := chunk.Value.Block[address.BlockIndex] - self.storage.StoreBlock(int64(bitmap_id), self.db, self.slice, int64(address.ChunkKey), int32(address.BlockIndex), int64(val)) - self.storage.StoreBlock(int64(bitmap_id), self.db, self.slice, COUNTER_KEY, 0, int64(bm.Count())) + self.storage.StoreBlock(int64(bitmap_id), self.db, self.frame, self.slice, filter, int64(address.ChunkKey), int32(address.BlockIndex), int64(val)) + self.storage.StoreBlock(int64(bitmap_id), self.db, self.frame, self.slice, filter, COUNTER_KEY, 0, int64(bm.Count())) } return change @@ -60,10 +62,10 @@ func (self *General) TopN(b IBitmap, n int, categories []int) []Pair { return nil } -func (self *General) Store(bitmap_id uint64, bm IBitmap) { +func (self *General) Store(bitmap_id uint64, bm IBitmap, filter int) { //oldbm:=self.Get(bitmap_id) //nbm = Union(oldbm, bm) - self.storage.Store(int64(bitmap_id), self.db, self.slice, bm.(*Bitmap)) + self.storage.Store(int64(bitmap_id), self.db, self.frame, self.slice, filter, bm.(*Bitmap)) self.bitmap_cache.Add(bitmap_id, bm) self.keys[bitmap_id] = 0 } @@ -83,7 +85,7 @@ func (self *General) getFileName() string { if base == "" { base = "." } - return fmt.Sprintf("%s/General.%s.%d", base, self.db, self.slice) + return fmt.Sprintf("%s/%s.%s.%d", base, self.db, self.frame, self.slice) } func (self *General) Persist() error { diff --git a/index/storage.go b/index/storage.go index 3f33a8522..5ed4cefaf 100644 --- a/index/storage.go +++ b/index/storage.go @@ -1,7 +1,7 @@ package index type Storage interface { - Fetch(bitmap_id uint64, db string, slice int) IBitmap - Store(id int64, db string, slice int, bitmap *Bitmap) error - StoreBlock(id int64, db string, slice int, chunk int64, block_index int32, block int64) error + Fetch(bitmap_id uint64, db string, frame string, slice int) (IBitmap, int) + Store(id int64, db string, frame string, slice int, filter int, bitmap *Bitmap) error + StoreBlock(id int64, db string, frame string, slice int, filter int, chunk int64, block_index int32, block int64) error } diff --git a/index/storage_cass.go b/index/storage_cass.go index 5f7d93451..9293c3e9e 100644 --- a/index/storage_cass.go +++ b/index/storage_cass.go @@ -16,7 +16,7 @@ type CassandraStorage struct { func BuildSchema() { /* "CREATE KEYSPACE IF NOT EXISTS hotbox WITH strategy_class = SimpleStrategy AND strategy_options:replication_factor = 1" - CREATE TABLE IF NOT EXISTS bitmap ( bitmap_id bigint, db varchar, slice int, ChunkKey bigint, BlockIndex int, block bigint, PRIMARY KEY ((bitmap_id, db, slice),ChunkKey,BlockIndex) ) + CREATE TABLE IF NOT EXISTS bitmap ( bitmap_id bigint, db varchar, frame varchar, slice int, filter int, ChunkKey bigint, BlockIndex int, block bigint, PRIMARY KEY ((bitmap_id, db, frame,slice),ChunkKey,BlockIndex) ) " */ @@ -41,7 +41,7 @@ func NewCassStorage(host, keyspace string) Storage { return obj } -func (c *CassandraStorage) Fetch(bitmap_id uint64, db string, slice int) IBitmap { +func (c *CassandraStorage) Fetch(bitmap_id uint64, db string, frame string, slice int) (IBitmap, int) { var dumb = COUNTERMASK last_key := int64(dumb) marker := int64(dumb) @@ -52,13 +52,15 @@ func (c *CassandraStorage) Fetch(bitmap_id uint64, db string, slice int) IBitmap chunk_key, block int64 block_index uint32 s8 uint8 + filter int ) log.Println("FETCHING ", bitmap_id, db, slice) bitmap := CreateRBBitmap() - iter := c.db.Query("SELECT Chunkkey,BlockIndex,block FROM bitmap WHERE bitmap_id=? AND db=? AND slice=? ", id, db, slice).Iter() + iter := c.db.Query("SELECT filter,Chunkkey,BlockIndex,block FROM bitmap WHERE bitmap_id=? AND db=? AND frame=? AND slice=? ", id, db, frame, slice).Iter() count := int64(0) - for iter.Scan(&chunk_key, &block_index, &block) { + + for iter.Scan(&filter, &chunk_key, &block_index, &block) { s8 = uint8(block_index) if chunk_key != marker { if chunk_key != last_key { @@ -74,17 +76,17 @@ func (c *CassandraStorage) Fetch(bitmap_id uint64, db string, slice int) IBitmap } bitmap.SetCount(uint64(count)) - return bitmap + return bitmap, filter } -func (c *CassandraStorage) Store(id int64, db string, slice int, bitmap *Bitmap) error { +func (c *CassandraStorage) Store(id int64, db string, frame string, slice int, filter int, bitmap *Bitmap) error { for i := bitmap.Min(); !i.Limit(); i = i.Next() { var chunk = i.Item() for idx, block := range chunk.Value.Block { block_index := int32(idx) iblock := int64(block) if iblock != 0 { - c.StoreBlock(id, db, slice, int64(chunk.Key), block_index, iblock) + c.StoreBlock(id, db, frame, slice, filter, int64(chunk.Key), block_index, iblock) } } } @@ -93,13 +95,14 @@ func (c *CassandraStorage) Store(id int64, db string, slice int, bitmap *Bitmap) var dumb = COUNTERMASK COUNTER_KEY := int64(dumb) - c.StoreBlock(id, db, slice, COUNTER_KEY, 0, cnt) + c.StoreBlock(id, db, frame, slice, filter, COUNTER_KEY, 0, cnt) return nil } -func (c *CassandraStorage) StoreBlock(id int64, db string, slice int, chunk int64, block_index int32, block int64) error { +func (c *CassandraStorage) StoreBlock(id int64, db string, frame string, slice int, filter int, chunk int64, block_index int32, block int64) error { - if err := c.db.Query(`INSERT INTO bitmap (bitmap_id, db, slice , ChunkKey, BlockIndex,block) VALUES (?,?,?, ?,?,?);`, id, db, slice, chunk, block_index, block).Exec(); err != nil { + if err := c.db.Query(` + INSERT INTO bitmap ( bitmap_id, db, frame, slice , filter, ChunkKey, BlockIndex, block) VALUES (?,?,?,?,?,?,?,?);`, id, db, frame, slice, filter, chunk, block_index, block).Exec(); err != nil { log.Println(err) log.Println("INSERT ", id, chunk, block_index) return err diff --git a/index/storage_kvlite.go b/index/storage_kvlite.go deleted file mode 100644 index 89864edfb..000000000 --- a/index/storage_kvlite.go +++ /dev/null @@ -1,133 +0,0 @@ -package index - -// #cgo CFLAGS:-mpopcnt - -import ( - "bytes" - "encoding/binary" - "fmt" - "os" - - "github.com/steveyen/gkvlite" -) - -type KVStorage struct { - db *gkvlite.Store - cc *gkvlite.Collection - evic_count int -} - -func NewKVStorage(path string, slice int, db string) (Storage, error) { - // log.Println("Hello") - obj := new(KVStorage) - - base_path := fmt.Sprintf("%s/%s", path, db) - os.MkdirAll(base_path, 0755) - - f, err := os.Create(fmt.Sprintf("%s/%d.kvlite", base_path, slice)) - s, err := gkvlite.NewStore(f) - obj.cc = s.SetCollection(db, nil) - - obj.db = s - - return obj, err -} - -func (self *KVStorage) Fetch(bitmap_id uint64, db string, slice int) IBitmap { - - var ( - chunk *Chunk - ) - last_key := COUNTER_KEY - marker := COUNTER_KEY - count := uint64(0) - - key, _ := toKeyBytes(int64(bitmap_id), slice, 0, 0) - - bitmap := CreateRBBitmap() - - self.cc.VisitItemsAscend(key, true, func(i *gkvlite.Item) bool { - bid, _, chunk_key, block_index, _ := fromKeyBytes(i.Key) - if bid != int64(bitmap_id) { - return false - } - block, _ := binary.Uvarint(i.Val) //just need to cast as a block - if chunk_key != marker { - if chunk_key != last_key { - chunk = &Chunk{uint64(chunk_key), BlockArray{}} - bitmap.AddChunk(chunk) - } - chunk.Value.Block[block_index] = block - - } else { - count = block - } - last_key = chunk_key - return true - }) - bitmap.SetCount(uint64(count)) - return bitmap -} - -func (self *KVStorage) Store(bitmap_id int64, db string, slice int, bitmap *Bitmap) error { - for i := bitmap.Min(); !i.Limit(); i = i.Next() { - var chunk = i.Item() - for idx, block := range chunk.Value.Block { - block_index := int32(idx) - iblock := int64(block) - if iblock != 0 { - self.StoreBlock(bitmap_id, db, slice, int64(chunk.Key), block_index, iblock) - } - } - } - cnt := int64(BitCount(bitmap)) - - //var dumb = COUNTERMASK - //COUNTER_KEY := int64(dumb) - - self.StoreBlock(bitmap_id, db, slice, COUNTER_KEY, 0, cnt) - self.db.Flush() - return nil - -} -func Bytes(src int64) []byte { - buf := new(bytes.Buffer) - err := binary.Write(buf, binary.LittleEndian, src) - if err != nil { - fmt.Println("binary.Write failed:", err) - } - return buf.Bytes() -} -func fromKeyBytes(src []byte) (bitmap_id int64, slice int, chunk_key int64, block_index int32, err error) { - buf := bytes.NewReader(src) - err = binary.Read(buf, binary.LittleEndian, &bitmap_id) - err = binary.Read(buf, binary.LittleEndian, &slice) - err = binary.Read(buf, binary.LittleEndian, &chunk_key) - err = binary.Read(buf, binary.LittleEndian, &block_index) - return -} - -func toKeyBytes(bitmap_id int64, slice int, chunk_key int64, block_index int32) ([]byte, error) { - buf := new(bytes.Buffer) - err := binary.Write(buf, binary.LittleEndian, bitmap_id) - err = binary.Write(buf, binary.LittleEndian, slice) - err = binary.Write(buf, binary.LittleEndian, chunk_key) - err = binary.Write(buf, binary.LittleEndian, block_index) - - return buf.Bytes(), err -} - -func (self *KVStorage) StoreBlock(bitmap_id int64, db string, slice int, chunk_key int64, block_index int32, block int64) error { - //key := fmt.Sprintf("%d%d:%d:%d", bitmap_id, slice, chunk_key, block_index) - key, _ := toKeyBytes(bitmap_id, slice, chunk_key, block_index) - value := Bytes(block) - - self.cc.Set(key, value) - self.evic_count += 1 - if self.evic_count > 10000 { - self.db.Flush() - self.cc.EvictSomeItems() - self.evic_count = 0 - } - return nil -} diff --git a/index/storage_mem.go b/index/storage_mem.go index 00b4f13da..bd8bf31a7 100644 --- a/index/storage_mem.go +++ b/index/storage_mem.go @@ -16,24 +16,24 @@ func NewMemoryStorage() Storage { return obj } -func (c *MemoryStorage) Fetch(bitmap_id uint64, db string, slice int) IBitmap { +func (c *MemoryStorage) Fetch(bitmap_id uint64, db string, frame string, slice int) (IBitmap, int) { // log.Println("hello") - key := fmt.Sprintf("%d:%s:%d", bitmap_id, db, slice) + key := fmt.Sprintf("%d:%s:%s:%d", bitmap_id, db, frame, slice) bitmap, found := c.db[key] if !found { bitmap = CreateRBBitmap().(*Bitmap) c.db[key] = bitmap } - return bitmap + return bitmap, 0 } -func (c *MemoryStorage) Store(bitmap_id int64, db string, slice int, bitmap *Bitmap) error { +func (c *MemoryStorage) Store(bitmap_id int64, db string, frame string, slice int, filter int, bitmap *Bitmap) error { //only use the cache and throw away everything return nil } -func (c *MemoryStorage) StoreBlock(bitmap_id int64, db string, slice int, chunk_key int64, block_index int32, block int64) error { +func (c *MemoryStorage) StoreBlock(bitmap_id int64, db string, frame string, slice int, filter int, chunk_key int64, block_index int32, block int64) error { //only use the cache and throw away everything return nil diff --git a/index/storage_test.go b/index/storage_test.go index e95c162f5..c654e4e45 100644 --- a/index/storage_test.go +++ b/index/storage_test.go @@ -13,35 +13,42 @@ import ( func TestStorage(t *testing.T) { db := "db" + frame := "main" slice := 0 + 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, slice, 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) + /* 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) - }) + }) + */ 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") - bm := storage.Fetch(bitmap_id, db, slice) + fmt.Println("FETCH") + bm, _ := storage.Fetch(bitmap_id, db, frame, slice) SetBit(bm, 0) SetBit(bm, 1) SetBit(bm, 2) - storage.Store(int64(bitmap_id), db, slice, bm.(*Bitmap)) - bm2 := storage.Fetch(bitmap_id, db, slice) + fmt.Println("STORE") + storage.Store(int64(bitmap_id), db, frame, slice, 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) diff --git a/query/parser.go b/query/parser.go index 08ceea04a..ea6565f11 100644 --- a/query/parser.go +++ b/query/parser.go @@ -39,7 +39,9 @@ func (qp *QueryParser) walkInputs(tokens []Token) ([]QueryInput, uint64, int) { panic(err) } } - bm := db.Bitmap{bitmap_id, frame_type} + var filter int //TRAVIS the is for the category + filter = 0 + bm := db.Bitmap{bitmap_id, frame_type, filter} return []QueryInput{&bm}, uint64(profile_id), 0 } diff --git a/remote/remote_test.go b/remote/remote_test.go index 5173bada6..2d809506b 100644 --- a/remote/remote_test.go +++ b/remote/remote_test.go @@ -30,27 +30,11 @@ func TestRemote(t *testing.T) { So(true, ShouldEqual, true) } }) - /* - Convey("BigFile", t, func() { - ssh, err := New("50.16.204.123:22", "todd", "id_dsa") + Convey("BigFile", t, func() { + ssh, _ := New("50.16.204.123:22", "todd", "id_dsa") - fo, err := os.Create("outbin") - if err != nil { - panic(err) - } - // close fo on exit and check for its returned error - defer func() { - if err := fo.Close(); err != nil { - panic(err) - } - }() - // make a write buffer - w := bufio.NewWriter(fo) - ssh.CopyFrom("pilosa-cruncher", w) - if err = w.Flush(); err != nil { - panic(err) - } + ssh.SimpleFileCopyTo("outbin", "tb") + ssh.SimpleFileCopyFrom("tb", "outbin2") - }) - */ + }) }