optimize intersection count

This commit moves the computation of the intersection count to
the `roaring` package so that no allocations are required. The
implementation operates at the roaring container level and has
specialized functions for array-array, array-bitmap, and
bitmap-bitmap container pairs.

bitmap.go

@@ -38,27 +38,7 @@ func (b *Bitmap) Merge(other *Bitmap) {
 
 // IntersectionCount returns the number of intersections between b and other.
 func (b *Bitmap) IntersectionCount(other *Bitmap) uint64 {
-	// OPTIMIZE: Implement roaring.Bitmap.IntersectionCount()
-
-	itr0 := roaring.NewBufIterator(b.data.Iterator())
-	itr1 := roaring.NewBufIterator(other.data.Iterator())
-
-	var n uint64
-	for {
-		v0, eof0 := itr0.Next()
-		v1, eof1 := itr1.Next()
-
-		if eof0 || eof1 {
-			break
-		} else if v0 < v1 {
-			itr1.Unread()
-		} else if v0 > v1 {
-			itr0.Unread()
-		} else {
-			n++
-		}
-	}
-	return n
+	return b.data.IntersectionCount(&other.data)
 }
 
 // Intersect returns the itersection of b and other.

popcnt.go

@@ -1,53 +0,0 @@
-package pilosa
-
-// bit population count, take from
-// https://code.google.com/p/go/issues/detail?id=4988#c11
-// credit: https://code.google.com/u/arnehormann/
-func popcntGo(x uint64) (n uint64) {
-	x -= (x >> 1) & 0x5555555555555555
-	x = (x>>2)&0x3333333333333333 + x&0x3333333333333333
-	x += x >> 4
-	x &= 0x0f0f0f0f0f0f0f0f
-	x *= 0x0101010101010101
-	return x >> 56
-}
-
-func popcntSliceGo(s []uint64) uint64 {
-	cnt := uint64(0)
-	for _, x := range s {
-		cnt += popcntGo(x)
-	}
-	return cnt
-}
-
-func popcntMaskSliceGo(s, m []uint64) uint64 {
-	cnt := uint64(0)
-	for i := range s {
-		cnt += popcntGo(s[i] &^ m[i])
-	}
-	return cnt
-}
-
-func popcntAndSliceGo(s, m []uint64) uint64 {
-	cnt := uint64(0)
-	for i := range s {
-		cnt += popcntGo(s[i] & m[i])
-	}
-	return cnt
-}
-
-func popcntOrSliceGo(s, m []uint64) uint64 {
-	cnt := uint64(0)
-	for i := range s {
-		cnt += popcntGo(s[i] | m[i])
-	}
-	return cnt
-}
-
-func popcntXorSliceGo(s, m []uint64) uint64 {
-	cnt := uint64(0)
-	for i := range s {
-		cnt += popcntGo(s[i] ^ m[i])
-	}
-	return cnt
-}

popcnt_amd64.s

@@ -1,109 +0,0 @@
-TEXT ·hasAsm(SB),4,$0
-MOVQ $1, AX
-CPUID
-SHRQ $23, CX
-ANDQ $1, CX
-MOVB CX, ret+0(FP)
-RET
-
-
-#define POPCNTQ_DX_DX BYTE $0xf3; BYTE $0x48; BYTE $0x0f; BYTE $0xb8; BYTE $0xd2
-
-TEXT ·popcntSliceAsm(SB),4,$0-32
-XORQ	AX, AX
-MOVQ	s+0(FP), SI
-MOVQ	s+8(FP), CX
-TESTQ	CX, CX
-JZ		popcntSliceEnd
-popcntSliceLoop:
-BYTE $0xf3; BYTE $0x48; BYTE $0x0f; BYTE $0xb8; BYTE $0x16 // POPCNTQ (SI), DX
-ADDQ	DX, AX
-ADDQ	$8, SI
-LOOP	popcntSliceLoop
-popcntSliceEnd:
-MOVQ	AX, ret+24(FP)
-RET
-
-TEXT ·popcntMaskSliceAsm(SB),4,$0-56
-XORQ	AX, AX
-MOVQ	s+0(FP), SI
-MOVQ	s+8(FP), CX
-TESTQ	CX, CX
-JZ		popcntMaskSliceEnd
-MOVQ	m+24(FP), DI
-popcntMaskSliceLoop:
-MOVQ	(DI), DX
-NOTQ	DX
-ANDQ	(SI), DX
-POPCNTQ_DX_DX
-ADDQ	DX, AX
-ADDQ	$8, SI
-ADDQ	$8, DI
-LOOP	popcntMaskSliceLoop
-popcntMaskSliceEnd:
-MOVQ	AX, ret+48(FP)
-RET
-
-TEXT ·popcntAndSliceAsm(SB),4,$0-56
-XORQ	AX, AX
-MOVQ	s+0(FP), SI
-MOVQ	s+8(FP), CX
-TESTQ	CX, CX
-JZ		popcntAndSliceEnd
-MOVQ	m+24(FP), DI
-popcntAndSliceLoop:
-MOVQ	(DI), DX
-ANDQ	(SI), DX
-POPCNTQ_DX_DX
-ADDQ	DX, AX
-ADDQ	$8, SI
-ADDQ	$8, DI
-LOOP	popcntAndSliceLoop
-popcntAndSliceEnd:
-MOVQ	AX, ret+48(FP)
-RET
-
-TEXT ·popcntOrSliceAsm(SB),4,$0-56
-XORQ	AX, AX
-MOVQ	s+0(FP), SI
-MOVQ	s+8(FP), CX
-TESTQ	CX, CX
-JZ		popcntOrSliceEnd
-MOVQ	m+24(FP), DI
-popcntOrSliceLoop:
-MOVQ	(DI), DX
-ORQ		(SI), DX
-POPCNTQ_DX_DX
-ADDQ	DX, AX
-ADDQ	$8, SI
-ADDQ	$8, DI
-LOOP	popcntOrSliceLoop
-popcntOrSliceEnd:
-MOVQ	AX, ret+48(FP)
-RET
-
-TEXT ·popcntXorSliceAsm(SB),4,$0-56
-XORQ	AX, AX
-MOVQ	s+0(FP), SI
-MOVQ	s+8(FP), CX
-TESTQ	CX, CX
-JZ		popcntXorSliceEnd
-MOVQ	m+24(FP), DI
-popcntXorSliceLoop:
-MOVQ	(DI), DX
-XORQ	(SI), DX
-POPCNTQ_DX_DX
-ADDQ	DX, AX
-ADDQ	$8, SI
-ADDQ	$8, DI
-LOOP	popcntXorSliceLoop
-popcntXorSliceEnd:
-MOVQ	AX, ret+48(FP)
-RET
-
-
-TEXT ·popcntAsm(SB),4,$0-16
-        MOVQ  x+0(FP), DX
-        POPCNTQ_DX_DX
-        MOVQ DX, ret+8(FP)
-        RET

popcnt_test.go

@@ -1,75 +0,0 @@
-package pilosa
-
-import "testing"
-
-func BenchmarkPopcntAsm(b *testing.B) {
-	// run the Fib function b.N times
-	for n := 0; n < b.N; n++ {
-		popcntAsm(0xdeadbeef)
-	}
-}
-
-func BenchmarkPopcntGo(b *testing.B) {
-	// run the Fib function b.N times
-	for n := 0; n < b.N; n++ {
-		popcntGo(0xdeadbeef)
-	}
-}
-
-func getData() []uint64 {
-	return []uint64{
-		0xdeadbeef,
-		0xdeadbeef,
-		0xdeadbeef,
-		0xdeadbeef,
-		0xdeadbeef,
-		0xdeadbeef,
-		0xdeadbeef,
-		0xdeadbeef,
-		0xdeadbeef,
-		0xdeadbeef,
-		0xdeadbeef,
-		0xdeadbeef,
-		0xdeadbeef,
-		0xdeadbeef,
-		0xdeadbeef,
-		0xdeadbeef,
-		0xdeadbeef,
-		0xdeadbeef,
-		0xdeadbeef,
-		0xdeadbeef,
-		0xdeadbeef,
-		0xdeadbeef,
-		0xdeadbeef,
-		0xdeadbeef,
-		0xdeadbeef,
-		0xdeadbeef,
-		0xdeadbeef,
-		0xdeadbeef,
-		0xdeadbeef,
-		0xdeadbeef,
-		0xdeadbeef,
-		0xdeadbeef,
-	}
-}
-
-func BenchmarkPopcntSliceGo(b *testing.B) {
-	d := getData()
-	for n := 0; n < b.N; n++ {
-		popcntSliceGo(d)
-	}
-}
-
-func BenchmarkPopcntSliceAsm(b *testing.B) {
-	d := getData()
-	for n := 0; n < b.N; n++ {
-		popcntSliceAsm(d)
-	}
-}
-
-func BenchmarkPopcntSlice(b *testing.B) {
-	d := getData()
-	for n := 0; n < b.N; n++ {
-		popcntSlice(d)
-	}
-}

roaring/assembly.go

@@ -5,3 +5,55 @@ func hasAsm() bool
 func BSFQ(memory uint64) int
 
 func POPCNTQ(memory uint64) int
+
+// bit population count, take from
+// https://code.google.com/p/go/issues/detail?id=4988#c11
+// credit: https://code.google.com/u/arnehormann/
+func popcntGo(x uint64) (n uint64) {
+	x -= (x >> 1) & 0x5555555555555555
+	x = (x>>2)&0x3333333333333333 + x&0x3333333333333333
+	x += x >> 4
+	x &= 0x0f0f0f0f0f0f0f0f
+	x *= 0x0101010101010101
+	return x >> 56
+}
+
+func popcntSliceGo(s []uint64) uint64 {
+	cnt := uint64(0)
+	for _, x := range s {
+		cnt += popcntGo(x)
+	}
+	return cnt
+}
+
+func popcntMaskSliceGo(s, m []uint64) uint64 {
+	cnt := uint64(0)
+	for i := range s {
+		cnt += popcntGo(s[i] &^ m[i])
+	}
+	return cnt
+}
+
+func popcntAndSliceGo(s, m []uint64) uint64 {
+	cnt := uint64(0)
+	for i := range s {
+		cnt += popcntGo(s[i] & m[i])
+	}
+	return cnt
+}
+
+func popcntOrSliceGo(s, m []uint64) uint64 {
+	cnt := uint64(0)
+	for i := range s {
+		cnt += popcntGo(s[i] | m[i])
+	}
+	return cnt
+}
+
+func popcntXorSliceGo(s, m []uint64) uint64 {
+	cnt := uint64(0)
+	for i := range s {
+		cnt += popcntGo(s[i] ^ m[i])
+	}
+	return cnt
+}

roaring/assembly_amd64.s

@@ -19,3 +19,104 @@ TEXT ·BSFQ(SB),NOSPLIT,$0-8
 	BSFQ BP, BX
 	MOVQ BX, ret+8(FP)
 	RET
+
+#define POPCNTQ_DX_DX BYTE $0xf3; BYTE $0x48; BYTE $0x0f; BYTE $0xb8; BYTE $0xd2
+
+TEXT ·popcntSliceAsm(SB),4,$0-32
+XORQ	AX, AX
+MOVQ	s+0(FP), SI
+MOVQ	s+8(FP), CX
+TESTQ	CX, CX
+JZ		popcntSliceEnd
+popcntSliceLoop:
+BYTE $0xf3; BYTE $0x48; BYTE $0x0f; BYTE $0xb8; BYTE $0x16 // POPCNTQ (SI), DX
+ADDQ	DX, AX
+ADDQ	$8, SI
+LOOP	popcntSliceLoop
+popcntSliceEnd:
+MOVQ	AX, ret+24(FP)
+RET
+
+TEXT ·popcntMaskSliceAsm(SB),4,$0-56
+XORQ	AX, AX
+MOVQ	s+0(FP), SI
+MOVQ	s+8(FP), CX
+TESTQ	CX, CX
+JZ		popcntMaskSliceEnd
+MOVQ	m+24(FP), DI
+popcntMaskSliceLoop:
+MOVQ	(DI), DX
+NOTQ	DX
+ANDQ	(SI), DX
+POPCNTQ_DX_DX
+ADDQ	DX, AX
+ADDQ	$8, SI
+ADDQ	$8, DI
+LOOP	popcntMaskSliceLoop
+popcntMaskSliceEnd:
+MOVQ	AX, ret+48(FP)
+RET
+
+TEXT ·popcntAndSliceAsm(SB),4,$0-56
+XORQ	AX, AX
+MOVQ	s+0(FP), SI
+MOVQ	s+8(FP), CX
+TESTQ	CX, CX
+JZ		popcntAndSliceEnd
+MOVQ	m+24(FP), DI
+popcntAndSliceLoop:
+MOVQ	(DI), DX
+ANDQ	(SI), DX
+POPCNTQ_DX_DX
+ADDQ	DX, AX
+ADDQ	$8, SI
+ADDQ	$8, DI
+LOOP	popcntAndSliceLoop
+popcntAndSliceEnd:
+MOVQ	AX, ret+48(FP)
+RET
+
+TEXT ·popcntOrSliceAsm(SB),4,$0-56
+XORQ	AX, AX
+MOVQ	s+0(FP), SI
+MOVQ	s+8(FP), CX
+TESTQ	CX, CX
+JZ		popcntOrSliceEnd
+MOVQ	m+24(FP), DI
+popcntOrSliceLoop:
+MOVQ	(DI), DX
+ORQ		(SI), DX
+POPCNTQ_DX_DX
+ADDQ	DX, AX
+ADDQ	$8, SI
+ADDQ	$8, DI
+LOOP	popcntOrSliceLoop
+popcntOrSliceEnd:
+MOVQ	AX, ret+48(FP)
+RET
+
+TEXT ·popcntXorSliceAsm(SB),4,$0-56
+XORQ	AX, AX
+MOVQ	s+0(FP), SI
+MOVQ	s+8(FP), CX
+TESTQ	CX, CX
+JZ		popcntXorSliceEnd
+MOVQ	m+24(FP), DI
+popcntXorSliceLoop:
+MOVQ	(DI), DX
+XORQ	(SI), DX
+POPCNTQ_DX_DX
+ADDQ	DX, AX
+ADDQ	$8, SI
+ADDQ	$8, DI
+LOOP	popcntXorSliceLoop
+popcntXorSliceEnd:
+MOVQ	AX, ret+48(FP)
+RET
+
+
+TEXT ·popcntAsm(SB),4,$0-16
+        MOVQ  x+0(FP), DX
+        POPCNTQ_DX_DX
+        MOVQ DX, ret+8(FP)
+        RET

roaring/assembly_asm.go

@@ -1,11 +1,9 @@
 // +build amd64
 
-package pilosa
+package roaring
 
 //go:noescape
 
-func hasAsm() bool
-
 var useAsm = hasAsm()
 
 //go:noescape

roaring/assembly_generic.go

@@ -1,6 +1,6 @@
 // +build !amd64
 
-package pilosa
+package roaring
 
 func popcntSlice(s []uint64) uint64        { return popcntSliceGo(s) }
 func popcntMaskSlice(s, m []uint64) uint64 { return popcntMaskSliceGo(s, m) }

roaring/assembly_test.go

@@ -51,3 +51,75 @@ func BenchmarkPopcount(b *testing.B) {
 		popcount(uint64(i))
 	}
 }
