From 8f4d7d50c36a99a8933ee79c9f0d3e9bc8dcc7bc Mon Sep 17 00:00:00 2001 From: Ben Johnson Date: Tue, 12 Apr 2016 19:17:47 -0600 Subject: [PATCH] active anti-entropy This commit adds active anti-entropy via a merkle tree structure. --- Godeps/Godeps.json | 4 +- client.go | 107 +++++++++++++ client_test.go | 39 +++++ db.go | 42 ++++++ fragment.go | 275 ++++++++++++++++++++++++++++++++++ fragment_test.go | 118 +++++++++++++++ frame.go | 11 ++ handler.go | 116 +++++++++++--- handler_test.go | 8 + index.go | 72 +++++++++ index_test.go | 91 +++++++++++ internal/internal.pb.go | 325 ++++++++++++++++++++++++++++++---------- internal/internal.proto | 15 ++ popcnt_asm.go | 7 +- popcnt_generic.go | 2 +- roaring/roaring.go | 117 +++++++++++++-- roaring/roaring_test.go | 63 ++++++++ 17 files changed, 1289 insertions(+), 123 deletions(-) diff --git a/Godeps/Godeps.json b/Godeps/Godeps.json index eb98c33b6..80b4f4dc2 100644 --- a/Godeps/Godeps.json +++ b/Godeps/Godeps.json @@ -21,8 +21,8 @@ }, { "ImportPath": "github.com/gogo/protobuf/proto", - "Comment": "v0.1-125-g82d16f7", - "Rev": "82d16f734d6d871204a3feb1a73cb220cc92574c" + "Comment": "v0.2-9-g4365f75", + "Rev": "4365f750fe246471f2a03ef5da5231c3565c5628" }, { "ImportPath": "github.com/golang/groupcache/lru", diff --git a/client.go b/client.go index 22fea42bd..15f4b401b 100644 --- a/client.go +++ b/client.go @@ -67,6 +67,29 @@ func (c *Client) SliceN() (uint64, error) { return rsp.SliceMax, nil } +// Schema returns all database and frame schema information. +func (c *Client) Schema() ([]*DBInfo, error) { + // Execute request against the host. + u := url.URL{ + Scheme: "http", + Host: c.host, + Path: "/schema", + } + resp, err := c.HTTPClient.Get(u.String()) + if err != nil { + return nil, err + } + defer resp.Body.Close() + + var rsp getSchemaResponse + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("http: status=%d", resp.StatusCode) + } else if err := json.NewDecoder(resp.Body).Decode(&rsp); err != nil { + return nil, fmt.Errorf("json decode: %s", err) + } + return rsp.DBs, nil +} + // SliceNodes returns a list of nodes that own a slice. func (c *Client) SliceNodes(slice uint64) ([]*Node, error) { // Execute request against the host. @@ -443,6 +466,90 @@ func (c *Client) RestoreFrame(host, db, frame string) error { return nil } +// FragmentBlocks returns a list of block checksums for a fragment on a host. +// Only returns blocks which contain data. +func (c *Client) FragmentBlocks(db, frame string, slice uint64) ([]FragmentBlock, error) { + u := url.URL{ + Scheme: "http", + Host: c.host, + Path: "/fragment/blocks", + RawQuery: url.Values{ + "db": {db}, + "frame": {frame}, + "slice": {strconv.FormatUint(slice, 10)}, + }.Encode(), + } + resp, err := c.HTTPClient.Get(u.String()) + if err != nil { + return nil, err + } + defer resp.Body.Close() + + // Return error if status is not OK. + switch resp.StatusCode { + case http.StatusOK: // ok + case http.StatusNotFound: + return nil, ErrFragmentNotFound + default: + return nil, fmt.Errorf("unexpected status: code=%d", resp.StatusCode) + } + + // Decode response object. + var rsp getFragmentBlocksResponse + if err := json.NewDecoder(resp.Body).Decode(&rsp); err != nil { + return nil, err + } + return rsp.Blocks, nil +} + +// MergeBlock sends data for a block for the remote host to merge. +// +// The remote host returns a list of bitmap/profile bit pairs for each bit +// that was set on the remote host but not sent by the client. These bits +// can be used by the caller to synchronize the local index. +func (c *Client) MergeBlock(db, frame string, slice uint64, block int, bitmapIDs, profileIDs []uint64) ([]uint64, []uint64, error) { + buf, err := proto.Marshal(&internal.MergeBlockRequest{ + DB: proto.String(db), + Frame: proto.String(frame), + Slice: proto.Uint64(slice), + Block: proto.Uint64(uint64(block)), + BitmapIDs: bitmapIDs, + ProfileIDs: profileIDs, + }) + if err != nil { + return nil, nil, err + } + + u := url.URL{Scheme: "http", Host: c.host, Path: "/fragment/block"} + req, err := http.NewRequest("PATCH", u.String(), bytes.NewReader(buf)) + if err != nil { + return nil, nil, err + } + req.Header.Set("Content-Type", "application/protobuf") + req.Header.Set("Content-Length", strconv.Itoa(len(buf))) + req.Header.Set("Accept", "application/protobuf") + + resp, err := c.HTTPClient.Do(req) + if err != nil { + return nil, nil, err + } + defer resp.Body.Close() + + // Return error if status is not OK. + if resp.StatusCode != http.StatusOK { + return nil, nil, fmt.Errorf("unexpected status: code=%d", resp.StatusCode) + } + + // Decode response object. + var rsp internal.MergeBlockResponse + if body, err := ioutil.ReadAll(resp.Body); err != nil { + return nil, nil, err + } else if err := proto.Unmarshal(body, &rsp); err != nil { + return nil, nil, err + } + return rsp.BitmapIDs, rsp.ProfileIDs, nil +} + // Bit represents the location of a single bit. type Bit struct { BitmapID uint64 diff --git a/client_test.go b/client_test.go index 3a4f24ab1..ee5ae7c25 100644 --- a/client_test.go +++ b/client_test.go @@ -5,6 +5,7 @@ import ( "reflect" "testing" + "github.com/davecgh/go-spew/spew" "github.com/umbel/pilosa" ) @@ -85,6 +86,44 @@ func TestClient_BackupRestore(t *testing.T) { } } +// Ensure client can retrieve a list of all checksums for blocks in a fragment. +func TestClient_FragmentBlocks(t *testing.T) { + idx := MustOpenIndex() + defer idx.Close() + + // Set two bits on blocks 0 & 3. + idx.MustCreateFragmentIfNotExists("d", "f", 0).SetBit(0, 1, nil, 0) + idx.MustCreateFragmentIfNotExists("d", "f", 0).SetBit(pilosa.HashBlockSize*3, 100, nil, 0) + + // Set a bit on a different slice. + idx.MustCreateFragmentIfNotExists("d", "f", 1).SetBit(0, 1, nil, 0) + + s := NewServer() + defer s.Close() + s.Handler.Host = s.Host() + s.Handler.Cluster = NewCluster(1) + s.Handler.Cluster.Nodes[0].Host = s.Host() + s.Handler.Index = idx.Index + + // Retrieve blocks. + c := MustNewClient(s.Host()) + blocks, err := c.FragmentBlocks("d", "f", 0) + if err != nil { + t.Fatal(err) + } else if len(blocks) != 2 { + t.Fatalf("unexpected blocks: %s", spew.Sdump(blocks)) + } else if blocks[0].ID != 0 { + t.Fatalf("unexpected block id(0): %d", blocks[0].ID) + } else if blocks[1].ID != 3 { + t.Fatalf("unexpected block id(1): %d", blocks[1].ID) + } + + // Verify data matches local blocks. + if a := idx.Fragment("d", "f", 0).Blocks(); !reflect.DeepEqual(a, blocks) { + t.Fatalf("blocks mismatch:\n\nexp=%s\n\ngot=%s\n\n", spew.Sdump(a), spew.Sdump(blocks)) + } +} + // Client represents a test wrapper for pilosa.Client. type Client struct { *pilosa.Client diff --git a/db.go b/db.go index 78d6d0c71..f4b2752d6 100644 --- a/db.go +++ b/db.go @@ -178,3 +178,45 @@ type dbSlice []*DB func (p dbSlice) Swap(i, j int) { p[i], p[j] = p[j], p[i] } func (p dbSlice) Len() int { return len(p) } func (p dbSlice) Less(i, j int) bool { return p[i].Name() < p[j].Name() } + +// DBInfo represents schema information for a database. +type DBInfo struct { + Name string `json:"name"` + Frames []*FrameInfo `json:"frames"` +} + +type dbInfoSlice []*DBInfo + +func (p dbInfoSlice) Swap(i, j int) { p[i], p[j] = p[j], p[i] } +func (p dbInfoSlice) Len() int { return len(p) } +func (p dbInfoSlice) Less(i, j int) bool { return p[i].Name < p[j].Name } + +// MergeSchemas combines databases and frames from a and b into one schema. +func MergeSchemas(a, b []*DBInfo) []*DBInfo { + // Generate a map from both schemas. + m := make(map[string]map[string]struct{}) + for _, dbs := range [][]*DBInfo{a, b} { + for _, db := range dbs { + if m[db.Name] == nil { + m[db.Name] = make(map[string]struct{}) + } + for _, frame := range db.Frames { + m[db.Name][frame.Name] = struct{}{} + } + } + } + + // Generate new schema from map. + dbs := make([]*DBInfo, 0, len(m)) + for db, frames := range m { + di := &DBInfo{Name: db} + for frame := range frames { + di.Frames = append(di.Frames, &FrameInfo{Name: frame}) + } + sort.Sort(frameInfoSlice(di.Frames)) + dbs = append(dbs, di) + } + sort.Sort(dbInfoSlice(dbs)) + + return dbs +} diff --git a/fragment.go b/fragment.go index 5e54fb6ac..83095d021 100644 --- a/fragment.go +++ b/fragment.go @@ -2,6 +2,9 @@ package pilosa import ( "archive/tar" + "bytes" + "crypto/sha1" + "encoding/binary" "errors" "fmt" "io" @@ -33,7 +36,12 @@ const ( // CacheExt is the file extension for persisted cache ids. CacheExt = ".cache" + // MinThreshold is the lowest count to use in a Top-N operation when + // looking for additional bitmap/count pairs. MinThreshold = 10 + + // HashBlockSize is the number of bitmaps in a merkle hash block. + HashBlockSize = 100 ) const ( @@ -63,6 +71,9 @@ type Fragment struct { // Bitmap cache. cache Cache + // Cached checksums for each block. + checksums map[int][]byte + // Close management wg sync.WaitGroup closing chan struct{} @@ -133,6 +144,9 @@ func (f *Fragment) Open() error { return err } + // Clear checksums. + f.checksums = make(map[int][]byte) + // Periodically flush cache. f.wg.Add(1) go func() { defer f.wg.Done(); f.monitorCacheFlush() }() @@ -264,6 +278,9 @@ func (f *Fragment) close() error { f.logger().Printf("fragment: error closing storage: err=%s, path=%s", err, f.path) } + // Remove checksums. + f.checksums = nil + return nil } @@ -351,6 +368,9 @@ func (f *Fragment) setBit(bitmapID, profileID uint64) (changed bool, bool error) return false, err } + // Invalidate block checksum. + delete(f.checksums, int(bitmapID/HashBlockSize)) + // If the number of operations exceeds the limit then snapshot. if err := f.incrementOpN(); err != nil { return false, err @@ -393,6 +413,9 @@ func (f *Fragment) ClearBit(bitmapID, profileID uint64) (bool, error) { return false, err } + // Invalidate block checksum. + delete(f.checksums, int(bitmapID/HashBlockSize)) + // Increment number of operations until snapshot is required. if err := f.incrementOpN(); err != nil { return false, err @@ -568,6 +591,170 @@ func (f *Fragment) Range(bitmapID uint64, start, end time.Time) *Bitmap { return bm } +// Checksum returns a checksum for the entire fragment. +// If two fragments have the same checksum then they have the same data. +func (f *Fragment) Checksum() []byte { + h := sha1.New() + for i, blockN := 0, f.BlockN(); i < blockN; i++ { + h.Write(f.BlockChecksum(i)) + } + return h.Sum(nil) +} + +// BlockN returns the number of blocks in the fragment. +func (f *Fragment) BlockN() int { + f.mu.Lock() + defer f.mu.Unlock() + return int(f.storage.Max() / (HashBlockSize * SliceWidth)) +} + +// BlockChecksum returns the checksum for a single block in the fragment. +// Returns nil if there is no data for the block. +func (f *Fragment) BlockChecksum(i int) []byte { + f.mu.Lock() + defer f.mu.Unlock() + + // Use the cached checksum, if available. + if chksum, ok := f.checksums[i]; ok { + return chksum + } + + // Otherwise calculate the checksum from the data on disk. + h := sha1.New() + var written bool + f.storage.ForEachRange(uint64(i)*HashBlockSize*SliceWidth, (uint64(i)+1)*HashBlockSize*SliceWidth, func(i uint64) { + // Write value to the hash. + var buf [8]byte + binary.BigEndian.PutUint64(buf[:], i) + h.Write(buf[:]) + + // Mark the block has having data. + written = true + }) + + // If no data was written then return a nil checksum. + if !written { + return nil + } + + // Cache checksum for later use. + chksum := h.Sum(nil)[:] + f.checksums[i] = chksum + + return chksum +} + +// InvalidateChecksums clears all cached block checksums. +func (f *Fragment) InvalidateChecksums() { + f.mu.Lock() + f.checksums = make(map[int][]byte) + f.mu.Unlock() +} + +// Blocks returns info for all blocks containing data. +func (f *Fragment) Blocks() []FragmentBlock { + var a []FragmentBlock + for i, blockN := 0, f.BlockN(); i <= blockN; i++ { + chksum := f.BlockChecksum(i) + if chksum == nil { + continue + } + + a = append(a, FragmentBlock{ + ID: i, + Checksum: chksum, + }) + } + return a +} + +// BlockBits returns bits in a block as bitmap & profile ID pairs. +func (f *Fragment) BlockBits(id int) (bitmapIDs, profileIDs []uint64) { + f.mu.Lock() + defer f.mu.Unlock() + + f.storage.ForEachRange(uint64(id)*HashBlockSize*SliceWidth, (uint64(id)+1)*HashBlockSize*SliceWidth, func(i uint64) { + bitmapIDs = append(bitmapIDs, i/SliceWidth) + profileIDs = append(profileIDs, i%SliceWidth) + }) + return +} + +// MergeBlock sets bit pairs on the fragment if they aren't already set. +// Bit pairs must be sorted in bitmap/profile order. Returns a set of changed bit pairs. +func (f *Fragment) MergeBlock(id int, bitmapIDs, profileIDs []uint64) (bids, pids []uint64, err error) { + // Ensure that both slices are of equal length. + if len(bitmapIDs) != len(profileIDs) { + return nil, nil, fmt.Errorf("bitmap/profile len mismatch: %d != %d", len(bitmapIDs), len(profileIDs)) + } + + f.mu.Lock() + defer f.mu.Unlock() + + // Track writes to be made separately so we aren't mutating while we iterate. + var queued [][2]uint64 + + // Only look at values within hash block range. + min := uint64(id) * HashBlockSize * SliceWidth + max := uint64(id+1) * HashBlockSize * SliceWidth + + // Buffer iterator so we can unread values. + // Add initial seek to buffer so we can just use Next() in the loop. + itr := roaring.NewBufIterator(f.storage.Iterator()) + if v := itr.Seek(min); !itr.EOF() { + itr.Unread(v) + } + + for i := 0; ; { + // Read local value into x. + // Mark as EOF if at the end of the hash block. + x := itr.Next() + xEOF := itr.EOF() + if !xEOF && x >= max { + itr.Unread(x) + x, xEOF = 0, true + } + + // Read next incoming value into y. + // Mark as EOF if at the end of the hash block. + var y uint64 + yEOF := i >= len(bitmapIDs) + if !yEOF { + y = (bitmapIDs[i] * SliceWidth) + profileIDs[i] + if y >= max { + y, yEOF = 0, true + } + } + + if xEOF && yEOF { // no more data + break + } else if yEOF || (!xEOF && x < y) { // local data + bids = append(bids, x/SliceWidth) + pids = append(pids, x%SliceWidth) + continue + } else if xEOF || (!yEOF && y < x) { // incoming data + if !xEOF { + itr.Unread(x) + } + i++ + queued = append(queued, [2]uint64{y / SliceWidth, y % SliceWidth}) + continue + } else { // local and incoming match, skip + i++ + continue + } + } + + // Set bits for queued writes. + for _, values := range queued { + if _, err := f.setBit(values[0], (f.slice*SliceWidth)+values[1]); err != nil { + return nil, nil, err + } + } + + return bids, pids, nil +} + // Import bulk imports a set of bits and then snapshots the storage. // This does not affect the fragment's cache. func (f *Fragment) Import(bitmapIDs, profileIDs []uint64) error { @@ -893,6 +1080,94 @@ func (f *Fragment) readCacheFromArchive(r io.Reader) error { return nil } +// FragmentBlock represents info about a subsection of the bitmaps in a block. +// This is used for comparing data in remote blocks for active anti-entropy. +type FragmentBlock struct { + ID int `json:"id"` + Checksum []byte `json:"checksum"` +} + +// FragmentSyncer syncs a local fragment to one on a remote host. +type FragmentSyncer struct { + Fragment *Fragment + Client *Client +} + +// SyncFragment compares checksums for the local and remote fragments and +// then merges any blocks which have differences. +func (s *FragmentSyncer) SyncFragment() error { + // Retrieve local blocks immediately to minimize read skew. + localBlocks := s.Fragment.Blocks() + + // Retrieve blocks. + remoteBlocks, err := s.Client.FragmentBlocks(s.Fragment.DB(), s.Fragment.Frame(), s.Fragment.Slice()) + if err != nil && err != ErrFragmentNotFound { + return err + } + + // Iterate over each block and merge if different. + for i, j := 0, 0; ; { + // Retrieve the next block for local & remote. + var a, b *FragmentBlock + if i < len(localBlocks) { + a = &localBlocks[i] + } + if j < len(remoteBlocks) { + b = &remoteBlocks[j] + } + + // Determine the next block to be merged. + var block *FragmentBlock + if a == nil && b == nil { + break + } else if a != nil && b == nil { // only local blocks remain + block, i = a, i+1 + } else if a == nil && b != nil { // only remote blocks remain + block, j = b, j+1 + } else if a.ID < b.ID { // lower local block id + block, i = a, i+1 + } else if a.ID > b.ID { // lower remote block id + block, j = b, j+1 + } else if !bytes.Equal(a.Checksum, b.Checksum) { // checksum mismatch + block, i, j = a, i+1, j+1 + } else { // blocks equal, skip + i, j = i+1, j+1 + continue + } + + // Synchronize block. + if err := s.syncBlock(block.ID); err != nil { + return fmt.Errorf("sync block: id=%d, err=%s", block.ID, err) + } + } + + return nil +} + +// syncBlock sends and receives all bitmaps for a given block. +// The remote bitmaps are merges it the local bitmaps. +func (s *FragmentSyncer) syncBlock(id int) error { + f := s.Fragment + + // Retrieve bitmaps for block. + bitmapIDs, profileIDs := f.BlockBits(id) + + // Send bitmaps to remote. + bids, pids, err := s.Client.MergeBlock(f.DB(), f.Frame(), f.Slice(), id, bitmapIDs, profileIDs) + if err != nil { + return err + } + + // Set any local bits which are not set in remote. + for i := range bids { + if _, err := f.SetBit(bids[i], (s.Fragment.Slice()*SliceWidth)+pids[i], nil, 0); err != nil { + return err + } + } + + return nil +} + func madvise(b []byte, advice int) (err error) { _, _, e1 := syscall.Syscall(syscall.SYS_MADVISE, uintptr(unsafe.Pointer(&b[0])), uintptr(len(b)), uintptr(advice)) if e1 != 0 { diff --git a/fragment_test.go b/fragment_test.go index 5f4604ce2..54b43f7f5 100644 --- a/fragment_test.go +++ b/fragment_test.go @@ -241,6 +241,83 @@ func TestFragment_TopN_BitmapIDs(t *testing.T) { } } +// Ensure fragment can return a checksum for its blocks. +func TestFragment_Checksum(t *testing.T) { + f := MustOpenFragment("d", "f", 0) + defer f.Close() + + // Retrieve checksum and set bits. + orig := f.Checksum() + if _, err := f.SetBit(1, 200, nil, 0); err != nil { + t.Fatal(err) + } else if _, err := f.SetBit(pilosa.HashBlockSize*2, 200, nil, 0); err != nil { + t.Fatal(err) + } + + // Ensure new checksum is different. + if chksum := f.Checksum(); bytes.Equal(chksum, orig) { + t.Fatalf("expected checksum to change: %x", chksum, orig) + } +} + +// Ensure fragment can return a checksum for a given block. +func TestFragment_BlockChecksum(t *testing.T) { + f := MustOpenFragment("d", "f", 0) + defer f.Close() + + // Retrieve initial checksum. + var chksum []byte + prev := f.Checksum() + + // Set first bit. + if _, err := f.SetBit(0, 0, nil, 0); err != nil { + t.Fatal(err) + } + chksum = f.BlockChecksum(0) + if bytes.Equal(chksum, prev) { + t.Fatalf("expected checksum to change: %x", chksum) + } + prev = chksum + + // Set bit on different bitmap. + if _, err := f.SetBit(20, 0, nil, 0); err != nil { + t.Fatal(err) + } + chksum = f.BlockChecksum(0) + if bytes.Equal(chksum, prev) { + t.Fatalf("expected checksum to change: %x", chksum) + } + prev = chksum + + // Set bit on different profile. + if _, err := f.SetBit(20, 100, nil, 0); err != nil { + t.Fatal(err) + } + chksum = f.BlockChecksum(0) + if bytes.Equal(chksum, prev) { + t.Fatalf("expected checksum to change: %x", chksum) + } +} + +// Ensure fragment returns an empty checksum if no data exists for a block. +func TestFragment_BlockChecksum_Empty(t *testing.T) { + f := MustOpenFragment("d", "f", 0) + defer f.Close() + + // Set bits on a different block. + if _, err := f.SetBit(1, 200, nil, 0); err != nil { + t.Fatal(err) + } + + // Ensure checksum for block 1 is blank. + if chksum := f.BlockChecksum(0); chksum == nil { + t.Fatalf("expected chksum(0)") + } + if chksum := f.BlockChecksum(1); chksum != nil { + t.Fatalf("expected empty checksum: %x", chksum) + } +} + // Ensure a fragment's cache can be persisted between restarts. func TestFragment_LRUCache_Persistence(t *testing.T) { f := MustOpenFragment("d", "f", 0) @@ -359,6 +436,33 @@ func TestFragment_WriteTo_ReadFrom(t *testing.T) { } } +func BenchmarkFragment_BlockChecksum_Fill1(b *testing.B) { benchmarkFragmentBlockChecksum(b, 0.01) } +func BenchmarkFragment_BlockChecksum_Fill10(b *testing.B) { benchmarkFragmentBlockChecksum(b, 0.10) } +func BenchmarkFragment_BlockChecksum_Fill50(b *testing.B) { benchmarkFragmentBlockChecksum(b, 0.50) } + +func benchmarkFragmentBlockChecksum(b *testing.B, fillPercent float64) { + f := MustOpenFragment("d", "f", 0) + defer f.Close() + + // Fill fragment. + bitmapIDs, profileIDs := GenerateImportFill(pilosa.HashBlockSize, fillPercent) + if err := f.Import(bitmapIDs, profileIDs); err != nil { + b.Fatal(err) + } + + b.ResetTimer() + b.ReportAllocs() + + // Calculate block checksum. + for i := 0; i < b.N; i++ { + f.InvalidateChecksums() + + if chksum := f.BlockChecksum(0); chksum == nil { + b.Fatal("expected checksum") + } + } +} + // Fragment is a test wrapper for pilosa.Fragment. type Fragment struct { *pilosa.Fragment @@ -453,3 +557,17 @@ func (s *BitmapAttrStore) BitmapAttrs(id uint64) (map[string]interface{}, error) func (s *BitmapAttrStore) SetBitmapAttrs(id uint64, m map[string]interface{}) { s.attrs[id] = m } + +// GenerateImportFill generates a set of bits pairs that evenly fill a fragment chunk. +func GenerateImportFill(bitmapN int, pct float64) (bitmapIDs, profileIDs []uint64) { + ipct := int(pct * 100) + for i := 0; i < SliceWidth*bitmapN; i++ { + if i%100 >= ipct { + continue + } + + bitmapIDs = append(bitmapIDs, uint64(i%SliceWidth)) + profileIDs = append(profileIDs, uint64(i/SliceWidth)) + } + return +} diff --git a/frame.go b/frame.go index 386c07298..d15e2e5d1 100644 --- a/frame.go +++ b/frame.go @@ -192,3 +192,14 @@ type frameSlice []*Frame func (p frameSlice) Swap(i, j int) { p[i], p[j] = p[j], p[i] } func (p frameSlice) Len() int { return len(p) } func (p frameSlice) Less(i, j int) bool { return p[i].Name() < p[j].Name() } + +// FrameInfo represents schema information for a frame. +type FrameInfo struct { + Name string `json:"name"` +} + +type frameInfoSlice []*FrameInfo + +func (p frameInfoSlice) Swap(i, j int) { p[i], p[j] = p[j], p[i] } +func (p frameInfoSlice) Len() int { return len(p) } +func (p frameInfoSlice) Less(i, j int) bool { return p[i].Name < p[j].Name } diff --git a/handler.go b/handler.go index 2162a0075..fff3cf6f1 100644 --- a/handler.go +++ b/handler.go @@ -112,6 +112,20 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { default: http.Error(w, "method not allowed", http.StatusMethodNotAllowed) } + case "/fragment/block": + switch r.Method { + case "PATCH": + h.handlePatchFragmentBlock(w, r) + default: + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + } + case "/fragment/blocks": + switch r.Method { + case "GET": + h.handleGetFragmentBlocks(w, r) + default: + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + } case "/frame/restore": switch r.Method { case "POST": @@ -133,35 +147,15 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { // handleGetSchema handles GET /schema requests. func (h *Handler) handleGetSchema(w http.ResponseWriter, r *http.Request) { - // Construct schema based on databases and frames. - var resp getSchemaResponse - for _, db := range h.Index.DBs() { - respDB := getSchemaDB{Name: db.Name()} - for _, frame := range db.Frames() { - respDB.Frames = append(respDB.Frames, getSchemaFrame{ - Name: frame.Name(), - }) - } - resp.DBs = append(resp.DBs, respDB) - } - - // Write JSON to response. - if err := json.NewEncoder(w).Encode(resp); err != nil { + if err := json.NewEncoder(w).Encode(getSchemaResponse{ + DBs: h.Index.Schema(), + }); err != nil { h.logger().Printf("write schema response error: %s", err) } } type getSchemaResponse struct { - DBs []getSchemaDB `json:"dbs"` -} - -type getSchemaDB struct { - Name string `json:"name"` - Frames []getSchemaFrame `json:"frames"` -} - -type getSchemaFrame struct { - Name string `json:"name"` + DBs []*DBInfo `json:"dbs"` } // handlePostQuery handles /query requests. @@ -498,6 +492,80 @@ func (h *Handler) handlePostFragmentData(w http.ResponseWriter, r *http.Request) } } +// handlePatchFragmentBlock handles PATCH /fragment/block requests. +func (h *Handler) handlePatchFragmentBlock(w http.ResponseWriter, r *http.Request) { + // Read request object. + var req internal.MergeBlockRequest + if body, err := ioutil.ReadAll(r.Body); err != nil { + http.Error(w, "ready body error", http.StatusBadRequest) + return + } else if err := proto.Unmarshal(body, &req); err != nil { + http.Error(w, "unmarshal body error", http.StatusBadRequest) + return + } + + // Retrieve fragment from index. + f, err := h.Index.CreateFragmentIfNotExists(req.GetDB(), req.GetFrame(), req.GetSlice()) + if err != nil { + http.Error(w, "create fragment error", http.StatusInternalServerError) + return + } + + // Merge data into block. + bids, pids, err := f.MergeBlock(int(req.GetBlock()), req.BitmapIDs, req.ProfileIDs) + if err != nil { + w.WriteHeader(http.StatusInternalServerError) + } + + // Encode response. + buf, err := proto.Marshal(&internal.MergeBlockResponse{ + BitmapIDs: bids, + ProfileIDs: pids, + Err: proto.String(errorString(err)), + }) + if err != nil { + h.logger().Printf("merge block response encoding error: %s", err) + return + } + + // Write response. + w.Header().Set("Content-Type", "application/protobuf") + w.Header().Set("Content-Length", strconv.Itoa(len(buf))) + w.Write(buf) +} + +// handleGetFragmentBlocks handles GET /fragment/blocks requests. +func (h *Handler) handleGetFragmentBlocks(w http.ResponseWriter, r *http.Request) { + // Read slice parameter. + q := r.URL.Query() + slice, err := strconv.ParseUint(q.Get("slice"), 10, 64) + if err != nil { + http.Error(w, "slice required", http.StatusBadRequest) + return + } + + // Retrieve fragment from index. + f := h.Index.Fragment(q.Get("db"), q.Get("frame"), slice) + if f == nil { + http.Error(w, "fragment not found", http.StatusNotFound) + return + } + + // Retrieve blocks. + blocks := f.Blocks() + + // Encode response. + if err := json.NewEncoder(w).Encode(getFragmentBlocksResponse{ + Blocks: blocks, + }); err != nil { + h.logger().Printf("block response encoding error: %s", err) + } +} + +type getFragmentBlocksResponse struct { + Blocks []FragmentBlock `json:"blocks"` +} + // handlePostFrameRestore handles POST /frame/restore requests. func (h *Handler) handlePostFrameRestore(w http.ResponseWriter, r *http.Request) { q := r.URL.Query() diff --git a/handler_test.go b/handler_test.go index 520967097..889bf7534 100644 --- a/handler_test.go +++ b/handler_test.go @@ -539,6 +539,14 @@ func NewServer() *Server { Handler: NewHandler(), } s.Server = httptest.NewServer(s.Handler.Handler) + + // Update handler to use hostname. + s.Handler.Host = s.Host() + + // Create a default cluster on the handler + s.Handler.Cluster = NewCluster(1) + s.Handler.Cluster.Nodes[0].Host = s.Host() + return s } diff --git a/index.go b/index.go index afa67aa63..067321bf6 100644 --- a/index.go +++ b/index.go @@ -85,6 +85,21 @@ func (i *Index) SliceN() uint64 { return sliceN } +// Schema returns schema data for all databases and frames. +func (i *Index) Schema() []*DBInfo { + var a []*DBInfo + for _, db := range i.DBs() { + di := &DBInfo{Name: db.Name()} + for _, frame := range db.Frames() { + di.Frames = append(di.Frames, &FrameInfo{ + Name: frame.Name(), + }) + } + a = append(a, di) + } + return a +} + // DBPath returns the path where a given database is stored. func (i *Index) DBPath(name string) string { return filepath.Join(i.path, name) } @@ -176,8 +191,65 @@ func (i *Index) CreateFragmentIfNotExists(db, frame string, slice uint64) (*Frag } return f.CreateFragmentIfNotExists(slice) } + func (i *Index) SetMax(newmax uint64) { i.mu.Lock() defer i.mu.Unlock() i.remoteMax = newmax } + +// IndexSyncer is an active anti-entropy tool that compares the local index +// with a remote index based on block checksums and resolves differences. +type IndexSyncer struct { + Index *Index + Client *Client +} + +// SyncIndex compares the index on host with the local index and resolves differences. +func (s *IndexSyncer) SyncIndex() error { + // Ensure slice range is in sync first. + if newmax, err := s.Client.SliceN(); err != nil { + return err + } else if newmax > s.Index.SliceN() { + s.Index.SetMax(newmax) + } + + // Retrieve schema data from remote node. + other, err := s.Client.Schema() + if err != nil { + return err + } + + // Merge with local schema. + dbs := MergeSchemas(s.Index.Schema(), other) + + // Iterate over schema in sorted order. + sliceN := s.Index.SliceN() + for _, di := range dbs { + for _, fi := range di.Frames { + for slice := uint64(0); slice <= sliceN; slice++ { + if err := s.syncFragment(di.Name, fi.Name, slice); err != nil { + return fmt.Errorf("sync error: db=%s, frame=%s, slice=%d, err=%s", di.Name, fi.Name, slice, err) + } + } + } + } + + return nil +} + +// syncFragment synchronizes +func (s *IndexSyncer) syncFragment(db, frame string, slice uint64) error { + // Ensure fragment exists locally. + f, err := s.Index.CreateFragmentIfNotExists(db, frame, slice) + if err != nil { + return err + } + + // Sync fragments together. + fs := FragmentSyncer{Fragment: f, Client: s.Client} + if err := fs.SyncFragment(); err != nil { + return err + } + return nil +} diff --git a/index_test.go b/index_test.go index 440416d47..ab8bd592a 100644 --- a/index_test.go +++ b/index_test.go @@ -3,10 +3,101 @@ package pilosa_test import ( "io/ioutil" "os" + "reflect" + "testing" "github.com/umbel/pilosa" ) +// Ensure index can sync with a remote index. +func TestIndexSyncer_SyncIndex(t *testing.T) { + // Create a local index. + idx0 := MustOpenIndex() + defer idx0.Close() + + // Create a remote index wrapped by an HTTP + idx1 := MustOpenIndex() + defer idx1.Close() + s := NewServer() + defer s.Close() + s.Handler.Index = idx1.Index + + // Set data on the local index. + f := idx0.MustCreateFragmentIfNotExists("d", "f", 0) + if _, err := f.SetBit(0, 10, nil, 0); err != nil { + t.Fatal(err) + } else if _, err := f.SetBit(2, 20, nil, 0); err != nil { + t.Fatal(err) + } else if _, err := f.SetBit(120, 10, nil, 0); err != nil { + t.Fatal(err) + } else if _, err := f.SetBit(200, 4, nil, 0); err != nil { + t.Fatal(err) + } + + f = idx0.MustCreateFragmentIfNotExists("d", "f0", 1) + if _, err := f.SetBit(9, SliceWidth+5, nil, 0); err != nil { + t.Fatal(err) + } + + // Set data on the remote index. + f = idx1.MustCreateFragmentIfNotExists("d", "f", 0) + if _, err := f.SetBit(0, 4000, nil, 0); err != nil { + t.Fatal(err) + } else if _, err := f.SetBit(3, 10, nil, 0); err != nil { + t.Fatal(err) + } else if _, err := f.SetBit(120, 10, nil, 0); err != nil { + t.Fatal(err) + } else if _, err := f.SetBit(350, 0, nil, 0); err != nil { + t.Fatal(err) + } + + f = idx1.MustCreateFragmentIfNotExists("y", "z", 3) + if _, err := f.SetBit(10, (3*SliceWidth)+4, nil, 0); err != nil { + t.Fatal(err) + } else if _, err := f.SetBit(10, (3*SliceWidth)+5, nil, 0); err != nil { + t.Fatal(err) + } else if _, err := f.SetBit(10, (3*SliceWidth)+7, nil, 0); err != nil { + t.Fatal(err) + } + + // Set up syncer. + syncer := pilosa.IndexSyncer{ + Index: idx0.Index, + Client: MustNewClient(s.Host()).Client, + } + if err := syncer.SyncIndex(); err != nil { + t.Fatal(err) + } + + // Verify data is the same on both nodes. + for i, idx := range []*Index{idx0, idx1} { + f := idx.Fragment("d", "f", 0) + if a := f.Bitmap(0).Bits(); !reflect.DeepEqual(a, []uint64{10, 4000}) { + t.Fatalf("unexpected bits(%d/0): %+v", i, a) + } else if a := f.Bitmap(2).Bits(); !reflect.DeepEqual(a, []uint64{20}) { + t.Fatalf("unexpected bits(%d/2): %+v", i, a) + } else if a := f.Bitmap(3).Bits(); !reflect.DeepEqual(a, []uint64{10}) { + t.Fatalf("unexpected bits(%d/3): %+v", i, a) + } else if a := f.Bitmap(120).Bits(); !reflect.DeepEqual(a, []uint64{10}) { + t.Fatalf("unexpected bits(%d/120): %+v", i, a) + } else if a := f.Bitmap(200).Bits(); !reflect.DeepEqual(a, []uint64{4}) { + t.Fatalf("unexpected bits(%d/200): %+v", i, a) + } else if a := f.Bitmap(350).Bits(); !reflect.DeepEqual(a, []uint64{0}) { + t.Fatalf("unexpected bits(%d/350): %+v", i, a) + } + + f = idx.Fragment("d", "f0", 1) + if a := f.Bitmap(9).Bits(); !reflect.DeepEqual(a, []uint64{SliceWidth + 5}) { + t.Fatalf("unexpected bits(%d/d/f0): %+v", i, a) + } + + f = idx.Fragment("y", "z", 3) + if a := f.Bitmap(10).Bits(); !reflect.DeepEqual(a, []uint64{(3 * SliceWidth) + 4, (3 * SliceWidth) + 5, (3 * SliceWidth) + 7}) { + t.Fatalf("unexpected bits(%d/y/z): %+v", i, a) + } + } +} + // Index is a test wrapper for pilosa.Index. type Index struct { *pilosa.Index diff --git a/internal/internal.pb.go b/internal/internal.pb.go index 475fa879e..638afb836 100644 --- a/internal/internal.pb.go +++ b/internal/internal.pb.go @@ -21,27 +21,36 @@ It has these top-level messages: QueryResult ImportRequest ImportResponse + MergeBlockRequest + MergeBlockResponse Cache SliceMaxResponse */ package internal import proto "github.com/gogo/protobuf/proto" +import fmt "fmt" import math "math" // Reference imports to suppress errors if they are not otherwise used. var _ = proto.Marshal +var _ = fmt.Errorf var _ = math.Inf +// This is a compile-time assertion to ensure that this generated file +// is compatible with the proto package it is being compiled against. +const _ = proto.GoGoProtoPackageIsVersion1 + type Bitmap struct { - Chunks []*Chunk `protobuf:"bytes,1,rep" json:"Chunks,omitempty"` - Attrs []*Attr `protobuf:"bytes,2,rep" json:"Attrs,omitempty"` + Chunks []*Chunk `protobuf:"bytes,1,rep,name=Chunks" json:"Chunks,omitempty"` + Attrs []*Attr `protobuf:"bytes,2,rep,name=Attrs" json:"Attrs,omitempty"` XXX_unrecognized []byte `json:"-"` } -func (m *Bitmap) Reset() { *m = Bitmap{} } -func (m *Bitmap) String() string { return proto.CompactTextString(m) } -func (*Bitmap) ProtoMessage() {} +func (m *Bitmap) Reset() { *m = Bitmap{} } +func (m *Bitmap) String() string { return proto.CompactTextString(m) } +func (*Bitmap) ProtoMessage() {} +func (*Bitmap) Descriptor() ([]byte, []int) { return fileDescriptorInternal, []int{0} } func (m *Bitmap) GetChunks() []*Chunk { if m != nil { @@ -58,14 +67,15 @@ func (m *Bitmap) GetAttrs() []*Attr { } type Chunk struct { - Key *uint64 `protobuf:"varint,1,req" json:"Key,omitempty"` - Value []uint64 `protobuf:"varint,2,rep" json:"Value,omitempty"` + Key *uint64 `protobuf:"varint,1,req,name=Key" json:"Key,omitempty"` + Value []uint64 `protobuf:"varint,2,rep,name=Value" json:"Value,omitempty"` XXX_unrecognized []byte `json:"-"` } -func (m *Chunk) Reset() { *m = Chunk{} } -func (m *Chunk) String() string { return proto.CompactTextString(m) } -func (*Chunk) ProtoMessage() {} +func (m *Chunk) Reset() { *m = Chunk{} } +func (m *Chunk) String() string { return proto.CompactTextString(m) } +func (*Chunk) ProtoMessage() {} +func (*Chunk) Descriptor() ([]byte, []int) { return fileDescriptorInternal, []int{1} } func (m *Chunk) GetKey() uint64 { if m != nil && m.Key != nil { @@ -82,14 +92,15 @@ func (m *Chunk) GetValue() []uint64 { } type Pair struct { - Key *uint64 `protobuf:"varint,1,req" json:"Key,omitempty"` - Count *uint64 `protobuf:"varint,2,req" json:"Count,omitempty"` + Key *uint64 `protobuf:"varint,1,req,name=Key" json:"Key,omitempty"` + Count *uint64 `protobuf:"varint,2,req,name=Count" json:"Count,omitempty"` XXX_unrecognized []byte `json:"-"` } -func (m *Pair) Reset() { *m = Pair{} } -func (m *Pair) String() string { return proto.CompactTextString(m) } -func (*Pair) ProtoMessage() {} +func (m *Pair) Reset() { *m = Pair{} } +func (m *Pair) String() string { return proto.CompactTextString(m) } +func (*Pair) ProtoMessage() {} +func (*Pair) Descriptor() ([]byte, []int) { return fileDescriptorInternal, []int{2} } func (m *Pair) GetKey() uint64 { if m != nil && m.Key != nil { @@ -106,14 +117,15 @@ func (m *Pair) GetCount() uint64 { } type Bit struct { - BitmapID *uint64 `protobuf:"varint,1,req" json:"BitmapID,omitempty"` - ProfileID *uint64 `protobuf:"varint,2,req" json:"ProfileID,omitempty"` + BitmapID *uint64 `protobuf:"varint,1,req,name=BitmapID" json:"BitmapID,omitempty"` + ProfileID *uint64 `protobuf:"varint,2,req,name=ProfileID" json:"ProfileID,omitempty"` XXX_unrecognized []byte `json:"-"` } -func (m *Bit) Reset() { *m = Bit{} } -func (m *Bit) String() string { return proto.CompactTextString(m) } -func (*Bit) ProtoMessage() {} +func (m *Bit) Reset() { *m = Bit{} } +func (m *Bit) String() string { return proto.CompactTextString(m) } +func (*Bit) ProtoMessage() {} +func (*Bit) Descriptor() ([]byte, []int) { return fileDescriptorInternal, []int{3} } func (m *Bit) GetBitmapID() uint64 { if m != nil && m.BitmapID != nil { @@ -130,14 +142,15 @@ func (m *Bit) GetProfileID() uint64 { } type Profile struct { - ID *uint64 `protobuf:"varint,1,req" json:"ID,omitempty"` - Attrs []*Attr `protobuf:"bytes,2,rep" json:"Attrs,omitempty"` + ID *uint64 `protobuf:"varint,1,req,name=ID" json:"ID,omitempty"` + Attrs []*Attr `protobuf:"bytes,2,rep,name=Attrs" json:"Attrs,omitempty"` XXX_unrecognized []byte `json:"-"` } -func (m *Profile) Reset() { *m = Profile{} } -func (m *Profile) String() string { return proto.CompactTextString(m) } -func (*Profile) ProtoMessage() {} +func (m *Profile) Reset() { *m = Profile{} } +func (m *Profile) String() string { return proto.CompactTextString(m) } +func (*Profile) ProtoMessage() {} +func (*Profile) Descriptor() ([]byte, []int) { return fileDescriptorInternal, []int{4} } func (m *Profile) GetID() uint64 { if m != nil && m.ID != nil { @@ -154,16 +167,17 @@ func (m *Profile) GetAttrs() []*Attr { } type Attr struct { - Key *string `protobuf:"bytes,1,req" json:"Key,omitempty"` - StringValue *string `protobuf:"bytes,2,opt" json:"StringValue,omitempty"` - UintValue *uint64 `protobuf:"varint,3,opt" json:"UintValue,omitempty"` - BoolValue *bool `protobuf:"varint,4,opt" json:"BoolValue,omitempty"` + Key *string `protobuf:"bytes,1,req,name=Key" json:"Key,omitempty"` + StringValue *string `protobuf:"bytes,2,opt,name=StringValue" json:"StringValue,omitempty"` + UintValue *uint64 `protobuf:"varint,3,opt,name=UintValue" json:"UintValue,omitempty"` + BoolValue *bool `protobuf:"varint,4,opt,name=BoolValue" json:"BoolValue,omitempty"` XXX_unrecognized []byte `json:"-"` } -func (m *Attr) Reset() { *m = Attr{} } -func (m *Attr) String() string { return proto.CompactTextString(m) } -func (*Attr) ProtoMessage() {} +func (m *Attr) Reset() { *m = Attr{} } +func (m *Attr) String() string { return proto.CompactTextString(m) } +func (*Attr) ProtoMessage() {} +func (*Attr) Descriptor() ([]byte, []int) { return fileDescriptorInternal, []int{5} } func (m *Attr) GetKey() string { if m != nil && m.Key != nil { @@ -194,13 +208,14 @@ func (m *Attr) GetBoolValue() bool { } type AttrMap struct { - Attrs []*Attr `protobuf:"bytes,1,rep" json:"Attrs,omitempty"` + Attrs []*Attr `protobuf:"bytes,1,rep,name=Attrs" json:"Attrs,omitempty"` XXX_unrecognized []byte `json:"-"` } -func (m *AttrMap) Reset() { *m = AttrMap{} } -func (m *AttrMap) String() string { return proto.CompactTextString(m) } -func (*AttrMap) ProtoMessage() {} +func (m *AttrMap) Reset() { *m = AttrMap{} } +func (m *AttrMap) String() string { return proto.CompactTextString(m) } +func (*AttrMap) ProtoMessage() {} +func (*AttrMap) Descriptor() ([]byte, []int) { return fileDescriptorInternal, []int{6} } func (m *AttrMap) GetAttrs() []*Attr { if m != nil { @@ -210,19 +225,20 @@ func (m *AttrMap) GetAttrs() []*Attr { } type QueryRequest struct { - DB *string `protobuf:"bytes,1,req" json:"DB,omitempty"` - Query *string `protobuf:"bytes,2,req" json:"Query,omitempty"` - Slices []uint64 `protobuf:"varint,3,rep" json:"Slices,omitempty"` - Profiles *bool `protobuf:"varint,4,opt" json:"Profiles,omitempty"` - Timestamp *int64 `protobuf:"varint,5,opt" json:"Timestamp,omitempty"` - Quantum *uint32 `protobuf:"varint,6,opt" json:"Quantum,omitempty"` - Remote *bool `protobuf:"varint,7,opt" json:"Remote,omitempty"` + DB *string `protobuf:"bytes,1,req,name=DB" json:"DB,omitempty"` + Query *string `protobuf:"bytes,2,req,name=Query" json:"Query,omitempty"` + Slices []uint64 `protobuf:"varint,3,rep,name=Slices" json:"Slices,omitempty"` + Profiles *bool `protobuf:"varint,4,opt,name=Profiles" json:"Profiles,omitempty"` + Timestamp *int64 `protobuf:"varint,5,opt,name=Timestamp" json:"Timestamp,omitempty"` + Quantum *uint32 `protobuf:"varint,6,opt,name=Quantum" json:"Quantum,omitempty"` + Remote *bool `protobuf:"varint,7,opt,name=Remote" json:"Remote,omitempty"` XXX_unrecognized []byte `json:"-"` } -func (m *QueryRequest) Reset() { *m = QueryRequest{} } -func (m *QueryRequest) String() string { return proto.CompactTextString(m) } -func (*QueryRequest) ProtoMessage() {} +func (m *QueryRequest) Reset() { *m = QueryRequest{} } +func (m *QueryRequest) String() string { return proto.CompactTextString(m) } +func (*QueryRequest) ProtoMessage() {} +func (*QueryRequest) Descriptor() ([]byte, []int) { return fileDescriptorInternal, []int{7} } func (m *QueryRequest) GetDB() string { if m != nil && m.DB != nil { @@ -274,15 +290,16 @@ func (m *QueryRequest) GetRemote() bool { } type QueryResponse struct { - Err *string `protobuf:"bytes,1,opt" json:"Err,omitempty"` - Results []*QueryResult `protobuf:"bytes,2,rep" json:"Results,omitempty"` - Profiles []*Profile `protobuf:"bytes,3,rep" json:"Profiles,omitempty"` + Err *string `protobuf:"bytes,1,opt,name=Err" json:"Err,omitempty"` + Results []*QueryResult `protobuf:"bytes,2,rep,name=Results" json:"Results,omitempty"` + Profiles []*Profile `protobuf:"bytes,3,rep,name=Profiles" json:"Profiles,omitempty"` XXX_unrecognized []byte `json:"-"` } -func (m *QueryResponse) Reset() { *m = QueryResponse{} } -func (m *QueryResponse) String() string { return proto.CompactTextString(m) } -func (*QueryResponse) ProtoMessage() {} +func (m *QueryResponse) Reset() { *m = QueryResponse{} } +func (m *QueryResponse) String() string { return proto.CompactTextString(m) } +func (*QueryResponse) ProtoMessage() {} +func (*QueryResponse) Descriptor() ([]byte, []int) { return fileDescriptorInternal, []int{8} } func (m *QueryResponse) GetErr() string { if m != nil && m.Err != nil { @@ -306,16 +323,17 @@ func (m *QueryResponse) GetProfiles() []*Profile { } type QueryResult struct { - Bitmap *Bitmap `protobuf:"bytes,1,opt" json:"Bitmap,omitempty"` - N *uint64 `protobuf:"varint,2,opt" json:"N,omitempty"` - Pairs []*Pair `protobuf:"bytes,3,rep" json:"Pairs,omitempty"` - Changed *bool `protobuf:"varint,4,opt" json:"Changed,omitempty"` + Bitmap *Bitmap `protobuf:"bytes,1,opt,name=Bitmap" json:"Bitmap,omitempty"` + N *uint64 `protobuf:"varint,2,opt,name=N" json:"N,omitempty"` + Pairs []*Pair `protobuf:"bytes,3,rep,name=Pairs" json:"Pairs,omitempty"` + Changed *bool `protobuf:"varint,4,opt,name=Changed" json:"Changed,omitempty"` XXX_unrecognized []byte `json:"-"` } -func (m *QueryResult) Reset() { *m = QueryResult{} } -func (m *QueryResult) String() string { return proto.CompactTextString(m) } -func (*QueryResult) ProtoMessage() {} +func (m *QueryResult) Reset() { *m = QueryResult{} } +func (m *QueryResult) String() string { return proto.CompactTextString(m) } +func (*QueryResult) ProtoMessage() {} +func (*QueryResult) Descriptor() ([]byte, []int) { return fileDescriptorInternal, []int{9} } func (m *QueryResult) GetBitmap() *Bitmap { if m != nil { @@ -346,17 +364,18 @@ func (m *QueryResult) GetChanged() bool { } type ImportRequest struct { - DB *string `protobuf:"bytes,1,req" json:"DB,omitempty"` - Frame *string `protobuf:"bytes,2,req" json:"Frame,omitempty"` - Slice *uint64 `protobuf:"varint,3,req" json:"Slice,omitempty"` - BitmapIDs []uint64 `protobuf:"varint,4,rep" json:"BitmapIDs,omitempty"` - ProfileIDs []uint64 `protobuf:"varint,5,rep" json:"ProfileIDs,omitempty"` + DB *string `protobuf:"bytes,1,req,name=DB" json:"DB,omitempty"` + Frame *string `protobuf:"bytes,2,req,name=Frame" json:"Frame,omitempty"` + Slice *uint64 `protobuf:"varint,3,req,name=Slice" json:"Slice,omitempty"` + BitmapIDs []uint64 `protobuf:"varint,4,rep,name=BitmapIDs" json:"BitmapIDs,omitempty"` + ProfileIDs []uint64 `protobuf:"varint,5,rep,name=ProfileIDs" json:"ProfileIDs,omitempty"` XXX_unrecognized []byte `json:"-"` } -func (m *ImportRequest) Reset() { *m = ImportRequest{} } -func (m *ImportRequest) String() string { return proto.CompactTextString(m) } -func (*ImportRequest) ProtoMessage() {} +func (m *ImportRequest) Reset() { *m = ImportRequest{} } +func (m *ImportRequest) String() string { return proto.CompactTextString(m) } +func (*ImportRequest) ProtoMessage() {} +func (*ImportRequest) Descriptor() ([]byte, []int) { return fileDescriptorInternal, []int{10} } func (m *ImportRequest) GetDB() string { if m != nil && m.DB != nil { @@ -394,13 +413,14 @@ func (m *ImportRequest) GetProfileIDs() []uint64 { } type ImportResponse struct { - Err *string `protobuf:"bytes,1,opt" json:"Err,omitempty"` + Err *string `protobuf:"bytes,1,opt,name=Err" json:"Err,omitempty"` XXX_unrecognized []byte `json:"-"` } -func (m *ImportResponse) Reset() { *m = ImportResponse{} } -func (m *ImportResponse) String() string { return proto.CompactTextString(m) } -func (*ImportResponse) ProtoMessage() {} +func (m *ImportResponse) Reset() { *m = ImportResponse{} } +func (m *ImportResponse) String() string { return proto.CompactTextString(m) } +func (*ImportResponse) ProtoMessage() {} +func (*ImportResponse) Descriptor() ([]byte, []int) { return fileDescriptorInternal, []int{11} } func (m *ImportResponse) GetErr() string { if m != nil && m.Err != nil { @@ -409,14 +429,105 @@ func (m *ImportResponse) GetErr() string { return "" } -type Cache struct { - BitmapIDs []uint64 `protobuf:"varint,1,rep" json:"BitmapIDs,omitempty"` +type MergeBlockRequest struct { + DB *string `protobuf:"bytes,1,req,name=DB" json:"DB,omitempty"` + Frame *string `protobuf:"bytes,2,req,name=Frame" json:"Frame,omitempty"` + Slice *uint64 `protobuf:"varint,3,req,name=Slice" json:"Slice,omitempty"` + Block *uint64 `protobuf:"varint,4,req,name=Block" json:"Block,omitempty"` + BitmapIDs []uint64 `protobuf:"varint,5,rep,name=BitmapIDs" json:"BitmapIDs,omitempty"` + ProfileIDs []uint64 `protobuf:"varint,6,rep,name=ProfileIDs" json:"ProfileIDs,omitempty"` XXX_unrecognized []byte `json:"-"` } -func (m *Cache) Reset() { *m = Cache{} } -func (m *Cache) String() string { return proto.CompactTextString(m) } -func (*Cache) ProtoMessage() {} +func (m *MergeBlockRequest) Reset() { *m = MergeBlockRequest{} } +func (m *MergeBlockRequest) String() string { return proto.CompactTextString(m) } +func (*MergeBlockRequest) ProtoMessage() {} +func (*MergeBlockRequest) Descriptor() ([]byte, []int) { return fileDescriptorInternal, []int{12} } + +func (m *MergeBlockRequest) GetDB() string { + if m != nil && m.DB != nil { + return *m.DB + } + return "" +} + +func (m *MergeBlockRequest) GetFrame() string { + if m != nil && m.Frame != nil { + return *m.Frame + } + return "" +} + +func (m *MergeBlockRequest) GetSlice() uint64 { + if m != nil && m.Slice != nil { + return *m.Slice + } + return 0 +} + +func (m *MergeBlockRequest) GetBlock() uint64 { + if m != nil && m.Block != nil { + return *m.Block + } + return 0 +} + +func (m *MergeBlockRequest) GetBitmapIDs() []uint64 { + if m != nil { + return m.BitmapIDs + } + return nil +} + +func (m *MergeBlockRequest) GetProfileIDs() []uint64 { + if m != nil { + return m.ProfileIDs + } + return nil +} + +type MergeBlockResponse struct { + Err *string `protobuf:"bytes,1,opt,name=Err" json:"Err,omitempty"` + BitmapIDs []uint64 `protobuf:"varint,2,rep,name=BitmapIDs" json:"BitmapIDs,omitempty"` + ProfileIDs []uint64 `protobuf:"varint,3,rep,name=ProfileIDs" json:"ProfileIDs,omitempty"` + XXX_unrecognized []byte `json:"-"` +} + +func (m *MergeBlockResponse) Reset() { *m = MergeBlockResponse{} } +func (m *MergeBlockResponse) String() string { return proto.CompactTextString(m) } +func (*MergeBlockResponse) ProtoMessage() {} +func (*MergeBlockResponse) Descriptor() ([]byte, []int) { return fileDescriptorInternal, []int{13} } + +func (m *MergeBlockResponse) GetErr() string { + if m != nil && m.Err != nil { + return *m.Err + } + return "" +} + +func (m *MergeBlockResponse) GetBitmapIDs() []uint64 { + if m != nil { + return m.BitmapIDs + } + return nil +} + +func (m *MergeBlockResponse) GetProfileIDs() []uint64 { + if m != nil { + return m.ProfileIDs + } + return nil +} + +type Cache struct { + BitmapIDs []uint64 `protobuf:"varint,1,rep,name=BitmapIDs" json:"BitmapIDs,omitempty"` + XXX_unrecognized []byte `json:"-"` +} + +func (m *Cache) Reset() { *m = Cache{} } +func (m *Cache) String() string { return proto.CompactTextString(m) } +func (*Cache) ProtoMessage() {} +func (*Cache) Descriptor() ([]byte, []int) { return fileDescriptorInternal, []int{14} } func (m *Cache) GetBitmapIDs() []uint64 { if m != nil { @@ -426,13 +537,14 @@ func (m *Cache) GetBitmapIDs() []uint64 { } type SliceMaxResponse struct { - SliceMax *uint64 `protobuf:"varint,1,req" json:"SliceMax,omitempty"` + SliceMax *uint64 `protobuf:"varint,1,req,name=SliceMax" json:"SliceMax,omitempty"` XXX_unrecognized []byte `json:"-"` } -func (m *SliceMaxResponse) Reset() { *m = SliceMaxResponse{} } -func (m *SliceMaxResponse) String() string { return proto.CompactTextString(m) } -func (*SliceMaxResponse) ProtoMessage() {} +func (m *SliceMaxResponse) Reset() { *m = SliceMaxResponse{} } +func (m *SliceMaxResponse) String() string { return proto.CompactTextString(m) } +func (*SliceMaxResponse) ProtoMessage() {} +func (*SliceMaxResponse) Descriptor() ([]byte, []int) { return fileDescriptorInternal, []int{15} } func (m *SliceMaxResponse) GetSliceMax() uint64 { if m != nil && m.SliceMax != nil { @@ -442,4 +554,57 @@ func (m *SliceMaxResponse) GetSliceMax() uint64 { } func init() { + proto.RegisterType((*Bitmap)(nil), "internal.Bitmap") + proto.RegisterType((*Chunk)(nil), "internal.Chunk") + proto.RegisterType((*Pair)(nil), "internal.Pair") + proto.RegisterType((*Bit)(nil), "internal.Bit") + proto.RegisterType((*Profile)(nil), "internal.Profile") + proto.RegisterType((*Attr)(nil), "internal.Attr") + proto.RegisterType((*AttrMap)(nil), "internal.AttrMap") + proto.RegisterType((*QueryRequest)(nil), "internal.QueryRequest") + proto.RegisterType((*QueryResponse)(nil), "internal.QueryResponse") + proto.RegisterType((*QueryResult)(nil), "internal.QueryResult") + proto.RegisterType((*ImportRequest)(nil), "internal.ImportRequest") + proto.RegisterType((*ImportResponse)(nil), "internal.ImportResponse") + proto.RegisterType((*MergeBlockRequest)(nil), "internal.MergeBlockRequest") + proto.RegisterType((*MergeBlockResponse)(nil), "internal.MergeBlockResponse") + proto.RegisterType((*Cache)(nil), "internal.Cache") + proto.RegisterType((*SliceMaxResponse)(nil), "internal.SliceMaxResponse") +} + +var fileDescriptorInternal = []byte{ + // 523 bytes of a gzipped FileDescriptorProto + 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x09, 0x6e, 0x88, 0x02, 0xff, 0x94, 0x93, 0x5b, 0x6f, 0xd3, 0x30, + 0x14, 0xc7, 0x95, 0xe6, 0xd6, 0x9e, 0xd0, 0xae, 0x35, 0x42, 0x44, 0x48, 0x13, 0x53, 0x86, 0x50, + 0xc5, 0xc3, 0x90, 0x26, 0xbe, 0x00, 0xed, 0x40, 0x9b, 0x50, 0xa7, 0x5d, 0x80, 0x67, 0xac, 0x62, + 0xda, 0xb0, 0x24, 0x0e, 0x8e, 0x23, 0xb1, 0x27, 0xbe, 0x3a, 0xc7, 0x8e, 0x9d, 0x66, 0x6a, 0x10, + 0xda, 0x53, 0xeb, 0xff, 0xb9, 0xfd, 0xfc, 0xf7, 0x09, 0x3c, 0x4f, 0x0b, 0xc9, 0x44, 0x41, 0xb3, + 0xb7, 0xf6, 0xcf, 0x49, 0x29, 0xb8, 0xe4, 0x64, 0x68, 0xcf, 0xc9, 0x39, 0x04, 0x8b, 0x54, 0xe6, + 0xb4, 0x24, 0x2f, 0x21, 0x58, 0x6e, 0xeb, 0xe2, 0xae, 0x8a, 0x9d, 0x23, 0x77, 0x1e, 0x9d, 0x1e, + 0x9c, 0xb4, 0x45, 0x5a, 0x27, 0x87, 0xe0, 0xbf, 0x97, 0x52, 0x54, 0xf1, 0x40, 0xc7, 0x27, 0xbb, + 0xb8, 0x92, 0x93, 0x63, 0xf0, 0x9b, 0xbc, 0x08, 0xdc, 0x4f, 0xec, 0x1e, 0xbb, 0x0c, 0xe6, 0x1e, + 0x19, 0x83, 0xff, 0x95, 0x66, 0x35, 0xd3, 0x45, 0x5e, 0x92, 0x80, 0x77, 0x45, 0x53, 0xb1, 0x97, + 0xb3, 0xe4, 0x75, 0x21, 0x31, 0x07, 0x8f, 0xc9, 0x1b, 0x70, 0x11, 0x89, 0x4c, 0x61, 0xd8, 0x90, + 0x5d, 0x9c, 0x99, 0xbc, 0x19, 0x8c, 0xae, 0x04, 0xff, 0x91, 0x66, 0x0c, 0xa5, 0x26, 0xf7, 0x1d, + 0x84, 0x46, 0x22, 0x00, 0x83, 0x36, 0xf3, 0x3f, 0xa8, 0x97, 0xe0, 0xa9, 0xdf, 0x2e, 0xc5, 0x88, + 0x3c, 0x85, 0xe8, 0x56, 0x8a, 0xb4, 0xd8, 0x58, 0x5e, 0x07, 0x45, 0x1c, 0xf9, 0x05, 0x6b, 0x1b, + 0xc9, 0x45, 0x49, 0x53, 0x2c, 0x38, 0xcf, 0x1a, 0xc9, 0x43, 0x69, 0x98, 0xcc, 0x21, 0x54, 0xfd, + 0x56, 0xe8, 0x62, 0x3b, 0xd9, 0xe9, 0x9d, 0xfc, 0x07, 0x9e, 0x5c, 0xd7, 0x4c, 0xdc, 0xdf, 0xb0, + 0x5f, 0x35, 0xab, 0xa4, 0x82, 0x3e, 0x5b, 0x18, 0x00, 0xb4, 0x41, 0xc7, 0xf4, 0xd5, 0x46, 0x64, + 0x02, 0xc1, 0x6d, 0x96, 0xae, 0x59, 0x85, 0x73, 0xd1, 0x3a, 0xe5, 0x87, 0xb9, 0x6a, 0xd5, 0x8c, + 0x55, 0x24, 0x9f, 0xd3, 0x1c, 0xdb, 0xd0, 0xbc, 0x8c, 0x7d, 0x94, 0x5c, 0x72, 0x00, 0xe1, 0x75, + 0x4d, 0x0b, 0x59, 0xe7, 0x71, 0x80, 0xc2, 0x58, 0x75, 0xb9, 0x61, 0x39, 0x97, 0x2c, 0x0e, 0x35, + 0x6a, 0x0a, 0x63, 0x03, 0x50, 0x95, 0xbc, 0xa8, 0x98, 0xf2, 0xe0, 0x83, 0x10, 0x88, 0xa0, 0xae, + 0xfb, 0x1a, 0x42, 0x0c, 0xd4, 0x99, 0xb4, 0xce, 0x3d, 0xdb, 0xf1, 0xdb, 0x32, 0x8c, 0x92, 0xe3, + 0x0e, 0x8b, 0xab, 0x13, 0x67, 0xbb, 0x44, 0x13, 0x49, 0x7e, 0x42, 0xd4, 0xad, 0x39, 0xb2, 0x9b, + 0xa6, 0x67, 0x45, 0xa7, 0xd3, 0x5d, 0x85, 0xd9, 0xc0, 0x11, 0x38, 0x97, 0xda, 0x77, 0xfd, 0x80, + 0x6a, 0x4f, 0x6c, 0xf7, 0x8e, 0x8d, 0x7a, 0x7d, 0xf0, 0x9a, 0xcb, 0x2d, 0x2d, 0x36, 0xec, 0xbb, + 0x79, 0x81, 0x6f, 0x30, 0xbe, 0xc8, 0x4b, 0x2e, 0xe4, 0x3f, 0x8c, 0xfd, 0x28, 0x68, 0xce, 0x8c, + 0xb1, 0x78, 0xd4, 0xc6, 0x62, 0x6f, 0xb3, 0x55, 0x76, 0xcf, 0x94, 0xb1, 0xca, 0x6a, 0xac, 0x6e, + 0x17, 0xad, 0x42, 0x67, 0xd5, 0xe6, 0x1e, 0xc2, 0xc4, 0x4e, 0xe8, 0x71, 0x2e, 0xa9, 0x60, 0xb6, + 0x62, 0x62, 0xc3, 0x16, 0x19, 0x5f, 0xdf, 0x3d, 0x1e, 0x02, 0x8f, 0xba, 0x12, 0x01, 0xf6, 0x98, + 0xfc, 0x1e, 0xa6, 0x40, 0x33, 0x9d, 0x03, 0xe9, 0x0e, 0xed, 0x7b, 0xd1, 0x07, 0x9d, 0x06, 0x3d, + 0x9d, 0xf4, 0x72, 0x25, 0x2f, 0xf0, 0x13, 0xa4, 0xeb, 0x2d, 0x7b, 0x98, 0xef, 0xe8, 0xd8, 0x2b, + 0x98, 0x6a, 0xd4, 0x15, 0xfd, 0xdd, 0xce, 0xc0, 0x65, 0xb4, 0x5a, 0xf3, 0xc9, 0xfd, 0x0d, 0x00, + 0x00, 0xff, 0xff, 0xde, 0x76, 0xa8, 0x39, 0x6c, 0x04, 0x00, 0x00, } diff --git a/internal/internal.proto b/internal/internal.proto index 127095f50..31106ab75 100644 --- a/internal/internal.proto +++ b/internal/internal.proto @@ -71,6 +71,21 @@ message ImportResponse { optional string Err = 1; } +message MergeBlockRequest { + required string DB = 1; + required string Frame = 2; + required uint64 Slice = 3; + required uint64 Block = 4; + repeated uint64 BitmapIDs = 5; + repeated uint64 ProfileIDs = 6; +} + +message MergeBlockResponse { + optional string Err = 1; + repeated uint64 BitmapIDs = 2; + repeated uint64 ProfileIDs = 3; +} + message Cache { repeated uint64 BitmapIDs = 1; } diff --git a/popcnt_asm.go b/popcnt_asm.go index b34613956..f12e6f254 100644 --- a/popcnt_asm.go +++ b/popcnt_asm.go @@ -29,7 +29,7 @@ func popcntOrSliceAsm(s, m []uint64) uint64 func popcntXorSliceAsm(s, m []uint64) uint64 //go:noescape -func popcntAsm(x uint64)uint64 +func popcntAsm(x uint64) uint64 func popcntSlice(s []uint64) uint64 { if useAsm { @@ -66,10 +66,9 @@ func popcntXorSlice(s, m []uint64) uint64 { return popcntXorSliceGo(s, m) } -func popcnt(x uint64) uint64{ - if useAsm { +func popcnt(x uint64) uint64 { + if useAsm { return popcntAsm(x) } return popcntGo(x) } - diff --git a/popcnt_generic.go b/popcnt_generic.go index 01d381385..322d07584 100644 --- a/popcnt_generic.go +++ b/popcnt_generic.go @@ -7,4 +7,4 @@ func popcntMaskSlice(s, m []uint64) uint64 { return popcntMaskSliceGo(s, m) } func popcntAndSlice(s, m []uint64) uint64 { return popcntAndSliceGo(s, m) } func popcntOrSlice(s, m []uint64) uint64 { return popcntOrSliceGo(s, m) } func popcntXorSlice(s, m []uint64) uint64 { return popcntXorSliceGo(s, m) } -func popcnt(s uint64) uint64 { return popcntGo(s)} +func popcnt(s uint64) uint64 { return popcntGo(s) } diff --git a/roaring/roaring.go b/roaring/roaring.go index a87d4f5c1..f6c3cf742 100644 --- a/roaring/roaring.go +++ b/roaring/roaring.go @@ -114,10 +114,22 @@ func (b *Bitmap) remove(v uint64) bool { return b.containers[i].remove(lowbits(v)) } +// Max returns the highest value in the bitmap. +// Returns zero if the bitmap is empty. +func (b *Bitmap) Max() uint64 { + if len(b.keys) == 0 { + return 0 + } + + hb := b.keys[len(b.keys)-1] + lb := b.containers[len(b.containers)-1].max() + return uint64(hb)<<16 | uint64(lb) +} + // Slice returns a slice of all integers in the bitmap. func (b *Bitmap) Slice() []uint64 { var a []uint64 - itr := b.iterator() + itr := b.Iterator() for v := itr.Seek(0); !itr.EOF(); v = itr.Next() { a = append(a, v) } @@ -127,7 +139,7 @@ func (b *Bitmap) Slice() []uint64 { // SliceRange returns a slice of integers between [start, end). func (b *Bitmap) SliceRange(start, end uint64) []uint64 { var a []uint64 - itr := b.iterator() + itr := b.Iterator() for v := itr.Seek(start); !itr.EOF() && v < end; v = itr.Next() { a = append(a, v) } @@ -136,7 +148,7 @@ func (b *Bitmap) SliceRange(start, end uint64) []uint64 { // ForEach executes fn for each value in the bitmap. func (b *Bitmap) ForEach(fn func(uint64)) { - itr := b.iterator() + itr := b.Iterator() for v := itr.Seek(0); !itr.EOF(); v = itr.Next() { fn(v) } @@ -144,7 +156,7 @@ func (b *Bitmap) ForEach(fn func(uint64)) { // ForEachRange executes fn for each value in the bitmap between [start, end). func (b *Bitmap) ForEachRange(start, end uint64, fn func(uint64)) { - itr := b.iterator() + itr := b.Iterator() for v := itr.Seek(start); !itr.EOF() && v < end; v = itr.Next() { fn(v) } @@ -291,20 +303,20 @@ func (b *Bitmap) writeOp(op *op) error { return nil } -// iterator returns an iterator for the bitmap. -func (b *Bitmap) iterator() *iterator { return &iterator{bitmap: b} } +// Iterator returns a new iterator for the bitmap. +func (b *Bitmap) Iterator() *Iterator { return &Iterator{bitmap: b} } -// iterator represents an iterator over a Bitmap. -type iterator struct { +// Iterator represents an iterator over a Bitmap. +type Iterator struct { bitmap *Bitmap i, j int } // EOF returns true if the iterator is at the end of the bitmap. -func (itr *iterator) EOF() bool { return itr.i >= len(itr.bitmap.containers) } +func (itr *Iterator) EOF() bool { return itr.i >= len(itr.bitmap.containers) } // Seek moves to the first value equal to or greater than v. -func (itr *iterator) Seek(seek uint64) uint64 { +func (itr *Iterator) Seek(seek uint64) uint64 { // Move to the correct container. itr.i = search64(itr.bitmap.keys, highbits(seek)) if itr.i < 0 { @@ -337,7 +349,7 @@ func (itr *iterator) Seek(seek uint64) uint64 { } // Next returns the next value in the bitmap. -func (itr *iterator) Next() uint64 { +func (itr *Iterator) Next() uint64 { // Iterate over containers until we find the next value or EOF. for { if itr.EOF() { @@ -380,7 +392,7 @@ func (itr *iterator) Next() uint64 { } // peek returns the current value. -func (itr *iterator) peek() uint64 { +func (itr *Iterator) peek() uint64 { key := itr.bitmap.keys[itr.i] c := itr.bitmap.containers[itr.i] if c.isArray() { @@ -389,6 +401,56 @@ func (itr *iterator) peek() uint64 { return uint64(key)<<16 | uint64(itr.j) } +// BufIterator wraps an iterator to provide the ability to unread values. +type BufIterator struct { + buf struct { + v uint64 + full bool + } + itr *Iterator +} + +// NewBufIterator returns a buffered iterator that wraps itr. +func NewBufIterator(itr *Iterator) *BufIterator { + return &BufIterator{itr: itr} +} + +// EOF returns true if the iterator is at the end of the bitmap. +func (itr *BufIterator) EOF() bool { + if itr.buf.full { + return false + } + return itr.itr.EOF() +} + +// Seek moves to the first value equal to or greater than v. +func (itr *BufIterator) Seek(seek uint64) uint64 { + itr.buf.v = 0 + itr.buf.full = false + return itr.itr.Seek(seek) +} + +// Next returns the next value in the bitmap. +// If a value has been buffered then it is returned and the buffer is cleared. +func (itr *BufIterator) Next() uint64 { + if itr.buf.full { + v := itr.buf.v + itr.buf.full = false + return v + } + return itr.itr.Next() +} + +// Unread pushes a value on to the buffer. +// Panics if the buffer is already full. +func (itr *BufIterator) Unread(v uint64) { + if itr.buf.full { + panic("roaring.BufIterator: buffer full") + } + itr.buf.v = v + itr.buf.full = true +} + // The maximum size of array containers. const arrayMaxSize = 4096 @@ -537,6 +599,37 @@ func (c *container) bitmapRemove(v uint16) bool { return true } +// max returns the maximum value in the container. +func (c *container) max() uint16 { + if c.isArray() { + return c.arrayMax() + } + return c.bitmapMax() +} + +func (c *container) arrayMax() uint16 { + return c.array[len(c.array)-1] +} + +func (c *container) bitmapMax() uint16 { + // Search bitmap in reverse order. + for i := len(c.bitmap) - 1; i >= 0; i-- { + // If value is zero then skip. + v := c.bitmap[i] + if v == 0 { + continue + } + + // Find the highest set bit. + for j := uint16(63); j >= 0; j-- { + if v&(1<