fix(hnsw): restore multi-threaded build; vendor USearch #735 fix

pkg/vectorindex/hnsw/build.go

@@ -52,30 +52,25 @@ type AddItem[T types.RealNumbers] struct {
 func NewHnswBuild[T types.RealNumbers](sqlproc *sqlexec.SqlProcess, uid string, nworker int32,
 	cfg vectorindex.IndexConfig, tblcfg vectorindex.IndexTableConfig) (info *HnswBuild[T], err error) {
 
-	/*
-		// estimate the number of worker threads
-		nthread := 0
-		if nworker <= 1 {
-			// single database thread and set nthread to ThreadsBuild
-			nthread = int(vectorindex.GetConcurrency(tblcfg.ThreadsBuild))
-		} else {
-			// multiple database worker threads
-			threadsbuild := vectorindex.GetConcurrencyForBuild(tblcfg.ThreadsBuild)
-			nthread = int(float64(threadsbuild) / float64(nworker))
-		}
-		if nthread < 1 {
-			nthread = 1
-		}
-	*/
-
-	// MatrixOne #24849 / USearch #735 (open): concurrent add() can orphan nodes —
-	// the vector is stored (contains() returns true) but the HNSW graph never links
-	// it, so search() can never reach it, producing flaky recall@1 (an exact match
-	// is intermittently missed). This is a real build race, not just HNSW
-	// approximation. Reproduced in pkg/vectorindex/hnsw/zz_orphan_test.go:
-	// multi-threaded build orphans ~1/30, single-threaded 0/30. Until the upstream
-	// race is fixed, force a single build thread for correctness.
-	nthread := 1
+	// estimate the number of worker threads
+	//
+	// MatrixOne #24849 / USearch #735: concurrent add() used to orphan nodes (a
+	// vector stored but never linked into the HNSW graph, so search() could not
+	// reach it — flaky recall@1). That race is fixed in our usearch build (the
+	// two-pass add: all forward links before any reverse link), so concurrent
+	// builds now match single-threaded reachability. Multi-threaded build restored.
+	nthread := 0
+	if nworker <= 1 {
+		// single database thread and set nthread to ThreadsBuild
+		nthread = int(vectorindex.GetConcurrency(tblcfg.ThreadsBuild))
+	} else {
+		// multiple database worker threads
+		threadsbuild := vectorindex.GetConcurrencyForBuild(tblcfg.ThreadsBuild)
+		nthread = int(float64(threadsbuild) / float64(nworker))
+	}
+	if nthread < 1 {
+		nthread = 1
+	}
 
 	info = &HnswBuild[T]{
 		uid:     uid,

pkg/vectorindex/hnsw/sync.go

@@ -93,10 +93,7 @@ func NewHnswSync[T types.RealNumbers](sqlproc *sqlexec.SqlProcess,
 		if err != nil {
 			return nil, err
 		}
-		// Force single-thread build until USearch #735 is fixed (concurrent add()
-		// orphans HNSW graph nodes -> flaky recall@1). See
-		// vectorindex.GetConcurrencyForSingleThreadBuild for the one-line revert.
-		idxtblcfg.ThreadsBuild = vectorindex.GetConcurrencyForSingleThreadBuild(val.(int64))
+		idxtblcfg.ThreadsBuild = vectorindex.GetConcurrencyForBuild(val.(int64))
 
 		idxcap, err := sqlproc.GetResolveVariableFunc()("hnsw_max_index_capacity", true, false)
 		if err != nil {
@@ -105,10 +102,7 @@ func NewHnswSync[T types.RealNumbers](sqlproc *sqlexec.SqlProcess,
 		indexCapacity = idxcap.(int64)
 	} else {
 
-		// Force single-thread build until USearch #735 is fixed (concurrent add()
-		// orphans HNSW graph nodes -> flaky recall@1). See
-		// vectorindex.GetConcurrencyForSingleThreadBuild for the one-line revert.
-		idxtblcfg.ThreadsBuild = vectorindex.GetConcurrencyForSingleThreadBuild(0)
+		idxtblcfg.ThreadsBuild = vectorindex.GetConcurrencyForBuild(0)
 		indexCapacity = 1000000
 	}
 

pkg/vectorindex/hnsw/zz_orphan_test.go

@@ -19,7 +19,6 @@ import (
 	"compress/gzip"
 	"fmt"
 	"os"
-	"runtime"
 	"strconv"
 	"strings"
 	"sync"
@@ -64,9 +63,10 @@ func zzBuild(keys []usearch.Key, vecs [][]float32, dim int, threads uint) *usear
 	c := usearch.DefaultConfig(uint(dim))
 	c.Quantization = usearch.F32
 	c.Metric = usearch.L2sq
+	// Match the BVT t2 case (vector_hnsw_async.sql): M 64 EF_CONSTRUCTION 200 EF_SEARCH 200.
 	c.Connectivity = 64
-	c.ExpansionAdd = 500
-	c.ExpansionSearch = 1000
+	c.ExpansionAdd = 200
+	c.ExpansionSearch = 200
 	idx, _ := usearch.NewIndex(c)
 	idx.Reserve(uint(len(keys)))
 	idx.ChangeThreadsAdd(threads)
@@ -99,20 +99,31 @@ func zzBuild(keys []usearch.Key, vecs [][]float32, dim int, threads uint) *usear
 	return idx
 }
 
-// TestZZBuildOrphan is a reference reproducer for USearch #735 (concurrent add()
-// orphans nodes): a multi-threaded build occasionally leaves id 0 unreachable in
-// search despite contains()==true; single-threaded never does. Kept to verify the
-// single-thread build workaround (build.go) and any upstream fix. Slow; needs SIFT.
+// TestZZBuildOrphan is a regression guard for USearch #735 (concurrent add()
+// orphans nodes): a multi-threaded build used to occasionally leave id 0
+// unreachable in search despite contains()==true. Our patched libusearch
+// (two-pass add: all forward links before any reverse link) fixes the race, so
+// an 8-thread build must now report 0 orphans — this asserts that and fails if a
+// future libusearch regresses it. Builds 30x with the same params as the BVT t2
+// case (M 64, EF_CONSTRUCTION 200, EF_SEARCH 200). Auto-skips when the SIFT data
+// file is absent (see zzLoadSift).
 func TestZZBuildOrphan(t *testing.T) {
-	t.Skip("USearch #735 reference repro; skipped by default — comment out this line to run manually")
 	keys, vecs, dim := zzLoadSift(t)
 	t.Logf("loaded %d vectors dim=%d id0=%d", len(keys), dim, keys[0])
 	q := vecs[0]
 	const iters = 30
-	for _, threads := range []uint{uint(runtime.NumCPU()), 1} {
+	for _, threads := range []uint{8} {
 		notTop1, missing, notContained := 0, 0, 0
 		for it := 0; it < iters; it++ {
-			idx := zzBuild(keys, vecs, dim, threads)
+			// Rotate the insertion order each iteration so a different key lands
+			// first and the thread chunks shift — exercises different concurrent
+			// add interleavings, like `load data ... parallel 'true'` loading rows
+			// in a non-deterministic order. Deterministic (no RNG); keys stay
+			// aligned with vecs.
+			off := (it * (len(keys) / iters)) % len(keys)
+			ik := append(append([]usearch.Key(nil), keys[off:]...), keys[:off]...)
+			iv := append(append([][]float32(nil), vecs[off:]...), vecs[:off]...)
+			idx := zzBuild(ik, iv, dim, threads)
 			contained, _ := idx.Contains(0)
 			rk, _, _ := idx.Search(q, 10)
 			rank := -1
@@ -141,5 +152,13 @@ func TestZZBuildOrphan(t *testing.T) {
 			idx.Destroy()
 		}
 		fmt.Printf("\n*** threads=%d : id0_not_top1=%d/%d  id0_missing_top10=%d/%d  id0_not_in_index=%d/%d ***\n", threads, notTop1, iters, missing, iters, notContained, iters)
+		// #735 regression guard: with the patched libusearch the build must never
+		// orphan id 0, at any thread count. notContained==0 always held (the vector
+		// is stored); the race only broke reachability, so missing/notTop1 are the
+		// real signal.
+		if missing > 0 || notTop1 > 0 || notContained > 0 {
+			t.Errorf("USearch #735 regression: threads=%d orphaned id0 — not_top1=%d/%d missing_top10=%d/%d not_in_index=%d/%d",
+				threads, notTop1, iters, missing, iters, notContained, iters)
+		}
 	}
 }

pkg/vectorindex/types.go

@@ -373,15 +373,3 @@ func SimulateDevices(devices []int, n int64) []int {
 	// all zeros -> every logical rank maps to physical device 0
 	return sim
 }
-
-// GetConcurrencyForSingleThreadBuild returns the build concurrency for the HNSW
-// write paths (CDC/sync). While MatrixOne #24849 / USearch #735 (open) is
-// unresolved, concurrent USearch add() can orphan graph nodes — the vector is
-// stored (contains() is true) but never linked into the HNSW graph, so search()
-// can never reach it, producing flaky recall@1. So every HNSW build/sync path
-// must add from a single thread (the model is likewise pinned to
-// ChangeThreadsAdd(1) in NewHnswModelForBuild). When usearch fixes the race,
-// this is a one-line revert: `return GetConcurrencyForBuild(nthread)`.
-func GetConcurrencyForSingleThreadBuild(nthread int64) int64 {
-	return 1
-}

test/distributed/cases/pessimistic_transaction/vector/vector_hnsw_async.result

@@ -27,8 +27,8 @@ load data infile {'filepath'='$resources/vector/sift128_base_10k.csv.gz', 'compr
 select count(*) from t2;
 count(*)
 10000
-select sleep(20);
-sleep(20)
+select sleep(30);
+sleep(30)
 0
 select * from t2 order by L2_DISTANCE(b, "[14, 2, 0, 0, 0, 2, 42, 55, 9, 1, 0, 0, 18, 100, 77, 32, 89, 1, 0, 0, 19, 85, 15, 68, 52, 4, 0, 0, 0, 0, 2, 28, 34, 13, 5, 12, 49, 40, 39, 37, 24, 2, 0, 0, 34, 83, 88, 28, 119, 20, 0, 0, 41, 39, 13, 62, 119, 16, 2, 0, 0, 0, 10, 42, 9, 46, 82, 79, 64, 19, 2, 5, 10, 35, 26, 53, 84, 32, 34, 9, 119, 119, 21, 3, 3, 11, 17, 14, 119, 25, 8, 5, 0, 0, 11, 22, 23, 17, 42, 49, 17, 12, 5, 5, 12, 78, 119, 90, 27, 0, 4, 2, 48, 92, 112, 85, 15, 0, 2, 7, 50, 36, 15, 11, 1, 0, 0, 7]") ASC LIMIT 1;
 a    b

test/distributed/cases/pessimistic_transaction/vector/vector_hnsw_async.sql

@@ -47,7 +47,7 @@ load data infile {'filepath'='$resources/vector/sift128_base_10k.csv.gz', 'compr
 
 select count(*) from t2;
 
-select sleep(20);
+select sleep(30);
 
 select * from t2 order by L2_DISTANCE(b, "[14, 2, 0, 0, 0, 2, 42, 55, 9, 1, 0, 0, 18, 100, 77, 32, 89, 1, 0, 0, 19, 85, 15, 68, 52, 4, 0, 0, 0, 0, 2, 28, 34, 13, 5, 12, 49, 40, 39, 37, 24, 2, 0, 0, 34, 83, 88, 28, 119, 20, 0, 0, 41, 39, 13, 62, 119, 16, 2, 0, 0, 0, 10, 42, 9, 46, 82, 79, 64, 19, 2, 5, 10, 35, 26, 53, 84, 32, 34, 9, 119, 119, 21, 3, 3, 11, 17, 14, 119, 25, 8, 5, 0, 0, 11, 22, 23, 17, 42, 49, 17, 12, 5, 5, 12, 78, 119, 90, 27, 0, 4, 2, 48, 92, 112, 85, 15, 0, 2, 7, 50, 36, 15, 11, 1, 0, 0, 7]") ASC LIMIT 1;