Avoid bitmap conversions in non-inplace array subtraction

arraycontainer.go

@@ -693,9 +693,27 @@ func (ac *arrayContainer) andNot(a container) container {
 }
 
 func (ac *arrayContainer) andNotRun16(rc *runContainer16) container {
-	acb := ac.toBitmapContainer()
-	rcb := rc.toBitmapContainer()
-	return acb.andNotBitmap(rcb)
+	if len(rc.iv) == 1 {
+		return ac.andNotInterval(rc.iv[0])
+	}
+	return ac.clone().(*arrayContainer).iandNotRun16(rc)
+}
+
+func (ac *arrayContainer) andNotInterval(iv interval16) container {
+	start := binarySearch(ac.content, iv.start)
+	if start < 0 {
+		start = -start - 1
+	}
+	end := binarySearch(ac.content, iv.last())
+	if end < 0 {
+		end = -end - 1
+	} else {
+		end++
+	}
+	answer := newArrayContainerSize(start + len(ac.content) - end)
+	copy(answer.content, ac.content[:start])
+	copy(answer.content[start:], ac.content[end:])
+	return answer
 }
 
 func (ac *arrayContainer) iandNot(a container) container {
@@ -763,6 +781,14 @@ func (ac *arrayContainer) iandNotRun16(rc *runContainer16) container {
 }
 
 func (ac *arrayContainer) andNotArray(value2 *arrayContainer) container {
+	if len(value2.content) > 0 {
+		start := value2.content[0]
+		last := value2.content[len(value2.content)-1]
+		if int(last)-int(start)+1 == len(value2.content) {
+			return ac.andNotInterval(interval16{start: start, length: uint16(len(value2.content) - 1)})
+		}
+	}
+
 	value1 := ac
 	desiredcapacity := value1.getCardinality()
 	answer := newArrayContainerCapacity(desiredcapacity)

arraycontainer_test.go

@@ -358,6 +358,80 @@ func TestArrayContainerIAndNot(t *testing.T) {
 	require.Equal(t, 2, ac.getCardinality())
 }
 
+func TestArrayContainerAndNotRun(t *testing.T) {
+	tests := []struct {
+		name      string
+		content   []uint16
+		intervals []interval16
+		want      []uint16
+	}{
+		{
+			name:    "empty run",
+			content: []uint16{0, 10, MaxUint16},
+			want:    []uint16{0, 10, MaxUint16},
+		},
+		{
+			name:      "empty array",
+			intervals: []interval16{newInterval16Range(0, 0)},
+			want:      []uint16{},
+		},
+		{
+			name:      "single run",
+			content:   []uint16{0, 2, 3, 4, 6, 8, 10},
+			intervals: []interval16{newInterval16Range(3, 8)},
+			want:      []uint16{0, 2, 10},
+		},
+		{
+			name:      "full run",
+			content:   []uint16{0, 2, MaxUint16},
+			intervals: []interval16{newInterval16Range(0, MaxUint16)},
+			want:      []uint16{},
+		},
+		{
+			name:      "run after array",
+			content:   []uint16{1, 3, 5},
+			intervals: []interval16{newInterval16Range(10, 20)},
+			want:      []uint16{1, 3, 5},
+		},
+		{
+			name:      "run before array",
+			content:   []uint16{10, 20, 30},
+			intervals: []interval16{newInterval16Range(0, 5)},
+			want:      []uint16{10, 20, 30},
+		},
+		{
+			name:    "multiple runs",
+			content: []uint16{0, 3, 4, 10, 12, 15, MaxUint16},
+			intervals: []interval16{
+				newInterval16Range(0, 0),
+				newInterval16Range(4, 10),
+				newInterval16Range(MaxUint16, MaxUint16),
+			},
+			want: []uint16{3, 12, 15},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			ac := &arrayContainer{content: append([]uint16(nil), test.content...)}
+			rc := newRunContainer16CopyIv(test.intervals)
+			originalContent := append([]uint16(nil), ac.content...)
+			originalIntervals := make([]interval16, len(rc.iv))
+			copy(originalIntervals, rc.iv)
+
+			got := ac.andNotRun16(rc).(*arrayContainer)
+			assert.Equal(t, test.want, got.content)
+			assert.Equal(t, originalContent, ac.content)
+			assert.Equal(t, originalIntervals, rc.iv)
+			assert.NotSame(t, ac, got)
+			if len(got.content) > 0 {
+				got.content[0]++
+				assert.Equal(t, originalContent, ac.content)
+			}
+		})
+	}
+}
+
 func TestArrayContainerIand(t *testing.T) {
 	a := NewBitmap()
 	a.AddRange(0, 200000)

roaring64/benchmark_test.go

@@ -135,3 +135,33 @@ func BenchmarkIteratorAlloc(b *testing.B) {
 		b.Fatalf("Cardinalities don't match: %d, %d", counter, expectedCardinality)
 	}
 }
+
+func BenchmarkAndNotArrayRun(b *testing.B) {
+	left, right := andNotArrayRunInputs()
+	if left.Stats().ArrayContainers != 1 || right.Stats().RunContainers != 1 {
+		b.Fatal("benchmark inputs do not use array/run containers")
+	}
+	if got := AndNot(left, right).GetCardinality(); got != 1024 {
+		b.Fatalf("unexpected result cardinality: got %d, want 1024", got)
+	}
+
+	b.ResetTimer()
+	for b.Loop() {
+		AndNot(left, right)
+	}
+}
+
+func BenchmarkAndNotArrayArray(b *testing.B) {
+	left, right := andNotArrayArrayInputs()
+	if left.Stats().ArrayContainers != 1 || right.Stats().ArrayContainers != 1 {
+		b.Fatal("benchmark inputs do not use array/array containers")
+	}
+	if got := AndNot(left, right).GetCardinality(); got != 1920 {
+		b.Fatalf("unexpected result cardinality: got %d, want 1920", got)
+	}
+
+	b.ResetTimer()
+	for b.Loop() {
+		AndNot(left, right)
+	}
+}

roaring64/roaring64_test.go

@@ -1823,6 +1823,71 @@ func TestAndNot(t *testing.T) {
 	assert.True(t, correct.Equals(rr))
 }
 
+func andNotArrayRunInputs() (*Bitmap, *Bitmap) {
+	const highKey = uint64(1) << 32
+
+	left := NewBitmap()
+	for value := uint64(0); value < 32768; value += 16 {
+		left.Add(highKey + value)
+	}
+
+	right := NewBitmap()
+	right.AddRange(highKey+8192, highKey+24576)
+	right.RunOptimize()
+	return left, right
+}
+
+func andNotArrayArrayInputs() (*Bitmap, *Bitmap) {
+	const highKey = uint64(1) << 32
+
+	left := NewBitmap()
+	for value := uint64(0); value < 32768; value += 16 {
+		left.Add(highKey + value)
+	}
+
+	right := NewBitmap()
+	for value := uint64(8192); value < 10240; value++ {
+		right.Add(highKey + value)
+	}
+	return left, right
+}
+
+func TestAndNotArrayRun(t *testing.T) {
+	left, right := andNotArrayRunInputs()
+	assert.EqualValues(t, 1, left.Stats().ArrayContainers)
+	assert.EqualValues(t, 1, right.Stats().RunContainers)
+
+	leftBefore := left.Clone()
+	rightBefore := right.Clone()
+	result := AndNot(left, right)
+
+	const highKey = uint64(1) << 32
+	assert.EqualValues(t, 1024, result.GetCardinality())
+	assert.True(t, result.Contains(highKey))
+	assert.True(t, result.Contains(highKey+24576))
+	assert.False(t, result.Contains(highKey+8192))
+	assert.True(t, left.Equals(leftBefore))
+	assert.True(t, right.Equals(rightBefore))
+}
+
+func TestAndNotArrayArray(t *testing.T) {
+	left, right := andNotArrayArrayInputs()
+	assert.EqualValues(t, 1, left.Stats().ArrayContainers)
+	assert.EqualValues(t, 1, right.Stats().ArrayContainers)
+
+	leftBefore := left.Clone()
+	rightBefore := right.Clone()
+	result := AndNot(left, right)
+
+	const highKey = uint64(1) << 32
+	assert.EqualValues(t, 1920, result.GetCardinality())
+	assert.True(t, result.Contains(highKey))
+	assert.True(t, result.Contains(highKey+32752))
+	assert.False(t, result.Contains(highKey+8192))
+	assert.True(t, left.Equals(leftBefore))
+	assert.True(t, right.Equals(rightBefore))
+}
+
 func TestStats(t *testing.T) {
 	t.Run("Test Stats with empty bitmap", func(t *testing.T) {
 		expectedStats := roaring.Statistics{}