Use advanceUntil for base in AndAny, improve tests

arraycontainer.go

@@ -892,7 +892,7 @@ func (ac *arrayContainer) resetTo(a container) {
 		ac.realloc(card)
 		cur := 0
 		for _, r := range x.iv {
-			for val := r.start; val <= r.start+r.length; val++ {
+			for val := r.start; val <= r.last(); val++ {
 				ac.content[cur] = val
 				cur++
 			}

benchmark_test.go

@@ -919,11 +919,13 @@ func BenchmarkBitmapReuseWithClear(b *testing.B) {
 func BenchmarkAndAny(b *testing.B) {
 
 	runSet := func(name string, base *Bitmap, filters []*Bitmap) {
-		var andFirstCard uint64
-		var orFirstCard uint64
-		var fastCard uint64
+		var (
+			andFirstCard, orFirstCard, andAnyCard uint64
+			andFirstRan, orFirstRan, andAnyRan    bool
+		)
 
 		b.Run(name+"_or-first", func(b *testing.B) {
+			andAnyRan = true
 			for n := 0; n < b.N; n++ {
 				clone := base.Clone()
 
@@ -935,6 +937,7 @@ func BenchmarkAndAny(b *testing.B) {
 		})
 
 		b.Run(name+"_and-first", func(b *testing.B) {
+			orFirstRan = true
 			for n := 0; n < b.N; n++ {
 				anded := make([]*Bitmap, 0, len(filters))
 
@@ -948,21 +951,22 @@ func BenchmarkAndAny(b *testing.B) {
 		})
 
 		b.Run(name+"_AndAny", func(b *testing.B) {
+			andAnyRan = true
 			for n := 0; n < b.N; n++ {
 				clone := base.Clone()
 
 				b.StartTimer()
 				clone.AndAny(filters...)
-				fastCard = clone.GetCardinality()
+				andAnyCard = clone.GetCardinality()
 				b.StopTimer()
 			}
 		})
 
-		if andFirstCard != orFirstCard {
-			b.Fatalf("Cardinalities don't match: %d, %d", andFirstCard, orFirstCard)
+		if andFirstRan && andAnyRan && andFirstCard != andAnyCard {
+			b.Fatalf("Cardinalities don't match: %d, %d", andFirstCard, andAnyCard)
 		}
-		if andFirstCard != fastCard {
-			b.Fatalf("Cardinalities don't match: %d, %d", andFirstCard, fastCard)
+		if orFirstRan && andAnyRan && orFirstCard != andAnyCard {
+			b.Fatalf("Cardinalities don't match: %d, %d", orFirstCard, andAnyCard)
 		}
 	}
 

fastaggregation.go

@@ -245,17 +245,19 @@ func (rb *Bitmap) AndAny(bitmaps ...*Bitmap) {
 	intersections := 0
 	keyContainers := make([]container, 0, len(filters))
 	var (
-		tmpArray  *arrayContainer
-		tmpBitmap *bitmapContainer
+		tmpArray   *arrayContainer
+		tmpBitmap  *bitmapContainer
+		minNextKey uint16
 	)
 
-	for ; basePos < rb.highlowcontainer.size() && len(filters) > 0; basePos++ {
+	for basePos < rb.highlowcontainer.size() && len(filters) > 0 {
 		baseKey := rb.highlowcontainer.getKeyAtIndex(basePos)
 
-		// accumulate containers for current key
-		// and exclude filters that do not have more related values
+		// accumulate containers for current key, find next minimal key in filters
+		// and exclude filters that do not have related values anymore
 		i := 0
 		maxPossibleOr := 0
+		minNextKey = MaxUint16
 		for _, f := range filters {
 			if f.key < baseKey {
 				f.pos = f.bitmap.advanceUntil(baseKey, f.pos)
@@ -269,14 +271,22 @@ func (rb *Bitmap) AndAny(bitmaps ...*Bitmap) {
 				cont := f.bitmap.getContainerAtIndex(f.pos)
 				keyContainers = append(keyContainers, cont)
 				maxPossibleOr += cont.getCardinality()
+
+				f.pos++
+				if f.pos == f.bitmap.size() {
+					continue
+				}
+				f.key = f.bitmap.getKeyAtIndex(f.pos)
 			}
 
+			minNextKey = minOfUint16(minNextKey, f.key)
 			filters[i] = f
 			i++
 		}
 		filters = filters[:i]
 
 		if len(keyContainers) == 0 {
+			basePos = rb.highlowcontainer.advanceUntil(minNextKey, basePos)
 			continue
 		}
 
@@ -315,6 +325,7 @@ func (rb *Bitmap) AndAny(bitmaps ...*Bitmap) {
 		}
 
 		keyContainers = keyContainers[:0]
+		basePos = rb.highlowcontainer.advanceUntil(minNextKey, basePos)
 	}
 
 	rb.highlowcontainer.resize(intersections)

fastaggregation_test.go

@@ -193,38 +193,75 @@ func TestFastAggregationsXOR_run(t *testing.T) {
 }
 
 func TestFastAggregationsAndAny(t *testing.T) {
+	base := NewBitmap()
 	rb1 := NewBitmap()
 	rb2 := NewBitmap()
 	rb3 := NewBitmap()
-	rb4 := NewBitmap()
-	for i := uint32(500); i < 75000; i++ {
+	// only one filter has some values
+	from := uint32(maxCapacity * 4)
+	for i := uint32(from); i < from+100; i += 2 {
 		rb1.Add(i)
 	}
-	for i := uint32(0); i < 1000000; i += 7 {
-		rb2.Add(i)
+	// only base has values
+	from = maxCapacity * 7
+	for i := uint32(from); i < from+100; i += 2 {
+		base.Add(i)
 	}
-	for i := uint32(0); i < 1000000; i += 1001 {
-		rb3.Add(i)
+	// base and one of filters have same values
+	from = maxCapacity * 8
+	for i := uint32(from); i < from+100; i += 2 {
+		base.Add(i)
+		rb1.Add(i)
 	}
-	for i := uint32(1000000); i < 2000000; i += 1001 {
+	// small union
+	from = maxCapacity * 10
+	for i := uint32(from); i < from+1000; i += 10 {
+		base.Add(i)
+		base.Add(i + i%3)
+
+		rb1.Add(i)
+		rb1.Add(i + 1)
+
+		rb2.Add(i + 2)
+		rb2.Add(i + i%7)
+
+		rb3.Add(200 + i)
+	}
+	// run filters
+	from = maxCapacity * 10
+	for i := uint32(from); i < from+1000; i += 3 {
+		base.Add(i)
+	}
+	for i := uint32(from); i < from+100; i++ {
 		rb1.Add(i)
+		rb2.Add(i + 333)
+		rb3.Add(i + 433)
 	}
-	for i := uint32(1000000); i < 2000000; i += 3 {
+	// large union
+	from = maxCapacity * 16
+	for i := uint32(from); i < from+arrayDefaultMaxSize*10; i += 3 {
+		base.Add(i)
+		base.Add(i + i%2 + 1)
 		rb2.Add(i)
-	}
-	for i := uint32(1000000); i < 2000000; i += 7 {
-		rb3.Add(i)
+		rb3.Add(i + 1)
 	}
 
-	rb4.Add(1001001)
+	// some extra base values
+	from = maxCapacity * 17
+	for i := uint32(from); i < from+1000; i++ {
+		base.Add(i)
+	}
 
+	base.RunOptimize()
 	rb1.RunOptimize()
+	rb2.RunOptimize()
+	rb3.RunOptimize()
 
-	base := rb1.Clone()
-	base.And(FastOr(rb2, rb3, rb4))
+	orFirst := base.Clone()
+	orFirst.And(FastOr(rb1, rb2, rb3))
 
-	fast := rb1.Clone()
-	fast.AndAny(rb2, rb3, rb4)
+	fast := base.Clone()
+	fast.AndAny(rb1, rb2, rb3)
 
-	assert.True(t, fast.Equals(base))
+	assert.True(t, fast.Equals(orFirst))
 }