Merge pull request #23 from Gouthamve/18-bytes-iterator

Implement Postings Iterator Over Bytes

index.go

@@ -698,20 +698,11 @@ func (r *indexReader) Postings(name, value string) (Postings, error) {
 		return nil, errors.Wrapf(errInvalidFlag, "section at %d", off)
 	}
 
-	// TODO(fabxc): just read into memory as an intermediate solution.
-	// Add iterator over serialized data.
-	var l []uint32
-
-	for len(b) > 0 {
-		if len(b) < 4 {
-			return nil, errors.Wrap(errInvalidSize, "plain postings entry")
-		}
-		l = append(l, binary.BigEndian.Uint32(b[:4]))
-
-		b = b[4:]
+	// Add iterator over the bytes.
+	if len(b)%4 != 0 {
+		return nil, errors.Wrap(errInvalidSize, "plain postings entry")
 	}
-
-	return &listPostings{list: l, idx: -1}, nil
+	return newBigEndianPostings(b), nil
 }
 
 type stringTuples struct {

postings.go

@@ -1,6 +1,7 @@
 package tsdb
 
 import (
+	"encoding/binary"
 	"sort"
 	"strings"
 )
@@ -240,6 +241,42 @@ func (it *listPostings) Err() error {
 	return nil
 }
 
+// bigEndianPostings implements the Postings interface over a byte stream of
+// big endian numbers.
+type bigEndianPostings struct {
+	list []byte
+	idx  int
+}
+
+func newBigEndianPostings(list []byte) *bigEndianPostings {
+	return &bigEndianPostings{list: list, idx: -1}
+}
+
+func (it *bigEndianPostings) At() uint32 {
+	idx := 4 * it.idx
+	return binary.BigEndian.Uint32(it.list[idx : idx+4])
+}
+
+func (it *bigEndianPostings) Next() bool {
+	it.idx++
+	return it.idx*4 < len(it.list)
+}
+
+func (it *bigEndianPostings) Seek(x uint32) bool {
+	num := len(it.list) / 4
+	// Do binary search between current position and end.
+	it.idx += sort.Search(num-it.idx, func(i int) bool {
+		idx := 4 * (it.idx + i)
+		val := binary.BigEndian.Uint32(it.list[idx : idx+4])
+		return val >= x
+	})
+	return it.idx*4 < len(it.list)
+}
+
+func (it *bigEndianPostings) Err() error {
+	return nil
+}
+
 type stringset map[string]struct{}
 
 func (ss stringset) set(s string) {

postings_test.go

@@ -1,8 +1,12 @@
 package tsdb
 
 import (
+	"encoding/binary"
+	"math/rand"
 	"reflect"
 	"testing"
+
+	"github.com/stretchr/testify/require"
 )
 
 type mockPostings struct {
@@ -181,3 +185,81 @@ func TestMerge(t *testing.T) {
 		}
 	}
 }
+
+func TestBigEndian(t *testing.T) {
+	num := 1000
+	// mock a list as postings
+	ls := make([]uint32, num)
+	ls[0] = 2
+	for i := 1; i < num; i++ {
+		ls[i] = ls[i-1] + uint32(rand.Int31n(25)) + 2
+	}
+
+	beLst := make([]byte, num*4)
+	for i := 0; i < num; i++ {
+		b := beLst[i*4 : i*4+4]
+		binary.BigEndian.PutUint32(b, ls[i])
+	}
+
+	t.Run("Iteration", func(t *testing.T) {
+		bep := newBigEndianPostings(beLst)
+		for i := 0; i < num; i++ {
+			require.True(t, bep.Next())
+			require.Equal(t, ls[i], bep.At())
+		}
+
+		require.False(t, bep.Next())
+		require.Nil(t, bep.Err())
+	})
+
+	t.Run("Seek", func(t *testing.T) {
+		table := []struct {
+			seek  uint32
+			val   uint32
+			found bool
+		}{
+			{
+				ls[0] - 1, ls[0], true,
+			},
+			{
+				ls[4], ls[4], true,
+			},
+			{
+				ls[500] - 1, ls[500], true,
+			},
+			{
+				ls[600] + 1, ls[601], true,
+			},
+			{
+				ls[600] + 1, ls[601], true,
+			},
+			{
+				ls[600] + 1, ls[601], true,
+			},
+			{
+				ls[0], ls[601], true,
+			},
+			{
+				ls[600], ls[601], true,
+			},
+			{
+				ls[999], ls[999], true,
+			},
+			{
+				ls[999] + 10, ls[999], false,
+			},
+		}
+
+		bep := newBigEndianPostings(beLst)
+		bep.Next()
+
+		for _, v := range table {
+			require.Equal(t, v.found, bep.Seek(v.seek))
+			// Once you seek beyond, At() will panic.
+			if v.found {
+				require.Equal(t, v.val, bep.At())
+				require.Nil(t, bep.Err())
+			}
+		}
+	})
+}