Better Tombstone handling in Iter

engines/leveldb/iter.go

@@ -2,58 +2,90 @@ package leveldb
 
 import (
 	"bytes"
-	"fmt"
 
 	honuiter "github.com/rotationalio/honu/iterator"
 	pb "github.com/rotationalio/honu/object"
+	opts "github.com/rotationalio/honu/options"
 	"github.com/syndtr/goleveldb/leveldb/iterator"
 	"google.golang.org/protobuf/proto"
 )
 
 // NewLevelDBIterator creates a new iterator that wraps a leveldb Iterator with object
 // management access and Honu-specific serialization.
-func NewLevelDBIterator(iter iterator.Iterator, namespace string) honuiter.Iterator {
-	return &ldbIterator{ldb: iter, namespace: namespace}
+func NewLevelDBIterator(iter iterator.Iterator, options *opts.Options) honuiter.Iterator {
+	return &ldbIterator{ldb: iter, options: options}
 }
 
 // Wraps the underlying leveldb iterator to provide object management access.
 type ldbIterator struct {
-	ldb       iterator.Iterator
-	namespace string
+	ldb     iterator.Iterator
+	options *opts.Options
 }
 
 // Type check for the ldbIterator
 var _ honuiter.Iterator = &ldbIterator{}
 
-func (i *ldbIterator) Next() bool   { return i.ldb.Next() }
-func (i *ldbIterator) Prev() bool   { return i.ldb.Prev() }
 func (i *ldbIterator) Error() error { return i.ldb.Error() }
 func (i *ldbIterator) Release()     { i.ldb.Release() }
 
+func (i *ldbIterator) Next() bool {
+	if ok := i.ldb.Next(); !ok {
+		return false
+	}
+
+	// If we aren't including Tombstones, we need to check if the next version is a
+	// tombstone before we know if we have a next value or not.
+	if !i.options.Tombstones {
+		if obj, err := i.Object(); err != nil || obj.Tombstone() {
+			return i.Next()
+		}
+	}
+	return true
+}
+
+func (i *ldbIterator) Prev() bool {
+	if ok := i.ldb.Prev(); !ok {
+		return false
+	}
+
+	// If we aren't including Tombstones, we need to check if the next version is a
+	// tombstone before we know if we have a next value or not.
+	if !i.options.Tombstones {
+		if obj, err := i.Object(); err != nil || obj.Tombstone() {
+			return i.Prev()
+		}
+	}
+	return true
+}
+
 func (i *ldbIterator) Seek(key []byte) bool {
+	// NOTE: no need to do tombstone checking in Seek because Next will be called.
 	// We need to prefix the seek with the correct namespace
-	key = prepend(i.namespace, key)
+	if i.options.Namespace != "" {
+		key = prepend(i.options.Namespace, key)
+	}
 	return i.ldb.Seek(key)
 }
 
 func (i *ldbIterator) Key() []byte {
 	// Fetch the key then split the namespace from the key
+	// Note that because the namespace itself might have colons in it, we
+	// strip off the namespace prefix then remove any preceding colons.
 	key := i.ldb.Key()
-	parts := bytes.SplitN(key, nssep, 2)
-	if len(parts) == 2 {
-		return parts[1]
+	if i.options.Namespace != "" {
+		prefix := prepend(i.options.Namespace, nil)
+		return bytes.TrimPrefix(key, prefix)
 	}
 	return key
 }
 
 func (i *ldbIterator) Value() []byte {
 	obj, err := i.Object()
 	if err != nil {
-		fmt.Println(err)
+		// NOTE: if err is not nil, it's up to the caller to get the error from Object
 		return nil
-	} else {
-		return obj.Data
 	}
+	return obj.Data
 }
 
 func (i *ldbIterator) Object() (obj *pb.Object, err error) {
@@ -65,5 +97,5 @@ func (i *ldbIterator) Object() (obj *pb.Object, err error) {
 }
 
 func (i *ldbIterator) Namespace() string {
-	return i.namespace
+	return i.options.Namespace
 }

engines/leveldb/leveldb.go

@@ -192,7 +192,7 @@ func (db *LevelDBEngine) Iter(prefix []byte, options *opts.Options) (i iterator.
 	if len(prefix) > 0 {
 		slice = util.BytesPrefix(prefix)
 	}
-	return NewLevelDBIterator(db.ldb.NewIterator(slice, options.LevelDBRead), options.Namespace), nil
+	return NewLevelDBIterator(db.ldb.NewIterator(slice, options.LevelDBRead), options), nil
 }
 
 var nssep = []byte("::")

engines/leveldb/leveldb_test.go

@@ -1,6 +1,7 @@
 package leveldb_test
 
 import (
+	"fmt"
 	"io/ioutil"
 	"os"
 	"testing"
@@ -15,12 +16,15 @@ import (
 	"google.golang.org/protobuf/proto"
 )
 
+// a test set of key/value pairs used to evaluate iteration
+// note because :: is the namespace separator in leveldb, we want to ensure that keys
+// with colons are correctly iterated on.
 var pairs = [][]string{
 	{"aa", "first"},
 	{"ab", "second"},
-	{"ba", "third"},
-	{"bb", "fourth"},
-	{"bc", "fifth"},
+	{"b::a", "third"},
+	{"b::b", "fourth"},
+	{"b::c", "fifth"},
 	{"ca", "sixth"},
 	{"cb", "seventh"},
 }
@@ -44,11 +48,19 @@ func setupLevelDBEngine(t testing.TB) (_ *leveldb.LevelDBEngine, path string) {
 		os.RemoveAll(tempDir)
 	}
 	require.NoError(t, err)
+
+	// Add a cleanup function to ensure the fixture is deleted after tests
+	t.Cleanup(func() {
+		// Teardown after finishing the test
+		engine.Close()
+		os.RemoveAll(tempDir)
+		fmt.Printf("cleaned up %s\n", tempDir)
+	})
+
 	return engine, tempDir
 }
 
-// Creates an options.Options struct with namespace set and returns
-// a pointer to it.
+// Creates an options.Options struct with namespace set and returns a pointer to it.
 func namespaceOpts(namespace string, t *testing.T) *options.Options {
 	opts, err := options.New(options.WithNamespace(namespace))
 	require.NoError(t, err)
@@ -78,18 +90,14 @@ func checkDelete(ldbStore engine.Store, opts *options.Options, key []byte, t *te
 	require.Empty(t, value)
 }
 
-func TestLeveldbEngine(t *testing.T) {
+func TestLevelDBEngine(t *testing.T) {
 	// Setup a levelDB Engine.
 	ldbEngine, ldbPath := setupLevelDBEngine(t)
 	require.Equal(t, "leveldb", ldbEngine.Engine())
 
 	// Ensure the db was created.
 	require.DirExists(t, ldbPath)
 
-	// Teardown after finishing the test.
-	defer os.RemoveAll(ldbPath)
-	defer ldbEngine.Close()
-
 	// Use a constant key to ensure namespaces
 	// are working correctly.
 	key := []byte("foo")
@@ -111,12 +119,8 @@ func TestLeveldbEngine(t *testing.T) {
 	}
 }
 
-func TestLeveldbTransactions(t *testing.T) {
-	ldbEngine, ldbPath := setupLevelDBEngine(t)
-
-	// Teardown after finishing the test
-	defer os.RemoveAll(ldbPath)
-	defer ldbEngine.Close()
+func TestLevelDBTransactions(t *testing.T) {
+	ldbEngine, _ := setupLevelDBEngine(t)
 
 	// Use a constant key to ensure namespaces
 	// are working correctly.
@@ -155,20 +159,9 @@ func TestLeveldbTransactions(t *testing.T) {
 }
 
 func TestLevelDBIter(t *testing.T) {
-	ldbEngine, ldbPath := setupLevelDBEngine(t)
-
-	// Teardown after finishing the test
-	defer os.RemoveAll(ldbPath)
-	defer ldbEngine.Close()
+	ldbEngine, _ := setupLevelDBEngine(t)
 
 	for _, namespace := range testNamespaces {
-		// TODO: figure out what to do with this testcase.
-		// Iter currently grabs the namespace by splitting
-		// on :: and grabbing the first string, so it only
-		// grabs "namespace".
-		if namespace == "namespace::with::colons" {
-			continue
-		}
 		// Add data to the database to iterate over.
 		opts := namespaceOpts(namespace, t)
 
@@ -223,7 +216,16 @@ func addIterPairsToDB(ldbStore engine.Store, opts *options.Options, pairs [][]st
 		obj := &pb.Object{
 			Key:       key,
 			Namespace: opts.Namespace,
-			Data:      value,
+			Version: &pb.Version{
+				Pid:       1,
+				Version:   1,
+				Region:    "testing",
+				Parent:    nil,
+				Tombstone: false,
+			},
+			Region: "testing",
+			Owner:  "testing",
+			Data:   value,
 		}
 
 		data, err := proto.Marshal(obj)