Bring Badger v1.5.0 into Dgraph (dgraph-io#2383)

Bring Badger v1.5.0 in. Keep all versions in posting list store by default. This would get rid of the `Purge...` invocations, instead of using the `SetWithDiscard` function to automatically discard older versions of the key.

Address dgraph-io#2382.

dgraph/cmd/bulk/shuffle.go

@@ -81,7 +81,7 @@ func readMapOutput(filename string, mapEntryCh chan<- *intern.MapEntry) {
  x.Check(err)
  sz, n := binary.Uvarint(buf)
  if n <= 0 {
- log.Fatal("Could not read uvarint: %d", n)
+ log.Fatalf("Could not read uvarint: %d", n)
  }
  x.Check2(r.Discard(n))
 

edgraph/server.go

@@ -11,6 +11,7 @@ import (
  "bytes"
  "fmt"
  "log"
+ "math"
  "math/rand"
  "os"
  "sync"
@@ -114,6 +115,7 @@ func (s *ServerState) initStorage() {
  opt.SyncWrites = true
  opt.Dir = Config.PostingDir
  opt.ValueDir = Config.PostingDir
+ opt.NumVersionsToKeep = math.MaxInt32
  switch Config.PostingTables {
  case "memorymap":
  opt.TableLoadingMode = options.MemoryMap

gql/parser.go

@@ -697,7 +697,7 @@ func parseRecurseArgs(it *lex.ItemIterator, gq *GraphQuery) error {
  key = strings.ToLower(item.Val)
 
  if ok := trySkipItemTyp(it, itemColon); !ok {
- return fmt.Errorf("Expected colon(:) after %s")
+ return fmt.Errorf("Expected colon(:) after %s", key)
  }
 
  if item, ok = tryParseItemType(it, itemName); !ok {

posting/index.go

@@ -131,7 +131,7 @@ func (txn *Txn) addIndexMutation(ctx context.Context, edge *intern.DirectedEdge,
  }
  return err
  }
- x.PredicateStats.Add(fmt.Sprintf("i.%s", edge.Attr), 1)
+ x.PredicateStats.Add("i."+edge.Attr, 1)
  return nil
 }
 
@@ -441,15 +441,14 @@ func deleteEntries(prefix []byte, remove func(key []byte) bool) error {
 
  txn := pstore.NewTransactionAt(version, true)
  // Purge doesn't delete anything, so write an empty pl
- txn.SetWithMeta(nkey, nil, BitEmptyPosting)
+ txn.SetWithDiscard(nkey, nil, BitEmptyPosting)
  wg.Add(1)
  err := txn.CommitAt(version, func(e error) {
  defer wg.Done()
  if e != nil {
  setError(e)
  return
  }
- pstore.PurgeVersionsBelow(nkey, version)
  })
  txn.Discard()
  if err != nil {

posting/list.go

@@ -642,7 +642,7 @@ func (l *List) Length(readTs, afterUid uint64) int {
 func doAsyncWrite(commitTs uint64, key []byte, data []byte, meta byte, f func(error)) {
  txn := pstore.NewTransactionAt(commitTs, true)
  defer txn.Discard()
- if err := txn.SetWithMeta(key, data, meta); err != nil {
+ if err := txn.SetWithDiscard(key, data, meta); err != nil {
  f(err)
  }
  if err := txn.CommitAt(commitTs, f); err != nil {
@@ -814,7 +814,6 @@ func (l *List) syncIfDirty(delFromCache bool) (committed bool, err error) {
  x.AssertTrue(atomic.LoadInt32(&l.deleteMe) == 1)
  lcache.delete(l.key)
  }
- pstore.PurgeVersionsBelow(l.key, minTs)
  }
 
  doAsyncWrite(minTs, l.key, data, meta, f)

posting/mvcc.go

@@ -422,6 +422,9 @@ func ReadPostingList(key []byte, it *badger.Iterator) (*List, error) {
  } else {
  x.Fatalf("unexpected meta: %d", item.UserMeta())
  }
+ if item.DiscardEarlierVersions() {
+ break
+ }
  it.Next()
  }