refactor storage

This commit moves individual storage engines to their own packages.
They can be registered and instantiated by a string name.

cmd/pilosa/config.go

@@ -4,6 +4,8 @@ import (
 	"time"
 
 	"github.com/umbel/pilosa/index"
+	_ "github.com/umbel/pilosa/index/storage"
+	"github.com/umbel/pilosa/index/storage/cassandra"
 	"github.com/umbel/pilosa/transport"
 	"github.com/umbel/pilosa/util"
 )
@@ -91,12 +93,12 @@ func NewConfig() Config {
 	c.HTTP.Port = transport.DefaultHTTPPort
 	c.Log.Path = DefaultLogPath
 	c.Storage.Backend = index.DefaultBackend
-	c.Storage.Hosts = index.DefaultStorageHosts[:]
-	c.Storage.Keyspace = index.DefaultStorageKeyspace
+	c.Storage.Hosts = cassandra.DefaultHosts
+	c.Storage.Keyspace = cassandra.DefaultKeyspace
 	c.Storage.FragmentBase = DefaultFragmentBase
 	c.Storage.SupportedFrames = DefaultSupportedFrames[:]
-	c.Storage.CassandraTimeWindow = Duration(index.DefaultCassandraTimeWindow)
-	c.Storage.CassandraMaxSizeBatch = index.DefaultCassandraMaxSizeBatch
+	c.Storage.CassandraTimeWindow = Duration(cassandra.DefaultFlushInterval)
+	c.Storage.CassandraMaxSizeBatch = cassandra.DefaultFlushThreshold
 	c.Statsd.Host = util.DefaultStatsdHost
 	c.ETCD.Hosts = DefaultETCDHosts[:]
 	return c

cmd/pilosa/main.go

@@ -108,9 +108,6 @@ func (m *Main) Run(args ...string) error {
 	util.StatsdHost = config.Statsd.Host
 	util.SetupStatsd()
 
-	// REMOVED(benbjohnson): config.SetupConfig()
-	index.SetupCassandra()
-
 	// Initialize logging.
 	logger, _ := log.LoggerFromConfigAsBytes([]byte(SeelogProductionConfig(config.Log.Path, *id, config.Log.Level)))
 	log.ReplaceLogger(logger)

core/etcd.go

@@ -299,7 +299,7 @@ func (self *TopologyMapper) remove_fragment(node *etcd.Node) error {
 func flatten(node *etcd.Node) []*etcd.Node {
 	nodes := []*etcd.Node{node}
 	for i := 0; i < len(node.Nodes); i++ {
-		nodes = append(nodes, flatten(&node.Nodes[i])...)
+		nodes = append(nodes, flatten(node.Nodes[i])...)
 	}
 	return nodes
 }

index/blocks.go

@@ -6,13 +6,18 @@ import (
 
 type Blocks []uint64
 
+// NewBlocks returns a 32-length Block.
+func NewBlocks() Blocks {
+	return make(Blocks, 32)
+}
+
 func (a Blocks) bitcount() uint64 {
 	fmt.Println("bitcount:", a)
 	return popcntSlice(a)
 }
 
 func (a Blocks) union(other Blocks) Blocks {
-	ret := make(Blocks, 32)
+	ret := NewBlocks()
 	for i, _ := range a {
 		ret[i] = a[i] | other[i]
 	}
@@ -20,7 +25,7 @@ func (a Blocks) union(other Blocks) Blocks {
 }
 
 func (a Blocks) invert() Blocks {
-	other := make(Blocks, 32)
+	other := NewBlocks()
 	for i, _ := range a {
 		other[i] = ^a[i]
 	}
@@ -28,7 +33,7 @@ func (a Blocks) invert() Blocks {
 }
 
 func (a Blocks) copy() Blocks {
-	other := make(Blocks, 32)
+	other := NewBlocks()
 	for i, _ := range a {
 		other[i] = a[i]
 	}
@@ -41,7 +46,7 @@ func (a Blocks) andcount(other Blocks) uint64 {
 }
 
 func (a Blocks) intersection(other Blocks) Blocks {
-	ret := make(Blocks, 32)
+	ret := NewBlocks()
 	for i, _ := range a {
 		ret[i] = a[i] & other[i]
 	}
@@ -49,7 +54,7 @@ func (a Blocks) intersection(other Blocks) Blocks {
 }
 
 func (a Blocks) difference(other Blocks) Blocks {
-	ret := make(Blocks, 32)
+	ret := NewBlocks()
 	for i, _ := range a {
 		ret[i] = a[i] &^ other[i]
 	}

index/brand.go

@@ -360,7 +360,7 @@ func (b *Brand) getFileName() string {
 
 func (b *Brand) Persist() error {
 	log.Info("Brand Persist:", b.getFileName())
-	b.storage.FlushBatch()
+	b.storage.Flush()
 	asize := len(b.bitmap_cache)
 
 	if asize == 0 {

index/brand_test.go

@@ -1,16 +1,18 @@
-package index
+package index_test
 
 import (
 	"math/rand"
 	"testing"
 
+	"github.com/umbel/pilosa/index"
+	"github.com/umbel/pilosa/index/storage/mem"
 	"github.com/umbel/pilosa/util"
 )
 
 var (
 	size      int
-	membrand  *Brand
-	cassbrand *Brand
+	membrand  *index.Brand
+	cassbrand *index.Brand
 )
 
 func init() {
@@ -19,7 +21,7 @@ func init() {
 	util.SetupStatsd()
 	// SetupCassandra()
 
-	membrand = NewBrand("db", "frame", 0, NewMemoryStorage(), size, size, 0)
+	membrand = index.NewBrand("db", "frame", 0, mem.NewStorage(), size, size, 0)
 	for i := uint64(0); i < uint64(size); i++ {
 		membrand.SetBit(i, 0, 1)
 	}
@@ -29,7 +31,7 @@ func init() {
 	// }
 }
 
-func benchmarkBrand(b *testing.B, size int, fill int, brand *Brand) {
+func benchmarkBrand(b *testing.B, size int, fill int, brand *index.Brand) {
 	println(b.N)
 	for i := 0; i < b.N; i++ {
 		bid := rand.Int() % size

index/fragment_container.go

@@ -369,25 +369,22 @@ type Fragment struct {
 	queue_size  int
 }
 
-func getStorage(db string, slice int, frame string, fid util.SUUID) Storage {
-	switch Backend {
-	default:
-		return NewMemoryStorage()
-	case "leveldb":
-		full_dir := fmt.Sprintf("%s/%s/%d/%s/%s", LevelDBPath, db, slice, frame, util.SUUID_to_Hex(fid))
-		return NewLevelDBStorage(full_dir)
-	case "cassandra":
-		return NewCassStorage()
-	}
-}
-
 func NewFragment(frag_id util.SUUID, db string, slice int, frame string) *Fragment {
 	var impl Pilosa
 	log.Warn(fmt.Sprintf("XXXXXXXXXXXXXXXXXXXXXXXXXXX(%s)", frame))
+
+	storage := NewStorage(Backend, StorageOptions{
+		DB:          db,
+		Slice:       slice,
+		Frame:       frame,
+		FragmentID:  frag_id,
+		LevelDBPath: LevelDBPath,
+	})
+
 	if strings.HasSuffix(frame, ".n") {
-		impl = NewBrand(db, frame, slice, getStorage(db, slice, frame, frag_id), 50000, 45000, 100)
+		impl = NewBrand(db, frame, slice, storage, 50000, 45000, 100)
 	} else {
-		impl = NewGeneral(db, frame, slice, getStorage(db, slice, frame, frag_id))
+		impl = NewGeneral(db, frame, slice, storage)
 	}
 
 	f := new(Fragment)

index/fragment_container_test.go

@@ -4,6 +4,7 @@ import (
 	"testing"
 
 	"github.com/umbel/pilosa/index"
+	_ "github.com/umbel/pilosa/index/storage"
 	"github.com/umbel/pilosa/util"
 )
 

index/storage.go

@@ -1,14 +1,53 @@
 package index
 
+import (
+	"github.com/umbel/pilosa/util"
+)
+
+// Storage represents
 type Storage interface {
-	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)
-	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()
-	Close()
+	Open() error
+	Close() error
+	Flush()
+
+	Fetch(bitmap_id uint64, db, frame string, slice int) (*Bitmap, uint64)
+	Store(id uint64, db, frame string, slice int, filter uint64, bitmap *Bitmap) error
+
+	StoreBlock(id uint64, db, frame string, slice int, filter uint64, chunk uint64, block_index int32, block uint64) error
+	RemoveBlock(id uint64, db, frame string, slice int, chunk uint64, block_index int32)
+
+	StoreBit(bid uint64, db, frame string, slice int, filter uint64, chunk uint64, block_index int32, block, count uint64)
+	RemoveBit(id uint64, db, frame string, slice int, filter uint64, chunk uint64, block_index int32, count uint64)
+}
+
+var storageFns = make(map[string]NewStorageFunc)
+
+// NewStorageFunc represents a function that instantiates a new storage engine.
+type NewStorageFunc func(opt StorageOptions) Storage
+
+// RegisterStorage registers a storage engine.
+func RegisterStorage(name string, fn NewStorageFunc) {
+	if storageFns[name] != nil {
+		panic("storage engine already registered: " + name)
+	}
+	storageFns[name] = fn
+}
+
+// NewStorage returns a new Storage instance by name.
+func NewStorage(name string, opt StorageOptions) Storage {
+	fn := storageFns[name]
+	if fn == nil {
+		panic("storage type not registered: " + name)
+	}
+	return fn(opt)
+}
+
+// StorageOptions represents the options passed to the storage engine.
+type StorageOptions struct {
+	DB         string
+	Slice      int
+	Frame      string
+	FragmentID util.SUUID
+
+	LevelDBPath string
 }

index/storage/cassandra/cassandra.go

@@ -0,0 +1,244 @@
+package cassandra
+
+import (
+	"time"
+
+	log "github.com/cihub/seelog"
+	"github.com/gocql/gocql"
+	"github.com/umbel/pilosa/index"
+	"github.com/umbel/pilosa/util"
+)
+
+func init() {
+	index.RegisterStorage("cassandra",
+		func(opt index.StorageOptions) index.Storage {
+			return NewStorage(opt)
+		},
+	)
+}
+
+// CREATE KEYSPACE IF NOT EXISTS pilosa WITH strategy_class = SimpleStrategy AND strategy_options:replication_factor = 1"
+// create keyspace if not exists pilosa with replication = {'class': 'SimpleStrategy', 'replication_factor' : 1} and durable_writes = true;
+// CREATE KEYSPACE pilosa WITH replication = {'class': 'NetworkTopologyStrategy', 'pilpang': '2'}  AND durable_writes = true;
+// CREATE TABLE IF NOT EXISTS bitmap (bitmap_id bigint, db varchar, frame varchar, slice int, filter int, chunkkey bigint, blockindex int, block bigint, PRIMARY KEY ((bitmap_id, db, frame, slice), chunkkey, blockindex) )
+
+// DefaultHosts are the default hosts in the cassandra cluster.
+var DefaultHosts = []string{"localhost"}
+
+const (
+	// DefaultKeyspace is the default keyspace used in cassandra.
+	DefaultKeyspace = "pilosa"
+
+	// DefaultFlushInterval is the default maximum time between flushes.
+	DefaultFlushInterval = 5 * time.Second
+
+	// DefaultFlushThreshold is the default maximum number of items to batch.
+	DefaultFlushThreshold = 15
+)
+
+// Storage represents Cassandra-backed storage for bitmaps.
+type Storage struct {
+	session *gocql.Session
+
+	batch     *gocql.Batch
+	batchTime time.Time
+	batchN    int
+
+	FlushInterval  time.Duration
+	FlushThreshold int
+
+	Hosts    []string
+	Keyspace string
+}
+
+// NewStorage returns a new, uninitialized instance of Storage.
+func NewStorage(opt index.StorageOptions) *Storage {
+	return &Storage{
+		batchTime: time.Now(),
+
+		FlushInterval:  DefaultFlushInterval,
+		FlushThreshold: DefaultFlushThreshold,
+
+		Hosts:    DefaultHosts,
+		Keyspace: DefaultKeyspace,
+	}
+}
+
+// Open opens the connection to the cassandra cluster.
+func (s *Storage) Open() error {
+	// Create cluster configuration.
+	config := gocql.NewCluster(s.Hosts...)
+	config.Keyspace = s.Keyspace
+	config.Consistency = gocql.One
+	config.Timeout = 5 * time.Second
+	config.RetryPolicy = &gocql.SimpleRetryPolicy{NumRetries: 10}
+
+	// Connect to cassandra.
+	session, err := config.CreateSession()
+	if err != nil {
+		return err
+	}
+	s.session = session
+
+	return nil
+}
+
+// Close closes the connection to the cassandra cluster.
+func (s *Storage) Close() error {
+	if s.session != nil {
+		s.Flush()
+		s.session.Close()
+	}
+
+	return nil
+}
+
+// Fetch returns a bitmap by ID.
+func (s *Storage) Fetch(bitmapID uint64, db string, frame string, slice int) (*index.Bitmap, uint64) {
+	bm := index.NewBitmap()
+
+	// Start benchmark.
+	start := time.Now()
+
+	// Create iterator over bitmap.
+	itr := s.session.Query("SELECT filter, Chunkkey, BlockIndex, block FROM bitmap WHERE bitmap_id=? AND db=? AND frame=? AND slice=? ",
+		u64toi64(bitmapID), db, frame, slice).Iter()
+
+	// Iterate over chunks and materialize bitmap object.
+	var chunk *index.Chunk
+	var chunkKey, block, count int64
+	var blockIndex uint32
+	var filter int
+	lastKey := int64(-1)
+
+	for itr.Scan(&filter, &chunkKey, &blockIndex, &block) {
+		if chunkKey != int64(-1) {
+			if chunkKey != lastKey {
+				chunk = &index.Chunk{uint64(chunkKey), index.NewBlocks()}
+				bm.AddChunk(chunk)
+			}
+			chunk.Value[uint8(blockIndex)] = uint64(block)
+		} else {
+			count = block
+		}
+		lastKey = chunkKey
+	}
+
+	util.SendTimer("cassandra_storage_Fetch", time.Since(start).Nanoseconds())
+	util.SendInc("cassandra_storage_Read")
+
+	// Set total bits set.
+	bm.SetCount(uint64(count))
+	return bm, uint64(filter)
+}
+
+// beginBatch starts a batch if one is not already in progress.
+func (s *Storage) beginBatch() {
+	if s.batch == nil {
+		s.batch = gocql.NewBatch(gocql.UnloggedBatch)
+	}
+	s.batchN++
+}
+
+// endBatch flushes a batch if one is in progress.
+func (s *Storage) endBatch() {
+	if s.batch == nil {
+		return
+	}
+
+	start := time.Now()
+	s.Flush()
+	util.SendTimer("cassandra_storage_EndBatch", time.Since(start).Nanoseconds())
+	util.SendInc("cassandra_storage_Write")
+}
+
+// Flush flushes the current batch to storage.
+func (s *Storage) Flush() {
+	start := time.Now()
+
+	// If batch exists then flush it.
+	if s.batch != nil {
+		if err := s.session.ExecuteBatch(s.batch); err != nil {
+			log.Warn("Batch ERROR: ", err)
+		}
+	}
+
+	// Clear batch and threshold time and count.
+	s.batch = nil
+	s.batchTime = time.Now()
+	s.batchN = 0
+
+	util.SendTimer("cassandra_storage_FlushBatch", time.Since(start).Nanoseconds())
+}
+
+// Store saves a bitmap to storage.
+func (s *Storage) Store(id uint64, db string, frame string, slice int, filter uint64, bm *index.Bitmap) error {
+	s.beginBatch()
+
+	for i := bm.ChunkIterator(); !i.Limit(); i = i.Next() {
+		var chunk = i.Item()
+		for idx, block := range chunk.Value {
+			if block != 0 {
+				s.StoreBlock(id, db, frame, slice, filter, chunk.Key, int32(idx), block)
+			}
+		}
+	}
+
+	s.StoreBlock(id, db, frame, slice, filter, index.CounterMask, 0, bm.BitCount())
+	s.endBatch()
+	return nil
+}
+
+// StoreBlock saves a block to storage.
+func (s *Storage) StoreBlock(bid uint64, db string, frame string, slice int, filter uint64, bchunk uint64, blockIndex int32, bblock uint64) error {
+	id := u64toi64(bid) // these functions ignore overflow
+	block := u64toi64(bblock)
+	chunk := u64toi64(bchunk)
+
+	if s.batch == nil {
+		s.beginBatch()
+	}
+
+	start := time.Now()
+	s.batch.Query(`INSERT INTO bitmap ( bitmap_id, db, frame, slice , filter, ChunkKey, BlockIndex, block) VALUES (?,?,?,?,?,?,?,?)  USING timestamp ?;`,
+		id, db, frame, slice, int(filter), chunk, blockIndex, block, start.UnixNano())
+	util.SendTimer("cassandra_storage_StoreBlock", time.Since(start).Nanoseconds())
+
+	return nil
+}
+
+// RemoveBlock deletes a block from storage.
+func (s *Storage) RemoveBlock(bid uint64, db string, frame string, slice int, bchunk uint64, blockIndex int32) {
+	log.Trace("RemoveBlock", bid, db, frame, slice, bchunk, blockIndex)
+
+	id := u64toi64(bid) //these fucntions ignore overflow
+	chunk := u64toi64(bchunk)
+
+	if s.batch == nil {
+		s.beginBatch()
+	}
+
+	startTime := time.Now()
+
+	s.batch.Query(`DELETE FROM bitmap USING TIMESTAMP ? WHERE bitmap_id=? AND db=? AND frame=? AND slice=? AND chunkkey=? AND blockindex=?`,
+		startTime.UnixNano(), id, db, frame, slice, chunk, blockIndex)
+
+	util.SendTimer("cassandra_storage_DeleteBlock", time.Since(startTime).Nanoseconds())
+}
+
+func (s *Storage) StoreBit(bid uint64, db string, frame string, slice int, filter uint64, chunk uint64, blockIndex int32, val, count uint64) {
+	s.beginBatch()
+	s.StoreBlock(bid, db, frame, slice, filter, chunk, blockIndex, val)
+	s.StoreBlock(bid, db, frame, slice, filter, index.CounterMask, 0, count)
+	s.endBatch()
+}
+
+func (s *Storage) RemoveBit(id uint64, db string, frame string, slice int, filter uint64, chunk uint64, blockIndex int32, count uint64) {
+	log.Trace("RemoveBit", id, db, frame, slice, chunk, blockIndex)
+	s.beginBatch()
+	s.RemoveBlock(id, db, frame, slice, chunk, blockIndex)
+	s.StoreBlock(id, db, frame, slice, filter, index.CounterMask, 0, count)
+	s.endBatch()
+}
+
+func u64toi64(v uint64) int64 { return int64(v) }

index/storage/cassandra/cassandra_test.go

@@ -0,0 +1 @@
+package cassandra_test

index/storage/leveldb/leveldb.go

@@ -0,0 +1,222 @@
+package leveldb
+
+import (
+	"bytes"
+	"encoding/binary"
+	"path/filepath"
+	"strconv"
+	"time"
+
+	"github.com/syndtr/goleveldb/leveldb"
+	. "github.com/syndtr/goleveldb/leveldb/util"
+	"github.com/umbel/pilosa/index"
+	"github.com/umbel/pilosa/util"
+)
+
+func init() {
+	index.RegisterStorage("leveldb",
+		func(opt index.StorageOptions) index.Storage {
+			return NewStorage(opt)
+		},
+	)
+}
+
+//go get github.com/syndtr/goleveldb/leveldb
+
+// Storage represents a LevelDB-backed storage engine.
+type Storage struct {
+	path string
+	db   *leveldb.DB
+
+	batch     *leveldb.Batch
+	batchTime time.Time
+	batchN    int
+}
+
+// NewStorage returns a new instance of Storage.
+func NewStorage(opt index.StorageOptions) *Storage {
+	path := filepath.Join(
+		opt.LevelDBPath,
+		opt.DB,
+		strconv.Itoa(opt.Slice),
+		opt.Frame,
+		util.SUUID_to_Hex(opt.FragmentID),
+	)
+
+	return &Storage{path: path}
+}
+
+// Path returns the path the storage was initialized with.
+func (s *Storage) Path() string { return s.path }
+
+// Open opens and initializes the storage.
+func (s *Storage) Open() error {
+	db, err := leveldb.OpenFile(s.path, nil)
+	if err != nil {
+		return err
+	}
+	s.db = db
+	s.batchTime = time.Now().Add(-time.Hour)
+
+	return nil
+}
+
+// Close flushes and closes the storage.
+func (s *Storage) Close() error {
+	s.Flush()
+	return s.db.Close()
+}
+
+func (s *Storage) Fetch(bitmapID uint64, db string, frame string, slice int) (*index.Bitmap, uint64) {
+	bm := index.NewBitmap()
+
+	// Begin benchmark.
+	start := time.Now()
+
+	// Create an iterator on the database.
+	iter := s.db.NewIterator(&Range{
+		Start: marshalKey(bitmapID, 0, 0),
+		Limit: marshalKey(bitmapID+1, 0, 0),
+	}, nil)
+	defer iter.Release()
+
+	// Iterate over blocks in database and create bitmap.
+	var chunk *index.Chunk
+	var filter, block, count uint64
+	lastKey := uint64(index.CounterMask)
+
+	for iter.Next() {
+		_, key, idx := unmarshalKey(iter.Key())
+		block, filter = unmarshalValue(iter.Value())
+		if key != index.CounterMask {
+			if key != lastKey {
+				chunk = &index.Chunk{key, index.NewBlocks()}
+				bm.AddChunk(chunk)
+			}
+			chunk.Value[idx] = block
+		} else {
+			count = block
+		}
+
+		lastKey = key
+	}
+
+	util.SendTimer("leveldb_storage_Fetch", time.Since(start).Nanoseconds())
+
+	// Set bit count on bitmap.
+	bm.SetCount(uint64(count))
+
+	return bm, uint64(filter)
+}
+
+// beginBatch starts a new batch if one is not already started.
+func (s *Storage) beginBatch() {
+	if s.batch == nil {
+		s.batch = &leveldb.Batch{}
+	}
+	s.batchN++
+}
+
+func (s *Storage) endBatch() {
+	if s.batch == nil {
+		return
+	}
+
+	start := time.Now()
+	if time.Since(s.batchTime) > 15*time.Second || s.batchN > 300 {
+		s.Flush()
+	}
+	util.SendTimer("leveldb_storage_EndBatch", time.Since(start).Nanoseconds())
+}
+
+func (s *Storage) Flush() {
+	start := time.Now()
+
+	// Flush the batch if one exists.
+	if s.batch != nil {
+		s.db.Write(s.batch, nil)
+	}
+
+	// Clear batch and reset timer and count.
+	s.batch = nil
+	s.batchTime = time.Now()
+	s.batchN = 0
+
+	util.SendTimer("leveldb_storage_FlushBatch", time.Since(start).Nanoseconds())
+}
+
+// Store saves a bitmap to storage.
+func (s *Storage) Store(bitmapID uint64, db string, frame string, slice int, filter uint64, bm *index.Bitmap) error {
+	s.beginBatch()
+
+	for itr := bm.ChunkIterator(); !itr.Limit(); itr = itr.Next() {
+		for idx, block := range itr.Item().Value {
+			if block != 0 {
+				s.StoreBlock(bitmapID, db, frame, slice, filter, itr.Item().Key, int32(idx), block)
+			}
+		}
+	}
+
+	s.StoreBlock(bitmapID, db, frame, slice, filter, index.CounterMask, 0, bm.BitCount())
+
+	s.endBatch()
+	return nil
+}
+
+// StoreBlock saves a block to storage.
+func (s *Storage) StoreBlock(bitmapID uint64, db string, frame string, slice int, filter uint64, chunk uint64, index int32, block uint64) error {
+	start := time.Now()
+	s.batch.Put(marshalKey(bitmapID, chunk, uint8(index)), marshalValue(block, filter))
+	util.SendTimer("leveldb_storage_StoreBlock", time.Since(start).Nanoseconds())
+	return nil
+}
+
+// RemoveBlock deletes a block from storage. This is a no-op in the LevelDB store.
+func (s *Storage) RemoveBlock(bitmapID uint64, db string, frame string, slice int, chunk uint64, blockIndex int32) {
+}
+
+// StoreBit sets a bit in storage.
+func (s *Storage) StoreBit(bitmapID uint64, db string, frame string, slice int, filter uint64, bchunk uint64, blockIndex int32, bblock, count uint64) {
+	s.beginBatch()
+	s.StoreBlock(bitmapID, db, frame, slice, filter, bchunk, blockIndex, bblock)
+	s.StoreBlock(bitmapID, db, frame, slice, filter, index.CounterMask, 0, count)
+	s.endBatch()
+}
+
+// RemoveBlock unsets a bit in storage. This is a no-op in the LevelDB store.
+func (s *Storage) RemoveBit(bitmapID uint64, db string, frame string, slice int, filter uint64, bchunk uint64, blockIndex int32, count uint64) {
+}
+
+// marshalKey encodes the id, chunk key, & block index to a byte slice.
+func marshalKey(id, key uint64, index uint8) []byte {
+	buf := new(bytes.Buffer)
+	binary.Write(buf, binary.LittleEndian, id)
+	binary.Write(buf, binary.LittleEndian, key)
+	binary.Write(buf, binary.LittleEndian, index)
+	return buf.Bytes()
+}
+
+// marshalValue encodes the block number and filter to a byte slice.
+func marshalValue(block, filter uint64) []byte {
+	buf := new(bytes.Buffer)
+	binary.Write(buf, binary.LittleEndian, block)
+	binary.Write(buf, binary.LittleEndian, filter)
+	return buf.Bytes()
+}
+
+// unmarshalKey decodes the id, chunk key, & block index from a byte slice.
+func unmarshalKey(v []byte) (id, key uint64, index uint8) {
+	buf := bytes.NewReader(v)
+	binary.Read(buf, binary.LittleEndian, &id)
+	binary.Read(buf, binary.LittleEndian, &key)
+	binary.Read(buf, binary.LittleEndian, &index)
+	return
+}
+
+// unmarshalValue decodes the block number and filter from a byte slice.
+func unmarshalValue(v []byte) (block, filter uint64) {
+	buf := bytes.NewReader(v)
+	binary.Read(buf, binary.LittleEndian, &block)
+	binary.Read(buf, binary.LittleEndian, &filter)
+	return
+}

index/storage/leveldb/leveldb_test.go

@@ -0,0 +1,55 @@
+package leveldb_test
+
+import (
+	"io/ioutil"
+	"os"
+	"testing"
+
+	"github.com/umbel/pilosa/index"
+	"github.com/umbel/pilosa/index/storage/leveldb"
+	"github.com/umbel/pilosa/util"
+)
+
+func init() {
+	util.SetupStatsd()
+}
+
+func TestStorage_Fetch(t *testing.T) {
+	s := MustOpenStorage()
+	defer s.Close()
+}
+
+// Storage represents a test wrapper for leveldb.Storage.
+type Storage struct {
+	*leveldb.Storage
+}
+
+// NewStorage returns a new instance of Storage with a temporary path.
+func NewStorage() *Storage {
+	// Create temporary path.
+	f, err := ioutil.TempFile("", "pilosa-leveldb-")
+	if err != nil {
+		panic(err)
+	}
+	f.Close()
+	os.Remove(f.Name())
+
+	return &Storage{leveldb.NewStorage(index.StorageOptions{
+		LevelDBPath: f.Name(),
+	})}
+}
+
+// MustOpenStorage returns a new, opened instance of Storage.
+func MustOpenStorage() *Storage {
+	s := NewStorage()
+	if err := s.Open(); err != nil {
+		panic(err)
+	}
+	return s
+}
+
+// Close closes the storage and removes the underlying data file.
+func (s *Storage) Close() error {
+	defer os.Remove(s.Path())
+	return s.Storage.Close()
+}

index/storage/mem/mem.go

@@ -0,0 +1,79 @@
+package mem
+
+import (
+	"fmt"
+
+	"github.com/umbel/pilosa/index"
+)
+
+func init() {
+	index.RegisterStorage("memory",
+		func(opt index.StorageOptions) index.Storage {
+			return NewStorage()
+		},
+	)
+}
+
+// Storage represents in-memory bitmap storage.
+type Storage struct {
+	db map[string]*index.Bitmap
+}
+
+// NewStorage returns a new instance of Storage.
+func NewStorage() *Storage {
+	return &Storage{
+		db: make(map[string]*index.Bitmap),
+	}
+}
+
+// Open initializes the storage.
+func (c *Storage) Open() error { return nil }
+
+// Close shuts down the storage.
+func (c *Storage) Close() error { return nil }
+
+// FlushBatch flushes the batch to disk. This is a no-op for in-memory storage.
+func (c *Storage) Flush() {}
+
+// Fetch retrieves a bitmap by id.
+func (c *Storage) Fetch(id uint64, db, frame string, slice int) (*index.Bitmap, uint64) {
+	key := fmt.Sprintf("%d:%s:%s:%d", id, db, frame, slice)
+
+	// Find bitmap by key.
+	b, ok := c.db[key]
+	if ok {
+		return b, 0
+	}
+
+	// If the bitmap doesn't exist then create a new one.
+	b = index.NewBitmap()
+	c.db[key] = b
+	return b, 0
+}
+
+// Store saves a bitmap to storage.
+// This is a no-op for in-memory storage because changes are stored in the cache.
+func (c *Storage) Store(id uint64, db, frame string, slice int, filter uint64, b *index.Bitmap) error {
+	return nil
+}
+
+// StoreBlock saves a block to storage.
+// This is a no-op for in-memory storage because changes are stored in the cache.
+func (c *Storage) StoreBlock(id uint64, db, frame string, slice int, filter uint64, chunk_key uint64, block_index int32, block uint64) error {
+	return nil
+}
+
+// RemoveBlock deletes a block from bitmap.
+// This is a no-op for in-memory storage because changes are stored in the cache.
+func (self *Storage) RemoveBlock(id uint64, db string, frame string, slice int, chunk uint64, block_index int32) {
+}
+
+// StoreBit sets a bit in a bitmap.
+// This is a no-op for in-memory storage because changes are stored in the cache.
+func (self *Storage) StoreBit(id uint64, db string, frame string, slice int, filter uint64, bchunk uint64, block_index int32, bblock, count uint64) {
+}
+
+// RemoveBit unsets a bit in a bitmap.
+// This is a no-op for in-memory storage because changes are stored in the cache.
+func (self *Storage) RemoveBit(id uint64, db string, frame string, slice int, filter uint64, chunk uint64, block_index int32, count uint64) {
+}

index/storage/mem/mem_test.go

@@ -0,0 +1,30 @@
+package mem_test
+
+import (
+	"testing"
+
+	"github.com/umbel/pilosa/index/storage/mem"
+)
+
+// Ensure a bitmap can be retrieved from storage.
+func TestStorage_Fetch(t *testing.T) {
+	s := mem.NewStorage()
+
+	// Retrieving a non-existent bitmap should create a new one.
+	b, _ := s.Fetch(1, "d", "f", 0)
+	if b == nil {
+		t.Fatal("expected bitmap")
+	}
+
+	// Retrieve the bitmap again.
+	other, _ := s.Fetch(1, "d", "f", 0)
+	if b != other {
+		t.Fatal("expected same bitmap")
+	}
+
+	// Retrieving a different bitmap should return a different reference.
+	b2, _ := s.Fetch(2, "d", "f", 0)
+	if b == b2 {
+		t.Fatal("expected new bitmap")
+	}
+}

index/storage/storage.go

@@ -0,0 +1,7 @@
+package storage
+
+import (
+	_ "github.com/umbel/pilosa/index/storage/cassandra"
+	_ "github.com/umbel/pilosa/index/storage/leveldb"
+	_ "github.com/umbel/pilosa/index/storage/mem"
+)

index/storage_cass.go

@@ -1,224 +0,0 @@
-package index
-
-// #cgo  CFLAGS:-mpopcnt
-
-import (
-	"time"
-
-	log "github.com/cihub/seelog"
-	"github.com/gocql/gocql"
-	"github.com/umbel/pilosa/util"
-)
-
-// DefaultStorageHosts are the hosts that stores the data.
-var DefaultStorageHosts = [...]string{"localhost"}
-
-// DefaultStorageKeyspace is the default keyspace used for storage.
-const DefaultStorageKeyspace = "pilosa"
-
-const DefaultCassandraTimeWindow = 5 * time.Second
-const DefaultCassandraMaxSizeBatch = 15
-
-var StorageHosts = DefaultStorageHosts[:]
-var StorageKeyspace = DefaultStorageKeyspace
-var CassandraTimeWindow = DefaultCassandraTimeWindow
-var CassandraMaxSizeBatch = DefaultCassandraMaxSizeBatch
-
-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
-	cass_flush_size       int
-}
-
-var cluster *gocql.ClusterConfig
-var session *gocql.Session
-
-func SetupCassandra() {
-	var err error
-	hosts := StorageHosts
-	keyspace := StorageKeyspace
-	cluster = gocql.NewCluster(hosts...)
-	cluster.Keyspace = keyspace
-	cluster.Consistency = gocql.One
-	cluster.Timeout = 5 * time.Second
-	cluster.RetryPolicy = &gocql.SimpleRetryPolicy{NumRetries: 10}
-	session, err = cluster.CreateSession()
-	if err != nil {
-		log.Warn(err)
-	}
-}
-
-func BuildSchema() {
-	/*
-		"CREATE KEYSPACE IF NOT EXISTS pilosa WITH strategy_class = SimpleStrategy AND strategy_options:replication_factor = 1"
-		create keyspace if not exists pilosa with replication = {'class': 'SimpleStrategy', 'replication_factor' : 1} and durable_writes = true;
-		CREATE KEYSPACE pilosa WITH replication = {'class': 'NetworkTopologyStrategy', 'pilpang': '2'}  AND durable_writes = true;
-
-		CREATE TABLE IF NOT EXISTS bitmap (bitmap_id bigint, db varchar, frame varchar, slice int, filter int, chunkkey bigint, blockindex int, block bigint, PRIMARY KEY ((bitmap_id, db, frame, slice), chunkkey, blockindex) )
-		"
-	*/
-
-}
-
-func NewCassStorage() Storage {
-	obj := new(CassandraStorage)
-	// cluster.CQLVersion = "3.0.0"
-	/*session, err := cluster.CreateSession()
-	if err != nil {
-		log.Warn(err)
-	}
-	*/
-
-	obj.db = session
-	obj.stmt = `INSERT INTO bitmap ( bitmap_id, db, frame, slice , filter, ChunkKey, BlockIndex, block) VALUES (?,?,?,?,?,?,?,?)  USING timestamp ?;`
-	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(CassandraTimeWindow.Seconds())
-	obj.cass_flush_size = CassandraMaxSizeBatch
-	return obj
-}
-
-func (c *CassandraStorage) Close() {
-}
-
-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 (
-		chunk            *Chunk
-		chunk_key, block int64
-		block_index      uint32
-		s8               uint8
-		filter           int
-	)
-
-	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)
-
-	for iter.Scan(&filter, &chunk_key, &block_index, &block) {
-		s8 = uint8(block_index)
-		if chunk_key != marker {
-			if chunk_key != last_key {
-				chunk = &Chunk{uint64(chunk_key), make(Blocks, 32)}
-				bitmap.AddChunk(chunk)
-			}
-			chunk.Value[s8] = uint64(block)
-
-		} else {
-			count = block
-		}
-		last_key = chunk_key
-
-	}
-	delta := time.Since(start)
-	util.SendTimer("cassandra_storage_Fetch", delta.Nanoseconds())
-	util.SendInc("cassandra_storage_Read")
-	bitmap.SetCount(uint64(count))
-	return bitmap, uint64(filter)
-}
-func (self *CassandraStorage) BeginBatch() {
-	if self.batch == nil {
-		self.batch = gocql.NewBatch(gocql.UnloggedBatch)
-	}
-	self.batch_counter++
-}
-func (self *CassandraStorage) runBatch(batch *gocql.Batch) {
-	if batch != nil {
-		err := self.db.ExecuteBatch(batch)
-		if err != nil {
-			log.Warn("Batch ERROR: ", err)
-		}
-	}
-}
-func (self *CassandraStorage) FlushBatch() {
-	start := time.Now()
-	self.runBatch(self.batch) //maybe this is crazy but i'll give it a whirl
-	self.batch = nil
-	self.batch_time = time.Now()
-	self.batch_counter = 0
-	delta := time.Since(start)
-	util.SendTimer("cassandra_storage_FlushBatch", delta.Nanoseconds())
-}
-func (self *CassandraStorage) EndBatch() {
-	start := time.Now()
-	if self.batch != nil {
-		self.FlushBatch()
-	} else {
-		log.Warn("NIL BATCH")
-	}
-	delta := time.Since(start)
-	util.SendTimer("cassandra_storage_EndBatch", delta.Nanoseconds())
-	util.SendInc("cassandra_storage_Write")
-
-}
-
-func (self *CassandraStorage) Store(id uint64, db string, frame string, slice int, filter uint64, bitmap *Bitmap) error {
-	self.BeginBatch()
-	for i := bitmap.ChunkIterator(); !i.Limit(); i = i.Next() {
-		var chunk = i.Item()
-		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 := bitmap.BitCount()
-	self.StoreBlock(id, db, frame, slice, filter, CounterMask, 0, cnt)
-	self.EndBatch()
-	return nil
-}
-
-func (self *CassandraStorage) StoreBlock(bid uint64, db string, frame string, slice int, filter uint64, bchunk uint64, block_index int32, bblock uint64) error {
-	id := util.Uint64ToInt64(bid) //these fucntions ignore overflow
-	block := util.Uint64ToInt64(bblock)
-	chunk := util.Uint64ToInt64(bchunk)
-
-	if self.batch == nil {
-		self.BeginBatch()
-	}
-	start := time.Now()
-	self.batch.Query(self.stmt, id, db, frame, slice, int(filter), chunk, block_index, block, start.UnixNano())
-	delta := time.Since(start)
-	util.SendTimer("cassandra_storage_StoreBlock", delta.Nanoseconds())
-	return nil
-}
-
-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) 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()
-	}
-	start := time.Now()
-
-	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())
-}

index/storage_leveldb.go

@@ -1,188 +0,0 @@
-package index
-
-// #cgo  CFLAGS:-mpopcnt
-
-import (
-	"bytes"
-	"encoding/binary"
-	"log"
-	"time"
-
-	"github.com/syndtr/goleveldb/leveldb"
-	. "github.com/syndtr/goleveldb/leveldb/util"
-	"github.com/umbel/pilosa/util"
-)
-
-type LevelDBStorage struct {
-	db            *leveldb.DB
-	batch         *leveldb.Batch
-	batch_time    time.Time
-	batch_counter int
-}
-
-//go get github.com/syndtr/goleveldb/leveldb
-func NewLevelDBStorage(file_path string) Storage {
-	obj := new(LevelDBStorage)
-	db, _ := leveldb.OpenFile(file_path, nil)
-	obj.db = db
-	obj.batch = nil
-	obj.batch_counter = 0
-	obj.batch_time = time.Now().Add(-time.Hour)
-	return obj
-}
-func encodeKey(id, chunk_key uint64, block_index uint8) []byte {
-	buf := new(bytes.Buffer)
-	binary.Write(buf, binary.LittleEndian, id)
-	binary.Write(buf, binary.LittleEndian, chunk_key)
-	binary.Write(buf, binary.LittleEndian, block_index)
-	return buf.Bytes()
-}
-
-func encodeValue(block, filter uint64) []byte {
-	buf := new(bytes.Buffer)
-	binary.Write(buf, binary.LittleEndian, block)
-	binary.Write(buf, binary.LittleEndian, filter)
-	return buf.Bytes()
-}
-
-func decodeKey(key []byte) (uint64, uint64, uint8) {
-	var (
-		id, chunk_key uint64
-		block_index   uint8
-	)
-	buf := bytes.NewReader(key)
-	binary.Read(buf, binary.LittleEndian, &id)
-	binary.Read(buf, binary.LittleEndian, &chunk_key)
-	binary.Read(buf, binary.LittleEndian, &block_index)
-	return id, chunk_key, block_index
-}
-
-func decodeValue(value []byte) (uint64, uint64) {
-	var (
-		block, filter uint64
-	)
-	buf := bytes.NewReader(value)
-	binary.Read(buf, binary.LittleEndian, &block)
-	binary.Read(buf, binary.LittleEndian, &filter)
-	return block, filter
-}
-
-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 := 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
-	for iter.Next() {
-		_, chunk_key, block_index := decodeKey(iter.Key())
-		block, filter = decodeValue(iter.Value())
-		if chunk_key != CounterMask {
-			if chunk_key != last_key {
-				chunk = &Chunk{chunk_key, make(Blocks, 32)}
-				bitmap.AddChunk(chunk)
-			}
-			chunk.Value[block_index] = block
-
-		} else {
-			count = block
-		}
-		last_key = chunk_key
-	}
-	iter.Release()
-
-	delta := time.Since(start)
-	util.SendTimer("leveldb_storage_Fetch", delta.Nanoseconds())
-	bitmap.SetCount(uint64(count))
-	return bitmap, uint64(filter)
-}
-
-func (self *LevelDBStorage) BeginBatch() {
-	if self.batch == nil {
-		self.batch = new(leveldb.Batch)
-	}
-	self.batch_counter++
-}
-func (self *LevelDBStorage) runBatch(batch *leveldb.Batch) {
-	if batch != nil {
-		self.db.Write(batch, nil)
-	}
-}
-func (self *LevelDBStorage) FlushBatch() {
-	start := time.Now()
-	self.runBatch(self.batch) //maybe this is crazy but i'll give it a whirl
-	self.batch = nil
-	self.batch_time = time.Now()
-	self.batch_counter = 0
-	delta := time.Since(start)
-	util.SendTimer("leveldb_storage_FlushBatch", delta.Nanoseconds())
-}
-func (self *LevelDBStorage) EndBatch() {
-	start := time.Now()
-	if self.batch != nil {
-		last := time.Since(self.batch_time)
-		if last*time.Second > 15 {
-			self.FlushBatch()
-		} else if self.batch_counter > 300 {
-			self.FlushBatch()
-		}
-	} else {
-		log.Println("NIL BATCH")
-	}
-	delta := time.Since(start)
-	util.SendTimer("leveldb_storage_EndBatch", delta.Nanoseconds())
-
-}
-
-func (self *LevelDBStorage) Store(id uint64, db string, frame string, slice int, filter uint64, bitmap *Bitmap) error {
-	self.BeginBatch()
-	for i := bitmap.ChunkIterator(); !i.Limit(); i = i.Next() {
-		var chunk = i.Item()
-		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)
-			}
-		}
-	}
-
-	self.StoreBlock(id, db, frame, slice, filter, CounterMask, 0, bitmap.BitCount())
-	self.EndBatch()
-	return nil
-}
-
-func (self *LevelDBStorage) StoreBlock(id uint64, db string, frame string, slice int, filter uint64, chunk uint64, block_index int32, block uint64) error {
-	if self.batch == nil {
-		panic("NIL BATCH")
-	}
-	start := time.Now()
-	self.batch.Put(encodeKey(id, chunk, uint8(block_index)), encodeValue(block, filter))
-	delta := time.Since(start)
-	util.SendTimer("leveldb_storage_StoreBlock", delta.Nanoseconds())
-	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() {
-	self.FlushBatch()
-	self.db.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.EndBatch()
-
-}

index/storage_mem.go

@@ -1,53 +0,0 @@
-package index
-
-// #cgo  CFLAGS:-mpopcnt
-
-import "fmt"
-
-type MemoryStorage struct {
-	db map[string]*Bitmap
-}
-
-func NewMemoryStorage() Storage {
-	obj := new(MemoryStorage)
-	obj.db = make(map[string]*Bitmap)
-
-	return obj
-}
-
-func (c *MemoryStorage) BeginBatch() {}
-
-func (c *MemoryStorage) Close() {}
-
-func (c *MemoryStorage) EndBatch() {}
-
-func (c *MemoryStorage) FlushBatch() {}
-
-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 = NewBitmap()
-		c.db[key] = bitmap
-	}
-	return bitmap, 0
-}
-
-func (c *MemoryStorage) Store(bitmap_id uint64, db string, frame string, slice int, filter uint64, bitmap *Bitmap) error {
-	//only use the cache and throw away everything
-	return nil
-}
-
-func (c *MemoryStorage) StoreBlock(bitmap_id uint64, db string, frame string, slice int, filter uint64, chunk_key uint64, block_index int32, block uint64) error {
-	//only use the cache and throw away everything
-	return nil
-}
-
-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) 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) {
-}

index/storage_test.go

@@ -56,7 +56,7 @@ func TestStorage(t *testing.T) {
 	// 	SetBit(bm, 2)
 	// 	fmt.Println("STORE")
 	// 	storage.Store(int64(bitmap_id), db, frame, slice, uint64(filter), bm.(*Bitmap))
-	// 	//storage.FlushBatch()
+	// 	//storage.Flush()
 	// 	fmt.Println("FETCH")
 	// 	bm2, _ := storage.Fetch(bitmap_id, db, frame, slice)
 	// 	So(BitCount(bm), ShouldEqual, BitCount(bm2))

util/constants.go

@@ -29,9 +29,3 @@ func ByteToInt64(data []byte) int64 {
 	binary.Read(buf, binary.BigEndian, &value)
 	return value
 }
-func Uint64ToInt64(before uint64) int64 {
-	return int64(before)
-}
-func Int64ToUint64(before int64) uint64 {
-	return uint64(before)
-}