From b66bedd1ef009de63f38314ecd1a6410d9d1c9ac Mon Sep 17 00:00:00 2001 From: Travis Turner Date: Mon, 19 Mar 2018 11:30:41 -0500 Subject: [PATCH 1/2] put BoltDB behind AttrStore interface --- attr.go | 522 +++++++++----------------------------------- boltdb/attrstore.go | 478 ++++++++++++++++++++++++++++++++++++++++ fragment.go | 2 +- frame.go | 9 +- holder.go | 7 + index.go | 14 +- server.go | 4 + server/server.go | 4 + test/attr.go | 7 +- test/holder.go | 3 + test/pilosa.go | 5 + view.go | 2 +- 12 files changed, 623 insertions(+), 434 deletions(-) create mode 100644 boltdb/attrstore.go diff --git a/attr.go b/attr.go index 437cab433..7e025ea5b 100644 --- a/attr.go +++ b/attr.go @@ -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,128 @@ const ( AttrTypeFloat = 4 ) -// AttrCache represents a cache for attributes. -type AttrCache struct { - mu sync.RWMutex - attrs map[uint64]map[string]interface{} +// AttrStoreGenerator represents an interface for generating an AttrStore. +type AttrStoreGenerator interface { + New(string) AttrStore } -// 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 +// 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) } -// 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 +func init() { + NopAttrStoreGenerator = &nopAttrStoreGenerator{ + Name: "NooP", + } + NopAttrStore = &nopAttrStore{} } -// AttrStore represents a storage layer for attributes. -type AttrStore struct { - mu sync.RWMutex - path string - db *bolt.DB - attrCache *AttrCache +// NopAttrStoreGenerator represents an AttrStoreGenerator that return a no-op AttrStore. +var NopAttrStoreGenerator AttrStoreGenerator + +// NopAttrStore represents an AttrStore that doesn't do anything. +var NopAttrStore AttrStore + +// nopAttrStoreGenerator represents a no-op implementation of the AttrStoreGenerator interface. +type nopAttrStoreGenerator struct { + Name string } -// NewAttrCache returns a new instance of AttrCache. -func NewAttrCache() *AttrCache { - return &AttrCache{ - attrs: make(map[uint64]map[string]interface{}), - } +// New is a no-op implementation of AttrStoreGenerator New method. +func (g *nopAttrStoreGenerator) New(string) AttrStore { + return &nopAttrStore{} } -// NewAttrStore returns a new instance of AttrStore. -func NewAttrStore(path string) *AttrStore { - return &AttrStore{ - path: path, - attrCache: NewAttrCache(), - } -} +// nopAttrStore represents a no-op implementation of the AttrStore interface. +type nopAttrStore struct{} -// 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 +225,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 } diff --git a/boltdb/attrstore.go b/boltdb/attrstore.go new file mode 100644 index 000000000..0ea35e663 --- /dev/null +++ b/boltdb/attrstore.go @@ -0,0 +1,478 @@ +// 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 +} + +// AttrStoreGenerator represents a bolt implementation of the AttrStoreGenerator interface. +type AttrStoreGenerator struct{} + +// New is a bolt implementation of AttrStoreGenerator New method. +func (g *AttrStoreGenerator) New(path string) pilosa.AttrStore { + return NewAttrStore(path) +} + +// AttrStore represents a storage layer for attributes. +type AttrStore struct { + mu sync.RWMutex + path string + db *bolt.DB + attrCache *AttrCache +} + +// NewAttrStoreGenerator returns a new instance of AttrStoreGenerator. +func NewAttrStoreGenerator() *AttrStoreGenerator { + return &AttrStoreGenerator{} +} + +// 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 + } + + 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 +} diff --git a/fragment.go b/fragment.go index 40a42dbe5..ab397af47 100644 --- a/fragment.go +++ b/fragment.go @@ -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 } diff --git a/frame.go b/frame.go index 069e0bc89..466d7ae96 100644 --- a/frame.go +++ b/frame.go @@ -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 { diff --git a/holder.go b/holder.go index 4c6c10423..2ebc1a639 100644 --- a/holder.go +++ b/holder.go @@ -55,6 +55,9 @@ type Holder struct { opened chan struct{} Broadcaster Broadcaster + + AttrStoreGenerator AttrStoreGenerator + // Close management wg sync.WaitGroup closing chan struct{} @@ -82,6 +85,8 @@ func NewHolder() *Holder { Broadcaster: NopBroadcaster, Stats: NopStatsClient, + AttrStoreGenerator: NopAttrStoreGenerator, + 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.AttrStoreGenerator = h.AttrStoreGenerator + index.columnAttrStore = h.AttrStoreGenerator.New(filepath.Join(index.path, ".data")) return index, nil } diff --git a/index.go b/index.go index fb445bb57..ba50f47e1 100644 --- a/index.go +++ b/index.go @@ -55,8 +55,10 @@ type Index struct { remoteMaxSlice uint64 remoteMaxInverseSlice uint64 + AttrStoreGenerator AttrStoreGenerator + // 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")), + AttrStoreGenerator: NopAttrStoreGenerator, + 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.AttrStoreGenerator.New(filepath.Join(f.path, ".data")) return f, nil } diff --git a/server.go b/server.go index 54bacda1d..c939b5ba1 100644 --- a/server.go +++ b/server.go @@ -74,6 +74,8 @@ type Server struct { GCNotifier GCNotifier + AttrStoreGenerator AttrStoreGenerator + // Background monitoring intervals. AntiEntropyInterval time.Duration MetricInterval time.Duration @@ -107,6 +109,8 @@ func NewServer() *Server { GCNotifier: NopGCNotifier, + AttrStoreGenerator: NopAttrStoreGenerator, + AntiEntropyInterval: DefaultAntiEntropyInterval, MetricInterval: 0, DiagnosticInterval: 0, diff --git a/server/server.go b/server/server.go index eac98121d..976e277c4 100644 --- a/server/server.go +++ b/server/server.go @@ -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" @@ -146,6 +147,9 @@ func (m *Command) SetupServer() error { // Configure data directory (for Cluster .topology) m.Server.Cluster.Path = m.Config.DataDir + m.Server.AttrStoreGenerator = boltdb.NewAttrStoreGenerator() + m.Server.Holder.AttrStoreGenerator = m.Server.AttrStoreGenerator + // Configure holder. m.Server.Logger().Printf("Using data from: %s\n", m.Config.DataDir) m.Server.Holder.Path = m.Config.DataDir diff --git a/test/attr.go b/test/attr.go index 9363011b1..de87144af 100644 --- a/test/attr.go +++ b/test/attr.go @@ -21,12 +21,13 @@ import ( "sync" "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 + *boltdb.AttrStore } // NewAttrStore returns a new instance of AttrStore. @@ -38,7 +39,7 @@ func NewAttrStore() *AttrStore { f.Close() os.Remove(f.Name()) - return &AttrStore{AttrStore: pilosa.NewAttrStore(f.Name())} + return &AttrStore{AttrStore: boltdb.NewAttrStore(f.Name())} } func BenchmarkAttrStore_Duplicate(b *testing.B) { diff --git a/test/holder.go b/test/holder.go index c9d21d2f2..2778cf8c2 100644 --- a/test/holder.go +++ b/test/holder.go @@ -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.AttrStoreGenerator = boltdb.NewAttrStoreGenerator() 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.AttrStoreGenerator = boltdb.NewAttrStoreGenerator() if err := h.Holder.Open(); err != nil { return err } diff --git a/test/pilosa.go b/test/pilosa.go index 2bfd0c1a6..f798e4a85 100644 --- a/test/pilosa.go +++ b/test/pilosa.go @@ -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.AttrStoreGenerator = boltdb.NewAttrStoreGenerator() + m.Server.Holder.AttrStoreGenerator = m.Server.AttrStoreGenerator 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.AttrStoreGenerator = boltdb.NewAttrStoreGenerator() + m.Server.Holder.AttrStoreGenerator = m.Server.AttrStoreGenerator m.Config = config // Run new program. diff --git a/view.go b/view.go index db2eb97f1..dfedebbe3 100644 --- a/view.go +++ b/view.go @@ -63,7 +63,7 @@ type View struct { broadcaster Broadcaster stats StatsClient - RowAttrStore *AttrStore + RowAttrStore AttrStore LogOutput io.Writer } From f9efa6c57928597fb3d0aa62aa757fa0562d338d Mon Sep 17 00:00:00 2001 From: Cody Soyland Date: Fri, 23 Mar 2018 13:51:50 -0500 Subject: [PATCH 2/2] Use type `func(string) AttrStore` in place of interface AttrStoreGenerator for simplicity --- attr.go | 19 +------------------ boltdb/attrstore.go | 15 +-------------- fragment_test.go | 2 +- holder.go | 8 ++++---- index.go | 8 ++++---- server.go | 4 ++-- server/server.go | 4 ++-- test/attr.go | 12 ++++++------ test/fragment.go | 6 +++--- test/holder.go | 4 ++-- test/pilosa.go | 8 ++++---- view_test.go | 6 +++--- 12 files changed, 33 insertions(+), 63 deletions(-) diff --git a/attr.go b/attr.go index 7e025ea5b..03ea4f43f 100644 --- a/attr.go +++ b/attr.go @@ -30,11 +30,6 @@ const ( AttrTypeFloat = 4 ) -// AttrStoreGenerator represents an interface for generating an AttrStore. -type AttrStoreGenerator interface { - New(string) AttrStore -} - // AttrStore represents an interface for handling row/column attributes. type AttrStore interface { Path() string @@ -48,25 +43,13 @@ type AttrStore interface { } func init() { - NopAttrStoreGenerator = &nopAttrStoreGenerator{ - Name: "NooP", - } NopAttrStore = &nopAttrStore{} } -// NopAttrStoreGenerator represents an AttrStoreGenerator that return a no-op AttrStore. -var NopAttrStoreGenerator AttrStoreGenerator - // NopAttrStore represents an AttrStore that doesn't do anything. var NopAttrStore AttrStore -// nopAttrStoreGenerator represents a no-op implementation of the AttrStoreGenerator interface. -type nopAttrStoreGenerator struct { - Name string -} - -// New is a no-op implementation of AttrStoreGenerator New method. -func (g *nopAttrStoreGenerator) New(string) AttrStore { +func NewNopAttrStore(string) AttrStore { return &nopAttrStore{} } diff --git a/boltdb/attrstore.go b/boltdb/attrstore.go index 0ea35e663..ebb539903 100644 --- a/boltdb/attrstore.go +++ b/boltdb/attrstore.go @@ -62,14 +62,6 @@ func (c *AttrCache) Set(id uint64, attrs map[string]interface{}) { c.attrs[id] = attrs } -// AttrStoreGenerator represents a bolt implementation of the AttrStoreGenerator interface. -type AttrStoreGenerator struct{} - -// New is a bolt implementation of AttrStoreGenerator New method. -func (g *AttrStoreGenerator) New(path string) pilosa.AttrStore { - return NewAttrStore(path) -} - // AttrStore represents a storage layer for attributes. type AttrStore struct { mu sync.RWMutex @@ -78,11 +70,6 @@ type AttrStore struct { attrCache *AttrCache } -// NewAttrStoreGenerator returns a new instance of AttrStoreGenerator. -func NewAttrStoreGenerator() *AttrStoreGenerator { - return &AttrStoreGenerator{} -} - // NewAttrCache returns a new instance of AttrCache. func NewAttrCache() *AttrCache { return &AttrCache{ @@ -91,7 +78,7 @@ func NewAttrCache() *AttrCache { } // NewAttrStore returns a new instance of AttrStore. -func NewAttrStore(path string) *AttrStore { +func NewAttrStore(path string) pilosa.AttrStore { return &AttrStore{ path: path, attrCache: NewAttrCache(), diff --git a/fragment_test.go b/fragment_test.go index 8a54d3b07..b86c10e3f 100644 --- a/fragment_test.go +++ b/fragment_test.go @@ -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) } diff --git a/holder.go b/holder.go index 2ebc1a639..63bbbbc86 100644 --- a/holder.go +++ b/holder.go @@ -56,7 +56,7 @@ type Holder struct { Broadcaster Broadcaster - AttrStoreGenerator AttrStoreGenerator + NewAttrStore func(string) AttrStore // Close management wg sync.WaitGroup @@ -85,7 +85,7 @@ func NewHolder() *Holder { Broadcaster: NopBroadcaster, Stats: NopStatsClient, - AttrStoreGenerator: NopAttrStoreGenerator, + NewAttrStore: NewNopAttrStore, CacheFlushInterval: DefaultCacheFlushInterval, @@ -377,8 +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.AttrStoreGenerator = h.AttrStoreGenerator - index.columnAttrStore = h.AttrStoreGenerator.New(filepath.Join(index.path, ".data")) + index.NewAttrStore = h.NewAttrStore + index.columnAttrStore = h.NewAttrStore(filepath.Join(index.path, ".data")) return index, nil } diff --git a/index.go b/index.go index ba50f47e1..707b87213 100644 --- a/index.go +++ b/index.go @@ -55,7 +55,7 @@ type Index struct { remoteMaxSlice uint64 remoteMaxInverseSlice uint64 - AttrStoreGenerator AttrStoreGenerator + NewAttrStore func(string) AttrStore // Column attribute storage and cache. columnAttrStore AttrStore @@ -85,8 +85,8 @@ func NewIndex(path, name string) (*Index, error) { remoteMaxSlice: 0, remoteMaxInverseSlice: 0, - AttrStoreGenerator: NopAttrStoreGenerator, - columnAttrStore: NopAttrStore, + NewAttrStore: NewNopAttrStore, + columnAttrStore: NopAttrStore, columnLabel: DefaultColumnLabel, @@ -531,7 +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.AttrStoreGenerator.New(filepath.Join(f.path, ".data")) + f.rowAttrStore = i.NewAttrStore(filepath.Join(f.path, ".data")) return f, nil } diff --git a/server.go b/server.go index c939b5ba1..7f6eeb6cb 100644 --- a/server.go +++ b/server.go @@ -74,7 +74,7 @@ type Server struct { GCNotifier GCNotifier - AttrStoreGenerator AttrStoreGenerator + NewAttrStore func(string) AttrStore // Background monitoring intervals. AntiEntropyInterval time.Duration @@ -109,7 +109,7 @@ func NewServer() *Server { GCNotifier: NopGCNotifier, - AttrStoreGenerator: NopAttrStoreGenerator, + NewAttrStore: NewNopAttrStore, AntiEntropyInterval: DefaultAntiEntropyInterval, MetricInterval: 0, diff --git a/server/server.go b/server/server.go index 976e277c4..8e1bb6b97 100644 --- a/server/server.go +++ b/server/server.go @@ -147,8 +147,8 @@ func (m *Command) SetupServer() error { // Configure data directory (for Cluster .topology) m.Server.Cluster.Path = m.Config.DataDir - m.Server.AttrStoreGenerator = boltdb.NewAttrStoreGenerator() - m.Server.Holder.AttrStoreGenerator = m.Server.AttrStoreGenerator + 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) diff --git a/test/attr.go b/test/attr.go index de87144af..16e8ff334 100644 --- a/test/attr.go +++ b/test/attr.go @@ -21,17 +21,17 @@ import ( "sync" "testing" + "github.com/pilosa/pilosa" "github.com/pilosa/pilosa/boltdb" ) // AttrStore represents a test wrapper for pilosa.AttrStore. type AttrStore struct { - //*pilosa.AttrStore - *boltdb.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) @@ -39,7 +39,7 @@ func NewAttrStore() *AttrStore { f.Close() os.Remove(f.Name()) - return &AttrStore{AttrStore: boltdb.NewAttrStore(f.Name())} + return &AttrStore{boltdb.NewAttrStore(f.Name())} } func BenchmarkAttrStore_Duplicate(b *testing.B) { @@ -75,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) } diff --git a/test/fragment.go b/test/fragment.go index b208e7a3a..39caa8db2 100644 --- a/test/fragment.go +++ b/test/fragment.go @@ -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 } diff --git a/test/holder.go b/test/holder.go index 2778cf8c2..59402cea0 100644 --- a/test/holder.go +++ b/test/holder.go @@ -39,7 +39,7 @@ func NewHolder() *Holder { h := &Holder{Holder: pilosa.NewHolder()} h.Path = path h.Holder.LogOutput = &h.LogOutput - h.Holder.AttrStoreGenerator = boltdb.NewAttrStoreGenerator() + h.Holder.NewAttrStore = boltdb.NewAttrStore return h } @@ -66,7 +66,7 @@ func (h *Holder) Reopen() error { h.Holder = pilosa.NewHolder() h.Holder.Path = path h.Holder.LogOutput = logOutput - h.Holder.AttrStoreGenerator = boltdb.NewAttrStoreGenerator() + h.Holder.NewAttrStore = boltdb.NewAttrStore if err := h.Holder.Open(); err != nil { return err } diff --git a/test/pilosa.go b/test/pilosa.go index f798e4a85..86623c239 100644 --- a/test/pilosa.go +++ b/test/pilosa.go @@ -50,8 +50,8 @@ func NewMain() *Main { m := &Main{Command: server.NewCommand(os.Stdin, os.Stdout, os.Stderr)} m.Server.Network = *Network - m.Server.AttrStoreGenerator = boltdb.NewAttrStoreGenerator() - m.Server.Holder.AttrStoreGenerator = m.Server.AttrStoreGenerator + m.Server.NewAttrStore = NewAttrStore + m.Server.Holder.NewAttrStore = NewAttrStore m.Config.DataDir = path m.Config.Bind = "http://localhost:0" m.Config.Cluster.Disabled = true @@ -139,8 +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.AttrStoreGenerator = boltdb.NewAttrStoreGenerator() - m.Server.Holder.AttrStoreGenerator = m.Server.AttrStoreGenerator + m.Server.NewAttrStore = boltdb.NewAttrStore + m.Server.Holder.NewAttrStore = m.Server.NewAttrStore m.Config = config // Run new program. diff --git a/view_test.go b/view_test.go index 64041aea0..87ed8e628 100644 --- a/view_test.go +++ b/view_test.go @@ -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 }