Method startWriting

dgraph/cmd/bulk/reduce.go:363–483

(ci *countIndexer, vi *vectorIndexer, writerCh chan *encodeRequest, closer *z.Closer)

}

func (r *reducer) startWriting(ci *countIndexer, vi *vectorIndexer, writerCh chan *encodeRequest, closer *z.Closer) {
	defer closer.Done()

	// Concurrently write split lists to a temporary badger.
	tmpWg := new(sync.WaitGroup)
	tmpWg.Add(1)
	go r.writeTmpSplits(ci, tmpWg)

	count := func(req *encodeRequest) {
		defer func() {
			if err := req.countBuf.Release(); err != nil {
				glog.Warningf("error in releasing buffer: %v", err)
			}
		}()
		if req.countBuf.IsEmpty() {
			return
		}

		// req.countBuf is already sorted.
		sz := req.countBuf.LenNoPadding()
		ci.countBuf.Grow(sz)

		if err := req.countBuf.SliceIterate(func(slice []byte) error {
			ce := countEntry(slice)
			ci.addCountEntry(ce)
			return nil
		}); err != nil {
			glog.Errorf("error while iterating over buf: %v", err)
			x.Check(err)
		}
	}

	// Process vector entries and insert into HNSW index
	vector := func(req *encodeRequest) {
		if req.vectorBuf == nil {
			return
		}
		defer func() {
			if err := req.vectorBuf.Release(); err != nil {
				glog.Warningf("error releasing vector buffer: %v", err)
			}
		}()
		if req.vectorBuf.IsEmpty() {
			return
		}

		// Iterate through vector entries and insert into HNSW
		if err := req.vectorBuf.SliceIterate(func(slice []byte) error {
			ve := unmarshalVectorEntry(slice)
			if ve == nil {
				vi.handleUnmarshalError()
				return nil
			}
			// Insert vector into HNSW and generate entries
			vi.addVectorEntry(ve)
			return nil
		}); err != nil {
			glog.Errorf("error processing vectors: %v", err)