feat(skiplist): Add sorted skiplist builder (#1693)

Add a Builder type in skiplist package which can be used to insert
sorted keys efficiently. Add a test and benchmark for it.

Author: ahsanbarkati

skl/arena.go

@@ -64,9 +64,7 @@ func (s *Arena) putNode(height int) uint32 {
 	// Pad the allocation with enough bytes to ensure pointer alignment.
 	l := uint32(MaxNodeSize - unusedSize + nodeAlign)
 	n := atomic.AddUint32(&s.n, l)
-	y.AssertTruef(int(n) <= len(s.buf),
-		"Arena too small, toWrite:%d newTotal:%d limit:%d",
-		l, n, len(s.buf))
+	y.AssertTrue(int(n) <= len(s.buf))
 
 	// Return the aligned offset.
 	m := (n - l + uint32(nodeAlign)) & ^uint32(nodeAlign)
@@ -80,9 +78,7 @@ func (s *Arena) putNode(height int) uint32 {
 func (s *Arena) putVal(v y.ValueStruct) uint32 {
 	l := uint32(v.EncodedSize())
 	n := atomic.AddUint32(&s.n, l)
-	y.AssertTruef(int(n) <= len(s.buf),
-		"Arena too small, toWrite:%d newTotal:%d limit:%d",
-		l, n, len(s.buf))
+	y.AssertTrue(int(n) <= len(s.buf))
 	m := n - l
 	v.Encode(s.buf[m:])
 	return m
@@ -91,9 +87,7 @@ func (s *Arena) putVal(v y.ValueStruct) uint32 {
 func (s *Arena) putKey(key []byte) uint32 {
 	l := uint32(len(key))
 	n := atomic.AddUint32(&s.n, l)
-	y.AssertTruef(int(n) <= len(s.buf),
-		"Arena too small, toWrite:%d newTotal:%d limit:%d",
-		l, n, len(s.buf))
+	y.AssertTrue(int(n) <= len(s.buf))
 	// m is the offset where you should write.
 	// n = new len - key len give you the offset at which you should write.
 	m := n - l

skl/skl.go

@@ -33,6 +33,7 @@ Key differences:
 package skl
 
 import (
+	"fmt"
 	"math"
 	"sync/atomic"
 	"unsafe"
@@ -522,3 +523,46 @@ func (s *UniIterator) Valid() bool { return s.iter.Valid() }
 
 // Close implements y.Interface (and frees up the iter's resources)
 func (s *UniIterator) Close() error { return s.iter.Close() }
+
+// Builder can be used to efficiently create a skiplist given that the keys are known to be in a
+// sorted order.
+type Builder struct {
+	s       *Skiplist
+	prev    [maxHeight + 1]*node
+	prevKey []byte
+}
+
+func NewBuilder(arenaSize int64) *Builder {
+	s := NewSkiplist(arenaSize)
+	b := &Builder{s: s}
+	for i := 0; i < maxHeight+1; i++ {
+		b.prev[i] = s.head
+	}
+	return b
+}
+
+const debug = false
+
+// Add must be used to add keys in a sorted order.
+func (b *Builder) Add(k []byte, v y.ValueStruct) {
+	if debug {
+		if len(b.prevKey) > 0 && y.CompareKeys(k, b.prevKey) <= 0 {
+			panic(fmt.Sprintf("new key: %s <= prev key: %s\n",
+				y.ParseKey(k), y.ParseKey(b.prevKey)))
+		}
+		b.prevKey = append(b.prevKey[:0], k...)
+	}
+	s := b.s
+	height := s.randomHeight()
+	if int32(height) > s.height {
+		s.height = int32(height)
+	}
+
+	x := newNode(s.arena, k, v, height)
+	nodeOffset := s.arena.getNodeOffset(x)
+	for i := 0; i < height; i++ {
+		node := b.prev[i]
+		node.tower[i] = nodeOffset
+		b.prev[i] = x
+	}
+}

skl/skl_test.go

@@ -17,6 +17,7 @@
 package skl
 
 import (
+	"bytes"
 	"encoding/binary"
 	"fmt"
 	"math/rand"
@@ -457,6 +458,39 @@ func randomKey(rng *rand.Rand) []byte {
 	return y.KeyWithTs(b, 0)
 }
 
+func TestBuilder(t *testing.T) {
+	N := 1 << 16
+	b := NewBuilder(32 << 20)
+	buf := make([]byte, 8)
+	for i := 0; i < N; i++ {
+		binary.BigEndian.PutUint64(buf, uint64(i))
+		key := y.KeyWithTs(buf, 0)
+		b.Add(key, y.ValueStruct{Value: []byte("00072")})
+	}
+	sl := b.s
+	for i := 0; i < N; i++ {
+		binary.BigEndian.PutUint64(buf, uint64(i))
+		key := y.KeyWithTs(buf, 0)
+		v := sl.Get(key)
+		require.NotNil(t, v.Value)
+		require.EqualValues(t, "00072", string(v.Value))
+	}
+	it := sl.NewIterator()
+	defer it.Close()
+
+	require.False(t, it.Valid())
+	it.SeekToFirst()
+	i := 0
+	for it.Valid() {
+		binary.BigEndian.PutUint64(buf, uint64(i))
+		key := y.KeyWithTs(buf, 0)
+		require.Equal(t, key, it.Key())
+		it.Next()
+		i++
+	}
+	require.Equal(t, N, i)
+}
+
 // Standard test. Some fraction is read. Some fraction is write. Writes have
 // to go through mutex lock.
 func BenchmarkReadWrite(b *testing.B) {
@@ -529,3 +563,49 @@ func BenchmarkWrite(b *testing.B) {
 		}
 	})
 }
+
+// $ go test -run=XXX -v -bench BenchmarkSortedWrites
+//
+// BenchmarkSortedWrites/builder-8         	11298607	       106 ns/op
+// BenchmarkSortedWrites/skiplist-8        	 2523454	       495 ns/op
+// BenchmarkSortedWrites/buffer-8          	10377798	       108 ns/op
+
+func BenchmarkSortedWrites(b *testing.B) {
+	b.Run("builder", func(b *testing.B) {
+		bl := NewBuilder(int64((b.N + 1) * MaxNodeSize))
+		buf := make([]byte, 8)
+		b.ResetTimer()
+		for i := 0; i < b.N; i++ {
+			binary.BigEndian.PutUint64(buf, uint64(i))
+			key := y.KeyWithTs(buf, 0)
+			bl.Add(key, y.ValueStruct{Value: []byte("00072")})
+		}
+	})
+
+	b.Run("skiplist", func(b *testing.B) {
+		bl := NewSkiplist(int64((b.N + 1) * MaxNodeSize))
+		buf := make([]byte, 8)
+		b.ResetTimer()
+		for i := 0; i < b.N; i++ {
+			binary.BigEndian.PutUint64(buf, uint64(i))
+			key := y.KeyWithTs(buf, 0)
+			bl.Put(key, y.ValueStruct{Value: []byte("00072")})
+		}
+	})
+
+	b.Run("buffer", func(b *testing.B) {
+		var bl bytes.Buffer
+		buf := make([]byte, 8)
+		b.ResetTimer()
+		for i := 0; i < b.N; i++ {
+			binary.BigEndian.PutUint64(buf, uint64(i))
+			key := y.KeyWithTs(buf, 0)
+			v := y.ValueStruct{Value: []byte("00072")}
+			vbuf := make([]byte, v.EncodedSize())
+			v.Encode(vbuf)
+
+			kv := append(key, vbuf...)
+			bl.Write(kv)
+		}
+	})
+}