diff --git a/index/bitmap.go b/index/bitmap.go index ebd51f73c..a236ddfa5 100644 --- a/index/bitmap.go +++ b/index/bitmap.go @@ -163,6 +163,9 @@ func Invert(a_bm IBitmap) IBitmap { return output } +func NewBitmap()IBitmap{ + return CreateRBBitmap() +} func Union(a_bm IBitmap, b_bm IBitmap) IBitmap { var a = a_bm.Min() diff --git a/index/commands.go b/index/commands.go new file mode 100644 index 000000000..8c120e16a --- /dev/null +++ b/index/commands.go @@ -0,0 +1,150 @@ +package index + +import ( + "encoding/json" + "log" + "fmt" +) + +type Rank struct { + Key, Count uint64 +} + +type RankList []Rank + +func (p RankList) Swap(i, j int) { p[i], p[j] = p[j], p[i] } +func (p RankList) Len() int { return len(p) } +func (p RankList) Less(i, j int) bool { return p[i].Count > p[j].Count } + + +type Command interface { + Execute(*Fragment)string + QueryType() string + Response() string + ResponseChannel() chan string +} + + +func BuildCommandFactory(req *RequestJSON,decoder *json.Decoder)Command{ + log.Println(req) + var result Command + + switch req.Request{ + default: + result=&CmdUnknown{make(chan string),req.Request} + case "UnionCount": + result= NewUnion(decoder) + case "IntersectCount": + result= NewIntersect(decoder) + } + return result + } + + +type CmdUnknown struct{ + result chan string + response string +} +func (cmd *CmdUnknown) Execute(f *Fragment)string { + return fmt.Sprintf(`{ "Unknown Command":"%s" }`,cmd.response) +} +func (cmd *CmdUnknown) QueryType()string { + return "UnknownCommand" +} +func (cmd *CmdUnknown) Response()string { + return <-cmd.result +} +func (cmd *CmdUnknown) ResponseChannel()chan string { + return cmd.result +} + +type CmdUnion struct{ + result chan string + bitmap_ids []uint64 +} +type Args struct { + Bitmaps[] uint64 +} + + +func NewUnion(decoder *json.Decoder) *CmdUnion{ + var f Args + decoder.Decode(&f) + + result:= &CmdUnion{make(chan string),f.Bitmaps} + return result +} +func (cmd *CmdUnion) Execute(f *Fragment)string { + bm:=f.impl.Union(cmd.bitmap_ids) + result := BitCount(bm) + return fmt.Sprintf(`{ "value":%d }`,result) +} +func (cmd *CmdUnion) QueryType()string { + return "UnionCount" +} +func (cmd *CmdUnion) Response()string { + return <-cmd.result +} +func (cmd *CmdUnion) ResponseChannel()chan string { + return cmd.result +} + + +type CmdIntersect struct{ + result chan string + bitmaps []uint64 +} + +func NewIntersect(decoder *json.Decoder) *CmdIntersect{ + var f Args + decoder.Decode(&f) + + result:= &CmdIntersect{ make(chan string), f.Bitmaps } + return result +} +func (cmd *CmdIntersect) Execute(f *Fragment)string { + bm:=f.impl.Intersect(cmd.bitmaps) + result := BitCount(bm) + return fmt.Sprintf(`{ "value":%d }`,result) +} +func (cmd *CmdIntersect) QueryType()string { + return "IntersectCount" +} +func (cmd *CmdIntersect) Response()string { + return <-cmd.result +} +func (cmd *CmdIntersect) ResponseChannel()chan string { + return cmd.result +} +/* +type CmdSetBit struct{ + result chan string + id uint64 + bit_pos uint64 +} + +func NewSetBit(decoder *json.Decoder) *CmdSetBit{ + var f interface{} + decoder.Decode(&f) + m := f.(map[string]interface{}) + + + result:= &CmdSetBit{make(chan string),f["bitmap_id"].(uint64),f["bit_pos"].(uint64)} + return result +} +func (cmd *CmdSetBit) Execute(f *Fragment)string { + bitmap := Get(f.impl,cmd.id) + SetBit(bitmap,cmd.bit_pos) + result := BitCount(bm) + return fmt.Sprintf(`{ "value":%d }`,result) +} +func (cmd *CmdSetBit) QueryType()string { + return "SetBit" +} +func (cmd *CmdSetBit) Response()string { + return <-cmd.result +} +func (cmd *CmdSetBit) ResponseChannel()chan string { + return cmd.result +} +*/ diff --git a/index/general.go b/index/general.go new file mode 100644 index 000000000..b6e345b8d --- /dev/null +++ b/index/general.go @@ -0,0 +1,59 @@ +package index + +import ( + "github.com/golang/groupcache/lru" +) + +type General struct { + bitmap_cache *lru.Cache + db string + slice int + storage Storage +} + +func NewGeneral(db string, slice int, s Storage) *General { + f := new(General) + f.bitmap_cache = lru.New(10000) + f.storage = s + f.slice = slice + f.db = db + return f + +} + +func (f *General) Union(bitmaps[]uint64) IBitmap { + result := NewBitmap() + for i,id:= range bitmaps{ + bm:=f.Get(id) + if i == 0{ + result = bm + } else { + result = Union(result,bm) + } + } + return result +} + +func (f *General) Intersect(bitmaps[]uint64) IBitmap{ + result := NewBitmap() + for i,id:= range bitmaps{ + bm:=f.Get(id) + if i == 0{ + result = bm + } else { + result = Intersection(result,bm) + } + } + return result +} + +func (f *General) Get(bitmap_id uint64) IBitmap { + bm, ok := f.bitmap_cache.Get(bitmap_id) + if ok { + return bm.(*Bitmap) + } + bm = f.storage.Fetch(bitmap_id, f.db,f.slice) + f.bitmap_cache.Add(bitmap_id, bm) + return bm.(*Bitmap) +} + diff --git a/index/server.go b/index/server.go new file mode 100644 index 000000000..1328029f0 --- /dev/null +++ b/index/server.go @@ -0,0 +1,122 @@ +package index + +import ( + "bytes" + "encoding/json" + "fmt" + "io/ioutil" + "net" + "net/http" + //"sort" + "time" + "log" +) + +var ( + fragments[] *Fragment +) +type Pilosa interface{ + Union([]uint64) IBitmap + Intersect([] uint64) IBitmap + // SetBit(id uint64, bit_pos int64)bool +} + +type RequestJSON struct { + Request string + FragmentIndex int + Args json.RawMessage +} +type Fragment struct { + requestChan chan Command + shardkey int + impl Pilosa +} + +func (f *Fragment) ServeFragment() { + for { + req := <-f.requestChan + start := time.Now() + answer := `""` + answer = req.Execute(f) + delta := time.Since(start) + var buffer bytes.Buffer + buffer.WriteString(`{ "results":`) + buffer.WriteString(answer) + buffer.WriteString(fmt.Sprintf(`,"query type": "%s"`, req.QueryType())) + buffer.WriteString(fmt.Sprintf(`, "elapsed": "%s"}`, delta)) + req.ResponseChannel() <- buffer.String() + } +} + +func handler(w http.ResponseWriter, r *http.Request) { + log.Println("GOT MESSAGE") + if r.Method == "POST" { + var f RequestJSON + + bin, _ := ioutil.ReadAll(r.Body) + err := json.Unmarshal(bin, &f) + + if err != nil { + w.Header().Set("Content-Type", "application/json") + fmt.Fprintf(w, fmt.Sprintf(`{ "error":"%s"}`, err)) + + } + decoder := json.NewDecoder(bytes.NewReader(f.Args)) + request := BuildCommandFactory(&f, decoder) + w.Header().Set("Content-Type", "application/json") + if request != nil { + output:=`{"Error":"Invalid FragmentIndex"}` + if f.FragmentIndex>>>DONE") + + + } + + func TestServer(t *testing.T) { + Convey("Run Server", t, func() { + Stop := make(chan bool) + Start := make(chan bool) + go StartServer(":8089",Stop,Start) + select{ + case <-Start: + simple() + case <-time.After(time.Duration(5) * time.Second): + } + Stop<- true + So(0, ShouldEqual, 0) + }) +} diff --git a/index/storage.go b/index/storage.go index 979539358..3f33a8522 100644 --- a/index/storage.go +++ b/index/storage.go @@ -1,7 +1,7 @@ package index type Storage interface { - Fetch(bitmap_id uint64, shard int32) IBitmap - Store(id int64, shard_key int32, bitmap *Bitmap) error - StoreBlock(id int64, shard_key int32, chunk int64, block_index int32, block int64) error + 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 } diff --git a/index/storage_cass.go b/index/storage_cass.go index c9010a167..f83bb5aa2 100644 --- a/index/storage_cass.go +++ b/index/storage_cass.go @@ -13,7 +13,8 @@ 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, shard_id int, ChunkKey bigint, BlockIndex int, block bigint, PRIMARY KEY ((bitmap_id, shard_id),ChunkKey,BlockIndex) )" + "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) ) + " */ } @@ -36,7 +37,7 @@ func NewCassStorage() Storage{ return obj } -func (c *CassandraStorage)Fetch( bitmap_id uint64, shard int32) IBitmap { +func (c *CassandraStorage)Fetch( bitmap_id uint64, db string, slice int) IBitmap { var dumb = COUNTERMASK last_key := int64(dumb) marker := int64(dumb) @@ -48,10 +49,10 @@ func (c *CassandraStorage)Fetch( bitmap_id uint64, shard int32) IBitmap { block_index uint32 s8 uint8 ) - log.Println("FETCHING ", bitmap_id, shard) + log.Println("FETCHING ", bitmap_id, db,slice) bitmap := CreateRBBitmap() - iter := c.db.Query("SELECT Chunkkey,BlockIndex,block FROM bitmap WHERE bitmap_id=? AND shard_id=? ", id, shard).Iter() + iter := c.db.Query("SELECT Chunkkey,BlockIndex,block FROM bitmap WHERE bitmap_id=? AND db=? AND slice=? ", id, db, slice).Iter() count := int64(0) for iter.Scan(&chunk_key, &block_index, &block) { s8 = uint8(block_index) @@ -72,14 +73,14 @@ func (c *CassandraStorage)Fetch( bitmap_id uint64, shard int32) IBitmap { return bitmap } -func (c *CassandraStorage) Store( id int64, shard_key int32, bitmap *Bitmap) error { +func (c *CassandraStorage) Store( 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 { - c.StoreBlock(id, shard_key, int64(chunk.Key), block_index, iblock) + c.StoreBlock(id, db,slice, int64(chunk.Key), block_index, iblock) } } } @@ -88,13 +89,13 @@ func (c *CassandraStorage) Store( id int64, shard_key int32, bitmap *Bitmap) err var dumb = COUNTERMASK COUNTER_KEY := int64(dumb) - c.StoreBlock(id, shard_key, COUNTER_KEY, 0, cnt) + c.StoreBlock(id, db,slice, COUNTER_KEY, 0, cnt) return nil } -func (c *CassandraStorage)StoreBlock(id int64, shard_key int32, chunk int64, block_index int32, block int64) error { +func (c *CassandraStorage)StoreBlock(id int64, db string, slice int, chunk int64, block_index int32, block int64) error { - if err := c.db.Query(`INSERT INTO bitmap (bitmap_id, shard_id, ChunkKey, BlockIndex,block) VALUES (?,?, ?,?,?);`, id, shard_key, chunk, block_index, block).Exec(); err != nil { + 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 { log.Println(err) log.Println("INSERT ", id, chunk, block_index) return err diff --git a/index/storage_mem.go b/index/storage_mem.go index ba000496c..700aa6df5 100644 --- a/index/storage_mem.go +++ b/index/storage_mem.go @@ -18,8 +18,8 @@ func NewMemoryStorage() Storage{ return obj } -func (c *MemoryStorage)Fetch( bitmap_id uint64, shard_key int32) IBitmap { - key := fmt.Sprintf("%d:%d",bitmap_id,shard_key) +func (c *MemoryStorage)Fetch( bitmap_id uint64, db string, slice int) IBitmap { + key := fmt.Sprintf("%d:%s:%d",bitmap_id,db,slice) bitmap,found := c.db[key] if !found{ bitmap = CreateRBBitmap().(*Bitmap) @@ -28,14 +28,14 @@ func (c *MemoryStorage)Fetch( bitmap_id uint64, shard_key int32) IBitmap { return bitmap } -func (c *MemoryStorage) Store( bitmap_id int64, shard_key int32, bitmap *Bitmap) error { - key := fmt.Sprintf("%d:%d",bitmap_id,shard_key) +func (c *MemoryStorage) Store( bitmap_id int64, db string, slice int, bitmap *Bitmap) error { + key := fmt.Sprintf("%d:%s:%d",bitmap_id,db,slice) c.db[key]= bitmap return nil } -func (c *MemoryStorage)StoreBlock(bitmap_id int64, shard_key int32, chunk_key int64, block_index int32, block int64) error { - bm := c.Fetch(uint64(bitmap_id),shard_key) +func (c *MemoryStorage)StoreBlock(bitmap_id int64, db string, slice int, chunk_key int64, block_index int32, block int64) error { + bm := c.Fetch(uint64(bitmap_id),db,slice) node := GetChunk(bm,uint64(chunk_key)) if node == nil{ node = &Chunk{uint64(chunk_key), BlockArray{}}