Add a function to memstore to implement a fast Exists method on db (#43)

Author: sebheitzmann

memstore/memstore.go

@@ -2,6 +2,7 @@ package memstore
 
 import (
 	"errors"
+
 	"github.com/thomasjungblut/go-sstables/skiplist"
 	"github.com/thomasjungblut/go-sstables/sstables"
 )
@@ -19,6 +20,8 @@ type MemStoreI interface {
 	Add(key []byte, value []byte) error
 	// Contains returns true when the given key exists, false otherwise
 	Contains(key []byte) bool
+	// IsTombstoned returns true when the given key exists and is tombstoned, false otherwise
+	IsTombstoned(key []byte) bool
 	// Get returns the values for the given key, if not exists returns a KeyNotFound error
 	// if the key exists (meaning it was added and deleted) it will return KeyTombstoned as an error
 	Get(key []byte) ([]byte, error)
@@ -81,6 +84,21 @@ func (m *MemStore) Contains(key []byte) bool {
 	return true
 }
 
+func (m *MemStore) IsTombstoned(key []byte) bool {
+	exist := m.skipListMap.Contains(key)
+	if !exist {
+		return false
+	}
+	element, err := m.skipListMap.Get(key)
+	if errors.Is(err, skiplist.NotFound) {
+		return false
+	}
+	if *element.value == nil {
+		return true
+	}
+	return false
+}
+
 func (m *MemStore) Get(key []byte) ([]byte, error) {
 	element, err := m.skipListMap.Get(key)
 	// we can return false if we didn't find it by error, or when the key is tomb-stoned

memstore/memstore_test.go

@@ -2,11 +2,12 @@ package memstore
 
 import (
 	"errors"
+	"os"
+	"testing"
+
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/thomasjungblut/go-sstables/sstables"
-	"os"
-	"testing"
 )
 
 func TestMemStoreAddHappyPath(t *testing.T) {
@@ -18,6 +19,8 @@ func TestMemStoreAddHappyPath(t *testing.T) {
 	err = m.Add([]byte("a"), []byte("aVal"))
 	assert.Nil(t, err)
 	assert.True(t, m.Contains([]byte("a")))
+	assert.False(t, m.IsTombstoned([]byte("a")))
+
 	val, err = m.Get([]byte("a"))
 	assert.Nil(t, err)
 	assert.Equal(t, []byte("aVal"), val)
@@ -80,6 +83,7 @@ func TestMemStoreDeleteTombstones(t *testing.T) {
 
 	err = m.Delete([]byte("a"))
 	assert.False(t, m.Contains([]byte("a")))
+	assert.True(t, m.IsTombstoned([]byte("a")))
 	// make sure that the value was changed under the hood
 	kv, err = m.skipListMap.Get([]byte("a"))
 	assert.Nil(t, err)

sstables/super_sstable_reader.go

@@ -16,12 +16,20 @@ type SuperSSTableReader struct {
 }
 
 func (s SuperSSTableReader) Contains(key []byte) bool {
-	// this can't be implemented using contains because NotFound is the same as false, thus we have to go via Get
-	_, err := s.Get(key)
-	if err != nil && errors.Is(err, NotFound) {
-		return false
+	// scanning from back to front to get the latest definitive answer
+	for i := len(s.readers) - 1; i >= 0; i-- {
+		// first check if key exist to return fast
+		keyExist := s.readers[i].Contains(key)
+		if !keyExist {
+			continue
+		}
+		// we have to check if the value is not tombstoned
+		// maybe had to be implemented in an IsTombstoned in sstableReader
+		res, _ := s.readers[i].Get(key)
+		return res != nil
 	}
-	return true
+
+	return false
 }
 
 func (s SuperSSTableReader) Get(key []byte) ([]byte, error) {