introduces ContainsMany

Author: aliszka

bitmap_opt.go

@@ -458,6 +458,50 @@ func orContainersInRange(a, b *Bitmap, bi, bn int, buf []uint16) {
 	}
 }
 
+func (ra *Bitmap) ContainsMany(xs []uint64) []uint64 {
+	if xs == nil {
+		xs = []uint64{}
+	}
+
+	if len(xs) == 0 || ra.IsEmpty() {
+		return xs[:0]
+	}
+	if len(xs) == 1 {
+		if ra.Contains(xs[0]) {
+			return xs
+		}
+		return xs[:0]
+	}
+
+	keyToContainer := map[uint64][]uint16{}
+	j := 0
+	for i, x := range xs {
+		key := x & mask
+		container, ok := keyToContainer[key]
+		if !ok {
+			if offset, has := ra.keys.getValue(key); has {
+				container = ra.getContainer(offset)
+			}
+			keyToContainer[key] = container
+		}
+		if container != nil {
+			y := uint16(x)
+			contains := false
+			switch container[indexType] {
+			case typeArray:
+				contains = array(container).has(y)
+			case typeBitmap:
+				contains = bitmap(container).has(y)
+			}
+			if contains {
+				xs[j] = xs[i]
+				j++
+			}
+		}
+	}
+	return xs[:j]
+}
+
 const minContainersForConcurrency = 16
 
 // AndToSuperset calculates intersection of current and incoming bitmap

bitmap_opt_test.go

@@ -871,6 +871,32 @@ func TestIssue_Or_NotMergeContainers(t *testing.T) {
 	})
 }
 
+func TestContainsMany(t *testing.T) {
+	t.Run("contains even", func(t *testing.T) {
+		maxI := 10
+		maxJ := 100
+
+		set := make([]uint64, 0, maxI*maxJ)
+		control := make([]uint64, 0, maxI*maxJ)
+		bm := NewBitmap()
+
+		for i := 0; i < maxI; i++ {
+			for j := 0; j < maxJ; j++ {
+				x := uint64(maxCardinality*i + j)
+				set = append(set, x)
+				if j%2 == 0 {
+					control = append(control, x)
+					bm.Set(x)
+				}
+			}
+		}
+
+		contained := bm.ContainsMany(set)
+		require.Len(t, contained, len(control))
+		require.ElementsMatch(t, contained, control)
+	})
+}
+
 func TestMergeToSuperset(t *testing.T) {
 	run := func(t *testing.T, bufs [][]uint16) {
 		containerThreshold := uint64(math.MaxUint16 + 1)