Use the new Stream API from Badger

ee/backup/backup.go

@@ -16,6 +16,7 @@ import (
 	"context"
 
 	"github.com/dgraph-io/badger"
+	bpb "github.com/dgraph-io/badger/pb"
 	"github.com/dgraph-io/dgraph/posting"
 	"github.com/dgraph-io/dgraph/protos/pb"
 	"github.com/dgraph-io/dgraph/stream"
@@ -43,17 +44,17 @@ func (r *Request) Process(ctx context.Context) error {
 
 	sl := stream.Lists{Stream: w, DB: r.DB}
 	sl.ChooseKeyFunc = nil
-	sl.ItemToKVFunc = func(key []byte, itr *badger.Iterator) (*pb.KV, error) {
+	sl.ItemToKVFunc = func(key []byte, itr *badger.Iterator) (*bpb.KV, error) {
 		item := itr.Item()
 		pk := x.Parse(key)
 		if pk.IsSchema() {
 			val, err := item.ValueCopy(nil)
 			if err != nil {
 				return nil, err
 			}
-			kv := &pb.KV{
+			kv := &bpb.KV{
 				Key:      key,
-				Val:      val,
+				Value:    val,
 				UserMeta: []byte{item.UserMeta()},
 				Version:  item.Version(),
 			}

ee/backup/writer.go

@@ -18,6 +18,7 @@ import (
 	"net/url"
 	"strings"
 
+	bpb "github.com/dgraph-io/badger/pb"
 	"github.com/dgraph-io/dgraph/protos/pb"
 	"github.com/dgraph-io/dgraph/x"
 
@@ -97,7 +98,7 @@ func (w *writer) flush() error {
 
 // write uses the data length as delimiter.
 // XXX: we could use CRC for restore.
-func (w *writer) write(kv *pb.KV) error {
+func (w *writer) write(kv *bpb.KV) error {
 	if err := binary.Write(w.h, binary.LittleEndian, uint64(kv.Size())); err != nil {
 		return err
 	}

posting/index.go

@@ -562,7 +562,7 @@ func (r *rebuild) Run(ctx context.Context) error {
 		// We choose to write the PL at r.startTs, so it won't be read by txns,
 		// which occurred before this schema mutation. Typically, we use
 		// kv.Version as the timestamp.
-		if err = writer.SetAt(kv.Key, kv.Val, kv.UserMeta[0], r.startTs); err != nil {
+		if err = writer.SetAt(kv.Key, kv.Value, kv.UserMeta[0], r.startTs); err != nil {
 			return err
 		}
 		// This locking is just to catch any future issues.  We shouldn't need

posting/list.go

@@ -30,6 +30,7 @@ import (
 	"github.com/dgryski/go-farm"
 	"github.com/golang/glog"
 
+	bpb "github.com/dgraph-io/badger/pb"
 	"github.com/dgraph-io/dgo/protos/api"
 	"github.com/dgraph-io/dgo/y"
 	"github.com/dgraph-io/dgraph/algo"
@@ -639,19 +640,19 @@ func doAsyncWrite(commitTs uint64, key []byte, data []byte, meta byte, f func(er
 	}
 }
 
-func (l *List) MarshalToKv() (*pb.KV, error) {
+func (l *List) MarshalToKv() (*bpb.KV, error) {
 	l.Lock()
 	defer l.Unlock()
 	if err := l.rollup(math.MaxUint64); err != nil {
 		return nil, err
 	}
 
-	kv := &pb.KV{}
+	kv := &bpb.KV{}
 	kv.Version = l.minTs
 	kv.Key = l.key
 	val, meta := marshalPostingList(l.plist)
 	kv.UserMeta = []byte{meta}
-	kv.Val = val
+	kv.Value = val
 	return kv, nil
 }
 

protos/Makefile

@@ -17,6 +17,7 @@
 PROTO_PATH := ${GOPATH}/src:.
 PROTO_PATH := ${PROTO_PATH}:${GOPATH}/src/github.com/dgraph-io/dgraph
 PROTO_PATH := ${PROTO_PATH}:${GOPATH}/src/github.com/dgraph-io/dgo/protos
+PROTO_PATH := ${PROTO_PATH}:${GOPATH}/src/github.com/dgraph-io/badger/pb
 
 .PHONY: help
 help:

protos/pb.proto

@@ -25,6 +25,7 @@ syntax = "proto3";
 package pb;
 
 import "api.proto";
+import "github.com/dgraph-io/badger/pb/pb.proto";
 
 /* import "gogoproto/gogo.proto"; */
 
@@ -224,18 +225,11 @@ message Proposal {
 }
 
 message KVS {
- repeated KV kv = 1;
+ repeated pb.KV kv = 1;
  // done used to indicate if the stream of KVS is over.
  bool done      = 2;
 }
 
-message KV {
-	bytes key      = 1;
-	bytes val      = 2;
-	bytes userMeta = 3;
-	uint64 version = 4;
-}
-
 // Posting messages.
 message Posting {
 	fixed64 uid = 1;