runcontainer: count or test before allocating in run-bitmap intersections

runContainer16.andBitmapContainer materialized the run as an 8 KiB bitmap
container and then intersected, so an empty or array-sized result paid
for a bitmap it never returned. It now counts the result with the existing
andBitmapContainerCardinality and builds only that: an empty array
container when the intersection is empty, an array of the right size when
it is small, and the bitmap only when the result is one. A run with more
than 64 intervals is ORed into a scratch bitmap on the stack, counted
there, and extracted in one pass instead.

bitmapContainer.iandRun16 did the same conversion on the argument side. It
now counts the same way, returns a small result as an array built from
the run's intervals, and otherwise clears the gaps between intervals in
place; past 64 intervals the run is built in stack scratch and ANDed in
one flat pass, since clearing gap by gap costs a call per interval.

runContainer16.intersects computed rc.and(a) and tested the result for
emptiness, allocating a container to answer a boolean; every mixed
pairing with a run reached it through the array and bitmap dispatchers.
It now walks the run's intervals against the other container, a masked
word scan for bitmaps, a galloping merge for arrays or a probe per value
when the array is much smaller than the interval list, an interval
overlap test for runs, each returning at the first shared value.

BenchmarkRunAndBitmap (added): And, in-place And and Intersects between a
run-optimized bitmap and a dense one over a single key. Xeon 8375C:
                                        before                          after
  And, empty result                     3.0 µs   8328 B/op  4 allocs    594 ns   104 B/op  2 allocs
  And, array result                     8.3 µs  14504 B/op  8 allocs    4.6 µs  6280 B/op  6 allocs
  And, bitmap result                    4.6 µs  16560 B/op  8 allocs    3.6 µs  8336 B/op  6 allocs
  And, 1024-interval run, array        11.9 µs  12456 B/op  8 allocs   10.0 µs  4232 B/op  6 allocs
  And, 1024-interval run, bitmap        9.5 µs  16560 B/op  8 allocs    7.6 µs  8336 B/op  6 allocs
  in-place And, bitmap result           2.4 µs   8224 B/op  2 allocs    290 ns     0 B/op  0 allocs
  in-place And, 1024-interval run       7.3 µs   8224 B/op  2 allocs    4.6 µs     0 B/op  0 allocs
  Intersects, disjoint                  3.0 µs   8248 B/op  3 allocs    121 ns     0 B/op  0 allocs
  Intersects, overlapping               8.3 µs  14392 B/op  4 allocs      8 ns     0 B/op  0 allocs
  Intersects, few values, 1024 intervals 1.3 µs    32 B/op  2 allocs     62 ns     0 B/op  0 allocs

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: tamirms

bitmapcontainer.go

