diff --git a/api.go b/api.go index ff9ce8dc3..648696601 100644 --- a/api.go +++ b/api.go @@ -2653,7 +2653,7 @@ 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) { +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) @@ -2662,7 +2662,26 @@ 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 @@ -2707,21 +2726,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) { +// details false: +// []uint64 // unkeyed index +// []string // keyed index +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") } @@ -2746,12 +2769,12 @@ func (api *API) MutexCheck(ctx context.Context, qcx *Qcx, indexName string, fiel 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 }) } @@ -2760,12 +2783,18 @@ 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. 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 @@ -2778,8 +2807,9 @@ func (api *API) MutexCheck(ctx context.Context, qcx *Qcx, indexName string, fiel // 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 { @@ -2843,11 +2873,27 @@ func (api *API) MutexCheck(ctx context.Context, qcx *Qcx, indexName string, fiel // 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() { + 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]) @@ -2855,32 +2901,50 @@ 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() { + 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]) @@ -2894,20 +2958,26 @@ 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 @@ -2917,7 +2987,9 @@ func (api *API) MutexCheck(ctx context.Context, qcx *Qcx, indexName string, fiel continue } for record, values := range v { - process(record, values) + if process(record, values) { + break processing + } } } } diff --git a/api_test.go b/api_test.go index bb2764aa2..146924b0b 100644 --- a/api_test.go +++ b/api_test.go @@ -966,10 +966,6 @@ 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) - 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{ @@ -980,6 +976,12 @@ func TestAPI_MutexCheck(t *testing.T) { (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 { @@ -1014,9 +1016,29 @@ 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) } } + + // 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] @@ -1111,7 +1133,7 @@ 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) } @@ -1155,6 +1177,28 @@ func TestAPI_MutexCheck(t *testing.T) { 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) + } + // 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.([]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)) + } }) } } diff --git a/client.go b/client.go index d3c20cb73..62f16e240 100644 --- a/client.go +++ b/client.go @@ -80,7 +80,7 @@ type InternalClient interface { RetrieveTranslatePartitionFromURI(ctx context.Context, index string, partition int, uri pnet.URI) (io.ReadCloser, error) ImportRoaring(ctx context.Context, uri *pnet.URI, index, field string, shard uint64, remote bool, req *ImportRoaringRequest) error ShardReader(ctx context.Context, index string, shard uint64) (io.ReadCloser, error) - MutexCheck(ctx context.Context, uri *pnet.URI, index string, field string) (map[uint64]map[uint64][]uint64, error) + MutexCheck(ctx context.Context, uri *pnet.URI, index string, field string, details bool, limit int) (map[uint64]map[uint64][]uint64, error) IDAllocDataReader(ctx context.Context) (io.ReadCloser, error) IndexTranslateDataReader(ctx context.Context, index string, partitionID int) (io.ReadCloser, error) @@ -214,7 +214,7 @@ func (n nopInternalClient) ImportRoaring(ctx context.Context, uri *pnet.URI, ind return nil } -func (n nopInternalClient) MutexCheck(ctx context.Context, uri *pnet.URI, index, field string) (map[uint64]map[uint64][]uint64, error) { +func (n nopInternalClient) MutexCheck(ctx context.Context, uri *pnet.URI, index, field string, details bool, limit int) (map[uint64]map[uint64][]uint64, error) { return nil, nil } diff --git a/field.go b/field.go index 34e7f2f22..793371272 100644 --- a/field.go +++ b/field.go @@ -1094,7 +1094,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") } @@ -1106,7 +1106,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. diff --git a/fragment.go b/fragment.go index 73da513c0..d545a1726 100644 --- a/fragment.go +++ b/fragment.go @@ -601,8 +601,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<= 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 diff --git a/roaring/filter_internal_test.go b/roaring/filter_internal_test.go index 6de5cb7c6..a46c891b0 100644 --- a/roaring/filter_internal_test.go +++ b/roaring/filter_internal_test.go @@ -350,16 +350,16 @@ 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}}, }, } @@ -370,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 { diff --git a/view.go b/view.go index e60105ceb..bb0eb3a91 100644 --- a/view.go +++ b/view.go @@ -17,6 +17,7 @@ package pilosa import ( "context" "fmt" + "math" "os" "path/filepath" "runtime" @@ -444,7 +445,7 @@ 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) @@ -465,7 +466,7 @@ 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 } @@ -477,11 +478,22 @@ 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 }