Method run

dgraph/cmd/bulk/mapper.go:220–292

(inputFormat chunker.InputFormat)

}

func (m *mapper) run(inputFormat chunker.InputFormat) {
	chunk := chunker.NewChunker(inputFormat, 1000)
	nquads := chunk.NQuads()
	go func() {
		for chunkMeta := range m.readerChunkCh {
			if err := chunk.Parse(chunkMeta.buf); err != nil {
				atomic.AddInt64(&m.prog.errCount, 1)
				if m.errorLog != nil {
					m.errorLog.Log(chunkMeta.filename, err, "")
				}
				if !m.opt.IgnoreErrors {
					x.Check(err)
				}
			}
		}
		aclOnce.Do(func() {
			if m.opt.Namespace != math.MaxUint64 && m.opt.Namespace != x.RootNamespace {
				// Insert ACL related RDFs force uploading the data into non-galaxy namespace.
				aclNquads := make([]*api.NQuad, 0)
				aclNquads = append(aclNquads, acl.CreateGroupNQuads(x.SuperAdminId)...)
				aclNquads = append(aclNquads, acl.CreateUserNQuads(x.GrootId, "password")...)
				aclNquads = append(aclNquads, &api.NQuad{
					Subject:   "_:newuser",
					Predicate: "dgraph.user.group",
					ObjectId:  "_:newgroup",
				})
				nquads.Push(aclNquads...)
			}
		})
		nquads.Flush()
	}()

	for nqs := range nquads.Ch() {
		for _, nq := range nqs {
			if err := facets.SortAndValidate(nq.Facets); err != nil {
				atomic.AddInt64(&m.prog.errCount, 1)
				if m.errorLog != nil {
					m.errorLog.Log("<facet_validation>", err, fmt.Sprintf("subject=%s predicate=%s", nq.Subject, nq.Predicate))
				}
				if !m.opt.IgnoreErrors {
					x.Check(err)
				}
			}

			m.processNQuad(dql.NQuad{NQuad: nq})
			atomic.AddInt64(&m.prog.nquadCount, 1)
		}

		for i := range m.shards {
			sh := &m.shards[i]
			if uint64(sh.cbuf.LenNoPadding()) >= m.opt.MapBufSize {
				sh.mu.Lock() // One write at a time.
				go m.writeMapEntriesToFile(sh.cbuf, i)
				// Clear the entries and encodedSize for the next batch.
				// Proactively allocate 32 slots to bootstrap the entries slice.
				sh.cbuf = newMapperBuffer(m.opt)
			}
		}