fix(rollups): cherry-pick-7609 (#8297)

* cherry-pick-7609 Fix(rollups): Fix splits in roll-up (#7609) Fix roll-up split generation by fixing the range of UIDs to keep in the first and second part of the split. * resolve conflict * resolve more conflicts * resolve more conflicts * remove newline * remove whitespace Co-authored-by: Ahsan Barkati <[email protected]>
hypermodeinc · Sep 26, 2022 · a26c125 · a26c125
1 parent 0c6071b
commit a26c125
Show file tree

Hide file tree

Showing 2 changed files with 52 additions and 5 deletions.
diff --git a/posting/list.go b/posting/list.go
@@ -567,11 +567,11 @@ func (l *List) setMutation(startTs uint64, data []byte) {
 // The function will loop until either the posting List is fully iterated, or you return a false
 // in the provided function, which will indicate to the function to break out of the iteration.
 //
-// 	pl.Iterate(..., func(p *pb.posting) error {
-//    // Use posting p
-//    return nil // to continue iteration.
-//    return errStopIteration // to break iteration.
-//  })
+//		pl.Iterate(..., func(p *pb.posting) error {
+//	   // Use posting p
+//	   return nil // to continue iteration.
+//	   return errStopIteration // to break iteration.
+//	 })
 func (l *List) Iterate(readTs uint64, afterUid uint64, f func(obj *pb.Posting) error) error {
 	l.RLock()
 	defer l.RUnlock()

diff --git a/posting/list_test.go b/posting/list_test.go
@@ -973,6 +973,53 @@ func createAndDeleteMultiPartList(t *testing.T, size int) (*List, int) {
 	return ol, commits
 }
 
+func TestLargePlistSplit(t *testing.T) {
+	key := x.DataKey(uuid.New().String(), 1331)
+	ol, err := getNew(key, ps, math.MaxUint64)
+	require.NoError(t, err)
+	b := make([]byte, 30<<20)
+	rand.Read(b)
+	for i := 1; i <= 2; i++ {
+		edge := &pb.DirectedEdge{
+			ValueId: uint64(i),
+			Facets:  []*api.Facet{{Key: strconv.Itoa(i)}},
+			Value:   b,
+		}
+		txn := Txn{StartTs: uint64(i)}
+		addMutationHelper(t, ol, edge, Set, &txn)
+		require.NoError(t, ol.commitMutation(uint64(i), uint64(i)+1))
+	}
+	_, err = ol.Rollup(nil)
+	require.NoError(t, err)
+
+	ol, err = getNew(key, ps, math.MaxUint64)
+	require.NoError(t, err)
+	b = make([]byte, 10<<20)
+	rand.Read(b)
+	for i := 0; i < 63; i++ {
+		edge := &pb.DirectedEdge{
+			Entity:  uint64(1 << uint32(i)),
+			ValueId: uint64(i),
+			Facets:  []*api.Facet{{Key: strconv.Itoa(i)}},
+			Value:   b,
+		}
+		txn := Txn{StartTs: uint64(i)}
+		addMutationHelper(t, ol, edge, Set, &txn)
+		require.NoError(t, ol.commitMutation(uint64(i), uint64(i)+1))
+	}
+
+	kvs, err := ol.Rollup(nil)
+	require.NoError(t, err)
+	require.NoError(t, writePostingListToDisk(kvs))
+	ol, err = getNew(key, ps, math.MaxUint64)
+	require.NoError(t, err)
+	//require.Nil(t, ol.plist.Bitmap)
+	require.Equal(t, 0, len(ol.plist.Postings))
+	t.Logf("Num splits: %d\n", len(ol.plist.Splits))
+	require.True(t, len(ol.plist.Splits) > 0)
+	verifySplits(t, ol.plist.Splits)
+}
+
 func TestDeleteStarMultiPartList(t *testing.T) {
 	numEdges := 10000