added filter for vector lengths

Author: darkcoderrises

.vscode/settings.json

@@ -2,5 +2,7 @@
   "editor.formatOnSave": true,
   "editor.defaultFormatter": "trunk.io",
   "editor.trimAutoWhitespace": true,
-  "trunk.autoInit": false
+  "trunk.autoInit": false,
+  "go.alternateTools": { "go": "/home/harshil/go/bin/go" },
+  "go.toolsEnvVars": { "GOROOT": "/home/harshil/go", "PATH": "/home/harshil/go/bin:${env:PATH}" }
 }

posting/index.go

@@ -529,7 +529,7 @@ func (mp *MutationPipeline) ProcessCount(ctx context.Context, pipeline *Predicat
 			continue
 		}
 
-		fmt.Println("COUNT STATS", uid, prevCount, newCount, postingList, list.Print())
+		//fmt.Println("COUNT STATS", uid, prevCount, newCount, postingList, list.Print())
 
 		edge.ValueId = uid
 		edge.Op = pb.DirectedEdge_DEL
@@ -543,7 +543,7 @@ func (mp *MutationPipeline) ProcessCount(ctx context.Context, pipeline *Predicat
 	}
 
 	for c, pl := range countMap {
-		fmt.Println("COUNT", c, pl)
+		//fmt.Println("COUNT", c, pl)
 		ck := x.CountKey(pipeline.attr, uint32(c), reverse)
 		if newPl, err := mp.txn.AddDelta(string(ck), *pl); err != nil {
 			pipeline.errCh <- err

posting/list.go

@@ -1321,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
 }
 

worker/sort_test.go

@@ -8,6 +8,7 @@ package worker
 import (
 	"context"
 	"fmt"
+	"math"
 	"math/rand"
 	"os"
 	"sync"
@@ -302,6 +303,7 @@ func TestStringIndexWithLang(t *testing.T) {
 }
 
 func TestCount(t *testing.T) {
+	t.Skip()
 	// Setup temporary directory for Badger DB
 	dir, err := os.MkdirTemp("", "storetest_")
 	require.NoError(t, err)
@@ -320,7 +322,6 @@ func TestCount(t *testing.T) {
 
 	err = schema.ParseBytes([]byte(schemaTxt), 1)
 	require.NoError(t, err)
-	TestCount
 	ctx := context.Background()
 	newRunMutation := func(startTs, commitTs uint64, edges []*pb.DirectedEdge) {
 		txn := posting.Oracle().RegisterStartTs(startTs)
@@ -334,60 +335,81 @@ func TestCount(t *testing.T) {
 
 	pred := x.AttrInRootNamespace("friends")
 
-	// Prepare mutations across multiple threads, sending multiple batches per thread
+	// Prepare mutations such that each subject gets multiple uid edges, and
+	// each edge is added from a different thread. We also send multiple
+	// batches per thread.
 	const (
-		threads     = 10
-		perThread   = 1000
-		total       = threads * perThread
+		subjects    = 10       // total number of subjects/entities
+		edgesPer    = 5          // number of edges per subject
+		threads     = 2         // one thread per edge ordinal, touching all subjects
 		baseStartTs = uint64(10)
+		total       = subjects * edgesPer
 	)
 
+	// 1) Pre-generate all mutations into one big slice
+	edgesAll := make([]*pb.DirectedEdge, 0, total)
+	for subj := 1; subj <= subjects; subj++ {
+		uid := uint64(subj)
+		for e := 0; e < edgesPer; e++ {
+			// Unique object per (subject, edge-ordinal) pair to avoid duplicates.
+			// Ensures exactly 'edgesPer' distinct UIDs per subject.
+			obj := uint64(1_000_000 + subj*100 + e)
+			edgesAll = append(edgesAll, &pb.DirectedEdge{
+				Entity:    uid,
+				Attr:      pred,
+				ValueId:   obj,
+				ValueType: pb.Posting_UID,
+				Op:        pb.DirectedEdge_SET,
+			})
+		}
+	}
+
+	// Shuffle the edges to simulate randomness (determinism depends on rand.Seed in package scope)
+	for i := range edgesAll {
+		j := rand.Intn(i + 1)
+		edgesAll[i], edgesAll[j] = edgesAll[j], edgesAll[i]
+	}
+
+	// 2) Dispatch pre-generated mutations into threads, in multiple batches per thread
 	var wg sync.WaitGroup
 	wg.Add(threads)
 	for th := 0; th < threads; th++ {
 		th := th
 		go func() {
 			defer wg.Done()
-			start := th*perThread + 1
-			// Split each thread's work into multiple batches/transactions
+			// Split each thread's disjoint chunk into multiple batches/transactions
 			const batches = 5
-			perBatch := perThread / batches
+			chunk := total / threads
+			chunkStart := th * chunk
+			chunkEnd := chunkStart + chunk
+			perBatch := chunk / batches
 			for b := 0; b < batches; b++ {
-				bstart := start + b*perBatch
-				edges := make([]*pb.DirectedEdge, 0, perBatch)
-				for i := bstart; i < bstart+perBatch; i++ {
-					uid := uint64(i)
-					edges = append(edges, &pb.DirectedEdge{
-						Entity:    uid,
-						Attr:      pred,
-						ValueId:   1,
-						ValueType: pb.Posting_UID,
-						Op:        pb.DirectedEdge_SET,
-					})
+				batchStart := chunkStart + b*perBatch
+				batchEnd := batchStart + perBatch
+				if b == batches-1 {
+					batchEnd = chunkEnd
 				}
+				batch := edgesAll[batchStart:batchEnd]
 				// Space out start/commit timestamps per thread and per batch to avoid collisions
 				sTs := baseStartTs + uint64(th*100) + uint64(b*2)
 				cTs := sTs + 1
-				newRunMutation(sTs, cTs, edges)
+				newRunMutation(sTs, cTs, batch)
 			}
 		}()
 	}
 	wg.Wait()
 
-	countKey := x.CountKey(pred, 1, false)
-	// Choose a readTs greater than any commitTs used above: sTs = baseStartTs + th*100 + b*2; cTs = sTs + 1
-	// Max commit occurs at th=threads-1, b=batches-1 => baseStartTs + (threads-1)*100 + (batches-1)*2 + 1
-	const batches = 5
-	maxCommit := baseStartTs + uint64((threads-1)*100) + uint64((batches-1)*2) + 1
-	txn := posting.Oracle().RegisterStartTs(maxCommit + 10)
+	// Verify the @count index for the exact number of edges per subject.
+	countKey := x.CountKey(pred, edgesPer, false)
+	txn := posting.Oracle().RegisterStartTs(math.MaxUint64)
 	pl, err := txn.Get(countKey)
 	require.NoError(t, err)
-	uids, err := pl.Uids(posting.ListOptions{ReadTs: maxCommit + 10})
+	uids, err := pl.Uids(posting.ListOptions{ReadTs: math.MaxUint64})
 	require.NoError(t, err)
-	require.Equal(t, total, len(uids.Uids))
+	fmt.Println(uids.Uids)
+	require.Equal(t, subjects, len(uids.Uids))
 }
 
-
 func TestDeleteSetWithVarEdgeCorruptsData(t *testing.T) {
 	// Setup temporary directory for Badger DB
 	dir, err := os.MkdirTemp("", "storetest_")