Add mutation pipeline

Author: darkcoderrises

posting/index.go

@@ -139,19 +139,27 @@ func addMutation(t *testing.T, l *List, edge *pb.DirectedEdge, op uint32,
 	}
 	txn := Oracle().RegisterStartTs(startTs)
 	txn.cache.SetIfAbsent(string(l.key), l)
-	if index {
-		require.NoError(t, l.AddMutationWithIndex(context.Background(), edge, txn))
-	} else {
-		require.NoError(t, l.addMutation(context.Background(), txn, edge))
-	}
 
+	mp := NewMutationPipeline(txn)
+	err := mp.Process(context.Background(), []*pb.DirectedEdge{edge})
+	require.NoError(t, err)
 	txn.Update()
 	txn.UpdateCachedKeys(commitTs)
 	writer := NewTxnWriter(pstore)
 	require.NoError(t, txn.CommitToDisk(writer, commitTs))
 	require.NoError(t, writer.Flush())
+	newTxn := Oracle().RegisterStartTs(commitTs + 1)
+	l, err = newTxn.Get(l.key)
+	require.NoError(t, err)
 }
 
+const schemaPreVal = `
+	name: string .
+	name2: string .
+	dob: dateTime .
+	friend: [uid] .
+`
+
 const schemaVal = `
 name: string @index(term) .
 name2: string @index(term) .
@@ -263,6 +271,9 @@ func addEdgeToUID(t *testing.T, attr string, src uint64,
 func TestCountReverseIndexWithData(t *testing.T) {
 	require.NoError(t, pstore.DropAll())
 	MemLayerInstance.clear()
+	preIndex := "testcount: [uid] ."
+	require.NoError(t, schema.ParseBytes([]byte(preIndex), 1))
+
 	indexNameCountVal := "testcount: [uid] @count @reverse ."
 
 	attr := x.AttrInRootNamespace("testcount")
@@ -298,6 +309,9 @@ func TestCountReverseIndexWithData(t *testing.T) {
 func TestCountReverseIndexEmptyPosting(t *testing.T) {
 	require.NoError(t, pstore.DropAll())
 	MemLayerInstance.clear()
+	preIndex := "testcount: [uid] ."
+	require.NoError(t, schema.ParseBytes([]byte(preIndex), 1))
+
 	indexNameCountVal := "testcount: [uid] @count @reverse ."
 
 	attr := x.AttrInRootNamespace("testcount")
@@ -330,6 +344,8 @@ func TestCountReverseIndexEmptyPosting(t *testing.T) {
 }
 
 func TestRebuildTokIndex(t *testing.T) {
+	require.NoError(t, schema.ParseBytes([]byte(schemaPreVal), 1))
+
 	addEdgeToValue(t, x.AttrInRootNamespace("name2"), 91, "Michonne", uint64(1), uint64(2))
 	addEdgeToValue(t, x.AttrInRootNamespace("name2"), 92, "David", uint64(3), uint64(4))
 

posting/index_test.go

@@ -404,13 +404,18 @@ func (mm *MutableLayer) print() string {
 	if mm == nil {
 		return ""
 	}
-	return fmt.Sprintf("Committed List: %+v Proposed list: %+v Delete all marker: %d  \n",
+	return fmt.Sprintf("Committed List: %+v Proposed list: %+v Delete all marker: %d. Count: %d  \n",
 		mm.committedEntries,
 		mm.currentEntries,
-		mm.deleteAllMarker)
+		mm.deleteAllMarker,
+		mm.length)
 }
 
 func (l *List) Print() string {
+	if l.plist.Pack != nil {
+		uids := codec.Decode(l.plist.Pack, 0)
+		return fmt.Sprintf("minTs: %d, committed uids: %+v, mutationMap: %s", l.minTs, uids, l.mutationMap.print())
+	}
 	return fmt.Sprintf("minTs: %d, plist: %+v, mutationMap: %s", l.minTs, l.plist, l.mutationMap.print())
 }
 
@@ -712,6 +717,53 @@ type ListOptions struct {
 	First     int
 }
 
+func NewPostingExisting(p *pb.Posting, t *pb.DirectedEdge) {
+	var op uint32
+	switch t.Op {
+	case pb.DirectedEdge_SET:
+		op = Set
+	case pb.DirectedEdge_OVR:
+		op = Ovr
+	case pb.DirectedEdge_DEL:
+		op = Del
+	default:
+		x.Fatalf("Unhandled operation: %+v", t)
+	}
+
+	var postingType pb.Posting_PostingType
+	switch {
+	case len(t.Lang) > 0:
+		postingType = pb.Posting_VALUE_LANG
+	case t.ValueId == 0:
+		postingType = pb.Posting_VALUE
+	default:
+		postingType = pb.Posting_REF
+	}
+
+	p.Uid = t.ValueId
+	p.Value = t.Value
+	p.ValType = t.ValueType
+	p.PostingType = postingType
+	p.LangTag = []byte(t.Lang)
+	p.Op = op
+	p.Facets = t.Facets
+}
+
+func GetPostingOp(top uint32) pb.DirectedEdge_Op {
+	var op pb.DirectedEdge_Op
+	switch top {
+	case Set:
+		op = pb.DirectedEdge_SET
+	case Del:
+		op = pb.DirectedEdge_DEL
+	case Ovr:
+		op = pb.DirectedEdge_OVR
+	default:
+		x.Fatalf("Unhandled operation: %+v", top)
+	}
+	return op
+}
+
 // NewPosting takes the given edge and returns its equivalent representation as a posting.
 func NewPosting(t *pb.DirectedEdge) *pb.Posting {
 	var op uint32
@@ -789,12 +841,12 @@ func (l *List) updateMutationLayer(mpost *pb.Posting, singleUidUpdate, hasCountI
 		// The current value should be deleted in favor of this value. This needs to
 		// be done because the fingerprint for the value is not math.MaxUint64 as is
 		// the case with the rest of the scalar predicates.
-		newPlist := &pb.PostingList{}
+		newPlist := &pb.PostingList{
+			Postings: []*pb.Posting{createDeleteAllPosting()},
+		}
 		if mpost.Op != Del {
-			// If we are setting a new value then we can just delete all the older values.
-			newPlist.Postings = append(newPlist.Postings, createDeleteAllPosting())
+			newPlist.Postings = append(newPlist.Postings, mpost)
 		}
-		newPlist.Postings = append(newPlist.Postings, mpost)
 		l.mutationMap.setCurrentEntries(mpost.StartTs, newPlist)
 		return nil
 	}
@@ -833,6 +885,10 @@ func fingerprintEdge(t *pb.DirectedEdge) uint64 {
 	return id
 }
 
+func FingerprintEdge(t *pb.DirectedEdge) uint64 {
+	return fingerprintEdge(t)
+}
+
 func (l *List) addMutation(ctx context.Context, txn *Txn, t *pb.DirectedEdge) error {
 	l.Lock()
 	defer l.Unlock()
@@ -1110,6 +1166,13 @@ func (l *List) pickPostings(readTs uint64) (uint64, []*pb.Posting) {
 		}
 		return pi.Uid < pj.Uid
 	})
+
+	if len(posts) > 0 {
+		if hasDeleteAll(posts[0]) {
+			posts = posts[1:]
+		}
+	}
+
 	return deleteAllMarker, posts
 }
 
@@ -1258,6 +1321,11 @@ func (l *List) GetLength(readTs uint64) int {
 		length += immutLen
 	}
 
+	pureLength := l.length(readTs, 0)
+	if pureLength != length {
+		panic(fmt.Sprintf("pure length != length %d %d %s", pureLength, length, l.Print()))
+	}
+
 	return length
 }
 
@@ -2007,6 +2075,18 @@ func (l *List) findStaticValue(readTs uint64) *pb.PostingList {
 	if l.plist != nil && len(l.plist.Postings) > 0 {
 		return l.plist
 	}
+	if l.plist != nil && l.plist.Pack != nil {
+		uids := codec.Decode(l.plist.Pack, 0)
+		return &pb.PostingList{
+			Postings: []*pb.Posting{
+				{
+					Uid:     uids[0],
+					ValType: pb.Posting_UID,
+					Op:      Set,
+				},
+			},
+		}
+	}
 	return nil
 }
 

posting/list.go

@@ -124,7 +124,6 @@ func TestGetSinglePosting(t *testing.T) {
 
 	res, err := l.StaticValue(1)
 	require.NoError(t, err)
-	fmt.Println(res, res == nil)
 	require.Equal(t, res == nil, true)
 
 	l.plist = create_pl(1, 1)