@@ -798,8 +798,26 @@ func (bc *bitmapContainer) iand(a container) container {
 }
 
 func (bc *bitmapContainer) iandRun16(rc *runContainer16) container {
-	rcb := newBitmapContainerFromRun(rc)
-	return bc.iandBitmap(rcb)
+	if len(rc.iv) > runAndScratchIntervals {
+		// Clearing gap by gap costs a call per interval; past this many the
+		// run is built in scratch and ANDed in one flat pass instead.
+		var scratch [bitmapContainerSize]uint64
+		for i := range rc.iv {
+			setBitmapRange(scratch[:], int(rc.iv[i].start), int(rc.iv[i].last())+1)
+		}
+		return bc.iandBitmap(&bitmapContainer{bitmap: scratch[:]})
+	}
+	card := rc.andBitmapContainerCardinality(bc)
+	if card <= arrayDefaultMaxSize {
+		answer := newArrayContainerCapacity(card)
+		for i := range rc.iv {
+			answer.content = appendBitmapRange(answer.content, bc.bitmap, int(rc.iv[i].start), int(rc.iv[i].last())+1)
+		}
+		return answer
+	}
+	clearBitmapGaps(bc.bitmap, rc.iv, 0, maxCapacity)
+	bc.cardinality = card
+	return bc
 }
 
 func (bc *bitmapContainer) iandArray(ac *arrayContainer) container {
@@ -831,6 +849,26 @@ func (bc *bitmapContainer) andArrayCardinality(value2 *arrayContainer) int {
 	return pos
 }
 
+func (bc *bitmapContainer) intersectsRange(start, end uint) bool {
+	if start >= end {
+		return false
+	}
+	firstword, endword := start/64, (end-1)/64
+	lo, hi := ^uint64(0)<<(start%64), ^uint64(0)>>((64-end)&63)
+	if firstword == endword {
+		return bc.bitmap[firstword]&lo&hi != 0
+	}
+	if bc.bitmap[firstword]&lo != 0 || bc.bitmap[endword]&hi != 0 {
+		return true
+	}
+	for _, w := range bc.bitmap[firstword+1 : endword] {
+		if w != 0 {
+			return true
+		}
+	}
+	return false
+}
+
 func (bc *bitmapContainer) getCardinalityInRange(start, end uint) int {
 	if start >= end {
 		return 0

runbitmap_and_test.go

@@ -0,0 +1,260 @@
+package roaring
+
+import (
+	"math/rand"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+// randomRun builds a canonical run container: up to n sorted intervals of
+// at most maxLen values separated by gaps of at most maxGap, optionally
+// touching 0 and 65535.
+func randomRun(rng *rand.Rand, n, maxLen, maxGap int, touchEdges bool) *runContainer16 {
+	rc := &runContainer16{}
+	pos := 0
+	if !touchEdges {
+		pos = rng.Intn(100)
+	}
+	for i := 0; i < n && pos < 65536; i++ {
+		length := min(1+rng.Intn(1+rng.Intn(maxLen)), 65536-pos)
+		rc.iv = append(rc.iv, interval16{start: uint16(pos), length: uint16(length - 1)})
+		pos += length + 1 + rng.Intn(maxGap)
+	}
+	if last := int(rc.maximum()); touchEdges && last != 65535 {
+		if start := 65535 - rng.Intn(500); last+2 <= start {
+			rc.iv = append(rc.iv, interval16{start: uint16(start), length: uint16(65535 - start)})
+		}
+	}
+	return rc
+}
+
+// randomBitmapContainer ANDs ands random words per position, so the density
+// is 1/2^ands.
+func randomBitmapContainer(rng *rand.Rand, ands int) *bitmapContainer {
+	bc := newBitmapContainer()
+	for i := range bc.bitmap {
+		w := rng.Uint64()
+		for k := 1; k < ands; k++ {
+			w &= rng.Uint64()
+		}
+		bc.bitmap[i] = w
+	}
+	bc.computeCardinality()
+	return bc
+}
+
+func requireAndResult(t *testing.T, want, got container) {
+	t.Helper()
+	require.True(t, want.equals(got), "want %d values, got %d", want.getCardinality(), got.getCardinality())
+	_, isArray := got.(*arrayContainer)
+	require.Equal(t, got.getCardinality() <= arrayDefaultMaxSize, isArray, "container kind must follow its cardinality")
+}
+
+// The reference is the old path: materialize the run, then the unchanged
+// bitmap-bitmap intersection.
+func TestRunAndBitmapContainer(t *testing.T) {
+	rng := rand.New(rand.NewSource(1))
+	for trial := 0; trial < 300; trial++ {
+		rc := randomRun(rng, 1+rng.Intn(40), 3000, 2000, trial%5 == 0)
+		bc := randomBitmapContainer(rng, 1+trial%4)
+		want := newBitmapContainerFromRun(rc).andBitmap(bc)
+		requireAndResult(t, want, rc.and(bc))
+		requireAndResult(t, want, bc.and(rc))
+		requireAndResult(t, want, bc.clone().iand(rc))
+	}
+}
+
+// Runs with more than runAndScratchIntervals intervals take the scratch
+// path; check it on regular and random runs against every density.
+func TestRunAndBitmapContainerManyIntervals(t *testing.T) {
+	rng := rand.New(rand.NewSource(21))
+	regular := func(stride, length int) *runContainer16 {
+		rc := &runContainer16{}
+		for start := 0; start+length <= 65536; start += stride {
+			rc.iv = append(rc.iv, interval16{start: uint16(start), length: uint16(length - 1)})
+		}
+		return rc
+	}
+	runs := []*runContainer16{regular(32, 8), regular(64, 1), regular(3, 2), regular(200, 100)}
+	for i := 0; i < 60; i++ {
+		runs = append(runs, randomRun(rng, 65+rng.Intn(400), 60, 120, false))
+	}
+	for i, rc := range runs {
+		require.Greater(t, len(rc.iv), runAndScratchIntervals, "run %d", i)
+		for _, bc := range []*bitmapContainer{randomBitmapContainer(rng, 6), randomBitmapContainer(rng, 2), randomBitmapContainer(rng, 1), newBitmapContainerwithRange(0, 65535)} {
+			want := newBitmapContainerFromRun(rc).andBitmap(bc)
+			requireAndResult(t, want, rc.and(bc))
+			requireAndResult(t, want, bc.and(rc))
+			requireAndResult(t, want, bc.clone().iand(rc))
+			require.Equal(t, !want.isEmpty(), rc.intersects(bc), "run %d", i)
+		}
+	}
+}
+
+func TestRunAndBitmapContainerEdges(t *testing.T) {
+	empty := newBitmapContainer()
+	full := newBitmapContainerwithRange(0, 65535)
+	single := &runContainer16{iv: []interval16{{start: 100, length: 9}}}
+	edges := &runContainer16{iv: []interval16{{start: 0, length: 0}, {start: 65535, length: 0}}}
+	atThreshold := &runContainer16{iv: []interval16{{start: 0, length: arrayDefaultMaxSize - 1}}}
+	pastThreshold := &runContainer16{iv: []interval16{{start: 0, length: arrayDefaultMaxSize}}}
+	for _, rc := range []*runContainer16{single, edges, atThreshold, pastThreshold} {
+		for _, bc := range []*bitmapContainer{empty, full} {
+			want := newBitmapContainerFromRun(rc).andBitmap(bc)
+			requireAndResult(t, want, rc.and(bc))
+			requireAndResult(t, want, bc.clone().iand(rc))
+		}
+	}
+}
+
+// An empty intersection allocates only its empty array container; the old
+// path allocated an 8 KiB bitmap to hold the run first.
+func TestRunAndBitmapContainerEmptyAllocation(t *testing.T) {
+	rc := &runContainer16{iv: []interval16{{start: 0, length: 999}, {start: 30000, length: 999}}}
+	bc := newBitmapContainerwithRange(2000, 20000)
+	require.LessOrEqual(t, testing.AllocsPerRun(100, func() {
+		if !rc.and(bc).isEmpty() {
+			t.Fatal("expected empty")
+		}
+	}), 1.0)
+	// The clone costs two allocations, the empty result one.
+	require.LessOrEqual(t, testing.AllocsPerRun(100, func() {
+		_ = bc.clone().iand(rc)
+	}), 3.0)
+}
+
+// The reference is the old definition, !rc.and(c).isEmpty().
+func TestRunContainerIntersects(t *testing.T) {
+	rng := rand.New(rand.NewSource(11))
+	for trial := 0; trial < 400; trial++ {
+		rc := randomRun(rng, 1+rng.Intn(30), 3000, 2000, trial%7 == 0)
+		ac := newArrayContainer()
+		for i := 0; i < rng.Intn(300); i++ {
+			ac.iadd(uint16(rng.Intn(65536)))
+		}
+		others := []container{randomBitmapContainer(rng, 1+trial%3*3), ac, randomRun(rng, 1+rng.Intn(30), 3000, 2000, trial%5 == 0)}
+		for _, c := range others {
+			want := !rc.and(c).isEmpty()
+			require.Equal(t, want, rc.intersects(c), "trial %d run vs %T", trial, c)
+			require.Equal(t, want, c.intersects(rc), "trial %d %T vs run", trial, c)
+		}
+	}
+	// A few values against many intervals take the search branch; the
+	// reference is the same definition.
+	many := randomRun(rng, 300, 60, 120, false)
+	require.Greater(t, len(many.iv), 200)
+	for trial := 0; trial < 200; trial++ {
+		few := newArrayContainer()
+		for i := 0; i < 1+rng.Intn(6); i++ {
+			few.iadd(uint16(rng.Intn(65536)))
+		}
+		want := !many.and(few).isEmpty()
+		require.Equal(t, want, many.intersects(few), "trial %d few vs many", trial)
+		require.Equal(t, want, few.intersects(many), "trial %d many vs few", trial)
+	}
+	// Edges: empty array, touching intervals, single values at both ends.
+	empty := newArrayContainer()
+	edges := &runContainer16{iv: []interval16{{start: 0, length: 0}, {start: 65535, length: 0}}}
+	require.False(t, edges.intersects(empty))
+	one := newArrayContainer()
+	one.iadd(65535)
+	require.True(t, edges.intersects(one))
+	touching := &runContainer16{iv: []interval16{{start: 1, length: 65533}}}
+	require.False(t, edges.intersects(touching))
+	require.True(t, edges.intersects(newBitmapContainerwithRange(0, 0)))
+}
+
+func TestRunContainerIntersectsAllocatesNothing(t *testing.T) {
+	rc := &runContainer16{iv: []interval16{{start: 0, length: 999}, {start: 30000, length: 999}}}
+	bc := newBitmapContainerwithRange(2000, 20000)
+	ac := newArrayContainerRange(2000, 3000)
+	other := &runContainer16{iv: []interval16{{start: 5000, length: 100}}}
+	for _, c := range []container{bc, ac, other} {
+		require.Zero(t, testing.AllocsPerRun(100, func() { _ = rc.intersects(c) }), "%T", c)
+	}
+}
+
+// runBitmap holds one run container at key 0 covering n intervals of the
+// given length, spaced stride apart.
+func runBitmap(n, length, stride int) *Bitmap {
+	bm := New()
+	for i := 0; i < n; i++ {
+		bm.AddRange(uint64(i*stride), uint64(i*stride+length))
+	}
+	bm.RunOptimize()
+	return bm
+}
+
+// BenchmarkRunAndBitmap intersects a run-optimized bitmap with a dense one
+// over a single key through the public API, one case per result kind, plus
+// the in-place form and the Intersects test.
+func BenchmarkRunAndBitmap(b *testing.B) {
+	rng := rand.New(rand.NewSource(7))
+	short := runBitmap(20, 300, 3000)  // 6000 values
+	long := runBitmap(4, 10000, 16000) // 40000 values
+	many := runBitmap(1024, 8, 64)     // 8192 values in 1024 intervals
+	gaps, quarter, half, dense := New(), New(), New(), New()
+	for v := uint32(0); v < 65536; v++ {
+		if !short.Contains(v) && rng.Intn(2) == 0 {
+			gaps.Add(v)
+		}
+		if rng.Intn(4) == 0 {
+			quarter.Add(v)
+		}
+		if rng.Intn(2) == 0 {
+			half.Add(v)
+		}
+		if rng.Intn(10) != 0 {
+			dense.Add(v)
+		}
+	}
+	for _, tc := range []struct {
+		name string
+		x, y *Bitmap
+	}{
+		{"and/empty", short, gaps},
+		{"and/array", short, half},
+		{"and/bitmap", long, dense},
+		{"and/many-intervals/array", many, quarter},
+		{"and/many-intervals/bitmap", many, dense},
+	} {
+		b.Run(tc.name, func(b *testing.B) {
+			b.ReportAllocs()
+			for b.Loop() {
+				_ = And(tc.x, tc.y)
+			}
+		})
+	}
+	for _, tc := range []struct {
+		name string
+		x    *Bitmap
+	}{
+		{"iand/bitmap", long},
+		{"iand/many-intervals", many},
+	} {
+		b.Run(tc.name, func(b *testing.B) {
+			b.ReportAllocs()
+			x := dense.Clone()
+			for b.Loop() {
+				x.And(tc.x)
+			}
+		})
+	}
+	few := New()
+	few.AddMany([]uint32{100, 30000, 60000})
+	for _, tc := range []struct {
+		name string
+		x, y *Bitmap
+	}{
+		{"intersects/disjoint", short, gaps},
+		{"intersects/overlapping", short, half},
+		{"intersects/few-values-many-intervals", many, few},
+	} {
+		b.Run(tc.name, func(b *testing.B) {
+			for b.Loop() {
+				_ = tc.x.Intersects(tc.y)
+			}
+		})
+	}
+}