Optimize lazy bitmap container unions

bitmapcontainer.go

@@ -651,16 +651,42 @@ func (bc *bitmapContainer) lazyORArray(value2 *arrayContainer) container {
 
 func (bc *bitmapContainer) lazyIORBitmap(value2 *bitmapContainer) container {
 	answer := bc
-	for k := 0; k < len(answer.bitmap); k++ {
-		answer.bitmap[k] = bc.bitmap[k] | value2.bitmap[k]
+	bitmap := answer.bitmap
+	other := value2.bitmap
+
+	// Bitmap containers always span bitmapContainerSize words. Prove the
+	// bounds once so the compiler can eliminate the checks in the unrolled loop.
+	_ = bitmap[bitmapContainerSize-1]
+	_ = other[bitmapContainerSize-1]
+	for k := 0; k < bitmapContainerSize; k += 4 {
+		bitmap[k] |= other[k]
+		bitmap[k+1] |= other[k+1]
+		bitmap[k+2] |= other[k+2]
+		bitmap[k+3] |= other[k+3]
 	}
-	bc.cardinality = invalidCardinality
+	answer.cardinality = invalidCardinality
 	return answer
 }
 
 func (bc *bitmapContainer) lazyORBitmap(value2 *bitmapContainer) container {
-	answer := bc.clone().(*bitmapContainer)
-	return answer.lazyIORBitmap(value2)
+	answer := newBitmapContainer()
+	bitmap := answer.bitmap
+	left := bc.bitmap
+	right := value2.bitmap
+
+	// Bitmap containers always span bitmapContainerSize words. Prove the
+	// bounds once so the compiler can eliminate the checks in the unrolled loop.
+	_ = bitmap[bitmapContainerSize-1]
+	_ = left[bitmapContainerSize-1]
+	_ = right[bitmapContainerSize-1]
+	for k := 0; k < bitmapContainerSize; k += 4 {
+		bitmap[k] = left[k] | right[k]
+		bitmap[k+1] = left[k+1] | right[k+1]
+		bitmap[k+2] = left[k+2] | right[k+2]
+		bitmap[k+3] = left[k+3] | right[k+3]
+	}
+	answer.cardinality = invalidCardinality
+	return answer
 }
 
 func (bc *bitmapContainer) xor(a container) container {

bitmapcontainer_bench_test.go

@@ -24,3 +24,47 @@ func BenchmarkBitmapContainerFillLeastSignificant16bits(b *testing.B) {
 		sink += x[pos-1]
 	}
 }
+
+// BenchmarkParOrBitmapContainers measures ParOr across four inputs with dense
+// bitmap containers at 64 shared keys using four workers.
+func BenchmarkParOrBitmapContainers(b *testing.B) {
+	const (
+		bitmapCount         = 4
+		containersPerBitmap = 64
+		parallelism         = 4
+	)
+
+	bitmaps := make([]*Bitmap, bitmapCount)
+	for i := range bitmaps {
+		words := make([]uint64, bitmapContainerSize*containersPerBitmap)
+		state := uint64(i + 1)
+		for j := range words {
+			state += 0x9e3779b97f4a7c15
+			word := state
+			word = (word ^ (word >> 30)) * 0xbf58476d1ce4e5b9
+			word = (word ^ (word >> 27)) * 0x94d049bb133111eb
+			words[j] = word ^ (word >> 31)
+		}
+		bitmaps[i] = FromDense(words, false)
+		for _, c := range bitmaps[i].highlowcontainer.containers {
+			if _, ok := c.(*bitmapContainer); !ok {
+				b.Fatal("workload did not produce bitmap containers")
+			}
+		}
+	}
+
+	expected := bitmaps[0].Clone()
+	for _, bitmap := range bitmaps[1:] {
+		expected.Or(bitmap)
+	}
+	expectedCardinality := expected.GetCardinality()
+
+	b.ReportAllocs()
+	b.ResetTimer()
+	for b.Loop() {
+		result := ParOr(parallelism, bitmaps...)
+		if result.GetCardinality() != expectedCardinality {
+			b.Fatal("unexpected cardinality")
+		}
+	}
+}