Add a truncate option to warn people about data loss.

backup.go

@@ -4,6 +4,7 @@ import (
 	"bufio"
 	"encoding/binary"
 	"io"
+	"log"
 	"sync"
 
 	"github.com/dgraph-io/badger/y"
@@ -44,7 +45,8 @@ func (db *DB) Backup(w io.Writer, since uint64) (uint64, error) {
 			}
 			val, err := item.Value()
 			if err != nil {
-				return err
+				log.Printf("Key [%x]. Error while fetching value [%v]\n", item.Key(), err)
+				continue
 			}
 
 			entry := &protos.KVPair{

cmd/badger/cmd/backup.go

@@ -24,6 +24,7 @@ import (
 )
 
 var backupFile string
+var truncate bool
 
 // backupCmd represents the backup command
 var backupCmd = &cobra.Command{
@@ -42,13 +43,16 @@ func init() {
 	RootCmd.AddCommand(backupCmd)
 	backupCmd.Flags().StringVarP(&backupFile, "backup-file", "f",
 		"badger.bak", "File to backup to")
+	backupCmd.Flags().BoolVarP(&truncate, "truncate", "t",
+		false, "Allow value log truncation if required.")
 }
 
 func doBackup(cmd *cobra.Command, args []string) error {
 	// Open DB
 	opts := badger.DefaultOptions
 	opts.Dir = sstDir
 	opts.ValueDir = vlogDir
+	opts.Truncate = truncate
 	db, err := badger.Open(opts)
 	if err != nil {
 		return err

db.go

@@ -167,6 +167,11 @@ func Open(opt Options) (db *DB, err error) {
 	opt.maxBatchSize = (15 * opt.MaxTableSize) / 100
 	opt.maxBatchCount = opt.maxBatchSize / int64(skl.MaxNodeSize)
 
+	if opt.ReadOnly {
+		// Can't truncate if the DB is read only.
+		opt.Truncate = false
+	}
+
 	for _, path := range []string{opt.Dir, opt.ValueDir} {
 		dirExists, err := exists(path)
 		if err != nil {

errors.go

@@ -88,6 +88,10 @@ var (
 
 	// ErrWindowsNotSupported is returned when opt.ReadOnly is used on Windows
 	ErrWindowsNotSupported = errors.New("Read-only mode is not supported on Windows")
+
+	// ErrTruncateNeeded is returned when the value log gets corrupt, and requires truncation of
+	// corrupt data to allow Badger to run properly.
+	ErrTruncateNeeded = errors.New("Data corruption detected. Value log truncate required to run DB. This would result in data loss.")
 )
 
 // Key length can't be more than uint16, as determined by table::header.

options.go

@@ -92,6 +92,9 @@ type Options struct {
 	// before and has vlog data to be replayed, ReadOnly will cause Open
 	// to fail with an appropriate message.
 	ReadOnly bool
+
+	// Truncate value log to delete corrupt data, if any. Would not truncate if ReadOnly is set.
+	Truncate bool
 }
 
 // DefaultOptions sets a list of recommended options for good performance.
@@ -115,4 +118,5 @@ var DefaultOptions = Options{
 	// MemoryMap to mmap() the value log files
 	ValueLogFileSize: 1 << 30,
 	ValueThreshold:   20,
+	Truncate:         false,
 }

value.go

@@ -180,7 +180,7 @@ type logEntry func(e Entry, vp valuePointer) error
 
 // iterate iterates over log file. It doesn't not allocate new memory for every kv pair.
 // Therefore, the kv pair is only valid for the duration of fn call.
-func (vlog *valueLog) iterate(lf *logFile, offset uint32, readOnly bool, fn logEntry) error {
+func (vlog *valueLog) iterate(lf *logFile, offset uint32, fn logEntry) error {
 	_, err := lf.fd.Seek(int64(offset), io.SeekStart)
 	if err != nil {
 		return y.Wrap(err)
@@ -299,7 +299,7 @@ func (vlog *valueLog) iterate(lf *logFile, offset uint32, readOnly bool, fn logE
 			}
 		}
 
-		if readOnly {
+		if vlog.opt.ReadOnly {
 			return ErrReplayNeeded
 		}
 		if err := fn(e, vp); err != nil {
@@ -310,11 +310,13 @@ func (vlog *valueLog) iterate(lf *logFile, offset uint32, readOnly bool, fn logE
 		}
 	}
 
-	if !readOnly && truncate && len(lf.fmap) == 0 {
+	if vlog.opt.Truncate && truncate && len(lf.fmap) == 0 {
 		// Only truncate if the file isn't mmaped. Otherwise, Windows would puke.
 		if err := lf.fd.Truncate(int64(validEndOffset)); err != nil {
 			return err
 		}
+	} else if truncate {
+		return ErrTruncateNeeded
 	}
 
 	return nil
@@ -385,7 +387,7 @@ func (vlog *valueLog) rewrite(f *logFile) error {
 		return nil
 	}
 
-	err := vlog.iterate(f, 0, false, func(e Entry, vp valuePointer) error {
+	err := vlog.iterate(f, 0, func(e Entry, vp valuePointer) error {
 		return fe(e)
 	})
 	if err != nil {
@@ -685,7 +687,7 @@ func (vlog *valueLog) Replay(ptr valuePointer, fn logEntry) error {
 			of = 0
 		}
 		f := vlog.filesMap[id]
-		err := vlog.iterate(f, of, vlog.opt.ReadOnly, fn)
+		err := vlog.iterate(f, of, fn)
 		if err != nil {
 			return errors.Wrapf(err, "Unable to replay value log: %q", f.path)
 		}
@@ -983,7 +985,7 @@ func (vlog *valueLog) doRunGC(gcThreshold float64, head valuePointer) (err error
 	start := time.Now()
 	y.AssertTrue(vlog.kv != nil)
 	s := new(y.Slice)
-	err = vlog.iterate(lf, 0, false, func(e Entry, vp valuePointer) error {
+	err = vlog.iterate(lf, 0, func(e Entry, vp valuePointer) error {
 		esz := float64(vp.Len) / (1 << 20) // in MBs. +4 for the CAS stuff.
 		skipped += esz
 		if skipped < skipFirstM {

value_test.go

@@ -357,6 +357,7 @@ func TestChecksums(t *testing.T) {
 
 	// Set up SST with K1=V1
 	opts := getTestOptions(dir)
+	opts.Truncate = true
 	opts.ValueLogFileSize = 100 * 1024 * 1024 // 100Mb
 	kv, err := Open(opts)
 	require.NoError(t, err)
@@ -439,6 +440,7 @@ func TestPartialAppendToValueLog(t *testing.T) {
 
 	// Create skeleton files.
 	opts := getTestOptions(dir)
+	opts.Truncate = true
 	opts.ValueLogFileSize = 100 * 1024 * 1024 // 100Mb
 	kv, err := Open(opts)
 	require.NoError(t, err)