fix(merge-operator): don't read the deleted keys dgraph-io#1675

Signed-off-by: thomassong <thomassong2012@gmail.com>

merge.go

@@ -20,8 +20,9 @@ import (
 	"sync"
 	"time"
 
-	"github.com/dgraph-io/badger/y"
 	"github.com/pkg/errors"
+
+	"github.com/dgraph-io/badger/y"
 )
 
 // MergeOperator represents a Badger merge operator.
@@ -69,6 +70,9 @@ func (op *MergeOperator) iterateAndMerge() (newVal []byte, latest uint64, err er
 	var numVersions int
 	for it.Rewind(); it.Valid(); it.Next() {
 		item := it.Item()
+		if item.IsDeletedOrExpired() {
+			break
+		}
 		numVersions++
 		if numVersions == 1 {
 			// This should be the newVal, considering this is the latest version.

merge_test.go

@@ -110,6 +110,39 @@ func TestGetMergeOperator(t *testing.T) {
 			require.Equal(t, uint64(6), bytesToUint64(res))
 		})
 	})
+
+	t.Run("Get after Delete", func(t *testing.T) {
+		key := []byte("merge")
+		runBadgerTest(t, nil, func(t *testing.T, db *DB) {
+			m := db.GetMergeOperator(key, add, 200*time.Millisecond)
+
+			err := m.Add(uint64ToBytes(1))
+			require.NoError(t, err)
+			m.Add(uint64ToBytes(2))
+			require.NoError(t, err)
+			m.Add(uint64ToBytes(3))
+			require.NoError(t, err)
+
+			m.Stop()
+			res, err := m.Get()
+			require.NoError(t, err)
+			require.Equal(t, uint64(6), bytesToUint64(res))
+
+			db.Update(func(txn *Txn) error {
+				return txn.Delete(key)
+			})
+
+			m = db.GetMergeOperator(key, add, 200*time.Millisecond)
+			err = m.Add(uint64ToBytes(1))
+			require.NoError(t, err)
+			m.Stop()
+
+			res, err = m.Get()
+			require.NoError(t, err)
+			require.Equal(t, uint64(1), bytesToUint64(res))
+		})
+	})
+
 	t.Run("Old keys should be removed after compaction", func(t *testing.T) {
 		dir, err := ioutil.TempDir("", "badger-test")
 		require.NoError(t, err)