fixed count bug

posting/index.go

@@ -522,8 +522,8 @@ func (txn *Txn) addMutationHelper(ctx context.Context, l *List, doUpdateIndex bo
 	switch {
 	case hasCountIndex:
 		countBefore, found, currPost = l.getPostingAndLengthNoSort(txn.StartTs, 0, getUID(t))
-		if countBefore == -1 {
-			return val, false, emptyCountParams, errors.Wrapf(ErrTsTooOld, "Add mutation count index")
+		if countBefore < 0 {
+			return val, false, emptyCountParams, errors.Wrapf(ErrTsTooOld, "Count less than. Add mutation count index")
 		}
 	case doUpdateIndex || delNonListPredicate:
 		found, currPost, err = l.findPosting(txn.StartTs, fingerprintEdge(t))
@@ -692,6 +692,7 @@ func (r *rebuilder) RunWithoutTemp(ctx context.Context) error {
 
 		l := new(List)
 		l.key = key
+		l.pk = pk
 		l.plist = new(pb.PostingList)
 
 		found := false

posting/list.go

@@ -77,6 +77,7 @@ const (
 type List struct {
 	x.SafeMutex
 	key         []byte
+	pk          x.ParsedKey
 	plist       *pb.PostingList
 	mutationMap *MutableLayer
 	minTs       uint64 // commit timestamp of immutable layer, reject reads before this ts.
@@ -751,14 +752,8 @@ func (l *List) updateMutationLayer(mpost *pb.Posting, singleUidUpdate bool) erro
 		newPlist := &pb.PostingList{}
 		newPlist.Postings = append(newPlist.Postings, mpost)
 
-		// Add the deletions in the existing plist because those postings are not picked
-		// up by iterating. Not doing so would result in delete operations that are not
-		// applied when the transaction is committed.
-		for _, post := range l.mutationMap.currentEntries.Postings {
-			if post.Op == Del && post.Uid != mpost.Uid {
-				newPlist.Postings = append(newPlist.Postings, post)
-			}
-		}
+		// Set current entries as nil so that they don't show up in the iterate.
+		l.mutationMap.setCurrentEntries(mpost.StartTs, &pb.PostingList{})
 
 		err := l.iterate(mpost.StartTs, 0, func(obj *pb.Posting) error {
 			// Ignore values which have the same uid as they will get replaced

posting/mvcc.go

@@ -20,6 +20,7 @@ import (
 	"bytes"
 	"context"
 	"encoding/hex"
+	"fmt"
 	"math"
 	"strconv"
 	"sync"
@@ -446,6 +447,7 @@ func initMemoryLayer(cacheSize int64, deleteOnUpdates bool) *MemoryLayer {
 			defer ticker.Stop()
 			for range ticker.C {
 				// Record the posting list cache hit ratio
+				fmt.Println(ml.cache.numCacheRead, ml.cache.numCacheReadFails, ml.cache.numCacheSave, ml.numDisksRead)
 				ostats.Record(context.Background(), x.PLCacheHitRatio.M(m.Ratio()))
 			}
 		}()
@@ -550,6 +552,7 @@ func ReadPostingList(key []byte, it *badger.Iterator) (*List, error) {
 	}
 
 	l := new(List)
+	l.pk = pk
 	l.key = key
 	l.plist = new(pb.PostingList)
 	l.minTs = 0
@@ -627,6 +630,7 @@ func copyList(l *List) *List {
 	l.AssertRLock()
 	// No need to clone the immutable layer or the key since mutations will not modify it.
 	lCopy := &List{
+		pk:    l.pk,
 		minTs: l.minTs,
 		maxTs: l.maxTs,
 		key:   l.key,

schema/parse.go

@@ -71,6 +71,9 @@ func parseDirective(it *lex.ItemIterator, schema *pb.SchemaUpdate, t types.TypeI
 		schema.Tokenizer = tokenizer
 		schema.IndexSpecs = vectorSpecs
 	case "count":
+		if !schema.List {
+			return next.Errorf("Cannot count for not list")
+		}
 		schema.Count = true
 	case "upsert":
 		schema.Upsert = true