Repository: dgraph-io/badger
Pull request #365 (2017-12-19 02:13:29)

Fix bug while picking value log file


<!-- Reviewable:start -->
This change is [<img src="https://reviewable.io/review_button.svg" height="34" align="absmiddle" alt="Reviewable"/>](https://reviewable.io/reviews/dgraph-io/badger/365)
<!-- Reviewable:end -->


janardhan1993: 



Review status: 0 of 1 files reviewed at latest revision, 1 unresolved discussion.

---

*[value.go, line 846 at r1](https://reviewable.io:443/reviews/dgraph-io/badger/365#-L0gY5pswsvHIxzCszv6:-L0gY5ptCS88M1GYp5Aa:bl7jtau) ([raw file](https://github.com/dgraph-io/badger/blob/ce2168f7a30ca3cc71463b08b18d6a21db72c76e/value.go#L846)):*
> ```Go
> 
> 	i := sort.Search(len(fids), func(i int) bool {
> 		return fids[i] >= head.Fid
> ```

Search uses binary search to find and return the smallest index i in [0, n) at which f(i) is true, assuming that on the range [0, n), f(i) == true implies f(i+1) == true. That is, Search requires that f is false for some (possibly empty) prefix of the input range [0, n) and then true for the (possibly empty) remainder;

---


*Comments from [Reviewable](https://reviewable.io:443/reviews/dgraph-io/badger/365)*
<!-- Sent from Reviewable.io -->

janardhan1993: What should be the behaviour of iterator with AllVersions ? Say we write key at version1 and then delete at version 2, now if we use iterator with all versions it gives only key at version 1. 
This is affecting the purge, since purgeolderversions sees only one version of key it doesn't delete it.

---

Review status: 0 of 1 files reviewed at latest revision, 1 unresolved discussion, some commit checks broke.

---



*Comments from [Reviewable](https://reviewable.io:443/reviews/dgraph-io/badger/365#-:-L0gYkCkKiA_M2gqZxDg:buvu7yo)*
<!-- Sent from Reviewable.io -->

manishrjain: 



Review status: 0 of 1 files reviewed at latest revision, 1 unresolved discussion, some commit checks broke.

---

*[value.go, line 846 at r1](https://reviewable.io:443/reviews/dgraph-io/badger/365#-L0gY5pswsvHIxzCszv6:-L0gb7xkw9fiAMVu1JYw:b-nplzaa) ([raw file](https://github.com/dgraph-io/badger/blob/ce2168f7a30ca3cc71463b08b18d6a21db72c76e/value.go#L846)):*
<details><summary><i>Previously, janardhan1993 (Janardhan Reddy) wrote…</i></summary><blockquote>

Search uses binary search to find and return the smallest index i in [0, n) at which f(i) is true, assuming that on the range [0, n), f(i) == true implies f(i+1) == true. That is, Search requires that f is false for some (possibly empty) prefix of the input range [0, n) and then true for the (possibly empty) remainder;
</blockquote></details>

Let's rewrite this logic. The idea is to find a value log which is < head.fid, and contains the largest amount of discardable data.

We can merge both, and do them as one step. Actually, I'm not sure if we still need the sort.Search anymore.

---


*Comments from [Reviewable](https://reviewable.io:443/reviews/dgraph-io/badger/365)*
<!-- Sent from Reviewable.io -->

janardhan1993: 



Review status: 0 of 6 files reviewed at latest revision, 1 unresolved discussion, some commit checks broke.

---

*[value.go, line 846 at r1](https://reviewable.io:443/reviews/dgraph-io/badger/365#-L0gY5pswsvHIxzCszv6:-L0hTeTMla6-fA0CAVEY:b-u4tko) ([raw file](https://github.com/dgraph-io/badger/blob/ce2168f7a30ca3cc71463b08b18d6a21db72c76e/value.go#L846)):*
<details><summary><i>Previously, manishrjain (Manish R Jain) wrote…</i></summary><blockquote>

Let's rewrite this logic. The idea is to find a value log which is < head.fid, and contains the largest amount of discardable data.

We can merge both, and do them as one step. Actually, I'm not sure if we still need the sort.Search anymore.
</blockquote></details>

in case of fallback we choose a file randomly below head. Then we would need this search.

---


*Comments from [Reviewable](https://reviewable.io:443/reviews/dgraph-io/badger/365)*
<!-- Sent from Reviewable.io -->

manishrjain: 



Review status: 0 of 6 files reviewed at latest revision, 1 unresolved discussion, some commit checks broke.

---

*[value.go, line 846 at r1](https://reviewable.io:443/reviews/dgraph-io/badger/365#-L0gY5pswsvHIxzCszv6:-L0hUn4Cbti3v_MFFyFR:b-hu450r) ([raw file](https://github.com/dgraph-io/badger/blob/ce2168f7a30ca3cc71463b08b18d6a21db72c76e/value.go#L846)):*
<details><summary><i>Previously, janardhan1993 (Janardhan Reddy) wrote…</i></summary><blockquote>

in case of fallback we choose a file randomly below head. Then we would need this search.
</blockquote></details>

Iterate over all, pick the first one. Still no need for this confusing search. Simplify until can't anymore.

---


*Comments from [Reviewable](https://reviewable.io:443/reviews/dgraph-io/badger/365)*
<!-- Sent from Reviewable.io -->

janardhan1993: 



Review status: 0 of 6 files reviewed at latest revision, 1 unresolved discussion.

---

*[value.go, line 846 at r1](https://reviewable.io:443/reviews/dgraph-io/badger/365#-L0gY5pswsvHIxzCszv6:-L0hYdEsNZ_K_-e7DJam:b-896fix) ([raw file](https://github.com/dgraph-io/badger/blob/ce2168f7a30ca3cc71463b08b18d6a21db72c76e/value.go#L846)):*
<details><summary><i>Previously, manishrjain (Manish R Jain) wrote…</i></summary><blockquote>

Iterate over all, pick the first one. Still no need for this confusing search. Simplify until can't anymore.
</blockquote></details>

Done.

---


*Comments from [Reviewable](https://reviewable.io:443/reviews/dgraph-io/badger/365)*
<!-- Sent from Reviewable.io -->

deepakjois: <img class="emoji" title=":lgtm:" alt=":lgtm:" align="absmiddle" src="https://reviewable.io/lgtm.png" height="20" width="61"/>

---

Reviewed 6 of 6 files at r2.
Review status: all files reviewed at latest revision, 1 unresolved discussion, some commit checks broke.

---



*Comments from [Reviewable](https://reviewable.io:443/reviews/dgraph-io/badger/365#-:-L0ha0xOWqOaZEepSDOW:bnfp4nl)*
<!-- Sent from Reviewable.io -->

manishrjain: <img class="emoji" title=":lgtm:" alt=":lgtm:" align="absmiddle" src="https://reviewable.io/lgtm.png" height="20" width="61"/>

---

Reviewed 6 of 6 files at r2.
Review status: all files reviewed at latest revision, all discussions resolved, some commit checks broke.

---



*Comments from [Reviewable](https://reviewable.io:443/reviews/dgraph-io/badger/365#-:-L0hdd8YtPLJ5TFYINrk:bnfp4nl)*
<!-- Sent from Reviewable.io -->


Patch:
diff --git a/db.go b/db.go
index 14d50f7..f03a5ca 100644
--- a/db.go
+++ b/db.go
@@ -919,6 +919,9 @@ func (db *DB) purgeVersionsBelow(txn *Txn, key []byte, ts uint64) error {
 		if !bytes.Equal(key, item.Key()) || item.Version() >= ts {
 			continue
 		}
+		if isDeletedOrExpired(item.meta, item.ExpiresAt()) {
+			continue
+		}
 
 		// Found an older version. Mark for deletion
 		entries = append(entries,
@@ -976,6 +979,9 @@ func (db *DB) PurgeOlderVersions() error {
 				lastKey = y.SafeCopy(lastKey, item.Key())
 				continue
 			}
+			if isDeletedOrExpired(item.meta, item.ExpiresAt()) {
+				continue
+			}
 			// Found an older version. Mark for deletion
 			e := &Entry{
 				Key:  y.KeyWithTs(lastKey, item.version),
diff --git a/iterator.go b/iterator.go
index 1938ec9..a7b44d6 100644
--- a/iterator.go
+++ b/iterator.go
@@ -333,14 +333,14 @@ func (it *Iterator) Next() {
 	}
 }
 
-func isDeletedOrExpired(vs y.ValueStruct) bool {
-	if vs.Meta&bitDelete > 0 {
+func isDeletedOrExpired(meta byte, expiresAt uint64) bool {
+	if meta&bitDelete > 0 {
 		return true
 	}
-	if vs.ExpiresAt == 0 {
+	if expiresAt == 0 {
 		return false
 	}
-	return vs.ExpiresAt <= uint64(time.Now().Unix())
+	return expiresAt <= uint64(time.Now().Unix())
 }
 
 // parseItem is a complex function because it needs to handle both forward and reverse iteration
@@ -375,11 +375,8 @@ func (it *Iterator) parseItem() bool {
 	}
 
 	if it.opt.AllVersions {
-		// First check if value has been expired.
-		if isDeletedOrExpired(mi.Value()) {
-			mi.Next()
-			return false
-		}
+		// Return deleted or expired values also, otherwise user can't figure out
+		// whether the key was deleted.
 		item := it.newItem()
 		it.fill(item)
 		setItem(item)
@@ -404,7 +401,8 @@ func (it *Iterator) parseItem() bool {
 
 FILL:
 	// If deleted, advance and return.
-	if isDeletedOrExpired(mi.Value()) {
+	vs := mi.Value()
+	if isDeletedOrExpired(vs.Meta, vs.ExpiresAt) {
 		mi.Next()
 		return false
 	}
diff --git a/transaction.go b/transaction.go
index 7617b3b..ac9d1a2 100644
--- a/transaction.go
+++ b/transaction.go
@@ -377,10 +377,7 @@ func (txn *Txn) Get(key []byte) (item *Item, rerr error) {
 	item = new(Item)
 	if txn.update {
 		if e, has := txn.pendingWrites[string(key)]; has && bytes.Equal(key, e.Key) {
-			if e.meta&bitDelete > 0 {
-				return nil, ErrKeyNotFound
-			}
-			if e.ExpiresAt > 0 && e.ExpiresAt <= uint64(time.Now().Unix()) {
+			if isDeletedOrExpired(e.meta, e.ExpiresAt) {
 				return nil, ErrKeyNotFound
 			}
 			// Fulfill from cache.
@@ -407,7 +404,7 @@ func (txn *Txn) Get(key []byte) (item *Item, rerr error) {
 	if vs.Value == nil && vs.Meta == 0 {
 		return nil, ErrKeyNotFound
 	}
-	if isDeletedOrExpired(vs) {
+	if isDeletedOrExpired(vs.Meta, vs.ExpiresAt) {
 		return nil, ErrKeyNotFound
 	}
 
diff --git a/value.go b/value.go
index 1108010..93331d6 100644
--- a/value.go
+++ b/value.go
@@ -842,23 +842,18 @@ func (vlog *valueLog) pickLog(head valuePointer) *logFile {
 		return nil
 	}
 
-	i := sort.Search(len(fids), func(i int) bool {
-		return fids[i] == head.Fid
-	})
-	if i == len(fids) {
-		return nil
-	}
-
 	// Pick a candidate that contains the largest amount of discardable data
 	candidate := struct {
 		fid     uint32
 		discard int64
 	}{math.MaxUint32, 0}
 	vlog.lfDiscardStats.Lock()
-	for j := 0; j < i; j++ {
-		fid := fids[j]
+	for _, fid := range fids {
+		if fid >= head.Fid {
+			break
+		}
 		if vlog.lfDiscardStats.m[fid] > candidate.discard {
-			candidate.fid = fids[j]
+			candidate.fid = fid
 			candidate.discard = vlog.lfDiscardStats.m[fid]
 		}
 	}
@@ -869,7 +864,17 @@ func (vlog *valueLog) pickLog(head valuePointer) *logFile {
 	}
 
 	// Fallback to randomly picking a log file
-	idx := rand.Intn(i) // Don’t include head.Fid. We pick a random file before it.
+	var idxHead int
+	for i, fid := range fids {
+		if fid == head.Fid {
+			idxHead = i
+			break
+		}
+	}
+	if idxHead == 0 { // Not found or first file
+		return nil
+	}
+	idx := rand.Intn(idxHead) // Don’t include head.Fid. We pick a random file before it.
 	if idx > 0 {
 		idx = rand.Intn(idx + 1) // Another level of rand to favor smaller fids.
 	}
@@ -881,7 +886,7 @@ func discardEntry(e Entry, vs y.ValueStruct) bool {
 		// Version not found. Discard.
 		return true
 	}
-	if isDeletedOrExpired(vs) {
+	if isDeletedOrExpired(vs.Meta, vs.ExpiresAt) {
 		return true
 	}
 	if (vs.Meta & bitValuePointer) == 0 {


Test patch:
diff --git a/db_test.go b/db_test.go
index f99878e..effd781 100644
--- a/db_test.go
+++ b/db_test.go
@@ -907,16 +907,18 @@ func TestPurgeVersionsBelow(t *testing.T) {
 		require.NoError(t, err)
 		require.NotEmpty(t, db.vlog.lfDiscardStats.m)
 
-		// Verify that there are only 2 versions left
+		// Verify that there are only 2 versions left, and versions
+		// below ts have been deleted.
 		db.View(func(txn *Txn) error {
 			it := txn.NewIterator(opts)
 			var count int
 			for it.Rewind(); it.Valid(); it.Next() {
-				count++
 				item := it.Item()
-				require.True(t, item.Version() >= ts,
-					"item version: %d older than ts: %d",
-					item.Version(), ts)
+				if item.Version() < ts {
+					require.True(t, item.meta&bitDelete > 0)
+				} else {
+					count++
+				}
 				require.Equal(t, []byte("answer"), item.Key())
 			}
 			require.Equal(t, 2, count)
@@ -925,6 +927,64 @@ func TestPurgeVersionsBelow(t *testing.T) {
 	})
 }
 
+func TestPurgeVersionsBelow2(t *testing.T) {
+	runBadgerTest(t, nil, func(t *testing.T, db *DB) {
+		// Do a set and delete of same key
+		err := db.Update(func(txn *Txn) error {
+			return txn.Set([]byte("key"), []byte("value"))
+		})
+		require.NoError(t, err)
+
+		err = db.Update(func(txn *Txn) error {
+			return txn.Delete([]byte("key"))
+		})
+		require.NoError(t, err)
+
+		opts := DefaultIteratorOptions
+		opts.AllVersions = true
+		opts.PrefetchValues = false
+		// Verify that there are 2 versions and record highest version
+		var ts uint64
+		db.View(func(txn *Txn) error {
+			it := txn.NewIterator(opts)
+			var count int
+			for it.Rewind(); it.Valid(); it.Next() {
+				count++
+				item := it.Item()
+				require.Equal(t, []byte("key"), item.Key())
+				if count == 1 {
+					ts = item.Version()
+					require.True(t, item.meta&bitDelete > 0)
+					continue
+				}
+				val, err := item.Value()
+				require.NoError(t, err)
+				require.Equal(t, val, []byte("value"))
+			}
+			require.Equal(t, 2, count)
+			return nil
+		})
+
+		// Delete all versions
+		err = db.PurgeVersionsBelow([]byte("key"), ts)
+		require.NoError(t, err)
+
+		// Verify everything has been deleted
+		db.View(func(txn *Txn) error {
+			it := txn.NewIterator(opts)
+			var count int
+			for it.Rewind(); it.Valid(); it.Next() {
+				item := it.Item()
+				require.Equal(t, []byte("key"), item.Key())
+				require.True(t, item.meta&bitDelete > 0)
+				count++
+			}
+			require.Equal(t, 2, count)
+			return nil
+		})
+	})
+}
+
 func TestPurgeOlderVersions(t *testing.T) {
 	runBadgerTest(t, nil, func(t *testing.T, db *DB) {
 		// Write two versions of a key
@@ -960,7 +1020,7 @@ func TestPurgeOlderVersions(t *testing.T) {
 		err = db.PurgeOlderVersions()
 		require.NoError(t, err)
 
-		// Verify that only one version is found
+		// Verify that only one non-deleted version is found
 		err = db.View(func(txn *Txn) error {
 			it := txn.NewIterator(opts)
 			var count int
@@ -968,15 +1028,74 @@ func TestPurgeOlderVersions(t *testing.T) {
 				count++
 				item := it.Item()
 				require.Equal(t, []byte("answer"), item.Key())
+				if count == 1 {
+					val, err := item.Value()
+					require.NoError(t, err)
+					require.Equal(t, []byte("43"), val)
+				} else {
+					require.True(t, item.meta&bitDelete > 0)
+				}
+			}
+			require.Equal(t, 2, count)
+			return nil
+		})
+		require.NoError(t, err)
+	})
+}
+
+func TestPurgeOlderVersions2(t *testing.T) {
+	runBadgerTest(t, nil, func(t *testing.T, db *DB) {
+		// Do a set and delete of same key
+		err := db.Update(func(txn *Txn) error {
+			return txn.Set([]byte("key"), []byte("value"))
+		})
+		require.NoError(t, err)
+
+		err = db.Update(func(txn *Txn) error {
+			return txn.Delete([]byte("key"))
+		})
+		require.NoError(t, err)
+
+		opts := DefaultIteratorOptions
+		opts.AllVersions = true
+		opts.PrefetchValues = false
+		// Verify that there are 2 versions
+		db.View(func(txn *Txn) error {
+			it := txn.NewIterator(opts)
+			var count int
+			for it.Rewind(); it.Valid(); it.Next() {
+				count++
+				item := it.Item()
+				require.Equal(t, []byte("key"), item.Key())
+				if count == 1 {
+					require.True(t, item.meta&bitDelete > 0)
+					continue
+				}
 				val, err := item.Value()
 				require.NoError(t, err)
-				t.Logf("Item value is %q", val)
-				//require.Equal(t, []byte("43"), val)
+				require.Equal(t, val, []byte("value"))
 			}
-			require.Equal(t, 1, count)
+			require.Equal(t, 2, count)
 			return nil
 		})
+
+		// Delete all versions
+		err = db.PurgeOlderVersions()
 		require.NoError(t, err)
+
+		// Verify everything has been deleted
+		db.View(func(txn *Txn) error {
+			it := txn.NewIterator(opts)
+			var count int
+			for it.Rewind(); it.Valid(); it.Next() {
+				item := it.Item()
+				require.Equal(t, []byte("key"), item.Key())
+				require.True(t, item.meta&bitDelete > 0)
+				count++
+			}
+			require.Equal(t, 2, count)
+			return nil
+		})
 	})
 }
 
diff --git a/transaction_test.go b/transaction_test.go
index d88e0b8..6c5264c 100644
--- a/transaction_test.go
+++ b/transaction_test.go
@@ -570,7 +570,7 @@ func TestTxnIterationEdgeCase3(t *testing.T) {
 	})
 }
 
-func TestIteratorAllVersionsButDeleted(t *testing.T) {
+func TestIteratorAllVersionsWithDeleted(t *testing.T) {
 	runBadgerTest(t, nil, func(t *testing.T, db *DB) {
 		// Write two keys
 		err := db.Update(func(txn *Txn) error {
@@ -599,16 +599,63 @@ func TestIteratorAllVersionsButDeleted(t *testing.T) {
 		opts.AllVersions = true
 		opts.PrefetchValues = false
 
-		// Verify that deleted key does not show up when AllVersions is set.
+		// Verify that deleted shows up when AllVersions is set.
 		err = db.View(func(txn *Txn) error {
 			it := txn.NewIterator(opts)
 			var count int
 			for it.Rewind(); it.Valid(); it.Next() {
 				count++
 				item := it.Item()
-				require.Equal(t, []byte("answer2"), item.Key())
+				if count == 1 {
+					require.Equal(t, []byte("answer1"), item.Key())
+					require.True(t, item.meta&bitDelete > 0)
+				} else {
+					require.Equal(t, []byte("answer2"), item.Key())
+				}
+			}
+			require.Equal(t, 2, count)
+			return nil
+		})
+		require.NoError(t, err)
+	})
+}
+
+func TestIteratorAllVersionsWithDeleted2(t *testing.T) {
+	runBadgerTest(t, nil, func(t *testing.T, db *DB) {
+		// Set and delete alternatively
+		for i := 0; i < 4; i++ {
+			err := db.Update(func(txn *Txn) error {
+				if i%2 == 0 {
+					txn.Set([]byte("key"), []byte("value"))
+					return nil
+				}
+				txn.Delete([]byte("key"))
+				return nil
+			})
+			require.NoError(t, err)
+		}
+
+		opts := DefaultIteratorOptions
+		opts.AllVersions = true
+		opts.PrefetchValues = false
+
+		// Verify that deleted shows up when AllVersions is set.
+		err := db.View(func(txn *Txn) error {
+			it := txn.NewIterator(opts)
+			var count int
+			for it.Rewind(); it.Valid(); it.Next() {
+				item := it.Item()
+				require.Equal(t, []byte("key"), item.Key())
+				if count%2 != 0 {
+					val, err := item.Value()
+					require.NoError(t, err)
+					require.Equal(t, val, []byte("value"))
+				} else {
+					require.True(t, item.meta&bitDelete > 0)
+				}
+				count++
 			}
-			require.Equal(t, 1, count)
+			require.Equal(t, 4, count)
 			return nil
 		})
 		require.NoError(t, err)