diff --git a/.circleci/config.yml b/.circleci/config.yml index ef6b7ce80..ce623b054 100644 --- a/.circleci/config.yml +++ b/.circleci/config.yml @@ -67,7 +67,7 @@ jobs: name: golang steps: - run: '[[ -n $CIRCLE_PULL_REQUEST ]] || circleci step halt || true' # Skip if this is not a pull request - - run: curl https://moleculacorp:$GITHUB_PERSONAL_ACCESS_TOKEN@api.github.com/repos/molecula/pilosa/pulls/$(basename $CIRCLE_PULL_REQUEST) | jq "[.labels[] | .name | startswith(\"changelog\")] | any" -e + - run: curl https://moleculacorp:$GITHUB_PERSONAL_ACCESS_TOKEN@api.github.com/repos/molecula/featurebase/pulls/$(basename $CIRCLE_PULL_REQUEST) | jq "[.labels[] | .name | startswith(\"changelog\")] | any" -e test-build-arm: executor: name: golang diff --git a/api.go b/api.go index d2c09fe85..a0074371c 100644 --- a/api.go +++ b/api.go @@ -2179,6 +2179,263 @@ 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) { + 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 +} + +// 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 +} + +// 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") + } + nodes := Nodes(api.cluster.nodes).Clone() + eg, _ := errgroup.WithContext(ctx) + + 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) + 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. + useIndexKeys := index.Keys() + useFieldKeys := field.Keys() + var indexKeys = map[uint64]string{} + var fieldKeys = map[uint64]string{} + var indexIDs []uint64 + var fieldIDs []uint64 + // build translation tables, if we need them + 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) + } + } + } + } + } + } + // we now have lists of the keys, so... + 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 { + indexKeys[indexIDs[i]] = indexKeyList[i] + } + } + 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 { + fieldKeys[fieldIDs[i]] = fieldKeyList[i] + } + } + + } + // 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 +} // TranslateFieldDB is an internal function to load the field keys database func (api *API) TranslateFieldDB(ctx context.Context, indexName, fieldName string, rd io.Reader) error { @@ -2249,11 +2506,13 @@ const ( apiIDReserve apiIDCommit apiIDReset + apiMutexCheck //Backported from 1681 Caution ) var methodsCommon = map[apiMethod]struct{}{ apiClusterMessage: {}, apiSetCoordinator: {}, + apiMutexCheck: {}, } var methodsResizing = map[apiMethod]struct{}{ diff --git a/api_test.go b/api_test.go index 3f6369476..bec6c8b1c 100644 --- a/api_test.go +++ b/api_test.go @@ -28,6 +28,7 @@ import ( "github.com/pilosa/pilosa/v2/boltdb" "github.com/pilosa/pilosa/v2/http" "github.com/pilosa/pilosa/v2/server" + "github.com/pilosa/pilosa/v2/shardwidth" "github.com/pilosa/pilosa/v2/test" ) @@ -622,3 +623,317 @@ func TestAPI_ClearFlagForImportAndImportValues(t *testing.T) { panic(fmt.Sprintf("expected %v, observed %v starting acct0 balance", acct0bal, 0)) } } + +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.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 { + fmt.Println("running:", indexData.indexName, fieldData.fieldName) + 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 0fc9c0fe0..1811acff4 100644 --- a/roaring/filter_internal_test.go +++ b/roaring/filter_internal_test.go @@ -349,3 +349,50 @@ 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/roaring/roaring.go b/roaring/roaring.go index 5504b9bf1..32d573911 100644 --- a/roaring/roaring.go +++ b/roaring/roaring.go @@ -4448,10 +4448,13 @@ func intersectionCallbackArrayArray(a, b *Container, fn func(uint16)) { if (na << 2) < nb { for _, va := range ca { for cb[0] < va { - if len(cb) > 8 && cb[0] < va { + // try to skip ahead a bit faster + for len(cb) > 7 && cb[7] < va { cb = cb[8:] } - cb = cb[1:] + for len(cb) > 0 && cb[0] < va { + cb = cb[1:] + } if len(cb) == 0 { return } @@ -4540,7 +4543,7 @@ func intersectionCallbackBitmapRun(a, b *Container, fn func(uint16)) { } } -func intersectionCallbackArrayBitmap(a, b *Container, fn func(uint16)) (n int32) { +func intersectionCallbackArrayBitmap(a, b *Container, fn func(uint16)) { statsHit("intersectionCount/ArrayBitmap") bitmap := b.bitmap() ln := len(bitmap) @@ -4550,9 +4553,10 @@ func intersectionCallbackArrayBitmap(a, b *Container, fn func(uint16)) (n int32) break } off := val % 64 - n += int32(bitmap[i]>>off) & 1 + if (bitmap[i]>>off)&1 != 0 { + fn(val) + } } - return n } func intersectionCallbackBitmapBitmap(a, b *Container, fn func(uint16)) { diff --git a/view.go b/view.go index 15173f5e4..0af36b71a 100644 --- a/view.go +++ b/view.go @@ -447,6 +447,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