diff --git a/api.go b/api.go index aa7cdc79a..ff9ce8dc3 100644 --- a/api.go +++ b/api.go @@ -2653,6 +2653,277 @@ 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) { + index := api.holder.Index(indexName) + if index == nil { + return nil, newNotFoundError(ErrIndexNotFound, indexName) + } + field := index.Field(fieldName) + if field == nil { + return nil, newNotFoundError(ErrFieldNotFound, fieldName) + } + return field.MutexCheck(ctx, qcx) +} + +// mergeIDLists merges a list of numeric IDs into another list, removing +// duplicates. +func mergeIDLists(dst []uint64, src []uint64) []uint64 { + dst = append(dst, src...) + sort.Slice(dst, func(i, j int) bool { + return dst[i] < dst[j] + }) + // dedup. + n := 0 + prev := dst[0] + for i := 0; i < len(dst); i++ { + if dst[i] != prev { + dst[n] = dst[i] + n++ + } + prev = dst[i] + } + return dst[:n] +} + +// mergeKeyLists merges a list of string IDs into another list, removing +// duplicates. +func mergeKeyLists(dst []string, src []string) []string { + dst = append(dst, src...) + sort.Slice(dst, func(i, j int) bool { + return dst[i] < dst[j] + }) + // dedup. + n := 0 + prev := dst[0] + for i := 0; i < len(dst); i++ { + if dst[i] != prev { + dst[n] = dst[i] + n++ + } + prev = dst[i] + } + return dst[:n] +} + +// 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) { + if err := api.validate(apiMutexCheck); err != nil { + return nil, errors.Wrap(err, "validating api method") + } + return api.mutexCheckThisNode(ctx, qcx, indexName, fieldName) +} + +// 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: +// 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) { + if err = api.validate(apiMutexCheck); err != nil { + return nil, errors.Wrap(err, "validating api method") + } + index, err := api.Index(ctx, indexName) + if err != nil { + return nil, err + } + field, err := api.Field(ctx, indexName, fieldName) + if err != nil { + return nil, err + } + if field.Type() != FieldTypeMutex { + return nil, errors.New("can only check mutex state for mutex fields") + } + // request data from other nodes as well + snap := topology.NewClusterSnapshot(api.cluster.noder, api.cluster.Hasher, api.cluster.ReplicaN) + eg, _ := errgroup.WithContext(ctx) + myID := api.NodeID() + results := make([]map[uint64]map[uint64][]uint64, len(snap.Nodes)) + for i, node := range snap.Nodes { + i := i // loop variable shadowing is a war crime + 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) + return err + }) + } else { + eg.Go(func() (err error) { + results[i], err = api.mutexCheckThisNode(ctx, qcx, indexName, fieldName) + return err + }) + } + } + err = eg.Wait() + if err != nil { + return nil, err + } + // 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() + var indexKeys = map[uint64]string{} + var fieldKeys = map[uint64]string{} + var indexIDs []uint64 + var fieldIDs []uint64 + // 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) + if useIndexKeys || useFieldKeys { + for _, nodeResults := range results { + for _, shardResults := range nodeResults { + for record, values := range shardResults { + if useIndexKeys { + if _, ok := indexKeys[record]; !ok { + indexKeys[record] = untranslated + indexIDs = append(indexIDs, record) + } + } + if useFieldKeys { + for _, value := range values { + if _, ok := fieldKeys[value]; !ok { + fieldKeys[value] = untranslated + fieldIDs = append(fieldIDs, value) + } + } + } + } + } + } + untranslatedKeys := 0 + // Obtain translation tables for the keys. + if useIndexKeys { + indexKeyList, err := api.cluster.translateIndexIDs(ctx, indexName, indexIDs) + if err != nil { + return nil, errors.Wrap(err, "translating index keys") + } + if len(indexKeyList) != len(indexIDs) { + return nil, fmt.Errorf("translating %d record IDs, got %d keys", len(indexIDs), len(indexKeyList)) + } + for i := range indexIDs { + if indexKeyList[i] != "" { + indexKeys[indexIDs[i]] = indexKeyList[i] + } else { + untranslatedKeys++ + } + } + } + if useFieldKeys { + fieldKeyList, err := api.cluster.translateFieldListIDs(field, fieldIDs) + if err != nil { + return nil, errors.Wrap(err, "translating index keys") + } + if len(fieldKeyList) != len(fieldIDs) { + return nil, fmt.Errorf("translating %d IDs, got %d keys", len(indexIDs), len(fieldKeyList)) + } + for i := range fieldIDs { + if fieldKeyList[i] != "" { + fieldKeys[fieldIDs[i]] = fieldKeyList[i] + } else { + untranslatedKeys++ + } + } + } + if untranslatedKeys > 0 { + api.server.logger.Warnf("translating mutex check results: %d key(s) untranslated", untranslatedKeys) + } + } + + // define the process functions. separated from above code just to make + // it easier to follow/compare them. + if useIndexKeys { + if useFieldKeys { + outMap := make(map[string][]string) + var valueKeys []string + result = outMap + process = func(recordID uint64, valueIDs []uint64) { + valueKeys = valueKeys[:0] + for _, id := range valueIDs { + valueKeys = append(valueKeys, fieldKeys[id]) + } + record := indexKeys[recordID] + if existing, ok := outMap[record]; ok { + outMap[record] = mergeKeyLists(existing, valueKeys) + } else { + // 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...) + } + } + } else { + outMap := make(map[string][]uint64) + result = outMap + process = func(recordID uint64, values []uint64) { + record := indexKeys[recordID] + if existing, ok := outMap[record]; ok { + outMap[record] = mergeIDLists(existing, values) + } else { + outMap[record] = values + } + } + } + } else { + if useFieldKeys { + outMap := make(map[uint64][]string) + var valueKeys []string + result = outMap + process = func(record uint64, valueIDs []uint64) { + valueKeys = valueKeys[:0] + for _, id := range valueIDs { + valueKeys = append(valueKeys, fieldKeys[id]) + } + if existing, ok := outMap[record]; ok { + outMap[record] = mergeKeyLists(existing, valueKeys) + } else { + // 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...) + } + } + } else { + outMap := make(map[uint64][]uint64) + result = outMap + process = func(record uint64, values []uint64) { + if existing, ok := outMap[record]; ok { + outMap[record] = mergeIDLists(existing, values) + } else { + outMap[record] = values + } + } + } + } + + for _, nodeResults := range results { + if len(nodeResults) == 0 { + continue + } + for _, v := range nodeResults { + if len(v) == 0 { + continue + } + for record, values := range v { + process(record, values) + } + } + } + return result, nil +} + type serverInfo struct { ShardWidth uint64 `json:"shardWidth"` ReplicaN int `json:"replicaN"` @@ -2713,6 +2984,7 @@ const ( apiIDReset apiPartitionNodes apiIngestOperations + apiMutexCheck ) var methodsCommon = map[apiMethod]struct{}{ @@ -2781,4 +3053,5 @@ var methodsNormal = map[apiMethod]struct{}{ apiIDReset: {}, apiPartitionNodes: {}, apiIngestOperations: {}, + apiMutexCheck: {}, } diff --git a/api_test.go b/api_test.go index 7e34f5be4..bb2764aa2 100644 --- a/api_test.go +++ b/api_test.go @@ -30,6 +30,7 @@ import ( "github.com/molecula/featurebase/v2/boltdb" "github.com/molecula/featurebase/v2/http" "github.com/molecula/featurebase/v2/server" + "github.com/molecula/featurebase/v2/shardwidth" "github.com/molecula/featurebase/v2/test" . "github.com/molecula/featurebase/v2/vprint" // nolint:staticcheck ) @@ -844,3 +845,316 @@ func TestAPI_IDAlloc(t *testing.T) { } }) } + +type mutexCheckIndex struct { + index *pilosa.Index + indexName string + createdAt int64 + fields map[bool]mutexCheckField +} + +type mutexCheckField struct { + fieldName string + field *pilosa.Field + 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.NodeID() + 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() + + 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{ + (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, + } + 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", len(expected)) + } + } + }) + } + 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< 0 { + return key.NeedData() + } + return key.RejectOne() +} + +func (b *BitmapMutexDupFilter) ConsiderData(key FilterKey, data *Container) FilterResult { + value, basePos := uint64(key)>>rowExponent, uint64(key&keyMask)<<16 + 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) + } else { + b.first[pos] = value + } + }) + 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 + } + } + return b.extra +} + // ApplyFilterToIterator is a simplistic implementation that applies a bitmap // filter to a ContainerIterator, returning an error if it encounters an error. // diff --git a/roaring/filter_internal_test.go b/roaring/filter_internal_test.go index 40abd7e1e..6de5cb7c6 100644 --- a/roaring/filter_internal_test.go +++ b/roaring/filter_internal_test.go @@ -347,5 +347,51 @@ func TestFilterWithRows(t *testing.T) { } }) } - +} + +func TestMutexDupFilter(t *testing.T) { + tests := []struct{ + pairs [][2]uint64 + expect map[uint64][]uint64 + }{ + { + 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}}, + expect: map[uint64][]uint64{0: {0, 1, 2}}, + }, + } + for num, test := range tests { + t.Run(fmt.Sprintf("case%d", num), func(t *testing.T) { + b := NewSliceBitmap() + for _, p := range test.pairs { + v := (p[1] << shardwidth.Exponent) | p[0] + b.DirectAdd(v) + } + dup := NewBitmapMutexDupFilter(0) + iter, _ := b.Containers.Iterator(0) + err := ApplyFilterToIterator(dup, iter) + if err != nil { + t.Fatalf("applying filter: %v", err) + } + expected := test.expect + got := dup.Report() + if len(expected) != len(got) { + t.Fatalf("expected %d entries in duplicate map, got %d", len(expected), len(got)) + } + for k, v := range expected { + gv := got[k] + if len(v) != len(gv) { + t.Fatalf("for id %d, expected %d (len %d), got %d (len %d)", k, v, len(v), gv, len(gv)) + } + for j := range v { + if gv[j] != v[j] { + t.Fatalf("for id %d, expected %d, got %d", k, v[j], gv[j]) + } + } + } + }) + } } diff --git a/view.go b/view.go index c38bd8572..e60105ceb 100644 --- a/view.go +++ b/view.go @@ -442,6 +442,50 @@ 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) { + // 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) + throttle := make(chan struct{}, runtime.NumCPU()) + frags := v.allFragments() + results := make([]map[uint64][]uint64, len(frags)) + for i, frag := range frags { + // local copies for the goroutine to use + i, frag := i, frag + eg.Go(func() error { + // limit simultaneous parallel goroutines associated with this + throttle <- struct{}{} + defer func() { + <-throttle + }() + tx, finisher, err := qcx.GetTx(Txo{Index: v.idx, Shard: frag.shard}) + if err != nil { + return err + } + defer finisher(&err) + results[i], err = frag.mutexCheck(tx) + if err != nil { + return err + } + return nil + }) + } + err := eg.Wait() + if err != nil { + return nil, err + } + out := map[uint64]map[uint64][]uint64{} + for i, result := range results { + if len(result) == 0 { + continue + } + out[frags[i].shard] = result + } + return out, nil +} + // setBit sets a bit within the view. func (v *view) setBit(txOrig Tx, rowID, columnID uint64) (changed bool, err error) { shard := columnID / ShardWidth