scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-10-09 12:27:52 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #1687 from seebs/core850

Browse source 
make details optional and support limits on mutex checks
This commit is contained in:
seebs 2021-09-09 16:17:53 -05:00 • committed by GitHub
commit 0d568b0d52
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
10 changed files with 275 additions and 64 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

141
api.go
View file

@ -2653,7 +2653,7 @@ func (api *API) RestoreShard(ctx context.Context, indexName string, shard uint64
return nil
}
func (api *API) mutexCheckThisNode(ctx context.Context, qcx *Qcx, indexName string, fieldName string) (map[uint64]map[uint64][]uint64, error) {
func (api *API) mutexCheckThisNode(ctx context.Context, qcx *Qcx, indexName string, fieldName string, details bool, limit int) (map[uint64]map[uint64][]uint64, error) {
index := api.holder.Index(indexName)
if index == nil {
return nil, newNotFoundError(ErrIndexNotFound, indexName)
@ -2662,7 +2662,26 @@ func (api *API) mutexCheckThisNode(ctx context.Context, qcx *Qcx, indexName stri
if field == nil {
return nil, newNotFoundError(ErrFieldNotFound, fieldName)
}
return field.MutexCheck(ctx, qcx)
results, err := field.MutexCheck(ctx, qcx, details, limit)
if err != nil {
return nil, err
}
if limit != 0 && len(results) > limit {
toDel := len(results) - limit
// yes, Go allows you to delete keys you've already seen while
// iterating a map. The spec says that if a value not-yet-reached
// is deleted during iteration, it may or may not appear; this
// carries the implication that deleting things during map iteration
// is safe.
for k := range results {
delete(results, k)
toDel--
if toDel == 0 {
break
}
}
}
return results, err
}
// mergeIDLists merges a list of numeric IDs into another list, removing
@ -2707,21 +2726,25 @@ func mergeKeyLists(dst []string, src []string) []string {
// MutexCheckNode checks for collisions in a given mutex field. The response is
// a map[shard]map[column]values, not translated.
func (api *API) MutexCheckNode(ctx context.Context, qcx *Qcx, indexName string, fieldName string) (map[uint64]map[uint64][]uint64, error) {
func (api *API) MutexCheckNode(ctx context.Context, qcx *Qcx, indexName string, fieldName string, details bool, limit int) (map[uint64]map[uint64][]uint64, error) {
if err := api.validate(apiMutexCheck); err != nil {
return nil, errors.Wrap(err, "validating api method")
}
return api.mutexCheckThisNode(ctx, qcx, indexName, fieldName)
return api.mutexCheckThisNode(ctx, qcx, indexName, fieldName, details, limit)
}
// MutexCheck checks a named field for mutex violations, returning a
// map of record IDs to values for records that have multiple values in the
// field. The return will be one of:
// details true:
// map[uint64][]uint64 // unkeyed index, unkeyed field
// map[uint64][]string // unkeyed index, keyed field
// map[string][]uint64 // keyed index, unkeyed field
// map[string][]string // keyed index, keyed field
func (api *API) MutexCheck(ctx context.Context, qcx *Qcx, indexName string, fieldName string) (result interface{}, err error) {
// details false:
// []uint64 // unkeyed index
// []string // keyed index
func (api *API) MutexCheck(ctx context.Context, qcx *Qcx, indexName string, fieldName string, details bool, limit int) (result interface{}, err error) {
if err = api.validate(apiMutexCheck); err != nil {
return nil, errors.Wrap(err, "validating api method")
}
@ -2746,12 +2769,12 @@ func (api *API) MutexCheck(ctx context.Context, qcx *Qcx, indexName string, fiel
if node.ID != myID {
node := node // loop variable shadowing again
eg.Go(func() (err error) {
results[i], err = api.server.defaultClient.MutexCheck(ctx, &node.URI, indexName, fieldName)
results[i], err = api.server.defaultClient.MutexCheck(ctx, &node.URI, indexName, fieldName, details, limit)
return err
})
} else {
eg.Go(func() (err error) {
results[i], err = api.mutexCheckThisNode(ctx, qcx, indexName, fieldName)
results[i], err = api.mutexCheckThisNode(ctx, qcx, indexName, fieldName, details, limit)
return err
})
}
@ -2760,12 +2783,18 @@ func (api *API) MutexCheck(ctx context.Context, qcx *Qcx, indexName string, fiel
if err != nil {
return nil, err
}
// Set this arbitrarily large so we don't have to be hand-checking for 0
// throughout.
if limit == 0 {
limit = math.MaxInt32
}
// We now have a series of maps from shards to maps of record IDs to
// values. But wait! Either the field, or the index, might be using keys,
// and want those translated. So we have to translate those. We'll create
// some tables.
useIndexKeys := index.Keys()
useFieldKeys := field.Keys()
// If we're not doing details, we won't translate field keys even if we could.
useFieldKeys := field.Keys() && details
var indexKeys = map[uint64]string{}
var fieldKeys = map[uint64]string{}
var indexIDs []uint64
@ -2778,8 +2807,9 @@ func (api *API) MutexCheck(ctx context.Context, qcx *Qcx, indexName string, fiel
// but what we can do is make a function which works with that map type
// given the raw integer values, and is a closure with an already-created
// map which has already been stashed in `result`. Because maps are
// reference-y, this should actually work.
var process func(uint64, []uint64)
// reference-y, this should actually work. This function returns true if
// it's hit the limit for length of results.
var process func(uint64, []uint64) bool
if useIndexKeys || useFieldKeys {
for _, nodeResults := range results {
for _, shardResults := range nodeResults {
@ -2801,6 +2831,10 @@ func (api *API) MutexCheck(ctx context.Context, qcx *Qcx, indexName string, fiel
}
}
}
// if context is done, return early.
if err := ctx.Err(); err != nil {
return nil, err
}
untranslatedKeys := 0
// Obtain translation tables for the keys.
if useIndexKeys {
@ -2819,6 +2853,10 @@ func (api *API) MutexCheck(ctx context.Context, qcx *Qcx, indexName string, fiel
}
}
}
// if context is done, return early.
if err := ctx.Err(); err != nil {
return nil, err
}
if useFieldKeys {
fieldKeyList, err := api.cluster.translateFieldListIDs(field, fieldIDs)
if err != nil {
@ -2840,14 +2878,37 @@ func (api *API) MutexCheck(ctx context.Context, qcx *Qcx, indexName string, fiel
}
}
// if context is done, return early.
if err := ctx.Err(); err != nil {
return nil, err
}
// define the process functions. separated from above code just to make
// it easier to follow/compare them.
if useIndexKeys {
if useFieldKeys {
if !details {
outMap := make(map[uint64]struct{})
outStrings := []string{}
// unlike a map, the slice won't get updated-in-place, so we have
// to assign to result after we're done
defer func() {
if err == nil {
result = outStrings
}
}()
process = func(recordID uint64, valueIDs []uint64) bool {
if _, ok := outMap[recordID]; ok {
return len(outMap) >= limit
}
outMap[recordID] = struct{}{}
outStrings = append(outStrings, indexKeys[recordID])
return len(outMap) >= limit
}
} else if useFieldKeys {
outMap := make(map[string][]string)
var valueKeys []string
result = outMap
process = func(recordID uint64, valueIDs []uint64) {
process = func(recordID uint64, valueIDs []uint64) bool {
valueKeys = valueKeys[:0]
for _, id := range valueIDs {
valueKeys = append(valueKeys, fieldKeys[id])
@ -2855,32 +2916,52 @@ func (api *API) MutexCheck(ctx context.Context, qcx *Qcx, indexName string, fiel
record := indexKeys[recordID]
if existing, ok := outMap[record]; ok {
outMap[record] = mergeKeyLists(existing, valueKeys)
} else {
} else if len(outMap) < limit {
// The append is so we can reuse this buffer safely,
// which matters if there's replication, because many
// cases won't need to copy the buffer, they'll just
// copy individual things from it.
outMap[record] = append([]string{}, valueKeys...)
}
return len(outMap) >= limit
}
} else {
outMap := make(map[string][]uint64)
result = outMap
process = func(recordID uint64, values []uint64) {
process = func(recordID uint64, values []uint64) bool {
record := indexKeys[recordID]
if existing, ok := outMap[record]; ok {
outMap[record] = mergeIDLists(existing, values)
} else {
} else if len(outMap) < limit {
outMap[record] = values
}
return len(outMap) >= limit
}
}
} else {
if useFieldKeys {
if !details {
outMap := make(map[uint64]struct{})
outIDs := []uint64{}
// unlike a map, the slice won't get updated-in-place, so we have
// to assign to result after we're done
defer func() {
if err == nil {
result = outIDs
}
}()
process = func(recordID uint64, valueIDs []uint64) bool {
if _, ok := outMap[recordID]; ok {
return len(outMap) >= limit
}
outMap[recordID] = struct{}{}
outIDs = append(outIDs, recordID)
return len(outMap) >= limit
}
} else if useFieldKeys {
outMap := make(map[uint64][]string)
var valueKeys []string
result = outMap
process = func(record uint64, valueIDs []uint64) {
process = func(record uint64, valueIDs []uint64) bool {
valueKeys = valueKeys[:0]
for _, id := range valueIDs {
valueKeys = append(valueKeys, fieldKeys[id])
@ -2894,34 +2975,54 @@ func (api *API) MutexCheck(ctx context.Context, qcx *Qcx, indexName string, fiel
// copy individual things from it.
outMap[record] = append([]string{}, valueKeys...)
}
return len(outMap) >= limit
}
} else {
outMap := make(map[uint64][]uint64)
result = outMap
process = func(record uint64, values []uint64) {
process = func(record uint64, values []uint64) bool {
if existing, ok := outMap[record]; ok {
outMap[record] = mergeIDLists(existing, values)
} else {
outMap[record] = values
}
return len(outMap) >= limit
}
}
}
// if you specify a limit, and you have *different* errors on different
// nodes, we will not check all of the nodes. otherwise there's no practical
// way to get the primary benefit of specifying a limit.
processing:
for _, nodeResults := range results {
if len(nodeResults) == 0 {
continue
}
// if context is done, return early.
if err := ctx.Err(); err != nil {
return nil, err
}
for _, v := range nodeResults {
if len(v) == 0 {
continue
}
counter := 0
for record, values := range v {
process(record, values)
counter++
if process(record, values) {
break processing
}
// every 65k items or so, check the context for done-ness
if counter%(1<<16) == 0 {
if err := ctx.Err(); err != nil {
return nil, err
}
}
}
}
}
return result, nil
return result, ctx.Err()
}
type serverInfo struct {

56
api_test.go
View file

@ -966,10 +966,6 @@ func TestAPI_MutexCheck(t *testing.T) {
qcx := m0.API.Txf().NewQcx()
defer qcx.Abort()
results, err := m0.API.MutexCheck(ctx, qcx, indexData.indexName, fieldData.fieldName)
if err != nil {
t.Fatalf("checking mutexes: %v", err)
}
// first two shards of each group of 4 should have a collision in
// position 1
expected := map[uint64]bool{
@ -980,6 +976,12 @@ func TestAPI_MutexCheck(t *testing.T) {
(8 << shardwidth.Exponent) + 1: true,
(9 << shardwidth.Exponent) + 1: true,
}
results, err := m0.API.MutexCheck(ctx, qcx, indexData.indexName, fieldData.fieldName, true, 0)
if err != nil {
t.Fatalf("checking mutexes: %v", err)
}
if keyedField {
mapped, ok := results.(map[uint64][]string)
if !ok {
@ -1014,9 +1016,29 @@ func TestAPI_MutexCheck(t *testing.T) {
}
}
if seen != len(expected) {
t.Fatalf("expected exactly %d records to have collisions", len(expected))
t.Fatalf("expected exactly %d records to have collisions, got %d", len(expected), seen)
}
}
// and let's try with no details and a limit of 3...
results, err = m0.API.MutexCheck(ctx, qcx, indexData.indexName, fieldData.fieldName, false, 3)
if err != nil {
t.Fatalf("checking mutexes: %v", err)
}
mapped, ok := results.([]uint64)
if !ok {
t.Fatalf("expected []uint64, got %T", results)
}
seen := 0
for _, k := range mapped {
seen++
if !expected[k] {
t.Fatalf("expected all collisions to be position 1 in shards (s %% 4 in [0,1]), got %d", k)
}
}
if seen != 3 {
t.Fatalf("expected results limited to 3, got %d", seen)
}
})
}
indexData = indexes[true]
@ -1111,14 +1133,11 @@ func TestAPI_MutexCheck(t *testing.T) {
qcx := m0.API.Txf().NewQcx()
defer qcx.Abort()
results, err := m0.API.MutexCheck(ctx, qcx, indexData.indexName, fieldData.fieldName)
results, err := m0.API.MutexCheck(ctx, qcx, indexData.indexName, fieldData.fieldName, true, 0)
if err != nil {
t.Fatalf("checking mutexes: %v", err)
}
if keyedField {
// this just sorta comes out this way with our hashing; these are
// the things which were in position 1 of their shards, and did
// not have a value which happens to map to 3.
mapped, ok := results.(map[string][]string)
if !ok {
t.Fatalf("expected map[string][]string, got %T", results)
@ -1155,6 +1174,25 @@ func TestAPI_MutexCheck(t *testing.T) {
t.Fatalf("expected exactly %d records to have collisions, got %d", len(expected), seen)
}
}
results, err = m0.API.MutexCheck(ctx, qcx, indexData.indexName, fieldData.fieldName, false, 3)
if err != nil {
t.Fatalf("checking mutexes: %v", err)
}
mapped, ok := results.([]string)
if !ok {
t.Fatalf("expected []string, got %T", results)
}
seen := 0
for _, k := range mapped {
seen++
if _, ok := expected[k]; !ok {
t.Fatalf("unexpected collision on key %q", k)
}
}
if seen != 3 {
t.Fatalf("expected results limited to 3, got %d", len(expected))
}
})
}
}

4
client.go
View file

@ -80,7 +80,7 @@ type InternalClient interface {
RetrieveTranslatePartitionFromURI(ctx context.Context, index string, partition int, uri pnet.URI) (io.ReadCloser, error)
ImportRoaring(ctx context.Context, uri *pnet.URI, index, field string, shard uint64, remote bool, req *ImportRoaringRequest) error
ShardReader(ctx context.Context, index string, shard uint64) (io.ReadCloser, error)
MutexCheck(ctx context.Context, uri *pnet.URI, index string, field string) (map[uint64]map[uint64][]uint64, error)
MutexCheck(ctx context.Context, uri *pnet.URI, index string, field string, details bool, limit int) (map[uint64]map[uint64][]uint64, error)
IDAllocDataReader(ctx context.Context) (io.ReadCloser, error)
IndexTranslateDataReader(ctx context.Context, index string, partitionID int) (io.ReadCloser, error)
@ -214,7 +214,7 @@ func (n nopInternalClient) ImportRoaring(ctx context.Context, uri *pnet.URI, ind
return nil
}
func (n nopInternalClient) MutexCheck(ctx context.Context, uri *pnet.URI, index, field string) (map[uint64]map[uint64][]uint64, error) {
func (n nopInternalClient) MutexCheck(ctx context.Context, uri *pnet.URI, index, field string, details bool, limit int) (map[uint64]map[uint64][]uint64, error) {
return nil, nil
}

4
field.go
View file

@ -1094,7 +1094,7 @@ func (f *Field) Row(tx Tx, rowID uint64) (*Row, error) {
// mutexCheck performs a sanity-check on the available fragments for a
// field. The return is map[column]map[shard][]values for collisions only.
func (f *Field) MutexCheck(ctx context.Context, qcx *Qcx) (map[uint64]map[uint64][]uint64, error) {
func (f *Field) MutexCheck(ctx context.Context, qcx *Qcx, details bool, limit int) (map[uint64]map[uint64][]uint64, error) {
if f.Type() != FieldTypeMutex {
return nil, errors.New("mutex check only valid for mutex fields")
}
@ -1106,7 +1106,7 @@ func (f *Field) MutexCheck(ctx context.Context, qcx *Qcx) (map[uint64]map[uint64
// so it has no bits set, so it has no extra bits set.
return nil, nil
}
return standard.mutexCheck(ctx, qcx)
return standard.mutexCheck(ctx, qcx, details, limit)
}
// SetBit sets a bit on a view within the field.

4
fragment.go
View file

@ -601,8 +601,8 @@ func (f *fragment) closeStorage() error {
// mutexCheck checks for any entries in fragment which violate the mutex
// property of having only one value set for a given column ID.
func (f *fragment) mutexCheck(tx Tx) (map[uint64][]uint64, error) {
dup := roaring.NewBitmapMutexDupFilter(f.shard << shardwidth.Exponent)
func (f *fragment) mutexCheck(tx Tx, details bool, limit int) (map[uint64][]uint64, error) {
dup := roaring.NewBitmapMutexDupFilter(f.shard<<shardwidth.Exponent, details, limit)
err := tx.ApplyFilter(f.index(), f.field(), f.view(), f.shard, 0, dup)
if err != nil {
return nil, err

6
http/client.go
View file

@ -287,11 +287,13 @@ func (c *InternalClient) IngestOperations(ctx context.Context, uri *pnet.URI, in
// MutexCheck uses the mutex-check endpoint to request mutex collision data
// from a single node. It produces per-shard results, and does not translate
// them.
func (c *InternalClient) MutexCheck(ctx context.Context, uri *pnet.URI, indexName string, fieldName string) (map[uint64]map[uint64][]uint64, error) {
func (c *InternalClient) MutexCheck(ctx context.Context, uri *pnet.URI, indexName string, fieldName string, details bool, limit int) (map[uint64]map[uint64][]uint64, error) {
if uri == nil {
uri = c.defaultURI
}
u := uri.Path(fmt.Sprintf("/internal/index/%s/field/%s/mutex-check", indexName, fieldName))
// This is not actually a "Path", but reworking this to support queries
// is messier than I have resources to pursue just now.
u := uri.Path(fmt.Sprintf("/internal/index/%s/field/%s/mutex-check?details=%t&limit=%d", indexName, fieldName, details, limit))
req, err := http.NewRequest("GET", u, nil)
if err != nil {
return nil, errors.Wrap(err, "creating request")

26
http/handler.go
View file

@ -2765,9 +2765,20 @@ func (h *Handler) handleGetMutexCheck(w http.ResponseWriter, r *http.Request) {
// Get index and field type to determine how to handle the
// import data.
indexName, fieldName := mux.Vars(r)["index"], mux.Vars(r)["field"]
q := r.URL.Query()
limit := 0
details := q.Get("details") == "true"
limitStr := q.Get("limit")
if limitStr != "" {
var err error
limit, err = strconv.Atoi(limitStr)
if err != nil {
http.Error(w, "limit must be numeric", http.StatusBadRequest)
}
}
qcx := h.api.Txf().NewQcx()
defer qcx.Abort()
out, err := h.api.MutexCheck(r.Context(), qcx, indexName, fieldName)
out, err := h.api.MutexCheck(r.Context(), qcx, indexName, fieldName, details, limit)
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
@ -2791,9 +2802,20 @@ func (h *Handler) handleInternalGetMutexCheck(w http.ResponseWriter, r *http.Req
// Get index and field type to determine how to handle the
// import data.
indexName, fieldName := mux.Vars(r)["index"], mux.Vars(r)["field"]
q := r.URL.Query()
limit := 0
details := q.Get("details") == "true"
limitStr := q.Get("limit")
if limitStr != "" {
var err error
limit, err = strconv.Atoi(limitStr)
if err != nil {
http.Error(w, "limit must be numeric", http.StatusBadRequest)
}
}
qcx := h.api.Txf().NewQcx()
defer qcx.Abort()
out, err := h.api.MutexCheckNode(r.Context(), qcx, indexName, fieldName)
out, err := h.api.MutexCheckNode(r.Context(), qcx, indexName, fieldName, details, limit)
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return

72
roaring/filter.go
View file

@ -767,18 +767,29 @@ func NewBitmapRangeFilter(min, max FilterKey, keyCallback func(FilterKey, int32)
// The slice is local-coordinates (first column 0), but the map is global
// coordinates (first column is whatever base was).
type BitmapMutexDupFilter struct {
base uint64 // the offset of 0 for this, used to accommodate shard offsets
extra map[uint64][]uint64 // extra values observed
first []uint64 // first values observed
base uint64 // the offset of 0 for this, used to accommodate shard offsets
extra map[uint64][]uint64 // extra values observed
first []uint64 // first values observed
details bool
limit int
done bool // if we have a limit, and we've hit it...
highKey FilterKey // ... we can stop after this many containers.
}
var _ BitmapFilter = &BitmapMutexDupFilter{}
func NewBitmapMutexDupFilter(base uint64) *BitmapMutexDupFilter {
func NewBitmapMutexDupFilter(base uint64, details bool, limit int) *BitmapMutexDupFilter {
filter := &BitmapMutexDupFilter{
base: base,
extra: map[uint64][]uint64{},
first: make([]uint64, 1<<shardwidth.Exponent),
base: base,
extra: map[uint64][]uint64{},
first: make([]uint64, 1<<shardwidth.Exponent),
details: details,
limit: limit,
}
if filter.limit == 0 {
// A limit of 0 is not a limit; set limit higher than possible number of
// values we could have.
filter.limit = 2 << shardwidth.Exponent
}
for i := range filter.first {
filter.first[i] = ^uint64(0)
@ -798,27 +809,52 @@ func (b *BitmapMutexDupFilter) ConsiderData(key FilterKey, data *Container) Filt
containerCallback(data, func(u uint16) {
pos := basePos + uint64(u)
if b.first[pos] != ^uint64(0) {
b.extra[pos+b.base] = append(b.extra[pos+b.base], value)
if b.details {
b.extra[pos+b.base] = append(b.extra[pos+b.base], value)
} else {
// no details, just annotate that it exists
b.extra[pos+b.base] = []uint64{}
}
} else {
b.first[pos] = value
}
})
if len(b.extra) >= b.limit {
if !b.done {
// we note which container we found the last value we needed in.
// We may still go over the limit, but we won't look at any *more*
// containers in this row.
//
// We can't just abort early because the records we already found
// could have more values.
b.done = true
b.highKey = key & keyMask
return key.RejectRow()
}
if (key & keyMask) >= b.highKey {
return key.RejectRow()
}
}
return key.MatchOne()
}
// Report returns the set of duplicate values identified.
func (b *BitmapMutexDupFilter) Report() map[uint64][]uint64 {
// copy values into extra, and remove them from first, so calling
// Report() again won't cause double-appends.
for k, v := range b.extra {
kpos := k % (1 << shardwidth.Exponent)
if b.first[kpos] != ^uint64(0) {
v = append(v, 0)
// prepend so the lowest value goes at the beginning
copy(v[1:], v[:])
v[0] = b.first[kpos]
b.first[kpos] = ^uint64(0)
b.extra[k] = v
// Report() again won't cause double-appends. We only have to do
// this if we've been asked for details; otherwise the list of
// known positions is sufficient.
if b.details {
for k, v := range b.extra {
kpos := k % (1 << shardwidth.Exponent)
if b.first[kpos] != ^uint64(0) {
v = append(v, 0)
// prepend so the lowest value goes at the beginning
copy(v[1:], v[:])
v[0] = b.first[kpos]
b.first[kpos] = ^uint64(0)
b.extra[k] = v
}
}
}
return b.extra

10
roaring/filter_internal_test.go
View file

@ -350,16 +350,16 @@ func TestFilterWithRows(t *testing.T) {
}
func TestMutexDupFilter(t *testing.T) {
tests := []struct{
pairs [][2]uint64
tests := []struct {
pairs [][2]uint64
expect map[uint64][]uint64
}{
{
pairs: [][2]uint64{{0, 0}, {1, 0}, {0, 1}},
pairs: [][2]uint64{{0, 0}, {1, 0}, {0, 1}},
expect: map[uint64][]uint64{0: {0, 1}},
},
{
pairs: [][2]uint64{{0, 0}, {1, 0}, {0, 1}, {0, 2}},
pairs: [][2]uint64{{0, 0}, {1, 0}, {0, 1}, {0, 2}},
expect: map[uint64][]uint64{0: {0, 1, 2}},
},
}
@ -370,7 +370,7 @@ func TestMutexDupFilter(t *testing.T) {
v := (p[1] << shardwidth.Exponent) | p[0]
b.DirectAdd(v)
}
dup := NewBitmapMutexDupFilter(0)
dup := NewBitmapMutexDupFilter(0, true, 9)
iter, _ := b.Containers.Iterator(0)
err := ApplyFilterToIterator(dup, iter)
if err != nil {

16
view.go
View file

@ -17,6 +17,7 @@ package pilosa
import (
"context"
"fmt"
"math"
"os"
"path/filepath"
"runtime"
@ -444,7 +445,7 @@ func (v *view) row(txOrig Tx, rowID uint64) (*Row, error) {
// mutexCheck checks all available fragments for duplicate values. The return
// is map[column]map[shard][]values for collisions only.
func (v *view) mutexCheck(ctx context.Context, qcx *Qcx) (map[uint64]map[uint64][]uint64, error) {
func (v *view) mutexCheck(ctx context.Context, qcx *Qcx, details bool, limit int) (map[uint64]map[uint64][]uint64, error) {
// We don't need the context, we just want the context-awareness on the error groups.
// It would be nice if the inner functions could use this too...
eg, _ := errgroup.WithContext(ctx)
@ -465,7 +466,7 @@ func (v *view) mutexCheck(ctx context.Context, qcx *Qcx) (map[uint64]map[uint64]
return err
}
defer finisher(&err)
results[i], err = frag.mutexCheck(tx)
results[i], err = frag.mutexCheck(tx, details, limit)
if err != nil {
return err
}
@ -477,11 +478,22 @@ func (v *view) mutexCheck(ctx context.Context, qcx *Qcx) (map[uint64]map[uint64]
return nil, err
}
out := map[uint64]map[uint64][]uint64{}
// We would use MaxInt here, but it's new with go 1.17. In practice if
// you have 2 billion duplicates you're sorta screwed anyway.
if limit == 0 {
limit = math.MaxInt32
}
count := 0
for i, result := range results {
if len(result) == 0 {
continue
}
out[frags[i].shard] = result
count += len(result)
// if we have enough, stop
if count > limit {
break
}
}
return out, nil
}