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

added batching to cassandra storage

Browse source
This commit is contained in:
Todd Gruben 2014-04-10 01:38:36 -05:00
parent 29f006f506
commit 51d5d97ef0
5 changed files with 41 additions and 13 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
index/brand.go
View file

@ -104,8 +104,10 @@ func (self *Brand) SetBit(bitmap_id uint64, bit_pos uint64, filter uint64) bool
change, chunk, address := SetBit(bm, bit_pos)
if change {
val := chunk.Value.Block[address.BlockIndex]
self.storage.BeginBatch()
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()))
self.storage.EndBatch()
if bm.Count() >= self.threshold_value {
self.Rank() //need to optimize this

2
index/general.go
View file

@ -52,8 +52,10 @@ func (self *General) SetBit(bitmap_id uint64, bit_pos uint64, filter uint64) boo
change, chunk, address := SetBit(bm, bit_pos)
if change {
val := chunk.Value.Block[address.BlockIndex]
self.storage.BeginBatch()
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()))
self.storage.EndBatch()
}
return change

2
index/storage.go
View file

@ -4,4 +4,6 @@ type Storage interface {
Fetch(bitmap_id uint64, db string, frame string, slice int) (IBitmap, uint64)
Store(id int64, db string, frame string, slice int, filter uint64, bitmap *Bitmap) error
StoreBlock(id int64, db string, frame string, slice int, filter uint64, chunk int64, block_index int32, block int64) error
BeginBatch()
EndBatch()
}

44
index/storage_cass.go
View file

@ -13,7 +13,9 @@ import (
)
type CassandraStorage struct {
db *gocql.Session
db *gocql.Session
batch *gocql.Batch
stmt string
}
func BuildSchema() {
@ -43,6 +45,8 @@ func NewCassStorage(host, keyspace string) Storage {
if err != nil {
}
obj.db = session
obj.stmt = `INSERT INTO bitmap ( bitmap_id, db, frame, slice , filter, ChunkKey, BlockIndex, block) VALUES (?,?,?,?,?,?,?,?);`
obj.batch = nil
return obj
}
@ -84,15 +88,31 @@ func (c *CassandraStorage) Fetch(bitmap_id uint64, db string, frame string, slic
bitmap.SetCount(uint64(count))
return bitmap, uint64(filter)
}
func (self *CassandraStorage) BeginBatch() {
self.batch = gocql.NewBatch(gocql.LoggedBatch)
}
func (self *CassandraStorage) EndBatch() {
start := time.Now()
if self.batch != nil {
self.db.ExecuteBatch(self.batch)
self.batch = nil
} else {
log.Println("NIL BATCH")
}
delta := time.Since(start)
util.SendTimer("cassandra_storage_EndBatch", delta.Nanoseconds())
func (c *CassandraStorage) Store(id int64, db string, frame string, slice int, filter uint64, bitmap *Bitmap) error {
}
func (self *CassandraStorage) Store(id int64, db string, frame string, slice int, filter uint64, bitmap *Bitmap) error {
self.BeginBatch()
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, frame, slice, filter, int64(chunk.Key), block_index, iblock)
self.StoreBlock(id, db, frame, slice, filter, int64(chunk.Key), block_index, iblock)
}
}
}
@ -101,20 +121,18 @@ func (c *CassandraStorage) Store(id int64, db string, frame string, slice int, f
var dumb = COUNTERMASK
COUNTER_KEY := int64(dumb)
c.StoreBlock(id, db, frame, slice, filter, COUNTER_KEY, 0, cnt)
self.StoreBlock(id, db, frame, slice, filter, COUNTER_KEY, 0, cnt)
self.EndBatch()
return nil
}
func (c *CassandraStorage) StoreBlock(id int64, db string, frame string, slice int, filter uint64, chunk int64, block_index int32, block int64) error {
start := time.Now()
var err error
if err = c.db.Query(`
INSERT INTO bitmap ( bitmap_id, db, frame, slice , filter, ChunkKey, BlockIndex, block) VALUES (?,?,?,?,?,?,?,?);`, id, db, frame, slice, int(filter), chunk, block_index, block).Consistency(gocql.One).Exec(); err != nil {
log.Println(err)
log.Println("INSERT ", id, chunk, block_index)
func (self *CassandraStorage) StoreBlock(id int64, db string, frame string, slice int, filter uint64, chunk int64, block_index int32, block int64) error {
if self.batch == nil {
panic("NIL BATCH")
}
start := time.Now()
self.batch.Query(self.stmt, id, db, frame, slice, int(filter), chunk, block_index, block)
delta := time.Since(start)
util.SendTimer("cassandra_storage_StoreBlock", delta.Nanoseconds())
return err
return nil
}

4
index/storage_mem.go
View file

@ -16,6 +16,10 @@ func NewMemoryStorage() Storage {
return obj
}
func (c *MemoryStorage) BeginBatch() {
}
func (c *MemoryStorage) EndBatch() {
}
func (c *MemoryStorage) Fetch(bitmap_id uint64, db string, frame string, slice int) (IBitmap, uint64) {
// log.Println("hello")