Avoid materializing unions in OrCardinality

orcardinality_test.go

@@ -0,0 +1,116 @@
+package roaring
+
+import "testing"
+
+func TestOrCardinalityContainers(t *testing.T) {
+	array := BitmapOf(1, 3, 63, 64, 65535)
+	bitmap := New()
+	for i := uint32(0); i < 65536; i += 2 {
+		bitmap.Add(i)
+	}
+	runs := New()
+	runs.AddRange(100, 60000)
+	otherKey := BitmapOf((1 << 16) + 1)
+	cases := []struct {
+		name   string
+		bitmap *Bitmap
+	}{
+		{"empty", New()}, {"array", array}, {"bitmap", bitmap},
+		{"run", runs}, {"other-key", otherKey},
+	}
+	for _, left := range cases {
+		for _, right := range cases {
+			t.Run(left.name+"/"+right.name, func(t *testing.T) {
+				x, y := left.bitmap.Clone(), right.bitmap.Clone()
+				want := Or(x, y).GetCardinality()
+				if got := x.OrCardinality(y); got != want {
+					t.Fatalf("got %d, want %d", got, want)
+				}
+				allocs := testing.AllocsPerRun(10, func() { x.OrCardinality(y) })
+				if allocs != 0 {
+					t.Errorf("OrCardinality allocated %g times", allocs)
+				}
+				if !x.Equals(left.bitmap) || !y.Equals(right.bitmap) {
+					t.Fatal("OrCardinality modified an input")
+				}
+			})
+		}
+	}
+}
+
+func BenchmarkOrCardinalityArray(b *testing.B) {
+	for _, tc := range []struct {
+		name                              string
+		leftSize, rightSize, step, offset int
+	}{
+		{"half-overlap", 4096, 4096, 2, 0},
+		{"identical", 4096, 4096, 2, -1},
+		{"interleaved", 4096, 4096, 2, 1},
+		{"separated", 4096, 4096, 2, 16384},
+		{"skew", 4096, 16, 512, -1},
+		{"tiny", 16, 16, 2, 0},
+	} {
+		b.Run(tc.name, func(b *testing.B) {
+			x, y := New(), New()
+			for high := uint32(0); high < 4; high++ {
+				for i := 0; i < tc.leftSize; i++ {
+					x.Add(high<<16 | uint32(i*2))
+				}
+				for i := 0; i < tc.rightSize; i++ {
+					v := i * tc.step
+					if tc.offset == 0 {
+						v = (i/2)*4 + i%2
+					} else if tc.offset > 0 {
+						v += tc.offset
+					}
+					y.Add(high<<16 | uint32(v))
+				}
+			}
+			want := Or(x, y).GetCardinality()
+			var got uint64
+			b.ReportAllocs()
+			for b.Loop() {
+				got = x.OrCardinality(y)
+			}
+			if got != want {
+				b.Fatalf("got %d, want %d", got, want)
+			}
+		})
+	}
+}
+
+func BenchmarkOrCardinalityContainers(b *testing.B) {
+	array, bitmap, runs := New(), New(), New()
+	for high := uint32(0); high < 4; high++ {
+		base := high << 16
+		for i := uint32(0); i < 65536; i += 2 {
+			bitmap.Add(base + i)
+		}
+		for i := uint32(0); i < 8192; i += 2 {
+			array.Add(base + i)
+		}
+		runs.AddRange(uint64(base)+100, uint64(base)+60000)
+	}
+	for _, tc := range []struct {
+		name string
+		x, y *Bitmap
+	}{
+		{"bitmap-bitmap", bitmap, bitmap.Clone()},
+		{"array-bitmap", array, bitmap},
+		{"run-bitmap", runs, bitmap},
+		{"run-array", runs, array},
+		{"run-run", runs, runs.Clone()},
+	} {
+		b.Run(tc.name, func(b *testing.B) {
+			want := Or(tc.x, tc.y).GetCardinality()
+			var got uint64
+			b.ReportAllocs()
+			for b.Loop() {
+				got = tc.x.OrCardinality(tc.y)
+			}
+			if got != want {
+				b.Fatalf("got %d, want %d", got, want)
+			}
+		})
+	}
+}

real_data_benchmark_test.go

@@ -7,6 +7,7 @@ import (
 	"io"
 	"os"
 	"path"
+	"runtime"
 	"strconv"
 	"strings"
 	"testing"
@@ -186,3 +187,41 @@ func BenchmarkRealDataFastOr(b *testing.B) {
 		return FastOr(bitmaps...).GetCardinality()
 	})
 }
+
+func BenchmarkRealDataOrCardinality(b *testing.B) {
+	if !benchRealData {
+		b.SkipNow()
+	}
+	for _, dataset := range realDatasets {
+		for _, optimize := range []bool{false, true} {
+			mode := "raw"
+			if optimize {
+				mode = "run"
+			}
+			b.Run(dataset+"/"+mode, func(b *testing.B) {
+				bitmaps, err := retrieveRealDataBitmaps(dataset, optimize)
+				if err != nil {
+					b.Fatal(err)
+				}
+				var want uint64
+				for j := 1; j < len(bitmaps); j++ {
+					x, y := bitmaps[j-1], bitmaps[j]
+					want += x.GetCardinality() + y.GetCardinality() - x.AndCardinality(y)
+				}
+				runtime.GC()
+				var got uint64
+				b.ReportAllocs()
+				for b.Loop() {
+					got = 0
+					for j := 1; j < len(bitmaps); j++ {
+						got += bitmaps[j-1].OrCardinality(bitmaps[j])
+					}
+				}
+				if got != want {
+					b.Fatalf("got %d, want %d", got, want)
+				}
+				b.ReportMetric(float64(len(bitmaps)-1), "pairs/op")
+			})
+		}
+	}
+}

