mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-09-12 07:41:02 +00:00
cleaup api and add keytranslation
This commit is contained in:
Todd Gruben
parent
dfea30e18a
commit
32ad1db2b2
6 changed files with 536 additions and 93 deletions
305
api.go
305
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 {
|
||||
|
|
|
|||
315
api_test.go
315
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<<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)
|
||||
if err != nil {
|
||||
t.Fatalf("checking mutexes: %v", err)
|
||||
}
|
||||
if keyedField {
|
||||
// this just sorta comes out this way with our hashing; these are
|
||||
// the things which were in position 1 of their shards, and did
|
||||
// not have a value which happens to map to 3.
|
||||
mapped, ok := results.(map[string][]string)
|
||||
if !ok {
|
||||
t.Fatalf("expected map[string][]string, got %T", results)
|
||||
}
|
||||
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)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -939,7 +939,7 @@ func CorruptAMutex(tb testing.TB, field *Field, qcx *Qcx) {
|
|||
}
|
||||
// set a bonus bit, bypassing the mutex handling
|
||||
frag.mu.Lock()
|
||||
_, err = frag.unprotectedSetBit(tx, 3, frag.shard<<shardwidth.Exponent)
|
||||
_, err = frag.unprotectedSetBit(tx, 3, (frag.shard<<shardwidth.Exponent)+1)
|
||||
frag.mu.Unlock()
|
||||
if err != nil {
|
||||
tb.Fatalf("setting bit: %v", err)
|
||||
|
|
|
|||
1
go.mod
1
go.mod
|
|
@ -14,6 +14,7 @@ require (
|
|||
github.com/fsnotify/fsnotify v1.4.9 // indirect
|
||||
github.com/glycerine/goconvey v0.0.0-20190410193231-58a59202ab31 // indirect
|
||||
github.com/glycerine/idem v0.0.0-20190127113923-7a8083893311
|
||||
github.com/glycerine/vprint v0.0.0-20200730000117-76cea49a68ea // indirect
|
||||
github.com/gogo/protobuf v1.2.1
|
||||
github.com/golang/protobuf v1.4.2
|
||||
github.com/google/go-cmp v0.5.2
|
||||
|
|
|
|||
2
go.sum
2
go.sum
|
|
@ -74,6 +74,8 @@ github.com/glycerine/goconvey v0.0.0-20190410193231-58a59202ab31 h1:gclg6gY70GLy
|
|||
github.com/glycerine/goconvey v0.0.0-20190410193231-58a59202ab31/go.mod h1:Ogl1Tioa0aV7gstGFO7KhffUsb9M4ydbEbbxpcEDc24=
|
||||
github.com/glycerine/idem v0.0.0-20190127113923-7a8083893311 h1:AAXH0ZvYIHHqU06ASy0H2tYAkAGrQlZvEy2QZrrtt4E=
|
||||
github.com/glycerine/idem v0.0.0-20190127113923-7a8083893311/go.mod h1:B72P/ZM99sNiCmaQJflpmMAF5LsDzStpLdWzn0+Vr2Y=
|
||||
github.com/glycerine/vprint v0.0.0-20200730000117-76cea49a68ea h1:Uiuhuh77mImdrAMjPfw2V8tWw4AF6r9dxbNkECo23SA=
|
||||
github.com/glycerine/vprint v0.0.0-20200730000117-76cea49a68ea/go.mod h1:q7RHAiHHxYrXtGEkX14OuACg+cHODdKnVvTgpBnOzHk=
|
||||
github.com/go-gl/glfw v0.0.0-20190409004039-e6da0acd62b1/go.mod h1:vR7hzQXu2zJy9AVAgeJqvqgH9Q5CA+iKCZ2gyEVpxRU=
|
||||
github.com/go-gl/glfw/v3.3/glfw v0.0.0-20200222043503-6f7a984d4dc4/go.mod h1:tQ2UAYgL5IevRw8kRxooKSPJfGvJ9fJQFa0TUsXzTg8=
|
||||
github.com/go-kit/kit v0.8.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2as=
|
||||
|
|
|
|||
|
|
@ -2492,7 +2492,7 @@ func (h *Handler) handleGetMutexCheck(w http.ResponseWriter, r *http.Request) {
|
|||
indexName, fieldName := mux.Vars(r)["index"], mux.Vars(r)["field"]
|
||||
qcx := h.api.Txf().NewQcx()
|
||||
defer qcx.Abort()
|
||||
out, err := h.api.MutexCheck(r.Context(), qcx, indexName, fieldName, false)
|
||||
out, err := h.api.MutexCheck(r.Context(), qcx, indexName, fieldName)
|
||||
if err != nil {
|
||||
http.Error(w, err.Error(), http.StatusInternalServerError)
|
||||
return
|
||||
|
|
@ -2518,7 +2518,7 @@ func (h *Handler) handleInternalGetMutexCheck(w http.ResponseWriter, r *http.Req
|
|||
indexName, fieldName := mux.Vars(r)["index"], mux.Vars(r)["field"]
|
||||
qcx := h.api.Txf().NewQcx()
|
||||
defer qcx.Abort()
|
||||
out, err := h.api.MutexCheck(r.Context(), qcx, indexName, fieldName, true)
|
||||
out, err := h.api.MutexCheckNode(r.Context(), qcx, indexName, fieldName)
|
||||
if err != nil {
|
||||
http.Error(w, err.Error(), http.StatusInternalServerError)
|
||||
return
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue