mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-09-05 08:10:50 +00:00
refactor index package
This commit does clean up and minor refactoring of the index package.
This commit is contained in:
Ben Johnson
parent
afbdebaac8
commit
c38d023cc1
14 changed files with 558 additions and 664 deletions
22
core/load.go
22
core/load.go
|
|
@ -13,21 +13,29 @@ import (
|
|||
"github.com/umbel/pilosa/index"
|
||||
)
|
||||
|
||||
func copy_raw(src [32]uint64) index.BlockArray {
|
||||
var o = make([]uint64, 32, 32)
|
||||
func copy_raw(src [32]uint64) index.Blocks {
|
||||
o := make(index.Blocks, 32)
|
||||
for k, v := range src {
|
||||
o[k] = v
|
||||
}
|
||||
return index.BlockArray{Block: o}
|
||||
return o
|
||||
}
|
||||
func sendBitmap(batcher *Batcher, bitmap index.IBitmap, db string, frame string, bitmap_id, filter uint64, slice int, finish chan error) {
|
||||
func sendBitmap(batcher *Batcher, bitmap *index.Bitmap, db string, frame string, bitmap_id, filter uint64, slice int, finish chan error) {
|
||||
if slice < 0 {
|
||||
log.Warn("Bad split", db, frame, slice, bitmap_id)
|
||||
finish <- errors.New("BadSplit")
|
||||
return
|
||||
}
|
||||
compressed_bitmap := bitmap.ToCompressString()
|
||||
results := batcher.Batch(db, frame, compressed_bitmap, bitmap_id, slice, filter)
|
||||
|
||||
// Compress bitmap.
|
||||
var b bytes.Buffer
|
||||
w := gzip.NewWriter(&b)
|
||||
w.Write(bitmap.ToBytes())
|
||||
w.Flush()
|
||||
w.Close()
|
||||
buf := base64.StdEncoding.EncodeToString(b.Bytes())
|
||||
|
||||
results := batcher.Batch(db, frame, buf, bitmap_id, slice, filter)
|
||||
finish <- results
|
||||
}
|
||||
|
||||
|
|
@ -50,7 +58,7 @@ func FromApiString(batcher *Batcher, db string, frame string, api_string string,
|
|||
}
|
||||
first := true
|
||||
bitmap := index.NewBitmap()
|
||||
last_slice := index.COUNTERMASK
|
||||
last_slice := index.CounterMask
|
||||
sent_count := 0
|
||||
finish := make(chan error)
|
||||
|
||||
|
|
|
|||
|
|
@ -18,31 +18,31 @@ func benchmarkDifferentCombinations(b *testing.B, op string, b1, b2 int, s1, s2
|
|||
rand.Seed(int64(c1))
|
||||
for i := 0; i < b1; i++ {
|
||||
bit += uint64(rand.Intn(s1) + 1)
|
||||
SetBit(m1, bit)
|
||||
m1.SetBit(bit)
|
||||
}
|
||||
|
||||
bit = 0
|
||||
rand.Seed(int64(c2))
|
||||
for i := 0; i < b2; i++ {
|
||||
bit += uint64(rand.Intn(s1) + 1)
|
||||
SetBit(m2, bit)
|
||||
m2.SetBit(bit)
|
||||
}
|
||||
|
||||
var f func(a_bm IBitmap, b_bm IBitmap) IBitmap
|
||||
var f func(b_bm *Bitmap) *Bitmap
|
||||
switch op {
|
||||
case "and":
|
||||
f = Intersection
|
||||
f = m1.Intersection
|
||||
case "or":
|
||||
f = Union
|
||||
f = m1.Union
|
||||
case "diff":
|
||||
f = Difference
|
||||
f = m1.Difference
|
||||
default:
|
||||
return
|
||||
}
|
||||
|
||||
b.ResetTimer()
|
||||
for i := 0; i < b.N; i++ {
|
||||
if f(m1, m2) == nil {
|
||||
if f(m2) == nil {
|
||||
b.Fatalf("Problem with %s benchmark at i = %d", op, i)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
703
index/bitmap.go
703
index/bitmap.go
|
|
@ -8,439 +8,225 @@ import (
|
|||
"encoding/base64"
|
||||
"encoding/binary"
|
||||
"encoding/gob"
|
||||
"io/ioutil"
|
||||
|
||||
log "github.com/cihub/seelog"
|
||||
"github.com/yasushi-saito/rbtree"
|
||||
)
|
||||
|
||||
const (
|
||||
MAX_HOT_SIZE = 50000
|
||||
BLOCK_SIZE = 5000
|
||||
START_IDX = MAX_HOT_SIZE - BLOCK_SIZE
|
||||
COUNTERMASK = uint64(0xffffffffffffffff)
|
||||
)
|
||||
const CounterMask = uint64(0xffffffffffffffff)
|
||||
|
||||
var (
|
||||
COUNTER_KEY int64
|
||||
)
|
||||
|
||||
func init() {
|
||||
var dumb = COUNTERMASK
|
||||
COUNTER_KEY = int64(dumb)
|
||||
|
||||
}
|
||||
|
||||
//
|
||||
type IntSet struct {
|
||||
set map[uint64]bool
|
||||
}
|
||||
|
||||
func NewIntSet() *IntSet {
|
||||
x := new(IntSet)
|
||||
x.set = make(map[uint64]bool)
|
||||
return x
|
||||
}
|
||||
|
||||
func (self *IntSet) Add(i uint64) bool {
|
||||
_, found := self.set[i]
|
||||
self.set[i] = true
|
||||
return !found //False if it existed already
|
||||
}
|
||||
|
||||
func (self *IntSet) Contains(i uint64) bool {
|
||||
_, found := self.set[i]
|
||||
return found //true if it existed already
|
||||
}
|
||||
|
||||
func (self *IntSet) Remove(i uint64) {
|
||||
delete(self.set, i)
|
||||
}
|
||||
|
||||
func (self *IntSet) Size() int {
|
||||
return len(self.set)
|
||||
}
|
||||
|
||||
type BlockArray struct {
|
||||
Block []uint64
|
||||
}
|
||||
|
||||
func (s *BlockArray) bitcount() uint64 {
|
||||
return popcntSlice(s.Block)
|
||||
}
|
||||
func BlockArray_union(a *BlockArray, b *BlockArray) BlockArray {
|
||||
var o = BlockArray{make([]uint64, 32, 32)}
|
||||
for i, _ := range a.Block {
|
||||
o.Block[i] = a.Block[i] | b.Block[i]
|
||||
}
|
||||
return o
|
||||
}
|
||||
|
||||
func BlockArray_invert(a *BlockArray) BlockArray {
|
||||
var o = BlockArray{make([]uint64, 32, 32)}
|
||||
for i, _ := range a.Block {
|
||||
o.Block[i] = ^a.Block[i]
|
||||
}
|
||||
return o
|
||||
}
|
||||
|
||||
func BlockArray_copy(a *BlockArray) BlockArray {
|
||||
var o = BlockArray{make([]uint64, 32, 32)}
|
||||
for i, _ := range a.Block {
|
||||
o.Block[i] = a.Block[i]
|
||||
}
|
||||
return o
|
||||
}
|
||||
|
||||
func BlockArray_andcount(a *BlockArray, b *BlockArray) uint64 {
|
||||
return popcntAndSliceAsm(a.Block, b.Block)
|
||||
}
|
||||
|
||||
func BlockArray_intersection(a *BlockArray, b *BlockArray) BlockArray {
|
||||
var o = BlockArray{make([]uint64, 32, 32)}
|
||||
for i, _ := range a.Block {
|
||||
o.Block[i] = a.Block[i] & b.Block[i]
|
||||
}
|
||||
return o
|
||||
}
|
||||
|
||||
func BlockArray_difference(a *BlockArray, b *BlockArray) BlockArray {
|
||||
var o = BlockArray{make([]uint64, 32, 32)}
|
||||
for i, _ := range a.Block {
|
||||
o.Block[i] = a.Block[i] &^ b.Block[i]
|
||||
}
|
||||
return o
|
||||
}
|
||||
|
||||
func (s *BlockArray) set_bit(BlockIndex uint8, bit uint8) bool {
|
||||
val := s.Block[BlockIndex] & (1 << bit)
|
||||
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
|
||||
}
|
||||
var CounterKey = int64(-1)
|
||||
|
||||
// Bitmap represents a bitmap broken up into Chunks.
|
||||
// Internally it is represented as a red-black tree of chunks.
|
||||
type Bitmap struct {
|
||||
nodes *rbtree.Tree
|
||||
tree *rbtree.Tree
|
||||
bcount uint64
|
||||
}
|
||||
|
||||
func Compare(a uint64, b uint64) int {
|
||||
if a < b {
|
||||
return -1
|
||||
} else if a > b {
|
||||
return 1
|
||||
// NewBitmap returns a new instance of Bitmap.
|
||||
func NewBitmap() *Bitmap {
|
||||
return &Bitmap{
|
||||
tree: rbtree.NewTree(rbtreeItemCompare),
|
||||
}
|
||||
return 0
|
||||
}
|
||||
func Clone(a_bm IBitmap) IBitmap {
|
||||
|
||||
var a = a_bm.Min()
|
||||
output := CreateRBBitmap()
|
||||
// Chunk returns the chunk within the bitmap.
|
||||
// Returns nil if the chunk key does not exist.
|
||||
func (b *Bitmap) Chunk(c *Chunk) *Chunk {
|
||||
if n := b.tree.Get(c); n != nil {
|
||||
return n.(*Chunk)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// AddChunk adds c to the bitmap.
|
||||
func (b *Bitmap) AddChunk(c *Chunk) { b.tree.Insert(c) }
|
||||
|
||||
// ChunkIterator returns an iterator for looping over the bitmap's chunks.
|
||||
func (b *Bitmap) ChunkIterator() *ChunkIterator {
|
||||
return &ChunkIterator{b.tree.Min()}
|
||||
}
|
||||
|
||||
// Clone returns a copy of b.
|
||||
func (b *Bitmap) Clone() *Bitmap {
|
||||
itr := b.ChunkIterator()
|
||||
other := NewBitmap()
|
||||
|
||||
for {
|
||||
if a.Limit() {
|
||||
if itr.Limit() {
|
||||
break
|
||||
}
|
||||
var a_node = a.Item()
|
||||
var o = BlockArray_copy(&a_node.Value)
|
||||
var o_node = &Chunk{a_node.Key, o}
|
||||
output.AddChunk(o_node)
|
||||
a = a.Next()
|
||||
|
||||
node := itr.Item()
|
||||
other.AddChunk(&Chunk{
|
||||
Key: node.Key,
|
||||
Value: node.Value.copy(),
|
||||
})
|
||||
|
||||
itr = itr.Next()
|
||||
}
|
||||
return output
|
||||
return other
|
||||
}
|
||||
|
||||
func IntersectionCount(a_bm IBitmap, b_bm IBitmap) uint64 {
|
||||
var a = a_bm.Min()
|
||||
var b = b_bm.Min()
|
||||
defer a.Close()
|
||||
defer b.Close()
|
||||
results := uint64(0)
|
||||
// IntersectionCount returns the number of itersections between b and other.
|
||||
func (b *Bitmap) IntersectionCount(other *Bitmap) uint64 {
|
||||
itr0 := b.ChunkIterator()
|
||||
itr1 := other.ChunkIterator()
|
||||
|
||||
results := uint64(0)
|
||||
for {
|
||||
if b.Limit() || a.Limit() {
|
||||
if itr1.Limit() || itr0.Limit() {
|
||||
break
|
||||
} else if a.Item().Key < b.Item().Key {
|
||||
a = a.Next()
|
||||
} else if a.Item().Key > b.Item().Key {
|
||||
b = b.Next()
|
||||
} else if a.Item().Key == b.Item().Key {
|
||||
var a_node = a.Item()
|
||||
var b_node = b.Item().Value
|
||||
results += BlockArray_andcount(&a_node.Value, &b_node)
|
||||
a = a.Next()
|
||||
b = b.Next()
|
||||
} else if itr0.Item().Key < itr1.Item().Key {
|
||||
itr0 = itr0.Next()
|
||||
} else if itr0.Item().Key > itr1.Item().Key {
|
||||
itr1 = itr1.Next()
|
||||
} else if itr0.Item().Key == itr1.Item().Key {
|
||||
results += itr0.Item().Value.andcount(itr1.Item().Value)
|
||||
itr0 = itr0.Next()
|
||||
itr1 = itr1.Next()
|
||||
}
|
||||
}
|
||||
return results
|
||||
}
|
||||
|
||||
func Intersection(a_bm IBitmap, b_bm IBitmap) IBitmap {
|
||||
var a = a_bm.Min()
|
||||
var b = b_bm.Min()
|
||||
defer a.Close()
|
||||
defer b.Close()
|
||||
output := CreateRBBitmap()
|
||||
// Intersection returns the itersection of b and other.
|
||||
func (b *Bitmap) Intersection(other *Bitmap) *Bitmap {
|
||||
itr0 := b.ChunkIterator()
|
||||
itr1 := other.ChunkIterator()
|
||||
|
||||
output := NewBitmap()
|
||||
for {
|
||||
if b.Limit() || a.Limit() {
|
||||
if itr1.Limit() || itr0.Limit() {
|
||||
break
|
||||
} else if a.Item().Key < b.Item().Key {
|
||||
a = a.Next()
|
||||
} else if a.Item().Key > b.Item().Key {
|
||||
b = b.Next()
|
||||
} else if a.Item().Key == b.Item().Key {
|
||||
var a_node = a.Item()
|
||||
var b_node = b.Item().Value
|
||||
var o = BlockArray_intersection(&a_node.Value, &b_node)
|
||||
var o_node = &Chunk{a_node.Key, o}
|
||||
output.AddChunk(o_node)
|
||||
a = a.Next()
|
||||
b = b.Next()
|
||||
} else if itr0.Item().Key < itr1.Item().Key {
|
||||
itr0 = itr0.Next()
|
||||
} else if itr0.Item().Key > itr1.Item().Key {
|
||||
itr1 = itr1.Next()
|
||||
} else if itr0.Item().Key == itr1.Item().Key {
|
||||
output.AddChunk(&Chunk{
|
||||
Key: itr0.Item().Key,
|
||||
Value: itr0.Item().Value.intersection(itr1.Item().Value),
|
||||
})
|
||||
itr0 = itr0.Next()
|
||||
itr1 = itr1.Next()
|
||||
}
|
||||
}
|
||||
return output
|
||||
}
|
||||
func Invert(a_bm IBitmap) IBitmap {
|
||||
output := CreateRBBitmap()
|
||||
for i := a_bm.Min(); !i.Limit(); i = i.Next() {
|
||||
var node = i.Item()
|
||||
var o = BlockArray_invert(&node.Value)
|
||||
var o_node = &Chunk{node.Key, o}
|
||||
output.AddChunk(o_node)
|
||||
|
||||
// Invert returns a bitwise inversion of b.
|
||||
func (b *Bitmap) Invert() *Bitmap {
|
||||
other := NewBitmap()
|
||||
for i := b.ChunkIterator(); !i.Limit(); i = i.Next() {
|
||||
other.AddChunk(&Chunk{
|
||||
Key: i.Item().Key,
|
||||
Value: i.Item().Value.invert(),
|
||||
})
|
||||
}
|
||||
return output
|
||||
return other
|
||||
|
||||
}
|
||||
func NewBitmap() IBitmap {
|
||||
return CreateRBBitmap()
|
||||
}
|
||||
|
||||
func Union(a_bm IBitmap, b_bm IBitmap) IBitmap {
|
||||
var a = a_bm.Min()
|
||||
var b = b_bm.Min()
|
||||
defer a.Close()
|
||||
defer b.Close()
|
||||
output := CreateRBBitmap()
|
||||
var o_last_Key = uint64(0xdeadbeef)
|
||||
// Union returns the bitwise union of b and other.
|
||||
func (b *Bitmap) Union(other *Bitmap) *Bitmap {
|
||||
itr0 := b.ChunkIterator()
|
||||
itr1 := other.ChunkIterator()
|
||||
|
||||
output := NewBitmap()
|
||||
eof := uint64(0xdeadbeef)
|
||||
|
||||
for {
|
||||
if a.Limit() && b.Limit() {
|
||||
if itr0.Limit() && itr1.Limit() {
|
||||
break
|
||||
} else if a.Limit() {
|
||||
if o_last_Key == b.Item().Key {
|
||||
} else if itr0.Limit() {
|
||||
if eof == itr1.Item().Key {
|
||||
break
|
||||
}
|
||||
var b_node = b.Item()
|
||||
var o_node = &Chunk{b_node.Key, b_node.Value}
|
||||
output.AddChunk(o_node)
|
||||
o_last_Key = o_node.Key
|
||||
b = b.Next()
|
||||
} else if b.Limit() {
|
||||
if o_last_Key == a.Item().Key {
|
||||
output.AddChunk(&Chunk{itr1.Item().Key, itr1.Item().Value})
|
||||
eof = itr1.Item().Key
|
||||
itr1 = itr1.Next()
|
||||
} else if itr1.Limit() {
|
||||
if eof == itr0.Item().Key {
|
||||
break
|
||||
}
|
||||
var a_node = a.Item()
|
||||
var o_node = &Chunk{a_node.Key, a_node.Value}
|
||||
output.AddChunk(o_node)
|
||||
o_last_Key = o_node.Key
|
||||
a = a.Next()
|
||||
} else if a.Item().Key < b.Item().Key {
|
||||
var a_node = a.Item()
|
||||
var o_node = &Chunk{a_node.Key, a_node.Value}
|
||||
output.AddChunk(o_node)
|
||||
o_last_Key = o_node.Key
|
||||
a = a.Next()
|
||||
} else if a.Item().Key > b.Item().Key {
|
||||
var b_node = b.Item()
|
||||
var o_node = &Chunk{b_node.Key, b_node.Value}
|
||||
output.AddChunk(o_node)
|
||||
o_last_Key = o_node.Key
|
||||
b = b.Next()
|
||||
} else if a.Item().Key == b.Item().Key {
|
||||
var a_node = a.Item()
|
||||
var b_node = b.Item().Value
|
||||
var o = BlockArray_union(&a_node.Value, &b_node)
|
||||
var o_node = &Chunk{a_node.Key, o}
|
||||
output.AddChunk(o_node)
|
||||
o_last_Key = o_node.Key
|
||||
a = a.Next()
|
||||
b = b.Next()
|
||||
output.AddChunk(&Chunk{itr0.Item().Key, itr0.Item().Value})
|
||||
eof = itr0.Item().Key
|
||||
itr0 = itr0.Next()
|
||||
} else if itr0.Item().Key < itr1.Item().Key {
|
||||
output.AddChunk(&Chunk{itr0.Item().Key, itr0.Item().Value})
|
||||
eof = itr0.Item().Key
|
||||
itr0 = itr0.Next()
|
||||
} else if itr0.Item().Key > itr1.Item().Key {
|
||||
output.AddChunk(&Chunk{itr1.Item().Key, itr1.Item().Value})
|
||||
eof = itr1.Item().Key
|
||||
itr1 = itr1.Next()
|
||||
} else if itr0.Item().Key == itr1.Item().Key {
|
||||
output.AddChunk(&Chunk{
|
||||
Key: itr0.Item().Key,
|
||||
Value: itr0.Item().Value.union(itr1.Item().Value),
|
||||
})
|
||||
eof = itr0.Item().Key
|
||||
itr0 = itr0.Next()
|
||||
itr1 = itr1.Next()
|
||||
} else {
|
||||
log.Warn("NEVER SHOULD BE HERE")
|
||||
break
|
||||
panic("unreachable")
|
||||
}
|
||||
}
|
||||
return output
|
||||
}
|
||||
|
||||
func Difference(a_bm IBitmap, b_bm IBitmap) IBitmap {
|
||||
var a = a_bm.Min()
|
||||
var b = b_bm.Min()
|
||||
defer a.Close()
|
||||
defer b.Close()
|
||||
output := CreateRBBitmap()
|
||||
var o_last_Key = uint64(0)
|
||||
|
||||
if o_last_Key != 0 {
|
||||
o_last_Key = uint64(0)
|
||||
}
|
||||
// Difference returns the diff of b and other.
|
||||
func (b *Bitmap) Difference(other *Bitmap) *Bitmap {
|
||||
itr0 := b.ChunkIterator()
|
||||
itr1 := other.ChunkIterator()
|
||||
|
||||
output := NewBitmap()
|
||||
for {
|
||||
if a.Limit() && b.Limit() {
|
||||
if itr0.Limit() && itr1.Limit() {
|
||||
break
|
||||
} else if a.Limit() {
|
||||
} else if itr0.Limit() {
|
||||
break
|
||||
} else if b.Limit() {
|
||||
var a_node = a.Item()
|
||||
var o_node = &Chunk{a_node.Key, a_node.Value}
|
||||
output.AddChunk(o_node)
|
||||
o_last_Key = o_node.Key
|
||||
a = a.Next()
|
||||
} else if a.Item().Key < b.Item().Key {
|
||||
var a_node = a.Item()
|
||||
var o_node = &Chunk{a_node.Key, a_node.Value}
|
||||
output.AddChunk(o_node)
|
||||
o_last_Key = o_node.Key
|
||||
a = a.Next()
|
||||
} else if a.Item().Key > b.Item().Key {
|
||||
var b_node = b.Item()
|
||||
o_last_Key = b_node.Key
|
||||
b = b.Next()
|
||||
} else if a.Item().Key == b.Item().Key {
|
||||
var a_node = a.Item()
|
||||
|
||||
var b_node = b.Item().Value
|
||||
var o = BlockArray_difference(&a_node.Value, &b_node)
|
||||
|
||||
var o_node = &Chunk{a_node.Key, o}
|
||||
|
||||
//could not add if all zero
|
||||
if o_node.Value.bitcount() > 0 {
|
||||
output.AddChunk(o_node)
|
||||
} else if itr1.Limit() {
|
||||
output.AddChunk(&Chunk{itr0.Item().Key, itr0.Item().Value})
|
||||
itr0 = itr0.Next()
|
||||
} else if itr0.Item().Key < itr1.Item().Key {
|
||||
output.AddChunk(&Chunk{itr0.Item().Key, itr0.Item().Value})
|
||||
itr0 = itr0.Next()
|
||||
} else if itr0.Item().Key > itr1.Item().Key {
|
||||
itr1 = itr1.Next()
|
||||
} else if itr0.Item().Key == itr1.Item().Key {
|
||||
chunk := &Chunk{
|
||||
Key: itr0.Item().Key,
|
||||
Value: itr0.Item().Value.difference(itr1.Item().Value),
|
||||
}
|
||||
o_last_Key = o_node.Key
|
||||
a = a.Next()
|
||||
b = b.Next()
|
||||
|
||||
// Could not add if all zero
|
||||
if chunk.Value.bitcount() > 0 {
|
||||
output.AddChunk(chunk)
|
||||
}
|
||||
|
||||
itr0 = itr0.Next()
|
||||
itr1 = itr1.Next()
|
||||
} else {
|
||||
log.Warn("NEVER SHOULD BE HERE")
|
||||
break
|
||||
panic("unreachable")
|
||||
}
|
||||
}
|
||||
return output
|
||||
}
|
||||
|
||||
type ChunkIterator interface {
|
||||
Limit() bool
|
||||
Item() *Chunk
|
||||
Next() ChunkIterator
|
||||
Dump()
|
||||
Close()
|
||||
}
|
||||
type RBNodeIterator struct {
|
||||
rbiterator rbtree.Iterator
|
||||
}
|
||||
|
||||
func (r *RBNodeIterator) Limit() bool {
|
||||
return r.rbiterator.Limit()
|
||||
}
|
||||
func (r *RBNodeIterator) Next() ChunkIterator {
|
||||
r.rbiterator = r.rbiterator.Next()
|
||||
return r
|
||||
}
|
||||
func (r *RBNodeIterator) Dump() {
|
||||
}
|
||||
|
||||
func (r *RBNodeIterator) Close() {
|
||||
}
|
||||
func (r *RBNodeIterator) Item() *Chunk {
|
||||
if r.rbiterator.Item() != nil {
|
||||
return r.rbiterator.Item().(*Chunk)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
func GetChunk(bm IBitmap, ChunkKey uint64) *Chunk {
|
||||
look := &Chunk{ChunkKey, BlockArray{make([]uint64, 32, 32)}}
|
||||
return bm.Get(look)
|
||||
}
|
||||
|
||||
type IBitmap interface {
|
||||
AddChunk(*Chunk)
|
||||
Min() ChunkIterator
|
||||
Get(*Chunk) *Chunk
|
||||
Len() int
|
||||
Inc()
|
||||
Dec()
|
||||
Count() uint64
|
||||
SetCount(uint64)
|
||||
Bits() []uint64
|
||||
BuildFromBits(bits []uint64)
|
||||
ToBytes() []byte
|
||||
FromBytes([]byte)
|
||||
ToCompressString() string
|
||||
ToRawCompressString() (string, int)
|
||||
FromCompressString(string)
|
||||
}
|
||||
|
||||
func NewRB() *rbtree.Tree {
|
||||
return rbtree.NewTree(func(a, b rbtree.Item) int { return Compare(a.(*Chunk).Key, b.(*Chunk).Key) })
|
||||
}
|
||||
|
||||
func CreateRBBitmap() IBitmap {
|
||||
return &Bitmap{nodes: NewRB(), bcount: 0}
|
||||
}
|
||||
|
||||
func (self *Bitmap) FromCompressString(str string) {
|
||||
compressed_data, err := base64.StdEncoding.DecodeString(str)
|
||||
if err != nil {
|
||||
log.Warn(err)
|
||||
return
|
||||
}
|
||||
reader, _ := gzip.NewReader(bytes.NewReader(compressed_data))
|
||||
data, _ := ioutil.ReadAll(reader)
|
||||
self.FromBytes(data)
|
||||
}
|
||||
|
||||
func (self *Bitmap) ToCompressString() string {
|
||||
var b bytes.Buffer
|
||||
w := gzip.NewWriter(&b)
|
||||
w.Write(self.ToBytes())
|
||||
w.Flush()
|
||||
w.Close()
|
||||
return base64.StdEncoding.EncodeToString(b.Bytes())
|
||||
}
|
||||
|
||||
func (self *Bitmap) AddChunk(a *Chunk) {
|
||||
self.nodes.Insert(a)
|
||||
}
|
||||
func (b *Bitmap) Min() ChunkIterator {
|
||||
return &RBNodeIterator{b.nodes.Min()}
|
||||
}
|
||||
func (b *Bitmap) Get(a *Chunk) *Chunk {
|
||||
n := b.nodes.Get(a)
|
||||
if n != nil {
|
||||
return n.(*Chunk)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
// ToRawCompressString returns a compressed, hex-encoded string of b.
|
||||
func (b *Bitmap) ToRawCompressString() (string, int) {
|
||||
var bt bytes.Buffer
|
||||
buf := gzip.NewWriter(&bt)
|
||||
binary.Write(buf, binary.LittleEndian, uint64(b.nodes.Len()))
|
||||
binary.Write(buf, binary.LittleEndian, uint64(b.tree.Len()))
|
||||
max_slice := 0
|
||||
for i := b.nodes.Min(); !i.Limit(); i = i.Next() {
|
||||
for i := b.tree.Min(); !i.Limit(); i = i.Next() {
|
||||
obj := i.Item().(*Chunk)
|
||||
max_slice = int(obj.Key)
|
||||
binary.Write(buf, binary.LittleEndian, obj.Key)
|
||||
for _, v := range obj.Value.Block {
|
||||
for _, v := range obj.Value {
|
||||
binary.Write(buf, binary.LittleEndian, v)
|
||||
}
|
||||
}
|
||||
|
|
@ -450,13 +236,12 @@ func (b *Bitmap) ToRawCompressString() (string, int) {
|
|||
return base64.StdEncoding.EncodeToString(bt.Bytes()), max_slice
|
||||
}
|
||||
|
||||
// ToBytes returns a gob-encoded byte slice of b.
|
||||
func (b *Bitmap) ToBytes() []byte {
|
||||
var (
|
||||
buf bytes.Buffer
|
||||
)
|
||||
var buf bytes.Buffer
|
||||
enc := gob.NewEncoder(&buf)
|
||||
enc.Encode(b.nodes.Len())
|
||||
for i := b.nodes.Min(); !i.Limit(); i = i.Next() {
|
||||
enc.Encode(b.tree.Len())
|
||||
for i := b.tree.Min(); !i.Limit(); i = i.Next() {
|
||||
obj := i.Item().(*Chunk)
|
||||
err := enc.Encode(obj)
|
||||
if err != nil {
|
||||
|
|
@ -466,34 +251,31 @@ func (b *Bitmap) ToBytes() []byte {
|
|||
return buf.Bytes()
|
||||
}
|
||||
|
||||
func (self *Bitmap) FromBytes(raw []byte) {
|
||||
// FromBytes decodes a gob-encoded byte slice into b.
|
||||
func (b *Bitmap) FromBytes(raw []byte) {
|
||||
buf := bytes.NewBuffer(raw)
|
||||
dec := gob.NewDecoder(buf)
|
||||
|
||||
var size int
|
||||
dec.Decode(&size)
|
||||
self.nodes = NewRB()
|
||||
b.tree = rbtree.NewTree(rbtreeItemCompare)
|
||||
for i := 0; i < size; i++ {
|
||||
var chunk Chunk
|
||||
dec.Decode(&chunk)
|
||||
self.AddChunk(&chunk)
|
||||
b.AddChunk(&chunk)
|
||||
}
|
||||
self.SetCount(BitCount(self))
|
||||
b.SetCount(b.BitCount())
|
||||
}
|
||||
|
||||
func (b *Bitmap) BuildFromBits(bits []uint64) {
|
||||
for _, v := range bits {
|
||||
SetBit(b, v)
|
||||
}
|
||||
}
|
||||
// Bits returns the bits in b as a slice of ints.
|
||||
func (b *Bitmap) Bits() []uint64 {
|
||||
result := make([]uint64, b.Count())
|
||||
|
||||
x := 0
|
||||
for i := b.Min(); !i.Limit(); i = i.Next() {
|
||||
for i := b.ChunkIterator(); !i.Limit(); i = i.Next() {
|
||||
item := i.Item()
|
||||
chunk := item.Key
|
||||
for bi, block := range item.Value.Block {
|
||||
for bi, block := range item.Value {
|
||||
for bit := uint(0); bit < 64; bit++ {
|
||||
if (block & (1 << bit)) != 0 {
|
||||
idx := chunk << 11
|
||||
|
|
@ -507,25 +289,101 @@ func (b *Bitmap) Bits() []uint64 {
|
|||
}
|
||||
return result
|
||||
}
|
||||
func (b *Bitmap) Len() int {
|
||||
return b.nodes.Len()
|
||||
}
|
||||
func (b *Bitmap) Inc() {
|
||||
b.bcount += 1
|
||||
}
|
||||
func (b *Bitmap) Dec() {
|
||||
if b.bcount > 0 {
|
||||
b.bcount -= 1
|
||||
|
||||
// SetBit sets the i-th bit of the bitmap.
|
||||
func (b *Bitmap) SetBit(i uint64) (bool, *Chunk, Address) {
|
||||
address := deref(i)
|
||||
|
||||
chunk := b.Chunk(&Chunk{address.ChunkKey, make(Blocks, 32)})
|
||||
if chunk == nil {
|
||||
chunk = &Chunk{address.ChunkKey, make(Blocks, 32)}
|
||||
b.AddChunk(chunk)
|
||||
}
|
||||
}
|
||||
func (b *Bitmap) SetCount(c uint64) {
|
||||
b.bcount = c
|
||||
|
||||
changed := chunk.Value.setBit(address.BlockIndex, address.Bit)
|
||||
if changed {
|
||||
b.bcount++
|
||||
}
|
||||
|
||||
return changed, chunk, address
|
||||
}
|
||||
|
||||
func (b *Bitmap) Count() uint64 {
|
||||
return b.bcount
|
||||
// ClearBit clears the i-th bit of the bitmap.
|
||||
func (b *Bitmap) ClearBit(i uint64) (bool, *Chunk, Address) {
|
||||
address := deref(i)
|
||||
|
||||
chunk := b.Chunk(&Chunk{address.ChunkKey, make(Blocks, 32)})
|
||||
if chunk == nil {
|
||||
return false, nil, address
|
||||
}
|
||||
|
||||
changed := chunk.Value.clearBit(address.BlockIndex, address.Bit)
|
||||
if changed && b.bcount > 0 {
|
||||
b.bcount--
|
||||
}
|
||||
|
||||
return changed, chunk, address
|
||||
}
|
||||
|
||||
// Len returns the number of chunks in b.
|
||||
func (b *Bitmap) Len() int { return b.tree.Len() }
|
||||
|
||||
// SetCount sets the number of set bits in the bitmap.
|
||||
func (b *Bitmap) SetCount(c uint64) { b.bcount = c }
|
||||
|
||||
// Count returns the number of set bits in the bitmap.
|
||||
func (b *Bitmap) Count() uint64 { return b.bcount }
|
||||
|
||||
// BitCount calculates the number of set bits in the bitmap from raw chunk data.
|
||||
func (b *Bitmap) BitCount() uint64 {
|
||||
var n uint64
|
||||
for i := b.ChunkIterator(); !i.Limit(); i = i.Next() {
|
||||
n += i.Item().Value.bitcount()
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
// Chunk represents a set of blocks in a Bitmap.
|
||||
type Chunk struct {
|
||||
Key uint64
|
||||
Value Blocks
|
||||
}
|
||||
|
||||
// ChunkIterator represents an object for iterating over chunks in a bitmap.
|
||||
type ChunkIterator struct {
|
||||
itr rbtree.Iterator
|
||||
}
|
||||
|
||||
// Limit return true when the iterator is at the end of iteration.
|
||||
func (r *ChunkIterator) Limit() bool {
|
||||
return r.itr.Limit()
|
||||
}
|
||||
|
||||
// Next moves the iterator to the next chunk.
|
||||
func (r *ChunkIterator) Next() *ChunkIterator {
|
||||
r.itr = r.itr.Next()
|
||||
return r
|
||||
}
|
||||
|
||||
// Item returns the current item that the iterator is pointing at.
|
||||
func (r *ChunkIterator) Item() *Chunk {
|
||||
if r.itr.Item() != nil {
|
||||
return r.itr.Item().(*Chunk)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func rbtreeItemCompare(a, b rbtree.Item) int {
|
||||
aKey, bKey := a.(*Chunk).Key, b.(*Chunk).Key
|
||||
if aKey < bKey {
|
||||
return -1
|
||||
} else if aKey > bKey {
|
||||
return 1
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
// Address represents a location for a given chunk/block/bit.
|
||||
type Address struct {
|
||||
ChunkKey uint64
|
||||
BlockIndex uint8
|
||||
|
|
@ -533,58 +391,9 @@ type Address struct {
|
|||
}
|
||||
|
||||
func deref(pos uint64) Address {
|
||||
ChunkKey := pos >> 11 // div by 2048
|
||||
var bucket_offset = pos & 0x7FF // mod by 2048
|
||||
BlockIndex := uint8(bucket_offset >> 6) // div by 64
|
||||
bit_offset := uint8(bucket_offset & 0x3F) // mod by 64
|
||||
return Address{ChunkKey, BlockIndex, bit_offset}
|
||||
}
|
||||
|
||||
func SetBit(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 {
|
||||
node = &Chunk{address.ChunkKey, BlockArray{make([]uint64, 32, 32)}}
|
||||
b.AddChunk(node)
|
||||
} else {
|
||||
node = item
|
||||
}
|
||||
data_changed := node.Value.set_bit(address.BlockIndex, address.Bit)
|
||||
if data_changed {
|
||||
b.Inc()
|
||||
}
|
||||
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
|
||||
i := b.Min()
|
||||
defer i.Close()
|
||||
for ; !i.Limit(); i = i.Next() {
|
||||
var item = i.Item()
|
||||
total += item.Value.bitcount()
|
||||
}
|
||||
return total
|
||||
chunkKey := pos >> 11 // div by 2048
|
||||
offset := pos & 0x7FF // mod by 2048
|
||||
blockIndex := uint8(offset >> 6) // div by 64
|
||||
bit_offset := uint8(offset & 0x3F) // mod by 64
|
||||
return Address{chunkKey, blockIndex, bit_offset}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,14 +0,0 @@
|
|||
package index
|
||||
|
||||
import (
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestRBBitmap_SetBit(t *testing.T) {
|
||||
bm := CreateRBBitmap()
|
||||
SetBit(bm, 0)
|
||||
ClearBit(bm, 0)
|
||||
if BitCount(bm) != 0 {
|
||||
t.Error("Should be 0")
|
||||
}
|
||||
}
|
||||
69
index/blocks.go
Normal file
69
index/blocks.go
Normal file
|
|
@ -0,0 +1,69 @@
|
|||
package index
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
)
|
||||
|
||||
type Blocks []uint64
|
||||
|
||||
func (a Blocks) bitcount() uint64 {
|
||||
fmt.Println("bitcount:", a)
|
||||
return popcntSlice(a)
|
||||
}
|
||||
|
||||
func (a Blocks) union(other Blocks) Blocks {
|
||||
ret := make(Blocks, 32)
|
||||
for i, _ := range a {
|
||||
ret[i] = a[i] | other[i]
|
||||
}
|
||||
return ret
|
||||
}
|
||||
|
||||
func (a Blocks) invert() Blocks {
|
||||
other := make(Blocks, 32)
|
||||
for i, _ := range a {
|
||||
other[i] = ^a[i]
|
||||
}
|
||||
return other
|
||||
}
|
||||
|
||||
func (a Blocks) copy() Blocks {
|
||||
other := make(Blocks, 32)
|
||||
for i, _ := range a {
|
||||
other[i] = a[i]
|
||||
}
|
||||
return other
|
||||
}
|
||||
|
||||
func (a Blocks) andcount(other Blocks) uint64 {
|
||||
println("andcount")
|
||||
return popcntAndSliceAsm(a, other)
|
||||
}
|
||||
|
||||
func (a Blocks) intersection(other Blocks) Blocks {
|
||||
ret := make(Blocks, 32)
|
||||
for i, _ := range a {
|
||||
ret[i] = a[i] & other[i]
|
||||
}
|
||||
return ret
|
||||
}
|
||||
|
||||
func (a Blocks) difference(other Blocks) Blocks {
|
||||
ret := make(Blocks, 32)
|
||||
for i, _ := range a {
|
||||
ret[i] = a[i] &^ other[i]
|
||||
}
|
||||
return ret
|
||||
}
|
||||
|
||||
func (a Blocks) setBit(i uint8, bit uint8) (changed bool) {
|
||||
val := a[i] & (1 << bit)
|
||||
a[i] |= 1 << bit
|
||||
return val == 0
|
||||
}
|
||||
|
||||
func (a Blocks) clearBit(i uint8, bit uint8) (changed bool) {
|
||||
val := a[i] & (1 << bit)
|
||||
a[i] &= ^(1 << bit)
|
||||
return val != 0
|
||||
}
|
||||
238
index/brand.go
238
index/brand.go
|
|
@ -26,7 +26,7 @@ type Pair struct {
|
|||
}
|
||||
type Rank struct {
|
||||
*Pair
|
||||
bitmap IBitmap
|
||||
bitmap *Bitmap
|
||||
category uint64
|
||||
}
|
||||
|
||||
|
|
@ -66,100 +66,102 @@ func NewBrand(db string, frame string, slice int, s Storage, threshold_len int,
|
|||
return f
|
||||
|
||||
}
|
||||
func (self *Brand) Clear() bool {
|
||||
self.bitmap_cache = make(map[uint64]*Rank)
|
||||
|
||||
func (b *Brand) Clear() bool {
|
||||
b.bitmap_cache = make(map[uint64]*Rank)
|
||||
return true
|
||||
}
|
||||
func (self *Brand) Exists(bitmap_id uint64) bool {
|
||||
_, ok := self.bitmap_cache[bitmap_id]
|
||||
|
||||
func (b *Brand) Exists(bitmap_id uint64) bool {
|
||||
_, ok := b.bitmap_cache[bitmap_id]
|
||||
return ok
|
||||
}
|
||||
func (self *Brand) Get(bitmap_id uint64) IBitmap {
|
||||
bm, ok := self.bitmap_cache[bitmap_id]
|
||||
if ok {
|
||||
|
||||
func (b *Brand) Get(bitmap_id uint64) *Bitmap {
|
||||
if bm, ok := b.bitmap_cache[bitmap_id]; ok {
|
||||
return bm.bitmap
|
||||
}
|
||||
//I should fetch the category here..need to come up with a good source
|
||||
b, filter := self.storage.Fetch(bitmap_id, self.db, self.frame, self.slice)
|
||||
self.cache_it(b, bitmap_id, filter)
|
||||
return b
|
||||
|
||||
// I should fetch the category here..need to come up with a good source
|
||||
bm, filter := b.storage.Fetch(bitmap_id, b.db, b.frame, b.slice)
|
||||
b.cache_it(bm, bitmap_id, filter)
|
||||
return bm
|
||||
}
|
||||
|
||||
func (self *Brand) Get_nocache(bitmap_id uint64) (IBitmap, uint64) {
|
||||
bm, ok := self.bitmap_cache[bitmap_id]
|
||||
if ok {
|
||||
func (b *Brand) Get_nocache(bitmap_id uint64) (*Bitmap, uint64) {
|
||||
if bm, ok := b.bitmap_cache[bitmap_id]; 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)
|
||||
return b.storage.Fetch(bitmap_id, b.db, b.frame, b.slice)
|
||||
}
|
||||
|
||||
func (self *Brand) GetFilter(bitmap_id, filter uint64) IBitmap {
|
||||
b, old_filter := self.storage.Fetch(bitmap_id, self.db, self.frame, self.slice)
|
||||
func (b *Brand) GetFilter(bitmap_id, filter uint64) *Bitmap {
|
||||
bm, old_filter := b.storage.Fetch(bitmap_id, b.db, b.frame, b.slice)
|
||||
if filter == 0 {
|
||||
filter = old_filter
|
||||
}
|
||||
self.cache_it(b, bitmap_id, filter)
|
||||
return b
|
||||
b.cache_it(bm, bitmap_id, filter)
|
||||
return bm
|
||||
}
|
||||
|
||||
func (self *Brand) cache_it(bm IBitmap, bitmap_id uint64, category uint64) {
|
||||
if bm.Count() >= self.threshold_value {
|
||||
self.bitmap_cache[bitmap_id] = &Rank{&Pair{bitmap_id, bm.Count()}, bm, category}
|
||||
if len(self.bitmap_cache) > self.threshold_length {
|
||||
log.Info("RANK:", len(self.bitmap_cache), self.threshold_length, self.threshold_value)
|
||||
self.Rank()
|
||||
self.trim()
|
||||
func (b *Brand) cache_it(bm *Bitmap, bitmap_id uint64, category uint64) {
|
||||
if bm.Count() >= b.threshold_value {
|
||||
b.bitmap_cache[bitmap_id] = &Rank{&Pair{bitmap_id, bm.Count()}, bm, category}
|
||||
if len(b.bitmap_cache) > b.threshold_length {
|
||||
log.Info("RANK:", len(b.bitmap_cache), b.threshold_length, b.threshold_value)
|
||||
b.Rank()
|
||||
b.trim()
|
||||
}
|
||||
}
|
||||
}
|
||||
func (self *Brand) trim() {
|
||||
for k, item := range self.bitmap_cache {
|
||||
if item.bitmap.Count() <= self.threshold_value {
|
||||
delete(self.bitmap_cache, k)
|
||||
|
||||
func (b *Brand) trim() {
|
||||
for k, item := range b.bitmap_cache {
|
||||
if item.bitmap.Count() <= b.threshold_value {
|
||||
delete(b.bitmap_cache, k)
|
||||
}
|
||||
}
|
||||
log.Info("TRIM:", len(self.bitmap_cache), self.threshold_length)
|
||||
|
||||
log.Info("TRIM:", len(b.bitmap_cache), b.threshold_length)
|
||||
}
|
||||
|
||||
func (self *Brand) SetBit(bitmap_id uint64, bit_pos uint64, filter uint64) bool {
|
||||
bm1, ok := self.bitmap_cache[bitmap_id]
|
||||
var bm IBitmap
|
||||
func (b *Brand) SetBit(bitmap_id uint64, bit_pos uint64, filter uint64) bool {
|
||||
bm1, ok := b.bitmap_cache[bitmap_id]
|
||||
var bm *Bitmap
|
||||
if ok {
|
||||
bm = bm1.bitmap
|
||||
} else {
|
||||
bm = self.GetFilter(bitmap_id, filter) //aways overwrites what is in cass filter type
|
||||
bm = b.GetFilter(bitmap_id, filter) //aways overwrites what is in cass filter type
|
||||
}
|
||||
change, chunk, address := SetBit(bm, bit_pos)
|
||||
change, chunk, address := bm.SetBit(bit_pos)
|
||||
if change {
|
||||
val := chunk.Value.Block[address.BlockIndex]
|
||||
self.storage.StoreBit(bitmap_id, self.db, self.frame, self.slice, filter, address.ChunkKey, int32(address.BlockIndex), val, bm.Count())
|
||||
self.rank_count++
|
||||
val := chunk.Value[address.BlockIndex]
|
||||
b.storage.StoreBit(bitmap_id, b.db, b.frame, b.slice, filter, address.ChunkKey, int32(address.BlockIndex), val, bm.Count())
|
||||
b.rank_count++
|
||||
}
|
||||
return change
|
||||
}
|
||||
|
||||
func (self *Brand) Rank() {
|
||||
func (b *Brand) Rank() {
|
||||
start := time.Now()
|
||||
var list RankList
|
||||
for k, item := range self.bitmap_cache {
|
||||
for k, item := range b.bitmap_cache {
|
||||
list = append(list, &Rank{&Pair{k, item.bitmap.Count()}, item.bitmap, item.category})
|
||||
}
|
||||
sort.Sort(list)
|
||||
self.rankings = list
|
||||
if len(list) > self.threshold_idx {
|
||||
item := list[self.threshold_idx]
|
||||
self.threshold_value = item.bitmap.Count()
|
||||
b.rankings = list
|
||||
if len(list) > b.threshold_idx {
|
||||
item := list[b.threshold_idx]
|
||||
b.threshold_value = item.bitmap.Count()
|
||||
} else {
|
||||
self.threshold_value = 1
|
||||
b.threshold_value = 1
|
||||
}
|
||||
|
||||
self.rank_count = 0
|
||||
b.rank_count = 0
|
||||
delta := time.Since(start)
|
||||
util.SendTimer("brand_Rank", delta.Nanoseconds())
|
||||
self.rank_time = start
|
||||
|
||||
b.rank_time = start
|
||||
}
|
||||
|
||||
func packagePairs(r RankList) []Pair {
|
||||
|
|
@ -170,11 +172,11 @@ func packagePairs(r RankList) []Pair {
|
|||
return res
|
||||
}
|
||||
|
||||
func (self *Brand) Stats() interface{} {
|
||||
func (b *Brand) Stats() interface{} {
|
||||
total := uint64(0)
|
||||
i := uint64(0)
|
||||
bit_total := uint64(0)
|
||||
for _, v := range self.bitmap_cache {
|
||||
for _, v := range b.bitmap_cache {
|
||||
total += uint64(v.bitmap.Len()) * uint64(256)
|
||||
i += 1
|
||||
bit_total += v.Count
|
||||
|
|
@ -188,41 +190,43 @@ func (self *Brand) Stats() interface{} {
|
|||
|
||||
stats := map[string]interface{}{
|
||||
"total size of cache in bytes": total,
|
||||
"number of bitmaps": len(self.bitmap_cache),
|
||||
"number of bitmaps": len(b.bitmap_cache),
|
||||
"avg size of bitmap in space(bytes)": avg_bytes,
|
||||
"avg size of bitmap in bits": avg_bits,
|
||||
"rank counter": self.rank_count,
|
||||
"threshold_value": self.threshold_value,
|
||||
"threshold_length": self.threshold_length,
|
||||
"threshold_idx": self.threshold_idx,
|
||||
"skip": self.skip}
|
||||
"rank counter": b.rank_count,
|
||||
"threshold_value": b.threshold_value,
|
||||
"threshold_length": b.threshold_length,
|
||||
"threshold_idx": b.threshold_idx,
|
||||
"skip": b.skip}
|
||||
return stats
|
||||
}
|
||||
func (self *Brand) Store(bitmap_id uint64, bm IBitmap, filter uint64) {
|
||||
self.storage.Store(bitmap_id, self.db, self.frame, self.slice, filter, bm.(*Bitmap))
|
||||
self.cache_it(bm, bitmap_id, filter)
|
||||
func (b *Brand) Store(bitmap_id uint64, bm *Bitmap, filter uint64) {
|
||||
b.storage.Store(bitmap_id, b.db, b.frame, b.slice, filter, bm)
|
||||
b.cache_it(bm, bitmap_id, filter)
|
||||
}
|
||||
func (self *Brand) checkRank() {
|
||||
if len(self.rankings) < 50 {
|
||||
self.Rank()
|
||||
} else if self.rank_count > 0 {
|
||||
last := time.Since(self.rank_time) * time.Second
|
||||
func (b *Brand) checkRank() {
|
||||
if len(b.rankings) < 50 {
|
||||
b.Rank()
|
||||
return
|
||||
}
|
||||
|
||||
if b.rank_count > 0 {
|
||||
last := time.Since(b.rank_time) * time.Second
|
||||
if last > 60*5 {
|
||||
self.Rank()
|
||||
b.Rank()
|
||||
}
|
||||
}
|
||||
}
|
||||
func (self *Brand) TopN(src_bitmap IBitmap, n int, categories []uint64) []Pair {
|
||||
self.checkRank()
|
||||
is := NewIntSet()
|
||||
func (b *Brand) TopN(src_bitmap *Bitmap, n int, categories []uint64) []Pair {
|
||||
b.checkRank()
|
||||
set := make(map[uint64]struct{})
|
||||
for _, v := range categories {
|
||||
is.Add(v)
|
||||
|
||||
set[v] = struct{}{}
|
||||
}
|
||||
|
||||
test := self.TopNCat(src_bitmap, n, is)
|
||||
return test
|
||||
return b.TopNCat(src_bitmap, n, set)
|
||||
}
|
||||
|
||||
func dump(r RankList, n int) {
|
||||
for i, v := range r {
|
||||
log.Info(i, v)
|
||||
|
|
@ -232,23 +236,23 @@ func dump(r RankList, n int) {
|
|||
}
|
||||
}
|
||||
|
||||
func (self *Brand) TopNAll(n int, categories []uint64) []Pair {
|
||||
func (b *Brand) TopNAll(n int, categories []uint64) []Pair {
|
||||
log.Trace("TopNAll")
|
||||
|
||||
self.checkRank()
|
||||
b.checkRank()
|
||||
results := make([]Pair, 0, 0)
|
||||
|
||||
category := NewIntSet()
|
||||
set := make(map[uint64]struct{})
|
||||
needCat := false
|
||||
for _, v := range categories {
|
||||
category.Add(v)
|
||||
set[v] = struct{}{}
|
||||
needCat = true
|
||||
|
||||
}
|
||||
|
||||
count := 0
|
||||
for _, pair := range self.rankings {
|
||||
for _, pair := range b.rankings {
|
||||
if needCat {
|
||||
if !category.Contains(pair.category) {
|
||||
if _, ok := set[pair.category]; !ok {
|
||||
continue
|
||||
}
|
||||
}
|
||||
|
|
@ -264,7 +268,7 @@ func (self *Brand) TopNAll(n int, categories []uint64) []Pair {
|
|||
return results
|
||||
}
|
||||
|
||||
func (self *Brand) TopNCat(src_bitmap IBitmap, n int, category *IntSet) []Pair {
|
||||
func (b *Brand) TopNCat(src_bitmap *Bitmap, n int, set map[uint64]struct{}) []Pair {
|
||||
breakout := 1000
|
||||
var (
|
||||
o *Rank
|
||||
|
|
@ -273,11 +277,10 @@ func (self *Brand) TopNCat(src_bitmap IBitmap, n int, category *IntSet) []Pair {
|
|||
counter := 0
|
||||
x := 0
|
||||
|
||||
needCat := category.Size() > 0
|
||||
for i, pair := range self.rankings {
|
||||
|
||||
needCat := (len(set) > 0)
|
||||
for i, pair := range b.rankings {
|
||||
if needCat {
|
||||
if !category.Contains(pair.category) {
|
||||
if _, ok := set[pair.category]; !ok {
|
||||
continue
|
||||
}
|
||||
}
|
||||
|
|
@ -285,7 +288,7 @@ func (self *Brand) TopNCat(src_bitmap IBitmap, n int, category *IntSet) []Pair {
|
|||
if counter > n {
|
||||
break
|
||||
}
|
||||
bc := IntersectionCount(src_bitmap, pair.bitmap)
|
||||
bc := src_bitmap.IntersectionCount(pair.bitmap)
|
||||
if bc > 0 {
|
||||
results = append(results, &Rank{&Pair{pair.Key, bc}, nil, pair.category})
|
||||
counter = counter + 1
|
||||
|
|
@ -307,11 +310,11 @@ func (self *Brand) TopNCat(src_bitmap IBitmap, n int, category *IntSet) []Pair {
|
|||
|
||||
results = append(results, o)
|
||||
|
||||
for i := x + 1; i < len(self.rankings); i++ {
|
||||
o = self.rankings[i]
|
||||
for i := x + 1; i < len(b.rankings); i++ {
|
||||
o = b.rankings[i]
|
||||
|
||||
if needCat {
|
||||
if !category.Contains(o.category) {
|
||||
if _, ok := set[o.category]; !ok {
|
||||
continue
|
||||
}
|
||||
counter = counter + 1
|
||||
|
|
@ -330,7 +333,7 @@ func (self *Brand) TopNCat(src_bitmap IBitmap, n int, category *IntSet) []Pair {
|
|||
|
||||
}
|
||||
|
||||
bc := IntersectionCount(src_bitmap, o.bitmap)
|
||||
bc := src_bitmap.IntersectionCount(o.bitmap)
|
||||
|
||||
if bc > current_threshold {
|
||||
if results[end-1].Count > bc {
|
||||
|
|
@ -346,35 +349,35 @@ func (self *Brand) TopNCat(src_bitmap IBitmap, n int, category *IntSet) []Pair {
|
|||
}
|
||||
return packagePairs(results[:end])
|
||||
}
|
||||
func (self *Brand) getFileName() string {
|
||||
func (b *Brand) getFileName() string {
|
||||
base := FragmentBase
|
||||
if base == "" {
|
||||
base = "."
|
||||
}
|
||||
|
||||
return fmt.Sprintf("%s/%s.%s.%d.json", base, self.db, self.frame, self.slice)
|
||||
return fmt.Sprintf("%s/%s.%s.%d.json", base, b.db, b.frame, b.slice)
|
||||
}
|
||||
|
||||
func (self *Brand) Persist() error {
|
||||
log.Info("Brand Persist:", self.getFileName())
|
||||
self.storage.FlushBatch()
|
||||
asize := len(self.bitmap_cache)
|
||||
func (b *Brand) Persist() error {
|
||||
log.Info("Brand Persist:", b.getFileName())
|
||||
b.storage.FlushBatch()
|
||||
asize := len(b.bitmap_cache)
|
||||
|
||||
if asize == 0 {
|
||||
log.Warn("Nothing to save ", self.getFileName())
|
||||
log.Warn("Nothing to save ", b.getFileName())
|
||||
return nil
|
||||
}
|
||||
w, err := util.Create(self.getFileName())
|
||||
w, err := util.Create(b.getFileName())
|
||||
if err != nil {
|
||||
log.Warn("Error opening outfile ", self.getFileName())
|
||||
log.Warn("Error opening outfile ", b.getFileName())
|
||||
log.Warn(err)
|
||||
return err
|
||||
}
|
||||
defer w.Close()
|
||||
defer self.storage.Close()
|
||||
defer b.storage.Close()
|
||||
|
||||
var list RankList
|
||||
for k, item := range self.bitmap_cache {
|
||||
for k, item := range b.bitmap_cache {
|
||||
list = append(list, &Rank{&Pair{k, item.bitmap.Count()}, item.bitmap, item.category})
|
||||
}
|
||||
|
||||
|
|
@ -391,12 +394,12 @@ func (self *Brand) Persist() error {
|
|||
return encoder.Encode(results)
|
||||
}
|
||||
|
||||
func (self *Brand) Load(requestChan chan Command, f *Fragment) {
|
||||
func (b *Brand) Load(requestChan chan Command, f *Fragment) {
|
||||
log.Warn("Brand Load")
|
||||
time.Sleep(time.Duration(rand.Intn(32)) * time.Second) //trying to avoid mass cassandra hit
|
||||
r, err := util.Open(self.getFileName())
|
||||
r, err := util.Open(b.getFileName())
|
||||
if err != nil {
|
||||
log.Warn("NO Brand Init File:", self.getFileName())
|
||||
log.Warn("NO Brand Init File:", b.getFileName())
|
||||
return
|
||||
}
|
||||
dec := json.NewDecoder(r)
|
||||
|
|
@ -416,30 +419,31 @@ func (self *Brand) Load(requestChan chan Command, f *Fragment) {
|
|||
}
|
||||
}
|
||||
|
||||
func (self *Brand) ClearBit(bitmap_id uint64, bit_pos uint64) bool {
|
||||
func (b *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
|
||||
bm1, ok := b.bitmap_cache[bitmap_id]
|
||||
var bm *Bitmap
|
||||
filter := uint64(0)
|
||||
if ok {
|
||||
bm = bm1.bitmap
|
||||
filter = bm1.category
|
||||
} else {
|
||||
bm, filter = self.Get_nocache(bitmap_id)
|
||||
bm, filter = b.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]
|
||||
changed, chunk, address := bm.ClearBit(bit_pos)
|
||||
if changed {
|
||||
val := chunk.Value[address.BlockIndex]
|
||||
if val == 0 {
|
||||
self.storage.RemoveBit(bitmap_id, self.db, self.frame, self.slice, filter, address.ChunkKey, int32(address.BlockIndex), bm.Count())
|
||||
b.storage.RemoveBit(bitmap_id, b.db, b.frame, b.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())
|
||||
b.storage.StoreBit(bitmap_id, b.db, b.frame, b.slice, filter, address.ChunkKey, int32(address.BlockIndex), val, bm.Count())
|
||||
}
|
||||
self.rank_count++
|
||||
b.rank_count++
|
||||
}
|
||||
return change
|
||||
|
||||
return changed
|
||||
}
|
||||
|
|
|
|||
|
|
@ -3,6 +3,7 @@ package index
|
|||
import (
|
||||
"bytes"
|
||||
"compress/gzip"
|
||||
"encoding/base64"
|
||||
"io/ioutil"
|
||||
"time"
|
||||
|
||||
|
|
@ -15,23 +16,21 @@ type Result struct {
|
|||
}
|
||||
|
||||
type Responder struct {
|
||||
result chan Result
|
||||
query_type string
|
||||
result chan Result
|
||||
queryType string
|
||||
}
|
||||
|
||||
func NewResponder(query_type string) *Responder {
|
||||
return &Responder{make(chan Result), query_type}
|
||||
}
|
||||
func (self *Responder) QueryType() string {
|
||||
return self.query_type
|
||||
}
|
||||
func (self *Responder) Response() Result {
|
||||
return <-self.result
|
||||
}
|
||||
func (self *Responder) ResponseChannel() chan Result {
|
||||
return self.result
|
||||
func NewResponder(queryType string) *Responder {
|
||||
return &Responder{
|
||||
result: make(chan Result),
|
||||
queryType: queryType,
|
||||
}
|
||||
}
|
||||
|
||||
func (r *Responder) QueryType() string { return r.queryType }
|
||||
func (r *Responder) Response() Result { return <-r.result }
|
||||
func (r *Responder) ResponseChannel() chan Result { return r.result }
|
||||
|
||||
type Calculation interface{}
|
||||
|
||||
type Command interface {
|
||||
|
|
@ -68,7 +67,7 @@ func (self *CmdCount) Execute(f *Fragment) Calculation {
|
|||
if ok == false {
|
||||
return uint64(0)
|
||||
}
|
||||
return BitCount(bm)
|
||||
return bm.BitCount()
|
||||
}
|
||||
|
||||
type CmdUnion struct {
|
||||
|
|
@ -238,9 +237,17 @@ func NewLoader(bitmap_id uint64, compressed_bitmap string, filter uint64) *CmdLo
|
|||
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, self.filter)
|
||||
buf, err := base64.StdEncoding.DecodeString(self.compressed_bitmap)
|
||||
if err != nil {
|
||||
log.Warn(err)
|
||||
return "ok"
|
||||
}
|
||||
reader, _ := gzip.NewReader(bytes.NewReader(buf))
|
||||
data, _ := ioutil.ReadAll(reader)
|
||||
|
||||
bm := NewBitmap()
|
||||
bm.FromBytes(data)
|
||||
f.impl.Store(self.bitmap_id, bm, self.filter)
|
||||
return "ok"
|
||||
}
|
||||
|
||||
|
|
@ -296,7 +303,7 @@ func NewMask(start, end uint64) *CmdMask {
|
|||
func (self *CmdMask) Execute(f *Fragment) Calculation {
|
||||
result := NewBitmap()
|
||||
for i := self.start; i < self.end; i++ {
|
||||
SetBit(result, i)
|
||||
result.SetBit(i)
|
||||
}
|
||||
return f.AllocHandle(result)
|
||||
}
|
||||
|
|
@ -325,7 +332,7 @@ func (self *CmdTopFill) Execute(f *Fragment) Calculation {
|
|||
res := f.intersect([]BitmapHandle{self.args.Handle, a})
|
||||
bm, ok := f.getBitmap(res)
|
||||
if ok {
|
||||
bc := BitCount(bm)
|
||||
bc := bm.BitCount()
|
||||
if bc > 0 {
|
||||
result = append(result, Pair{v, bc})
|
||||
}
|
||||
|
|
|
|||
|
|
@ -343,13 +343,13 @@ func (self *FragmentContainer) AddFragment(db string, frame string, slice int, i
|
|||
}
|
||||
|
||||
type Pilosa interface {
|
||||
Get(id uint64) IBitmap
|
||||
Get(id uint64) *Bitmap
|
||||
SetBit(id uint64, bit_pos uint64, filter uint64) bool
|
||||
ClearBit(id uint64, bit_pos uint64) bool
|
||||
TopN(b IBitmap, n int, categories []uint64) []Pair
|
||||
TopN(b *Bitmap, n int, categories []uint64) []Pair
|
||||
TopNAll(n int, categories []uint64) []Pair
|
||||
Clear() bool
|
||||
Store(bitmap_id uint64, bm IBitmap, filter uint64)
|
||||
Store(bitmap_id uint64, bm *Bitmap, filter uint64)
|
||||
Stats() interface{}
|
||||
Persist() error
|
||||
Load(requestChan chan Command, fragment *Fragment)
|
||||
|
|
@ -401,10 +401,10 @@ func NewFragment(frag_id util.SUUID, db string, slice int, frame string) *Fragme
|
|||
return f
|
||||
}
|
||||
|
||||
func (self *Fragment) getBitmap(bitmap BitmapHandle) (IBitmap, bool) {
|
||||
func (self *Fragment) getBitmap(bitmap BitmapHandle) (*Bitmap, bool) {
|
||||
bm, ok := self.cache.Get(bitmap)
|
||||
if ok && bm != nil {
|
||||
return bm.(IBitmap), ok
|
||||
return bm.(*Bitmap), ok
|
||||
}
|
||||
return NewBitmap(), false //cache fail but return ting em
|
||||
}
|
||||
|
|
@ -430,7 +430,7 @@ func (self *Fragment) NewHandle(bitmap_id uint64) BitmapHandle {
|
|||
return self.AllocHandle(bm)
|
||||
//given a bitmap_id return a newly allocated handle
|
||||
}
|
||||
func (self *Fragment) AllocHandle(bm IBitmap) BitmapHandle {
|
||||
func (self *Fragment) AllocHandle(bm *Bitmap) BitmapHandle {
|
||||
handle := self.nextHandle()
|
||||
self.cache.Add(handle, bm)
|
||||
return handle
|
||||
|
|
@ -452,7 +452,7 @@ func (self *Fragment) union(bitmaps []BitmapHandle) BitmapHandle {
|
|||
if i == 0 {
|
||||
result = bm
|
||||
} else {
|
||||
result = Union(result, bm)
|
||||
result = result.Union(bm)
|
||||
}
|
||||
}
|
||||
return self.AllocHandle(result)
|
||||
|
|
@ -464,20 +464,20 @@ func (self *Fragment) build_time_range_bitmap(bitmap_id uint64, start, end time.
|
|||
if i == 0 {
|
||||
result = bm
|
||||
} else {
|
||||
result = Union(result, bm)
|
||||
result = result.Union(bm)
|
||||
}
|
||||
}
|
||||
return self.AllocHandle(result)
|
||||
}
|
||||
|
||||
func (self *Fragment) intersect(bitmaps []BitmapHandle) BitmapHandle {
|
||||
var result IBitmap
|
||||
var result *Bitmap
|
||||
for i, id := range bitmaps {
|
||||
bm, _ := self.getBitmap(id)
|
||||
if i == 0 {
|
||||
result = Clone(bm)
|
||||
result = bm.Clone()
|
||||
} else {
|
||||
result = Intersection(result, bm)
|
||||
result = result.Intersection(bm)
|
||||
}
|
||||
}
|
||||
return self.AllocHandle(result)
|
||||
|
|
@ -490,7 +490,7 @@ func (self *Fragment) difference(bitmaps []BitmapHandle) BitmapHandle {
|
|||
if i == 0 {
|
||||
result = bm
|
||||
} else {
|
||||
result = Difference(result, bm)
|
||||
result = result.Difference(bm)
|
||||
}
|
||||
}
|
||||
return self.AllocHandle(result)
|
||||
|
|
|
|||
|
|
@ -233,17 +233,30 @@ func TestFragmentContainer_Clear(t *testing.T) {
|
|||
}
|
||||
}
|
||||
|
||||
// Ensure a fragment can be loaded from a compressed form.
|
||||
func TestFragmentContainer_LoadBitmap(t *testing.T) {
|
||||
// Ensure a fragment can be marshaled and unmarshaled to and from bytes.
|
||||
func TestFragmentContainer_FromBytes(t *testing.T) {
|
||||
fc := NewFragmentContainer()
|
||||
fc.AddFragment("25", "b.n", 0, 2)
|
||||
|
||||
// Set bits for a bitmap.
|
||||
for i := 0; i < 4096; i++ {
|
||||
fc.MustSetBit(2, 1, uint64(i), 0)
|
||||
}
|
||||
|
||||
// Marshal to bytes.
|
||||
buf, err := fc.GetBytes(2, fc.MustGet(2, 1))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// Load a bitmap from compressed data.
|
||||
buf := "H4sIAAAJbogA/2JmYWBR+9/IzMjI6pxRmpfN+L+JgZGJkdk7tZKRjYGRNSwxpzSV8X8LAwOD8v9moDIup5z85GzHoqLESpAwI1AjWITxfxtQjdj/ViZGRo7o2NLMvBIzE5Ag0BiGf4zq/5uYGBV+/IeCUQZWBiikNP83AQN1NKwIMRgYAAAAAP//AQAA//9U05AivAIAAA=="
|
||||
fc.LoadBitmap(2, 1029, buf, 0)
|
||||
bh, err := fc.FromBytes(2, buf)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// Load and count bits.
|
||||
if n := fc.MustCount(2, fc.MustGet(2, 1029)); n != 4096 {
|
||||
if n := fc.MustCount(2, bh); n != 4096 {
|
||||
t.Fatalf("unexpected bit count: %d", n)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -40,7 +40,7 @@ func (self *General) Exists(bitmap_id uint64) bool {
|
|||
return ok
|
||||
|
||||
}
|
||||
func (self *General) Get(bitmap_id uint64) IBitmap {
|
||||
func (self *General) Get(bitmap_id uint64) *Bitmap {
|
||||
bm, ok := self.bitmap_cache.Get(bitmap_id)
|
||||
if ok && bm != nil {
|
||||
return bm.(*Bitmap)
|
||||
|
|
@ -52,20 +52,20 @@ func (self *General) Get(bitmap_id uint64) IBitmap {
|
|||
}
|
||||
func (self *General) SetBit(bitmap_id uint64, bit_pos uint64, filter uint64) bool {
|
||||
bm := self.Get(bitmap_id)
|
||||
change, chunk, address := SetBit(bm, bit_pos)
|
||||
change, chunk, address := bm.SetBit(bit_pos)
|
||||
if change {
|
||||
val := chunk.Value.Block[address.BlockIndex]
|
||||
val := chunk.Value[address.BlockIndex]
|
||||
self.storage.StoreBit(bitmap_id, self.db, self.frame, self.slice, filter, address.ChunkKey, int32(address.BlockIndex), val, bm.Count())
|
||||
}
|
||||
return change
|
||||
}
|
||||
func (self *General) TopN(b IBitmap, n int, categories []uint64) []Pair {
|
||||
func (self *General) TopN(b *Bitmap, n int, categories []uint64) []Pair {
|
||||
var empty []Pair
|
||||
|
||||
return empty
|
||||
}
|
||||
func (self *General) Store(bitmap_id uint64, bm IBitmap, filter uint64) {
|
||||
self.storage.Store(bitmap_id, self.db, self.frame, self.slice, filter, bm.(*Bitmap))
|
||||
func (self *General) Store(bitmap_id uint64, bm *Bitmap, filter uint64) {
|
||||
self.storage.Store(bitmap_id, self.db, self.frame, self.slice, filter, bm)
|
||||
self.bitmap_cache.Add(bitmap_id, bm)
|
||||
self.keys[bitmap_id] = 0
|
||||
}
|
||||
|
|
@ -152,9 +152,9 @@ func (self *General) ClearBit(bitmap_id uint64, bit_pos uint64) bool {
|
|||
if bm.Count() == 0 {
|
||||
return false
|
||||
}
|
||||
change, chunk, address := ClearBit(bm, bit_pos)
|
||||
change, chunk, address := bm.ClearBit(bit_pos)
|
||||
if change {
|
||||
val := chunk.Value.Block[address.BlockIndex]
|
||||
val := chunk.Value[address.BlockIndex]
|
||||
if val == 0 {
|
||||
self.storage.RemoveBit(bitmap_id, self.db, self.frame, self.slice, uint64(0), address.ChunkKey, int32(address.BlockIndex), bm.Count())
|
||||
} else {
|
||||
|
|
@ -164,7 +164,7 @@ func (self *General) ClearBit(bitmap_id uint64, bit_pos uint64) bool {
|
|||
return change
|
||||
}
|
||||
|
||||
func (self *General) Get_nocache(bitmap_id uint64) IBitmap {
|
||||
func (self *General) Get_nocache(bitmap_id uint64) *Bitmap {
|
||||
bm, ok := self.bitmap_cache.Get(bitmap_id)
|
||||
if ok && bm != nil {
|
||||
return bm.(*Bitmap)
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
package index
|
||||
|
||||
type Storage interface {
|
||||
Fetch(bitmap_id uint64, db string, frame string, slice int) (IBitmap, uint64)
|
||||
Fetch(bitmap_id uint64, db string, frame string, slice int) (*Bitmap, 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, chunk uint64, block_index int32, block, count uint64)
|
||||
|
|
|
|||
|
|
@ -87,10 +87,9 @@ func NewCassStorage() Storage {
|
|||
|
||||
func (c *CassandraStorage) Close() {
|
||||
}
|
||||
func (c *CassandraStorage) Fetch(bitmap_id uint64, db string, frame string, slice int) (IBitmap, uint64) {
|
||||
var dumb = COUNTERMASK
|
||||
last_key := int64(dumb)
|
||||
marker := int64(dumb)
|
||||
|
||||
func (c *CassandraStorage) Fetch(bitmap_id uint64, db string, frame string, slice int) (*Bitmap, uint64) {
|
||||
last_key, marker := int64(-1), int64(-1)
|
||||
var id = util.Uint64ToInt64(bitmap_id)
|
||||
start := time.Now()
|
||||
var (
|
||||
|
|
@ -101,7 +100,7 @@ func (c *CassandraStorage) Fetch(bitmap_id uint64, db string, frame string, slic
|
|||
filter int
|
||||
)
|
||||
|
||||
bitmap := CreateRBBitmap()
|
||||
bitmap := NewBitmap()
|
||||
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)
|
||||
|
||||
|
|
@ -109,10 +108,10 @@ func (c *CassandraStorage) Fetch(bitmap_id uint64, db string, frame string, slic
|
|||
s8 = uint8(block_index)
|
||||
if chunk_key != marker {
|
||||
if chunk_key != last_key {
|
||||
chunk = &Chunk{uint64(chunk_key), BlockArray{make([]uint64, 32, 32)}}
|
||||
chunk = &Chunk{uint64(chunk_key), make(Blocks, 32)}
|
||||
bitmap.AddChunk(chunk)
|
||||
}
|
||||
chunk.Value.Block[s8] = uint64(block)
|
||||
chunk.Value[s8] = uint64(block)
|
||||
|
||||
} else {
|
||||
count = block
|
||||
|
|
@ -164,17 +163,17 @@ func (self *CassandraStorage) EndBatch() {
|
|||
|
||||
func (self *CassandraStorage) Store(id uint64, db string, frame string, slice int, filter uint64, bitmap *Bitmap) error {
|
||||
self.BeginBatch()
|
||||
for i := bitmap.Min(); !i.Limit(); i = i.Next() {
|
||||
for i := bitmap.ChunkIterator(); !i.Limit(); i = i.Next() {
|
||||
var chunk = i.Item()
|
||||
for idx, block := range chunk.Value.Block {
|
||||
for idx, block := range chunk.Value {
|
||||
block_index := int32(idx)
|
||||
if block != 0 {
|
||||
self.StoreBlock(id, db, frame, slice, filter, chunk.Key, block_index, block)
|
||||
}
|
||||
}
|
||||
}
|
||||
cnt := BitCount(bitmap)
|
||||
self.StoreBlock(id, db, frame, slice, filter, COUNTERMASK, 0, cnt)
|
||||
cnt := bitmap.BitCount()
|
||||
self.StoreBlock(id, db, frame, slice, filter, CounterMask, 0, cnt)
|
||||
self.EndBatch()
|
||||
return nil
|
||||
}
|
||||
|
|
@ -197,7 +196,7 @@ func (self *CassandraStorage) StoreBlock(bid uint64, db string, frame string, sl
|
|||
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.StoreBlock(bid, db, frame, slice, filter, CounterMask, 0, count)
|
||||
self.EndBatch()
|
||||
}
|
||||
|
||||
|
|
@ -205,7 +204,7 @@ func (self *CassandraStorage) RemoveBit(id uint64, db string, frame string, slic
|
|||
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.StoreBlock(id, db, frame, slice, filter, CounterMask, 0, count)
|
||||
self.EndBatch()
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -67,28 +67,28 @@ func decodeValue(value []byte) (uint64, uint64) {
|
|||
return block, filter
|
||||
}
|
||||
|
||||
func (self *LevelDBStorage) Fetch(bitmap_id uint64, db string, frame string, slice int) (IBitmap, uint64) {
|
||||
func (self *LevelDBStorage) Fetch(bitmap_id uint64, db string, frame string, slice int) (*Bitmap, uint64) {
|
||||
start := time.Now()
|
||||
var (
|
||||
chunk *Chunk
|
||||
filter, block, last_key uint64
|
||||
)
|
||||
|
||||
bitmap := CreateRBBitmap()
|
||||
bitmap := NewBitmap()
|
||||
count := uint64(0)
|
||||
start_key := encodeKey(bitmap_id, 0, 0)
|
||||
limit_key := encodeKey(bitmap_id+1, 0, 0)
|
||||
iter := self.db.NewIterator(&Range{Start: start_key, Limit: limit_key}, nil)
|
||||
last_key = COUNTERMASK
|
||||
last_key = CounterMask
|
||||
for iter.Next() {
|
||||
_, chunk_key, block_index := decodeKey(iter.Key())
|
||||
block, filter = decodeValue(iter.Value())
|
||||
if chunk_key != COUNTERMASK {
|
||||
if chunk_key != CounterMask {
|
||||
if chunk_key != last_key {
|
||||
chunk = &Chunk{chunk_key, BlockArray{make([]uint64, 32, 32)}}
|
||||
chunk = &Chunk{chunk_key, make(Blocks, 32)}
|
||||
bitmap.AddChunk(chunk)
|
||||
}
|
||||
chunk.Value.Block[block_index] = block
|
||||
chunk.Value[block_index] = block
|
||||
|
||||
} else {
|
||||
count = block
|
||||
|
|
@ -142,18 +142,17 @@ func (self *LevelDBStorage) EndBatch() {
|
|||
|
||||
func (self *LevelDBStorage) Store(id uint64, db string, frame string, slice int, filter uint64, bitmap *Bitmap) error {
|
||||
self.BeginBatch()
|
||||
for i := bitmap.Min(); !i.Limit(); i = i.Next() {
|
||||
for i := bitmap.ChunkIterator(); !i.Limit(); i = i.Next() {
|
||||
var chunk = i.Item()
|
||||
for idx, block := range chunk.Value.Block {
|
||||
for idx, block := range chunk.Value {
|
||||
block_index := int32(idx)
|
||||
if block != 0 {
|
||||
self.StoreBlock(id, db, frame, slice, filter, chunk.Key, block_index, block)
|
||||
}
|
||||
}
|
||||
}
|
||||
cnt := BitCount(bitmap)
|
||||
|
||||
self.StoreBlock(id, db, frame, slice, filter, COUNTERMASK, 0, cnt)
|
||||
self.StoreBlock(id, db, frame, slice, filter, CounterMask, 0, bitmap.BitCount())
|
||||
self.EndBatch()
|
||||
return nil
|
||||
}
|
||||
|
|
@ -183,7 +182,7 @@ func (self *LevelDBStorage) Close() {
|
|||
func (self *LevelDBStorage) StoreBit(bid uint64, db string, frame string, slice int, filter uint64, bchunk uint64, block_index int32, bblock, count uint64) {
|
||||
self.BeginBatch()
|
||||
self.StoreBlock(bid, db, frame, slice, filter, bchunk, block_index, bblock)
|
||||
self.StoreBlock(bid, db, frame, slice, filter, COUNTERMASK, 0, count)
|
||||
self.StoreBlock(bid, db, frame, slice, filter, CounterMask, 0, count)
|
||||
self.EndBatch()
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -23,11 +23,11 @@ func (c *MemoryStorage) EndBatch() {}
|
|||
|
||||
func (c *MemoryStorage) FlushBatch() {}
|
||||
|
||||
func (c *MemoryStorage) Fetch(bitmap_id uint64, db string, frame string, slice int) (IBitmap, uint64) {
|
||||
func (c *MemoryStorage) Fetch(bitmap_id uint64, db string, frame string, slice int) (*Bitmap, uint64) {
|
||||
key := fmt.Sprintf("%d:%s:%s:%d", bitmap_id, db, frame, slice)
|
||||
bitmap, found := c.db[key]
|
||||
if !found {
|
||||
bitmap = CreateRBBitmap().(*Bitmap)
|
||||
bitmap = NewBitmap()
|
||||
c.db[key] = bitmap
|
||||
}
|
||||
return bitmap, 0
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue