Ensure bitValuePointer flag is cleared for LSM entry values written…

… to LSM (#1313)

When restoring a backup that was written with a lower `ValueThreshold`
to a DB instance with a higher `ValueThreshold`, some entry values
originally written to the value log in the backup DB will be written
to the LSM along with the key when restored to the new DB.

The `meta` field for those entries is not updated to reflect the
correct value storage state. Those entries still have the
`bitValuePointer` flag set.

That causes `iterator.prefetch` to fail while looking up the value
for those entries. This fix ensures the `meta` `bitValuePointer`
is cleared when entry values are stored in the LSM.

The original error can be reproduced by running the included
`TestLSMVPClear` test after reverting the line 687 `db.go` change
to clear the `bitValuePointer` flag.

backup_test.go

@@ -485,3 +485,58 @@ func TestBackupLoadIncremental(t *testing.T) {
 	})
 	require.NoError(t, err, "%v %v", updates, actual)
 }
+
+func TestBackupBitClear(t *testing.T) {
+	dir, err := ioutil.TempDir("", "badger-test")
+	require.NoError(t, err)
+	defer removeDir(dir)
+
+	opt := getTestOptions(dir)
+	opt.ValueThreshold = 10 // This is important
+	db, err := Open(opt)
+	require.NoError(t, err)
+
+	key := []byte("foo")
+	val := []byte(fmt.Sprintf("%0100d", 1))
+	require.Greater(t, len(val), db.opt.ValueThreshold)
+
+	err = db.Update(func(txn *Txn) error {
+		e := NewEntry(key, val)
+		// Value > valueTheshold so bitValuePointer will be set.
+		return txn.SetEntry(e)
+	})
+	require.NoError(t, err)
+
+	// Use different directory.
+	dir, err = ioutil.TempDir("", "badger-test")
+	require.NoError(t, err)
+	defer removeDir(dir)
+
+	bak, err := ioutil.TempFile(dir, "badgerbak")
+	require.NoError(t, err)
+	_, err = db.Backup(bak, 0)
+	require.NoError(t, err)
+	require.NoError(t, bak.Close())
+	require.NoError(t, db.Close())
+
+	opt = getTestOptions(dir)
+	opt.ValueThreshold = 200 // This is important.
+	db, err = Open(opt)
+	require.NoError(t, err)
+	defer db.Close()
+
+	bak, err = os.Open(bak.Name())
+	require.NoError(t, err)
+	defer bak.Close()
+
+	require.NoError(t, db.Load(bak, 16))
+
+	require.NoError(t, db.View(func(txn *Txn) error {
+		e, err := txn.Get(key)
+		require.NoError(t, err)
+		v, err := e.ValueCopy(nil)
+		require.NoError(t, err)
+		require.Equal(t, val, v)
+		return nil
+	}))
+}

db.go

@@ -679,8 +679,12 @@ func (db *DB) writeToLSM(b *request) error {
 		if db.shouldWriteValueToLSM(*entry) { // Will include deletion / tombstone case.
 			db.mt.Put(entry.Key,
 				y.ValueStruct{
-					Value:     entry.Value,
-					Meta:      entry.meta,
+					Value: entry.Value,
+					// Ensure value pointer flag is removed. Otherwise, the value will fail
+					// to be retrieved during iterator prefetch. `bitValuePointer` is only
+					// known to be set in write to LSM when the entry is loaded from a backup
+					// with lower ValueThreshold and its value was stored in the value log.
+					Meta:      entry.meta &^ bitValuePointer,
 					UserMeta:  entry.UserMeta,
 					ExpiresAt: entry.ExpiresAt,
 				})