perf(core): Implement parallel merge sorted algo (#9429)

```
goos: linux
goarch: amd64
pkg: github.com/hypermodeinc/dgraph/v25/algo
cpu: 11th Gen Intel(R) Core(TM) i7-1185G7 @ 3.00GHz
BenchmarkMergeSorted/MergeSorted-8         	       4	 349424640 ns/op	45692916 B/op	   20059 allocs/op
BenchmarkMergeSorted/MergeSortedParallel-8 	       9	 173713513 ns/op	124697028 B/op	   22970 allocs/op
BenchmarkMergeSorted/MergeSortedNew-8      	       8	 131558511 ns/op	52276942 B/op	   20236 allocs/op
```

Author: harshil-goel

algo/uidlist.go

@@ -8,6 +8,7 @@ package algo
 import (
 	"container/heap"
 	"sort"
+	"sync"
 
 	"github.com/hypermodeinc/dgraph/v25/codec"
 	"github.com/hypermodeinc/dgraph/v25/protos/pb"
@@ -360,41 +361,61 @@ func Difference(u, v *pb.List) *pb.List {
 	return &pb.List{Uids: out}
 }
 
-// MergeSorted merges sorted lists.
-func MergeSorted(lists []*pb.List) *pb.List {
+func MergeSortedMoreMem(lists []*pb.List) *pb.List {
+	numThreads := 10
+	if len(lists) > numThreads*numThreads {
+		k := numThreads
+		res := []*pb.List{}
+		var wg sync.WaitGroup
+		var mutex sync.Mutex
+		wg.Add(k)
+		for i := 0; i < k; i++ {
+			go func() {
+				defer wg.Done()
+				end := (i + 1) * len(lists) / k
+				if end > len(lists) {
+					end = len(lists)
+				}
+				result := MergeSortedMoreMem(lists[i*len(lists)/k : end])
+				mutex.Lock()
+				res = append(res, result)
+				mutex.Unlock()
+			}()
+		}
+		wg.Wait()
+		return MergeSortedMoreMem(res)
+	} else {
+		return internalMergeSort(lists)
+	}
+}
+
+func internalMergeSortWithBuffer(lists []*pb.List, buffer []uint64) *pb.List {
 	if len(lists) == 0 {
-		return new(pb.List)
+		return &pb.List{Uids: buffer[:0]}
 	}
 
 	h := &uint64Heap{}
 	heap.Init(h)
-	maxSz := 0
 
 	for i, l := range lists {
-		if l == nil {
+		if l == nil || len(l.Uids) == 0 {
 			continue
 		}
-		lenList := len(l.Uids)
-		if lenList > 0 {
-			heap.Push(h, elem{
-				val:     l.Uids[0],
-				listIdx: i,
-			})
-			if lenList > maxSz {
-				maxSz = lenList
-			}
-		}
+		heap.Push(h, elem{
+			val:     l.Uids[0],
+			listIdx: i,
+		})
 	}
 
-	// Our final output. Give it an approximate capacity as copies are expensive.
-	output := make([]uint64, 0, maxSz)
-	// idx[i] is the element we are looking at for lists[i].
+	// Use the provided buffer
+	output := buffer[:0]
 	idx := make([]int, len(lists))
-	var last uint64   // Last element added to sorted / final output.
-	for h.Len() > 0 { // While heap is not empty.
-		me := (*h)[0] // Peek at the top element in heap.
+	var last uint64
+
+	for h.Len() > 0 {
+		me := (*h)[0]
 		if len(output) == 0 || me.val != last {
-			output = append(output, me.val) // Add if unique.
+			output = append(output, me.val)
 			last = me.val
 		}
 		l := lists[me.listIdx]
@@ -404,12 +425,122 @@ func MergeSorted(lists []*pb.List) *pb.List {
 			idx[me.listIdx]++
 			val := l.Uids[idx[me.listIdx]]
 			(*h)[0].val = val
-			heap.Fix(h, 0) // Faster than Pop() followed by Push().
+			heap.Fix(h, 0)
 		}
 	}
+
 	return &pb.List{Uids: output}
 }
 
+// MergeSorted merges sorted lists.
+func internalMergeSort(lists []*pb.List) *pb.List {
+	sz := 0
+	for _, l := range lists {
+		if l == nil || len(l.Uids) == 0 {
+			continue
+		}
+		sz += len(l.Uids)
+	}
+	buffer := make([]uint64, 0, sz)
+	return internalMergeSortWithBuffer(lists, buffer)
+}
+
+func MergeSorted(lists []*pb.List) *pb.List {
+	// Calculate total capacity needed
+	totalCap := 0
+	for _, l := range lists {
+		if l != nil {
+			totalCap += len(l.Uids)
+		}
+	}
+
+	// Pre-allocate one big buffer
+	bigBuffer := make([]uint64, totalCap)
+	return mergeSortedWithBuffer(lists, bigBuffer)
+}
+
+const numThreads = 10
+const numListPerThread = 10
+
+func mergeSortedWithBuffer(lists []*pb.List, buffer []uint64) *pb.List {
+	if len(lists) < numThreads*numListPerThread {
+		return internalMergeSort(lists)
+	}
+
+	// Calculate how much buffer each goroutine needs
+	chunkSizes := make([]int, numThreads)
+	totalNeeded := 0
+
+	chunkSize := (len(lists) + numThreads - 1) / numThreads
+
+	for i := 0; i < numThreads; i++ {
+		start := i * chunkSize
+		end := (i + 1) * chunkSize
+		if end > len(lists) {
+			end = len(lists)
+		}
+
+		if start > len(lists) {
+			continue
+		}
+
+		chunkCap := 0
+		for j := start; j < end; j++ {
+			if lists[j] != nil {
+				chunkCap += len(lists[j].Uids)
+			}
+		}
+		chunkSizes[i] = chunkCap
+		totalNeeded += chunkCap
+	}
+
+	// Calculate buffer offsets for each goroutine
+	bufferOffsets := make([]int, numThreads)
+	bufferOffsets[0] = 0
+	for i := 1; i < numThreads; i++ {
+		bufferOffsets[i] = bufferOffsets[i-1] + chunkSizes[i-1]
+	}
+
+	// Distribute buffer slices to each goroutine
+	intermediateResults := make([]*pb.List, numThreads)
+	var wg sync.WaitGroup
+
+	wg.Add(numThreads)
+	for i := 0; i < numThreads; i++ {
+		go func(idx int) {
+			defer wg.Done()
+			start := idx * chunkSize
+			end := (idx + 1) * chunkSize
+			if end > len(lists) {
+				end = len(lists)
+			}
+			if start > len(lists) {
+				return
+			}
+
+			// Give this goroutine its slice of the big buffer
+			bufferStart := bufferOffsets[idx]
+			bufferEnd := bufferStart + chunkSizes[idx]
+			goroutineBuffer := buffer[bufferStart:bufferEnd:bufferEnd][:0]
+			result := internalMergeSortWithBuffer(lists[start:end], goroutineBuffer)
+			intermediateResults[idx] = result
+		}(i)
+	}
+	wg.Wait()
+
+	// Filter out nil results
+	validResults := make([]*pb.List, 0, numThreads)
+	for _, result := range intermediateResults {
+		if result != nil && len(result.Uids) > 0 {
+			validResults = append(validResults, result)
+		}
+	}
+
+	// Use the remaining part of buffer for final merge
+	finalBuffer := make([]uint64, 0, totalNeeded)
+	return internalMergeSortWithBuffer(validResults, finalBuffer)
+}
+
 // IndexOf performs a binary search on the uids slice and returns the index at
 // which it finds the uid, else returns -1
 func IndexOf(u *pb.List, uid uint64) int {

algo/uidlist_test.go

@@ -72,6 +72,60 @@ func TestMergeSorted6(t *testing.T) {
 		[]uint64{1, 2, 11, 12, 13, 14, 15, 16, 17, 18, 20, 25})
 }
 
+func BenchmarkMergeSorted(b *testing.B) {
+	createList := func(n int) *pb.List {
+		list := make([]uint64, n)
+		for i := range list {
+			list[i] = uint64(rand.Int63())
+		}
+		sort.Slice(list, func(i, j int) bool {
+			return list[i] < list[j]
+		})
+		return &pb.List{Uids: list}
+	}
+
+	input := []*pb.List{}
+	for i := 0; i < 10000; i++ {
+		input = append(input, createList(100))
+	}
+
+	b.Run("MergeSorted", func(b *testing.B) {
+		for i := 0; i < b.N; i++ {
+			internalMergeSort(input)
+		}
+	})
+	b.Run("MergeSortedParallel", func(b *testing.B) {
+		for i := 0; i < b.N; i++ {
+			MergeSortedMoreMem(input)
+		}
+	})
+	b.Run("MergeSortedNew", func(b *testing.B) {
+		for i := 0; i < b.N; i++ {
+			MergeSorted(input)
+		}
+	})
+}
+
+func TestMergeSortedRandom(t *testing.T) {
+	createList := func(n int) *pb.List {
+		list := make([]uint64, n)
+		for i := range list {
+			list[i] = uint64(rand.Int63())
+		}
+		sort.Slice(list, func(i, j int) bool {
+			return list[i] < list[j]
+		})
+		return &pb.List{Uids: list}
+	}
+
+	input := []*pb.List{}
+	for i := 0; i < 1000; i++ {
+		input = append(input, createList(1000))
+	}
+
+	require.Equal(t, MergeSorted(input).Uids, internalMergeSort(input).Uids)
+}
+
 func TestMergeSorted7(t *testing.T) {
 	input := []*pb.List{
 		newList([]uint64{5, 6, 7}),