+
+func BenchmarkPopcntAsm(b *testing.B) {
+	// run the Fib function b.N times
+	for n := 0; n < b.N; n++ {
+		popcntAsm(0xdeadbeef)
+	}
+}
+
+func BenchmarkPopcntGo(b *testing.B) {
+	// run the Fib function b.N times
+	for n := 0; n < b.N; n++ {
+		popcntGo(0xdeadbeef)
+	}
+}
+
+func getData() []uint64 {
+	return []uint64{
+		0xdeadbeef,
+		0xdeadbeef,
+		0xdeadbeef,
+		0xdeadbeef,
+		0xdeadbeef,
+		0xdeadbeef,
+		0xdeadbeef,
+		0xdeadbeef,
+		0xdeadbeef,
+		0xdeadbeef,
+		0xdeadbeef,
+		0xdeadbeef,
+		0xdeadbeef,
+		0xdeadbeef,
+		0xdeadbeef,
+		0xdeadbeef,
+		0xdeadbeef,
+		0xdeadbeef,
+		0xdeadbeef,
+		0xdeadbeef,
+		0xdeadbeef,
+		0xdeadbeef,
+		0xdeadbeef,
+		0xdeadbeef,
+		0xdeadbeef,
+		0xdeadbeef,
+		0xdeadbeef,
+		0xdeadbeef,
+		0xdeadbeef,
+		0xdeadbeef,
+		0xdeadbeef,
+		0xdeadbeef,
+	}
+}
+
+func BenchmarkPopcntSliceGo(b *testing.B) {
+	d := getData()
+	for n := 0; n < b.N; n++ {
+		popcntSliceGo(d)
+	}
+}
+
+func BenchmarkPopcntSliceAsm(b *testing.B) {
+	d := getData()
+	for n := 0; n < b.N; n++ {
+		popcntSliceAsm(d)
+	}
+}
+
+func BenchmarkPopcntSlice(b *testing.B) {
+	d := getData()
+	for n := 0; n < b.N; n++ {
+		popcntSlice(d)
+	}
+}

