scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-08-28 10:54:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #1169 from travisturner/attrstore-interface

Browse source 
put BoltDB behind AttrStore interface
This commit is contained in:
Travis Turner 2018-03-23 16:39:19 -05:00 • committed by GitHub
commit 7dd41399cb
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
15 changed files with 606 additions and 447 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

513
attr.go
View file

@ -16,23 +16,12 @@ package pilosa
import (
"bytes"
"encoding/binary"
"fmt"
"sort"
"sync"
"time"
"github.com/cespare/xxhash"
"github.com/boltdb/bolt"
"github.com/gogo/protobuf/proto"
"github.com/pilosa/pilosa/internal"
)
// AttrBlockSize is the size of attribute blocks for anti-entropy.
const AttrBlockSize = 100
// Attribute data type enum.
const (
AttrTypeString = 1
@ -41,313 +30,111 @@ const (
AttrTypeFloat = 4
)
// AttrCache represents a cache for attributes.
type AttrCache struct {
mu sync.RWMutex
attrs map[uint64]map[string]interface{}
// AttrStore represents an interface for handling row/column attributes.
type AttrStore interface {
Path() string
Open() error
Close() error
Attrs(id uint64) (m map[string]interface{}, err error)
SetAttrs(id uint64, m map[string]interface{}) error
SetBulkAttrs(m map[uint64]map[string]interface{}) error
Blocks() ([]AttrBlock, error)
BlockData(i uint64) (map[uint64]map[string]interface{}, error)
}
// Get returns the cached attributes for a given id.
func (c *AttrCache) Get(id uint64) map[string]interface{} {
c.mu.RLock()
defer c.mu.RUnlock()
attrs := c.attrs[id]
if attrs == nil {
return nil
}
// Make a copy for safety
ret := make(map[string]interface{})
for k, v := range attrs {
ret[k] = v
}
return ret
func init() {
NopAttrStore = &nopAttrStore{}
}
// Set updates the cached attributes for a given id.
func (c *AttrCache) Set(id uint64, attrs map[string]interface{}) {
c.mu.Lock()
defer c.mu.Unlock()
c.attrs[id] = attrs
// NopAttrStore represents an AttrStore that doesn't do anything.
var NopAttrStore AttrStore
func NewNopAttrStore(string) AttrStore {
return &nopAttrStore{}
}
// AttrStore represents a storage layer for attributes.
type AttrStore struct {
mu sync.RWMutex
path string
db *bolt.DB
attrCache *AttrCache
}
// nopAttrStore represents a no-op implementation of the AttrStore interface.
type nopAttrStore struct{}
// NewAttrCache returns a new instance of AttrCache.
func NewAttrCache() *AttrCache {
return &AttrCache{
attrs: make(map[uint64]map[string]interface{}),
}
}
// NewAttrStore returns a new instance of AttrStore.
func NewAttrStore(path string) *AttrStore {
return &AttrStore{
path: path,
attrCache: NewAttrCache(),
}
}
// Path returns path to the store's data file.
func (s *AttrStore) Path() string { return s.path }
// Open opens and initializes the store.
func (s *AttrStore) Open() error {
// Open storage.
db, err := bolt.Open(s.path, 0666, &bolt.Options{Timeout: 1 * time.Second})
if err != nil {
return err
}
s.db = db
// Initialize database.
if err := s.db.Update(func(tx *bolt.Tx) error {
if _, err := tx.CreateBucketIfNotExists([]byte("attrs")); err != nil {
return err
}
return nil
}); err != nil {
return err
}
// Path is a no-op implementation of AttrStore Path method.
func (s *nopAttrStore) Path() string { return "" }
// Open is a no-op implementation of AttrStore Open method.
func (s *nopAttrStore) Open() error {
return nil
}
// Close closes the store.
func (s *AttrStore) Close() error {
if s.db != nil {
s.db.Close()
}
// Close is a no-op implementation of AttrStore Close method.
func (s *nopAttrStore) Close() error {
return nil
}
// Attrs returns a set of attributes by ID.
func (s *AttrStore) Attrs(id uint64) (m map[string]interface{}, err error) {
s.mu.RLock()
defer s.mu.RUnlock()
// Check cache for map.
if m = s.attrCache.Get(id); m != nil {
return m, nil
}
// Find attributes from storage.
if err = s.db.View(func(tx *bolt.Tx) error {
m, err = txAttrs(tx, id)
if err != nil {
return err
}
return nil
}); err != nil {
return nil, err
}
// Add to cache.
s.attrCache.Set(id, m)
return
// Attrs is a no-op implementation of AttrStore Attrs method.
func (s *nopAttrStore) Attrs(id uint64) (m map[string]interface{}, err error) {
return nil, nil
}
// SetAttrs sets attribute values for a given ID.
func (s *AttrStore) SetAttrs(id uint64, m map[string]interface{}) error {
// Ignore empty maps.
if len(m) == 0 {
return nil
}
// Check if the attributes already exist under a read-only lock.
if attr, err := s.Attrs(id); err != nil {
return err
} else if attr != nil && mapContains(attr, m) {
return nil
}
// Obtain write lock.
s.mu.Lock()
defer s.mu.Unlock()
var attr map[string]interface{}
if err := s.db.Update(func(tx *bolt.Tx) error {
tmp, err := txUpdateAttrs(tx, id, m)
if err != nil {
return err
}
attr = tmp
return nil
}); err != nil {
return err
}
// Swap attributes map in cache.
s.attrCache.Set(id, attr)
// SetAttrs is a no-op implementation of AttrStore SetAttrs method.
func (s *nopAttrStore) SetAttrs(id uint64, m map[string]interface{}) error {
return nil
}
// SetBulkAttrs sets attribute values for a set of ids.
func (s *AttrStore) SetBulkAttrs(m map[uint64]map[string]interface{}) error {
s.mu.Lock()
defer s.mu.Unlock()
// SetBulkAttrs is a no-op implementation of AttrStore SetBulkAttrs method.
func (s *nopAttrStore) SetBulkAttrs(m map[uint64]map[string]interface{}) error {
return nil
}
attrs := make(map[uint64]map[string]interface{})
if err := s.db.Update(func(tx *bolt.Tx) error {
// Collect and sort keys.
ids := make([]uint64, 0, len(m))
for id := range m {
ids = append(ids, id)
// Blocks is a no-op implementation of AttrStore Blocks method.
func (s *nopAttrStore) Blocks() ([]AttrBlock, error) {
return nil, nil
}
// BlockData is a no-op implementation of AttrStore BlockData method.
func (s *nopAttrStore) BlockData(i uint64) (map[uint64]map[string]interface{}, error) {
return nil, nil
}
// AttrBlock represents a checksummed block of the attribute store.
type AttrBlock struct {
ID uint64 `json:"id"`
Checksum []byte `json:"checksum"`
}
// AttrBlocks represents a list of blocks.
type AttrBlocks []AttrBlock
// Diff returns a list of block ids that are different or are new in other.
// Block lists must be in sorted order.
func (a AttrBlocks) Diff(other []AttrBlock) []uint64 {
var ids []uint64
for {
// Read next block from each list.
var blk0, blk1 *AttrBlock
if len(a) > 0 {
blk0 = &a[0]
}
if len(other) > 0 {
blk1 = &other[0]
}
sort.Sort(uint64Slice(ids))
// Update attributes for each id.
for _, id := range ids {
attr, err := txUpdateAttrs(tx, id, m[id])
if err != nil {
return err
// Exit if "a" contains no more blocks.
if blk0 == nil {
return ids
}
// Add block ID if it's different or if it's only in "a".
if blk1 == nil || blk0.ID < blk1.ID {
ids = append(ids, blk0.ID)
a = a[1:]
} else if blk1.ID < blk0.ID {
other = other[1:]
} else {
if !bytes.Equal(blk0.Checksum, blk1.Checksum) {
ids = append(ids, blk0.ID)
}
attrs[id] = attr
}
return nil
}); err != nil {
return err
}
// Swap attributes map in cache.
for id, attr := range attrs {
s.attrCache.Set(id, attr)
}
return nil
}
// Blocks returns a list of all blocks in the store.
func (s *AttrStore) Blocks() ([]AttrBlock, error) {
tx, err := s.db.Begin(false)
if err != nil {
return nil, err
}
defer tx.Rollback()
// Wrap cursor to segment by block.
cur := newBlockCursor(tx.Bucket([]byte("attrs")).Cursor(), AttrBlockSize)
// Iterate over each block.
var blocks []AttrBlock
for cur.nextBlock() {
block := AttrBlock{ID: cur.blockID()}
// Compute checksum of every key/value in block.
h := xxhash.New()
for k, v := cur.next(); k != nil; k, v = cur.next() {
h.Write(k)
h.Write(v)
}
block.Checksum = h.Sum(nil)
// Append block.
blocks = append(blocks, block)
}
return blocks, nil
}
// BlockData returns all data for a single block.
func (s *AttrStore) BlockData(i uint64) (map[uint64]map[string]interface{}, error) {
m := make(map[uint64]map[string]interface{})
// Start read-only transaction.
tx, err := s.db.Begin(false)
if err != nil {
return nil, err
}
defer tx.Rollback()
// Move to the start of the block.
min := u64tob(uint64(i) * AttrBlockSize)
max := u64tob(uint64(i+1) * AttrBlockSize)
cur := tx.Bucket([]byte("attrs")).Cursor()
for k, v := cur.Seek(min); k != nil; k, v = cur.Next() {
// Exit if we're past the end of the block.
if bytes.Compare(k, max) != -1 {
break
}
// Decode attribute map and associate with id.
var pb internal.AttrMap
if err := proto.Unmarshal(v, &pb); err != nil {
return nil, err
}
m[btou64(k)] = decodeAttrs(pb.GetAttrs())
}
return m, nil
}
// txAttrs returns a map of attributes for an id.
func txAttrs(tx *bolt.Tx, id uint64) (map[string]interface{}, error) {
v := tx.Bucket([]byte("attrs")).Get(u64tob(id))
if v == nil {
return emptyMap, nil
}
var pb internal.AttrMap
if err := proto.Unmarshal(v, &pb); err != nil {
return nil, err
}
return decodeAttrs(pb.GetAttrs()), nil
}
// txUpdateAttrs updates the attributes for an id.
// Returns the new combined set of attributes for the id.
func txUpdateAttrs(tx *bolt.Tx, id uint64, m map[string]interface{}) (map[string]interface{}, error) {
attr, err := txAttrs(tx, id)
if err != nil {
return nil, err
}
// Create a new map if it is empty so we don't update emptyMap.
if len(attr) == 0 {
attr = make(map[string]interface{}, len(m))
}
// Merge attributes with original values.
// Nil values should delete keys.
for k, v := range m {
if v == nil {
delete(attr, k)
continue
}
switch v := v.(type) {
case int:
attr[k] = int64(v)
case uint:
attr[k] = int64(v)
case uint64:
attr[k] = int64(v)
case string, int64, bool, float64:
attr[k] = v
default:
return nil, fmt.Errorf("invalid attr type: %T", v)
a, other = a[1:], other[1:]
}
}
// Marshal and save new values.
buf, err := proto.Marshal(&internal.AttrMap{Attrs: encodeAttrs(attr)})
if err != nil {
return nil, err
}
if err := tx.Bucket([]byte("attrs")).Put(u64tob(id), buf); err != nil {
return nil, err
}
return attr, nil
}
func encodeAttrs(m map[string]interface{}) []*internal.Attr {
@ -421,136 +208,16 @@ func cloneAttrs(m map[string]interface{}) map[string]interface{} {
return other
}
// u64tob encodes v to big endian encoding.
func u64tob(v uint64) []byte {
b := make([]byte, 8)
binary.BigEndian.PutUint64(b, v)
return b
// EncodeAttrs encodes an attribute map into a byte slice.
func EncodeAttrs(attr map[string]interface{}) ([]byte, error) {
return proto.Marshal(&internal.AttrMap{Attrs: encodeAttrs(attr)})
}
// btou64 decodes b from big endian encoding.
func btou64(b []byte) uint64 { return binary.BigEndian.Uint64(b) }
// emptyMap is a reusable map that contains no keys.
var emptyMap = make(map[string]interface{})
// AttrBlock represents a checksummed block of the attribute store.
type AttrBlock struct {
ID uint64 `json:"id"`
Checksum []byte `json:"checksum"`
}
// AttrBlocks represents a list of blocks.
type AttrBlocks []AttrBlock
// Diff returns a list of block ids that are different or are new in other.
// Block lists must be in sorted order.
func (a AttrBlocks) Diff(other []AttrBlock) []uint64 {
var ids []uint64
for {
// Read next block from each list.
var blk0, blk1 *AttrBlock
if len(a) > 0 {
blk0 = &a[0]
}
if len(other) > 0 {
blk1 = &other[0]
}
// Exit if "a" contains no more blocks.
if blk0 == nil {
return ids
}
// Add block ID if it's different or if it's only in "a".
if blk1 == nil || blk0.ID < blk1.ID {
ids = append(ids, blk0.ID)
a = a[1:]
} else if blk1.ID < blk0.ID {
other = other[1:]
} else {
if !bytes.Equal(blk0.Checksum, blk1.Checksum) {
ids = append(ids, blk0.ID)
}
a, other = a[1:], other[1:]
}
}
}
// blockCursor represents a cursor for iterating over blocks of a bolt bucket.
type blockCursor struct {
cur *bolt.Cursor
base uint64
n uint64
buf struct {
key []byte
value []byte
filled bool
}
}
// newBlockCursor returns a new block cursor that wraps cur using n sized blocks.
func newBlockCursor(c *bolt.Cursor, n int) blockCursor {
cur := blockCursor{
cur: c,
n: uint64(n),
}
cur.buf.key, cur.buf.value = c.First()
cur.buf.filled = true
return cur
}
// blockID returns the current block ID. Only valid after call to nextBlock().
func (cur *blockCursor) blockID() uint64 { return cur.base }
// nextBlock moves the cursor to the next block.
// Returns true if another block exists, otherwise returns false.
func (cur *blockCursor) nextBlock() bool {
if cur.buf.key == nil {
return false
}
cur.base = binary.BigEndian.Uint64(cur.buf.key) / cur.n
return true
}
// next returns the next key/value within the block.
// Returns nils at the end of the block.
func (cur *blockCursor) next() (key, value []byte) {
// Use buffered value, if set.
if cur.buf.filled {
key, value = cur.buf.key, cur.buf.value
cur.buf.filled = false
return key, value
}
// Read next key.
key, value = cur.cur.Next()
// Fill buffer for EOF.
if key == nil {
cur.buf.key, cur.buf.value, cur.buf.filled = key, value, false
return nil, nil
}
// Parse key and buffer if outside of block.
id := binary.BigEndian.Uint64(key)
if id/cur.n > cur.base {
cur.buf.key, cur.buf.value, cur.buf.filled = key, value, true
return nil, nil
}
return key, value
}
// mapContains returns true if all keys & values of subset are in m.
func mapContains(m, subset map[string]interface{}) bool {
for k, v := range subset {
value, ok := m[k]
if !ok || value != v {
return false
}
}
return true
// DecodeAttrs decodes a byte slice into an attribute map.
func DecodeAttrs(v []byte) (map[string]interface{}, error) {
var pb internal.AttrMap
if err := proto.Unmarshal(v, &pb); err != nil {
return nil, err
}
return decodeAttrs(pb.GetAttrs()), nil
}

465
boltdb/attrstore.go Normal file
View file

@ -0,0 +1,465 @@
// Copyright 2017 Pilosa Corp.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package boltdb
import (
"bytes"
"encoding/binary"
"fmt"
"sort"
"sync"
"time"
"github.com/cespare/xxhash"
"github.com/boltdb/bolt"
"github.com/pilosa/pilosa"
)
// AttrBlockSize is the size of attribute blocks for anti-entropy.
const AttrBlockSize = 100
// AttrCache represents a cache for attributes.
type AttrCache struct {
mu sync.RWMutex
attrs map[uint64]map[string]interface{}
}
// Get returns the cached attributes for a given id.
func (c *AttrCache) Get(id uint64) map[string]interface{} {
c.mu.RLock()
defer c.mu.RUnlock()
attrs := c.attrs[id]
if attrs == nil {
return nil
}
// Make a copy for safety
ret := make(map[string]interface{})
for k, v := range attrs {
ret[k] = v
}
return ret
}
// Set updates the cached attributes for a given id.
func (c *AttrCache) Set(id uint64, attrs map[string]interface{}) {
c.mu.Lock()
defer c.mu.Unlock()
c.attrs[id] = attrs
}
// AttrStore represents a storage layer for attributes.
type AttrStore struct {
mu sync.RWMutex
path string
db *bolt.DB
attrCache *AttrCache
}
// NewAttrCache returns a new instance of AttrCache.
func NewAttrCache() *AttrCache {
return &AttrCache{
attrs: make(map[uint64]map[string]interface{}),
}
}
// NewAttrStore returns a new instance of AttrStore.
func NewAttrStore(path string) pilosa.AttrStore {
return &AttrStore{
path: path,
attrCache: NewAttrCache(),
}
}
// Path returns path to the store's data file.
func (s *AttrStore) Path() string { return s.path }
// Open opens and initializes the store.
func (s *AttrStore) Open() error {
// Open storage.
db, err := bolt.Open(s.path, 0666, &bolt.Options{Timeout: 1 * time.Second})
if err != nil {
return err
}
s.db = db
// Initialize database.
if err := s.db.Update(func(tx *bolt.Tx) error {
if _, err := tx.CreateBucketIfNotExists([]byte("attrs")); err != nil {
return err
}
return nil
}); err != nil {
return err
}
return nil
}
// Close closes the store.
func (s *AttrStore) Close() error {
if s.db != nil {
s.db.Close()
}
return nil
}
// Attrs returns a set of attributes by ID.
func (s *AttrStore) Attrs(id uint64) (m map[string]interface{}, err error) {
s.mu.RLock()
defer s.mu.RUnlock()
// Check cache for map.
if m = s.attrCache.Get(id); m != nil {
return m, nil
}
// Find attributes from storage.
if err = s.db.View(func(tx *bolt.Tx) error {
m, err = txAttrs(tx, id)
if err != nil {
return err
}
return nil
}); err != nil {
return nil, err
}
// Add to cache.
s.attrCache.Set(id, m)
return
}
// SetAttrs sets attribute values for a given ID.
func (s *AttrStore) SetAttrs(id uint64, m map[string]interface{}) error {
// Ignore empty maps.
if len(m) == 0 {
return nil
}
// Check if the attributes already exist under a read-only lock.
if attr, err := s.Attrs(id); err != nil {
return err
} else if attr != nil && mapContains(attr, m) {
return nil
}
// Obtain write lock.
s.mu.Lock()
defer s.mu.Unlock()
var attr map[string]interface{}
if err := s.db.Update(func(tx *bolt.Tx) error {
tmp, err := txUpdateAttrs(tx, id, m)
if err != nil {
return err
}
attr = tmp
return nil
}); err != nil {
return err
}
// Swap attributes map in cache.
s.attrCache.Set(id, attr)
return nil
}
// SetBulkAttrs sets attribute values for a set of ids.
func (s *AttrStore) SetBulkAttrs(m map[uint64]map[string]interface{}) error {
s.mu.Lock()
defer s.mu.Unlock()
attrs := make(map[uint64]map[string]interface{})
if err := s.db.Update(func(tx *bolt.Tx) error {
// Collect and sort keys.
ids := make([]uint64, 0, len(m))
for id := range m {
ids = append(ids, id)
}
sort.Sort(uint64Slice(ids))
// Update attributes for each id.
for _, id := range ids {
attr, err := txUpdateAttrs(tx, id, m[id])
if err != nil {
return err
}
attrs[id] = attr
}
return nil
}); err != nil {
return err
}
// Swap attributes map in cache.
for id, attr := range attrs {
s.attrCache.Set(id, attr)
}
return nil
}
// Blocks returns a list of all blocks in the store.
func (s *AttrStore) Blocks() ([]pilosa.AttrBlock, error) {
tx, err := s.db.Begin(false)
if err != nil {
return nil, err
}
defer tx.Rollback()
// Wrap cursor to segment by block.
cur := newBlockCursor(tx.Bucket([]byte("attrs")).Cursor(), AttrBlockSize)
// Iterate over each block.
var blocks []pilosa.AttrBlock
for cur.nextBlock() {
block := pilosa.AttrBlock{ID: cur.blockID()}
// Compute checksum of every key/value in block.
h := xxhash.New()
for k, v := cur.next(); k != nil; k, v = cur.next() {
h.Write(k)
h.Write(v)
}
block.Checksum = h.Sum(nil)
// Append block.
blocks = append(blocks, block)
}
return blocks, nil
}
// BlockData returns all data for a single block.
func (s *AttrStore) BlockData(i uint64) (map[uint64]map[string]interface{}, error) {
m := make(map[uint64]map[string]interface{})
// Start read-only transaction.
tx, err := s.db.Begin(false)
if err != nil {
return nil, err
}
defer tx.Rollback()
// Move to the start of the block.
min := u64tob(uint64(i) * AttrBlockSize)
max := u64tob(uint64(i+1) * AttrBlockSize)
cur := tx.Bucket([]byte("attrs")).Cursor()
for k, v := cur.Seek(min); k != nil; k, v = cur.Next() {
// Exit if we're past the end of the block.
if bytes.Compare(k, max) != -1 {
break
}
// Decode attribute map and associate with id.
attrs, err := pilosa.DecodeAttrs(v)
if err != nil {
return nil, err
}
m[btou64(k)] = attrs
}
return m, nil
}
// txAttrs returns a map of attributes for an id.
func txAttrs(tx *bolt.Tx, id uint64) (map[string]interface{}, error) {
v := tx.Bucket([]byte("attrs")).Get(u64tob(id))
if v == nil {
return emptyMap, nil
}
return pilosa.DecodeAttrs(v)
}
// txUpdateAttrs updates the attributes for an id.
// Returns the new combined set of attributes for the id.
func txUpdateAttrs(tx *bolt.Tx, id uint64, m map[string]interface{}) (map[string]interface{}, error) {
attr, err := txAttrs(tx, id)
if err != nil {
return nil, err
}
// Create a new map if it is empty so we don't update emptyMap.
if len(attr) == 0 {
attr = make(map[string]interface{}, len(m))
}
// Merge attributes with original values.
// Nil values should delete keys.
for k, v := range m {
if v == nil {
delete(attr, k)
continue
}
switch v := v.(type) {
case int:
attr[k] = int64(v)
case uint:
attr[k] = int64(v)
case uint64:
attr[k] = int64(v)
case string, int64, bool, float64:
attr[k] = v
default:
return nil, fmt.Errorf("invalid attr type: %T", v)
}
}
// Marshal and save new values.
buf, err := pilosa.EncodeAttrs(attr)
if err != nil {
return nil, err
}
if err := tx.Bucket([]byte("attrs")).Put(u64tob(id), buf); err != nil {
return nil, err
}
return attr, nil
}
// u64tob encodes v to big endian encoding.
func u64tob(v uint64) []byte {
b := make([]byte, 8)
binary.BigEndian.PutUint64(b, v)
return b
}
// btou64 decodes b from big endian encoding.
func btou64(b []byte) uint64 { return binary.BigEndian.Uint64(b) }
// emptyMap is a reusable map that contains no keys.
var emptyMap = make(map[string]interface{})
// mapContains returns true if all keys & values of subset are in m.
func mapContains(m, subset map[string]interface{}) bool {
for k, v := range subset {
value, ok := m[k]
if !ok || value != v {
return false
}
}
return true
}
// uint64Slice represents a sortable slice of uint64 numbers.
type uint64Slice []uint64
func (p uint64Slice) Swap(i, j int) { p[i], p[j] = p[j], p[i] }
func (p uint64Slice) Len() int { return len(p) }
func (p uint64Slice) Less(i, j int) bool { return p[i] < p[j] }
// merge combines p and other to a unique sorted set of values.
// p and other must both have unique sets and be sorted.
func (p uint64Slice) merge(other []uint64) []uint64 {
ret := make([]uint64, 0, len(p))
i, j := 0, 0
for i < len(p) && j < len(other) {
a, b := p[i], other[j]
if a == b {
ret = append(ret, a)
i, j = i+1, j+1
} else if a < b {
ret = append(ret, a)
i++
} else {
ret = append(ret, b)
j++
}
}
if i < len(p) {
ret = append(ret, p[i:]...)
} else if j < len(other) {
ret = append(ret, other[j:]...)
}
return ret
}
// blockCursor represents a cursor for iterating over blocks of a bolt bucket.
type blockCursor struct {
cur *bolt.Cursor
base uint64
n uint64
buf struct {
key []byte
value []byte
filled bool
}
}
// newBlockCursor returns a new block cursor that wraps cur using n sized blocks.
func newBlockCursor(c *bolt.Cursor, n int) blockCursor {
cur := blockCursor{
cur: c,
n: uint64(n),
}
cur.buf.key, cur.buf.value = c.First()
cur.buf.filled = true
return cur
}
// blockID returns the current block ID. Only valid after call to nextBlock().
func (cur *blockCursor) blockID() uint64 { return cur.base }
// nextBlock moves the cursor to the next block.
// Returns true if another block exists, otherwise returns false.
func (cur *blockCursor) nextBlock() bool {
if cur.buf.key == nil {
return false
}
cur.base = binary.BigEndian.Uint64(cur.buf.key) / cur.n
return true
}
// next returns the next key/value within the block.
// Returns nils at the end of the block.
func (cur *blockCursor) next() (key, value []byte) {
// Use buffered value, if set.
if cur.buf.filled {
key, value = cur.buf.key, cur.buf.value
cur.buf.filled = false
return key, value
}
// Read next key.
key, value = cur.cur.Next()
// Fill buffer for EOF.
if key == nil {
cur.buf.key, cur.buf.value, cur.buf.filled = key, value, false
return nil, nil
}
// Parse key and buffer if outside of block.
id := binary.BigEndian.Uint64(key)
if id/cur.n > cur.base {
cur.buf.key, cur.buf.value, cur.buf.filled = key, value, true
return nil, nil
}
return key, value
}

2
fragment.go
View file

@ -108,7 +108,7 @@ type Fragment struct {
// Row attribute storage.
// This is set by the parent frame unless overridden for testing.
RowAttrStore *AttrStore
RowAttrStore AttrStore
stats StatsClient
}

2
fragment_test.go
View file

@ -702,7 +702,7 @@ func TestFragment_TopN_CacheSize(t *testing.T) {
Fragment: frag,
RowAttrStore: test.MustOpenAttrStore(),
}
f.Fragment.RowAttrStore = f.RowAttrStore.AttrStore
f.Fragment.RowAttrStore = f.RowAttrStore
if err := f.Open(); err != nil {
panic(err)
}

9
frame.go
View file

@ -51,7 +51,7 @@ type Frame struct {
views map[string]*View
// Row attribute storage and cache
rowAttrStore *AttrStore
rowAttrStore AttrStore
broadcaster Broadcaster
Stats StatsClient
@ -80,8 +80,9 @@ func NewFrame(path, index, name string) (*Frame, error) {
index: index,
name: name,
views: make(map[string]*View),
rowAttrStore: NewAttrStore(filepath.Join(path, ".data")),
views: make(map[string]*View),
rowAttrStore: NopAttrStore,
broadcaster: NopBroadcaster,
Stats: NopStatsClient,
@ -108,7 +109,7 @@ func (f *Frame) Index() string { return f.index }
func (f *Frame) Path() string { return f.path }
// RowAttrStore returns the attribute storage.
func (f *Frame) RowAttrStore() *AttrStore { return f.rowAttrStore }
func (f *Frame) RowAttrStore() AttrStore { return f.rowAttrStore }
// MaxSlice returns the max slice in the frame.
func (f *Frame) MaxSlice() uint64 {

7
holder.go
View file

@ -55,6 +55,9 @@ type Holder struct {
opened chan struct{}
Broadcaster Broadcaster
NewAttrStore func(string) AttrStore
// Close management
wg sync.WaitGroup
closing chan struct{}
@ -82,6 +85,8 @@ func NewHolder() *Holder {
Broadcaster: NopBroadcaster,
Stats: NopStatsClient,
NewAttrStore: NewNopAttrStore,
CacheFlushInterval: DefaultCacheFlushInterval,
LogOutput: os.Stderr,
@ -372,6 +377,8 @@ func (h *Holder) newIndex(path, name string) (*Index, error) {
index.LogOutput = h.LogOutput
index.Stats = h.Stats.WithTags(fmt.Sprintf("index:%s", index.Name()))
index.broadcaster = h.Broadcaster
index.NewAttrStore = h.NewAttrStore
index.columnAttrStore = h.NewAttrStore(filepath.Join(index.path, ".data"))
return index, nil
}

14
index.go
View file

@ -55,8 +55,10 @@ type Index struct {
remoteMaxSlice uint64
remoteMaxInverseSlice uint64
NewAttrStore func(string) AttrStore
// Column attribute storage and cache.
columnAttrStore *AttrStore
columnAttrStore AttrStore
// InputDefinitions by name.
inputDefinitions map[string]*InputDefinition
@ -83,7 +85,8 @@ func NewIndex(path, name string) (*Index, error) {
remoteMaxSlice: 0,
remoteMaxInverseSlice: 0,
columnAttrStore: NewAttrStore(filepath.Join(path, ".data")),
NewAttrStore: NewNopAttrStore,
columnAttrStore: NopAttrStore,
columnLabel: DefaultColumnLabel,
@ -100,7 +103,7 @@ func (i *Index) Name() string { return i.name }
func (i *Index) Path() string { return i.path }
// ColumnAttrStore returns the storage for column attributes.
func (i *Index) ColumnAttrStore() *AttrStore { return i.columnAttrStore }
func (i *Index) ColumnAttrStore() AttrStore { return i.columnAttrStore }
// SetColumnLabel sets the column label. Persists to meta file on update.
func (i *Index) SetColumnLabel(v string) error {
@ -256,9 +259,7 @@ func (i *Index) Close() error {
defer i.mu.Unlock()
// Close the attribute store.
if i.columnAttrStore != nil {
i.columnAttrStore.Close()
}
i.columnAttrStore.Close()
// Close all frames.
for _, f := range i.frames {
@ -530,6 +531,7 @@ func (i *Index) newFrame(path, name string) (*Frame, error) {
f.LogOutput = i.LogOutput
f.Stats = i.Stats.WithTags(fmt.Sprintf("frame:%s", name))
f.broadcaster = i.broadcaster
f.rowAttrStore = i.NewAttrStore(filepath.Join(f.path, ".data"))
return f, nil
}

4
server.go
View file

@ -74,6 +74,8 @@ type Server struct {
GCNotifier GCNotifier
NewAttrStore func(string) AttrStore
// Background monitoring intervals.
AntiEntropyInterval time.Duration
MetricInterval time.Duration
@ -107,6 +109,8 @@ func NewServer() *Server {
GCNotifier: NopGCNotifier,
NewAttrStore: NewNopAttrStore,
AntiEntropyInterval: DefaultAntiEntropyInterval,
MetricInterval: 0,
DiagnosticInterval: 0,

4
server/server.go
View file

@ -33,6 +33,7 @@ import (
"crypto/tls"
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/boltdb"
"github.com/pilosa/pilosa/gcnotify"
"github.com/pilosa/pilosa/gopsutil"
"github.com/pilosa/pilosa/gossip"
@ -147,6 +148,9 @@ func (m *Command) SetupServer() error {
// Configure data directory (for Cluster .topology)
m.Server.Cluster.Path = m.Config.DataDir
m.Server.NewAttrStore = boltdb.NewAttrStore
m.Server.Holder.NewAttrStore = boltdb.NewAttrStore
// Configure holder.
m.Server.Logger().Printf("Using data from: %s\n", m.Config.DataDir)
m.Server.Holder.Path = m.Config.DataDir

11
test/attr.go
View file

@ -22,15 +22,16 @@ import (
"testing"
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/boltdb"
)
// AttrStore represents a test wrapper for pilosa.AttrStore.
type AttrStore struct {
*pilosa.AttrStore
pilosa.AttrStore
}
// NewAttrStore returns a new instance of AttrStore.
func NewAttrStore() *AttrStore {
func NewAttrStore(string) pilosa.AttrStore {
f, err := ioutil.TempFile("", "pilosa-attr-")
if err != nil {
panic(err)
@ -38,7 +39,7 @@ func NewAttrStore() *AttrStore {
f.Close()
os.Remove(f.Name())
return &AttrStore{AttrStore: pilosa.NewAttrStore(f.Name())}
return &AttrStore{boltdb.NewAttrStore(f.Name())}
}
func BenchmarkAttrStore_Duplicate(b *testing.B) {
@ -74,8 +75,8 @@ func BenchmarkAttrStore_Duplicate(b *testing.B) {
}
// MustOpenAttrStore returns a new, opened attribute store at a temporary path. Panic on error.
func MustOpenAttrStore() *AttrStore {
s := NewAttrStore()
func MustOpenAttrStore() pilosa.AttrStore {
s := NewAttrStore("")
if err := s.Open(); err != nil {
panic(err)
}

6
test/fragment.go
View file

@ -27,7 +27,7 @@ const SliceWidth = pilosa.SliceWidth
// Fragment is a test wrapper for pilosa.Fragment.
type Fragment struct {
*pilosa.Fragment
RowAttrStore *AttrStore
RowAttrStore pilosa.AttrStore
}
// NewFragment returns a new instance of Fragment with a temporary path.
@ -43,7 +43,7 @@ func NewFragment(index, frame, view string, slice uint64, cacheType string) *Fra
RowAttrStore: MustOpenAttrStore(),
}
f.Fragment.CacheType = cacheType
f.Fragment.RowAttrStore = f.RowAttrStore.AttrStore
f.Fragment.RowAttrStore = f.RowAttrStore
return f
}
@ -78,7 +78,7 @@ func (f *Fragment) Reopen() error {
f.Fragment = pilosa.NewFragment(path, f.Index(), f.Frame(), f.View(), f.Slice())
f.Fragment.CacheType = cacheType
f.Fragment.RowAttrStore = f.RowAttrStore.AttrStore
f.Fragment.RowAttrStore = f.RowAttrStore
if err := f.Open(); err != nil {
return err
}

3
test/holder.go
View file

@ -20,6 +20,7 @@ import (
"os"
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/boltdb"
)
// Holder is a test wrapper for pilosa.Holder.
@ -38,6 +39,7 @@ func NewHolder() *Holder {
h := &Holder{Holder: pilosa.NewHolder()}
h.Path = path
h.Holder.LogOutput = &h.LogOutput
h.Holder.NewAttrStore = boltdb.NewAttrStore
return h
}
@ -64,6 +66,7 @@ func (h *Holder) Reopen() error {
h.Holder = pilosa.NewHolder()
h.Holder.Path = path
h.Holder.LogOutput = logOutput
h.Holder.NewAttrStore = boltdb.NewAttrStore
if err := h.Holder.Open(); err != nil {
return err
}

5
test/pilosa.go
View file

@ -25,6 +25,7 @@ import (
"testing"
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/boltdb"
"github.com/pilosa/pilosa/gossip"
"github.com/pilosa/pilosa/server"
"github.com/pkg/errors"
@ -49,6 +50,8 @@ func NewMain() *Main {
m := &Main{Command: server.NewCommand(os.Stdin, os.Stdout, os.Stderr)}
m.Server.Network = *Network
m.Server.NewAttrStore = NewAttrStore
m.Server.Holder.NewAttrStore = NewAttrStore
m.Config.DataDir = path
m.Config.Bind = "http://localhost:0"
m.Config.Cluster.Disabled = true
@ -136,6 +139,8 @@ func (m *Main) Reopen() error {
config := m.Config
m.Command = server.NewCommand(os.Stdin, os.Stdout, os.Stderr)
m.Server.Network = *Network
m.Server.NewAttrStore = boltdb.NewAttrStore
m.Server.Holder.NewAttrStore = m.Server.NewAttrStore
m.Config = config
// Run new program.

2
view.go
View file

@ -63,7 +63,7 @@ type View struct {
broadcaster Broadcaster
stats StatsClient
RowAttrStore *AttrStore
RowAttrStore AttrStore
LogOutput io.Writer
}

6
view_test.go
View file

@ -26,7 +26,7 @@ import (
// View is a test wrapper for pilosa.View.
type View struct {
*pilosa.View
RowAttrStore *test.AttrStore
RowAttrStore pilosa.AttrStore
}
// NewView returns a new instance of View with a temporary path.
@ -40,7 +40,7 @@ func NewView(index, frame, name string) *View {
View: pilosa.NewView(path, index, frame, name, pilosa.DefaultCacheSize),
RowAttrStore: test.MustOpenAttrStore(),
}
v.View.RowAttrStore = v.RowAttrStore.AttrStore
v.View.RowAttrStore = v.RowAttrStore
return v
}
@ -68,7 +68,7 @@ func (v *View) Reopen() error {
}
v.View = pilosa.NewView(path, v.Index(), v.Frame(), v.Name(), pilosa.DefaultCacheSize)
v.View.RowAttrStore = v.RowAttrStore.AttrStore
v.View.RowAttrStore = v.RowAttrStore
if err := v.Open(); err != nil {
return err
}