From d2dc8f391ff88504ed9eaedaa37c6d8767ad945c Mon Sep 17 00:00:00 2001 From: Todd Gruben Date: Thu, 9 Sep 2021 10:11:51 -0500 Subject: [PATCH] backport 1687 --- api.go | 172 ++++++++++++--- api_test.go | 366 +++++++++++++++++++++++++++++++- client.go | 4 +- field.go | 4 +- fragment.go | 4 +- http/client.go | 7 +- http/handler.go | 28 ++- roaring/filter.go | 73 +++++-- roaring/filter_internal_test.go | 11 +- view.go | 19 +- 10 files changed, 621 insertions(+), 67 deletions(-) diff --git a/api.go b/api.go index 34548b794..2ce422608 100644 --- a/api.go +++ b/api.go @@ -2179,7 +2179,7 @@ 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) { +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) @@ -2188,7 +2188,27 @@ 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 @@ -2233,21 +2253,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) { +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") } @@ -2264,22 +2288,21 @@ func (api *API) MutexCheck(ctx context.Context, qcx *Qcx, indexName string, fiel } nodes := Nodes(api.cluster.nodes).Clone() eg, _ := errgroup.WithContext(ctx) + myID := api.Node().ID 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) + 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 }) } @@ -2288,23 +2311,33 @@ 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. + // 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 var fieldIDs []uint64 - // build translation tables, if we need them + // 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) + // 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 { @@ -2326,7 +2359,12 @@ func (api *API) MutexCheck(ctx context.Context, qcx *Qcx, indexName string, fiel } } } - // we now have lists of the keys, so... + // if context is done, return early. + if err := ctx.Err(); err != nil { + return nil, err + } + untranslatedKeys := 0 + // Obtain translation tables for the keys. if useIndexKeys { indexKeyList, err := api.cluster.translateIndexIDs(ctx, indexName, indexIDs) if err != nil { @@ -2336,9 +2374,17 @@ func (api *API) MutexCheck(ctx context.Context, qcx *Qcx, indexName string, fiel 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 indexKeyList[i] != "" { + indexKeys[indexIDs[i]] = indexKeyList[i] + } else { + untranslatedKeys++ + } } } + // if context is done, return early. + if err := ctx.Err(); err != nil { + return nil, err + } if useFieldKeys { fieldKeyList, err := api.cluster.translateFieldListIDs(field, fieldIDs) if err != nil { @@ -2348,19 +2394,49 @@ func (api *API) MutexCheck(ctx context.Context, qcx *Qcx, indexName string, fiel return nil, fmt.Errorf("translating %d IDs, got %d keys", len(indexIDs), len(fieldKeyList)) } for i := range fieldIDs { - fieldKeys[fieldIDs[i]] = fieldKeyList[i] + if fieldKeyList[i] != "" { + fieldKeys[fieldIDs[i]] = fieldKeyList[i] + } else { + untranslatedKeys++ + } } } - + if untranslatedKeys > 0 { + api.server.logger.Printf("translating mutex check results: %d key(s) untranslated", untranslatedKeys) + } } + + // if context is done, return early. + if err := ctx.Err(); err != nil { + return nil, err + } + // 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() { + if err == nil { + 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]) @@ -2368,32 +2444,52 @@ 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() { + if err == nil { + 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]) @@ -2407,34 +2503,54 @@ 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 } + // if context is done, return early. + if err := ctx.Err(); err != nil { + return nil, err + } for _, v := range nodeResults { if len(v) == 0 { continue } + counter := 0 for record, values := range v { - process(record, values) + counter++ + if process(record, values) { + break processing + } + // every 65k items or so, check the context for done-ness + if counter%(1<<16) == 0 { + if err := ctx.Err(); err != nil { + return nil, err + } + } } } } - return result, nil + return result, ctx.Err() } // TranslateFieldDB is an internal function to load the field keys database diff --git a/api_test.go b/api_test.go index bec6c8b1c..d5b1d26a4 100644 --- a/api_test.go +++ b/api_test.go @@ -637,6 +637,345 @@ type mutexCheckField struct { 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 { + 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() + + // 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, + } + + 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 { + 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, 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] + 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<= 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 1811acff4..92b5372bb 100644 --- a/roaring/filter_internal_test.go +++ b/roaring/filter_internal_test.go @@ -349,18 +349,17 @@ 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}}, }, } @@ -371,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 0af36b71a..af2753127 100644 --- a/view.go +++ b/view.go @@ -17,6 +17,7 @@ package pilosa import ( "context" "fmt" + "math" "os" "path/filepath" "runtime" @@ -449,7 +450,8 @@ 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) @@ -470,7 +472,8 @@ 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 } @@ -482,11 +485,23 @@ 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 }