roaring/internal_test.go

@@ -0,0 +1,43 @@
+package roaring
+
+import (
+	"reflect"
+	"testing"
+)
+
+// Ensure iterator returns values from a bitmap.
+func TestBitmapIterator(t *testing.T) {
+	for i, tt := range []struct {
+		bitmap []uint64
+		values []uint16
+	}{
+		// Empty
+		{
+			bitmap: []uint64{6}, // 0110
+			values: []uint16{1, 2},
+		},
+
+		// Single uint64 bitmap
+		{
+			bitmap: []uint64{6}, // 0110
+			values: []uint16{1, 2},
+		},
+
+		// Multi uint64 bitmap
+		{
+			bitmap: []uint64{1 << 63, 1, 0, 1, 3 << 62},
+			values: []uint16{63, 64, 192, 318, 319},
+		},
+	} {
+		itr := newBitmapIterator(tt.bitmap)
+
+		var a []uint16
+		for v, eof := itr.next(); !eof; v, eof = itr.next() {
+			a = append(a, v)
+		}
+
+		if !reflect.DeepEqual(a, tt.values) {
+			t.Errorf("%d. unexpected values: exp=%+v, got=%+v", i, a, tt.values)
+		}
+	}
+}

roaring/roaring.go

@@ -185,6 +185,23 @@ func (b *Bitmap) insertAt(key uint64, c *container, i int) {
 	b.containers[i] = c
 }
 
