fix bug where a count query and bitmap query could return different numbers

There was a case where the Bitmap iterator logic could skip over a bit in a run
container if 1. the run container was not the first container in the bitmap, and
2. The first run in the run container had only one bit.

The bug was due to how the iterator was initialized with iterator.Seek(0) which
sets up the initial values of itr.i,j,k based on the type of the first
container. It was failing to set itr.k to -1 unless the first container was an
RLE container. itr.k is only used by RLE containers in the iterator, and must be
set to -1 when an RLE container is encountered. When Iterator.Next() encountered
the run container and itr.k was set to 0, it checked to see if itr.k <= run.last
- run.first, and if so it assumes that it was finished with the run and moved to
the next one. run.last - run.first is 0 in the case of a single bit run, so that
bit was skipped. After this, itr.k is set to -1 and all further iteration
proceeds as expected.
This commit is contained in:
Matthew Jaffee 2018-02-06 10:02:53 -06:00
parent 902a2daf76
commit cc8733eedb
No known key found for this signature in database
GPG key ID: 51C676AF9FFCDB87
3 changed files with 153 additions and 6 deletions

View file

@ -15,6 +15,7 @@
package pilosa_test
import (
"fmt"
"reflect"
"testing"
@ -23,14 +24,34 @@ import (
// Ensure a bitmap can be merged
func TestBitmap_Merge(t *testing.T) {
bm1 := pilosa.NewBitmap(1, 2, 3, SliceWidth+1, 2*SliceWidth)
bm2 := pilosa.NewBitmap(3, 4, 5)
bm1.Merge(bm2)
if bm1.Count() != 7 {
t.Fatalf("Count after merge %d != 7\n", bm1.Count())
tests := []struct {
bm1 *pilosa.Bitmap
bm2 *pilosa.Bitmap
exp uint64
}{
{
bm1: pilosa.NewBitmap(1, 2, 3, SliceWidth+1, 2*SliceWidth),
bm2: pilosa.NewBitmap(3, 4, 5),
exp: 7,
},
{
bm1: pilosa.NewBitmap(),
bm2: pilosa.NewBitmap(2, 66000, 70000, 70001, 70002, 70003, 70004),
exp: 7,
},
}
for i, test := range tests {
t.Run(fmt.Sprintf("#%d:", i), func(t *testing.T) {
test.bm1.Merge(test.bm2)
if cnt := test.bm1.Count(); cnt != test.exp {
t.Fatalf("merged count %d is not %d", cnt, test.exp)
}
if length := len(test.bm1.Bits()); uint64(length) != test.exp {
t.Fatalf("merged length %d is not %d", length, test.exp)
}
})
}
}
// Ensure a bitmap can Xor'ed

View file

@ -840,6 +840,10 @@ func (itr *Iterator) eof() bool { return itr.i >= len(itr.bitmap.containers) }
// Seek moves to the first value equal to or greater than `seek`.
func (itr *Iterator) Seek(seek uint64) {
// k should always be -1 unless we're seeking into a run container. Then the
// "if c.isRun" section will take care of it.
itr.k = -1
// Move to the correct container.
itr.i = search64(itr.bitmap.keys, highbits(seek))
if itr.i < 0 {

View file

@ -2244,6 +2244,44 @@ func TestIteratorRuns(t *testing.T) {
}
}
func TestIteratorVarious(t *testing.T) {
tests := []struct {
bm *Bitmap
exp uint64
}{
{
bm: NewBitmap(3, 4, 5),
exp: 3,
},
{
bm: bitmapVariousContainers(),
exp: 61221,
},
{
bm: NewBitmap(2, 66000, 70000, 70001, 70002, 70003, 70004),
exp: 7,
},
}
for i, test := range tests {
test.bm.Optimize()
t.Run(fmt.Sprintf("#%d:", i), func(t *testing.T) {
if cnt := test.bm.Count(); cnt != test.exp {
t.Fatalf("merged count %d is not %d", cnt, test.exp)
}
iter := test.bm.Iterator()
bits := make([]uint64, 0, test.bm.Count())
for v, eof := iter.Next(); !eof; v, eof = iter.Next() {
bits = append(bits, v)
}
if length := len(bits); uint64(length) != test.exp {
t.Fatalf("length %d is not %d", length, test.exp)
}
})
}
}
func TestRunBinSearchContains(t *testing.T) {
tests := []struct {
runs []interval16
@ -2547,3 +2585,87 @@ func bitmapEvens() []uint64 {
}
return bitmap
}
var containerWidth uint64 = 65536
// rleCont returns a slice of numbers all in the range starting from
// container_width*num, and ending at container_width*(num+1)-1. If left is
// true, then the first 100 bits will be set, if mid is true, 100 bits in the
// middle will be set, if right is true, the last 100 bits will be set.
// sets 100 bits per true
func rleCont(num int, left, mid, right bool) []uint64 {
ret := make([]uint64, 0)
base := containerWidth * uint64(num)
if left {
for i := uint64(0); i < 100; i++ {
ret = append(ret, base+i)
}
}
if mid {
for i := containerWidth / 2; i < containerWidth/2+100; i++ {
ret = append(ret, base+i)
}
}
if right {
for i := containerWidth - 100; i < containerWidth; i++ {
ret = append(ret, base+i)
}
}
return ret
}
// sets 2 bits per true.
func arrCont(num int, left, mid, right bool) []uint64 {
ret := make([]uint64, 0)
base := containerWidth * uint64(num)
if left {
ret = append(ret, base+0, base+2)
}
if mid {
half := containerWidth / 2
ret = append(ret, base+half, base+half+2)
}
if right {
ret = append(ret, base+containerWidth-3, base+containerWidth-1)
}
return ret
}
// sets 6667 bits per true.
func bitCont(num int, left, mid, right bool) []uint64 {
ret := make([]uint64, 0)
base := containerWidth * uint64(num)
if left {
for i := uint64(0); i < 20001; i += 3 {
ret = append(ret, base+i)
}
}
if mid {
for i := uint64(21000); i < 41001; i += 3 {
ret = append(ret, base+i)
}
}
if right {
for i := uint64(45537); i <= 65535; i += 3 {
ret = append(ret, base+i)
}
}
return ret
}
func bitmapVariousContainers() *Bitmap {
bits := make([]uint64, 0)
bits = append(bits, rleCont(0, true, true, true)...)
bits = append(bits, rleCont(1, true, true, true)...)
bits = append(bits, arrCont(2, true, true, true)...)
bits = append(bits, arrCont(3, true, true, true)...)
bits = append(bits, bitCont(4, true, true, true)...)
bits = append(bits, bitCont(5, true, true, true)...)
bits = append(bits, rleCont(6, true, true, true)...)
bits = append(bits, bitCont(7, true, true, true)...)
bits = append(bits, arrCont(8, true, true, true)...)
bits = append(bits, rleCont(9, true, true, true)...)
bm := NewBitmap(bits...)
bm.Optimize()
return bm
}