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

added filter(category) to data model

Browse source
This commit is contained in:
Todd Gruben 2014-04-01 15:42:35 -05:00
parent a85a469252
commit ef56ef6af3
17 changed files with 117 additions and 300 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
core/batch.go
View file

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

9
core/http.go
View file

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

2
core/query.go
View file

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

1
db/topology.go
View file

@ -413,4 +413,5 @@ func (d *Database) GetSliceForProfile(profile_id uint64) (*Slice, error) {
type Bitmap struct {
Id uint64
FrameType string
Filter int
}

2
dispatch/dispatch.go
View file

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

32
index/brand.go
View file

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

14
index/commands.go
View file

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

85
index/fragment_container.go
View file

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

10
index/fragment_container_test.go
View file

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

18
index/general.go
View file

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

6
index/storage.go
View file

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

23
index/storage_cass.go
View file

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

133
index/storage_kvlite.go
View file

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

10
index/storage_mem.go
View file

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

37
index/storage_test.go
View file

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

4
query/parser.go
View file

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

26
remote/remote_test.go
View file

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