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

Merge pull request #1448 from jaffee/unexport-cache-stuff

Browse source 
Unexport cache stuff
This commit is contained in:
Matthew Jaffee 2018-07-02 17:17:48 -05:00 • committed by GitHub
commit fe80b92e9a
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 170 additions and 184 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

158
cache.go
View file

@ -31,8 +31,8 @@ const (
thresholdFactor = 1.1
)
// Cache represents a cache of counts.
type Cache interface {
// cache represents a cache of counts.
type cache interface {
Add(id uint64, n uint64)
BulkAdd(id uint64, n uint64)
Get(id uint64) uint64
@ -48,22 +48,22 @@ type Cache interface {
Recalculate()
// Returns an ordered list of the top ranked bitmaps.
Top() []BitmapPair
Top() []bitmapPair
// SetStats defines the stats client used in the cache.
SetStats(s StatsClient)
}
// LRUCache represents a least recently used Cache implementation.
type LRUCache struct {
// lruCache represents a least recently used Cache implementation.
type lruCache struct {
cache *lru.Cache
counts map[uint64]uint64
stats StatsClient
}
// NewLRUCache returns a new instance of LRUCache.
func NewLRUCache(maxEntries uint32) *LRUCache {
c := &LRUCache{
// newLRUCache returns a new instance of LRUCache.
func newLRUCache(maxEntries uint32) *lruCache {
c := &lruCache{
cache: lru.New(int(maxEntries)),
counts: make(map[uint64]uint64),
stats: NopStatsClient,
@ -73,34 +73,34 @@ func NewLRUCache(maxEntries uint32) *LRUCache {
}
// BulkAdd adds a count to the cache unsorted. You should Invalidate after completion.
func (c *LRUCache) BulkAdd(id, n uint64) {
func (c *lruCache) BulkAdd(id, n uint64) {
c.Add(id, n)
}
// Add adds a count to the cache.
func (c *LRUCache) Add(id, n uint64) {
func (c *lruCache) Add(id, n uint64) {
c.cache.Add(id, n)
c.counts[id] = n
}
// Get returns a count for a given id.
func (c *LRUCache) Get(id uint64) uint64 {
func (c *lruCache) Get(id uint64) uint64 {
n, _ := c.cache.Get(id)
nn, _ := n.(uint64)
return nn
}
// Len returns the number of items in the cache.
func (c *LRUCache) Len() int { return c.cache.Len() }
func (c *lruCache) Len() int { return c.cache.Len() }
// Invalidate is a no-op.
func (c *LRUCache) Invalidate() {}
func (c *lruCache) Invalidate() {}
// Recalculate is a no-op.
func (c *LRUCache) Recalculate() {}
func (c *lruCache) Recalculate() {}
// IDs returns a list of all IDs in the cache.
func (c *LRUCache) IDs() []uint64 {
func (c *lruCache) IDs() []uint64 {
a := make([]uint64, 0, len(c.counts))
for id := range c.counts {
a = append(a, id)
@ -110,33 +110,33 @@ func (c *LRUCache) IDs() []uint64 {
}
// Top returns all counts in the cache.
func (c *LRUCache) Top() []BitmapPair {
a := make([]BitmapPair, 0, len(c.counts))
func (c *lruCache) Top() []bitmapPair {
a := make([]bitmapPair, 0, len(c.counts))
for id, n := range c.counts {
a = append(a, BitmapPair{
a = append(a, bitmapPair{
ID: id,
Count: uint64(n),
})
}
sort.Sort(BitmapPairs(a))
sort.Sort(bitmapPairs(a))
return a
}
// SetStats defines the stats client used in the cache.
func (c *LRUCache) SetStats(s StatsClient) {
func (c *lruCache) SetStats(s StatsClient) {
c.stats = s
}
func (c *LRUCache) onEvicted(key lru.Key, _ interface{}) { delete(c.counts, key.(uint64)) }
func (c *lruCache) onEvicted(key lru.Key, _ interface{}) { delete(c.counts, key.(uint64)) }
// Ensure LRUCache implements Cache.
var _ Cache = &LRUCache{}
var _ cache = &lruCache{}
// RankCache represents a cache with sorted entries.
type RankCache struct {
// rankCache represents a cache with sorted entries.
type rankCache struct {
mu sync.Mutex
entries map[uint64]uint64
rankings []BitmapPair // cached, ordered list
rankings []bitmapPair // cached, ordered list
updateN int
updateTime time.Time
@ -155,8 +155,8 @@ type RankCache struct {
}
// NewRankCache returns a new instance of RankCache.
func NewRankCache(maxEntries uint32) *RankCache {
return &RankCache{
func NewRankCache(maxEntries uint32) *rankCache {
return &rankCache{
maxEntries: maxEntries,
thresholdBuffer: int(thresholdFactor * float64(maxEntries)),
entries: make(map[uint64]uint64),
@ -165,7 +165,7 @@ func NewRankCache(maxEntries uint32) *RankCache {
}
// Add adds a count to the cache.
func (c *RankCache) Add(id uint64, n uint64) {
func (c *rankCache) Add(id uint64, n uint64) {
c.mu.Lock()
defer c.mu.Unlock()
// Ignore if the column count is below the threshold.
@ -179,7 +179,7 @@ func (c *RankCache) Add(id uint64, n uint64) {
}
// BulkAdd adds a count to the cache unsorted. You should Invalidate after completion.
func (c *RankCache) BulkAdd(id uint64, n uint64) {
func (c *rankCache) BulkAdd(id uint64, n uint64) {
c.mu.Lock()
defer c.mu.Unlock()
if n < c.thresholdValue {
@ -190,21 +190,21 @@ func (c *RankCache) BulkAdd(id uint64, n uint64) {
}
// Get returns a count for a given id.
func (c *RankCache) Get(id uint64) uint64 {
func (c *rankCache) Get(id uint64) uint64 {
c.mu.Lock()
defer c.mu.Unlock()
return c.entries[id]
}
// Len returns the number of items in the cache.
func (c *RankCache) Len() int {
func (c *rankCache) Len() int {
c.mu.Lock()
defer c.mu.Unlock()
return len(c.entries)
}
// IDs returns a list of all IDs in the cache.
func (c *RankCache) IDs() []uint64 {
func (c *rankCache) IDs() []uint64 {
c.mu.Lock()
defer c.mu.Unlock()
a := make([]uint64, 0, len(c.entries))
@ -216,21 +216,21 @@ func (c *RankCache) IDs() []uint64 {
}
// Invalidate recalculates the entries by rank.
func (c *RankCache) Invalidate() {
func (c *rankCache) Invalidate() {
c.mu.Lock()
defer c.mu.Unlock()
c.invalidate()
}
// Recalculate rebuilds the cache.
func (c *RankCache) Recalculate() {
func (c *rankCache) Recalculate() {
c.mu.Lock()
defer c.mu.Unlock()
c.stats.Count("cache.recalculate", 1, 1.0)
c.recalculate()
}
func (c *RankCache) invalidate() {
func (c *rankCache) invalidate() {
// Don't invalidate more than once every X seconds.
// TODO: consider making this configurable.
if time.Since(c.updateTime).Seconds() < 10 {
@ -240,23 +240,23 @@ func (c *RankCache) invalidate() {
c.recalculate()
}
func (c *RankCache) recalculate() {
func (c *rankCache) recalculate() {
// Convert cache to a sorted list.
rankings := make([]BitmapPair, 0, len(c.entries))
rankings := make([]bitmapPair, 0, len(c.entries))
for id, cnt := range c.entries {
rankings = append(rankings, BitmapPair{
rankings = append(rankings, bitmapPair{
ID: id,
Count: cnt,
})
}
sort.Sort(BitmapPairs(rankings))
sort.Sort(bitmapPairs(rankings))
// Store the count of the item at the threshold index.
c.rankings = rankings
length := len(c.rankings)
c.stats.Gauge("RankCache", float64(length), 1.0)
var removeItems []BitmapPair // cached, ordered list
var removeItems []bitmapPair // cached, ordered list
if length > int(c.maxEntries) {
c.thresholdValue = rankings[c.maxEntries].Count
removeItems = c.rankings[c.maxEntries:]
@ -278,38 +278,38 @@ func (c *RankCache) recalculate() {
}
// SetStats defines the stats client used in the cache.
func (c *RankCache) SetStats(s StatsClient) {
func (c *rankCache) SetStats(s StatsClient) {
c.stats = s
}
// Top returns an ordered list of pairs.
func (c *RankCache) Top() []BitmapPair { return c.rankings }
func (c *rankCache) Top() []bitmapPair { return c.rankings }
// WriteTo writes the cache to w.
func (c *RankCache) WriteTo(w io.Writer) (n int64, err error) {
func (c *rankCache) WriteTo(w io.Writer) (n int64, err error) {
panic("FIXME: TODO")
}
// ReadFrom read from r into the cache.
func (c *RankCache) ReadFrom(r io.Reader) (n int64, err error) {
func (c *rankCache) ReadFrom(r io.Reader) (n int64, err error) {
panic("FIXME: TODO")
}
// Ensure RankCache implements Cache.
var _ Cache = &RankCache{}
var _ cache = &rankCache{}
// BitmapPair represents a id/count pair with an associated identifier.
type BitmapPair struct {
// bitmapPair represents a id/count pair with an associated identifier.
type bitmapPair struct {
ID uint64
Count uint64
}
// BitmapPairs is a sortable list of BitmapPair objects.
type BitmapPairs []BitmapPair
// bitmapPairs is a sortable list of BitmapPair objects.
type bitmapPairs []bitmapPair
func (p BitmapPairs) Swap(i, j int) { p[i], p[j] = p[j], p[i] }
func (p BitmapPairs) Len() int { return len(p) }
func (p BitmapPairs) Less(i, j int) bool { return p[i].Count > p[j].Count }
func (p bitmapPairs) Swap(i, j int) { p[i], p[j] = p[j], p[i] }
func (p bitmapPairs) Len() int { return len(p) }
func (p bitmapPairs) Less(i, j int) bool { return p[i].Count > p[j].Count }
// Pair holds an id/count pair.
type Pair struct {
@ -341,14 +341,14 @@ func (p Pairs) Swap(i, j int) { p[i], p[j] = p[j], p[i] }
func (p Pairs) Len() int { return len(p) }
func (p Pairs) Less(i, j int) bool { return p[i].Count > p[j].Count }
// PairHeap is a heap implementation over a group of Pairs.
type PairHeap struct {
// pairHeap is a heap implementation over a group of Pairs.
type pairHeap struct {
Pairs
}
// Less implemets the Sort interface.
// reports whether the element with index i should sort before the element with index j.
func (p PairHeap) Less(i, j int) bool { return p.Pairs[i].Count < p.Pairs[j].Count }
func (p pairHeap) Less(i, j int) bool { return p.Pairs[i].Count < p.Pairs[j].Count }
// Push appends the element onto the Pair slice.
func (p *Pairs) Push(x interface{}) {
@ -461,60 +461,52 @@ func (p uint64Slice) merge(other []uint64) []uint64 {
return ret
}
// BitmapCache provides an interface for caching full bitmaps.
type BitmapCache interface {
// bitmapCache provides an interface for caching full bitmaps.
type bitmapCache interface {
Fetch(id uint64) (*Row, bool)
Add(id uint64, b *Row)
}
// SimpleCache implements BitmapCache
// simpleCache implements BitmapCache
// it is meant to be a short-lived cache for cases where writes are continuing to access
// the same row within a short time frame (i.e. good for write-heavy loads)
// A read-heavy use case would cause the cache to get bigger, potentially causing the
// node to run out of memory.
type SimpleCache struct {
type simpleCache struct {
cache map[uint64]*Row
}
// Fetch retrieves the bitmap at the id in the cache.
func (s *SimpleCache) Fetch(id uint64) (*Row, bool) {
func (s *simpleCache) Fetch(id uint64) (*Row, bool) {
m, ok := s.cache[id]
return m, ok
}
// Add adds the bitmap to the cache, keyed on the id.
func (s *SimpleCache) Add(id uint64, b *Row) {
func (s *simpleCache) Add(id uint64, b *Row) {
s.cache[id] = b
}
// NopCache represents a no-op Cache implementation.
type NopCache struct {
// nopCache represents a no-op Cache implementation.
type nopCache struct {
stats StatsClient
}
// Ensure NopCache implements Cache.
var _ Cache = &NopCache{}
// NewNopCache returns a new instance of NopCache.
func NewNopCache() *NopCache {
return &NopCache{
stats: NopStatsClient,
}
var globalNopCache cache = nopCache{
stats: NopStatsClient,
}
func (c *NopCache) Add(id uint64, n uint64) {}
func (c *NopCache) BulkAdd(id uint64, n uint64) {}
func (c *NopCache) Get(id uint64) uint64 { return 0 }
func (c *NopCache) IDs() []uint64 { return make([]uint64, 0) }
func (c nopCache) Add(uint64, uint64) {}
func (c nopCache) BulkAdd(uint64, uint64) {}
func (c nopCache) Get(uint64) uint64 { return 0 }
func (c nopCache) IDs() []uint64 { return []uint64{} }
func (c *NopCache) Invalidate() {}
func (c *NopCache) Len() int { return 0 }
func (c *NopCache) Recalculate() {
}
func (c *NopCache) SetStats(s StatsClient) {
c.stats = s
}
func (c nopCache) Invalidate() {}
func (c nopCache) Len() int { return 0 }
func (c nopCache) Recalculate() {}
func (c nopCache) SetStats(StatsClient) {}
func (c *NopCache) Top() []BitmapPair {
return []BitmapPair{}
func (c nopCache) Top() []bitmapPair {
return []bitmapPair{}
}

74
client.go
View file

@ -75,66 +75,62 @@ var _ InternalQueryClient = NewNopInternalQueryClient()
type NopInternalClient struct{}
func NewNopInternalClient() *NopInternalClient {
return &NopInternalClient{}
func NewNopInternalClient() NopInternalClient {
return NopInternalClient{}
}
var _ InternalClient = NewNopInternalClient()
func (n *NopInternalClient) MaxShardByIndex(ctx context.Context) (map[string]uint64, error) {
func (n NopInternalClient) MaxShardByIndex(context.Context) (map[string]uint64, error) {
return nil, nil
}
func (n *NopInternalClient) Schema(ctx context.Context) ([]*IndexInfo, error) {
func (n NopInternalClient) Schema(ctx context.Context) ([]*IndexInfo, error) { return nil, nil }
func (n NopInternalClient) CreateIndex(ctx context.Context, index string, opt IndexOptions) error {
return nil
}
func (n NopInternalClient) FragmentNodes(ctx context.Context, index string, shard uint64) ([]*Node, error) {
return nil, nil
}
func (n *NopInternalClient) CreateIndex(ctx context.Context, index string, opt IndexOptions) error {
return nil
}
func (n *NopInternalClient) FragmentNodes(ctx context.Context, index string, shard uint64) ([]*Node, error) {
func (n NopInternalClient) Query(ctx context.Context, index string, queryRequest *internal.QueryRequest) (*internal.QueryResponse, error) {
return nil, nil
}
func (n *NopInternalClient) Query(ctx context.Context, index string, queryRequest *internal.QueryRequest) (*internal.QueryResponse, error) {
func (n NopInternalClient) QueryNode(ctx context.Context, uri *URI, index string, queryRequest *internal.QueryRequest) (*internal.QueryResponse, error) {
return nil, nil
}
func (n *NopInternalClient) QueryNode(ctx context.Context, uri *URI, index string, queryRequest *internal.QueryRequest) (*internal.QueryResponse, error) {
func (n NopInternalClient) Import(ctx context.Context, index, field string, shard uint64, bits []Bit) error {
return nil
}
func (n NopInternalClient) ImportK(ctx context.Context, index, field string, bits []Bit) error {
return nil
}
func (n NopInternalClient) EnsureIndex(ctx context.Context, name string, options IndexOptions) error {
return nil
}
func (n NopInternalClient) EnsureField(ctx context.Context, indexName string, fieldName string) error {
return nil
}
func (n NopInternalClient) ImportValue(ctx context.Context, index, field string, shard uint64, vals []FieldValue) error {
return nil
}
func (n NopInternalClient) ExportCSV(ctx context.Context, index, field string, shard uint64, w io.Writer) error {
return nil
}
func (n NopInternalClient) CreateField(ctx context.Context, index, field string) error { return nil }
func (n NopInternalClient) FragmentBlocks(ctx context.Context, uri *URI, index, field string, shard uint64) ([]FragmentBlock, error) {
return nil, nil
}
func (n *NopInternalClient) Import(ctx context.Context, index, field string, shard uint64, bits []Bit) error {
return nil
}
func (n *NopInternalClient) ImportK(ctx context.Context, index, field string, bits []Bit) error {
return nil
}
func (n *NopInternalClient) EnsureIndex(ctx context.Context, name string, options IndexOptions) error {
return nil
}
func (n *NopInternalClient) EnsureField(ctx context.Context, indexName string, fieldName string) error {
return nil
}
func (n *NopInternalClient) ImportValue(ctx context.Context, index, field string, shard uint64, vals []FieldValue) error {
return nil
}
func (n *NopInternalClient) ExportCSV(ctx context.Context, index, field string, shard uint64, w io.Writer) error {
return nil
}
func (n *NopInternalClient) CreateField(ctx context.Context, index, field string) error {
return nil
}
func (n *NopInternalClient) FragmentBlocks(ctx context.Context, uri *URI, index, field string, shard uint64) ([]FragmentBlock, error) {
return nil, nil
}
func (n *NopInternalClient) BlockData(ctx context.Context, uri *URI, index, field string, shard uint64, block int) ([]uint64, []uint64, error) {
func (n NopInternalClient) BlockData(ctx context.Context, uri *URI, index, field string, shard uint64, block int) ([]uint64, []uint64, error) {
return nil, nil, nil
}
func (n *NopInternalClient) ColumnAttrDiff(ctx context.Context, uri *URI, index string, blks []AttrBlock) (map[uint64]map[string]interface{}, error) {
func (n NopInternalClient) ColumnAttrDiff(ctx context.Context, uri *URI, index string, blks []AttrBlock) (map[uint64]map[string]interface{}, error) {
return nil, nil
}
func (n *NopInternalClient) RowAttrDiff(ctx context.Context, uri *URI, index, field string, blks []AttrBlock) (map[uint64]map[string]interface{}, error) {
func (n NopInternalClient) RowAttrDiff(ctx context.Context, uri *URI, index, field string, blks []AttrBlock) (map[uint64]map[string]interface{}, error) {
return nil, nil
}
func (n *NopInternalClient) SendMessage(ctx context.Context, uri *URI, pb proto.Message) error {
func (n NopInternalClient) SendMessage(ctx context.Context, uri *URI, pb proto.Message) error {
return nil
}
func (n *NopInternalClient) RetrieveShardFromURI(ctx context.Context, index, field string, shard uint64, uri URI) (io.ReadCloser, error) {
func (n NopInternalClient) RetrieveShardFromURI(ctx context.Context, index, field string, shard uint64, uri URI) (io.ReadCloser, error) {
return nil, nil
}

36
fragment.go
View file

@ -82,14 +82,14 @@ type fragment struct {
// Cache for row counts.
CacheType string // passed in by field
cache Cache
cache cache
CacheSize uint32
// Stats reporting.
maxRowID uint64
// Cache containing full rows (not just counts).
rowCache BitmapCache
rowCache bitmapCache
// Cached checksums for each block.
checksums map[int][]byte
@ -217,7 +217,7 @@ func (f *fragment) openStorage() error {
// Attach the file to the bitmap to act as a write-ahead log.
f.storage.OpWriter = f.file
f.rowCache = &SimpleCache{make(map[uint64]*Row)}
f.rowCache = &simpleCache{make(map[uint64]*Row)}
return nil
@ -230,9 +230,9 @@ func (f *fragment) openCache() error {
case CacheTypeRanked:
f.cache = NewRankCache(f.CacheSize)
case CacheTypeLRU:
f.cache = NewLRUCache(f.CacheSize)
f.cache = newLRUCache(f.CacheSize)
case CacheTypeNone:
f.cache = NewNopCache()
f.cache = globalNopCache
return nil
default:
return ErrInvalidCacheType
@ -897,7 +897,7 @@ func (f *fragment) top(opt topOptions) ([]Pair, error) {
}
// Iterate over rankings and add to results until we have enough.
results := &PairHeap{}
results := &pairHeap{}
for _, pair := range pairs {
rowID, cnt := pair.ID, pair.Count
@ -1001,7 +1001,7 @@ func (f *fragment) top(opt topOptions) ([]Pair, error) {
return r, nil
}
func (f *fragment) topBitmapPairs(rowIDs []uint64) []BitmapPair {
func (f *fragment) topBitmapPairs(rowIDs []uint64) []bitmapPair {
// Don't retrieve from storage if CacheTypeNone.
if f.CacheType == CacheTypeNone {
return f.cache.Top()
@ -1015,11 +1015,11 @@ func (f *fragment) topBitmapPairs(rowIDs []uint64) []BitmapPair {
}
// Otherwise retrieve specific rows.
pairs := make([]BitmapPair, 0, len(rowIDs))
pairs := make([]bitmapPair, 0, len(rowIDs))
for _, rowID := range rowIDs {
// Look up cache first, if available.
if n := f.cache.Get(rowID); n > 0 {
pairs = append(pairs, BitmapPair{
pairs = append(pairs, bitmapPair{
ID: rowID,
Count: n,
})
@ -1029,13 +1029,13 @@ func (f *fragment) topBitmapPairs(rowIDs []uint64) []BitmapPair {
row := f.row(rowID)
if row.Count() > 0 {
// Otherwise load from storage.
pairs = append(pairs, BitmapPair{
pairs = append(pairs, bitmapPair{
ID: rowID,
Count: row.Count(),
})
}
}
sort.Sort(BitmapPairs(pairs))
sort.Sort(bitmapPairs(pairs))
return pairs
}
@ -1193,18 +1193,18 @@ func (f *fragment) mergeBlock(id int, data []pairSet) (sets, clears []pairSet, e
maxColumnID := uint64(ShardWidth)
// Create buffered iterator for local block.
itrs := make([]*BufIterator, 1, len(data)+1)
itrs[0] = NewBufIterator(
NewLimitIterator(
NewRoaringIterator(f.storage.Iterator()), maxRowID, maxColumnID,
itrs := make([]*bufIterator, 1, len(data)+1)
itrs[0] = newBufIterator(
newLimitIterator(
newRoaringIterator(f.storage.Iterator()), maxRowID, maxColumnID,
),
)
// Append buffered iterators for each incoming block.
for i := range data {
var itr Iterator = NewSliceIterator(data[i].rowIDs, data[i].columnIDs)
itr = NewLimitIterator(itr, maxRowID, maxColumnID)
itrs = append(itrs, NewBufIterator(itr))
var itr iterator = newSliceIterator(data[i].rowIDs, data[i].columnIDs)
itr = newLimitIterator(itr, maxRowID, maxColumnID)
itrs = append(itrs, newBufIterator(itr))
}
// Seek to initial pair.

8
fragment_internal_test.go
View file

@ -891,7 +891,7 @@ func TestFragment_LRUCache_Persistence(t *testing.T) {
}
// Verify correct cache type and size.
if cache, ok := f.cache.(*LRUCache); !ok {
if cache, ok := f.cache.(*lruCache); !ok {
t.Fatalf("unexpected cache: %T", f.cache)
} else if cache.Len() != 1000 {
t.Fatalf("unexpected cache len: %d", cache.Len())
@ -903,7 +903,7 @@ func TestFragment_LRUCache_Persistence(t *testing.T) {
}
// Re-verify correct cache type and size.
if cache, ok := f.cache.(*LRUCache); !ok {
if cache, ok := f.cache.(*lruCache); !ok {
t.Fatalf("unexpected cache: %T", f.cache)
} else if cache.Len() != 1000 {
t.Fatalf("unexpected cache len: %d", cache.Len())
@ -941,7 +941,7 @@ func TestFragment_RankCache_Persistence(t *testing.T) {
}
// Verify correct cache type and size.
if cache, ok := f.cache.(*RankCache); !ok {
if cache, ok := f.cache.(*rankCache); !ok {
t.Fatalf("unexpected cache: %T", f.cache)
} else if cache.Len() != 1000 {
t.Fatalf("unexpected cache len: %d", cache.Len())
@ -956,7 +956,7 @@ func TestFragment_RankCache_Persistence(t *testing.T) {
f = index.Field("f").view(ViewStandard).Fragment(0)
// Re-verify correct cache type and size.
if cache, ok := f.cache.(*RankCache); !ok {
if cache, ok := f.cache.(*rankCache); !ok {
t.Fatalf("unexpected cache: %T", f.cache)
} else if cache.Len() != 1000 {
t.Fatalf("unexpected cache len: %d", cache.Len())

68
iterator.go
View file

@ -20,37 +20,37 @@ import (
"github.com/pilosa/pilosa/roaring"
)
// Iterator is an interface for looping over row/column pairs.
type Iterator interface {
// iterator is an interface for looping over row/column pairs.
type iterator interface {
Seek(rowID, columnID uint64)
Next() (rowID, columnID uint64, eof bool)
}
// BufIterator wraps an iterator to provide the ability to unread values.
type BufIterator struct {
// bufIterator wraps an iterator to provide the ability to unread values.
type bufIterator struct {
buf struct {
rowID uint64
columnID uint64
eof bool
full bool
}
itr Iterator
itr iterator
}
// NewBufIterator returns a buffered iterator that wraps itr.
func NewBufIterator(itr Iterator) *BufIterator {
return &BufIterator{itr: itr}
// newBufIterator returns a buffered iterator that wraps itr.
func newBufIterator(itr iterator) *bufIterator {
return &bufIterator{itr: itr}
}
// Seek moves to the first pair equal to or greater than pseek/bseek.
func (itr *BufIterator) Seek(rowID, columnID uint64) {
func (itr *bufIterator) Seek(rowID, columnID uint64) {
itr.buf.full = false
itr.itr.Seek(rowID, columnID)
}
// Next returns the next pair in the row.
// If a value has been buffered then it is returned and the buffer is cleared.
func (itr *BufIterator) Next() (rowID, columnID uint64, eof bool) {
func (itr *bufIterator) Next() (rowID, columnID uint64, eof bool) {
if itr.buf.full {
itr.buf.full = false
return itr.buf.rowID, itr.buf.columnID, itr.buf.eof
@ -63,7 +63,7 @@ func (itr *BufIterator) Next() (rowID, columnID uint64, eof bool) {
}
// Peek reads the next value but leaves it on the buffer.
func (itr *BufIterator) Peek() (rowID, columnID uint64, eof bool) {
func (itr *bufIterator) Peek() (rowID, columnID uint64, eof bool) {
rowID, columnID, eof = itr.Next()
itr.Unread()
return
@ -71,25 +71,25 @@ func (itr *BufIterator) Peek() (rowID, columnID uint64, eof bool) {
// Unread pushes previous pair on to the buffer.
// Panics if the buffer is already full.
func (itr *BufIterator) Unread() {
func (itr *bufIterator) Unread() {
if itr.buf.full {
panic("pilosa.BufIterator: buffer full")
}
itr.buf.full = true
}
// LimitIterator wraps an Iterator and limits it to a max column/row pair.
type LimitIterator struct {
itr Iterator
// limitIterator wraps an Iterator and limits it to a max column/row pair.
type limitIterator struct {
itr iterator
maxRowID uint64
maxColumnID uint64
eof bool
}
// NewLimitIterator returns a new LimitIterator.
func NewLimitIterator(itr Iterator, maxRowID, maxColumnID uint64) *LimitIterator {
return &LimitIterator{
// newLimitIterator returns a new LimitIterator.
func newLimitIterator(itr iterator, maxRowID, maxColumnID uint64) *limitIterator {
return &limitIterator{
itr: itr,
maxRowID: maxRowID,
maxColumnID: maxColumnID,
@ -97,11 +97,11 @@ func NewLimitIterator(itr Iterator, maxRowID, maxColumnID uint64) *LimitIterator
}
// Seek moves the underlying iterator to a column/row pair.
func (itr *LimitIterator) Seek(rowID, columnID uint64) { itr.itr.Seek(rowID, columnID) }
func (itr *limitIterator) Seek(rowID, columnID uint64) { itr.itr.Seek(rowID, columnID) }
// Next returns the next row/column ID pair.
// If the underlying iterator returns a pair higher than the max then EOF is returned.
func (itr *LimitIterator) Next() (rowID, columnID uint64, eof bool) {
func (itr *limitIterator) Next() (rowID, columnID uint64, eof bool) {
// Always return EOF once it is reached by limit or the underlying iterator.
if itr.eof {
return 0, 0, true
@ -118,22 +118,22 @@ func (itr *LimitIterator) Next() (rowID, columnID uint64, eof bool) {
return rowID, columnID, false
}
// SliceIterator iterates over a pair of row/column ID slices.
type SliceIterator struct {
// sliceIterator iterates over a pair of row/column ID slices.
type sliceIterator struct {
rowIDs []uint64
columnIDs []uint64
i, n int
}
// NewSliceIterator returns an iterator to iterate over a set of row/column ID pairs.
// newSliceIterator returns an iterator to iterate over a set of row/column ID pairs.
// Both slices MUST have an equal length. Otherwise the function will panic.
func NewSliceIterator(rowIDs, columnIDs []uint64) *SliceIterator {
func newSliceIterator(rowIDs, columnIDs []uint64) *sliceIterator {
if len(columnIDs) != len(rowIDs) {
panic(fmt.Sprintf("pilosa.SliceIterator: pair length mismatch: %d != %d", len(rowIDs), len(columnIDs)))
}
return &SliceIterator{
return &sliceIterator{
rowIDs: rowIDs,
columnIDs: columnIDs,
@ -143,7 +143,7 @@ func NewSliceIterator(rowIDs, columnIDs []uint64) *SliceIterator {
// Seek moves the cursor to a given pair.
// If the pair is not found, the iterator seeks to the next pair.
func (itr *SliceIterator) Seek(bseek, pseek uint64) {
func (itr *sliceIterator) Seek(bseek, pseek uint64) {
for i := 0; i < itr.n; i++ {
rowID := itr.rowIDs[i]
columnID := itr.columnIDs[i]
@ -159,7 +159,7 @@ func (itr *SliceIterator) Seek(bseek, pseek uint64) {
}
// Next returns the next row/column ID pair.
func (itr *SliceIterator) Next() (rowID, columnID uint64, eof bool) {
func (itr *sliceIterator) Next() (rowID, columnID uint64, eof bool) {
if itr.i >= itr.n {
return 0, 0, true
}
@ -171,24 +171,24 @@ func (itr *SliceIterator) Next() (rowID, columnID uint64, eof bool) {
return rowID, columnID, false
}
// RoaringIterator converts a roaring.Iterator to output column/row pairs.
type RoaringIterator struct {
// roaringIterator converts a roaring.Iterator to output column/row pairs.
type roaringIterator struct {
itr *roaring.Iterator
}
// NewRoaringIterator returns a new iterator wrapping itr.
func NewRoaringIterator(itr *roaring.Iterator) *RoaringIterator {
return &RoaringIterator{itr: itr}
// newRoaringIterator returns a new iterator wrapping itr.
func newRoaringIterator(itr *roaring.Iterator) *roaringIterator {
return &roaringIterator{itr: itr}
}
// Seek moves the cursor to a pair matching bseek/pseek.
// If the pair is not found then it moves to the next pair.
func (itr *RoaringIterator) Seek(bseek, pseek uint64) {
func (itr *roaringIterator) Seek(bseek, pseek uint64) {
itr.itr.Seek((bseek * ShardWidth) + pseek)
}
// Next returns the next column/row ID pair.
func (itr *RoaringIterator) Next() (rowID, columnID uint64, eof bool) {
func (itr *roaringIterator) Next() (rowID, columnID uint64, eof bool) {
v, eof := itr.itr.Next()
return v / ShardWidth, v % ShardWidth, eof
}

10
iterator_test.go
View file

@ -12,19 +12,17 @@
// See the License for the specific language governing permissions and
// limitations under the License.
package pilosa_test
package pilosa
import (
"reflect"
"testing"
"github.com/pilosa/pilosa"
)
// Ensure slice iterator and iterate over a set of pairs.
func TestSliceIterator(t *testing.T) {
// Initialize iterator.
itr := pilosa.NewSliceIterator(
itr := newSliceIterator(
[]uint64{0, 0, 2, 4},
[]uint64{0, 1, 0, 10},
)
@ -48,7 +46,7 @@ func TestSliceIterator(t *testing.T) {
// Ensure buffered iterator can unread values on to the buffer.
func TestBufIterator(t *testing.T) {
itr := pilosa.NewBufIterator(pilosa.NewSliceIterator(
itr := newBufIterator(newSliceIterator(
[]uint64{0, 0, 1, 2},
[]uint64{1, 3, 0, 100},
))
@ -77,7 +75,7 @@ func TestBufIterator_DoubleFillPanic(t *testing.T) {
func() {
defer func() { v = recover() }()
itr := pilosa.NewBufIterator(pilosa.NewSliceIterator(nil, nil))
itr := newBufIterator(newSliceIterator(nil, nil))
itr.Unread()
itr.Unread()
}()