diff --git a/index/bitmap.go b/index/bitmap.go index eb0d4db41..5b690243a 100644 --- a/index/bitmap.go +++ b/index/bitmap.go @@ -142,11 +142,17 @@ func (s *BlockArray) set_bit(BlockIndex uint8, bit uint8) bool { s.Block[BlockIndex] |= 1 << bit return val == 0 } +func (s *BlockArray) clear_bit(BlockIndex uint8, bit uint8) bool { + val := s.Block[BlockIndex] & (1 << bit) + s.Block[BlockIndex] &= ^(1 << bit) + return val != 0 +} type Chunk struct { Key uint64 Value BlockArray } + type Bitmap struct { nodes *rbtree.Tree bcount uint64 @@ -399,6 +405,7 @@ type IBitmap interface { Get(*Chunk) *Chunk Len() int Inc() + Dec() Count() uint64 SetCount(uint64) Bits() []uint64 @@ -514,6 +521,11 @@ func (b *Bitmap) Len() int { func (b *Bitmap) Inc() { b.bcount += 1 } +func (b *Bitmap) Dec() { + if b.bcount > 0 { + b.bcount -= 1 + } +} func (b *Bitmap) SetCount(c uint64) { b.bcount = c } @@ -554,6 +566,25 @@ func SetBit(b IBitmap, position uint64) (bool, *Chunk, Address) { } return data_changed, node, address } + +func ClearBit(b IBitmap, position uint64) (bool, *Chunk, Address) { + //Chunk,Chunk_index,bit_offset :=deref(position) + address := deref(position) + + item := GetChunk(b, address.ChunkKey) + var node *Chunk + if item == nil { + return false, nil, address + } else { + node = item + } + data_changed := node.Value.clear_bit(address.BlockIndex, address.Bit) + if data_changed { + b.Dec() + } + return data_changed, node, address +} + func BitCount(b IBitmap) uint64 { var total uint64 total = 0 diff --git a/index/brand.go b/index/brand.go index 2696b79ee..895ffe887 100644 --- a/index/brand.go +++ b/index/brand.go @@ -87,6 +87,15 @@ func (self *Brand) Get(bitmap_id uint64) IBitmap { return b } +func (self *Brand) Get_nocache(bitmap_id uint64) (IBitmap, uint64) { + bm, ok := self.bitmap_cache[bitmap_id] + if ok { + return bm.bitmap, bm.category + } + //I should fetch the category here..need to come up with a good source + return self.storage.Fetch(bitmap_id, self.db, self.frame, self.slice) +} + func (self *Brand) GetFilter(bitmap_id, filter uint64) IBitmap { b, old_filter := self.storage.Fetch(bitmap_id, self.db, self.frame, self.slice) if filter == 0 { @@ -417,3 +426,35 @@ func (self *Brand) Load(requestChan chan Command, f *Fragment) { } } + +func (self *Brand) ClearBit(bitmap_id uint64, bit_pos uint64) bool { + bm1, ok := self.bitmap_cache[bitmap_id] + var bm IBitmap + filter := uint64(0) + if ok { + bm = bm1.bitmap + filter = bm1.category + } else { + bm, filter = self.Get_nocache(bitmap_id) + if bm.Count() == 0 { + return false //nothing to unset + } + } + + change, chunk, address := ClearBit(bm, bit_pos) + if change { + val := chunk.Value.Block[address.BlockIndex] + /* self.storage.BeginBatch() + self.storage.StoreBlock(bitmap_id, self.db, self.frame, self.slice, filter, address.ChunkKey, int32(address.BlockIndex), val) + self.storage.StoreBlock(bitmap_id, self.db, self.frame, self.slice, filter, COUNTERMASK, 0, bm.Count()) + self.storage.EndBatch() + */ + if val == 0 { + self.storage.RemoveBlock(bitmap_id, self.db, self.frame, self.slice, filter, address.ChunkKey, int32(address.BlockIndex)) + } else { + self.storage.StoreBit(bitmap_id, self.db, self.frame, self.slice, filter, address.ChunkKey, int32(address.BlockIndex), val, bm.Count()) + } + self.rank_count++ + } + return change +} diff --git a/index/commands.go b/index/commands.go index f91cb6a24..24b34809d 100644 --- a/index/commands.go +++ b/index/commands.go @@ -348,3 +348,17 @@ func NewTopNAll(n int, categories []uint64) *CmdTopNAll { func (self *CmdTopNAll) Execute(f *Fragment) Calculation { return f.TopNAll(self.n, self.categories) } + +type CmdClearBit struct { + *Responder + bitmap_id uint64 + bit_pos uint64 +} + +func NewClearBit(bitmap_id uint64, bit_pos uint64) *CmdClearBit { + result := &CmdClearBit{NewResponder("ClearBit"), bitmap_id, bit_pos} + return result +} +func (self *CmdClearBit) Execute(f *Fragment) Calculation { + return f.impl.ClearBit(self.bitmap_id, self.bit_pos) +} diff --git a/index/fragment_container.go b/index/fragment_container.go index 0766703e3..48b7c9741 100644 --- a/index/fragment_container.go +++ b/index/fragment_container.go @@ -270,6 +270,18 @@ func (self *FragmentContainer) SetBit(frag_id util.SUUID, bitmap_id uint64, pos return false, errors.New("Invalid Bitmap Handle") } +func (self *FragmentContainer) ClearBit(frag_id util.SUUID, bitmap_id uint64, pos uint64) (bool, error) { + if fragment, found := self.GetFragment(frag_id); found { + request := NewClearBit(bitmap_id, pos) + fragment.requestChan <- request + result := request.Response() + util.SendTimer("fragmant_container_ClearBit", result.exec_time.Nanoseconds()) + util.SendInc("fragmant_container_ClearBit") + return result.answer.(bool), nil + } + return false, errors.New("Invalid Bitmap Handle") +} + func (self *FragmentContainer) Clear(frag_id util.SUUID) (bool, error) { if fragment, found := self.GetFragment(frag_id); found { request := NewClear() @@ -295,6 +307,7 @@ func (self *FragmentContainer) AddFragment(db string, frame string, slice int, i type Pilosa interface { Get(id uint64) IBitmap SetBit(id uint64, bit_pos uint64, filter uint64) bool + ClearBit(id uint64, bit_pos uint64) bool TopN(b IBitmap, n int, categories []uint64) []Pair TopNAll(n int, categories []uint64) []Pair Clear() bool diff --git a/index/general.go b/index/general.go index 3dbf7d727..25041e4b5 100644 --- a/index/general.go +++ b/index/general.go @@ -156,3 +156,29 @@ func (self *General) TopNAll(n int, categories []uint64) []Pair { } return results } + +func (self *General) ClearBit(bitmap_id uint64, bit_pos uint64) bool { + bm := self.Get_nocache(bitmap_id) + if bm.Count() == 0 { + return false + } + change, chunk, address := ClearBit(bm, bit_pos) + 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)) + } else { + self.storage.StoreBit(bitmap_id, self.db, self.frame, self.slice, uint64(0), address.ChunkKey, int32(address.BlockIndex), val, bm.Count()) + } + } + return change +} + +func (self *General) Get_nocache(bitmap_id uint64) IBitmap { + bm, ok := self.bitmap_cache.Get(bitmap_id) + if ok && bm != nil { + return bm.(*Bitmap) + } + bm, _ = self.storage.Fetch(bitmap_id, self.db, self.frame, self.slice) + return bm.(*Bitmap) +} diff --git a/index/storage.go b/index/storage.go index ad0f67ab2..8be339c9d 100644 --- a/index/storage.go +++ b/index/storage.go @@ -4,7 +4,8 @@ type Storage interface { Fetch(bitmap_id uint64, db string, frame string, slice int) (IBitmap, uint64) 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, bchunk uint64, block_index int32, bblock, count uint64) + 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 BeginBatch() EndBatch() FlushBatch() diff --git a/index/storage_cass.go b/index/storage_cass.go index de198c530..3c8b839ed 100644 --- a/index/storage_cass.go +++ b/index/storage_cass.go @@ -17,6 +17,7 @@ type CassandraStorage struct { db *gocql.Session batch *gocql.Batch stmt string + dstmt string batch_time time.Time batch_counter int cass_time_window_secs float64 @@ -55,6 +56,7 @@ 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.batch = nil obj.batch_time = time.Now() obj.batch_counter = 0 @@ -207,6 +209,10 @@ func (self *CassandraStorage) StoreBit(bid uint64, db string, frame string, slic 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 diff --git a/index/storage_leveldb.go b/index/storage_leveldb.go index 1e0077d3e..95980d29f 100644 --- a/index/storage_leveldb.go +++ b/index/storage_leveldb.go @@ -170,6 +170,9 @@ 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) Close() { self.FlushBatch() diff --git a/index/storage_mem.go b/index/storage_mem.go index d06d1575d..398c6a146 100644 --- a/index/storage_mem.go +++ b/index/storage_mem.go @@ -48,3 +48,6 @@ 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 +}