roaring.go

@@ -1443,8 +1443,7 @@ main:
 					}
 					s2 = x2.highlowcontainer.getKeyAtIndex(pos2)
 				} else {
-					// TODO: could be faster if we did not have to materialize the container
-					answer += uint64(rb.highlowcontainer.getContainerAtIndex(pos1).or(x2.highlowcontainer.getContainerAtIndex(pos2)).getCardinality())
+					answer += uint64(rb.highlowcontainer.getContainerAtIndex(pos1).orCardinality(x2.highlowcontainer.getContainerAtIndex(pos2)))
 					pos1++
 					pos2++
 					if (pos1 == length1) || (pos2 == length2) {

roaring64/orcardinality_test.go

@@ -0,0 +1,60 @@
+package roaring64
+
+import "testing"
+
+func TestOrCardinalityContainers(t *testing.T) {
+	const base uint64 = 1 << 40
+	array := BitmapOf(base+1, base+3, base+63, base+64, base+65535)
+	bitmap := New()
+	for i := uint64(0); i < 65536; i += 2 {
+		bitmap.Add(base + i)
+	}
+	runs := New()
+	runs.AddRange(base+100, base+60000)
+	otherKey := BitmapOf(base + (1 << 48) + 1)
+	cases := []struct {
+		name   string
+		bitmap *Bitmap
+	}{
+		{"empty", New()}, {"array", array}, {"bitmap", bitmap},
+		{"run", runs}, {"other-key", otherKey},
+	}
+	for _, left := range cases {
+		for _, right := range cases {
+			t.Run(left.name+"/"+right.name, func(t *testing.T) {
+				x, y := left.bitmap.Clone(), right.bitmap.Clone()
+				want := Or(x, y).GetCardinality()
+				if got := x.OrCardinality(y); got != want {
+					t.Fatalf("got %d, want %d", got, want)
+				}
+				allocs := testing.AllocsPerRun(10, func() { x.OrCardinality(y) })
+				if allocs != 0 {
+					t.Errorf("OrCardinality allocated %g times", allocs)
+				}
+				if !x.Equals(left.bitmap) || !y.Equals(right.bitmap) {
+					t.Fatal("OrCardinality modified an input")
+				}
+			})
+		}
+	}
+}
+
+func BenchmarkOrCardinalityArray(b *testing.B) {
+	x, y := New(), New()
+	for high := uint64(1); high <= 4; high++ {
+		for low := uint64(0); low < 8192; low += 4 {
+			x.Add(high<<32 | low)
+			x.Add(high<<32 | (low + 2))
+			y.Add(high<<32 | low)
+			y.Add(high<<32 | (low + 1))
+		}
+	}
+	var got uint64
+	b.ReportAllocs()
+	for b.Loop() {
+		got = x.OrCardinality(y)
+	}
+	if want := uint64(4 * 6144); got != want {
+		b.Fatalf("got %d, want %d", got, want)
+	}
+}

roaring64/roaring64.go

@@ -513,8 +513,7 @@ main:
 					}
 					s2 = x2.highlowcontainer.getKeyAtIndex(pos2)
 				} else {
-					// TODO: could be faster if we did not have to materialize the container
-					answer += roaring.Or(rb.highlowcontainer.getContainerAtIndex(pos1), x2.highlowcontainer.getContainerAtIndex(pos2)).GetCardinality()
+					answer += rb.highlowcontainer.getContainerAtIndex(pos1).OrCardinality(x2.highlowcontainer.getContainerAtIndex(pos2))
 					pos1++
 					pos2++
 					if (pos1 == length1) || (pos2 == length2) {

setutil.go

@@ -122,51 +122,7 @@ func exclusiveUnion2by2(set1 []uint16, set2 []uint16, buffer []uint16) int {
 
 // union2by2Cardinality computes the cardinality of the union
 func union2by2Cardinality(set1 []uint16, set2 []uint16) int {
-	pos := 0
-	k1 := 0
-	k2 := 0
-	if 0 == len(set2) {
-		return len(set1)
-	}
-	if 0 == len(set1) {
-		return len(set2)
-	}
-	s1 := set1[k1]
-	s2 := set2[k2]
-	for {
-		if s1 < s2 {
-			pos++
-			k1++
-			if k1 >= len(set1) {
-				pos += len(set2) - k2
-				break
-			}
-			s1 = set1[k1]
-		} else if s1 == s2 {
-			pos++
-			k1++
-			k2++
-			if k1 >= len(set1) {
-				pos += len(set2) - k2
-				break
-			}
-			if k2 >= len(set2) {
-				pos += len(set1) - k1
-				break
-			}
-			s1 = set1[k1]
-			s2 = set2[k2]
-		} else { // if (set1[k1]>set2[k2])
-			pos++
-			k2++
-			if k2 >= len(set2) {
-				pos += len(set1) - k1
-				break
-			}
-			s2 = set2[k2]
-		}
-	}
-	return pos
+	return len(set1) + len(set2) - intersection2by2Cardinality(set1, set2)
 }
 
 func intersection2by2(