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

backport 1687

Browse source
This commit is contained in:
Todd Gruben 2021-09-09 10:11:51 -05:00
parent bfd2856e67
commit d2dc8f391f
10 changed files with 621 additions and 67 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

172
api.go
View file

@ -2179,7 +2179,7 @@ func (api *API) TranslateIndexDB(ctx context.Context, indexName string, partitio
_, err := store.ReadFrom(rd)
return err
}
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)
@ -2188,7 +2188,27 @@ 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
@ -2233,21 +2253,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) {
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")
}
@ -2264,22 +2288,21 @@ func (api *API) MutexCheck(ctx context.Context, qcx *Qcx, indexName string, fiel
}
nodes := Nodes(api.cluster.nodes).Clone()
eg, _ := errgroup.WithContext(ctx)
myID := api.Node().ID
results := make([]map[uint64]map[uint64][]uint64, len(nodes))
myID := api.Node().ID
// vprint.VV("MyID %#v\n", myID)
for i, node := range nodes {
i := i // loop variable shadowing is a war crime
// vprint.VV("Compare %#v with %v", node.ID, myID)
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
})
}
@ -2288,23 +2311,33 @@ 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.
// 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
var fieldIDs []uint64
// build translation tables, if we need them
// We'll use the string "untranslated" as our default value and overwrite
// it with translations. We do check for missing translation values in
// our returns, but just in case, you know?
untranslated := "untranslated"
// We don't know which of four map types we want to be working with,
// 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 {
@ -2326,7 +2359,12 @@ func (api *API) MutexCheck(ctx context.Context, qcx *Qcx, indexName string, fiel
}
}
}
// we now have lists of the keys, so...
// 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 {
indexKeyList, err := api.cluster.translateIndexIDs(ctx, indexName, indexIDs)
if err != nil {
@ -2336,9 +2374,17 @@ func (api *API) MutexCheck(ctx context.Context, qcx *Qcx, indexName string, fiel
return nil, fmt.Errorf("translating %d record IDs, got %d keys", len(indexIDs), len(indexKeyList))
}
for i := range indexIDs {
indexKeys[indexIDs[i]] = indexKeyList[i]
if indexKeyList[i] != "" {
indexKeys[indexIDs[i]] = indexKeyList[i]
} else {
untranslatedKeys++
}
}
}
// 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 {
@ -2348,19 +2394,49 @@ func (api *API) MutexCheck(ctx context.Context, qcx *Qcx, indexName string, fiel
return nil, fmt.Errorf("translating %d IDs, got %d keys", len(indexIDs), len(fieldKeyList))
}
for i := range fieldIDs {
fieldKeys[fieldIDs[i]] = fieldKeyList[i]
if fieldKeyList[i] != "" {
fieldKeys[fieldIDs[i]] = fieldKeyList[i]
} else {
untranslatedKeys++
}
}
}
if untranslatedKeys > 0 {
api.server.logger.Printf("translating mutex check results: %d key(s) untranslated", untranslatedKeys)
}
}
// 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])
@ -2368,32 +2444,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])
@ -2407,34 +2503,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()
}
// TranslateFieldDB is an internal function to load the field keys database

366
api_test.go
View file

