Method processLoadFile

dgraph/cmd/live/run.go:470–585

(ctx context.Context, rd *bufio.Reader, ck chunker.Chunker)

}

func (l *loader) processLoadFile(ctx context.Context, rd *bufio.Reader, ck chunker.Chunker) error {
	nqbuf := ck.NQuads()
	errCh := make(chan error, 1)
	// Spin a goroutine to push NQuads to mutation channel.
	go func() {
		var err error
		defer func() {
			errCh <- err
		}()
		buffer := make([]*api.NQuad, 0, opt.bufferSize*opt.batchSize)

		drain := func() {
			// We collect opt.bufferSize requests and preprocess them. For the requests
			// to not conflict between themselves, we sort them on the basis of their predicates.
			// Predicates with count index will conflict among themselves, so we keep them at
			// end, making room for other predicates to load quickly.
			sort.Slice(buffer, func(i, j int) bool {
				iPred := sch.preds[x.NamespaceAttr(buffer[i].Namespace, buffer[i].Predicate)]
				jPred := sch.preds[x.NamespaceAttr(buffer[j].Namespace, buffer[j].Predicate)]
				t := func(a *Predicate) int {
					if a != nil && a.Count {
						return 1
					}
					return 0
				}

				// Sorts the nquads on basis of their predicates, while keeping the
				// predicates with count index later than those without it.
				if t(iPred) != t(jPred) {
					return t(iPred) < t(jPred)
				}
				return buffer[i].Predicate < buffer[j].Predicate
			})
			for len(buffer) > 0 {
				sz := opt.batchSize
				if len(buffer) < opt.batchSize {
					sz = len(buffer)
				}
				mu := &request{Mutation: &api.Mutation{Set: buffer[:sz]}}
				l.reqs <- mu
				buffer = buffer[sz:]
			}
		}

		for nqs := range nqbuf.Ch() {
			if len(nqs) == 0 {
				continue
			}

			for _, nq := range nqs {
				if !opt.preserveNs {
					// If do not preserve namespace, use the namespace passed through
					// `--force-namespace` flag.
					nq.Namespace = opt.namespaceToLoad
				}
				if _, ok := l.namespaces[nq.Namespace]; !ok {
					err = errors.Errorf("Cannot load nquad:%+v as its namespace doesn't exist.", nq)
					return