Some optimizations in intersection of lists. (#370)

Some optimizations in intersection of lists:

1. Stopping intersection if some list has reached its end.
2. Use a double for loop to enhance usage of locality caching.

algo/uidlist.go

@@ -19,19 +19,25 @@ func ApplyFilter(u *task.List, f func(uint64, int) bool) {
 	u.Uids = out
 }
 
-// IntersectWith intersects u with v. The update is made to u. It assumes
-// that u, v are sorted, which are not always the case.
+// IntersectWith intersects u with v. The update is made to u.
+// u, v should be sorted.
 func IntersectWith(u, v *task.List) {
 	out := u.Uids[:0]
 	n := len(u.Uids)
 	m := len(v.Uids)
-	var k int
-	for i := 0; i < n; i++ {
+	for i, k := 0, 0; i < n && k < m; {
 		uid := u.Uids[i]
-		for ; k < m && v.Uids[k] < uid; k++ {
-		}
-		if k < m && v.Uids[k] == uid {
+		vid := v.Uids[k]
+		if uid > vid {
+			for k = k + 1; k < m && v.Uids[k] < uid; k++ {
+			}
+		} else if uid == vid {
 			out = append(out, uid)
+			k++
+			i++
+		} else {
+			for i = i + 1; i < n && u.Uids[i] < vid; i++ {
+			}
 		}
 	}
 	u.Uids = out
@@ -73,7 +79,9 @@ func IntersectSorted(lists []*task.List) *task.List {
 	// lptrs[j] is the element we are looking at for lists[j].
 	lptrs := make([]int, len(lists))
 	shortList := lists[minLenIdx]
-	for i := 0; i < len(shortList.Uids); i++ {
+	elemsLeft := true  // If some list has no elems left, we can't intersect more.
+
+	for i := 0; i < len(shortList.Uids) && elemsLeft; i++ {
 		val := shortList.Uids[i]
 		if i > 0 && val == shortList.Uids[i-1] {
 			x.AssertTruef(false, "We shouldn't have duplicates in UIDLists")
@@ -94,6 +102,7 @@ func IntersectSorted(lists []*task.List) *task.List {
 
 			lptrs[j] = ljp
 			if ljp >= lsz || lj.Uids[ljp] > val {
+				elemsLeft = ljp < lsz
 				skip = true
 				break
 			}

algo/uidlist_test.go

@@ -144,6 +144,15 @@ func TestIntersectSorted5(t *testing.T) {
 	require.Empty(t, IntersectSorted(input).Uids)
 }
 
+func TestIntersectSorted6(t *testing.T) {
+	input := []*task.List{
+		newList([]uint64{10, 12, 13}),
+		newList([]uint64{2, 3, 4, 13}),
+		newList([]uint64{4, 5, 6}),
+	}
+	require.Empty(t, IntersectSorted(input).Uids)
+}
+
 func TestUIDListIntersect1(t *testing.T) {
 	u := newList([]uint64{1, 2, 3})
 	v := newList([]uint64{})
@@ -179,6 +188,27 @@ func TestUIDListIntersect5(t *testing.T) {
 	require.Equal(t, u.Uids, []uint64{3})
 }
 
+func TestUIDListIntersectDupFirst(t *testing.T) {
+	u := newList([]uint64{1, 1, 2, 3})
+	v := newList([]uint64{1, 2})
+	IntersectWith(u, v)
+	require.Equal(t, []uint64{1, 2}, u.Uids)
+}
+
+func TestUIDListIntersectDupBoth(t *testing.T) {
+	u := newList([]uint64{1, 1, 2, 3, 5})
+	v := newList([]uint64{1, 1, 2, 4})
+	IntersectWith(u, v)
+	require.Equal(t, []uint64{1, 1, 2}, u.Uids)
+}
+
+func TestUIDListIntersectDupSecond(t *testing.T) {
+	u := newList([]uint64{1, 2, 3, 5})
+	v := newList([]uint64{1, 1, 2, 4})
+	IntersectWith(u, v)
+	require.Equal(t, []uint64{1, 2}, u.Uids)
+}
+
 func TestApplyFilterUint(t *testing.T) {
 	u := newList([]uint64{1, 2, 3, 4, 5})
 	ApplyFilter(u, func(a uint64, idx int) bool { return (a % 2) == 1 })