+// IntersectionCount returns the number of intersections between b and other.
+func (b *Bitmap) IntersectionCount(other *Bitmap) uint64 {
+	var n uint64
+	for i, j := 0, 0; i < len(b.containers) && i < len(other.containers); {
+		ki, kj := b.keys[i], other.keys[j]
+		if ki < kj {
+			i++
+		} else if ki > kj {
+			j++
+		} else {
+			n += intersectionCount(b.containers[i], other.containers[j])
+			i, j = i+1, j+1
+		}
+	}
+	return n
+}
+
 // WriteTo writes b to w.
 func (b *Bitmap) WriteTo(w io.Writer) (n int64, err error) {
 	// Build header before writing individual container blocks.
@@ -693,6 +710,64 @@ func (c *container) size() int {
 	return len(c.bitmap) * 8
 }
 
+func intersectionCount(a, b *container) uint64 {
+	if a.isArray() {
+		if b.isArray() {
+			return intersectionCountArrayArray(a, b)
+		} else {
+			return intersectionCountArrayBitmap(a, b)
+		}
+	} else {
+		if b.isArray() {
+			return intersectionCountArrayBitmap(b, a)
+		} else {
+			return intersectionCountBitmapBitmap(a, b)
+		}
+	}
+}
+
+func intersectionCountArrayArray(a, b *container) (n uint64) {
+	na, nb := len(a.array), len(b.array)
+	for i, j := 0, 0; i < na && j < nb; {
+		va, vb := a.array[i], b.array[j]
+		if va < vb {
+			i++
+		} else if va > vb {
+			j++
+		} else {
+			n++
+			i, j = i+1, j+1
+		}
+	}
+	return n
+}
+
+func intersectionCountArrayBitmap(a, b *container) (n uint64) {
+	itr := newBufIterator(newBitmapIterator(b.bitmap))
+	for i := 0; i < len(a.array); {
+		va := a.array[i]
+		vb, eof := itr.next()
+		if eof {
+			break
+		}
+
+		if va < vb {
+			i++
+			itr.unread()
+		} else if va > vb {
+			// nop
+		} else {
+			n++
+			i++
+		}
+	}
+	return n
+}
+
+func intersectionCountBitmapBitmap(a, b *container) (n uint64) {
+	return popcntAndSliceGo(a.bitmap, b.bitmap)
+}
+
 // opType represents a type of operation.
 type opType uint8
 
@@ -871,3 +946,79 @@ func popcount(x uint64) (n uint64) {
 	x *= 0x0101010101010101
 	return x >> 56
 }
+
+// bitmapIterator represents an iterator over container bitmap values.
+type bitmapIterator struct {
+	bitmap []uint64
+	i      int
+}
+
+func newBitmapIterator(bitmap []uint64) *bitmapIterator {
+	return &bitmapIterator{
+		bitmap: bitmap,
+		i:      -1,
+	}
+}
+
+// next returns the next value in the bitmap.
+// Returns eof as true if there are no values left in the iterator.
+func (itr *bitmapIterator) next() (v uint16, eof bool) {
+	if itr.i+1 >= len(itr.bitmap)*64 {
+		return 0, true
+	}
+	itr.i++
+
+	// Find first non-zero bit in current bitmap, if possible.
+	hb := int(itr.i / 64)
+	lb := itr.bitmap[hb] >> (uint(itr.i) % 64)
+	if lb != 0 {
+		itr.i = int(itr.i) + trailingZeroN(lb)
+		return uint16(itr.i), false
+	}
+
+	// Otherwise iterate through remaining bitmaps to find next bit.
+	for hb++; hb < len(itr.bitmap); hb++ {
+		if itr.bitmap[hb] != 0 {
+			itr.i = int(hb*64) + trailingZeroN(itr.bitmap[hb])
+			return uint16(itr.i), false
+		}
+	}
+
+	return 0, true
+}
+
+// bufBitmapIterator wraps an iterator to provide the ability to unread values.
+type bufBitmapIterator struct {
+	buf struct {
+		v    uint16
+		eof  bool
+		full bool
+	}
+	itr *bitmapIterator
+}
+
+// newBufBitmapIterator returns a buffered iterator that wraps a bitmapIterator.
+func newBufIterator(itr *bitmapIterator) *bufBitmapIterator {
+	return &bufBitmapIterator{itr: itr}
+}
+
+// next returns the next pair in the bitmap.
+// If a value has been buffered then it is returned and the buffer is cleared.
+func (itr *bufBitmapIterator) next() (v uint16, eof bool) {
+	if itr.buf.full {
+		itr.buf.full = false
+		return itr.buf.v, itr.buf.eof
+	}
+
+	// Read value onto buffer in case of unread.
+	itr.buf.v, itr.buf.eof = itr.itr.next()
+	return itr.buf.v, itr.buf.eof
+}
+
+// unread pushes previous pair on to the buffer. Panics if the buffer is already full.
+func (itr *bufBitmapIterator) unread() {
+	if itr.buf.full {
+		panic("roaring.bufBitmapIterator: buffer full")
+	}
+	itr.buf.full = true
+}

roaring/roaring_test.go

@@ -80,6 +80,55 @@ func TestBitmap_Max(t *testing.T) {
 	}
 }
 