@ -637,6 +637,345 @@ type mutexCheckField struct {
createdAt int64
}
func TestAPI_MutexCheck(t *testing.T) {
c := test.MustRunCluster(t, 3)
defer c.Close()
m0 := c.GetNode(0)
nodesByID := make(map[string]*test.Command, 3)
qcxsByID := make(map[string]*pilosa.Qcx, 3)
for i := 0; i < 3; i++ {
node := c.GetNode(i)
id := node.API.Node().ID
nodesByID[id] = node
}
indexes := make(map[bool]mutexCheckIndex)
ctx := context.Background()
for _, keyedIndex := range []bool{false, true} {
indexName := fmt.Sprintf("i%t", keyedIndex)
index, err := m0.API.CreateIndex(ctx, indexName, pilosa.IndexOptions{Keys: keyedIndex, TrackExistence: true})
if err != nil {
t.Fatalf("creating index: %v", err)
}
if index.CreatedAt() == 0 {
t.Fatal("index createdAt is empty")
}
indexData := mutexCheckIndex{indexName: indexName, index: index, fields: make(map[bool]mutexCheckField), createdAt: index.CreatedAt()}
for _, keyedField := range []bool{false, true} {
fieldName := fmt.Sprintf("f%t", keyedField)
var field *pilosa.Field
if keyedField {
field, err = m0.API.CreateField(ctx, indexName, fieldName, pilosa.OptFieldTypeMutex(pilosa.CacheTypeNone, 0), pilosa.OptFieldKeys())
} else {
field, err = m0.API.CreateField(ctx, indexName, fieldName, pilosa.OptFieldTypeMutex(pilosa.CacheTypeNone, 0))
}
if err != nil {
t.Fatalf("creating field: %v", err)
}
if field.CreatedAt() == 0 {
t.Fatal("field createdAt is empty")
}
indexData.fields[keyedField] = mutexCheckField{fieldName: fieldName, field: field, createdAt: field.CreatedAt()}
}
indexes[keyedIndex] = indexData
}
rowIDs := []uint64{0, 1, 2, 3}
colIDs := []uint64{0, 1, 2, 3}
rowKeysBase := []string{"v0", "v1", "v2", "v3"}
colKeysBase := []string{"c0", "c1", "c2", "c3"}
const nShards = 10
// now, try the same thing for each combination of keyed/unkeyed. we
// share code between keyed/unkeyed fields, but for indexes, the logic
// is fundamentally different because we can't know shards in advance.
indexData := indexes[false]
for keyedField, fieldData := range indexData.fields {
t.Run(fmt.Sprintf("%s-%s", indexData.indexName, fieldData.fieldName), func(t *testing.T) {
for id, node := range nodesByID {
qcxsByID[id] = node.API.Txf().NewQcx()
}
for shard := uint64(0); shard < nShards; shard++ {
// restore row/col ID values which can get altered by imports
for i := range rowIDs {
rowIDs[i] = uint64(i)
colIDs[i] = (shard << shardwidth.Exponent) + uint64(i) + (shard % 4)
}
req := &pilosa.ImportRequest{
Index: indexData.indexName,
IndexCreatedAt: indexData.createdAt,
Field: fieldData.fieldName,
FieldCreatedAt: fieldData.createdAt,
Shard: shard,
ColumnIDs: colIDs,
}
if keyedField {
req.RowKeys = rowKeysBase
} else {
req.RowIDs = rowIDs
}
nodesForShard, err := m0.API.ShardNodes(ctx, indexData.indexName, shard)
if err != nil {
t.Fatalf("obtaining shard list: %v", err)
}
if len(nodesForShard) < 1 {
t.Fatalf("no nodes for shard %d", shard)
}
node := nodesByID[nodesForShard[0].ID]
if err := node.API.Import(ctx, qcxsByID[nodesForShard[0].ID], req); err != nil {
t.Fatalf("importing data: %v", err)
}
}
// and then we break the mutex and close the Qcxs
for id, node := range nodesByID {
field, err := node.API.Field(ctx, indexData.indexName, fieldData.fieldName)
if err != nil {
t.Fatalf("requesting field %s from node %s: %v", fieldData.fieldName, id, err)
}
pilosa.CorruptAMutex(t, field, qcxsByID[id])
err = qcxsByID[id].Finish()
if err != nil {
t.Fatalf("closing out transaction on node %s: %v", id, err)
}
}
qcx := m0.API.Txf().NewQcx()
defer qcx.Abort()
// first two shards of each group of 4 should have a collision in
// position 1
expected := map[uint64]bool{
(0 << shardwidth.Exponent) + 1: true,
(1 << shardwidth.Exponent) + 1: true,
(4 << shardwidth.Exponent) + 1: true,
(5 << shardwidth.Exponent) + 1: true,
(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 {
t.Fatalf("expected map[uint64][]string, got %T", results)
}
seen := 0
for k, v := range mapped {
seen++
if !expected[k] {
t.Fatalf("expected all collisions to be 1 shards (s %% 4 in [0,1]), got %d", k)
}
if len(v) != 2 {
t.Fatalf("expected exactly two collisions")
}
}
if seen != len(expected) {
t.Fatalf("expected exactly %d records to have collisions", len(expected))
}
} else {
mapped, ok := results.(map[uint64][]uint64)
if !ok {
t.Fatalf("expected map[uint64][]uint64, got %T", results)
}
seen := 0
for k, v := range mapped {
seen++
if !expected[k] {
t.Fatalf("expected all collisions to be 1 shards (s %% 4 in [0,1]), got %d", k)
}
if len(v) != 2 {
t.Fatalf("expected exactly two collisions")
}
}
if seen != 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]
for keyedField, fieldData := range indexData.fields {
t.Run(fmt.Sprintf("%s-%s", indexData.indexName, fieldData.fieldName), func(t *testing.T) {
for id, node := range nodesByID {
qcxsByID[id] = node.API.Txf().NewQcx()
}
req := &pilosa.ImportRequest{
Index: indexData.indexName,
IndexCreatedAt: indexData.createdAt,
Field: fieldData.fieldName,
FieldCreatedAt: fieldData.createdAt,
Shard: 0, // ignored when using keys
}
rowKeys := make([]string, 0, len(rowKeysBase)*nShards)
colKeys := make([]string, 0, len(rowKeysBase)*nShards)
rowIDs = rowIDs[:0]
for shard := uint64(0); shard < nShards; shard++ {
for i := range rowKeysBase {
colKeys = append(colKeys, fmt.Sprintf("s%d-%s", shard, colKeysBase[i]))
if keyedField {
rowKeys = append(rowKeys, rowKeysBase[i])
} else {
rowIDs = append(rowIDs, uint64(i))
}
}
}
req.ColumnKeys = colKeys
if keyedField {
req.RowKeys = rowKeys
} else {
req.RowIDs = rowIDs
}
var id string
var node *test.Command
for id, node = range nodesByID {
break
}
if err := node.API.Import(ctx, qcxsByID[id], req); err != nil {
t.Fatalf("importing data: %v", err)
}
expected, err := node.API.FindIndexKeys(ctx, indexData.indexName, colKeys...)
if err != nil {
t.Fatalf("looking up index keys: %v", err)
}
for key, id := range expected {
// CorruptAMutex should only corrupt things in position 1 of their
// shards...
if id%(1<<shardwidth.Exponent) != 1 {
delete(expected, key)
}
}
if keyedField {
fieldValues, err := node.API.FindFieldKeys(ctx, indexData.indexName, fieldData.fieldName, rowKeys...)
if err != nil {
t.Fatalf("looking up field keys: %v", err)
}
// Figure out which key got the value 3, delete any records
// which would have had that key, because they won't be
// conflicts.
for key, value := range fieldValues {
if value == 3 {
for offset, baseKey := range rowKeysBase {
if baseKey == key {
for i := offset; i < len(rowKeys); i += len(rowKeysBase) {
delete(expected, colKeys[i])
}
}
}
}
}
} else {
// we set rowKeys to 0-1-2-... for rowKeysBase items, which
// tells us which keys we expect to be 3 already.
for i := 3; i < len(rowIDs); i += len(rowKeysBase) {
delete(expected, colKeys[i])
}
}
// and then we break the mutex and close the Qcxs
for id, node := range nodesByID {
field, err := node.API.Field(ctx, indexData.indexName, fieldData.fieldName)
if err != nil {
t.Fatalf("requesting field %s from node %s: %v", fieldData.fieldName, id, err)
}
pilosa.CorruptAMutex(t, field, qcxsByID[id])
err = qcxsByID[id].Finish()
if err != nil {
t.Fatalf("closing out transaction on node %s: %v", id, err)
}
}
qcx := m0.API.Txf().NewQcx()
defer qcx.Abort()
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[string][]string)
if !ok {
t.Fatalf("expected map[string][]string, got %T", results)
}
seen := 0
for k, v := range mapped {
seen++
if _, ok := expected[k]; !ok {
t.Fatalf("unexpected collision on key %q", k)
}
if len(v) != 2 {
t.Fatalf("expected exactly two collisions")
}
}
if seen != len(expected) {
t.Fatalf("expected exactly %d records to have collisions, got %d", len(expected), seen)
}
} else {
mapped, ok := results.(map[string][]uint64)
if !ok {
t.Fatalf("expected map[string][]uint64, got %T", results)
}
seen := 0
for k, v := range mapped {
seen++
if _, ok := expected[k]; !ok {
t.Fatalf("unexpected collision on key %q", k)
}
if len(v) != 2 {
t.Fatalf("expected exactly two collisions")
}
}
if seen != len(expected) {
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))
}
})
}
}
/*
func TestAPI_MutexCheck(t *testing.T) {
c := test.MustRunCluster(t, 3)
defer c.Close()
@ -745,7 +1084,8 @@ 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)
}
@ -775,9 +1115,28 @@ 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)
}
} else {
}
// 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)
}
}
else {
mapped, ok := results.(map[uint64][]uint64)
if !ok {
t.Fatalf("expected map[uint64][]uint64, got %T", results)
@ -937,3 +1296,4 @@ func TestAPI_MutexCheck(t *testing.T) {
})
}
}
*/

4
client.go
View file

@ -85,7 +85,7 @@ type InternalClient interface {
GetNodeUsage(ctx context.Context, uri *URI) (map[string]NodeUsage, error)
GetPastQueries(ctx context.Context, uri *URI) ([]PastQueryStatus, error)
MutexCheck(ctx context.Context, uri *URI, index string, field string) (map[uint64]map[uint64][]uint64, error)
MutexCheck(ctx context.Context, uri *URI, index string, field string, details bool, limit int) (map[uint64]map[uint64][]uint64, error)
}
//===============
@ -254,6 +254,6 @@ func (n nopInternalClient) GetPastQueries(ctx context.Context, uri *URI) ([]Past
return nil, nil
}
func (n nopInternalClient) MutexCheck(ctx context.Context, uri *URI, index, field string) (map[uint64]map[uint64][]uint64, error) {
func (n nopInternalClient) MutexCheck(ctx context.Context, uri *URI, index, field string, details bool, limit int) (map[uint64]map[uint64][]uint64, error) {
return nil, nil
}

4
field.go
View file

@ -1231,7 +1231,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")
}
@ -1243,7 +1243,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

@ -587,8 +587,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

7
http/client.go
View file

@ -135,11 +135,14 @@ func (c *InternalClient) Schema(ctx context.Context) ([]*pilosa.IndexInfo, error
// MutexCheck uses the mutex-check endpoint to request mutex collision data
// from a single node.
func (c *InternalClient) MutexCheck(ctx context.Context, uri *pilosa.URI, indexName string, fieldName string) (map[uint64]map[uint64][]uint64, error) {
func (c *InternalClient) MutexCheck(ctx context.Context, uri *pilosa.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")

28
http/handler.go
View file

@ -2490,9 +2490,21 @@ 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
@ -2516,9 +2528,21 @@ 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

73
roaring/filter.go
View file

@ -767,18 +767,30 @@ 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 +810,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

11
roaring/filter_internal_test.go
View file

@ -349,18 +349,17 @@ 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}},
},
}
@ -371,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 {

19
view.go
View file

@ -17,6 +17,7 @@ package pilosa
import (
"context"
"fmt"
"math"
"os"
"path/filepath"
"runtime"
@ -449,7 +450,8 @@ 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)
@ -470,7 +472,8 @@ 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
}
@ -482,11 +485,23 @@ 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
}