diff --git a/attr.go b/attr.go index 1425676cf..c85f77cad 100644 --- a/attr.go +++ b/attr.go @@ -265,6 +265,8 @@ func txUpdateAttrs(tx *bolt.Tx, id uint64, m map[string]interface{}) (map[string attr[k] = uint64(v) case uint: attr[k] = uint64(v) + case float64: + attr[k] = uint64(v) case int64: attr[k] = uint64(v) case string, uint64, bool: diff --git a/bitmap.go b/bitmap.go index 55664b24b..2e27ded7a 100644 --- a/bitmap.go +++ b/bitmap.go @@ -174,6 +174,22 @@ func (b *Bitmap) InvalidateCount() { } } +//increment the bitmap cached counter, note this is an optimization that assumes that the caller is aware the size increased +func (b *Bitmap) IncrementCount(i uint64) { + seg := b.segment(i / SliceWidth) + if seg != nil { + seg.n++ + } +} +func (b *Bitmap) DecrementCount(i uint64) { + seg := b.segment(i / SliceWidth) + if seg != nil { + if seg.n > 0 { + seg.n-- + } + } +} + // Count returns the number of set bits in the bitmap. func (b *Bitmap) Count() uint64 { var n uint64 @@ -312,7 +328,6 @@ func (s *BitmapSegment) Difference(other *BitmapSegment) *BitmapSegment { // SetBit sets the i-th bit of the bitmap. func (s *BitmapSegment) SetBit(i uint64) (changed bool) { s.ensureWritable() - changed, _ = s.data.Add(i) if changed { s.n++ diff --git a/cache.go b/cache.go index a1cba3dec..7c2e96f2d 100644 --- a/cache.go +++ b/cache.go @@ -5,6 +5,7 @@ import ( "fmt" "io" "sort" + "sync" "time" "github.com/golang/groupcache/lru" @@ -14,6 +15,7 @@ import ( // Cache represents a cache for bitmap counts. type Cache interface { Add(bitmapID uint64, n uint64) + BulkAdd(bitmapID uint64, n uint64) Get(bitmapID uint64) uint64 Len() int @@ -43,6 +45,10 @@ func NewLRUCache(maxEntries int) *LRUCache { return c } +func (c *LRUCache) BulkAdd(bitmapID, n uint64) { + c.Add(bitmapID, n) +} + // Add adds a bitmap to the cache. func (c *LRUCache) Add(bitmapID, n uint64) { c.cache.Add(bitmapID, n) @@ -92,6 +98,7 @@ var _ Cache = &LRUCache{} // RankCache represents a cache with sorted entries. type RankCache struct { + mu sync.Mutex entries map[uint64]uint64 rankings []BitmapPair // cached, ordered list @@ -112,33 +119,47 @@ func NewRankCache() *RankCache { // Add adds a bitmap to the cache. func (c *RankCache) Add(bitmapID uint64, n uint64) { + c.mu.Lock() + defer c.mu.Unlock() // Ignore if the bit count on the bitmap is below the threshold. if n < c.ThresholdValue { return } - // Add to cache. c.entries[bitmapID] = n - // If size is larger than the threshold then trim it. - if len(c.entries) > c.ThresholdLength { - c.update() - for id, n := range c.entries { - if n <= c.ThresholdValue { - delete(c.entries, id) - } - } + c.invalidate() +} + +// BulkAdd adds a bitmap to the cache unsorted. You should Invalidate after completion. +func (c *RankCache) BulkAdd(bitmapID uint64, n uint64) { + c.mu.Lock() + defer c.mu.Unlock() + if n < c.ThresholdValue { + return } + + c.entries[bitmapID] = n } // Get returns a bitmap with a given id. -func (c *RankCache) Get(bitmapID uint64) uint64 { return c.entries[bitmapID] } +func (c *RankCache) Get(bitmapID uint64) uint64 { + c.mu.Lock() + defer c.mu.Unlock() + return c.entries[bitmapID] +} // Len returns the number of items in the cache. -func (c *RankCache) Len() int { return len(c.entries) } +func (c *RankCache) Len() int { + c.mu.Lock() + defer c.mu.Unlock() + return len(c.entries) +} // BitmapIDs returns a list of all bitmap IDs in the cache. func (c *RankCache) BitmapIDs() []uint64 { + c.mu.Lock() + defer c.mu.Unlock() a := make([]uint64, 0, len(c.entries)) for id := range c.entries { a = append(a, id) @@ -147,22 +168,25 @@ func (c *RankCache) BitmapIDs() []uint64 { return a } -// Invalidate reorders the entries, if necessary. -func (c *RankCache) Invalidate() { - // Update if there aren't many items or it hasn't been updated recently. - if len(c.rankings) < 50 || (c.updateN > 0 && time.Since(c.updateTime) > 5*time.Minute) { - c.update() - } -} - // update reorders the entries by rank. -func (c *RankCache) update() { +func (c *RankCache) Invalidate() { + c.mu.Lock() + defer c.mu.Unlock() + c.invalidate() + +} +func (c *RankCache) invalidate() { + // Don't invalidate more than once every X seconds. + // TODO: consider making this configurable. + if time.Now().Sub(c.updateTime).Seconds() < 10 { + return + } // Convert cache to a sorted list. rankings := make([]BitmapPair, 0, len(c.entries)) - for id, n := range c.entries { + for id, cnt := range c.entries { rankings = append(rankings, BitmapPair{ ID: id, - Count: n, + Count: cnt, }) } sort.Sort(BitmapPairs(rankings)) @@ -177,6 +201,15 @@ func (c *RankCache) update() { // Reset counters. c.updateTime, c.updateN = time.Now(), 0 + + // If size is larger than the threshold then trim it. + if len(c.entries) > c.ThresholdLength { + for id, cnt := range c.entries { + if cnt <= c.ThresholdValue { + delete(c.entries, id) + } + } + } } // Top returns an ordered list of bitmaps. @@ -233,6 +266,26 @@ 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 } +type PairHeap struct { + Pairs +} + +func (p PairHeap) Less(i, j int) bool { return p.Pairs[i].Count < p.Pairs[j].Count } + +func (h *Pairs) Push(x interface{}) { + // Push and Pop use pointer receivers because they modify the slice's length, + // not just its contents. + *h = append(*h, x.(Pair)) +} + +func (h *Pairs) Pop() interface{} { + old := *h + n := len(old) + x := old[n-1] + *h = old[0 : n-1] + return x +} + // Add merges other into p and returns a new slice. func (p Pairs) Add(other []Pair) []Pair { // Create lookup of key/counts. @@ -327,3 +380,27 @@ func (p uint64Slice) merge(other []uint64) []uint64 { return ret } + +// BitmapCache provides an interface for caching full bitmaps. +type BitmapCache interface { + Fetch(id uint64) (*Bitmap, bool) + Add(id uint64, b *Bitmap) +} + +// SimpleCache implements BitmapCache +// it is meant to be a short-lived cache for cases where writes are continuing to access +// the same bit 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 { + cache map[uint64]*Bitmap +} + +func (s *SimpleCache) Fetch(id uint64) (*Bitmap, bool) { + m, ok := s.cache[id] + return m, ok +} + +func (s *SimpleCache) Add(id uint64, b *Bitmap) { + s.cache[id] = b +} diff --git a/executor.go b/executor.go index 0ba1e7017..774e2443f 100644 --- a/executor.go +++ b/executor.go @@ -52,13 +52,15 @@ func (e *Executor) Execute(ctx context.Context, db string, q *pql.Query, slices // If slices aren't specified, then include all of them. if len(slices) == 0 { - // Round up the number of slices. - maxSlice := e.Index.DB(db).MaxSlice() + if needsSlices(q.Calls) { + // Round up the number of slices. + maxSlice := e.Index.DB(db).MaxSlice() - // Generate a slices of all slices. - slices = make([]uint64, maxSlice+1) - for i := range slices { - slices[i] = uint64(i) + // Generate a slices of all slices. + slices = make([]uint64, maxSlice+1) + for i := range slices { + slices[i] = uint64(i) + } } } @@ -168,6 +170,10 @@ func (e *Executor) executeBitmapCallSlice(ctx context.Context, db string, c *pql // requeries to retrieve the full counts for each of the top results. func (e *Executor) executeTopN(ctx context.Context, db string, c *pql.Call, slices []uint64, opt *ExecOptions) ([]Pair, error) { bitmapIDs, _ := c.Args["ids"].([]uint64) + var n uint64 + if nval, ok := c.Args["n"]; ok { + n = nval.(uint64) + } // Execute original query. pairs, err := e.executeTopNSlices(ctx, db, c, slices, opt) @@ -180,21 +186,29 @@ func (e *Executor) executeTopN(ctx context.Context, db string, c *pql.Call, slic if len(pairs) == 0 || len(bitmapIDs) > 0 || opt.Remote { return pairs, nil } - // Only the original caller should refetch the full counts. other := c.Clone() - other.Args["n"] = 0 + + // Double the size of n for other calls in order to... + // TODO: travis review + other.Args["n"] = len(bitmapIDs) * 2 ids := Pairs(pairs).Keys() sort.Sort(uint64Slice(ids)) other.Args["ids"] = ids - return e.executeTopNSlices(ctx, db, other, slices, opt) + trimmedList, err := e.executeTopNSlices(ctx, db, other, slices, opt) + if err != nil { + return nil, err + } + + if n != 0 && int(n) < len(trimmedList) { + trimmedList = trimmedList[0:n] + } + return trimmedList, nil } func (e *Executor) executeTopNSlices(ctx context.Context, db string, c *pql.Call, slices []uint64, opt *ExecOptions) ([]Pair, error) { - n, _ := c.Args["n"].(uint64) - // Execute calls in bulk on each remote node and merge. mapFn := func(slice uint64) (interface{}, error) { return e.executeTopNSlice(ctx, db, c, slice) @@ -215,11 +229,6 @@ func (e *Executor) executeTopNSlices(ctx context.Context, db string, c *pql.Call // Sort final merged results. sort.Sort(Pairs(results)) - // Only keep the top n after sorting. - if n > 0 && len(results) > int(n) { - results = results[0:n] - } - return results, nil } @@ -954,6 +963,7 @@ func (e *Executor) mapper(ctx context.Context, ch chan mapResponse, nodes []*Nod if n.Host == e.Host { resp.result, resp.err = e.mapperLocal(ctx, nodeSlices, mapFn, reduceFn) } else if !opt.Remote { + results, err := e.exec(ctx, n, db, &pql.Query{Calls: []*pql.Call{c}}, nodeSlices, opt) if len(results) > 0 { resp.result = results[0] @@ -1052,3 +1062,21 @@ func hasOnlySetBitmapAttrs(calls []*pql.Call) bool { } return true } + +func needsSlices(calls []*pql.Call) bool { + if len(calls) == 0 { + return false + } + for _, call := range calls { + switch call.Name { + case "ClearBit", "Profile", "SetBit", "SetBitmapAttrs", "SetProfileAttrs": + continue + case "Count", "TopN": + return true + // default catches Bitmap calls + default: + return true + } + } + return false +} diff --git a/executor_test.go b/executor_test.go index 50be72abb..1cc0985ac 100644 --- a/executor_test.go +++ b/executor_test.go @@ -260,6 +260,40 @@ func TestExecutor_Execute_TopN_fill(t *testing.T) { } } +// Ensure +func TestExecutor_Execute_TopN_fill_small(t *testing.T) { + idx := MustOpenIndex() + defer idx.Close() + + idx.MustCreateFragmentIfNotExists("d", "f", 0).SetBit(0, 0) + idx.MustCreateFragmentIfNotExists("d", "f", 1).SetBit(0, SliceWidth) + idx.MustCreateFragmentIfNotExists("d", "f", 2).SetBit(0, 2*SliceWidth) + idx.MustCreateFragmentIfNotExists("d", "f", 3).SetBit(0, 3*SliceWidth) + idx.MustCreateFragmentIfNotExists("d", "f", 4).SetBit(0, 4*SliceWidth) + + idx.MustCreateFragmentIfNotExists("d", "f", 0).SetBit(1, 0) + idx.MustCreateFragmentIfNotExists("d", "f", 0).SetBit(1, 1) + + idx.MustCreateFragmentIfNotExists("d", "f", 1).SetBit(2, SliceWidth) + idx.MustCreateFragmentIfNotExists("d", "f", 1).SetBit(2, SliceWidth+1) + + idx.MustCreateFragmentIfNotExists("d", "f", 2).SetBit(3, 2*SliceWidth) + idx.MustCreateFragmentIfNotExists("d", "f", 2).SetBit(3, 2*SliceWidth+1) + + idx.MustCreateFragmentIfNotExists("d", "f", 3).SetBit(4, 3*SliceWidth) + idx.MustCreateFragmentIfNotExists("d", "f", 3).SetBit(4, 3*SliceWidth+1) + + // Execute query. + e := NewExecutor(idx.Index, NewCluster(1)) + if result, err := e.Execute(context.Background(), "d", MustParse(`TopN(frame=f, n=1)`), nil, nil); err != nil { + t.Fatal(err) + } else if !reflect.DeepEqual(result, []interface{}{[]pilosa.Pair{ + {Key: 0, Count: 5}, + }}) { + t.Fatalf("unexpected result: %s", spew.Sdump(result)) + } +} + // Ensure a TopN() query with a source bitmap can be executed. func TestExecutor_Execute_TopN_Src(t *testing.T) { idx := MustOpenIndex() @@ -293,6 +327,52 @@ func TestExecutor_Execute_TopN_Src(t *testing.T) { } } +//Ensure TopN handles Attribute filters +func TestExecutor_Execute_TopN_Attr(t *testing.T) { + // + idx := MustOpenIndex() + defer idx.Close() + idx.MustCreateFragmentIfNotExists("d", "f", 0).SetBit(0, 0) + idx.MustCreateFragmentIfNotExists("d", "f", 0).SetBit(0, 1) + idx.MustCreateFragmentIfNotExists("d", "f", 1).SetBit(10, SliceWidth) + + if err := idx.Frame("d", "f").BitmapAttrStore().SetAttrs(10, map[string]interface{}{"category": uint64(123)}); err != nil { + t.Fatal(err) + } + e := NewExecutor(idx.Index, NewCluster(1)) + if result, err := e.Execute(context.Background(), "d", MustParse(`TopN(frame="f", n=1, field="category", filters=[123])`), nil, nil); err != nil { + t.Fatal(err) + } else if !reflect.DeepEqual(result, []interface{}{[]pilosa.Pair{ + {Key: 10, Count: 1}, + }}) { + t.Fatalf("unexpected result: %s", spew.Sdump(result)) + } + +} + +//Ensure TopN handles Attribute filters with source bitmap +func TestExecutor_Execute_TopN_Attr_Src(t *testing.T) { + // + idx := MustOpenIndex() + defer idx.Close() + idx.MustCreateFragmentIfNotExists("d", "f", 0).SetBit(0, 0) + idx.MustCreateFragmentIfNotExists("d", "f", 0).SetBit(0, 1) + idx.MustCreateFragmentIfNotExists("d", "f", 1).SetBit(10, SliceWidth) + + if err := idx.Frame("d", "f").BitmapAttrStore().SetAttrs(10, map[string]interface{}{"category": uint64(123)}); err != nil { + t.Fatal(err) + } + e := NewExecutor(idx.Index, NewCluster(1)) + if result, err := e.Execute(context.Background(), "d", MustParse(`TopN(Bitmap(id=10,frame=f),frame="f", n=1, field="category", filters=[123])`), nil, nil); err != nil { + t.Fatal(err) + } else if !reflect.DeepEqual(result, []interface{}{[]pilosa.Pair{ + {Key: 10, Count: 1}, + }}) { + t.Fatalf("unexpected result: %s", spew.Sdump(result)) + } + +} + // Ensure a range query can be executed. func TestExecutor_Execute_Range(t *testing.T) { idx := MustOpenIndex() diff --git a/fragment.go b/fragment.go index 1181f1739..e70dab80c 100644 --- a/fragment.go +++ b/fragment.go @@ -4,6 +4,7 @@ import ( "archive/tar" "bufio" "bytes" + "container/heap" "context" "crypto/sha1" "encoding/binary" @@ -49,7 +50,7 @@ const ( const ( // DefaultFragmentMaxOpN is the default value for Fragment.MaxOpN. - DefaultFragmentMaxOpN = 1000 + DefaultFragmentMaxOpN = 2000 ) // Fragment represents the intersection of a frame and slice in a database. @@ -68,9 +69,12 @@ type Fragment struct { storageData []byte opN int // number of ops since snapshot - // Bitmap cache. + // Cache for bitmap counts. cache Cache + // Cache containing full bitmaps (not just counts). + bitmapCache BitmapCache + // Cached checksums for each block. checksums map[int][]byte @@ -203,6 +207,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.bitmapCache = &SimpleCache{make(map[uint64]*Bitmap)} return nil @@ -239,9 +244,11 @@ func (f *Fragment) openCache() error { // Read in all bitmaps by ID. // This will cause them to be added to the cache. for _, bitmapID := range pb.BitmapIDs { - n := f.storage.CountRange(bitmapID*SliceWidth, (bitmapID+1)*SliceWidth) - f.cache.Add(bitmapID, n) + //n := f.storage.CountRange(bitmapID*SliceWidth, (bitmapID+1)*SliceWidth) + n := f.bitmap(bitmapID, true, true).Count() + f.cache.BulkAdd(bitmapID, n) } + f.cache.Invalidate() return nil } @@ -305,26 +312,37 @@ func (f *Fragment) logger() *log.Logger { return log.New(f.LogOutput, "", log.Ls func (f *Fragment) Bitmap(bitmapID uint64) *Bitmap { f.mu.Lock() defer f.mu.Unlock() - return f.bitmap(bitmapID) + return f.bitmap(bitmapID, true, true) } -func (f *Fragment) bitmap(bitmapID uint64) *Bitmap { +func (f *Fragment) bitmap(bitmapID uint64, checkBitmapCache bool, updateBitmapCache bool) *Bitmap { + + if checkBitmapCache { + r, ok := f.bitmapCache.Fetch(bitmapID) + if ok && r != nil { + return r + } + } + // Only use a subset of the containers. // NOTE: The start & end ranges must be divisible by data := f.storage.OffsetRange(f.slice*SliceWidth, bitmapID*SliceWidth, (bitmapID+1)*SliceWidth) // Reference bitmap subrange in storage. + // We Clone() data because otherwise bm will contains pointers to containers in storage. + // This causes unexpected results when we cache the bitmap and try to use it later. bm := &Bitmap{ segments: []BitmapSegment{{ - data: *data, + data: *data.Clone(), slice: f.slice, writable: false, }}, } bm.InvalidateCount() - // Update cache. - f.cache.Add(bitmapID, bm.Count()) + if updateBitmapCache { + f.bitmapCache.Add(bitmapID, bm) + } return bm } @@ -337,31 +355,39 @@ func (f *Fragment) SetBit(bitmapID, profileID uint64) (changed bool, err error) return f.setBit(bitmapID, profileID) } -func (f *Fragment) setBit(bitmapID, profileID uint64) (changed bool, bool error) { - // Determine the position of the bit in the storage. +func (f *Fragment) setBit(bitmapID, profileID uint64) (changed bool, err error) { changed = false + // Determine the position of the bit in the storage. pos, err := f.pos(bitmapID, profileID) if err != nil { return false, err } // Write to storage. + if changed, err = f.storage.Add(pos); err != nil { return false, err } + // Don't update the cache if nothing changed. + if !changed { + return changed, nil + } + // Invalidate block checksum. delete(f.checksums, int(bitmapID/HashBlockSize)) - // If the number of operations exceeds the limit then snapshot. + // Increment number of operations until snapshot is required. if err := f.incrementOpN(); err != nil { return false, err } + // Get the bitmap from bitmapCache or fragment.storage. + bm := f.bitmap(bitmapID, true, true) + bm.SetBit(profileID) + // Update the cache. - if f.bitmap(bitmapID).SetBit(profileID) { - changed = true - } + f.cache.Add(bitmapID, bm.Count()) f.stats.Count("setN", 1) @@ -376,7 +402,8 @@ func (f *Fragment) ClearBit(bitmapID, profileID uint64) (bool, error) { return f.clearBit(bitmapID, profileID) } -func (f *Fragment) clearBit(bitmapID, profileID uint64) (bool, error) { +func (f *Fragment) clearBit(bitmapID, profileID uint64) (changed bool, err error) { + changed = false // Determine the position of the bit in the storage. pos, err := f.pos(bitmapID, profileID) if err != nil { @@ -384,11 +411,15 @@ func (f *Fragment) clearBit(bitmapID, profileID uint64) (bool, error) { } // Write to storage. - changed, err := f.storage.Remove(pos) - if err != nil { + if changed, err = f.storage.Remove(pos); err != nil { return false, err } + // Don't update the cache if nothing changed. + if !changed { + return changed, nil + } + // Invalidate block checksum. delete(f.checksums, int(bitmapID/HashBlockSize)) @@ -397,10 +428,12 @@ func (f *Fragment) clearBit(bitmapID, profileID uint64) (bool, error) { return false, err } + // Get the bitmap from bitmapCache or fragment.storage. + bm := f.bitmap(bitmapID, true, true) + bm.ClearBit(profileID) + // Update the cache. - if f.bitmap(bitmapID).ClearBit(profileID) { - return true, nil - } + f.cache.Add(bitmapID, bm.Count()) f.stats.Count("clearN", 1) @@ -453,7 +486,8 @@ func (f *Fragment) Top(opt TopOptions) ([]Pair, error) { } // Iterate over rankings and add to results until we have enough. - results := make([]Pair, 0, opt.N) + //results := make(PairHeap, 0, opt.N) + results := &PairHeap{} for _, pair := range pairs { bitmapID, n := pair.ID, pair.Count @@ -477,7 +511,7 @@ func (f *Fragment) Top(opt TopOptions) ([]Pair, error) { } // The initial n pairs should simply be added to the results. - if opt.N == 0 || len(results) < opt.N { + if opt.N == 0 || results.Len() < opt.N { // Calculate count and append. count := n if opt.Src != nil { @@ -486,26 +520,22 @@ func (f *Fragment) Top(opt TopOptions) ([]Pair, error) { if count == 0 { continue } - results = append(results, Pair{Key: bitmapID, Count: count}) + heap.Push(results, Pair{Key: bitmapID, Count: count}) // If we reach the requested number of pairs and we are not computing // intersections then simply exit. If we are intersecting then sort // and then only keep pairs that are higher than the lowest count. - if opt.N > 0 && len(results) == opt.N { + if opt.N > 0 && results.Len() == opt.N { if opt.Src == nil { break } - sort.Sort(Pairs(results)) } continue } // Retrieve the lowest count we have. // If it's too low then don't try finding anymore pairs. - threshold := results[len(results)-1].Count - if threshold < MinThreshold { - break - } + threshold := results.Pairs[0].Count // If the bitmap doesn't have enough bits set before the intersection // then we can assume that any remaing bitmaps also have a count too low. @@ -515,22 +545,22 @@ func (f *Fragment) Top(opt TopOptions) ([]Pair, error) { // Calculate the intersecting bit count and skip if it's below our // last bitmap in our current result set. + count := opt.Src.IntersectionCount(f.Bitmap(bitmapID)) if count < threshold { continue } - // Swap out the last pair for this new count. - results[len(results)-1] = Pair{Key: bitmapID, Count: count} - - // If it's count is also higher than the second to last item then resort. - if len(results) >= 2 && count > results[len(results)-2].Count { - sort.Sort(Pairs(results)) - } + heap.Push(results, Pair{Key: bitmapID, Count: count}) } - - sort.Sort(Pairs(results)) - return results, nil + r := make(Pairs, results.Len(), results.Len()) + x := results.Len() + i := 1 + for results.Len() > 0 { + r[x-i] = heap.Pop(results).(Pair) + i++ + } + return r, nil } func (f *Fragment) topBitmapPairs(bitmapIDs []uint64) []BitmapPair { @@ -543,23 +573,27 @@ func (f *Fragment) topBitmapPairs(bitmapIDs []uint64) []BitmapPair { } // Otherwise retrieve specific bitmaps. - pairs := make([]BitmapPair, len(bitmapIDs)) - for i, bitmapID := range bitmapIDs { + pairs := make([]BitmapPair, 0, len(bitmapIDs)) + for _, bitmapID := range bitmapIDs { // Look up cache first, if available. if n := f.cache.Get(bitmapID); n > 0 { - pairs[i] = BitmapPair{ + pairs = append(pairs, BitmapPair{ ID: bitmapID, Count: n, - } + }) continue } - // Otherwise load from storage. - pairs[i] = BitmapPair{ - ID: bitmapID, - Count: f.Bitmap(bitmapID).Count(), + bm := f.Bitmap(bitmapID) + if bm.Count() > 0 { + // Otherwise load from storage. + pairs = append(pairs, BitmapPair{ + ID: bitmapID, + Count: bm.Count(), + }) } } + sort.Sort(BitmapPairs(pairs)) return pairs } @@ -838,7 +872,6 @@ func (f *Fragment) Import(bitmapIDs, profileIDs []uint64) error { // Process every bit. // If an error occurs then reopen the storage. lastID := uint64(0) - bmCounter := 0 if err := func() error { set := make(map[uint64]struct{}) for i := range bitmapIDs { @@ -851,7 +884,7 @@ func (f *Fragment) Import(bitmapIDs, profileIDs []uint64) error { } // Write to storage. - changed, err := f.storage.Add(pos) + _, err = f.storage.Add(pos) if err != nil { return err } @@ -863,9 +896,6 @@ func (f *Fragment) Import(bitmapIDs, profileIDs []uint64) error { lastID = bitmapID set[bitmapID] = struct{}{} } - if changed { - bmCounter += 1 - } // Invalidate block checksum. delete(f.checksums, int(bitmapID/HashBlockSize)) @@ -873,7 +903,10 @@ func (f *Fragment) Import(bitmapIDs, profileIDs []uint64) error { // Update cache counts for all bitmaps. for bitmapID := range set { - f.cache.Add(bitmapID, f.bitmap(bitmapID).Count()) + // Import should ALWAYS have bitmap() load a new bm from fragment.storage + // because the bitmap that's in bitmapCache hasn't been updated with + // this import's data. + f.cache.BulkAdd(bitmapID, f.bitmap(bitmapID, false, false).Count()) } f.cache.Invalidate() @@ -912,10 +945,15 @@ func (f *Fragment) Snapshot() error { defer f.mu.Unlock() return f.snapshot() } +func track(start time.Time, name string, logger *log.Logger) { + elapsed := time.Since(start) + logger.Printf("%s took %s", name, elapsed) +} func (f *Fragment) snapshot() error { logger := f.logger() logger.Printf("fragment: snapshotting %s/%s/%d", f.db, f.frame, f.slice) + defer track(time.Now(), fmt.Sprintf("fragment: snapshot complete %s/%s/%d", f.db, f.frame, f.slice), logger) // Create a temporary file to snapshot to. snapshotPath := f.path + SnapshotExt @@ -1216,7 +1254,6 @@ func (s *FragmentSyncer) SyncFragment() error { // Determine replica set. nodes := s.Cluster.FragmentNodes(s.Fragment.DB(), s.Fragment.Slice()) if len(nodes) == 1 { - //fmt.Println("no place to replicate", s.Fragment.DB(), s.Fragment.Frame(), s.Fragment.Slice()) return nil } diff --git a/handler.go b/handler.go index 381cfeab7..1ccdf3711 100644 --- a/handler.go +++ b/handler.go @@ -205,7 +205,10 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { http.NotFound(w, r) } - h.logger().Printf("%s %s %.03fs", r.Method, r.URL.String(), time.Since(t).Seconds()) + dif := time.Since(t).Seconds() + if dif > 90 { + h.logger().Printf("%s %s %.03fs", r.Method, r.URL.String(), dif) + } } // handleGetSchema handles GET /schema requests. diff --git a/roaring/roaring.go b/roaring/roaring.go index 2a0dc57d1..e74739311 100644 --- a/roaring/roaring.go +++ b/roaring/roaring.go @@ -444,33 +444,56 @@ func (b *Bitmap) removeEmptyContainers() { i++ } } +func (b *Bitmap) countEmptyContainers() int { + result := 0 + for i := 0; i < len(b.containers); { + c := b.containers[i] + + if c.n == 0 { + result++ + } + i++ + } + return result +} // WriteTo writes b to w. func (b *Bitmap) WriteTo(w io.Writer) (n int64, err error) { // Remove empty containers before persisting. - b.removeEmptyContainers() + //b.removeEmptyContainers() + containerCount := len(b.keys) - b.countEmptyContainers() // Build header before writing individual container blocks. - buf := make([]byte, headerSize+(len(b.keys)*(4+8+4))) + buf := make([]byte, headerSize+(containerCount*(4+8+4))) binary.LittleEndian.PutUint32(buf[0:], cookie) - binary.LittleEndian.PutUint32(buf[4:], uint32(len(b.keys))) - + binary.LittleEndian.PutUint32(buf[4:], uint32(containerCount)) + empty := 0 // Encode keys and cardinality. for i, key := range b.keys { c := b.containers[i] // Verify container count before writing. - count := c.count() - assert(c.count() == c.n, "cannot write container count, mismatch: count=%d, n=%d", count, c.n) - - binary.LittleEndian.PutUint64(buf[headerSize+i*12:], uint64(key)) - binary.LittleEndian.PutUint32(buf[headerSize+i*12+8:], uint32(c.n-1)) + // TODO: instead of commenting this out, we need to make it a configuration option + //count := c.count() + //assert(c.count() == c.n, "cannot write container count, mismatch: count=%d, n=%d", count, c.n) + if c.n > 0 { + binary.LittleEndian.PutUint64(buf[headerSize+(i-empty)*12:], uint64(key)) + binary.LittleEndian.PutUint32(buf[headerSize+(i-empty)*12+8:], uint32(c.n-1)) + } else { + empty++ + } } // Write the offset for each container block. offset := uint32(len(buf)) + empty = 0 for i, c := range b.containers { - binary.LittleEndian.PutUint32(buf[headerSize+(len(b.keys)*12)+(i*4):], uint32(offset)) + + if c.n > 0 { + binary.LittleEndian.PutUint32(buf[headerSize+(containerCount*12)+((i-empty)*4):], uint32(offset)) + } else { + empty++ + } offset += uint32(c.size()) } @@ -483,10 +506,12 @@ func (b *Bitmap) WriteTo(w io.Writer) (n int64, err error) { // Write each container block. for _, c := range b.containers { - nn, err := c.WriteTo(w) - n += nn - if err != nil { - return n, err + if c.n > 0 { + nn, err := c.WriteTo(w) + n += nn + if err != nil { + return n, err + } } } @@ -532,9 +557,10 @@ func (b *Bitmap) UnmarshalBinary(data []byte) error { c := b.containers[i] if c.n <= ArrayMaxSize { c.array = (*[0xFFFFFFF]uint32)(unsafe.Pointer(&data[offset]))[:c.n] - for _, v := range c.array { - assert(lowbits(uint64(v)) == v, "array value out of range: %d", v) - } + // TODO: instead of commenting this out, we need to make it a configuration option + //for _, v := range c.array { + // assert(lowbits(uint64(v)) == v, "array value out of range: %d", v) + //} opsOffset = int(offset) + len(c.array)*4 } else { c.bitmap = (*[0xFFFFFFF]uint64)(unsafe.Pointer(&data[offset]))[:bitmapN] @@ -542,8 +568,9 @@ func (b *Bitmap) UnmarshalBinary(data []byte) error { } // Verify container count on load. - count := c.count() - assert(c.count() == c.n, "container count mismatch: count=%d, n=%d", count, c.n) + // TODO: instead of commenting this out, we need to make it a configuration option + //count := c.count() + //assert(c.count() == c.n, "container count mismatch: count=%d, n=%d", count, c.n) } // Read ops log until the end of the file. @@ -1074,9 +1101,10 @@ func (c *container) arrayWriteTo(w io.Writer) (n int64, err error) { } // Verify all elements are valid. - for _, v := range c.array { - assert(lowbits(uint64(v)) == v, "cannot write array value out of range: %d", v) - } + // TODO: instead of commenting this out, we need to make it a configuration option + //for _, v := range c.array { + // assert(lowbits(uint64(v)) == v, "cannot write array value out of range: %d", v) + //} nn, err := w.Write((*[0xFFFFFFF]byte)(unsafe.Pointer(&c.array[0]))[:4*c.n]) return int64(nn), err