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

refactor delete bit

Browse source
This commit is contained in:
Todd Gruben 2015-03-04 23:17:36 +00:00
parent 2524ff2a51
commit 8b2398b8e5
8 changed files with 58 additions and 77 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

25
core/service.go
View file

@ -57,19 +57,11 @@ func NewService() *Service {
return service
}
func (self *Service) PrepareLogging() {
func (self *Service) getProduction() string {
base_path := config.GetString("log_path")
if base_path == "" {
base_path = "/tmp"
}
/*
f, err := os.OpenFile(fmt.Sprintf("%s/%s.%s", base_path, self.name, self.Id), os.O_RDWR|os.O_CREATE|os.O_APPEND, 0666)
if err != nil {
log.Warn("error opening file: %v", err)
}
//defer f.Close()
log.SetOutput(f)
*/
fname := fmt.Sprintf("%s/%s.%s", base_path, self.name, self.Id)
//<seelog minlevel="debug" maxlevel="error">
log_level := config.GetStringDefault("log_level", "info")
@ -80,7 +72,20 @@ func (self *Service) PrepareLogging() {
</outputs>
</seelog>
`, log_level, fname)
logger, _ := log.LoggerFromConfigAsBytes([]byte(prod_config))
return prod_config
}
func (self *Service) getDev() string {
return `<seelog>
<outputs>
<console />
</outputs>
</seelog>
`
}
func (self *Service) PrepareLogging() {
logger, _ := log.LoggerFromConfigAsBytes([]byte(self.getProduction()))
// logger, _ := log.LoggerFromConfigAsBytes([]byte(self.getDev()))
log.ReplaceLogger(logger)
}

3
index/brand.go
View file

@ -428,6 +428,7 @@ func (self *Brand) Load(requestChan chan Command, f *Fragment) {
}
func (self *Brand) ClearBit(bitmap_id uint64, bit_pos uint64) bool {
log.Trace("ClearBit", bitmap_id, bit_pos)
bm1, ok := self.bitmap_cache[bitmap_id]
var bm IBitmap
filter := uint64(0)
@ -450,7 +451,7 @@ func (self *Brand) ClearBit(bitmap_id uint64, bit_pos uint64) bool {
self.storage.EndBatch()
*/
if val == 0 {
self.storage.RemoveBlock(bitmap_id, self.db, self.frame, self.slice, filter, address.ChunkKey, int32(address.BlockIndex))
self.storage.RemoveBit(bitmap_id, self.db, self.frame, self.slice, filter, address.ChunkKey, int32(address.BlockIndex), bm.Count())
} else {
self.storage.StoreBit(bitmap_id, self.db, self.frame, self.slice, filter, address.ChunkKey, int32(address.BlockIndex), val, bm.Count())
}

2
index/general.go
View file

@ -165,7 +165,7 @@ func (self *General) ClearBit(bitmap_id uint64, bit_pos uint64) bool {
if change {
val := chunk.Value.Block[address.BlockIndex]
if val == 0 {
self.storage.RemoveBlock(bitmap_id, self.db, self.frame, self.slice, uint64(0), address.ChunkKey, int32(address.BlockIndex))
self.storage.RemoveBit(bitmap_id, self.db, self.frame, self.slice, uint64(0), address.ChunkKey, int32(address.BlockIndex), bm.Count())
} else {
self.storage.StoreBit(bitmap_id, self.db, self.frame, self.slice, uint64(0), address.ChunkKey, int32(address.BlockIndex), val, bm.Count())
}

3
index/storage.go
View file

@ -5,7 +5,8 @@ type Storage interface {
Store(id uint64, db string, frame string, slice int, filter uint64, bitmap *Bitmap) error
StoreBlock(id uint64, db string, frame string, slice int, filter uint64, chunk uint64, block_index int32, block uint64) error
StoreBit(bid uint64, db string, frame string, slice int, filter uint64, chunk uint64, block_index int32, block, count uint64)
RemoveBlock(id uint64, db string, frame string, slice int, filter uint64, chunk uint64, block_index int32) error
RemoveBit(id uint64, db string, frame string, slice int, filter uint64, chunk uint64, block_index int32, count uint64)
RemoveBlock(id uint64, db string, frame string, slice int, chunk uint64, block_index int32)
BeginBatch()
EndBatch()
FlushBatch()

81
index/storage_cass.go
View file

@ -22,7 +22,6 @@ type CassandraStorage struct {
batch_counter int
cass_time_window_secs float64
cass_flush_size int
cass_queue CassQueue
}
var cluster *gocql.ClusterConfig
@ -46,6 +45,7 @@ func BuildSchema() {
*/
}
func NewCassStorage() Storage {
obj := new(CassandraStorage)
// cluster.CQLVersion = "3.0.0"
@ -56,14 +56,12 @@ func NewCassStorage() Storage {
obj.db = session
obj.stmt = `INSERT INTO bitmap ( bitmap_id, db, frame, slice , filter, ChunkKey, BlockIndex, block) VALUES (?,?,?,?,?,?,?,?) USING timestamp ?;`
obj.dstmt = `DELETE BlockIndex,block FROM bitmap WHERE bitmap_id=? AND db=? AND frame=? AND slice=? AND filter=? AND ChunkKey=? AND BlockIndex=?;`
obj.dstmt = `DELETE FROM bitmap USING TIMESTAMP ? WHERE bitmap_id=? AND db=? AND frame=? AND slice=? AND chunkkey=? AND blockindex=?`
obj.batch = nil
obj.batch_time = time.Now()
obj.batch_counter = 0
obj.cass_time_window_secs = float64(config.GetIntDefault("cassandra_time_window_secs", 5))
obj.cass_flush_size = config.GetIntDefault("cassandra_max_size_batch", 15)
obj.cass_queue = NewCassQueue()
go obj.asyncStore()
return obj
}
@ -116,7 +114,10 @@ func (self *CassandraStorage) BeginBatch() {
}
func (self *CassandraStorage) runBatch(batch *gocql.Batch) {
if batch != nil {
go self.db.ExecuteBatch(batch)
err := self.db.ExecuteBatch(batch)
if err != nil {
log.Warn("Batch ERROR", err)
}
}
}
func (self *CassandraStorage) FlushBatch() {
@ -181,56 +182,32 @@ func (self *CassandraStorage) StoreBlock(bid uint64, db string, frame string, sl
return nil
}
type CassRecord struct {
bitmap_id uint64
db string
frame string
slice int
filter uint64
chunk uint64
block_index int32
val uint64
count uint64
func (self *CassandraStorage) StoreBit(bid uint64, db string, frame string, slice int, filter uint64, chunk uint64, block_index int32, val, count uint64) {
self.BeginBatch()
self.StoreBlock(bid, db, frame, slice, filter, chunk, block_index, val)
self.StoreBlock(bid, db, frame, slice, filter, COUNTERMASK, 0, count)
self.EndBatch()
}
func (self *CassandraStorage) asyncStore() {
for {
rec, term := self.cass_queue.Pop()
if term {
break
}
func (self *CassandraStorage) RemoveBit(id uint64, db string, frame string, slice int, filter uint64, chunk uint64, block_index int32, count uint64) {
log.Trace("RemoveBit", id, db, frame, slice, chunk, block_index)
self.BeginBatch()
self.RemoveBlock(id, db, frame, slice, chunk, block_index)
self.StoreBlock(id, db, frame, slice, filter, COUNTERMASK, 0, count)
self.EndBatch()
}
func (self *CassandraStorage) RemoveBlock(bid uint64, db string, frame string, slice int, bchunk uint64, block_index int32) {
log.Trace("RemoveBBlock", bid, db, frame, slice, bchunk, block_index)
id := util.Uint64ToInt64(bid) //these fucntions ignore overflow
chunk := util.Uint64ToInt64(bchunk)
if self.batch == nil {
self.BeginBatch()
self.StoreBlock(rec.bitmap_id, rec.db, rec.frame, rec.slice, rec.filter, rec.chunk, rec.block_index, rec.val)
self.StoreBlock(rec.bitmap_id, rec.db, rec.frame, rec.slice, rec.filter, COUNTERMASK, 0, rec.count)
self.EndBatch()
}
}
start := time.Now()
func (self *CassandraStorage) StoreBit(bid uint64, db string, frame string, slice int, filter uint64, bchunk uint64, block_index int32, bblock, count uint64) {
rec := CassRecord{bid, db, frame, slice, filter, bchunk, block_index, bblock, count}
self.cass_queue.Push(rec)
}
func (self *CassandraStorage) RemoveBlock(id uint64, db string, frame string, slice int, filter uint64, chunk uint64, block_index int32) error {
return nil
}
type CassQueue struct {
size int64
buffer chan CassRecord
}
func NewCassQueue() CassQueue {
//return CassQueue{0, make(chan CassRecord, 4096)}
return CassQueue{0, make(chan CassRecord)}
}
func (self *CassQueue) Push(rec CassRecord) {
// atomic.AddInt64(&self.size, 1)
self.buffer <- rec
}
func (self *CassQueue) Pop() (CassRecord, bool) {
ret := <-self.buffer
// atomic.AddInt64(&self.size, -1)
return ret, false
self.batch.Query(self.dstmt, start.UnixNano(), id, db, frame, slice, chunk, block_index)
delta := time.Since(start)
util.SendTimer("cassandra_storage_DeleteBlock", delta.Nanoseconds())
}

7
index/storage_leveldb.go
View file

@ -170,8 +170,11 @@ func (self *LevelDBStorage) StoreBlock(id uint64, db string, frame string, slice
util.SendTimer("leveldb_storage_StoreBlock", delta.Nanoseconds())
return nil
}
func (self *LevelDBStorage) RemoveBlock(id uint64, db string, frame string, slice int, filter uint64, chunk uint64, block_index int32) error {
return nil
func (self *LevelDBStorage) RemoveBlock(id uint64, db string, frame string, slice int, chunk uint64, block_index int32) {
}
func (self *LevelDBStorage) RemoveBit(bid uint64, db string, frame string, slice int, filter uint64, bchunk uint64, block_index int32, count uint64) {
}
func (self *LevelDBStorage) Close() {

6
index/storage_mem.go
View file

@ -48,6 +48,8 @@ func (c *MemoryStorage) StoreBlock(bitmap_id uint64, db string, frame string, sl
}
func (self *MemoryStorage) StoreBit(bid uint64, db string, frame string, slice int, filter uint64, bchunk uint64, block_index int32, bblock, count uint64) {
}
func (self *MemoryStorage) RemoveBlock(id uint64, db string, frame string, slice int, filter uint64, chunk uint64, block_index int32) error {
return nil
func (self *MemoryStorage) RemoveBit(id uint64, db string, frame string, slice int, filter uint64, chunk uint64, block_index int32, count uint64) {
}
func (self *MemoryStorage) RemoveBlock(id uint64, db string, frame string, slice int, chunk uint64, block_index int32) {
}

8
query/planner.go
View file

@ -464,26 +464,18 @@ func validateRange(Args map[string]interface{}) error {
func (self *QueryPlanner) buildTree(query *Query, slice int) (QueryTree, error) {
log.Trace("QueryPlanner.buildTree", query, slice)
var tree QueryTree
spew.Dump(query)
// handle SET operation regardless of the slice
if query.Operation == "set" {
println("SET")
log.Warn("SET")
tree = &SetQueryTree{&db.Bitmap{query.Args["id"].(uint64), query.Args["frame"].(string), query.Args["filter"].(uint64)}, query.Args["profile_id"].(uint64)}
return tree, nil
}
if query.Operation == "clear" {
println("CLEAR")
log.Warn("CLEAR")
tree = &ClearQueryTree{&db.Bitmap{query.Args["id"].(uint64), query.Args["frame"].(string), query.Args["filter"].(uint64)}, query.Args["profile_id"].(uint64)}
return tree, nil
}
if query.Operation == "recall" {
tree = &RecallQueryTree{query.Args["stash"].(Stash)}
return tree, nil
}