diff --git a/cache.go b/cache.go index b03e4951c..746934c47 100644 --- a/cache.go +++ b/cache.go @@ -262,6 +262,8 @@ func (c *rankCache) invalidate() { // The cache will remain flagged as dirty and will be recalculated if Top is called. // This may cause unexpected memory growth, so record it in metrics for debugging purposes. c.stats.Count(MetricInvalidateCacheSkipped, 1, 1.0) + // Ensure that we're marked as dirty even if we weren't otherwise. + c.dirty = true return } c.stats.Count(MetricInvalidateCache, 1, 1.0) diff --git a/fragment.go b/fragment.go index 5a9d0b438..f8f928581 100644 --- a/fragment.go +++ b/fragment.go @@ -2525,7 +2525,7 @@ func (f *fragment) importPositions(tx Tx, set, clear []uint64, rowSet map[uint64 } if f.CacheType != CacheTypeNone { - f.cache.Recalculate() + f.cache.Invalidate() } return nil } @@ -2823,7 +2823,7 @@ func (f *fragment) unprotectedImportRoaring(ctx context.Context, tx Tx, data []b } // we only set this if we need to update the cache if anyChanged { - f.cache.Recalculate() + f.cache.Invalidate() } span, _ = tracing.StartSpanFromContext(ctx, "importRoaring.incrementOpN") diff --git a/fragment_internal_test.go b/fragment_internal_test.go index 6afaf4d2a..54e3bb863 100644 --- a/fragment_internal_test.go +++ b/fragment_internal_test.go @@ -5558,7 +5558,7 @@ func requireMutexSampleData(tb testing.TB) { // a few mutex tests want common largeish pools of mutex data type mutexSampleData struct { name string - colIDs, rowIDs [2][]uint64 + colIDs, rowIDs [3][]uint64 } // scratchSpace copies the values over corresponding entries in slices, @@ -5628,125 +5628,123 @@ var mutexDensities = []mutexDensity{ {"64K", 16}, // {"32K", 15}, // 50-50 // {"16K", 14}, // 1/4 - // {"4K", 12}, // a fair number of things + // {"8K", 13}, + // {"4K", 12}, // a fair number of things + {"1K", 10}, // {"1", 0}, // about one per container // {"empty", -14}, // almost none } var mutexSizes = []mutexSize{ - {"4r", 2}, - {"16r", 4}, - {"256r", 8}, + // {"4r", 2}, + // {"16r", 4}, + // {"256r", 8}, + {"2Kr", 11}, // {"65Kr", 16}, } var mutexCaches = []string{ - // "ranked", + "ranked", "none", } -const mutexSampleDataSize = ShardWidth << 1 +const mutexSampleDataSize = ShardWidth * len(mutexSampleData{}.colIDs) -// prepareMutexSampleData creates two sets of data for each density and +// prepareMutexSampleData creates multiple sets of data for each density and // number of rows, so that we can test performance when overwriting also. func prepareMutexSampleData(tb testing.TB) { myrand := rand.New(rand.NewSource(9)) for _, d := range mutexDensities { + // at density 16, we want everything to be adjacent. + // at density 0, we want about 65k between items. + // The average spacing we want is 1<<(16 - density), + // so random numbers between 0 and twice that would + // be close, but we never want 0, so, subtract 1 from + // "twice that", then add 1 to the result. + // + // So for density 16, we compute spacing of 1, then + // draw random numbers in [0,1), and add 1 to them. + spacing := ((1 << (16 - d.density)) * 2) - 1 for _, s := range mutexSizes { rng := newMutexSampleRange(d.density, s.rows) col := uint64(0) - // at density 16, we want everything to be adjacent. - // at density 0, we want about 65k between items. - // The average spacing we want is 1<<(16 - density), - // so random numbers between 0 and twice that would - // be close, but we never want 0, so, subtract 1 from - // "twice that", then add 1 to the result. - // - // So for density 16, we compute spacing of 1, then - // draw random numbers in [0,1), and add 1 to them. - spacing := ((1 << (16 - d.density)) * 2) - 1 + rows := (int64(1) << s.rows) - expected := mutexSampleDataSize - if (ShardWidth / spacing) < mutexSampleDataSize { - expected = (ShardWidth / spacing) * 2 - if expected < 2 { - expected = 2 - } - } + colIDs := make([]uint64, mutexSampleDataSize) rowIDs := make([]uint64, mutexSampleDataSize) data := &mutexSampleData{name: d.name + "/" + s.name} prev := uint64(0) generated := 0 - for i := 0; i < expected; i++ { - col += uint64(myrand.Int63n(int64(spacing))) + 1 + for idx := 0; int(prev) < len(data.colIDs); idx++ { + if spacing > 1 { + col += uint64(myrand.Int63n(int64(spacing))) + 1 + } else { + col++ + } // can only import one fragment at a time, // though! if col/ShardWidth > prev { - data.colIDs[prev] = colIDs[generated:i:i] - data.rowIDs[prev] = rowIDs[generated:i:i] - generated = i + data.colIDs[prev] = colIDs[generated:idx:idx] + data.rowIDs[prev] = rowIDs[generated:idx:idx] + generated = idx prev = col / ShardWidth if int(prev) >= len(data.colIDs) { break } } row := uint64(myrand.Int63n(rows)) - colIDs[i] = col % ShardWidth - rowIDs[i] = row - } - if int(prev) < len(data.colIDs) { - data.colIDs[prev] = colIDs[generated:expected:expected] - data.rowIDs[prev] = rowIDs[generated:expected:expected] + colIDs[idx] = col % ShardWidth + rowIDs[idx] = row } sampleMutexData[rng] = data } } } -var importBatchSizes = []int{65536} +var importBatchSizes = []int{40, 80, 240, 2048} func TestImportMutexSampleData(t *testing.T) { requireMutexSampleData(t) var scratchCols []uint64 var scratchRows []uint64 for rng, data := range sampleMutexData { - // skip the larger ones, they'll be slow - if rng.rows() > 256 { - continue - } scratchCols, scratchRows = data.scratchSpace(0, scratchCols, scratchRows) t.Run(data.name, func(t *testing.T) { - for _, batchSize := range importBatchSizes { - t.Run(fmt.Sprintf("%d", batchSize), func(t *testing.T) { - f, _, tx := mustOpenMutexFragment(t, "i", "f", viewStandard, 0, "") - defer f.Clean(t) - // Set import. - var err error - for i := 0; i < len(scratchCols); i += batchSize { - max := i + batchSize - if len(scratchCols) < max { - max = len(scratchCols) - } - err = f.bulkImport(tx, scratchRows[i:max:max], scratchCols[i:max:max], &ImportOptions{}) - if err != nil { - t.Fatalf("bulk importing ids [%d:%d]: %v", i, max, err) - } - } - count := uint64(0) - for k := uint32(0); k < rng.rows(); k++ { - count += f.mustRow(tx, uint64(k)).Count() - } - if int(count) != len(data.colIDs[0]) { - t.Fatalf("for %d rows, %d density: expected %d results, got %d", - rng.rows(), rng.density(), len(data.colIDs[0]), count) - } - }) + batchSize := 16384 + f, _, tx := mustOpenMutexFragment(t, "i", "f", viewStandard, 0, "") + defer f.Clean(t) + // Set import. + var err error + for i := 0; i < len(scratchCols); i += batchSize { + max := i + batchSize + if len(scratchCols) < max { + max = len(scratchCols) + } + err = f.bulkImport(tx, scratchRows[i:max:max], scratchCols[i:max:max], &ImportOptions{}) + if err != nil { + t.Fatalf("bulk importing ids [%d:%d]: %v", i, max, err) + } + } + count := uint64(0) + for k := uint32(0); k < rng.rows(); k++ { + count += f.mustRow(tx, uint64(k)).Count() + } + if int(count) != len(data.colIDs[0]) { + t.Fatalf("for %d rows, %d density: expected %d results, got %d", + rng.rows(), rng.density(), len(data.colIDs[0]), count) } }) } } +// BenchmarkImportMutexSampleData tries to time importing mutex data. +// The tricky part is defining a meaningful b.N that can apply across +// different batch sizes, densities, and so on. So, basically, we take +// b.N, and multiply by 65536, to get "N containers" of data, meaning +// that the amount of data we want to process is independent of all of +// the other factors. But for sparse data sets, that means rewriting +// the same data a number of times, which isn't ideal. func BenchmarkImportMutexSampleData(b *testing.B) { requireMutexSampleData(b) var cols []uint64 @@ -5757,16 +5755,27 @@ func BenchmarkImportMutexSampleData(b *testing.B) { var frag *fragment var tx Tx var idx *Index - benchmarkOneFragmentImports := func(b *testing.B, i int) { - cols, rows = data.scratchSpace(i, cols, rows) - for i := 0; i < len(cols) && i < (batchSize*b.N); i += batchSize { - max := i + batchSize + benchmarkOneFragmentImports := func(b *testing.B, idx int) { + cols, rows = data.scratchSpace(idx, cols, rows) + toDo := b.N << 16 + start := 0 + for toDo > 0 { + max := start + batchSize if len(cols) < max { max = len(cols) } - err := frag.bulkImport(tx, rows[i:max:max], cols[i:max:max], &ImportOptions{}) + err := frag.bulkImport(tx, rows[start:max:max], cols[start:max:max], &ImportOptions{}) if err != nil { - b.Fatalf("bulk importing ids [%d:%d]: %v", i, max, err) + b.Fatalf("bulk importing ids [%d:%d]: %v", start, max, err) + } + toDo -= (max - start) + start = max + if start >= len(cols) { + start = 0 + b.StopTimer() + // recreate data again because bulkImport overwrote it + cols, rows = data.scratchSpace(idx, cols, rows) + b.StartTimer() } } } diff --git a/http/handler.go b/http/handler.go index 94f14a011..ad795efb4 100644 --- a/http/handler.go +++ b/http/handler.go @@ -260,7 +260,6 @@ func (h *Handler) populateValidators() { h.validators["GetTransaction"] = queryValidationSpecRequired() h.validators["PostTransaction"] = queryValidationSpecRequired() h.validators["PostFinishTransaction"] = queryValidationSpecRequired() - h.validators["Inspect"] = queryValidationSpecRequired().Optional("indexes", "fields", "views", "shards", "checksum", "containers") } @@ -397,7 +396,6 @@ func newRouter(handler *Handler) http.Handler { router.HandleFunc("/index/{index}/field/{field}/import-roaring/{shard}", handler.handlePostImportRoaring).Methods("POST").Name("PostImportRoaring") router.HandleFunc("/index/{index}/query", handler.handlePostQuery).Methods("POST").Name("PostQuery") router.HandleFunc("/info", handler.handleGetInfo).Methods("GET").Name("GetInfo") - router.HandleFunc("/inspect", handler.handleInspect).Methods("GET").Name("Inspect") router.HandleFunc("/recalculate-caches", handler.handleRecalculateCaches).Methods("POST").Name("RecalculateCaches") router.HandleFunc("/schema", handler.handleGetSchema).Methods("GET").Name("GetSchema") router.HandleFunc("/schema/details", handler.handleGetSchemaDetails).Methods("GET").Name("GetSchemaDetails") @@ -803,37 +801,6 @@ func (h *Handler) handleGetInfo(w http.ResponseWriter, r *http.Request) { } } -func (h *Handler) handleInspect(w http.ResponseWriter, r *http.Request) { - if !validHeaderAcceptJSON(r.Header) { - http.Error(w, "JSON only acceptable response", http.StatusNotAcceptable) - return - } - q := r.URL.Query() - _, checksum := q["checksum"] - _, containers := q["containers"] - req := pilosa.InspectRequest{ - HolderFilterParams: pilosa.HolderFilterParams{ - Indexes: q.Get("indexes"), - Fields: q.Get("fields"), - Views: q.Get("views"), - Shards: q.Get("shards"), - }, - InspectRequestParams: pilosa.InspectRequestParams{ - Checksum: checksum, - Containers: containers, - }, - } - info, err := h.api.Inspect(r.Context(), &req) - if err != nil { - http.Error(w, fmt.Sprintf("inspect request: %v", err), http.StatusBadRequest) - return - } - w.Header().Set("Content-Type", "application/json") - if err := json.NewEncoder(w).Encode(info); err != nil { - h.logger.Errorf("write inspect response error: %s", err) - } -} - type getSchemaResponse struct { Indexes []*pilosa.IndexInfo `json:"indexes"` } diff --git a/lattice b/lattice index 7871b74db..7ea3d77f8 160000 --- a/lattice +++ b/lattice @@ -1 +1 @@ -Subproject commit 7871b74dbe857cb034d3298e46de90c06faaca36 +Subproject commit 7ea3d77f89771a06cbe59867f9135436fc8ea3b6 diff --git a/roaring/filter.go b/roaring/filter.go index 61f9b3840..1f2018ad1 100644 --- a/roaring/filter.go +++ b/roaring/filter.go @@ -610,16 +610,23 @@ func (b *BitmapBitmapFilter) ConsiderData(key FilterKey, data *Container) Filter pos := key & keyMask base := uint64(key << 16) filter := b.containers[pos] - if filter == nil || !IntersectionAny(data, filter) { + if filter == nil { key.RejectUntilOffset(b.nextOffsets[pos]) } - matching := intersect(data, filter) - offsets := matching.Slice() - for _, v := range offsets { + var lastErr error + matched := false + intersectionCallback(data, filter, func(v uint16) { + matched = true err := b.callback(base + uint64(v)) if err != nil { - return key.Fail(err) + lastErr = err } + }) + if lastErr != nil { + return key.Fail(lastErr) + } + if !matched { + return key.RejectUntilOffset(b.nextOffsets[pos]) } return key.MatchOneUntilOffset(b.nextOffsets[pos]) } diff --git a/roaring/roaring.go b/roaring/roaring.go index bd2ee9d46..1cd579593 100644 --- a/roaring/roaring.go +++ b/roaring/roaring.go @@ -3032,6 +3032,58 @@ func BitmapCountRange(bitmap []uint64, start, end int32) int32 { return int32(n) } +func callbackBits(w uint64, base uint16, fn func(uint16)) { + bit := uint16(0) + for w != 0 { + trail := bits.TrailingZeros64(w) + bit += uint16(trail) + w >>= (trail + 1) + fn(base + bit) + } +} + +func bitmapCallbackRange(bitmap []uint64, start, end int32, fn func(uint16)) { + if roaringParanoia { + if start > end { + panic(fmt.Sprintf("counting in range but %v > %v", start, end)) + } + } + i, j := start/64, end/64 + // Special case when start and end fall in the same word. + if i == j { + offi, offj := uint(start%64), uint(64-end%64) + w := (bitmap[i] >> offi) << (offj + offi) + if w != 0 { + callbackBits(w, uint16(i)*64, fn) + } + } + + // Count partial starting word. + if off := uint(start) % 64; off != 0 { + w := (bitmap[i] >> off) << off + if w != 0 { + callbackBits(w, (uint16(i) * 64), fn) + } + i++ + } + + // Count words in between. + for ; i < j; i++ { + if bitmap[i] != 0 { + callbackBits(bitmap[i], uint16(i)*64, fn) + } + } + + // Count partial ending word. + if j < int32(len(bitmap)) { + off := 64 - (uint(end) % 64) + w := (bitmap[j] << off) >> off + if w != 0 { + callbackBits(w, uint16(j)*64, fn) + } + } +} + // RunCountRange returns the ranged bit count for RLE pairs. func RunCountRange(runs []Interval16, start, end int32) (n int32) { if roaringParanoia { @@ -4224,6 +4276,73 @@ func intersectionAnyBitmapBitmap(a, b *Container) bool { return false } +func containerCallback(a *Container, fn func(uint16)) { + if a.N() == 0 { + return + } + switch { + case a.isArray(): + values := a.array() + for _, v := range values { + fn(v) + } + case a.isBitmap(): + values := a.bitmap() + for i, w := range values { + if w == 0 { + continue + } + callbackBits(w, uint16(i)*64, fn) + } + case a.isRun(): + values := a.runs() + for _, r := range values { + for i := int(r.Start); i <= int(r.Last); i++ { + fn(uint16(i)) + } + } + } +} + +func intersectionCallback(a, b *Container, fn func(uint16)) { + if a.N() == MaxContainerVal+1 { + containerCallback(b, fn) + return + } + if b.N() == MaxContainerVal+1 { + containerCallback(a, fn) + return + } + if a.N() == 0 || b.N() == 0 { + return + } + if a.isArray() { + if b.isArray() { + intersectionCallbackArrayArray(a, b, fn) + } else if b.isRun() { + intersectionCallbackArrayRun(a, b, fn) + } else { + intersectionCallbackArrayBitmap(a, b, fn) + } + } else if a.isRun() { + if b.isArray() { + intersectionCallbackArrayRun(b, a, fn) + } else if b.isRun() { + intersectionCallbackRunRun(a, b, fn) + } else { + intersectionCallbackBitmapRun(b, a, fn) + } + } else { + if b.isArray() { + intersectionCallbackArrayBitmap(b, a, fn) + } else if b.isRun() { + intersectionCallbackBitmapRun(a, b, fn) + } else { + intersectionCallbackBitmapBitmap(a, b, fn) + } + } +} + func intersectionCount(a, b *Container) int32 { if a.N() == MaxContainerVal+1 { return b.N() @@ -4363,6 +4482,137 @@ func intersectionCountBitmapBitmap(a, b *Container) (n int32) { return int32(popcountAndSlice(a.bitmap(), b.bitmap())) } +func intersectionCallbackArrayArray(a, b *Container, fn func(uint16)) { + statsHit("intersectionCallback/ArrayArray") + ca, cb := a.array(), b.array() + na, nb := len(ca), len(cb) + if na > nb { + ca, cb = cb, ca + na, nb = nb, na // nolint: staticcheck, ineffassign + } + if (na << 2) < nb { + for _, va := range ca { + for cb[0] < va { + if len(cb) > 8 && cb[0] < va { + cb = cb[8:] + } + cb = cb[1:] + if len(cb) == 0 { + return + } + } + if cb[0] == va { + fn(va) + } + } + return + } + j := 0 + for _, va := range ca { + for cb[j] < va { + j++ + if j >= nb { + return + } + } + if cb[j] == va { + fn(va) + } + } +} + +func intersectionCallbackArrayRun(a, b *Container, fn func(uint16)) { + statsHit("intersectionCallback/ArrayRun") + array, runs := a.array(), b.runs() + na, nb := len(array), len(runs) + for i, j := 0, 0; i < na && j < nb; { + va, vb := array[i], runs[j] + if va < vb.Start { + i++ + } else if va >= vb.Start && va <= vb.Last { + i++ + fn(va) + } else if va > vb.Last { + j++ + } + } +} + +func intersectionCallbackRunRun(a, b *Container, fn func(uint16)) { + statsHit("intersectionCount/RunRun") + ra, rb := a.runs(), b.runs() + na, nb := len(ra), len(rb) + for i, j := 0, 0; i < na && j < nb; { + va, vb := ra[i], rb[j] + if va.Last < vb.Start { + // |--va--| |--vb--| + i++ + } else if va.Start > vb.Last { + // |--vb--| |--va--| + j++ + } else if va.Last > vb.Last && va.Start >= vb.Start { + // |--vb-|-|-va--| + for i := int(va.Start); i <= int(vb.Last); i++ { + fn(uint16(i)) + } + j++ + } else if va.Last > vb.Last && va.Start < vb.Start { + // |--va|--vb--|--| + for i := int(vb.Start); i <= int(vb.Last); i++ { + fn(uint16(i)) + } + j++ + } else if va.Last <= vb.Last && va.Start >= vb.Start { + // |--vb|--va--|--| + for i := int(va.Start); i <= int(va.Last); i++ { + fn(uint16(i)) + } + i++ + } else if va.Last <= vb.Last && va.Start < vb.Start { + // |--va-|-|-vb--| + for i := int(vb.Start); i <= int(va.Last); i++ { + fn(uint16(i)) + } + i++ + } + } +} + +func intersectionCallbackBitmapRun(a, b *Container, fn func(uint16)) { + statsHit("intersectionCount/BitmapRun") + for _, iv := range b.runs() { + bitmapCallbackRange(a.bitmap(), int32(iv.Start), int32(iv.Last)+1, fn) + } +} + +func intersectionCallbackArrayBitmap(a, b *Container, fn func(uint16)) (n int32) { + statsHit("intersectionCount/ArrayBitmap") + bitmap := b.bitmap() + ln := len(bitmap) + for _, val := range a.array() { + i := int(val >> 6) + if i >= ln { + break + } + off := val % 64 + n += int32(bitmap[i]>>off) & 1 + } + return n +} + +func intersectionCallbackBitmapBitmap(a, b *Container, fn func(uint16)) { + statsHit("intersectionCount/BitmapBitmap") + ab, bb := a.bitmap(), b.bitmap() + for i := range ab { + w := ab[i] & bb[i] + if w == 0 { + continue + } + base := uint16(i) * 64 + callbackBits(w, base, fn) + } +} + func intersect(a, b *Container) (c *Container) { if roaringParanoia { defer func() { c.CheckN() }()