diff --git a/api.go b/api.go index 4fca1a776..a0074371c 100644 --- a/api.go +++ b/api.go @@ -2179,15 +2179,6 @@ func (api *API) TranslateIndexDB(ctx context.Context, indexName string, partitio _, err := store.ReadFrom(rd) return err } - -// 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 { - idx := api.holder.Index(indexName) - field := idx.Field(fieldName) - store := field.TranslateStore() - _, 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 { @@ -2200,82 +2191,86 @@ func (api *API) mutexCheckThisNode(ctx context.Context, qcx *Qcx, indexName stri return field.MutexCheck(ctx, qcx) } -// mergeMutexCollisions adds collisions to an existing map if they aren't already -// present. It modifies dst. -func mergeMutexCollisions(dst, src map[uint64][]uint64) { - for k, v := range src { - if len(v) == 0 { - continue +// 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++ } - existing := dst[k] - if len(existing) == 0 { - dst[k] = v - continue - } - // existing and v are both non-empty lists of collisions for this key. - // but if replication is working, both nodes should have the SAME list - // of collisions, so it's probably worth special-casing that check: - if len(existing) == len(v) { - different := false - for i := range existing { - if v[i] != existing[i] { - different = true - break - } - } - // yay, we can just ignore this - if !different { - continue - } - } - // combine... - existing = append(existing, v...) - // sort... - sort.Slice(existing, func(i, j int) bool { - return existing[i] < existing[j] - }) - // dedup. - n := 0 - prev := existing[0] - for i := 0; i < len(existing); i++ { - if existing[i] != prev { - existing[n] = existing[i] - n++ - } - prev = existing[i] - } - dst[k] = existing[:n] + prev = dst[i] } + return dst } -// MutexCheck checks for collisions in a given mutex field. The response is -// a map[shard]map[column]values. -func (api *API) MutexCheck(ctx context.Context, qcx *Qcx, indexName string, fieldName string, remote bool) (map[uint64]map[uint64][]uint64, error) { +// 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") } - // short path: if this is an internal remote request, only try to solve the - // question for these shards. - if remote { - out, err := api.mutexCheckThisNode(ctx, qcx, indexName, fieldName) - return out, err - } - var nodes []*Node - if !remote { - nodes = Nodes(api.cluster.nodes).Clone() - } else { - nodes = []*Node{api.cluster.nodeByID(api.server.nodeID)} - } + return api.mutexCheckThisNode(ctx, qcx, indexName, fieldName) +} - /* - // request data from other nodes as well - snap := topology.NewClusterSnapshot(api.cluster.noder, api.cluster.Hasher, api.cluster.ReplicaN) - */ +// 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) - myID := api.server.nodeID + 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) { @@ -2289,36 +2284,166 @@ func (api *API) MutexCheck(ctx context.Context, qcx *Qcx, indexName string, fiel }) } } - err := eg.Wait() + err = eg.Wait() if err != nil { return nil, err } - var out map[uint64]map[uint64][]uint64 + // 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 } - if out == nil { - out = nodeResults - continue - } - for k, v := range nodeResults { + for _, v := range nodeResults { if len(v) == 0 { continue } - existing := out[k] - // results from two nodes. might be normal with replication. - if len(existing) == 0 { - out[k] = v - continue + for record, values := range v { + process(record, values) } - // we have two maps for this shard. whee. - mergeMutexCollisions(existing, v) - // we don't store it back into out[k] because it was already - // a non-empty map, so stores to it update it. yay? } } - return out, nil + 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 { + idx := api.holder.Index(indexName) + field := idx.Field(fieldName) + store := field.TranslateStore() + _, err := store.ReadFrom(rd) + return err } type serverInfo 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<