added single get call

Author: darkcoderrises

posting/list_test.go

@@ -435,6 +435,41 @@ func TestAddMutation_mrjn1(t *testing.T) {
 	require.Equal(t, 0, ol.Length(txn.StartTs, 0))
 }
 
+func TestReadSingleValue(t *testing.T) {
+	defer setMaxListSize(maxListSize)
+	maxListSize = math.MaxInt32
+
+	// We call pl.Iterate and then stop iterating in the first loop when we are reading
+	// single values. This test confirms that the two functions, getFirst from this file
+	// and GetSingeValueForKey works without an issue.
+
+	key := x.DataKey(x.GalaxyAttr("value"), 1240)
+	ol, err := getNew(key, ps, math.MaxUint64)
+	require.NoError(t, err)
+	N := int(1e2)
+	for i := 2; i <= N; i += 2 {
+		edge := &pb.DirectedEdge{
+			Value: []byte("ho hey there" + strconv.Itoa(i)),
+		}
+		txn := Txn{StartTs: uint64(i)}
+		addMutationHelper(t, ol, edge, Set, &txn)
+		require.NoError(t, ol.commitMutation(uint64(i), uint64(i)+1))
+		kData := ol.getMutation(uint64(i))
+		writer := NewTxnWriter(pstore)
+		if err := writer.SetAt(key, kData, BitDeltaPosting, uint64(i)); err != nil {
+			require.NoError(t, err)
+		}
+		writer.Flush()
+
+		for j := 3; j < i+6; j++ {
+			k, err, _ := GetSingleValueForKey(key, uint64(j))
+			require.NoError(t, err)
+			checkValue(t, ol, string(k.Postings[0].Value), uint64(j))
+		}
+
+	}
+}
+
 func TestRollupMaxTsIsSet(t *testing.T) {
 	defer setMaxListSize(maxListSize)
 	maxListSize = math.MaxInt32

posting/lists.go

@@ -195,6 +195,25 @@ func (lc *LocalCache) getInternal(key []byte, readFromDisk bool) (*List, error)
 	return lc.SetIfAbsent(skey, pl), nil
 }
 
