mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-08-28 02:44:59 +00:00
drop anti-entropy feature, since it doesn't work
The anti-entropy feature has never actually worked. We've been talking about removing it or replacing it for ages, but haven't had a concrete motivation. But the anti-entropy interface is the sole user of several components of the Tx interface, and now that we're trying to replace that interface, being able to drop those components has some appeal, so let's remove the one thing that used them, in the hopes that this will simplify life. This also lets us drop ForEach and ForEachRange, which were barely used at all. The one surviving usage (CSV export) can be handled by using the container iterator we already have, and making ContainerCallback exported so we can use it to just call things for every bit.
This commit is contained in:
Seebs
seebs
parent
dd30168b1c
commit
fff9ddc1f5
24 changed files with 71 additions and 2156 deletions
82
api.go
82
api.go
|
|
@ -761,9 +761,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.
|
||||
|
|
@ -803,66 +819,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")
|
||||
|
|
|
|||
26
catcher.go
26
catcher.go
|
|
@ -26,10 +26,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 {
|
||||
|
|
@ -149,28 +145,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() {
|
||||
|
|
|
|||
30
cluster.go
30
cluster.go
|
|
@ -51,9 +51,6 @@ type cluster struct { // nolint: maligned
|
|||
holder *Holder
|
||||
broadcaster broadcaster
|
||||
|
||||
abortAntiEntropyCh chan struct{}
|
||||
muAntiEntropy sync.Mutex
|
||||
|
||||
translationSyncer TranslationSyncer
|
||||
|
||||
mu sync.RWMutex
|
||||
|
|
@ -95,33 +92,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()
|
||||
}
|
||||
|
|
|
|||
|
|
@ -7,7 +7,6 @@ import (
|
|||
"reflect"
|
||||
"testing"
|
||||
"testing/quick"
|
||||
"time"
|
||||
|
||||
"github.com/molecula/featurebase/v3/disco"
|
||||
pnet "github.com/molecula/featurebase/v3/net"
|
||||
|
|
@ -179,63 +178,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")
|
||||
}
|
||||
})
|
||||
}
|
||||
|
|
|
|||
681
fragment.go
681
fragment.go
|
|
@ -6,9 +6,7 @@ import (
|
|||
"bytes"
|
||||
"container/heap"
|
||||
"context"
|
||||
"encoding/binary"
|
||||
"fmt"
|
||||
"hash"
|
||||
"io"
|
||||
"math"
|
||||
"math/bits"
|
||||
|
|
@ -20,11 +18,8 @@ import (
|
|||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/cespare/xxhash"
|
||||
"github.com/gogo/protobuf/proto"
|
||||
"github.com/molecula/featurebase/v3/disco"
|
||||
"github.com/molecula/featurebase/v3/logger"
|
||||
pnet "github.com/molecula/featurebase/v3/net"
|
||||
"github.com/molecula/featurebase/v3/pb"
|
||||
"github.com/molecula/featurebase/v3/pql"
|
||||
"github.com/molecula/featurebase/v3/roaring"
|
||||
|
|
@ -176,8 +171,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 {
|
||||
|
|
@ -1325,16 +1319,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) {
|
||||
|
|
@ -1516,270 +1500,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 {
|
||||
|
|
@ -2814,22 +2534,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)
|
||||
|
|
@ -3094,389 +2798,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)
|
||||
|
|
|
|||
|
|
@ -1145,36 +1145,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))
|
||||
|
|
@ -1387,110 +1357,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))
|
||||
|
|
@ -1591,30 +1457,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)
|
||||
|
|
|
|||
116
holder.go
116
holder.go
|
|
@ -201,9 +201,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
|
||||
}
|
||||
|
|
@ -1241,117 +1240,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 {
|
||||
|
|
|
|||
403
holder_test.go
403
holder_test.go
|
|
@ -2,16 +2,11 @@
|
|||
package pilosa_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"math"
|
||||
"os"
|
||||
"reflect"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
pilosa "github.com/molecula/featurebase/v3"
|
||||
"github.com/molecula/featurebase/v3/pql"
|
||||
"github.com/molecula/featurebase/v3/test"
|
||||
"github.com/pkg/errors"
|
||||
)
|
||||
|
|
@ -201,401 +196,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())
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
|
|
|
|||
|
|
@ -2761,62 +2761,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.
|
||||
|
|
|
|||
|
|
@ -1131,102 +1131,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")
|
||||
|
|
|
|||
|
|
@ -1180,40 +1180,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)
|
||||
|
|
|
|||
64
iterator.go
64
iterator.go
|
|
@ -3,8 +3,6 @@ package pilosa
|
|||
|
||||
import (
|
||||
"fmt"
|
||||
|
||||
"github.com/molecula/featurebase/v3/roaring"
|
||||
)
|
||||
|
||||
// iterator is an interface for looping over row/column pairs.
|
||||
|
|
@ -65,46 +63,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
|
||||
|
|
@ -157,25 +115,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
|
||||
}
|
||||
|
|
|
|||
15
rbf.go
15
rbf.go
|
|
@ -15,7 +15,6 @@ import (
|
|||
txkey "github.com/molecula/featurebase/v3/short_txkey"
|
||||
"github.com/molecula/featurebase/v3/storage"
|
||||
|
||||
"github.com/molecula/featurebase/v3/vprint"
|
||||
"github.com/pkg/errors"
|
||||
)
|
||||
|
||||
|
|
@ -361,14 +360,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))
|
||||
}
|
||||
|
|
@ -395,12 +386,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)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1059,7 +1059,7 @@ func TestCursor_RemoveCells(t *testing.T) {
|
|||
//f, err := os.OpenFile("before.dot", os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 066)
|
||||
}
|
||||
|
||||
//These aren't test i'm just using to generate graphs to look at structure
|
||||
// These aren't test i'm just using to generate graphs to look at structure
|
||||
func TestCursor_PlayContainer(t *testing.T) {
|
||||
db := MustOpenDB(t)
|
||||
defer MustCloseDB(t, db)
|
||||
|
|
@ -1169,85 +1169,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)
|
||||
|
|
|
|||
104
rbf/tx.go
104
rbf/tx.go
|
|
@ -5,7 +5,6 @@ import (
|
|||
"bufio"
|
||||
"fmt"
|
||||
"io"
|
||||
"math"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
|
|
@ -1415,109 +1414,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<<j) == 0 {
|
||||
continue
|
||||
}
|
||||
|
||||
v := cell.Key<<16 | (uint64(i) * 64) | uint64(j)
|
||||
if v < start {
|
||||
continue
|
||||
} else if v > 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<<j) == 0 {
|
||||
continue
|
||||
}
|
||||
|
||||
v := cell.Key<<16 | (uint64(i) * 64) | uint64(j)
|
||||
if v < start {
|
||||
continue
|
||||
} else if v > 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()
|
||||
|
|
|
|||
|
|
@ -897,7 +897,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 {
|
||||
|
|
|
|||
|
|
@ -1363,31 +1363,49 @@ func (b *Bitmap) unionIntoTargetSingle(target *Bitmap, other *Bitmap) {
|
|||
// the bitmap at a specific key, ^ symbol represents the bitmaps current container iteration position,
|
||||
// and the - symbol represents a container that is at the current iteration position, but has been marked as "handled".
|
||||
//
|
||||
// ---------------------------- | ---------------------------- | ----------------------------
|
||||
// ---------------------------- | ---------------------------- | ----------------------------
|
||||
//
|
||||
// Bitmap 1 |___X____________X__________| | |___X____________X__________| | |___X____________X__________|
|
||||
// ^ | _ |
|
||||
// ---------------------------- | ---------------------------- | ----------------------------
|
||||
//
|
||||
// ^ | _ |
|
||||
// ---------------------------- | ---------------------------- | ----------------------------
|
||||
//
|
||||
// Bitmap 2 |_______X________X______X___| | |_______X_______________X___| | |_______X_______________X___|
|
||||
// ^ | ^ |
|
||||
// ---------------------------- | ---------------------------- | ----------------------------
|
||||
//
|
||||
// ^ | ^ |
|
||||
// ---------------------------- | ---------------------------- | ----------------------------
|
||||
//
|
||||
// Bitmap 3 |_______X___________________| | |_______X___________________| | |_______X___________________|
|
||||
// ^ | ^ |
|
||||
// ---------------------------- | ---------------------------- | ----------------------------
|
||||
//
|
||||
// ^ | ^ |
|
||||
// ---------------------------- | ---------------------------- | ----------------------------
|
||||
//
|
||||
// Bitmap 4 |___X_______________________| | |___X_______________________| | |___X_______________________|
|
||||
// ^ | _ |
|
||||
//
|
||||
// ^ | _ |
|
||||
//
|
||||
// ------------------------------------------------------------------------------------------------------------------------
|
||||
// ---------------------------- | ---------------------------- | ----------------------------
|
||||
//
|
||||
// ---------------------------- | ---------------------------- | ----------------------------
|
||||
//
|
||||
// Bitmap 1 |___X____________X__________| | |___X____________X__________| | |___X____________X__________|
|
||||
// _ | ^ | _
|
||||
// ---------------------------- | ---------------------------- | ----------------------------
|
||||
//
|
||||
// _ | ^ | _
|
||||
// ---------------------------- | ---------------------------- | ----------------------------
|
||||
//
|
||||
// Bitmap 2 |_______X_______________X___| | |_______X_______________X___| | |_______X_______________X___|
|
||||
// _ | ^ | ^
|
||||
// ---------------------------- | ---------------------------- | ----------------------------
|
||||
//
|
||||
// _ | ^ | ^
|
||||
// ---------------------------- | ---------------------------- | ----------------------------
|
||||
//
|
||||
// Bitmap 3 |_______X___________________| | |_______X___________________| | |_______X___________________|
|
||||
// _ | |
|
||||
// ---------------------------- | ---------------------------- | ----------------------------
|
||||
//
|
||||
// _ | |
|
||||
// ---------------------------- | ---------------------------- | ----------------------------
|
||||
//
|
||||
// Bitmap 4 |___X_______________________| | |___X_______________________| | |___X_______________________|
|
||||
// _
|
||||
//
|
||||
// _
|
||||
func (b *Bitmap) unionInPlace(others ...*Bitmap) {
|
||||
const staticSize = 20
|
||||
var (
|
||||
|
|
@ -4391,7 +4409,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
|
||||
}
|
||||
|
|
@ -4421,11 +4439,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 {
|
||||
|
|
@ -6743,7 +6761,7 @@ func xorCompare(x *xorstm) (r1 Interval16, hasData bool) {
|
|||
return r1, hasData
|
||||
}
|
||||
|
||||
//stm is state machine used to "xor" iterate over runs.
|
||||
// stm is state machine used to "xor" iterate over runs.
|
||||
type xorstm struct {
|
||||
vaValid, vbValid bool
|
||||
va, vb Interval16
|
||||
|
|
|
|||
|
|
@ -4658,7 +4658,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))
|
||||
}
|
||||
|
|
|
|||
88
server.go
88
server.go
|
|
@ -69,7 +69,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
|
||||
|
|
@ -159,15 +158,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 {
|
||||
|
|
@ -448,7 +438,6 @@ func NewServer(opts ...ServerOption) (*Server, error) {
|
|||
|
||||
gcNotifier: NopGCNotifier,
|
||||
|
||||
antiEntropyInterval: 0,
|
||||
metricInterval: 0,
|
||||
diagnosticInterval: 0,
|
||||
viewsRemovalInterval: time.Hour,
|
||||
|
|
@ -485,7 +474,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 {
|
||||
|
|
@ -651,10 +639,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() }()
|
||||
|
|
@ -804,12 +791,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.
|
||||
|
|
@ -916,73 +897,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) {
|
||||
|
|
|
|||
|
|
@ -144,6 +144,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"`
|
||||
|
|
|
|||
|
|
@ -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),
|
||||
|
|
|
|||
|
|
@ -16,8 +16,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)
|
||||
}
|
||||
|
|
@ -25,7 +24,6 @@ func TestMonitorAntiEntropyZero(t *testing.T) {
|
|||
|
||||
ch := make(chan struct{})
|
||||
go func() {
|
||||
s.monitorAntiEntropy()
|
||||
close(ch)
|
||||
}()
|
||||
|
||||
|
|
|
|||
50
stattx.go
50
stattx.go
|
|
@ -150,8 +150,6 @@ const (
|
|||
kRemove
|
||||
kContains
|
||||
kContainerIterator
|
||||
kForEach
|
||||
kForEachRange
|
||||
kCount
|
||||
kMax
|
||||
kMin
|
||||
|
|
@ -187,10 +185,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:
|
||||
|
|
@ -212,16 +206,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
|
||||
|
||||
|
|
@ -415,40 +399,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
|
||||
|
||||
|
|
|
|||
13
tx.go
13
tx.go
|
|
@ -50,11 +50,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
|
||||
|
|
@ -108,14 +103,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)
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue