diff --git a/api.go b/api.go index dccd527cb..c0d7d3257 100644 --- a/api.go +++ b/api.go @@ -763,9 +763,25 @@ func (api *API) ExportCSV(ctx context.Context, indexName string, fieldName strin return cw.Write([]string{rowStr, colStr}) } - // Iterate over each column. - if err := f.forEachBit(tx, fn); err != nil { - return errors.Wrap(err, "writing CSV") + citer, _, err := tx.ContainerIterator(indexName, fieldName, viewStandard, shard, 0) + if err != nil { + return err + } + var row, hi uint64 + var failed error + process := func(u uint16) { + if err := fn(row, hi|uint64(u)); err != nil { + failed = err + } + } + for citer.Next() { + key, c := citer.Value() + hi = key << 16 + row, hi = (hi / ShardWidth), (shard*ShardWidth)+(hi%ShardWidth) + roaring.ContainerCallback(c, process) + if failed != nil { + return errors.Wrap(err, "writing CSV") + } } // Ensure data is flushed. @@ -805,66 +821,6 @@ func (api *API) PartitionNodes(ctx context.Context, partitionID int) ([]*disco.N return snap.PartitionNodes(partitionID), nil } -// FragmentBlockData is an endpoint for internal usage. It is not guaranteed to -// return anything useful. Currently it returns protobuf encoded row and column -// ids from a "block" which is a subdivision of a fragment. -func (api *API) FragmentBlockData(ctx context.Context, body io.Reader) (_ []byte, err error) { - span, _ := tracing.StartSpanFromContext(ctx, "API.FragmentBlockData") - defer span.Finish() - - if err := api.validate(apiFragmentBlockData); err != nil { - return nil, errors.Wrap(err, "validating api method") - } - - reqBytes, err := io.ReadAll(body) - if err != nil { - return nil, NewBadRequestError(errors.Wrap(err, "read body error")) - } - var req BlockDataRequest - if err := api.Serializer.Unmarshal(reqBytes, &req); err != nil { - return nil, NewBadRequestError(errors.Wrap(err, "unmarshal body error")) - } - - // Retrieve fragment from holder. - f := api.holder.fragment(req.Index, req.Field, req.View, req.Shard) - if f == nil { - return nil, ErrFragmentNotFound - } - - var resp = BlockDataResponse{} - resp.RowIDs, resp.ColumnIDs, err = f.blockData(int(req.Block)) - if err != nil { - return nil, err - } - - // Encode response. - buf, err := api.Serializer.Marshal(&resp) - if err != nil { - return nil, errors.Wrap(err, "merge block response encoding error") - } - - return buf, nil -} - -// FragmentBlocks returns the checksums and block ids for all blocks in the specified fragment. -func (api *API) FragmentBlocks(ctx context.Context, indexName, fieldName, viewName string, shard uint64) ([]FragmentBlock, error) { - span, _ := tracing.StartSpanFromContext(ctx, "API.FragmentBlocks") - defer span.Finish() - - if err := api.validate(apiFragmentBlocks); err != nil { - return nil, errors.Wrap(err, "validating api method") - } - - // Retrieve fragment from holder. - f := api.holder.fragment(indexName, fieldName, viewName, shard) - if f == nil { - return nil, ErrFragmentNotFound - } - - // Retrieve blocks. - return f.Blocks() -} - // FragmentData returns all data in the specified fragment. func (api *API) FragmentData(ctx context.Context, indexName, fieldName, viewName string, shard uint64) (io.WriterTo, error) { span, _ := tracing.StartSpanFromContext(ctx, "API.FragmentData") diff --git a/catcher.go b/catcher.go index 40be70121..35b0bc06f 100644 --- a/catcher.go +++ b/catcher.go @@ -27,10 +27,6 @@ func init() { var _ Tx = (*catcherTx)(nil) -func (c *catcherTx) NewTxIterator(index, field, view string, shard uint64) *roaring.Iterator { - return c.b.NewTxIterator(index, field, view, shard) -} - func (c *catcherTx) ImportRoaringBits(index, field, view string, shard uint64, rit roaring.RoaringIterator, clear bool, log bool, rowSize uint64) (changed int, rowSet map[uint64]int, err error) { defer func() { if r := recover(); r != nil { @@ -150,28 +146,6 @@ func (c *catcherTx) ContainerIterator(index, field, view string, shard uint64, f return c.b.ContainerIterator(index, field, view, shard, firstRoaringContainerKey) } -func (c *catcherTx) ForEach(index, field, view string, shard uint64, fn func(i uint64) error) error { - - defer func() { - if r := recover(); r != nil { - vprint.AlwaysPrintf("see ForEach() PanicOn '%v' at '%v'", r, vprint.Stack()) - vprint.PanicOn(r) - } - }() - return c.b.ForEach(index, field, view, shard, fn) -} - -func (c *catcherTx) ForEachRange(index, field, view string, shard uint64, start, end uint64, fn func(uint64) error) error { - - defer func() { - if r := recover(); r != nil { - vprint.AlwaysPrintf("see ForEachRange() PanicOn '%v' at '%v'", r, vprint.Stack()) - vprint.PanicOn(r) - } - }() - return c.b.ForEachRange(index, field, view, shard, start, end, fn) -} - func (c *catcherTx) Count(index, field, view string, shard uint64) (uint64, error) { defer func() { diff --git a/cluster.go b/cluster.go index 753402654..c3e8b6ecc 100644 --- a/cluster.go +++ b/cluster.go @@ -53,9 +53,6 @@ type cluster struct { // nolint: maligned holder *Holder broadcaster broadcaster - abortAntiEntropyCh chan struct{} - muAntiEntropy sync.Mutex - translationSyncer TranslationSyncer mu sync.RWMutex @@ -97,33 +94,6 @@ func newCluster() *cluster { } } -// initializeAntiEntropy is called by the anti entropy routine when it starts. -// If the AE channel is created without a routine reading from it, cluster will -// block indefinitely when calling abortAntiEntropy(). -func (c *cluster) initializeAntiEntropy() { - c.mu.Lock() - c.abortAntiEntropyCh = make(chan struct{}) - c.mu.Unlock() -} - -// abortAntiEntropyQ checks whether the cluster wants to abort the anti entropy -// process (so that it can resize). It does not block. -func (c *cluster) abortAntiEntropyQ() bool { - select { - case <-c.abortAntiEntropyCh: - return true - default: - return false - } -} - -// abortAntiEntropy blocks until the anti-entropy routine calls abortAntiEntropyQ -func (c *cluster) abortAntiEntropy() { - if c.abortAntiEntropyCh != nil { - c.abortAntiEntropyCh <- struct{}{} - } -} - func (c *cluster) primaryNode() *disco.Node { return c.unprotectedPrimaryNode() } diff --git a/cluster_internal_test.go b/cluster_internal_test.go index c8f738699..51e7d4293 100644 --- a/cluster_internal_test.go +++ b/cluster_internal_test.go @@ -8,7 +8,6 @@ import ( "reflect" "testing" "testing/quick" - "time" "github.com/featurebasedb/featurebase/v3/disco" pnet "github.com/featurebasedb/featurebase/v3/net" @@ -180,63 +179,3 @@ func TestCluster_Nodes(t *testing.T) { } }) } - -func TestAE(t *testing.T) { - t.Run("AbortDoesn'tBlockUninitialized", func(t *testing.T) { - c := newCluster() - ch := make(chan struct{}) - go func() { - c.abortAntiEntropy() - close(ch) - }() - defer c.abortAntiEntropyQ() // avoid leaking a goroutine. - select { - case <-ch: - return - case <-time.After(time.Second): - t.Fatalf("aborting anti entropy on a new cluster blocked") - } - }) - - t.Run("AbortBlocksInitialized", func(t *testing.T) { - c := newCluster() - c.initializeAntiEntropy() - - ch := make(chan struct{}) - go func() { - c.abortAntiEntropy() - close(ch) - }() - defer c.abortAntiEntropyQ() // avoid leak of goroutine. - select { - case <-ch: - t.Fatalf("aborting anti entropy on an initialized cluster didn't block") - case <-time.After(time.Microsecond * 100): - } - }) - - t.Run("AbortAntiEntropyQ", func(t *testing.T) { - c := newCluster() - c.initializeAntiEntropy() - if c.abortAntiEntropyQ() { - t.Fatalf("abortAntiEntropyQ should report false when abort not called") - } - go func() { - for { - if c.abortAntiEntropyQ() { - break - } - } - }() - ch := make(chan struct{}) - go func() { - c.abortAntiEntropy() - close(ch) - }() - select { - case <-ch: - case <-time.After(time.Second): - t.Fatalf("abort should not have blocked this long") - } - }) -} diff --git a/fragment.go b/fragment.go index 8f9ac082f..d4420d7d6 100644 --- a/fragment.go +++ b/fragment.go @@ -7,9 +7,7 @@ import ( "bytes" "container/heap" "context" - "encoding/binary" "fmt" - "hash" "io" "math" "math/bits" @@ -177,8 +175,7 @@ func (f *fragment) bitDepth() (uint64, error) { } type FragmentInfo struct { - BitmapInfo roaring.BitmapInfo - BlockChecksums []FragmentBlock `json:"BlockChecksums,omitempty"` + BitmapInfo roaring.BitmapInfo } func (f *fragment) Index() *Index { @@ -1326,16 +1323,6 @@ func (f *fragment) pos(rowID, columnID uint64) (uint64, error) { return pos(rowID, columnID), nil } -// forEachBit executes fn for every bit set in the fragment. -// Errors returned from fn are passed through. -func (f *fragment) forEachBit(tx Tx, fn func(rowID, columnID uint64) error) error { - f.mu.Lock() - defer f.mu.Unlock() - return tx.ForEach(f.index(), f.field(), f.view(), f.shard, func(i uint64) error { - return fn(i/ShardWidth, (f.shard*ShardWidth)+(i%ShardWidth)) - }) -} - // top returns the top rows from the fragment. // If opt.Src is specified then only rows which intersect src are returned. func (f *fragment) top(tx Tx, opt topOptions) ([]Pair, error) { @@ -1517,270 +1504,6 @@ type topOptions struct { TanimotoThreshold uint64 } -// Checksum returns a checksum for the entire fragment. -// If two fragments have the same checksum then they have the same data. -func (f *fragment) Checksum() ([]byte, error) { - h := xxhash.New() - - blocks, err := f.Blocks() - if err != nil { - return nil, err - } - - for _, block := range blocks { - _, _ = h.Write(block.Checksum) - } - return h.Sum(nil), nil -} - -// InvalidateChecksums clears all cached block checksums. -func (f *fragment) InvalidateChecksums() { - f.mu.Lock() - f.checksums = make(map[int][]byte) - f.mu.Unlock() -} - -// Blocks returns info for all blocks containing data. -func (f *fragment) Blocks() ([]FragmentBlock, error) { - f.mu.Lock() - defer f.mu.Unlock() - - var a []FragmentBlock - - idx := f.holder.Index(f.index()) - if idx == nil { - err := fmt.Errorf("index() was nil in fragment.Blocks(): f.index()='%v'", f.index()) - vprint.PanicOn(err) - return nil, err - } - tx := idx.holder.txf.NewTx(Txo{Write: !writable, Index: idx, Fragment: f, Shard: f.shard}) - defer tx.Rollback() - // no Commit below, b/c is read-only. - - itr := tx.NewTxIterator(f.index(), f.field(), f.view(), f.shard) - defer itr.Close() - - itr.Seek(0) - - // Initialize block hasher. - h := newBlockHasher() - - // Iterate over each value in the fragment. - v, eof := itr.Next() - if eof { - return nil, nil - } - blockID := int(v / (HashBlockSize * ShardWidth)) - for { - // Check for multiple block checksums in a row. - if n := f.readContiguousChecksums(&a, blockID); n > 0 { - itr.Seek(uint64(blockID+n) * HashBlockSize * ShardWidth) - v, eof = itr.Next() - if eof { - break - } - blockID = int(v / (HashBlockSize * ShardWidth)) - continue - } - - // Reset hasher. - h.blockID = blockID - h.Reset() - - // Read all values for the block. - for ; ; v, eof = itr.Next() { - // Once we hit the next block, save the value for the next iteration. - blockID = int(v / (HashBlockSize * ShardWidth)) - if blockID != h.blockID || eof { - break - } - - h.WriteValue(v) - } - - // Cache checksum. - chksum := h.Sum() - f.checksums[h.blockID] = chksum // the only place checksums is added to. - - // Append block. - a = append(a, FragmentBlock{ - ID: h.blockID, - Checksum: chksum, - }) - - // Exit if we're at the end. - if eof { - break - } - } - - return a, nil -} - -// readContiguousChecksums appends multiple checksums in a row and returns the count added. -func (f *fragment) readContiguousChecksums(a *[]FragmentBlock, blockID int) (n int) { - for i := 0; ; i++ { - chksum := f.checksums[blockID+i] - if chksum == nil { - return i - } - - *a = append(*a, FragmentBlock{ - ID: blockID + i, - Checksum: chksum, - }) - } -} - -// blockData returns bits in a block as row & column ID pairs. -func (f *fragment) blockData(id int) (rowIDs, columnIDs []uint64, err error) { - f.mu.Lock() - defer f.mu.Unlock() - - idx := f.holder.Index(f.index()) - tx := idx.holder.txf.NewTx(Txo{Write: !writable, Index: idx, Shard: f.shard}) - defer tx.Rollback() - // readonly, so no Commit() - - if err := tx.ForEachRange(f.index(), f.field(), f.view(), f.shard, uint64(id)*HashBlockSize*ShardWidth, (uint64(id)+1)*HashBlockSize*ShardWidth, func(i uint64) error { - rowIDs = append(rowIDs, i/ShardWidth) - columnIDs = append(columnIDs, i%ShardWidth) - return nil - }); err != nil { - return nil, nil, err - } - return rowIDs, columnIDs, nil -} - -// 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 set. The function returns the -// diff per incoming block so that all can be in sync. -func (f *fragment) mergeBlock(tx Tx, 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].rowIDs) != len(data[i].columnIDs) { - return nil, nil, fmt.Errorf("pair set mismatch(idx=%d): %d != %d", i, len(data[i].rowIDs), len(data[i].columnIDs)) - } - } - - f.mu.Lock() - defer f.mu.Unlock() - - // Track sets and clears for all blocks (including local). - sets = make([]pairSet, len(data)+1) - clears = make([]pairSet, len(data)+1) - - // Limit upper row/column pair. - maxRowID := (uint64(id+1) * HashBlockSize) - 1 - maxColumnID := uint64(ShardWidth) - 1 - - // Create buffered iterator for local block. - bm, err := tx.RoaringBitmap(f.index(), f.field(), f.view(), f.shard) - if err != nil { - return nil, nil, err - } - itrs := make([]*bufIterator, 1, len(data)+1) - itrs[0] = newBufIterator( - newLimitIterator( - newRoaringIterator(bm.Iterator()), maxRowID, maxColumnID, - ), - ) - - // Append buffered iterators for each incoming block. - for i := range data { - var itr iterator = newSliceIterator(data[i].rowIDs, data[i].columnIDs) - itr = newLimitIterator(itr, maxRowID, maxColumnID) - itrs = append(itrs, newBufIterator(itr)) - } - - // 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 { - rowID uint64 - columnID uint64 - } - - // 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.rowID, min.columnID, hasData = bid, pid, true - } else if bid < min.rowID || (bid == min.rowID && pid < min.columnID) { // lower pair - min.rowID, min.columnID = bid, pid - } - } - - // If all iterators are EOF then exit. - if !hasData { - break - } - - // Determine consensus of point. - var setN int - for i, itr := range itrs { - bid, pid, eof := itr.Next() - - values[i] = !eof && bid == min.rowID && pid == min.columnID - 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].rowIDs = append(sets[i].rowIDs, min.rowID) - sets[i].columnIDs = append(sets[i].columnIDs, min.columnID) - } else { - clears[i].rowIDs = append(clears[i].rowIDs, min.rowID) - clears[i].columnIDs = append(clears[i].columnIDs, min.columnID) - } - } - } - - rowSet := make(map[uint64]struct{}, len(sets[0].columnIDs)) - // compute positions directly, replacing columnIDs with the computed - // positions - for i := range sets[0].columnIDs { - rowSet[sets[0].rowIDs[i]] = struct{}{} - sets[0].columnIDs[i] += sets[0].rowIDs[i] * ShardWidth - } - for i := range clears[0].columnIDs { - rowSet[clears[0].rowIDs[i]] = struct{}{} - clears[0].columnIDs[i] += clears[0].rowIDs[i] * ShardWidth - } - err = f.importPositions(tx, sets[0].columnIDs, clears[0].columnIDs, rowSet) - - return sets[1:], clears[1:], err -} - // bulkImport bulk imports a set of bits. // The cache is updated to reflect the new data. func (f *fragment) bulkImport(tx Tx, rowIDs, columnIDs []uint64, options *ImportOptions) error { @@ -2815,22 +2538,6 @@ func (f *fragment) unprotectedUnionRows(ctx context.Context, tx Tx, rows []uint6 } } -// blockToRoaringData converts a fragment block into a roaring.Bitmap -// which represents a portion of the data within a single shard. -// TODO: it seems like we should be able to get the -// block data as roaring without having to go through -// this rows/columns step. -func (f *fragment) blockToRoaringData(block int) ([]byte, error) { - rowIDs, columnIDs, err := f.blockData(block) - if err != nil { - return nil, err - } - return bitsToRoaringData(pairSet{ - columnIDs: columnIDs, - rowIDs: rowIDs, - }) -} - type rowIterator interface { // TODO(kuba) linter suggests to use io.Seeker // Seek(offset int64, whence int) (int64, error) @@ -3095,389 +2802,6 @@ func (it *setRowIterator) Next() (r *Row, rowID uint64, _ *int64, wrapped bool, return r, rowID, nil, wrapped, nil } -// FragmentBlock represents info about a subsection of the rows in a block. -// This is used for comparing data in remote blocks for active anti-entropy. -type FragmentBlock struct { - ID int `json:"id"` - Checksum []byte `json:"checksum"` -} - -type blockHasher struct { - blockID int - buf [8]byte - hash hash.Hash -} - -func newBlockHasher() blockHasher { - return blockHasher{ - blockID: -1, - hash: xxhash.New(), - } -} -func (h *blockHasher) Reset() { - h.hash.Reset() -} - -func (h *blockHasher) Sum() []byte { - return h.hash.Sum(nil)[:] -} - -func (h *blockHasher) WriteValue(v uint64) { - binary.BigEndian.PutUint64(h.buf[:], v) - _, _ = h.hash.Write(h.buf[:]) -} - -// fragmentSyncer syncs a local fragment to one on a remote host. -type fragmentSyncer struct { - Fragment *fragment - - Node *disco.Node - Cluster *cluster - - // FieldType helps determine which method of syncing to use. - FieldType string - - Closing <-chan struct{} -} - -// isClosing returns true if the closing channel is closed. -func (s *fragmentSyncer) isClosing() bool { - select { - case <-s.Closing: - return true - default: - return false - } -} - -// syncFragment compares checksums for the local and remote fragments and -// then merges any blocks which have differences. -func (s *fragmentSyncer) syncFragment() error { - span, ctx := tracing.StartSpanFromContext(context.Background(), "FragmentSyncer.syncFragment") - defer span.Finish() - - // Create a snapshot of the cluster to use for node/partition calculations. - snap := s.Cluster.NewSnapshot() - - // Determine replica set. - nodes := snap.ShardNodes(s.Fragment.index(), s.Fragment.shard) - if len(nodes) == 1 { - return nil - } - - // This is here solely to prevent unnecessary work; - // if this node isn't the primary replica, there's no need - // to continue processing int/decimal fields. - if nodes[0].ID != s.Node.ID { - switch s.FieldType { - case FieldTypeInt, FieldTypeDecimal, FieldTypeTimestamp: - return nil - } - } - - // Create a set of blocks. - blockSets := make([][]FragmentBlock, 0, len(nodes)) - for _, node := range nodes { - // Read local blocks. - if node.ID == s.Node.ID { - b, err := s.Fragment.Blocks() // comes from Tx store, creates its own Tx. - if err != nil { - return err - } - blockSets = append(blockSets, b) - continue - } - - // Retrieve remote blocks. - blocks, err := s.Cluster.InternalClient.FragmentBlocks(ctx, &node.URI, s.Fragment.index(), s.Fragment.field(), s.Fragment.view(), s.Fragment.shard) - if err != nil && err != ErrFragmentNotFound { - return errors.Wrap(err, "getting blocks") - } - blockSets = append(blockSets, blocks) - - // Verify sync is not prematurely closing. - if s.isClosing() { - return nil - } - } - - // 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 - } - } - - // Exit loop if no blocks are left. - if blockID == -1 { - break - } - - // 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 - } - - // If we've gotten here, it means that the block differs - // between nodes. If this particular fragment is part of an - // `int` or `decimal` field, then instead of using a consensus - // to determine which bits to update, we consider the primary - // replica to be correct, and overwrite the non-primary replicas - // with the primary's data. - - s.Fragment.holder.Logger.Debugf("sync block from primary: index='%v' field='%v' view='%v' shard='%v' id=%d", s.Fragment.index(), s.Fragment.field(), s.Fragment.view(), s.Fragment.shard, blockID) - - switch s.FieldType { - case FieldTypeInt, FieldTypeDecimal, FieldTypeTimestamp: - // Synchronize block from the primary replica. - if err := s.syncBlockFromPrimary(blockID); err != nil { - return fmt.Errorf("sync block from primary: id=%d, err=%s", blockID, err) - } - s.Fragment.stats.CountWithCustomTags(MetricBlockRepair, 1, 1.0, []string{"primary:true"}) - default: - // Synchronize block. - if err := s.syncBlock(blockID); err != nil { - return fmt.Errorf("sync block: id=%d, err=%s", blockID, err) - } - s.Fragment.stats.CountWithCustomTags(MetricBlockRepair, 1, 1.0, []string{"primary:false"}) - } - } - - return nil -} - -// syncBlockFromPrimary sends all rows for a given block -// from the primary replica to non-primary replicas. -// Since this is pushing updates out to replicas, it only -// runs on the primary replica. -// Returns an error if any remote hosts are unreachable. -func (s *fragmentSyncer) syncBlockFromPrimary(id int) error { - span, ctx := tracing.StartSpanFromContext(context.Background(), "FragmentSyncer.syncBlockFromPrimary") - defer span.Finish() - - f := s.Fragment - - // Create a snapshot of the cluster to use for node/partition calculations. - snap := s.Cluster.NewSnapshot() - - // Determine replica set. Return early if this is not - // the primary node. - nodes := snap.ShardNodes(f.index(), f.shard) - if s.Node.ID != nodes[0].ID { - f.holder.Logger.Debugf("non-primary replica expecting sync from primary: %s, index=%s, field=%s, shard=%d", nodes[0].ID, f.index(), f.field(), f.shard) - return nil - } - - // Get the local block represented as roaring data. - localData, err := f.blockToRoaringData(id) - if err != nil { - return errors.Wrap(err, "converting block to roaring data") - } - - // Verify sync is not prematurely closing. - if s.isClosing() { - return nil - } - - // Create the overwrite request to be sent to non-primary replicas. - overwriteReq := &ImportRoaringRequest{ - Action: RequestActionOverwrite, - Block: id, - Views: map[string][]byte{cleanViewName(f.view()): localData}, - } - - // Write updates to remote blocks. - for _, node := range nodes { - if s.Node.ID == node.ID { - continue - } - - uri := &node.URI - if err := s.Cluster.InternalClient.ImportRoaring(ctx, uri, f.index(), f.field(), f.shard, true, overwriteReq); err != nil { - return errors.Wrap(err, "sending roaring data (overwrite)") - } - } - - return nil -} - -// syncBlock sends and receives all rows for a given block. -// Returns an error if any remote hosts are unreachable. -func (s *fragmentSyncer) syncBlock(id int) error { - span, ctx := tracing.StartSpanFromContext(context.Background(), "FragmentSyncer.syncBlock") - defer span.Finish() - - f := s.Fragment - - // Create a snapshot of the cluster to use for node/partition calculations. - snap := s.Cluster.NewSnapshot() - - // Read pairs from each remote block. - var uris []*pnet.URI - var pairSets []pairSet - for _, node := range snap.ShardNodes(f.index(), f.shard) { - if s.Node.ID == node.ID { - continue - } - - // Verify sync is not prematurely closing. - if s.isClosing() { - return nil - } - - uri := &node.URI - uris = append(uris, uri) - - // Only sync the standard block. - // Does a remote fetch - rowIDs, columnIDs, err := s.Cluster.InternalClient.BlockData(ctx, &node.URI, f.index(), f.field(), f.view(), f.shard, id) - if err != nil { - return errors.Wrap(err, "getting block") - } - - pairSets = append(pairSets, pairSet{ - columnIDs: columnIDs, - rowIDs: rowIDs, - }) - } - - // Verify sync is not prematurely closing. - if s.isClosing() { - return nil - } - - idx := f.holder.Index(f.index()) - tx := idx.holder.txf.NewTx(Txo{Write: writable, Index: idx, Shard: f.shard}) - defer tx.Rollback() - - // Merge blocks together. - sets, clears, err := f.mergeBlock(tx, id, pairSets) - if err != nil { - return errors.Wrap(err, "merging") - } - - // no safeCopy needed here. We are not leaking data outside the tx, because - // sets and clears only contain columnIDs. - - err = tx.Commit() - if err != nil { - return err - } - - // Write updates to remote blocks. - for i := 0; i < len(uris); i++ { - set, clear := sets[i], clears[i] - - // Handle Sets. - if len(set.columnIDs) > 0 { - setData, err := bitsToRoaringData(set) - if err != nil { - return errors.Wrap(err, "converting bits to roaring data (set)") - } - - setReq := &ImportRoaringRequest{ - Action: RequestActionSet, - Views: map[string][]byte{cleanViewName(f.view()): setData}, - } - - if err := s.Cluster.InternalClient.ImportRoaring(ctx, uris[i], f.index(), f.field(), f.shard, true, setReq); err != nil { - return errors.Wrap(err, "sending roaring data (set)") - } - } - - // Handle Clears. - if len(clear.columnIDs) > 0 { - clearData, err := bitsToRoaringData(clear) - if err != nil { - return errors.Wrap(err, "converting bits to roaring data (clear)") - } - - clearReq := &ImportRoaringRequest{ - Action: RequestActionClear, - Views: map[string][]byte{cleanViewName(f.view()): clearData}, - } - - if err := s.Cluster.InternalClient.ImportRoaring(ctx, uris[i], f.index(), f.field(), f.shard, true, clearReq); err != nil { - return errors.Wrap(err, "sending roaring data (clear)") - } - } - } - - return nil -} - -// cleanViewName converts a view name into the equivalent -// string required by the external api. Because views are -// not exposed externally, the conversion looks like this: -// "standard" -> "" -// "standard_YYYYMMDD" -> "YYYYMMDD" -// "other" -> "other" (there is currently not a use for this) -func cleanViewName(v string) string { - viewPrefix := viewStandard + "_" - if strings.HasPrefix(v, viewPrefix) { - return v[len(viewPrefix):] - } else if v == viewStandard { - return "" - } - return v -} - -// bitsToRoaringData converts a pairSet into a roaring.Bitmap -// which represents the data within a single shard. -func bitsToRoaringData(ps pairSet) ([]byte, error) { - bmp := roaring.NewBitmap() - for j := 0; j < len(ps.columnIDs); j++ { - bmp.DirectAdd(ps.rowIDs[j]*ShardWidth + (ps.columnIDs[j] % ShardWidth)) - } - - var buf bytes.Buffer - _, err := bmp.WriteTo(&buf) - if err != nil { - return nil, errors.Wrap(err, "writing to buffer") - } - - return buf.Bytes(), nil -} - -// pairSet is a list of equal length row and column id lists. -type pairSet struct { - rowIDs []uint64 - columnIDs []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 -} - // pos returns the row position of a row/column pair. func pos(rowID, columnID uint64) uint64 { return (rowID * ShardWidth) + (columnID % ShardWidth) diff --git a/fragment_internal_test.go b/fragment_internal_test.go index e39cfc0dc..7d56ee11f 100644 --- a/fragment_internal_test.go +++ b/fragment_internal_test.go @@ -1146,36 +1146,6 @@ func BenchmarkFragment_RepeatedSmallValueImports(b *testing.B) { } } -// Ensure a fragment can iterate over all bits in order. -func TestFragment_ForEachBit(t *testing.T) { - f, idx, tx := mustOpenFragment(t) - _ = idx - defer f.Clean(t) - - // Set bits on the fragment. - if _, err := f.setBit(tx, 100, 20); err != nil { - t.Fatal(err) - } else if _, err := f.setBit(tx, 2, 38); err != nil { - t.Fatal(err) - } else if _, err := f.setBit(tx, 2, 37); err != nil { - t.Fatal(err) - } - - // Iterate over bits. - var result [][2]uint64 - if err := f.forEachBit(tx, func(rowID, columnID uint64) error { - result = append(result, [2]uint64{rowID, columnID}) - return nil - }); err != nil { - t.Fatal(err) - } - - // Verify bits are correct. - if !reflect.DeepEqual(result, [][2]uint64{{2, 37}, {2, 38}, {100, 20}}) { - t.Fatalf("unexpected result: %#v", result) - } -} - // Ensure a fragment can return the top n results. func TestFragment_Top(t *testing.T) { f, idx, tx := mustOpenFragment(t, OptFieldTypeSet(CacheTypeRanked, DefaultCacheSize)) @@ -1388,110 +1358,6 @@ func TestFragment_TopN_CacheSize(t *testing.T) { } } -// Ensure fragment can return a checksum for its blocks. -func TestFragment_Checksum(t *testing.T) { - f, idx, tx := mustOpenFragment(t) - _ = idx - defer f.Clean(t) - tx.Rollback() // allow f.Checksum to make its read tx. - - // Retrieve checksum and set bits. - orig, err := f.Checksum() - if err != nil { - t.Fatal(err) - } - - tx = idx.holder.txf.NewTx(Txo{Write: writable, Index: idx, Fragment: f, Shard: f.shard}) - defer tx.Rollback() - - if _, err := f.setBit(tx, 1, 200); err != nil { - t.Fatal(err) - } else if _, err := f.setBit(tx, HashBlockSize*2, 200); err != nil { - t.Fatal(err) - } - PanicOn(tx.Commit()) - - // Ensure new checksum is different. - if chksum, err := f.Checksum(); err != nil { - t.Fatal(err) - } else if bytes.Equal(chksum, orig) { - t.Fatalf("expected checksum to change: %x - %x", chksum, orig) - } -} - -// Ensure fragment can return a checksum for a given block. -func TestFragment_Blocks(t *testing.T) { - f, idx, tx := mustOpenFragment(t) - _ = idx - defer f.Clean(t) - - // Retrieve initial checksum. - var prev []FragmentBlock - - // Set first bit. - if _, err := f.setBit(tx, 0, 0); err != nil { - t.Fatal(err) - } - PanicOn(tx.Commit()) - blocks, err := f.Blocks() // FAIL: TestFragment_Blocks b/c 0 blocks back - if err != nil { - t.Fatal(err) - } else if blocks[0].Checksum == nil { - t.Fatalf("expected checksum: %x", blocks[0].Checksum) - } - prev = blocks - - tx = idx.holder.txf.NewTx(Txo{Write: true, Index: idx, Fragment: f, Shard: f.shard}) - // Set bit on different row. - if _, err := f.setBit(tx, 20, 0); err != nil { - t.Fatal(err) - } - PanicOn(tx.Commit()) - blocks, err = f.Blocks() - if err != nil { - t.Fatal(err) - } else if bytes.Equal(blocks[0].Checksum, prev[0].Checksum) { - t.Fatalf("expected checksum to change: %x", blocks[0].Checksum) - } - prev = blocks - - // Set bit on different column. - tx = idx.holder.txf.NewTx(Txo{Write: true, Index: idx, Fragment: f, Shard: f.shard}) - defer tx.Rollback() - if _, err := f.setBit(tx, 20, 100); err != nil { - t.Fatal(err) - } - PanicOn(tx.Commit()) - blocks, err = f.Blocks() - if err != nil { - t.Fatal(err) - } else if bytes.Equal(blocks[0].Checksum, prev[0].Checksum) { - t.Fatalf("expected checksum to change: %x", blocks[0].Checksum) - } -} - -// Ensure fragment returns an empty checksum if no data exists for a block. -func TestFragment_Blocks_Empty(t *testing.T) { - f, idx, tx := mustOpenFragment(t) - _ = idx - defer f.Clean(t) - - // Set bits on a different block. - if _, err := f.setBit(tx, 100, 1); err != nil { - t.Fatal(err) - } - PanicOn(tx.Commit()) // f.Blocks() will start a new Tx, so the SetBit needs to be visible before that. - - // Ensure checksum for block 1 is blank. - if blocks, err := f.Blocks(); err != nil { - t.Fatal(err) - } else if len(blocks) != 1 { - t.Fatalf("unexpected block count: %d", len(blocks)) - } else if blocks[0].ID != 1 { - t.Fatalf("unexpected block id: %d", blocks[0].ID) - } -} - // Ensure a fragment's cache can be persisted between restarts. func TestFragment_LRUCache_Persistence(t *testing.T) { f, idx, tx := mustOpenFragment(t, OptFieldTypeSet(CacheTypeLRU, 0)) @@ -1592,30 +1458,6 @@ func TestFragment_WriteTo_ReadFrom(t *testing.T) { } } -func BenchmarkFragment_Blocks(b *testing.B) { - if *FragmentPath == "" { - b.Skip("no fragment specified") - } - - // Open the fragment specified by the path. Note that newFragment - // is overriding the usual holder-to-fragment path logic... - _, _, _, _, f := newTestFragment(b) - if err := f.Open(); err != nil { - b.Fatal(err) - } - defer f.Clean(b) - - // Reset timer and execute benchmark. - b.ResetTimer() - for i := 0; i < b.N; i++ { - if a, err := f.Blocks(); err != nil { - b.Fatal(err) - } else if len(a) == 0 { - b.Fatal("no blocks in fragment") - } - } -} - func BenchmarkFragment_IntersectionCount(b *testing.B) { f, idx, tx := mustOpenFragment(b) defer f.Clean(b) diff --git a/holder.go b/holder.go index 78009e104..4b5d270be 100644 --- a/holder.go +++ b/holder.go @@ -202,9 +202,8 @@ type HolderConfig struct { StatsClient stats.StatsClient Logger logger.Logger - StorageConfig *storage.Config - RBFConfig *rbfcfg.Config - AntiEntropyInterval time.Duration + StorageConfig *storage.Config + RBFConfig *rbfcfg.Config LookupDBDSN string } @@ -1242,117 +1241,6 @@ type holderSyncer struct { Closing <-chan struct{} } -// IsClosing returns true if the syncer has been asked to close. -func (s *holderSyncer) IsClosing() bool { - if s.Cluster.abortAntiEntropyQ() { - return true - } - select { - case <-s.Closing: - return true - default: - return false - } -} - -// SyncHolder compares the holder on host with the local holder and resolves differences. -func (s *holderSyncer) SyncHolder() error { - s.mu.Lock() // only allow one instance of SyncHolder to be running at a time - defer s.mu.Unlock() - ti := time.Now() - - // Create a snapshot of the cluster to use for node/partition calculations. - snap := s.Cluster.NewSnapshot() - - schema, err := s.Holder.Schema() - if err != nil { - return errors.Wrap(err, "getting schema") - } - - // Iterate over schema in sorted order. - for _, di := range schema { - // Verify syncer has not closed. - if s.IsClosing() { - return nil - } - - tf := time.Now() - for _, fi := range di.Fields { - // Verify syncer has not closed. - if s.IsClosing() { - return nil - } - - for _, vi := range fi.Views { - // Verify syncer has not closed. - if s.IsClosing() { - return nil - } - - itr := s.Holder.Index(di.Name).AvailableShards(includeRemote).Iterator() - itr.Seek(0) - for shard, eof := itr.Next(); !eof; shard, eof = itr.Next() { - // Ignore shards that this host doesn't own. - if !snap.OwnsShard(s.Node.ID, di.Name, shard) { - continue - } - - // Verify syncer has not closed. - if s.IsClosing() { - return nil - } - - // Sync fragment if own it. - if err := s.syncFragment(di.Name, fi.Name, vi.Name, shard); err != nil { - return fmt.Errorf("fragment sync error: index=%s, field=%s, view=%s, shard=%d, err=%s", di.Name, fi.Name, vi.Name, shard, err) - } - } - } - s.Stats.Timing(MetricSyncFieldDurationSeconds, time.Since(tf), 1.0) - tf = time.Now() // reset tf - } - s.Stats.Timing(MetricSyncIndexDurationSeconds, time.Since(ti), 1.0) - ti = time.Now() // reset ti - } - - return nil -} - -// syncFragment synchronizes a fragment with the rest of the cluster. -func (s *holderSyncer) syncFragment(index, field, view string, shard uint64) error { - // Retrieve local field. - f := s.Holder.Field(index, field) - if f == nil { - return newNotFoundError(ErrFieldNotFound, field) - } - - // Ensure view exists locally. - v, err := f.createViewIfNotExists(view) - if err != nil { - return errors.Wrap(err, "creating view") - } - - // Ensure fragment exists locally. - frag, err := v.CreateFragmentIfNotExists(shard) - if err != nil { - return errors.Wrap(err, "creating fragment") - } - - // Sync fragments together. - fs := fragmentSyncer{ - Fragment: frag, - Node: s.Node, - Cluster: s.Cluster, - FieldType: f.Type(), - Closing: s.Closing, - } - if err := fs.syncFragment(); err != nil { - return errors.Wrap(err, "syncing fragment") - } - - return nil -} - // resetTranslationSync reinitializes streaming sync of translation data. func (s *holderSyncer) resetTranslationSync() error { if s.stopInitializeReplicationCh == nil { diff --git a/holder_test.go b/holder_test.go index 0e99958ae..7345745d5 100644 --- a/holder_test.go +++ b/holder_test.go @@ -3,13 +3,9 @@ package pilosa_test import ( - "context" - "math" "os" - "reflect" "strings" "testing" - "time" pilosa "github.com/featurebasedb/featurebase/v3" "github.com/featurebasedb/featurebase/v3/pql" @@ -202,401 +198,3 @@ func TestHolder_DeleteIndex(t *testing.T) { t.Fatal("expected i1 files to still exist", err) } } - -// Ensure holder can sync with a remote holder. -func TestHolderSyncer_SyncHolder(t *testing.T) { - c := test.MustUnsharedCluster(t, 2) - c.GetIdleNode(0).Config.Cluster.ReplicaN = 2 - c.GetIdleNode(0).Config.AntiEntropy.Interval = 0 - c.GetIdleNode(1).Config.Cluster.ReplicaN = 2 - c.GetIdleNode(1).Config.AntiEntropy.Interval = 0 - err := c.Start() - - if err != nil { - t.Fatalf("starting cluster: %v", err) - } - defer c.Close() - - _, err = c.GetNode(0).API.CreateIndex(context.Background(), c.Idx(), pilosa.IndexOptions{}) - if err != nil { - t.Fatalf("creating index i: %v", err) - } - _, err = c.GetNode(0).API.CreateIndex(context.Background(), c.Idx("y"), pilosa.IndexOptions{}) - if err != nil { - t.Fatalf("creating index y: %v", err) - } - _, err = c.GetNode(0).API.CreateField(context.Background(), c.Idx(), "f", pilosa.OptFieldTypeSet(pilosa.DefaultCacheType, pilosa.DefaultCacheSize)) - if err != nil { - t.Fatalf("creating field f: %v", err) - } - _, err = c.GetNode(0).API.CreateField(context.Background(), c.Idx(), "f0", pilosa.OptFieldTypeSet(pilosa.DefaultCacheType, pilosa.DefaultCacheSize)) - if err != nil { - t.Fatalf("creating field f0: %v", err) - } - _, err = c.GetNode(0).API.CreateField(context.Background(), c.Idx("y"), "z", pilosa.OptFieldTypeMutex(pilosa.DefaultCacheType, pilosa.DefaultCacheSize)) - if err != nil { - t.Fatalf("creating field z in y: %v", err) - } - _, err = c.GetNode(0).API.CreateField(context.Background(), c.Idx("y"), "b", pilosa.OptFieldTypeBool()) - if err != nil { - t.Fatalf("creating field b in y: %v", err) - } - - hldr0 := &test.Holder{Holder: c.GetNode(0).Server.Holder()} - hldr1 := &test.Holder{Holder: c.GetNode(1).Server.Holder()} - - // Set data on the local holder. - hldr0.SetBit(c.Idx(), "f", 0, 10) - hldr0.SetBit(c.Idx(), "f", 2, 20) - hldr0.SetBit(c.Idx(), "f", 120, 10) - hldr0.SetBit(c.Idx(), "f", 200, 4) - - hldr0.SetBit(c.Idx(), "f0", 9, ShardWidth+5) - - // Set a bit to create the fragment. - hldr0.SetBit(c.Idx("y"), "z", 0, 0) - hldr0.SetBit(c.Idx("y"), "b", 0, 0) // rowID = 0 means false - - // Set data on the remote holder. - hldr1.SetBit(c.Idx(), "f", 0, 4000) - hldr1.SetBit(c.Idx(), "f", 3, 10) - hldr1.SetBit(c.Idx(), "f", 120, 10) - - hldr1.SetBit(c.Idx("y"), "z", 10, (3*ShardWidth)+4) - hldr1.SetBit(c.Idx("y"), "z", 10, (3*ShardWidth)+5) - hldr1.SetBit(c.Idx("y"), "z", 10, (3*ShardWidth)+7) - - hldr1.SetBit(c.Idx("y"), "b", 1, (3*ShardWidth)+4) // true - hldr1.SetBit(c.Idx("y"), "b", 0, (3*ShardWidth)+5) // false - hldr1.SetBit(c.Idx("y"), "b", 1, (3*ShardWidth)+7) // true - - err = c.GetNode(0).Server.SyncData() - if err != nil { - t.Fatalf("syncing node 0: %v", err) - } - err = c.GetNode(1).Server.SyncData() - if err != nil { - t.Fatalf("syncing node 1: %v", err) - } - - // Verify data is the same on both nodes. - for i, hldr := range []*test.Holder{hldr0, hldr1} { - if a := hldr.Row(c.Idx(), "f", 0).Columns(); !reflect.DeepEqual(a, []uint64{10, 4000}) { - t.Errorf("unexpected columns(%d/0): %+v", i, a) - } - if a := hldr.Row(c.Idx(), "f", 2).Columns(); !reflect.DeepEqual(a, []uint64{20}) { - t.Errorf("unexpected columns(%d/2): %+v", i, a) - } - if a := hldr.Row(c.Idx(), "f", 3).Columns(); !reflect.DeepEqual(a, []uint64{10}) { - t.Errorf("unexpected columns(%d/3): %+v", i, a) - } - if a := hldr.Row(c.Idx(), "f", 120).Columns(); !reflect.DeepEqual(a, []uint64{10}) { - t.Errorf("unexpected columns(%d/120): %+v", i, a) - } - if a := hldr.Row(c.Idx(), "f", 200).Columns(); !reflect.DeepEqual(a, []uint64{4}) { - t.Errorf("unexpected columns(%d/200): %+v", i, a) - } - - if a := hldr.Row(c.Idx(), "f0", 9).Columns(); !reflect.DeepEqual(a, []uint64{ShardWidth + 5}) { - t.Errorf("unexpected columns(%d/d/f0): %+v", i, a) - } - - if a := hldr.Row(c.Idx("y"), "z", 10).Columns(); !reflect.DeepEqual(a, []uint64{(3 * ShardWidth) + 4, (3 * ShardWidth) + 5, (3 * ShardWidth) + 7}) { - t.Errorf("unexpected columns(%d/y/z): %+v", i, a) - } - - if a := hldr.Row(c.Idx("y"), "b", 0).Columns(); !reflect.DeepEqual(a, []uint64{0, (3 * ShardWidth) + 5}) { - t.Errorf("unexpected false columns(%d/y/b): %+v", i, a) - } - if a := hldr.Row(c.Idx("y"), "b", 1).Columns(); !reflect.DeepEqual(a, []uint64{(3 * ShardWidth) + 4, (3 * ShardWidth) + 7}) { - t.Errorf("unexpected true columns(%d/y/b): %+v", i, a) - } - } -} - -// Ensure holder can sync with a remote holder and respects -// the row boundaries of the block. -func TestHolderSyncer_BlockIteratorLimits(t *testing.T) { - c := test.MustUnsharedCluster(t, 3) - c.GetIdleNode(0).Config.Cluster.ReplicaN = 3 - c.GetIdleNode(0).Config.AntiEntropy.Interval = 0 - c.GetIdleNode(1).Config.Cluster.ReplicaN = 3 - c.GetIdleNode(1).Config.AntiEntropy.Interval = 0 - c.GetIdleNode(2).Config.Cluster.ReplicaN = 3 - c.GetIdleNode(2).Config.AntiEntropy.Interval = 0 - err := c.Start() - if err != nil { - t.Fatalf("starting cluster: %v", err) - } - defer c.Close() - - _, err = c.GetNode(0).API.CreateIndex(context.Background(), c.Idx(), pilosa.IndexOptions{}) - if err != nil { - t.Fatalf("creating index i: %v", err) - } - _, err = c.GetNode(0).API.CreateField(context.Background(), c.Idx(), "f", pilosa.OptFieldTypeSet(pilosa.DefaultCacheType, pilosa.DefaultCacheSize)) - if err != nil { - t.Fatalf("creating field f: %v", err) - } - - blockEdge := uint64(pilosa.HashBlockSize) - - hldr0 := &test.Holder{Holder: c.GetNode(0).Server.Holder()} - hldr1 := &test.Holder{Holder: c.GetNode(1).Server.Holder()} - hldr2 := &test.Holder{Holder: c.GetNode(2).Server.Holder()} - - // Set data on the local holder. - hldr0.SetBit(c.Idx(), "f", blockEdge-1, 10) - hldr0.SetBit(c.Idx(), "f", blockEdge, 20) - - // Set the same data on one of the replicas - // so that we have a quorum. - hldr1.SetBit(c.Idx(), "f", blockEdge-1, 10) - hldr1.SetBit(c.Idx(), "f", blockEdge, 20) - - // Leave the third replica empty to force a block merge. - // - err = c.GetNode(0).Server.SyncData() - if err != nil { - t.Fatalf("syncing node 0: %v", err) - } - - // Verify data is the same on all nodes. - for i, hldr := range []*test.Holder{hldr0, hldr1, hldr2} { - if a := hldr.Row(c.Idx(), "f", blockEdge-1).Columns(); !reflect.DeepEqual(a, []uint64{10}) { - t.Errorf("unexpected columns(%d/block 0): %+v", i, a) - } - if a := hldr.Row(c.Idx(), "f", blockEdge).Columns(); !reflect.DeepEqual(a, []uint64{20}) { - t.Errorf("unexpected columns(%d/block 1): %+v", i, a) - } - } -} - -// Ensure holder correctly handles clears during block sync. -func TestHolderSyncer_Clears(t *testing.T) { - c := test.MustUnsharedCluster(t, 3) - c.GetIdleNode(0).Config.Cluster.ReplicaN = 3 - c.GetIdleNode(0).Config.AntiEntropy.Interval = 0 - c.GetIdleNode(1).Config.Cluster.ReplicaN = 3 - c.GetIdleNode(1).Config.AntiEntropy.Interval = 0 - c.GetIdleNode(2).Config.Cluster.ReplicaN = 3 - c.GetIdleNode(2).Config.AntiEntropy.Interval = 0 - err := c.Start() - if err != nil { - t.Fatalf("starting cluster: %v", err) - } - defer c.Close() - - _, err = c.GetNode(0).API.CreateIndex(context.Background(), c.Idx(), pilosa.IndexOptions{}) - if err != nil { - t.Fatalf("creating index i: %v", err) - } - _, err = c.GetNode(0).API.CreateField(context.Background(), c.Idx(), "f", pilosa.OptFieldTypeSet(pilosa.DefaultCacheType, pilosa.DefaultCacheSize)) - if err != nil { - t.Fatalf("creating field f: %v", err) - } - - hldr0 := &test.Holder{Holder: c.GetNode(0).Server.Holder()} - hldr1 := &test.Holder{Holder: c.GetNode(1).Server.Holder()} - hldr2 := &test.Holder{Holder: c.GetNode(2).Server.Holder()} - - // Set data on the local holder that should be cleared - // because it's the only instance of this value. - hldr0.SetBit(c.Idx(), "f", 0, 30) - - // Set similar data on the replicas, but - // different from what's on local. This should end - // up being set on all replicas - hldr1.SetBit(c.Idx(), "f", 0, 20) - hldr2.SetBit(c.Idx(), "f", 0, 20) - - err = c.GetNode(0).Server.SyncData() - if err != nil { - t.Fatalf("syncing node 0: %v", err) - } - - // Verify data is the same on all nodes. - for i, hldr := range []*test.Holder{hldr0, hldr1, hldr2} { - if a := hldr.Row(c.Idx(), "f", 0).Columns(); !reflect.DeepEqual(a, []uint64{20}) { - t.Errorf("unexpected columns(%d): %+v", i, a) - } - } -} - -// Ensure holder can sync time quantum views with a remote holder. -func TestHolderSyncer_TimeQuantum(t *testing.T) { - c := test.MustUnsharedCluster(t, 2) - c.GetIdleNode(0).Config.Cluster.ReplicaN = 2 - c.GetIdleNode(0).Config.AntiEntropy.Interval = 0 - c.GetIdleNode(1).Config.Cluster.ReplicaN = 2 - c.GetIdleNode(1).Config.AntiEntropy.Interval = 0 - err := c.Start() - if err != nil { - t.Fatalf("starting cluster: %v", err) - } - defer c.Close() - - quantum := "D" - - _, err = c.GetNode(0).API.CreateIndex(context.Background(), c.Idx(), pilosa.IndexOptions{}) - if err != nil { - t.Fatalf("creating index i: %v", err) - } - _, err = c.GetNode(0).API.CreateField(context.Background(), c.Idx(), "f", pilosa.OptFieldTypeTime(pilosa.TimeQuantum(quantum), "0")) - if err != nil { - t.Fatalf("creating field f: %v", err) - } - - hldr0 := &test.Holder{Holder: c.GetNode(0).Server.Holder()} - hldr1 := &test.Holder{Holder: c.GetNode(1).Server.Holder()} - - // Set data on the local holder for node0. - t1 := time.Date(2018, 8, 1, 12, 30, 0, 0, time.UTC) - t2 := time.Date(2018, 8, 2, 12, 30, 0, 0, time.UTC) - hldr0.SetBitTime(c.Idx(), "f", 0, 1, &t1) - hldr0.SetBitTime(c.Idx(), "f", 0, 2, &t2) - - // Set data on node1. - hldr1.SetBitTime(c.Idx(), "f", 0, 22, &t2) - - err = c.GetNode(0).Server.SyncData() - if err != nil { - t.Fatalf("syncing node 0: %v", err) - } - - // Verify data is the same on both nodes. - for i, hldr := range []*test.Holder{hldr0, hldr1} { - if a := hldr.RowTime(c.Idx(), "f", 0, t1, quantum).Columns(); !reflect.DeepEqual(a, []uint64{1}) { - t.Errorf("unexpected columns(%d/0): %+v", i, a) - } - if a := hldr.RowTime(c.Idx(), "f", 0, t2, quantum).Columns(); !reflect.DeepEqual(a, []uint64{2, 22}) { - t.Errorf("unexpected columns(%d/0): %+v", i, a) - } - } -} - -// Ensure holder can sync integer views with a remote holder. -func TestHolderSyncer_IntField(t *testing.T) { - t.Run("BasicSync", func(t *testing.T) { - c := test.MustUnsharedCluster(t, 2) - c.GetIdleNode(0).Config.Cluster.ReplicaN = 2 - c.GetIdleNode(0).Config.AntiEntropy.Interval = 0 - c.GetIdleNode(1).Config.Cluster.ReplicaN = 2 - c.GetIdleNode(1).Config.AntiEntropy.Interval = 0 - err := c.Start() - if err != nil { - t.Fatalf("starting cluster: %v", err) - } - defer c.Close() - - var idx0 *pilosa.Index - idx0, err = c.GetNode(0).API.CreateIndex(context.Background(), c.Idx(), pilosa.IndexOptions{}) - _ = idx0 - if err != nil { - t.Fatalf("creating index i: %v", err) - } - _, err = c.GetNode(0).API.CreateField(context.Background(), c.Idx(), "f", pilosa.OptFieldTypeInt(0, 100)) - if err != nil { - t.Fatalf("creating field f: %v", err) - } - - hldr0 := &test.Holder{Holder: c.GetNode(0).Server.Holder()} - hldr1 := &test.Holder{Holder: c.GetNode(1).Server.Holder()} - - // Set data on the local holder for node0. columnID=1, value=1 - hldr0.SetValue(c.Idx(), "f", 1, 1) - - // in c0 expect the 1 bit - - // Set data on node1. columnID=2, value=2 - idx1 := hldr1.SetValue(c.Idx(), "f", 2, 2) - _ = idx1 - - err = c.GetNode(0).Server.SyncData() - if err != nil { - t.Fatalf("syncing node 0: %v", err) - } - - // expect 3 rows, the 1 bit + 2 rows for the 2 value as BSI. But, we only see that c0 overwrote c1. - - // Problem is: data at c1 was replaced by c0, instead of being merged with existing c1. - // Problem is: data at c0 did not receive and merge the c1 data. - - // Verify data is the same on both nodes. - for i, hldr := range []*test.Holder{hldr0, hldr1} { - if a, exists := hldr.Value(c.Idx(), "f", 1); !exists || a != 1 { - // expects exists==true, a==1 - t.Errorf("unexpected value(node%d/0): a:%d, exists: %v", i, a, exists) - } - if a, exists := hldr.Value(c.Idx(), "f", 2); exists { - t.Errorf("unexpected value(node%d/1): a:%d, exists: %v", i, a, exists) - } - } - }) - - t.Run("MultiShard", func(t *testing.T) { - t.Skip() // skipping due to changed partitioning strategy - c := test.MustUnsharedCluster(t, 2) - c.GetIdleNode(0).Config.Cluster.ReplicaN = 2 - c.GetIdleNode(0).Config.AntiEntropy.Interval = 0 - c.GetIdleNode(1).Config.Cluster.ReplicaN = 2 - c.GetIdleNode(1).Config.AntiEntropy.Interval = 0 - err := c.Start() - if err != nil { - t.Fatalf("starting cluster: %v", err) - } - defer c.Close() - - var idx0 *pilosa.Index - _ = idx0 - idx0, err = c.GetNode(0).API.CreateIndex(context.Background(), c.Idx(), pilosa.IndexOptions{}) - _ = idx0 - if err != nil { - t.Fatalf("creating index i: %v", err) - } - _, err = c.GetNode(0).API.CreateField(context.Background(), c.Idx(), "f", pilosa.OptFieldTypeInt(math.MinInt64, math.MaxInt64)) - if err != nil { - t.Fatalf("creating field f: %v", err) - } - - hldr0 := &test.Holder{Holder: c.GetNode(0).Server.Holder()} - hldr1 := &test.Holder{Holder: c.GetNode(1).Server.Holder()} - - // Set data on the local holder for node0. - hldr0.SetValue(c.Idx(), "f", 1*pilosa.ShardWidth, 11) - hldr0.SetValue(c.Idx(), "f", 3*pilosa.ShardWidth, 32) - hldr0.SetValue(c.Idx(), "f", 4*pilosa.ShardWidth, math.MinInt32) - hldr0.SetValue(c.Idx(), "f", 7*pilosa.ShardWidth, math.MinInt32) - - // Set data on node1. - hldr1.SetValue(c.Idx(), "f", 0*pilosa.ShardWidth, 2) - hldr1.SetValue(c.Idx(), "f", 2*pilosa.ShardWidth, 22) - hldr1.SetValue(c.Idx(), "f", 4*pilosa.ShardWidth, math.MaxInt32) - hldr1.SetValue(c.Idx(), "f", 7*pilosa.ShardWidth, math.MaxInt32) - - // Primary for shards (for index c.Idx()): - // node0: [0,3,7] - // node1: [1,2,4] - - err = c.GetNode(0).Server.SyncData() - if err != nil { - t.Fatalf("syncing node 0: %v", err) - } - err = c.GetNode(1).Server.SyncData() - if err != nil { - t.Fatalf("syncing node 1: %v", err) - } - - // dump the rbf keys for both c0 and c1 - - // Verify data is the same on both nodes. - for i, hldr := range []*test.Holder{hldr0, hldr1} { - if a := hldr.Range(c.Idx(), "f", pql.GT, 0); !reflect.DeepEqual(a.Columns(), []uint64{2 * pilosa.ShardWidth, 3 * pilosa.ShardWidth, 4 * pilosa.ShardWidth}) { - t.Errorf("unexpected columns(node%d/0): %d", i, a.Columns()) - } - if a := hldr.Range(c.Idx(), "f", pql.LT, 0); !reflect.DeepEqual(a.Columns(), []uint64{7 * pilosa.ShardWidth}) { - t.Errorf("unexpected columns(node%d/0): %d", i, a.Columns()) - } - } - }) -} diff --git a/http_handler.go b/http_handler.go index 8131da3e1..1b7d16220 100644 --- a/http_handler.go +++ b/http_handler.go @@ -2763,62 +2763,12 @@ func (h *Handler) handleGetNodes(w http.ResponseWriter, r *http.Request) { // handleGetFragmentBlockData handles GET /internal/fragment/block/data requests. func (h *Handler) handleGetFragmentBlockData(w http.ResponseWriter, r *http.Request) { - buf, err := h.api.FragmentBlockData(r.Context(), r.Body) - if err != nil { - if _, ok := err.(BadRequestError); ok { - http.Error(w, err.Error(), http.StatusBadRequest) - } else if errors.Cause(err) == ErrFragmentNotFound { - http.Error(w, err.Error(), http.StatusNotFound) - } else { - http.Error(w, err.Error(), http.StatusInternalServerError) - } - return - } - - // Write response. - w.Header().Set("Content-Type", "application/protobuf") - w.Header().Set("Content-Length", strconv.Itoa(len(buf))) - _, err = w.Write(buf) - if err != nil { - h.logger.Errorf("writing fragment/block/data response: %v", err) - } + http.Error(w, "fragment blocks feature removed", http.StatusNotFound) } // handleGetFragmentBlocks handles GET /internal/fragment/blocks requests. func (h *Handler) handleGetFragmentBlocks(w http.ResponseWriter, r *http.Request) { - if !validHeaderAcceptJSON(r.Header) { - http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable) - return - } - // Read shard parameter. - q := r.URL.Query() - shard, err := strconv.ParseUint(q.Get("shard"), 10, 64) - if err != nil { - http.Error(w, "shard required", http.StatusBadRequest) - return - } - - blocks, err := h.api.FragmentBlocks(r.Context(), q.Get("index"), q.Get("field"), q.Get("view"), shard) - if err != nil { - if errors.Cause(err) == ErrFragmentNotFound { - http.Error(w, err.Error(), http.StatusNotFound) - } else { - http.Error(w, err.Error(), http.StatusInternalServerError) - } - return - } - - // Encode response. - w.Header().Set("Content-Type", "application/json") - if err := json.NewEncoder(w).Encode(getFragmentBlocksResponse{ - Blocks: blocks, - }); err != nil { - h.logger.Errorf("block response encoding error: %s", err) - } -} - -type getFragmentBlocksResponse struct { - Blocks []FragmentBlock `json:"blocks"` + http.Error(w, "fragment blocks feature removed", http.StatusNotFound) } // handleGetFragmentData handles GET /internal/fragment/data requests. diff --git a/internal_client.go b/internal_client.go index 4938b459e..6dfa599f8 100644 --- a/internal_client.go +++ b/internal_client.go @@ -1133,102 +1133,6 @@ func (c *InternalClient) CreateFieldWithOptions(ctx context.Context, index, fiel return errors.Wrap(resp.Body.Close(), "closing response body") } -// FragmentBlocks returns a list of block checksums for a fragment on a host. -// Only returns blocks which contain data. -func (c *InternalClient) FragmentBlocks(ctx context.Context, uri *pnet.URI, index, field, view string, shard uint64) ([]FragmentBlock, error) { - span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.FragmentBlocks") - defer span.Finish() - - if uri == nil { - uri = c.defaultURI - } - u := uriPathToURL(uri, "/internal/fragment/blocks") - u.RawQuery = url.Values{ - "index": {index}, - "field": {field}, - "view": {view}, - "shard": {strconv.FormatUint(shard, 10)}, - }.Encode() - - // Build request. - req, err := http.NewRequest("GET", u.String(), nil) - if err != nil { - return nil, errors.Wrap(err, "creating request") - } - - req.Header.Set("User-Agent", "pilosa/"+Version) - req.Header.Set("Accept", "application/json") - AddAuthToken(ctx, &req.Header) - - // Execute request. - resp, err := c.executeRequest(req.WithContext(ctx)) - if err != nil { - // Return the appropriate error. - if resp != nil && resp.StatusCode == http.StatusNotFound { - return nil, ErrFragmentNotFound - } - return nil, err - } - defer resp.Body.Close() - - // Decode response object. - var rsp getFragmentBlocksResponse - if err := json.NewDecoder(resp.Body).Decode(&rsp); err != nil { - return nil, errors.Wrap(err, "decoding") - } - return rsp.Blocks, nil -} - -// BlockData returns row/column id pairs for a block. -func (c *InternalClient) BlockData(ctx context.Context, uri *pnet.URI, index, field, view string, shard uint64, block int) ([]uint64, []uint64, error) { - span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.BlockData") - defer span.Finish() - - if uri == nil { - panic("need to pass a URI to BlockData") - } - buf, err := c.serializer.Marshal(&BlockDataRequest{ - Index: index, - Field: field, - View: view, - Shard: shard, - Block: uint64(block), - }) - if err != nil { - return nil, nil, errors.Wrap(err, "marshaling") - } - - u := uriPathToURL(uri, "/internal/fragment/block/data") - req, err := http.NewRequest("GET", u.String(), bytes.NewReader(buf)) - if err != nil { - return nil, nil, errors.Wrap(err, "creating request") - } - req.Header.Set("Content-Type", "application/protobuf") - req.Header.Set("Content-Length", strconv.Itoa(len(buf))) - req.Header.Set("Accept", "application/protobuf") - req.Header.Set("X-Pilosa-Row", "roaring") - req.Header.Set("User-Agent", "pilosa/"+Version) - AddAuthToken(ctx, &req.Header) - - resp, err := c.executeRequest(req.WithContext(ctx)) - if err != nil { - if resp != nil && resp.StatusCode == http.StatusNotFound { - return nil, nil, nil - } - return nil, nil, err - } - defer resp.Body.Close() - - // Decode response object. - var rsp BlockDataResponse - if body, err := io.ReadAll(resp.Body); err != nil { - return nil, nil, errors.Wrap(err, "reading") - } else if err := c.serializer.Unmarshal(body, &rsp); err != nil { - return nil, nil, errors.Wrap(err, "unmarshalling") - } - return rsp.RowIDs, rsp.ColumnIDs, nil -} - // SendMessage posts a message synchronously. func (c *InternalClient) SendMessage(ctx context.Context, uri *pnet.URI, msg []byte) error { span, ctx := tracing.StartSpanFromContext(ctx, "InternalClient.SendMessage") diff --git a/internal_client_test.go b/internal_client_test.go index dc61b1881..fd7bc3d8f 100644 --- a/internal_client_test.go +++ b/internal_client_test.go @@ -1181,40 +1181,6 @@ func TestClient_ImportExistence(t *testing.T) { }) } -// Ensure client can retrieve a list of all checksums for blocks in a fragment. -func TestClient_FragmentBlocks(t *testing.T) { - cluster := test.MustRunCluster(t, 1) - defer cluster.Close() - cmd := cluster.GetNode(0) - - holder := cmd.Server.Holder() - hldr := test.Holder{Holder: holder} - - hldr.SetBit(cluster.Idx(), "f", 0, 1) - hldr.SetBit(cluster.Idx(), "f", pilosa.HashBlockSize*3, 100) - - // Set a bit on a different shard. - hldr.SetBit(cluster.Idx(), "f", 0, 1) - c := MustNewClient(cmd.URL(), pilosa.GetHTTPClient(nil)) - blocks, err := c.FragmentBlocks(context.Background(), nil, cluster.Idx(), "f", "standard", 0) - if err != nil { - t.Fatal(err) - } else if len(blocks) != 2 { - t.Fatalf("unexpected blocks: %s", spew.Sdump(blocks)) - } else if blocks[0].ID != 0 { - t.Fatalf("unexpected block id(0): %d", blocks[0].ID) - } else if blocks[1].ID != 3 { - t.Fatalf("unexpected block id(1): %d", blocks[1].ID) - } - - // Verify data matches local blocks. - if a, err := cmd.API.FragmentBlocks(context.Background(), cluster.Idx(), "f", "standard", 0); err != nil { - t.Fatal(err) - } else if !reflect.DeepEqual(a, blocks) { - t.Fatalf("blocks mismatch:\n\nexp=%s\n\ngot=%s\n\n", spew.Sdump(a), spew.Sdump(blocks)) - } -} - // Try to request translation data which won't exist. func TestClient_IndexTranslateDataReader(t *testing.T) { cluster := test.MustRunCluster(t, 1) diff --git a/iterator.go b/iterator.go index abca0be3e..dd490bb26 100644 --- a/iterator.go +++ b/iterator.go @@ -66,46 +66,6 @@ func (itr *bufIterator) Unread() { itr.buf.full = true } -// limitIterator wraps an Iterator and limits it to a max column/row pair. -type limitIterator struct { - itr iterator - maxRowID uint64 - maxColumnID uint64 - - eof bool -} - -// newLimitIterator returns a new LimitIterator. -func newLimitIterator(itr iterator, maxRowID, maxColumnID uint64) *limitIterator { // nolint: unparam - return &limitIterator{ - itr: itr, - maxRowID: maxRowID, - maxColumnID: maxColumnID, - } -} - -// Seek moves the underlying iterator to a column/row pair. -func (itr *limitIterator) Seek(rowID, columnID uint64) { itr.itr.Seek(rowID, columnID) } - -// Next returns the next row/column ID pair. -// If the underlying iterator returns a pair higher than the max then EOF is returned. -func (itr *limitIterator) Next() (rowID, columnID 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). - rowID, columnID, eof = itr.itr.Next() - if eof || rowID > itr.maxRowID || (rowID == itr.maxRowID && columnID > itr.maxColumnID) { - itr.eof = true - return 0, 0, true - } - - return rowID, columnID, false -} - // sliceIterator iterates over a pair of row/column ID slices. type sliceIterator struct { rowIDs []uint64 @@ -158,25 +118,3 @@ func (itr *sliceIterator) Next() (rowID, columnID uint64, eof bool) { itr.i++ return rowID, columnID, false } - -// roaringIterator converts a roaring.Iterator to output column/row 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 * ShardWidth) + pseek) -} - -// Next returns the next column/row ID pair. -func (itr *roaringIterator) Next() (rowID, columnID uint64, eof bool) { - v, eof := itr.itr.Next() - return v / ShardWidth, v % ShardWidth, eof -} diff --git a/rbf.go b/rbf.go index 5c042d676..d5c460e01 100644 --- a/rbf.go +++ b/rbf.go @@ -362,14 +362,6 @@ func (tx *RBFTx) ContainerIterator(index, field, view string, shard uint64, key return tx.tx.ContainerIterator(rbfName(index, field, view, shard), key) } -func (tx *RBFTx) ForEach(index, field, view string, shard uint64, fn func(i uint64) error) error { - return tx.tx.ForEach(rbfName(index, field, view, shard), fn) -} - -func (tx *RBFTx) ForEachRange(index, field, view string, shard uint64, start, end uint64, fn func(uint64) error) error { - return tx.tx.ForEachRange(rbfName(index, field, view, shard), start, end, fn) -} - func (tx *RBFTx) Count(index, field, view string, shard uint64) (uint64, error) { return tx.tx.Count(rbfName(index, field, view, shard)) } @@ -396,12 +388,6 @@ func (tx *RBFTx) ImportRoaringBits(index, field, view string, shard uint64, rit return tx.tx.ImportRoaringBits(rbfName(index, field, view, shard), rit, clear, log, rowSize) } -func (tx *RBFTx) NewTxIterator(index, field, view string, shard uint64) *roaring.Iterator { - b, err := tx.RoaringBitmap(index, field, view, shard) - vprint.PanicOn(err) - return b.Iterator() -} - func (tx *RBFTx) ApplyFilter(index, field, view string, shard uint64, ckey uint64, filter roaring.BitmapFilter) (err error) { return tx.tx.ApplyFilter(rbfName(index, field, view, shard), ckey, filter) } diff --git a/rbf/cursor_test.go b/rbf/cursor_test.go index ad2b536b6..1052f08d0 100644 --- a/rbf/cursor_test.go +++ b/rbf/cursor_test.go @@ -1172,85 +1172,6 @@ func TestCursor_GenerateAll(t *testing.T) { } } -// test ForEachRange handles Bitmaps, because BitmapPtr -// case was missing -func TestForEachRange(t *testing.T) { - db := MustOpenDB(t) - defer MustCloseDB(t, db) - tx := MustBegin(t, db, true) - defer tx.Rollback() - - if err := tx.CreateBitmap("x"); err != nil { - t.Fatal(err) - } - - rb := roaring.NewBitmap() - bits := make([]uint64, rbf.BitmapN) - n := 0 - for i := range bits { - bits[i] = 15 // ^uint64(0) - n += 4 - if n > rbf.ArrayMaxSize { - break - } - } - - rb.Put(0, roaring.NewContainerBitmap(n, bits)) - crun := roaring.NewContainerRun([]roaring.Interval16{{Start: 0, Last: 1<<16 - 1}}) - rb.Put(1, crun) - rb.Put(2, roaring.NewContainerArray([]uint16{1, 1024, 1<<16 - 1})) - - // setup to delete random bits down - values := rb.Slice() - valmap := make(map[uint64]bool) - for _, v := range values { - valmap[v] = true - } - - _, err := tx.AddRoaring("x", rb) - if err != nil { - t.Errorf("Add Roaring Failed %v", err) - } - - c, err := tx.Cursor("x") - if err != nil { - t.Fatal(err) - } - c.DebugSlowCheckAllPages() - - if err := c.First(); err != nil { - t.Fatal(err) - } - - exists, err := c.Contains(0x3) - if err != nil { - t.Fatalf("ERR:%v", err) - } - if !exists { - t.Fatalf("Should Contain %v", 0x3) - } - - exists, err = c.Contains(0x4) - if err != nil { - t.Fatalf("ERR:%v", err) - } - if exists { - t.Fatalf("Should Not Contain %v", 0x4) - } - - c.DebugSlowCheckAllPages() - - _ = tx.ForEach("x", func(i uint64) error { - delete(valmap, i) - return nil - }) - - // check it is empty - if len(valmap) != 0 { - t.Fatalf("expected empty container, but see %v values left: '%#v'", len(valmap), valmap) - } -} - func TestCursor_PutContainer(t *testing.T) { t.Run("BitmapToArray", func(t *testing.T) { db := MustOpenDB(t) diff --git a/rbf/tx.go b/rbf/tx.go index 78d37b988..1a52c69d6 100644 --- a/rbf/tx.go +++ b/rbf/tx.go @@ -6,7 +6,6 @@ import ( "bufio" "fmt" "io" - "math" "sort" "strings" "sync" @@ -1416,109 +1415,6 @@ func (tx *Tx) ApplyFilter(name string, key uint64, filter roaring.BitmapFilter) return f.ApplyFilter() } -func (tx *Tx) ForEach(name string, fn func(i uint64) error) error { - return tx.ForEachRange(name, 0, math.MaxUint64, fn) -} - -func (tx *Tx) ForEachRange(name string, start, end uint64, fn func(uint64) error) error { - tx.mu.RLock() - defer tx.mu.RUnlock() - - c, err := tx.cursor(name) - if err == ErrBitmapNotFound { - return nil - } else if err != nil { - return err - } - defer c.Close() - - if _, err := c.Seek(highbits(start)); err != nil { - return err - } - - for { - if err := c.Next(); err == io.EOF { - return nil - } else if err != nil { - return err - } - - elem := &c.stack.elems[c.stack.top] - leafPage, _, err := c.tx.readPage(elem.pgno) - if err != nil { - return err - } - cell := readLeafCell(leafPage, elem.index) - - switch cell.Type { - case ContainerTypeArray: - for _, lo := range toArray16(cell.Data) { - v := cell.Key<<16 | uint64(lo) - if v < start { - continue - } else if v > end { - return nil - } else if err := fn(v); err != nil { - return err - } - } - case ContainerTypeRLE: - for _, r := range toInterval16(cell.Data) { - for lo := int(r.Start); lo <= int(r.Last); lo++ { - v := cell.Key<<16 | uint64(lo) - if v < start { - continue - } else if v > end { - return nil - } else if err := fn(v); err != nil { - return err - } - } - } - case ContainerTypeBitmap: - for i, bits := range toArray64(cell.Data) { - for j := uint(0); j < 64; j++ { - if bits&(1< end { - return nil - } else if err := fn(v); err != nil { - return err - } - } - } - case ContainerTypeBitmapPtr: - _, bm, err := c.tx.leafCellBitmap(toPgno(cell.Data)) - if err != nil { - return err - } - for i, bits := range bm { - for j := uint(0); j < 64; j++ { - if bits&(1< end { - return nil - } else if err := fn(v); err != nil { - return err - } - } - } - default: - vprint.PanicOn(fmt.Sprintf("invalid container type: %d", cell.Type)) - } - } -} - func (tx *Tx) Count(name string) (uint64, error) { tx.mu.RLock() defer tx.mu.RUnlock() diff --git a/roaring/filter.go b/roaring/filter.go index c7a37ee1b..68c76163b 100644 --- a/roaring/filter.go +++ b/roaring/filter.go @@ -898,7 +898,7 @@ func (b *BitmapMutexDupFilter) ConsiderKey(key FilterKey, n int32) FilterResult func (b *BitmapMutexDupFilter) ConsiderData(key FilterKey, data *Container) FilterResult { value, basePos := uint64(key)>>rowExponent, uint64(key&keyMask)<<16 - containerCallback(data, func(u uint16) { + ContainerCallback(data, func(u uint16) { pos := basePos + uint64(u) if b.first[pos] != ^uint64(0) { if b.details { diff --git a/roaring/roaring.go b/roaring/roaring.go index 5603199c3..76e5e734a 100644 --- a/roaring/roaring.go +++ b/roaring/roaring.go @@ -4407,7 +4407,7 @@ func intersectionAnyBitmapBitmap(a, b *Container) bool { return false } -func containerCallback(a *Container, fn func(uint16)) { +func ContainerCallback(a *Container, fn func(uint16)) { if a.N() == 0 { return } @@ -4437,11 +4437,11 @@ func containerCallback(a *Container, fn func(uint16)) { func intersectionCallback(a, b *Container, fn func(uint16)) { if a.N() == MaxContainerVal+1 { - containerCallback(b, fn) + ContainerCallback(b, fn) return } if b.N() == MaxContainerVal+1 { - containerCallback(a, fn) + ContainerCallback(a, fn) return } if a.N() == 0 || b.N() == 0 { diff --git a/roaring/roaring_internal_test.go b/roaring/roaring_internal_test.go index 7317f2ac1..c86bfc046 100644 --- a/roaring/roaring_internal_test.go +++ b/roaring/roaring_internal_test.go @@ -4659,7 +4659,7 @@ func TestContainerCallback(t *testing.T) { for _, c1 := range ci { got = got[:0] expected = c1.Slice() - containerCallback(c1, hit) + ContainerCallback(c1, hit) if len(got) != len(expected) { complain(t, "wrong length (%d vs %d)", len(expected), len(got)) } diff --git a/server.go b/server.go index 536c44fa5..67a250307 100644 --- a/server.go +++ b/server.go @@ -70,7 +70,6 @@ type Server struct { // nolint: maligned nodeID string uri pnet.URI grpcURI pnet.URI - antiEntropyInterval time.Duration metricInterval time.Duration diagnosticInterval time.Duration viewsRemovalInterval time.Duration @@ -160,15 +159,6 @@ func OptServerDataDir(dir string) ServerOption { } } -// OptServerAntiEntropyInterval is a functional option on Server -// used to set the anti-entropy interval. -func OptServerAntiEntropyInterval(interval time.Duration) ServerOption { - return func(s *Server) error { - s.antiEntropyInterval = interval - return nil - } -} - // OptServerViewsRemovalInterval is a functional option on Server // used to set the ttl removal interval. func OptServerViewsRemovalInterval(interval time.Duration) ServerOption { @@ -449,7 +439,6 @@ func NewServer(opts ...ServerOption) (*Server, error) { gcNotifier: NopGCNotifier, - antiEntropyInterval: 0, metricInterval: 0, diagnosticInterval: 0, viewsRemovalInterval: time.Hour, @@ -486,7 +475,6 @@ func NewServer(opts ...ServerOption) (*Server, error) { return nil, errors.Wrap(err, "applying option") } } - s.holderConfig.AntiEntropyInterval = s.antiEntropyInterval memTotal, err := s.systemInfo.MemTotal() if err != nil { @@ -652,10 +640,9 @@ func (s *Server) Open() error { return errors.Wrap(err, "setting nodeState") } - if ok := s.addToWaitGroup(4); !ok { + if ok := s.addToWaitGroup(3); !ok { return fmt.Errorf("closing server while opening server is NOT allowed") } - go func() { defer s.wg.Done(); s.monitorAntiEntropy() }() go func() { defer s.wg.Done(); s.monitorRuntime() }() go func() { defer s.wg.Done(); s.monitorDiagnostics() }() go func() { defer s.wg.Done(); s.monitorViewsRemoval() }() @@ -805,12 +792,6 @@ func (s *Server) Close() error { // NodeID returns the server's node id. func (s *Server) NodeID() string { return s.nodeID } -// SyncData manually invokes the anti entropy process which makes sure that this -// node has the data from all replicas across the cluster. -func (s *Server) SyncData() error { - return errors.Wrap(s.syncer.SyncHolder(), "syncing holder") -} - // monitorResetTranslationSync is a background process which // listens for events indicating the need to reset the translation // sync processes. @@ -917,73 +898,6 @@ func (s *Server) ViewsRemoval(ctx context.Context) { } } -func (s *Server) monitorAntiEntropy() { - // %% begin sonarcloud ignore %% - // This code isn't really used anymore because of problems with the design, - // but we haven't taken it out yet. But there's no code coverage of it. - if s.antiEntropyInterval == 0 || s.cluster.ReplicaN <= 1 { - return // anti entropy disabled - } - s.cluster.initializeAntiEntropy() - - ticker := time.NewTicker(s.antiEntropyInterval) - defer ticker.Stop() - - s.logger.Infof("holder sync monitor initializing (%s interval)", s.antiEntropyInterval) - - // Initialize syncer with local holder and remote client. - for { - // Wait for tick or a close. - select { - case <-s.closing: - return - case <-s.cluster.abortAntiEntropyCh: - // receive here so we don't block resizing - // ... note that resizing is gone now, but I don't know whether we still need this. - continue - case <-ticker.C: - s.holder.Stats.Count(MetricAntiEntropy, 1, 1.0) - } - t := time.Now() - - // We used to check for resizing before doing anti-entropy, but resizing is out - // so we don't otherwise care about state. - _, err := s.cluster.State() - if err != nil { - s.logger.Printf("cluster state error: err=%s", err) - continue - } - - // Sync holders. - s.logger.Infof("holder sync beginning") - s.cluster.muAntiEntropy.Lock() - if err := s.syncer.SyncHolder(); err != nil { - s.cluster.muAntiEntropy.Unlock() - s.logger.Errorf("holder sync error: err=%s", err) - continue - } - s.cluster.muAntiEntropy.Unlock() - - // Record successful sync in log. - s.logger.Infof("holder sync complete") - dif := time.Since(t) - s.holder.Stats.Timing(MetricAntiEntropyDurationSeconds, dif, 1.0) - - // Drain tick channel since we just finished anti-entropy. If the AE - // process took a long time, we don't want them to pile up on each - // other. - for { - select { - case <-ticker.C: - continue - default: - } - break - } - } - // %% end sonarcloud ignore %% -} - // receiveMessage represents an implementation of BroadcastHandler. func (s *Server) receiveMessage(m Message) error { switch obj := m.(type) { diff --git a/server/config.go b/server/config.go index fc90d31f9..351cf8406 100644 --- a/server/config.go +++ b/server/config.go @@ -145,6 +145,7 @@ type Config struct { PrimaryURL string `toml:"primary-url"` } `toml:"translation"` + // AntiEntropy config is now deprecated AntiEntropy struct { Interval toml.Duration `toml:"interval"` } `toml:"anti-entropy"` diff --git a/server/server.go b/server/server.go index ec5536775..dfd23e520 100644 --- a/server/server.go +++ b/server/server.go @@ -405,6 +405,9 @@ func (m *Command) SetupServer() error { if m.Config.Translation.PrimaryURL != "" { m.logger.Infof("DEPRECATED: The primary-url configuration option is no longer used.") } + if m.Config.AntiEntropy.Interval != 0 { + m.logger.Infof("DEPRECATED: The anti-entropy configuration option is no longer used.") + } // Handle renamed and deprecated config parameter longQueryTime := m.Config.LongQueryTime if m.Config.Cluster.LongQueryTime >= 0 { @@ -446,7 +449,6 @@ func (m *Command) SetupServer() error { } serverOptions := []pilosa.ServerOption{ - pilosa.OptServerAntiEntropyInterval(time.Duration(m.Config.AntiEntropy.Interval)), pilosa.OptServerLongQueryTime(time.Duration(longQueryTime)), pilosa.OptServerDataDir(m.Config.DataDir), pilosa.OptServerReplicaN(m.Config.Cluster.ReplicaN), diff --git a/server_internal_test.go b/server_internal_test.go index eecea6ca0..a35ab1370 100644 --- a/server_internal_test.go +++ b/server_internal_test.go @@ -17,8 +17,7 @@ func TestMonitorAntiEntropyZero(t *testing.T) { t.Fatalf("getting temp dir: %v", err) } cfg := &storage.Config{FsyncEnabled: false, Backend: storage.DefaultBackend} - s, err := NewServer(OptServerDataDir(td), - OptServerAntiEntropyInterval(0), OptServerStorageConfig(cfg)) + s, err := NewServer(OptServerDataDir(td), OptServerStorageConfig(cfg)) if err != nil { t.Fatalf("making new server: %v", err) } @@ -26,7 +25,6 @@ func TestMonitorAntiEntropyZero(t *testing.T) { ch := make(chan struct{}) go func() { - s.monitorAntiEntropy() close(ch) }() diff --git a/stattx.go b/stattx.go index 08457a5e5..ec40537c0 100644 --- a/stattx.go +++ b/stattx.go @@ -151,8 +151,6 @@ const ( kRemove kContains kContainerIterator - kForEach - kForEachRange kCount kMax kMin @@ -188,10 +186,6 @@ func (k kall) String() string { return "kContains" case kContainerIterator: return "kContainerIterator" - case kForEach: - return "kForEach" - case kForEachRange: - return "kForEachRange" case kCount: return "kCount" case kMax: @@ -213,16 +207,6 @@ func (k kall) String() string { var _ Tx = (*statTx)(nil) -func (c *statTx) NewTxIterator(index, field, view string, shard uint64) *roaring.Iterator { - me := kNewTxIterator - - t0 := time.Now() - defer func() { - c.stats.add(me, time.Since(t0)) - }() - return c.b.NewTxIterator(index, field, view, shard) -} - func (c *statTx) ImportRoaringBits(index, field, view string, shard uint64, rit roaring.RoaringIterator, clear bool, log bool, rowSize uint64) (changed int, rowSet map[uint64]int, err error) { me := kImportRoaringBits @@ -416,40 +400,6 @@ func (c *statTx) ApplyRewriter(index, field, view string, shard uint64, ckey uin return c.b.ApplyRewriter(index, field, view, shard, ckey, filter) } -func (c *statTx) ForEach(index, field, view string, shard uint64, fn func(i uint64) error) error { - me := kForEach - - t0 := time.Now() - defer func() { - c.stats.add(me, time.Since(t0)) - }() - - defer func() { - if r := recover(); r != nil { - vprint.AlwaysPrintf("see ForEach() PanicOn '%v' at '%v'", r, vprint.Stack()) - vprint.PanicOn(r) - } - }() - return c.b.ForEach(index, field, view, shard, fn) -} - -func (c *statTx) ForEachRange(index, field, view string, shard uint64, start, end uint64, fn func(uint64) error) error { - me := kForEachRange - - t0 := time.Now() - defer func() { - c.stats.add(me, time.Since(t0)) - }() - - defer func() { - if r := recover(); r != nil { - vprint.AlwaysPrintf("see ForEachRange() PanicOn '%v' at '%v'", r, vprint.Stack()) - vprint.PanicOn(r) - } - }() - return c.b.ForEachRange(index, field, view, shard, start, end, fn) -} - func (c *statTx) Count(index, field, view string, shard uint64) (uint64, error) { me := kCount diff --git a/tx.go b/tx.go index f85ef2fa9..3a4079f5b 100644 --- a/tx.go +++ b/tx.go @@ -51,11 +51,6 @@ type Tx interface { // Commit makes the updates in the Tx visible to subsequent transactions. Commit() error - // NewTxIterator returns a *roaring.Iterator whose Next() method will - // successively return each uint64 stored in the conceptual roaring.Bitmap - // for the specified fragment. - NewTxIterator(index, field, view string, shard uint64) *roaring.Iterator - // ContainerIterator loops over the containers in the conceptual // roaring.Bitmap for the specified fragment. // Calling Next() on the returned roaring.ContainerIterator gives @@ -109,14 +104,6 @@ type Tx interface { // Contains tests if the uint64 v is stored in the fragment's Bitmap. Contains(index, field, view string, shard uint64, v uint64) (exists bool, err error) - // ForEach calls function `fn` on every value (bit set) in the Bitmap for - // the fragment. - ForEach(index, field, view string, shard uint64, fn func(i uint64) error) error - - // ForEachRange calls function `fn` on every value (bit set) in the Bitmap for - // the fragment, limited to the [start, end) range. - ForEachRange(index, field, view string, shard uint64, start, end uint64, fn func(uint64) error) error - // Count returns the count of hot bits on the fragment. Count(index, field, view string, shard uint64) (uint64, error)