active anti-entropy

This commit adds active anti-entropy via a merkle tree structure.

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",

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

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

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
+}

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 {

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
+}

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 }

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()

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
 }
 

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
+}

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

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,
 }

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;
 }

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)
 }
-

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) }

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<<j) != 0 {
+				return uint16(i)*64 + j
+			}
+		}
+	}
+	return 0
+}
+
 // convertToArray converts the values in the bitmap to array values.
 func (c *container) convertToArray() {
 	c.array = make([]uint16, 0, c.n)

roaring/roaring_test.go

@@ -68,6 +68,18 @@ func TestBitmap_ForEachRange(t *testing.T) {
 	}
 }
 
+// Ensure bitmap can return the highest value.
+func TestBitmap_Max(t *testing.T) {
+	bm := roaring.NewBitmap()
+	for i := uint64(1000); i <= 100000; i++ {
+		bm.Add(i)
+
+		if v := bm.Max(); v != i {
+			t.Fatalf("max: got=%d; want=%d", v, i)
+		}
+	}
+}
+
 func TestBitmap_Quick_Array1(t *testing.T)     { testBitmapQuick(t, 1000, 1000, 2000) }
 func TestBitmap_Quick_Array2(t *testing.T)     { testBitmapQuick(t, 10000, 0, 1000) }
 func TestBitmap_Quick_Bitmap1(t *testing.T)    { testBitmapQuick(t, 10000, 0, 10000) }
@@ -198,6 +210,57 @@ func testBitmapMarshalQuick(t *testing.T, n int, min, max uint64, sorted bool) {
 	})
 }
 
+// Ensure iterator can iterate over all the values on the bitmap.
+func TestIterator(t *testing.T) {
+	itr := roaring.NewBitmap(1, 2, 3).Iterator()
+
+	var a []uint64
+	for v := itr.Seek(0); !itr.EOF(); v = itr.Next() {
+		a = append(a, v)
+	}
+
+	if !reflect.DeepEqual(a, []uint64{1, 2, 3}) {
+		t.Fatalf("unexpected values: %+v", a)
+	}
+}
+
+// Ensure buffered iterator can unread values on to the buffer.
+func TestBufIterator(t *testing.T) {
+	itr := roaring.NewBufIterator(roaring.NewBitmap(1, 2, 3).Iterator())
+	if v := itr.Seek(1); v != 1 {
+		t.Fatalf("unexpected seek: %d", v)
+	} else if v := itr.Next(); v != 2 {
+		t.Fatalf("unexpected next: %d", v)
+	}
+
+	itr.Unread(10)
+	if v := itr.Next(); v != 10 {
+		t.Fatalf("unexpected next(buffered): %d", v)
+	}
+
+	if v := itr.Next(); v != 3 {
+		t.Fatalf("unexpected next: %d", v)
+	} else if itr.Next(); !itr.EOF() {
+		t.Fatal("expected eof")
+	}
+}
+
+// Ensure buffered iterator will panic if unreading onto a full buffer.
+func TestBufIterator_DoubleFillPanic(t *testing.T) {
+	var v interface{}
+	func() {
+		defer func() { v = recover() }()
+
+		itr := roaring.NewBufIterator(roaring.NewBitmap(1, 2, 3).Iterator())
+		itr.Unread(1)
+		itr.Unread(2)
+	}()
+
+	if !reflect.DeepEqual(v, "roaring.BufIterator: buffer full") {
+		t.Fatalf("unexpected panic value: %#v", v)
+	}
+}
+
 // GenerateUint64Slice generates between [0, n) random uint64 numbers between min and max.
 func GenerateUint64Slice(n int, min, max uint64, sorted bool, rand *rand.Rand) []uint64 {
 	a := make([]uint64, rand.Intn(n))