scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-08-28 02:44:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

consensus block merge

Browse source 
This commit refactors the anti-entropy system to fetch data from
all replicated blocks and only set/clear bits which deviate from
the consensus between all blocks.

An example of this is if 3 nodes had the following bits set for
a single bitmap:

	Node A: 1 2 3
	Node B:   2   4
	Node C: 1 2   4

Then only bits which are set on a majority will be set. In this
case bits 1, 2, & 4 are set but 3 only exists on a single node.

The node performing the merge would then determine the following
set/clear diffs for each node:

	Node A: clear(3), set(4)
	Node B: set(1)
	Node C: none

Once the merge is performed and all nodes receive their diff
instructions then the nodes will be in sync:

	Node A: 1 2 4
	Node B: 1 2 4
	Node C: 1 2 4

There still exists situations where bits can be reset. If Node A
is up and Node B & C are down then Node A's bits will be reset
once B & C come back online. We should add write consistency
settings for incoming writes so that we can ensure that a quorum
is written to before returning a success. This is outside the
scope of this commit though.
This commit is contained in:
Ben Johnson 2016-04-21 15:55:44 -06:00
parent 616248707b
commit de698aa03e
14 changed files with 666 additions and 392 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

39
client.go
View file

@ -116,7 +116,7 @@ func (c *Client) SliceNodes(slice uint64) ([]*Node, error) {
}
// ExecuteQuery executes query against db on the server.
func (c *Client) ExecuteQuery(db, query string) (result interface{}, err error) {
func (c *Client) ExecuteQuery(db, query string, allowRedirect bool) (result interface{}, err error) {
if db == "" {
return nil, ErrDatabaseRequired
} else if query == "" {
@ -125,8 +125,9 @@ func (c *Client) ExecuteQuery(db, query string) (result interface{}, err error)
// Encode query request.
buf, err := proto.Marshal(&internal.QueryRequest{
DB: proto.String(db),
Query: proto.String(query),
DB: proto.String(db),
Query: proto.String(query),
Remote: proto.Bool(!allowRedirect),
})
if err != nil {
return nil, fmt.Errorf("marshal: %s", err)
@ -502,26 +503,20 @@ func (c *Client) FragmentBlocks(db, frame string, slice uint64) ([]FragmentBlock
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,
// BlockData returns bitmap/profile id pairs for a block.
func (c *Client) BlockData(db, frame string, slice uint64, block int) ([]uint64, []uint64, error) {
buf, err := proto.Marshal(&internal.BlockDataRequest{
DB: proto.String(db),
Frame: proto.String(frame),
Slice: proto.Uint64(slice),
Block: proto.Uint64(uint64(block)),
})
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))
u := url.URL{Scheme: "http", Host: c.host, Path: "/fragment/block/data"}
req, err := http.NewRequest("GET", u.String(), bytes.NewReader(buf))
if err != nil {
return nil, nil, err
}
@ -536,12 +531,16 @@ func (c *Client) MergeBlock(db, frame string, slice uint64, block int, bitmapIDs
defer resp.Body.Close()
// Return error if status is not OK.
if resp.StatusCode != http.StatusOK {
switch resp.StatusCode {
case http.StatusOK: // fallthrough
case http.StatusNotFound:
return nil, nil, nil
default:
return nil, nil, fmt.Errorf("unexpected status: code=%d", resp.StatusCode)
}
// Decode response object.
var rsp internal.MergeBlockResponse
var rsp internal.BlockDataResponse
if body, err := ioutil.ReadAll(resp.Body); err != nil {
return nil, nil, err
} else if err := proto.Unmarshal(body, &rsp); err != nil {

40
cmd/pilosa/main.go
View file

@ -196,8 +196,9 @@ func (m *Main) Run(args ...string) error {
// Serve HTTP.
go func() { http.Serve(ln, h) }()
// Start anti-entropy background workers.
m.startAntiEntropyMonitors()
// Start anti-entropy background worker.
m.wg.Add(1)
go func() { defer m.wg.Done(); m.monitorAntiEntropy() }()
// Sync up max slice if more than one node
if len(m.Cluster.Nodes) > 1 {
@ -226,26 +227,11 @@ func (m *Main) Run(args ...string) error {
return nil
}
func (m *Main) startAntiEntropyMonitors() {
for _, node := range m.Cluster.Nodes {
// Skip this node.
if node.Host == m.Host {
continue
}
m.wg.Add(1)
go func(node *pilosa.Node) {
defer m.wg.Done()
m.monitorAntiEntropy(node)
}(node)
}
}
func (m *Main) monitorAntiEntropy(node *pilosa.Node) {
func (m *Main) monitorAntiEntropy() {
ticker := time.NewTicker(time.Duration(m.Config.AntiEntropy.Interval))
defer ticker.Stop()
m.logger().Printf("index sync monitor initializing: host=%s", node.Host)
m.logger().Printf("index sync monitor initializing")
for {
// Wait for tick or a close.
@ -255,28 +241,22 @@ func (m *Main) monitorAntiEntropy(node *pilosa.Node) {
case <-ticker.C:
}
m.logger().Printf("index sync beginning: host=%s", node.Host)
// Set up remote client.
client, err := pilosa.NewClient(node.Host)
if err != nil {
m.logger().Printf("anti-entropy client error: host=%s", node.Host)
continue
}
m.logger().Printf("index sync beginning")
// Initialize syncer with local index and remote client.
var syncer pilosa.IndexSyncer
syncer.Index = m.index
syncer.Client = client
syncer.Host = m.Host
syncer.Cluster = m.Cluster
// Sync indexes.
if err := syncer.SyncIndex(); err != nil {
m.logger().Printf("index sync error: host=%s, err=%s", node.Host, err)
m.logger().Printf("index sync error: err=%s", err)
continue
}
// Record successful sync in log.
m.logger().Printf("index sync complete: host=%s", node.Host)
m.logger().Printf("index sync complete")
}
}

2
cmd/pilosactl/main.go
View file

@ -631,7 +631,7 @@ func (cmd *BenchCommand) runSetBit(client *pilosa.Client) error {
q := fmt.Sprintf(`SetBit(id=%d, frame="%s", profileID=%d)`, bitmapID, cmd.Frame, profileID)
if _, err := client.ExecuteQuery(cmd.Database, q); err != nil {
if _, err := client.ExecuteQuery(cmd.Database, q, true); err != nil {
return err
}
}

323
fragment.go
View file

@ -400,7 +400,10 @@ func (f *Fragment) setTimeBit(bitmapID, profileID uint64, t time.Time, q TimeQua
func (f *Fragment) ClearBit(bitmapID, profileID uint64) (bool, error) {
f.mu.Lock()
defer f.mu.Unlock()
return f.clearBit(bitmapID, profileID)
}
func (f *Fragment) clearBit(bitmapID, profileID uint64) (bool, error) {
// Determine the position of the bit in the storage.
pos, err := f.pos(bitmapID, profileID)
if err != nil {
@ -668,8 +671,8 @@ func (f *Fragment) Blocks() []FragmentBlock {
return a
}
// BlockBits returns bits in a block as bitmap & profile ID pairs.
func (f *Fragment) BlockBits(id int) (bitmapIDs, profileIDs []uint64) {
// BlockData returns bits in a block as bitmap & profile ID pairs.
func (f *Fragment) BlockData(id int) (bitmapIDs, profileIDs []uint64) {
f.mu.Lock()
defer f.mu.Unlock()
@ -680,79 +683,130 @@ func (f *Fragment) BlockBits(id int) (bitmapIDs, profileIDs []uint64) {
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))
// MergeBlock compares the block's bits and computes a diff with another set of block bits.
// The state of a bit is determined by consensus from all blocks being considered.
//
// For example, if 3 blocks are compared and two have a set bit and one has a
// cleared bit then the bit is considered cleared. The function returns the
// diff per incoming block so that all can be in sync.
func (f *Fragment) MergeBlock(id int, data []PairSet) (sets, clears []PairSet, err error) {
// Ensure that all pair sets are of equal length.
for i := range data {
if len(data[i].BitmapIDs) != len(data[i].ProfileIDs) {
return nil, nil, fmt.Errorf("pair set mismatch(idx=%d): %d != %d", i, len(data[i].BitmapIDs), len(data[i].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
// Track sets and clears for all blocks (including local).
sets = make([]PairSet, len(data)+1)
clears = make([]PairSet, len(data)+1)
// Only look at values within hash block range.
min := uint64(id) * HashBlockSize * SliceWidth
max := uint64(id+1) * HashBlockSize * SliceWidth
// Limit upper bitmap/profile pair.
maxBitmapID := uint64(id+1) * HashBlockSize
maxProfileID := uint64(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)
// Create buffered iterator for local block.
itrs := make([]*BufIterator, 1, len(data)+1)
itrs[0] = NewBufIterator(
NewLimitIterator(
NewRoaringIterator(f.storage.Iterator()), maxBitmapID, maxProfileID,
),
)
// Append buffered iterators for each incoming block.
for i := range data {
var itr Iterator = NewSliceIterator(data[i].BitmapIDs, data[i].ProfileIDs)
itr = NewLimitIterator(itr, maxBitmapID, maxProfileID)
itrs = append(itrs, NewBufIterator(itr))
}
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
// Seek to initial pair.
for _, itr := range itrs {
itr.Seek(uint64(id)*HashBlockSize, 0)
}
// Determine the number of blocks needed to meet consensus.
// If there is an even split then a set is used.
majorityN := (len(itrs) + 1) / 2
// Iterate over all values in all iterators to determine differences.
values := make([]bool, len(itrs))
for {
var min struct {
bitmapID uint64
profileID uint64
}
// 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
// Find the lowest pair.
var hasData bool
for _, itr := range itrs {
bid, pid, eof := itr.Peek()
if eof { // no more data
continue
} else if !hasData { // first pair
min.bitmapID, min.profileID, hasData = bid, pid, true
} else if bid < min.bitmapID || (bid == min.bitmapID && pid < min.profileID) { // lower pair
min.bitmapID, min.profileID = bid, pid
}
}
if xEOF && yEOF { // no more data
// If all iterators are EOF then exit.
if !hasData {
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)
}
// Determine consensus of point.
var setN int
for i, itr := range itrs {
bid, pid, eof := itr.Next()
values[i] = !eof && bid == min.bitmapID && pid == min.profileID
if values[i] {
setN++ // set
} else {
itr.Unread() // clear
}
}
// Determine consensus value.
newValue := setN >= majorityN
// Add a diff for any node with a different value.
for i := range itrs {
// Value matches, ignore.
if values[i] == newValue {
continue
}
// Append to either the set or clear diff.
if newValue {
sets[i].BitmapIDs = append(sets[i].BitmapIDs, min.bitmapID)
sets[i].ProfileIDs = append(sets[i].ProfileIDs, min.profileID)
} else {
clears[i].BitmapIDs = append(sets[i].BitmapIDs, min.bitmapID)
clears[i].ProfileIDs = append(sets[i].ProfileIDs, min.profileID)
}
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 {
// Set local bits.
for i := range sets[0].ProfileIDs {
if _, err := f.setBit(sets[0].BitmapIDs[i], (f.Slice()*SliceWidth)+sets[0].ProfileIDs[i]); err != nil {
return nil, nil, err
}
}
return bids, pids, nil
// Clear local bits.
for i := range clears[0].ProfileIDs {
if _, err := f.clearBit(clears[0].BitmapIDs[i], (f.Slice()*SliceWidth)+clears[0].ProfileIDs[i]); err != nil {
return nil, nil, err
}
}
return sets[1:], clears[1:], nil
}
// Import bulk imports a set of bits and then snapshots the storage.
@ -1090,54 +1144,77 @@ type FragmentBlock struct {
// FragmentSyncer syncs a local fragment to one on a remote host.
type FragmentSyncer struct {
Fragment *Fragment
Client *Client
Host string
Cluster *Cluster
}
// 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()
// Determine replica set.
nodes := s.Cluster.SliceNodes(s.Fragment.Slice())
// Retrieve blocks.
remoteBlocks, err := s.Client.FragmentBlocks(s.Fragment.DB(), s.Fragment.Frame(), s.Fragment.Slice())
if err != nil && err != ErrFragmentNotFound {
return err
// Create a set of blocks.
blockSets := make([][]FragmentBlock, 0, len(nodes))
for _, node := range nodes {
// Read local blocks.
if node.Host == s.Host {
blockSets = append(blockSets, s.Fragment.Blocks())
continue
}
// Retrieve remote blocks.
client, err := NewClient(node.Host)
if err != nil {
return err
}
blocks, err := client.FragmentBlocks(s.Fragment.DB(), s.Fragment.Frame(), s.Fragment.Slice())
if err != nil && err != ErrFragmentNotFound {
return err
}
blockSets = append(blockSets, blocks)
}
// 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]
// Iterate over all blocks and find differences.
checksums := make([][]byte, len(nodes))
for {
// Find min block id.
blockID := -1
for _, blocks := range blockSets {
if len(blocks) == 0 {
continue
} else if blockID == -1 || blocks[0].ID < blockID {
blockID = blocks[0].ID
}
}
// Determine the next block to be merged.
var block *FragmentBlock
if a == nil && b == nil {
// Exit loop if no blocks are left.
if blockID == -1 {
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
}
// Read the checksum for the current block.
for i, blocks := range blockSets {
// Clear checksum if the next block for the node doesn't match current ID.
if len(blocks) == 0 || blocks[0].ID != blockID {
checksums[i] = nil
continue
}
// Otherwise set checksum and move forward.
checksums[i] = blocks[0].Checksum
blockSets[i] = blockSets[i][1:]
}
// Ignore if all the blocks on each node match.
if byteSlicesEqual(checksums) {
continue
}
// Synchronize block.
if err := s.syncBlock(block.ID); err != nil {
return fmt.Errorf("sync block: id=%d, err=%s", block.ID, err)
if err := s.syncBlock(blockID); err != nil {
return fmt.Errorf("sync block: id=%d, err=%s", blockID, err)
}
}
@ -1145,22 +1222,62 @@ func (s *FragmentSyncer) SyncFragment() error {
}
// syncBlock sends and receives all bitmaps for a given block.
// The remote bitmaps are merges it the local bitmaps.
// Returns an error if any remote hosts are unreachable.
func (s *FragmentSyncer) syncBlock(id int) error {
f := s.Fragment
// Retrieve bitmaps for block.
bitmapIDs, profileIDs := f.BlockBits(id)
// Read pairs from each remote block.
var pairSets []PairSet
var clients []*Client
for _, node := range s.Cluster.SliceNodes(f.Slice()) {
if s.Host == node.Host {
continue
}
// Send bitmaps to remote.
bids, pids, err := s.Client.MergeBlock(f.DB(), f.Frame(), f.Slice(), id, bitmapIDs, profileIDs)
client, err := NewClient(node.Host)
if err != nil {
return err
}
clients = append(clients, client)
bitmapIDs, profileIDs, err := client.BlockData(f.DB(), f.Frame(), f.Slice(), id)
if err != nil {
return err
}
pairSets = append(pairSets, PairSet{
ProfileIDs: profileIDs,
BitmapIDs: bitmapIDs,
})
}
// Merge blocks together.
sets, clears, err := f.MergeBlock(id, pairSets)
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 {
// Write updates to remote blocks.
for i := 0; i < len(clients); i++ {
set, clear := sets[i], clears[i]
// Ignore if there are no differences.
if len(set.ProfileIDs) == 0 && len(clear.ProfileIDs) == 0 {
continue
}
// Generate query with sets & clears.
var buf bytes.Buffer
for j := 0; j < len(set.ProfileIDs); j++ {
fmt.Fprintf(&buf, "SetBit(frame=%q, id=%d, profileID=%d)\n", f.Frame(), set.BitmapIDs[j], (f.Slice()*SliceWidth)+set.ProfileIDs[j])
}
for j := 0; j < len(clear.ProfileIDs); j++ {
fmt.Fprintf(&buf, "ClearBit(frame=%q, id=%d, profileID=%d)\n", f.Frame(), clear.BitmapIDs[j], (f.Slice()*SliceWidth)+clear.ProfileIDs[j])
}
// Execute query.
_, err := clients[i].ExecuteQuery(f.DB(), buf.String(), false)
if err != nil {
return err
}
}
@ -1175,3 +1292,23 @@ func madvise(b []byte, advice int) (err error) {
}
return
}
// PairSet is a list of equal length bitmap and profile id lists.
type PairSet struct {
BitmapIDs []uint64
ProfileIDs []uint64
}
// byteSlicesEqual returns true if all slices are equal.
func byteSlicesEqual(a [][]byte) bool {
if len(a) == 0 {
return true
}
for _, v := range a[1:] {
if !bytes.Equal(a[0], v) {
return false
}
}
return true
}

40
handler.go
View file

@ -112,13 +112,6 @@ 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":
@ -126,6 +119,13 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
default:
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
}
case "/fragment/block/data":
switch r.Method {
case "GET":
h.handleGetFragmentBlockData(w, r)
default:
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
}
case "/frame/restore":
switch r.Method {
case "POST":
@ -492,10 +492,10 @@ 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) {
// handleGetFragmentData handles GET /fragment/block/data requests.
func (h *Handler) handleGetFragmentBlockData(w http.ResponseWriter, r *http.Request) {
// Read request object.
var req internal.MergeBlockRequest
var req internal.BlockDataRequest
if body, err := ioutil.ReadAll(r.Body); err != nil {
http.Error(w, "ready body error", http.StatusBadRequest)
return
@ -505,24 +505,20 @@ func (h *Handler) handlePatchFragmentBlock(w http.ResponseWriter, r *http.Reques
}
// 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)
f := h.Index.Fragment(req.GetDB(), req.GetFrame(), req.GetSlice())
if f == nil {
http.Error(w, ErrFragmentNotFound.Error(), http.StatusNotFound)
return
}
// Merge data into block.
bids, pids, err := f.MergeBlock(int(req.GetBlock()), req.BitmapIDs, req.ProfileIDs)
if err != nil {
w.WriteHeader(http.StatusInternalServerError)
// Read data
var resp internal.BlockDataResponse
if f != nil {
resp.BitmapIDs, resp.ProfileIDs = f.BlockData(int(req.GetBlock()))
}
// Encode response.
buf, err := proto.Marshal(&internal.MergeBlockResponse{
BitmapIDs: bids,
ProfileIDs: pids,
Err: proto.String(errorString(err)),
})
buf, err := proto.Marshal(&resp)
if err != nil {
h.logger().Printf("merge block response encoding error: %s", err)
return

7
handler_test.go
View file

@ -551,8 +551,11 @@ func NewServer() *Server {
}
// Host returns the hostname of the running server.
func (s *Server) Host() string {
u, err := url.Parse(s.URL)
func (s *Server) Host() string { return MustParseURLHost(s.URL) }
// MustParseURLHost parses rawurl and returns the hostname. Panic on error.
func MustParseURLHost(rawurl string) string {
u, err := url.Parse(rawurl)
if err != nil {
panic(err)
}

37
index.go
View file

@ -201,33 +201,26 @@ func (i *Index) SetMax(newmax uint64) {
// 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
Index *Index
Host string
Cluster *Cluster
}
// 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)
sliceN := s.Index.SliceN()
// Iterate over schema in sorted order.
sliceN := s.Index.SliceN()
for _, di := range dbs {
for _, di := range s.Index.Schema() {
for _, fi := range di.Frames {
for slice := uint64(0); slice <= sliceN; slice++ {
// Ignore slices that this host doesn't own.
if !s.Cluster.OwnsSlice(s.Host, slice) {
continue
}
// Sync fragment if own it.
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)
}
@ -247,7 +240,11 @@ func (s *IndexSyncer) syncFragment(db, frame string, slice uint64) error {
}
// Sync fragments together.
fs := FragmentSyncer{Fragment: f, Client: s.Client}
fs := FragmentSyncer{
Fragment: f,
Host: s.Host,
Cluster: s.Cluster,
}
if err := fs.SyncFragment(); err != nil {
return err
}

28
index_test.go
View file

@ -7,10 +7,13 @@ import (
"testing"
"github.com/umbel/pilosa"
"github.com/umbel/pilosa/pql"
)
// Ensure index can sync with a remote index.
func TestIndexSyncer_SyncIndex(t *testing.T) {
cluster := NewCluster(2)
// Create a local index.
idx0 := MustOpenIndex()
defer idx0.Close()
@ -21,6 +24,17 @@ func TestIndexSyncer_SyncIndex(t *testing.T) {
s := NewServer()
defer s.Close()
s.Handler.Index = idx1.Index
s.Handler.Executor.ExecuteFn = func(db string, query *pql.Query, slices []uint64, opt *pilosa.ExecOptions) ([]interface{}, error) {
e := pilosa.NewExecutor(idx1.Index)
e.Host = cluster.Nodes[1].Host
e.Cluster = cluster
return e.Execute(db, query, slices, opt)
}
// Mock 2-node, fully replicated cluster.
cluster.ReplicaN = 2
cluster.Nodes[0].Host = "localhost:0"
cluster.Nodes[1].Host = MustParseURLHost(s.URL)
// Set data on the local index.
f := idx0.MustCreateFragmentIfNotExists("d", "f", 0)
@ -39,6 +53,8 @@ func TestIndexSyncer_SyncIndex(t *testing.T) {
t.Fatal(err)
}
idx0.MustCreateFragmentIfNotExists("y", "z", 0)
// Set data on the remote index.
f = idx1.MustCreateFragmentIfNotExists("d", "f", 0)
if _, err := f.SetBit(0, 4000, nil, 0); err != nil {
@ -47,8 +63,6 @@ func TestIndexSyncer_SyncIndex(t *testing.T) {
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)
@ -60,10 +74,14 @@ func TestIndexSyncer_SyncIndex(t *testing.T) {
t.Fatal(err)
}
// Set highest slice.
idx0.SetMax(3)
// Set up syncer.
syncer := pilosa.IndexSyncer{
Index: idx0.Index,
Client: MustNewClient(s.Host()).Client,
Index: idx0.Index,
Host: cluster.Nodes[0].Host,
Cluster: cluster,
}
if err := syncer.SyncIndex(); err != nil {
t.Fatal(err)
@ -82,8 +100,6 @@ func TestIndexSyncer_SyncIndex(t *testing.T) {
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)

146
internal/internal.pb.go
View file

@ -21,8 +21,8 @@ It has these top-level messages:
QueryResult
ImportRequest
ImportResponse
MergeBlockRequest
MergeBlockResponse
BlockDataRequest
BlockDataResponse
Cache
SliceMaxResponse
*/
@ -429,90 +429,66 @@ func (m *ImportResponse) GetErr() string {
return ""
}
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:"-"`
type BlockDataRequest 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"`
XXX_unrecognized []byte `json:"-"`
}
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 *BlockDataRequest) Reset() { *m = BlockDataRequest{} }
func (m *BlockDataRequest) String() string { return proto.CompactTextString(m) }
func (*BlockDataRequest) ProtoMessage() {}
func (*BlockDataRequest) Descriptor() ([]byte, []int) { return fileDescriptorInternal, []int{12} }
func (m *MergeBlockRequest) GetDB() string {
func (m *BlockDataRequest) GetDB() string {
if m != nil && m.DB != nil {
return *m.DB
}
return ""
}
func (m *MergeBlockRequest) GetFrame() string {
func (m *BlockDataRequest) GetFrame() string {
if m != nil && m.Frame != nil {
return *m.Frame
}
return ""
}
func (m *MergeBlockRequest) GetSlice() uint64 {
func (m *BlockDataRequest) GetSlice() uint64 {
if m != nil && m.Slice != nil {
return *m.Slice
}
return 0
}
func (m *MergeBlockRequest) GetBlock() uint64 {
func (m *BlockDataRequest) 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"`
type BlockDataResponse struct {
BitmapIDs []uint64 `protobuf:"varint,1,rep,name=BitmapIDs" json:"BitmapIDs,omitempty"`
ProfileIDs []uint64 `protobuf:"varint,2,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 *BlockDataResponse) Reset() { *m = BlockDataResponse{} }
func (m *BlockDataResponse) String() string { return proto.CompactTextString(m) }
func (*BlockDataResponse) ProtoMessage() {}
func (*BlockDataResponse) 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 {
func (m *BlockDataResponse) GetBitmapIDs() []uint64 {
if m != nil {
return m.BitmapIDs
}
return nil
}
func (m *MergeBlockResponse) GetProfileIDs() []uint64 {
func (m *BlockDataResponse) GetProfileIDs() []uint64 {
if m != nil {
return m.ProfileIDs
}
@ -566,45 +542,45 @@ func init() {
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((*BlockDataRequest)(nil), "internal.BlockDataRequest")
proto.RegisterType((*BlockDataResponse)(nil), "internal.BlockDataResponse")
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,
// 514 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x09, 0x6e, 0x88, 0x02, 0xff, 0x94, 0x53, 0x5b, 0x8b, 0xd3, 0x40,
0x18, 0x25, 0x4d, 0xda, 0xb4, 0x5f, 0x6c, 0xb7, 0x1d, 0x11, 0x83, 0xb0, 0xb8, 0xcc, 0x8a, 0x14,
0x1f, 0x56, 0x58, 0x7c, 0xf2, 0xcd, 0xb6, 0x8a, 0xcb, 0xb2, 0xcb, 0x5e, 0xd4, 0x67, 0x87, 0x3a,
0x6e, 0xe3, 0x26, 0x99, 0x38, 0x99, 0x80, 0x7d, 0xf2, 0xaf, 0xfb, 0xcd, 0x2d, 0x8d, 0x58, 0x11,
0x9f, 0xda, 0x39, 0xdf, 0xe5, 0x9c, 0x39, 0x39, 0x03, 0x8f, 0xb3, 0x52, 0x71, 0x59, 0xb2, 0xfc,
0xa5, 0xff, 0x73, 0x52, 0x49, 0xa1, 0x04, 0x19, 0xfa, 0x33, 0x7d, 0x0f, 0x83, 0x45, 0xa6, 0x0a,
0x56, 0x91, 0xa7, 0x30, 0x58, 0x6e, 0x9a, 0xf2, 0xbe, 0x4e, 0x83, 0xa3, 0x70, 0x9e, 0x9c, 0x1e,
0x9c, 0xb4, 0x43, 0x06, 0x27, 0x87, 0xd0, 0x7f, 0xa3, 0x94, 0xac, 0xd3, 0x9e, 0xa9, 0x4f, 0x76,
0x75, 0x0d, 0xd3, 0x63, 0xe8, 0xdb, 0xbe, 0x04, 0xc2, 0x73, 0xbe, 0xc5, 0x2d, 0xbd, 0x79, 0x44,
0xc6, 0xd0, 0xff, 0xc4, 0xf2, 0x86, 0x9b, 0xa1, 0x88, 0x52, 0x88, 0xae, 0x58, 0x26, 0xff, 0xe8,
0x59, 0x8a, 0xa6, 0x54, 0xd8, 0x83, 0x47, 0xfa, 0x02, 0x42, 0x94, 0x44, 0xa6, 0x30, 0xb4, 0xca,
0xce, 0x56, 0xae, 0x6f, 0x06, 0xa3, 0x2b, 0x29, 0xbe, 0x66, 0x39, 0x47, 0xc8, 0xf6, 0xbe, 0x82,
0xd8, 0x41, 0x04, 0xa0, 0xd7, 0x76, 0xfe, 0x43, 0xea, 0x25, 0x44, 0xfa, 0xb7, 0xab, 0x62, 0x44,
0x1e, 0x42, 0x72, 0xab, 0x64, 0x56, 0xde, 0x79, 0xbd, 0x01, 0x82, 0x48, 0xf9, 0x11, 0x67, 0x2d,
0x14, 0x22, 0x64, 0x54, 0x2c, 0x84, 0xc8, 0x2d, 0x14, 0x21, 0x34, 0xa4, 0x73, 0x88, 0xf5, 0xbe,
0x0b, 0x74, 0xb1, 0x65, 0x0e, 0xf6, 0x32, 0xff, 0x84, 0x07, 0xd7, 0x0d, 0x97, 0xdb, 0x1b, 0xfe,
0xbd, 0xe1, 0xb5, 0xd2, 0xa2, 0x57, 0x0b, 0x27, 0x00, 0x6d, 0x30, 0x35, 0x73, 0xb5, 0x11, 0x99,
0xc0, 0xe0, 0x36, 0xcf, 0xd6, 0xbc, 0x46, 0x5e, 0xb4, 0x4e, 0xfb, 0xe1, 0xae, 0x5a, 0x5b, 0x5a,
0xad, 0xe4, 0x43, 0x56, 0xe0, 0x1a, 0x56, 0x54, 0x69, 0x1f, 0xa1, 0x90, 0x1c, 0x40, 0x7c, 0xdd,
0xb0, 0x52, 0x35, 0x45, 0x3a, 0x40, 0x60, 0xac, 0xb7, 0xdc, 0xf0, 0x42, 0x28, 0x9e, 0xc6, 0x46,
0x6a, 0x06, 0x63, 0x27, 0xa0, 0xae, 0x44, 0x59, 0x73, 0xed, 0xc1, 0x5b, 0x29, 0x51, 0x82, 0xbe,
0xee, 0x73, 0x88, 0xb1, 0xd0, 0xe4, 0xca, 0x3b, 0xf7, 0x68, 0xa7, 0xdf, 0x8f, 0x61, 0x95, 0x1c,
0x77, 0xb4, 0x84, 0xa6, 0x71, 0xb6, 0x6b, 0x74, 0x15, 0xfa, 0x0d, 0x92, 0xee, 0xcc, 0x91, 0x4f,
0x9a, 0xe1, 0x4a, 0x4e, 0xa7, 0xbb, 0x09, 0x97, 0xc0, 0x11, 0x04, 0x97, 0xc6, 0x77, 0xf3, 0x01,
0x75, 0x4e, 0xfc, 0xf6, 0x8e, 0x8d, 0x26, 0x3e, 0x78, 0xcd, 0xe5, 0x86, 0x95, 0x77, 0xfc, 0x8b,
0xfb, 0x02, 0x9f, 0x61, 0x7c, 0x56, 0x54, 0x42, 0xaa, 0xbf, 0x18, 0xfb, 0x4e, 0xb2, 0x82, 0x3b,
0x63, 0xf1, 0x68, 0x8c, 0xc5, 0xdd, 0x2e, 0x55, 0x3e, 0x67, 0xda, 0x58, 0x6d, 0x35, 0x4e, 0xb7,
0x41, 0xab, 0xd1, 0x59, 0x9d, 0xdc, 0x43, 0x98, 0x78, 0x86, 0x3d, 0xce, 0xd1, 0x73, 0x98, 0x2e,
0x72, 0xb1, 0xbe, 0x5f, 0x31, 0xc5, 0xfe, 0x5f, 0x03, 0x1e, 0xcd, 0x34, 0xf2, 0xeb, 0x54, 0xbf,
0x86, 0x59, 0x67, 0x99, 0xa3, 0xfb, 0x4d, 0x67, 0xb0, 0x47, 0xa7, 0x7d, 0x61, 0x4f, 0xf0, 0x31,
0xb1, 0xf5, 0x66, 0x5f, 0x3f, 0x7d, 0x06, 0x53, 0xc3, 0x7a, 0xc1, 0x7e, 0xb4, 0x6b, 0x31, 0x56,
0x1e, 0xb3, 0x8f, 0xe7, 0x57, 0x00, 0x00, 0x00, 0xff, 0xff, 0xc3, 0xc6, 0x93, 0x16, 0x36, 0x04,
0x00, 0x00,
}

11
internal/internal.proto
View file

@ -71,19 +71,16 @@ message ImportResponse {
optional string Err = 1;
}
message MergeBlockRequest {
message BlockDataRequest {
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 BlockDataResponse {
repeated uint64 BitmapIDs = 1;
repeated uint64 ProfileIDs = 2;
}
message Cache {

180
iterator.go Normal file
View file

@ -0,0 +1,180 @@
package pilosa
import (
"fmt"
"github.com/umbel/pilosa/roaring"
)
// Iterator is an interface for looping over bitmap/profile pairs.
type Iterator interface {
Seek(bitmapID, profileID uint64)
Next() (bitmapID, profileID uint64, eof bool)
}
// BufIterator wraps an iterator to provide the ability to unread values.
type BufIterator struct {
buf struct {
bitmapID uint64
profileID uint64
eof bool
full bool
}
itr Iterator
}
// NewBufIterator returns a buffered iterator that wraps itr.
func NewBufIterator(itr Iterator) *BufIterator {
return &BufIterator{itr: itr}
}
// Seek moves to the first pair equal to or greater than pseek/bseek.
func (itr *BufIterator) Seek(bitmapID, profileID uint64) {
itr.buf.full = false
itr.itr.Seek(bitmapID, profileID)
}
// Next returns the next pair in the bitmap.
// If a value has been buffered then it is returned and the buffer is cleared.
func (itr *BufIterator) Next() (bitmapID, profileID uint64, eof bool) {
if itr.buf.full {
itr.buf.full = false
return itr.buf.bitmapID, itr.buf.profileID, itr.buf.eof
}
// Read values onto buffer in case of unread.
itr.buf.bitmapID, itr.buf.profileID, itr.buf.eof = itr.itr.Next()
return itr.buf.bitmapID, itr.buf.profileID, itr.buf.eof
}
// Peek reads the next value but leaves it on the buffer.
func (itr *BufIterator) Peek() (bitmapID, profileID uint64, eof bool) {
bitmapID, profileID, eof = itr.Next()
itr.Unread()
return
}
// Unread pushes previous pair on to the buffer.
// Panics if the buffer is already full.
func (itr *BufIterator) Unread() {
if itr.buf.full {
panic("pilosa.BufIterator: buffer full")
}
itr.buf.full = true
}
// LimitIterator wraps an Iterator and limits it to a max profile/bitmap pair.
type LimitIterator struct {
itr Iterator
maxBitmapID uint64
maxProfileID uint64
eof bool
}
// NewLimitIterator returns a new LimitIterator.
func NewLimitIterator(itr Iterator, maxBitmapID, maxProfileID uint64) *LimitIterator {
return &LimitIterator{
itr: itr,
maxBitmapID: maxBitmapID,
maxProfileID: maxProfileID,
}
}
// Seek moves the underlying iterator to a profile/bitmap pair.
func (itr *LimitIterator) Seek(bitmapID, profileID uint64) { itr.itr.Seek(bitmapID, profileID) }
// Next returns the next bitmap/profile ID pair.
// If the underlying iterator returns a pair higher than the max then EOF is returned.
func (itr *LimitIterator) Next() (bitmapID, profileID uint64, eof bool) {
// Always return EOF once it is reached by limit or the underlying iterator.
if itr.eof {
return 0, 0, true
}
// Retrieve pair from underlying iterator.
// Mark as EOF if it is beyond the limit (or at EOF).
bitmapID, profileID, eof = itr.itr.Next()
if eof || bitmapID > itr.maxBitmapID || (bitmapID == itr.maxBitmapID && profileID > itr.maxProfileID) {
itr.eof = true
return 0, 0, true
}
return bitmapID, profileID, false
}
// SliceIterator iterates over a pair of bitmap/profile ID slices.
type SliceIterator struct {
bitmapIDs []uint64
profileIDs []uint64
i, n int
}
// NewSliceIterator returns an iterator to iterate over a set of bitmap/profile ID pairs.
// Both slices MUST have an equal length. Otherwise the function will panic.
func NewSliceIterator(bitmapIDs, profileIDs []uint64) *SliceIterator {
if len(profileIDs) != len(bitmapIDs) {
panic(fmt.Sprintf("pilosa.SliceIterator: pair length mismatch: %d != %d", len(bitmapIDs), len(profileIDs)))
}
return &SliceIterator{
bitmapIDs: bitmapIDs,
profileIDs: profileIDs,
n: len(bitmapIDs),
}
}
// Seek moves the cursor to a given pair.
// If the pair is not found, the iterator seeks to the next pair.
func (itr *SliceIterator) Seek(bseek, pseek uint64) {
for i := 0; i < itr.n; i++ {
bitmapID := itr.bitmapIDs[i]
profileID := itr.profileIDs[i]
if (bseek == bitmapID && pseek <= profileID) || bseek < bitmapID {
itr.i = i
return
}
}
// Seek to the end of the slice if all values are less than seek pair.
itr.i = itr.n
}
// Next returns the next bitmap/profile ID pair.
func (itr *SliceIterator) Next() (bitmapID, profileID uint64, eof bool) {
if itr.i >= itr.n {
return 0, 0, true
}
bitmapID = itr.bitmapIDs[itr.i]
profileID = itr.profileIDs[itr.i]
itr.i++
return bitmapID, profileID, false
}
// RoaringIterator converts a roaring.Iterator to output profile/bitmap pairs.
type RoaringIterator struct {
itr *roaring.Iterator
}
// NewRoaringIterator returns a new iterator wrapping itr.
func NewRoaringIterator(itr *roaring.Iterator) *RoaringIterator {
return &RoaringIterator{itr: itr}
}
// Seek moves the cursor to a pair matching bseek/pseek.
// If the pair is not found then it moves to the next pair.
func (itr *RoaringIterator) Seek(bseek, pseek uint64) {
itr.itr.Seek((bseek * SliceWidth) + pseek)
}
// Next returns the next profile/bitmap ID pair.
func (itr *RoaringIterator) Next() (bitmapID, profileID uint64, eof bool) {
v, eof := itr.itr.Next()
return v / SliceWidth, v % SliceWidth, eof
}

74
iterator_test.go Normal file
View file

@ -0,0 +1,74 @@
package pilosa_test
import (
"reflect"
"testing"
"github.com/umbel/pilosa"
)
// Ensure slice iterator and iterate over a set of pairs.
func TestSliceIterator(t *testing.T) {
// Initialize iterator.
itr := pilosa.NewSliceIterator(
[]uint64{0, 0, 2, 4},
[]uint64{0, 1, 0, 10},
)
// Iterate over all pairs.
var pairs [][2]uint64
for pid, bid, eof := itr.Next(); !eof; pid, bid, eof = itr.Next() {
pairs = append(pairs, [2]uint64{pid, bid})
}
// Verify pairs output correctly.
if !reflect.DeepEqual(pairs, [][2]uint64{
{0, 0},
{0, 1},
{2, 0},
{4, 10},
}) {
t.Fatalf("unexpected pairs: %+v", pairs)
}
}
// Ensure buffered iterator can unread values on to the buffer.
func TestBufIterator(t *testing.T) {
itr := pilosa.NewBufIterator(pilosa.NewSliceIterator(
[]uint64{0, 0, 1, 2},
[]uint64{1, 3, 0, 100},
))
itr.Seek(0, 2)
if pid, bid, eof := itr.Next(); pid != 0 || bid != 3 || eof {
t.Fatalf("unexpected seek: (%d, %d, %v)", pid, bid, eof)
} else if pid, bid, eof := itr.Next(); pid != 1 || bid != 0 || eof {
t.Fatalf("unexpected next: (%d, %d, %v)", pid, bid, eof)
}
itr.Unread()
if pid, bid, eof := itr.Next(); pid != 1 || bid != 0 || eof {
t.Fatalf("unexpected next(buffered): (%d, %d, %v)", pid, bid, eof)
}
if pid, bid, eof := itr.Next(); pid != 2 || bid != 100 || eof {
t.Fatalf("unexpected next: (%d, %d, %v)", pid, bid, eof)
} else if _, _, eof := itr.Next(); !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 := pilosa.NewBufIterator(pilosa.NewSliceIterator(nil, nil))
itr.Unread()
itr.Unread()
}()
if !reflect.DeepEqual(v, "pilosa.BufIterator: buffer full") {
t.Fatalf("unexpected panic value: %#v", v)
}
}

91
roaring/roaring.go
View file

@ -130,7 +130,8 @@ func (b *Bitmap) Max() uint64 {
func (b *Bitmap) Slice() []uint64 {
var a []uint64
itr := b.Iterator()
for v := itr.Seek(0); !itr.EOF(); v = itr.Next() {
itr.Seek(0)
for v, eof := itr.Next(); !eof; v, eof = itr.Next() {
a = append(a, v)
}
return a
@ -140,7 +141,8 @@ func (b *Bitmap) Slice() []uint64 {
func (b *Bitmap) SliceRange(start, end uint64) []uint64 {
var a []uint64
itr := b.Iterator()
for v := itr.Seek(start); !itr.EOF() && v < end; v = itr.Next() {
itr.Seek(start)
for v, eof := itr.Next(); !eof && v < end; v, eof = itr.Next() {
a = append(a, v)
}
return a
@ -149,7 +151,8 @@ 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()
for v := itr.Seek(0); !itr.EOF(); v = itr.Next() {
itr.Seek(0)
for v, eof := itr.Next(); !eof; v, eof = itr.Next() {
fn(v)
}
}
@ -157,7 +160,8 @@ 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()
for v := itr.Seek(start); !itr.EOF() && v < end; v = itr.Next() {
itr.Seek(start)
for v, eof := itr.Next(); !eof && v < end; v, eof = itr.Next() {
fn(v)
}
}
@ -312,18 +316,18 @@ type Iterator struct {
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) }
// 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) }
// 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) {
// Move to the correct container.
itr.i = search64(itr.bitmap.keys, highbits(seek))
if itr.i < 0 {
itr.i = -itr.i - 1
}
if itr.EOF() {
return 0
if itr.eof() {
return
}
// Move to the correct value index inside the array container.
@ -335,25 +339,26 @@ func (itr *Iterator) Seek(seek uint64) uint64 {
itr.j = -itr.j - 1
}
if itr.j < len(c.array) {
return itr.peek()
itr.j--
return
}
// If it's at the end of the container then move to the next one.
itr.i, itr.j = itr.i+1, -1
return itr.Next()
return
}
// If it's a bitmap container then move to index before the value and call next().
itr.j = int(lb) - 1
return itr.Next()
}
// Next returns the next value in the bitmap.
func (itr *Iterator) Next() uint64 {
// Returns eof as true if there are no values left in the iterator.
func (itr *Iterator) Next() (v uint64, eof bool) {
// Iterate over containers until we find the next value or EOF.
for {
if itr.EOF() {
return 0
if itr.eof() {
return 0, true
}
// Move to the next item in the container if it's an array container.
@ -364,7 +369,7 @@ func (itr *Iterator) Next() uint64 {
continue
}
itr.j++
return itr.peek()
return itr.peek(), false
}
// Move to the next possible index in the bitmap container.
itr.j++
@ -379,14 +384,14 @@ func (itr *Iterator) Next() uint64 {
lb := c.bitmap[hb] >> (uint(itr.j) % 64)
if lb != 0 {
itr.j = int(itr.j) + trailingZeroN(lb)
return itr.peek()
return itr.peek(), false
}
// Otherwise iterate through remaining bitmaps to find next bit.
for hb++; hb < len(c.bitmap); hb++ {
if c.bitmap[hb] != 0 {
itr.j = int(hb*64) + trailingZeroN(c.bitmap[hb])
return itr.peek()
return itr.peek(), false
}
}
@ -405,56 +410,6 @@ 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

40
roaring/roaring_test.go
View file

@ -213,9 +213,10 @@ 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()
itr.Seek(0)
var a []uint64
for v := itr.Seek(0); !itr.EOF(); v = itr.Next() {
for v, eof := itr.Next(); !eof; v, eof = itr.Next() {
a = append(a, v)
}
@ -224,43 +225,6 @@ func TestIterator(t *testing.T) {
}
}
// 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))