Correction skiplist should skip tombstoned keys on traversal

Author: Alex Gaetano Padula

k4.go

@@ -117,8 +117,8 @@ type WALIterator struct {
 
 // KV mainly used for serialization
 type KV struct {
-	Key   []byte
-	Value []byte
+	Key   []byte // Binary array of key
+	Value []byte // Binary array of keys value
 }
 
 // Open opens a new K4 instance at the specified directory.
@@ -185,7 +185,7 @@ func Open(directory string, memtableFlushThreshold int, compactionInterval int,
 	// We open sstable files in the configured directory
 	k4.loadSSTables()
 
-	// If logging is set we will open a loggin file, so we can write to it
+	// If logging is set we will open a logging file, so we can write to it
 	if logging {
 		// Create log file
 		logFile, err := os.OpenFile(directory+string(os.PathSeparator)+LOG_EXTENSION, os.O_RDWR|os.O_CREATE|os.O_APPEND, 0644)
@@ -312,18 +312,18 @@ func (k4 *K4) backgroundWalWriter() {
 
 // serializeOp serializes an operation
 func serializeOp(op OPR_CODE, key, value []byte) []byte {
-	var buf bytes.Buffer
-
-	// use gob
+	var buf bytes.Buffer // create a buffer
 
-	enc := gob.NewEncoder(&buf)
+	enc := gob.NewEncoder(&buf) // create a new encoder with the buffer
 
+	// create an operation struct and initialize it
 	operation := Operation{
 		Op:    op,
 		Key:   key,
 		Value: value,
 	}
 
+	// encode the operation
 	err := enc.Encode(&operation)
 	if err != nil {
 		return nil
@@ -887,7 +887,7 @@ func (txn *Transaction) Rollback(k4 *K4) error {
 		op := txn.ops[i]
 		switch op.Op {
 		case PUT:
-			err := k4.appendToWALQueue(PUT, op.Key, op.Value)
+			err := k4.appendToWALQueue(PUT, op.Key, []byte(TOMBSTONE_VALUE))
 			if err != nil {
 				return err
 			}

skiplist/skiplist.go

@@ -120,20 +120,10 @@ func (sl *SkipList) Insert(key, value []byte, ttl *time.Duration) {
 
 	// Traverse the skip list to find the position of the key
 	for i := sl.level; i >= 0; i-- {
-		for current.forward[i] != nil && string(current.forward[i].key) < string(key) {
-			if current.forward[i].IsExpired() {
-				//sl.Delete(current.forward[i].key)
-				// We mark as tombstone and delete later
-				current.forward[i].ttl = nil
-
-				// Subtraction of the size of the old value
-				sl.size -= current.forward[i].Size()
-				// Mark as tombstone
-				current.forward[i].value = []byte(TOMBSTONE_VALUE)
-
-				// Add the size of the tombstone
-				sl.size += current.forward[i].Size()
-
+		for current.forward[i] != nil && (string(current.forward[i].key) < string(key) || bytes.Equal(current.forward[i].value, []byte(TOMBSTONE_VALUE))) {
+			if current.forward[i].IsExpired() || bytes.Equal(current.forward[i].value, []byte(TOMBSTONE_VALUE)) {
+				// Skip nodes with tombstone values
+				current = current.forward[i]
 			} else {
 				current = current.forward[i]
 			}
@@ -181,19 +171,10 @@ func (sl *SkipList) Delete(key []byte) {
 	current := sl.header
 
 	for i := sl.level; i >= 0; i-- {
-		for current.forward[i] != nil && string(current.forward[i].key) < string(key) {
-			if current.forward[i].IsExpired() {
-				//sl.Delete(current.forward[i].key)
-				// We mark as tombstone and delete later
-				current.forward[i].ttl = nil
-
-				// Subtraction of the size of the old value
-				sl.size -= current.forward[i].Size()
-				// Mark as tombstone
-				current.forward[i].value = []byte(TOMBSTONE_VALUE)
-
-				// Add the size of the tombstone
-				sl.size += current.forward[i].Size()
+		for current.forward[i] != nil && (string(current.forward[i].key) < string(key) || bytes.Equal(current.forward[i].value, []byte(TOMBSTONE_VALUE))) {
+			if current.forward[i].IsExpired() || bytes.Equal(current.forward[i].value, []byte(TOMBSTONE_VALUE)) {
+				// Skip nodes with tombstone values
+				current = current.forward[i]
 			} else {
 				current = current.forward[i]
 			}
@@ -222,18 +203,10 @@ func (sl *SkipList) Delete(key []byte) {
 func (sl *SkipList) Search(key []byte) ([]byte, bool) {
 	current := sl.header
 	for i := sl.level; i >= 0; i-- {
-		for current.forward[i] != nil && string(current.forward[i].key) < string(key) {
+		for current.forward[i] != nil && (string(current.forward[i].key) < string(key) || bytes.Equal(current.forward[i].value, []byte(TOMBSTONE_VALUE))) {
 			if current.forward[i].IsExpired() || bytes.Equal(current.forward[i].value, []byte(TOMBSTONE_VALUE)) {
-				// Mark as tombstone and delete later
-				current.forward[i].ttl = nil
-
-				// Subtraction of the size of the old value
-				sl.size -= current.forward[i].Size()
-				// Mark as tombstone
-				current.forward[i].value = []byte(TOMBSTONE_VALUE)
-
-				// Add the size of the tombstone
-				sl.size += current.forward[i].Size()
+				// Skip nodes with tombstone values
+				current = current.forward[i]
 			} else {
 				current = current.forward[i]
 			}
@@ -242,16 +215,6 @@ func (sl *SkipList) Search(key []byte) ([]byte, bool) {
 	current = current.forward[0]
 	if current != nil && string(current.key) == string(key) {
 		if current.IsExpired() || bytes.Equal(current.value, []byte(TOMBSTONE_VALUE)) {
-			// Mark as tombstone and delete later
-			current.ttl = nil
-
-			// Subtraction of the size of the old value
-			sl.size -= current.Size()
-			// Mark as tombstone
-			current.value = []byte(TOMBSTONE_VALUE)
-
-			// Add the size of the tombstone
-			sl.size += current.Size()
 			return nil, false
 		}
 		return current.value, true

skiplist/skiplist_test.go

@@ -158,4 +158,13 @@ func TestInsertTombstone(t *testing.T) {
 	if found {
 		t.Fatal("Expected key to be deleted")
 	}
+
+	key = []byte("key2")
+	value = []byte("$tombstone")
+
+	sl.Insert(key, value, nil)
+	_, found = sl.Search(key)
+	if found {
+		t.Fatal("Expected key to be deleted")
+	}
 }