Merge branch 'master' into available-shards

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)

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")

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()
 			}
 		}
 	}

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"`
 }

lattice

@@ -1 +1 @@
-Subproject commit 7871b74dbe857cb034d3298e46de90c06faaca36
+Subproject commit 7ea3d77f89771a06cbe59867f9135436fc8ea3b6

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])
 }

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() }()