Update release/v3.2011 branch

README.md

@@ -54,11 +54,11 @@ For more details on our version naming schema please read [Choosing a version](#
 To start using Badger, install Go 1.12 or above. Badger v2 needs go modules. Run the following command to retrieve the library.
 
 ```sh
-$ go get github.com/dgraph-io/badger/v2
+$ go get github.com/dgraph-io/badger/v3
 ```
 This will retrieve the library.
 
-##### Note: Badger does not directly use CGO but it relies on https://github.com/DataDog/zstd for compression and it requires gcc/cgo. If you wish to use badger without gcc/cgo, you can run `CGO_ENABLED=0 go get github.com/dgraph-io/badger/v2` which will download badger without the support for ZSTD compression algorithm.
+##### Note: Badger does not directly use CGO but it relies on https://github.com/DataDog/zstd for compression and it requires gcc/cgo. If you wish to use badger without gcc/cgo, you can run `CGO_ENABLED=0 go get github.com/dgraph-io/badger/v3` which will download badger without the support for ZSTD compression algorithm.
 
 #### Installing Badger Command Line Tool
 
@@ -162,6 +162,7 @@ Below is a list of known projects that use Badger:
 * [emitter](https://github.com/emitter-io/emitter) - Scalable, low latency, distributed pub/sub broker with message storage, uses MQTT, gossip and badger.
 * [OctoSQL](https://github.com/cube2222/octosql) - Query tool that allows you to join, analyse and transform data from multiple databases using SQL.
 * [Dkron](https://dkron.io/) - Distributed, fault tolerant job scheduling system.
+* [smallstep/certificates](https://github.com/smallstep/certificates) - Step-ca is an online certificate authority for secure, automated certificate management.
 * [Sandglass](https://github.com/celrenheit/sandglass) - distributed, horizontally scalable, persistent, time sorted message queue.
 * [TalariaDB](https://github.com/grab/talaria) - Grab's Distributed, low latency time-series database.
 * [Sloop](https://github.com/salesforce/sloop) - Salesforce's Kubernetes History Visualization Project.

backup.go

@@ -23,8 +23,9 @@ import (
 	"encoding/binary"
 	"io"
 
-	"github.com/dgraph-io/badger/v2/pb"
-	"github.com/dgraph-io/badger/v2/y"
+	"github.com/dgraph-io/badger/v3/pb"
+	"github.com/dgraph-io/badger/v3/y"
+	"github.com/dgraph-io/ristretto/z"
 	"github.com/golang/protobuf/proto"
 )
 
@@ -61,6 +62,7 @@ func (db *DB) Backup(w io.Writer, since uint64) (uint64, error) {
 func (stream *Stream) Backup(w io.Writer, since uint64) (uint64, error) {
 	stream.KeyToList = func(key []byte, itr *Iterator) (*pb.KVList, error) {
 		list := &pb.KVList{}
+		a := itr.Alloc
 		for ; itr.Valid(); itr.Next() {
 			item := itr.Item()
 			if !bytes.Equal(item.Key(), key) {
@@ -76,7 +78,10 @@ func (stream *Stream) Backup(w io.Writer, since uint64) (uint64, error) {
 			if !item.IsDeletedOrExpired() {
 				// No need to copy value, if item is deleted or expired.
 				var err error
-				valCopy, err = item.ValueCopy(nil)
+				err = item.Value(func(val []byte) error {
+					valCopy = a.Copy(val)
+					return nil
+				})
 				if err != nil {
 					stream.db.opt.Errorf("Key [%x, %d]. Error while fetching value [%v]\n",
 						item.Key(), item.Version(), err)
@@ -86,13 +91,14 @@ func (stream *Stream) Backup(w io.Writer, since uint64) (uint64, error) {
 
 			// clear txn bits
 			meta := item.meta &^ (bitTxn | bitFinTxn)
-			kv := &pb.KV{
-				Key:       item.KeyCopy(nil),
+			kv := y.NewKV(a)
+			*kv = pb.KV{
+				Key:       a.Copy(item.Key()),
 				Value:     valCopy,
-				UserMeta:  []byte{item.UserMeta()},
+				UserMeta:  a.Copy([]byte{item.UserMeta()}),
 				Version:   item.Version(),
 				ExpiresAt: item.ExpiresAt(),
-				Meta:      []byte{meta},
+				Meta:      a.Copy([]byte{meta}),
 			}
 			list.Kv = append(list.Kv, kv)
 
@@ -115,7 +121,11 @@ func (stream *Stream) Backup(w io.Writer, since uint64) (uint64, error) {
 	}
 
 	var maxVersion uint64
-	stream.Send = func(list *pb.KVList) error {
+	stream.Send = func(buf *z.Buffer) error {
+		list, err := BufferToKVList(buf)
+		if err != nil {
+			return err
+		}
 		out := list.Kv[:0]
 		for _, kv := range list.Kv {
 			if maxVersion < kv.Version {

backup_test.go

@@ -28,7 +28,7 @@ import (
 	"testing"
 	"time"
 
-	"github.com/dgraph-io/badger/v2/pb"
+	"github.com/dgraph-io/badger/v3/pb"
 	"github.com/stretchr/testify/require"
 )
 

badger/cmd/backup.go

@@ -21,7 +21,7 @@ import (
 	"math"
 	"os"
 
-	"github.com/dgraph-io/badger/v2"
+	"github.com/dgraph-io/badger/v3"
 	"github.com/spf13/cobra"
 )
 

badger/cmd/bank.go

@@ -30,9 +30,10 @@ import (
 	"sync/atomic"
 	"time"
 
-	"github.com/dgraph-io/badger/v2"
-	"github.com/dgraph-io/badger/v2/pb"
-	"github.com/dgraph-io/badger/v2/y"
+	"github.com/dgraph-io/badger/v3"
+	"github.com/dgraph-io/badger/v3/pb"
+	"github.com/dgraph-io/badger/v3/y"
+	"github.com/dgraph-io/ristretto/z"
 	"github.com/spf13/cobra"
 )
 
@@ -359,7 +360,7 @@ func runTest(cmd *cobra.Command, args []string) error {
 		WithNumVersionsToKeep(int(math.MaxInt32)).
 		WithBlockCacheSize(1 << 30).
 		WithIndexCacheSize(1 << 30)
- 
+
 	if verbose {
 		opts = opts.WithLoggingLevel(badger.DEBUG)
 	}
@@ -498,13 +499,15 @@ func runTest(cmd *cobra.Command, args []string) error {
 				batch := tmpDb.NewWriteBatch()
 
 				stream := db.NewStream()
-				stream.Send = func(list *pb.KVList) error {
-					for _, kv := range list.Kv {
-						if err := batch.Set(kv.Key, kv.Value); err != nil {
+				stream.Send = func(buf *z.Buffer) error {
+					err := buf.SliceIterate(func(s []byte) error {
+						var kv pb.KV
+						if err := kv.Unmarshal(s); err != nil {
 							return err
 						}
-					}
-					return nil
+						return batch.Set(kv.Key, kv.Value)
+					})
+					return err
 				}
 				y.Check(stream.Orchestrate(context.Background()))
 				y.Check(batch.Flush())

badger/cmd/flatten.go

@@ -20,7 +20,7 @@ import (
 	"fmt"
 	"math"
 
-	"github.com/dgraph-io/badger/v2"
+	"github.com/dgraph-io/badger/v3"
 	"github.com/spf13/cobra"
 )
 

badger/cmd/info.go

@@ -30,9 +30,9 @@ import (
 
 	"github.com/pkg/errors"
 
-	"github.com/dgraph-io/badger/v2"
-	"github.com/dgraph-io/badger/v2/table"
-	"github.com/dgraph-io/badger/v2/y"
+	"github.com/dgraph-io/badger/v3"
+	"github.com/dgraph-io/badger/v3/table"
+	"github.com/dgraph-io/badger/v3/y"
 	humanize "github.com/dustin/go-humanize"
 	"github.com/spf13/cobra"
 )
@@ -99,6 +99,8 @@ func handleInfo(cmd *cobra.Command, args []string) error {
 	db, err := badger.Open(badger.DefaultOptions(sstDir).
 		WithValueDir(vlogDir).
 		WithReadOnly(opt.readOnly).
+		WithBlockCacheSize(100 << 20).
+		WithIndexCacheSize(200 << 20).
 		WithEncryptionKey([]byte(opt.encryptionKey)))
 	if err != nil {
 		return y.Wrap(err, "failed to open database")
@@ -434,7 +436,7 @@ func printInfo(dir, valueDir string) error {
 		totalSSTSize += sz
 	}
 
-	fmt.Printf("Total SST size: %8s\n", hbytes(totalSSTSize))
+	fmt.Printf("Total SST size: %10s\n", hbytes(totalSSTSize))
 	fmt.Printf("Value log size: %10s\n", hbytes(valueLogSize))
 	fmt.Println()
 	totalExtra := numExtra + numValueDirExtra

badger/cmd/read_bench.go

@@ -25,11 +25,12 @@ import (
 	"time"
 
 	humanize "github.com/dustin/go-humanize"
+	"github.com/pkg/errors"
 	"github.com/spf13/cobra"
 
-	"github.com/dgraph-io/badger/v2"
-	"github.com/dgraph-io/badger/v2/pb"
-	"github.com/dgraph-io/badger/v2/y"
+	"github.com/dgraph-io/badger/v3"
+	"github.com/dgraph-io/badger/v3/pb"
+	"github.com/dgraph-io/badger/v3/y"
 	"github.com/dgraph-io/ristretto/z"
 )
 
@@ -79,7 +80,7 @@ func init() {
 
 // Scan the whole database using the iterators
 func fullScanDB(db *badger.DB) {
-	txn := db.NewTransaction(false)
+	txn := db.NewTransactionAt(math.MaxUint64, false)
 	defer txn.Discard()
 
 	startTime = time.Now()
@@ -111,7 +112,7 @@ func readBench(cmd *cobra.Command, args []string) error {
 		WithBlockCacheSize(blockCacheSize << 20).
 		WithIndexCacheSize(indexCacheSize << 20)
 	fmt.Printf("Opening badger with options = %+v\n", opt)
-	db, err := badger.Open(opt)
+	db, err := badger.OpenManaged(opt)
 	if err != nil {
 		return y.Wrapf(err, "unable to open DB")
 	}
@@ -205,21 +206,30 @@ func getSampleKeys(db *badger.DB) ([][]byte, error) {
 		return l, nil
 	}
 
+	errStop := errors.Errorf("Stop iterating")
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	stream.Send = func(l *pb.KVList) error {
+	stream.Send = func(buf *z.Buffer) error {
 		if count >= sampleSize {
 			return nil
 		}
-		for _, kv := range l.Kv {
+		err := buf.SliceIterate(func(s []byte) error {
+			var kv pb.KV
+			if err := kv.Unmarshal(s); err != nil {
+				return err
+			}
 			keys = append(keys, kv.Key)
 			count++
 			if count >= sampleSize {
 				cancel()
-				return nil
+				return errStop
 			}
+			return nil
+		})
+		if err == errStop || err == nil {
+			return nil
 		}
-		return nil
+		return err
 	}
 
 	if err := stream.Orchestrate(ctx); err != nil && err != context.Canceled {

badger/cmd/restore.go

@@ -22,7 +22,7 @@ import (
 	"os"
 	"path"
 
-	"github.com/dgraph-io/badger/v2"
+	"github.com/dgraph-io/badger/v3"
 	"github.com/spf13/cobra"
 )
 

badger/cmd/rotate.go

@@ -21,7 +21,7 @@ import (
 	"os"
 	"time"
 
-	"github.com/dgraph-io/badger/v2"
+	"github.com/dgraph-io/badger/v3"
 
 	"github.com/spf13/cobra"
 )

badger/cmd/rotate_test.go

@@ -22,8 +22,8 @@ import (
 	"os"
 	"testing"
 
-	"github.com/dgraph-io/badger/v2"
-	"github.com/dgraph-io/badger/v2/y"
+	"github.com/dgraph-io/badger/v3"
+	"github.com/dgraph-io/badger/v3/y"
 	"github.com/stretchr/testify/require"
 )
 

badger/cmd/stream.go

@@ -22,9 +22,9 @@ import (
 	"math"
 	"os"
 
-	"github.com/dgraph-io/badger/v2"
-	"github.com/dgraph-io/badger/v2/options"
-	"github.com/dgraph-io/badger/v2/y"
+	"github.com/dgraph-io/badger/v3"
+	"github.com/dgraph-io/badger/v3/options"
+	"github.com/dgraph-io/badger/v3/y"
 	"github.com/pkg/errors"
 	"github.com/spf13/cobra"
 )
@@ -39,40 +39,29 @@ This command streams the contents of this DB into another DB with the given opti
 }
 
 var outDir string
+var outFile string
 var compressionType uint32
 
 func init() {
 	// TODO: Add more options.
 	RootCmd.AddCommand(streamCmd)
 	streamCmd.Flags().StringVarP(&outDir, "out", "o", "",
 		"Path to output DB. The directory should be empty.")
+	streamCmd.Flags().StringVarP(&outFile, "", "f", "",
+		"Run a backup to this file.")
 	streamCmd.Flags().BoolVarP(&readOnly, "read_only", "", true,
 		"Option to open input DB in read-only mode")
 	streamCmd.Flags().IntVarP(&numVersions, "num_versions", "", 0,
 		"Option to configure the maximum number of versions per key. "+
 			"Values <= 0 will be considered to have the max number of versions.")
-	streamCmd.Flags().Uint32VarP(&compressionType, "compression", "", 0,
+	streamCmd.Flags().Uint32VarP(&compressionType, "compression", "", 1,
 		"Option to configure the compression type in output DB. "+
 			"0 to disable, 1 for Snappy, and 2 for ZSTD.")
 	streamCmd.Flags().StringVarP(&keyPath, "encryption-key-file", "e", "",
 		"Path of the encryption key file.")
 }
 
 func stream(cmd *cobra.Command, args []string) error {
-	// Check that outDir doesn't exist or is empty.
-	if _, err := os.Stat(outDir); err == nil {
-		f, err := os.Open(outDir)
-		if err != nil {
-			return err
-		}
-		defer f.Close()
-
-		_, err = f.Readdirnames(1)
-		if err != io.EOF {
-			return errors.Errorf("cannot run stream tool on non-empty output directory %s", outDir)
-		}
-	}
-
 	// Options for input DB.
 	if numVersions <= 0 {
 		numVersions = math.MaxInt32
@@ -94,20 +83,44 @@ func stream(cmd *cobra.Command, args []string) error {
 		return errors.Errorf(
 			"compression value must be one of 0 (disabled), 1 (Snappy), or 2 (ZSTD)")
 	}
-	outOpt := inOpt.
-		WithDir(outDir).
-		WithValueDir(outDir).
-		WithNumVersionsToKeep(numVersions).
-		WithCompression(options.CompressionType(compressionType)).
-		WithReadOnly(false)
-
 	inDB, err := badger.OpenManaged(inOpt)
 	if err != nil {
 		return y.Wrapf(err, "cannot open DB at %s", sstDir)
 	}
 	defer inDB.Close()
 
-	err = inDB.StreamDB(outOpt)
+	stream := inDB.NewStreamAt(math.MaxUint64)
+
+	if len(outDir) > 0 {
+		if _, err := os.Stat(outDir); err == nil {
+			f, err := os.Open(outDir)
+			if err != nil {
+				return err
+			}
+			defer f.Close()
+
+			_, err = f.Readdirnames(1)
+			if err != io.EOF {
+				return errors.Errorf(
+					"cannot run stream tool on non-empty output directory %s", outDir)
+			}
+		}
+
+		stream.LogPrefix = "DB.Stream"
+		outOpt := inOpt.
+			WithDir(outDir).
+			WithValueDir(outDir).
+			WithNumVersionsToKeep(numVersions).
+			WithCompression(options.CompressionType(compressionType)).
+			WithReadOnly(false)
+		err = inDB.StreamDB(outOpt)
+
+	} else if len(outFile) > 0 {
+		stream.LogPrefix = "DB.Backup"
+		f, err := os.OpenFile(outFile, os.O_RDWR|os.O_CREATE, 0666)
+		y.Check(err)
+		_, err = stream.Backup(f, 0)
+	}
 	fmt.Println("Done.")
 	return err
 }