+func (lc *LocalCache) GetSingleItem(key []byte) (*List, error) {
+	pl, err, _ := GetSingleValueForKey(key, lc.startTs)
+	if err != nil {
+		return nil, err
+	}
+
+	l := new(List)
+	l.key = key
+	l.plist = pl
+
+	lc.RLock()
+	if delta, ok := lc.deltas[string(key)]; ok && len(delta) > 0 {
+		l.setMutation(lc.startTs, delta)
+	}
+	lc.RUnlock()
+
+	return l, nil
+}
+
 // Get retrieves the cached version of the list associated with the given key.
 func (lc *LocalCache) Get(key []byte) (*List, error) {
 	return lc.getInternal(key, true)

posting/mvcc.go

@@ -457,6 +457,31 @@ func ReadPostingList(key []byte, it *badger.Iterator) (*List, error) {
 	return l, nil
 }
 
+func GetSingleValueForKey(key []byte, readTs uint64) (*pb.PostingList, error, int) {
+	//fmt.Println("KEY:", key)
+	txn := pstore.NewTransactionAt(readTs, false)
+	item, err := txn.Get(key)
+	if err != nil {
+		return nil, err, 0
+	}
+	pl := &pb.PostingList{}
+	k := 0
+
+	err = item.Value(func(val []byte) error {
+		k = len(key) + len(val)
+		if err := pl.Unmarshal(val); err != nil {
+			return err
+		}
+		return nil
+	})
+
+	if err != nil {
+		return nil, err, 0
+	}
+
+	return pl, nil, k
+}
+
 func getNew(key []byte, pstore *badger.DB, readTs uint64) (*List, error) {
 	cachedVal, ok := lCache.Get(key)
 	if ok {

worker/task.go

@@ -391,49 +391,68 @@ func (qs *queryState) handleValuePostings(ctx context.Context, args funcArgs) er
 			key := x.DataKey(q.Attr, q.UidList.Uids[i])
 
 			// Get or create the posting list for an entity, attribute combination.
-			pl, err := qs.cache.Get(key)
-			if err != nil {
-				return err
-			}
+			var pl *posting.List
+			pickMultiplePostings := q.ExpandAll || (listType && len(q.Langs) == 0)
+
+			var vals []types.Val
+			fcs := &pb.FacetsList{FacetsList: make([]*pb.Facets, 0)} // TODO Figure out how it is stored
 
-			// If count is being requested, there is no need to populate value and facets matrix.
-			if q.DoCount {
-				count, err := countForValuePostings(args, pl, facetsTree, listType)
-				if err != nil && err != posting.ErrNoValue {
+			if pickMultiplePostings {
+				pl, err = qs.cache.Get(key)
+				if err != nil {
 					return err
 				}
-				out.Counts = append(out.Counts, uint32(count))
-				// Add an empty UID list to make later processing consistent.
-				out.UidMatrix = append(out.UidMatrix, &pb.List{})
-				continue
-			}
+				// If count is being requested, there is no need to populate value and facets matrix.
+				if q.DoCount {
+					count, err := countForValuePostings(args, pl, facetsTree, listType)
+					if err != nil && err != posting.ErrNoValue {
+						return err
+					}
+					out.Counts = append(out.Counts, uint32(count))
+					// Add an empty UID list to make later processing consistent.
+					out.UidMatrix = append(out.UidMatrix, &pb.List{})
+					continue
+				}
 
-			vals, fcs, err := retrieveValuesAndFacets(args, pl, facetsTree, listType)
-			switch {
-			case err == posting.ErrNoValue || (err == nil && len(vals) == 0):
-				// This branch is taken when the value does not exist in the pl or
-				// the number of values retrieved is zero (there could still be facets).
-				// We add empty lists to the UidMatrix, FaceMatrix, ValueMatrix and
-				// LangMatrix so that all these data structure have predictable layouts.
-				out.UidMatrix = append(out.UidMatrix, &pb.List{})
-				out.FacetMatrix = append(out.FacetMatrix, &pb.FacetsList{})
-				out.ValueMatrix = append(out.ValueMatrix,
-					&pb.ValueList{Values: []*pb.TaskValue{}})
 				if q.ExpandAll {
-					// To keep the cardinality same as that of ValueMatrix.
-					out.LangMatrix = append(out.LangMatrix, &pb.LangList{})
+					langTags, err := pl.GetLangTags(args.q.ReadTs)
+					if err != nil {
+						return err
+					}
+					out.LangMatrix = append(out.LangMatrix, &pb.LangList{Lang: langTags})
+				}
+
+				vals, fcs, err = retrieveValuesAndFacets(args, pl, facetsTree, listType)
+				switch {
+				case err == posting.ErrNoValue || (err == nil && len(vals) == 0):
+					// This branch is taken when the value does not exist in the pl or
+					// the number of values retrieved is zero (there could still be facets).
+					// We add empty lists to the UidMatrix, FaceMatrix, ValueMatrix and
+					// LangMatrix so that all these data structure have predictable layouts.
+					out.UidMatrix = append(out.UidMatrix, &pb.List{})
+					out.FacetMatrix = append(out.FacetMatrix, &pb.FacetsList{})
+					out.ValueMatrix = append(out.ValueMatrix,
+						&pb.ValueList{Values: []*pb.TaskValue{}})
+					if q.ExpandAll {
+						// To keep the cardinality same as that of ValueMatrix.
+						out.LangMatrix = append(out.LangMatrix, &pb.LangList{})
+					}
+					continue
+				case err != nil:
+					return err
+				}
+
+			} else {
+				pl, err = qs.cache.GetSingleItem(key)
+				if err != nil {
+					return err
 				}
-				continue
-			case err != nil:
-				return err
-			}
 
-			if q.ExpandAll {
-				langTags, err := pl.GetLangTags(args.q.ReadTs)
+				val, err := pl.AllValues(q.ReadTs)
 				if err != nil {
 					return err
 				}
-				out.LangMatrix = append(out.LangMatrix, &pb.LangList{Lang: langTags})
+				vals = append(vals, val...)
 			}
 
 			uidList := new(pb.List)