+// Ensure bitmap can return the number of intersecting bits in two bitmaps.
+func TestBitmap_IntersectionCount_ArrayArray(t *testing.T) {
+	bm0 := roaring.NewBitmap(0, 1000001, 1000002, 1000003)
+	bm1 := roaring.NewBitmap(0, 50000, 1000001, 1000002)
+
+	if n := bm0.IntersectionCount(bm1); n != 3 {
+		t.Fatalf("unexpected n: %d", n)
+	} else if n := bm1.IntersectionCount(bm0); n != 3 {
+		t.Fatalf("unexpected n (reverse): %d", n)
+	}
+}
+
+// Ensure bitmap can return the number of intersecting bits in two bitmaps.
+func TestBitmap_IntersectionCount_ArrayBitmap(t *testing.T) {
+	bm0 := roaring.NewBitmap(1, 70, 200, 4097, 4098)
+	bm1 := roaring.NewBitmap()
+	for i := uint64(0); i <= 10000; i += 2 {
+		bm1.Add(i)
+	}
+
+	if n := bm0.IntersectionCount(bm1); n != 3 {
+		t.Fatalf("unexpected n: %d", n)
+	} else if n := bm1.IntersectionCount(bm0); n != 3 {
+		t.Fatalf("unexpected n (reverse): %d", n)
+	}
+}
+
+// Ensure bitmap can return the number of intersecting bits in two bitmaps.
+func TestBitmap_IntersectionCount_BitmapBitmap(t *testing.T) {
+	bm0 := roaring.NewBitmap()
+	bm1 := roaring.NewBitmap()
+	for i := uint64(0); i <= 10000; i += 2 {
+		bm0.Add(i)
+		bm1.Add(i + 1)
+	}
+
+	bm0.Add(1000)
+	bm1.Add(1000)
+
+	bm0.Add(2000)
+	bm1.Add(2000)
+
+	if n := bm0.IntersectionCount(bm1); n != 2 {
+		t.Fatalf("unexpected n: %d", n)
+	} else if n := bm1.IntersectionCount(bm0); n != 2 {
+		t.Fatalf("unexpected n (reverse): %d", n)
+	}
+}
+
 func TestBitmap_Quick_Array1(t *testing.T)     { testBitmapQuick(t, 1000, 1000, 2000) }
 func TestBitmap_Quick_Array2(t *testing.T)     { testBitmapQuick(t, 10000, 0, 1000) }
 func TestBitmap_Quick_Bitmap1(t *testing.T)    { testBitmapQuick(t, 10000, 0, 10000) }