From de698aa03e877a8fd9c0fed337e89c286e2de525 Mon Sep 17 00:00:00 2001 From: Ben Johnson Date: Thu, 21 Apr 2016 15:55:44 -0600 Subject: [PATCH] consensus block merge 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. --- client.go | 39 +++-- cmd/pilosa/main.go | 40 ++--- cmd/pilosactl/main.go | 2 +- fragment.go | 323 ++++++++++++++++++++++++++++------------ handler.go | 40 +++-- handler_test.go | 7 +- index.go | 37 +++-- index_test.go | 28 +++- internal/internal.pb.go | 146 ++++++++---------- internal/internal.proto | 11 +- iterator.go | 180 ++++++++++++++++++++++ iterator_test.go | 74 +++++++++ roaring/roaring.go | 91 +++-------- roaring/roaring_test.go | 40 +---- 14 files changed, 666 insertions(+), 392 deletions(-) create mode 100644 iterator.go create mode 100644 iterator_test.go diff --git a/client.go b/client.go index 15f4b401b..3a0f221c0 100644 --- a/client.go +++ b/client.go @@ -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 { diff --git a/cmd/pilosa/main.go b/cmd/pilosa/main.go index 5c7971e6a..7393a51fa 100644 --- a/cmd/pilosa/main.go +++ b/cmd/pilosa/main.go @@ -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") } } diff --git a/cmd/pilosactl/main.go b/cmd/pilosactl/main.go index 43c361a1b..a7f0a3ddf 100644 --- a/cmd/pilosactl/main.go +++ b/cmd/pilosactl/main.go @@ -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 } } diff --git a/fragment.go b/fragment.go index 545f7aa8d..55f80c15c 100644 --- a/fragment.go +++ b/fragment.go @@ -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 +} diff --git a/handler.go b/handler.go index cb4a99f04..07a7f7349 100644 --- a/handler.go +++ b/handler.go @@ -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 diff --git a/handler_test.go b/handler_test.go index 889bf7534..c2b3c885e 100644 --- a/handler_test.go +++ b/handler_test.go @@ -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) } diff --git a/index.go b/index.go index 067321bf6..c09a47ac3 100644 --- a/index.go +++ b/index.go @@ -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 } diff --git a/index_test.go b/index_test.go index ab8bd592a..a19253592 100644 --- a/index_test.go +++ b/index_test.go @@ -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) diff --git a/internal/internal.pb.go b/internal/internal.pb.go index 638afb836..32869d702 100644 --- a/internal/internal.pb.go +++ b/internal/internal.pb.go @@ -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, } diff --git a/internal/internal.proto b/internal/internal.proto index 31106ab75..096620fcc 100644 --- a/internal/internal.proto +++ b/internal/internal.proto @@ -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 { diff --git a/iterator.go b/iterator.go new file mode 100644 index 000000000..4293378e6 --- /dev/null +++ b/iterator.go @@ -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 +} diff --git a/iterator_test.go b/iterator_test.go new file mode 100644 index 000000000..d18ee102e --- /dev/null +++ b/iterator_test.go @@ -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) + } +} diff --git a/roaring/roaring.go b/roaring/roaring.go index b425a3690..aba63077a 100644 --- a/roaring/roaring.go +++ b/roaring/roaring.go @@ -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 diff --git a/roaring/roaring_test.go b/roaring/roaring_test.go index 8c1bb9ef7..da65803e8 100644 --- a/roaring/roaring_test.go +++ b/roaring/roaring_test.go @@ -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))