Method wait

dgraph/cmd/bulk/vector_indexer.go:444–481

wait waits for all vector indexing operations to complete. It flushes any remaining writes and commits all HNSW mutations from LocalCache to vectorTmpDb.

()

// wait waits for all vector indexing operations to complete.
// It flushes any remaining writes and commits all HNSW mutations from LocalCache to vectorTmpDb.
func (vi *vectorIndexer) wait() {
	// Flush any remaining vector posting list writes
	if err := vi.flushWriteBatch(); err != nil {
		vi.handleError(err, fmt.Sprintf("shard=%d", vi.shardId), "<vector_flush>")
	}

	vi.mu.Lock()
	defer vi.mu.Unlock()

	// If no indexers were created (no vectors arrived), nothing to commit
	if len(vi.txns) == 0 {
		glog.Infof("Vector indexer wait completed for shard %d (no vectors processed)", vi.shardId)
		return
	}

	// Commit HNSW mutations from LocalCache to vectorTmpDb
	// This is critical - without this, HNSW edge data stays in memory and never gets persisted
	// Step 1: Call Update() to move mutations from plists to deltas
	// Step 2: Call CommitToDisk() to write deltas to BadgerDB
	writer := posting.NewTxnWriter(vi.tmpDb)
	for pred, txn := range vi.txns {
		// Update moves mutations from posting lists to delta cache
		txn.Update()
		if err := txn.CommitToDisk(writer, vi.state.writeTs); err != nil {
			vi.handleError(err,
				fmt.Sprintf("predicate=%s shard=%d", pred, vi.shardId),
				"<vector_commit>",
			)
		} else {
			glog.Infof("Committed HNSW data for predicate %s to tmpDb (shard %d)", pred, vi.shardId)
		}
	}
	if err := writer.Flush(); err != nil {
		vi.handleError(err, fmt.Sprintf("shard=%d", vi.shardId), "<vector_commit>")
	}

	glog.Infof("Vector indexer wait completed for shard %d with %d predicates", vi.shardId, len(vi.vectorPreds))
}