Method writeMapEntriesToFile

dgraph/cmd/bulk/mapper.go:146–218

(cbuf *z.Buffer, shardIdx int)

}

func (m *mapper) writeMapEntriesToFile(cbuf *z.Buffer, shardIdx int) {
	defer func() {
		m.shards[shardIdx].mu.Unlock() // Locked by caller.
		if err := cbuf.Release(); err != nil {
			glog.Warningf("error in releasing buffer: %v", err)
		}
	}()

	cbuf.SortSlice(func(ls, rs []byte) bool {
		lhs := MapEntry(ls)
		rhs := MapEntry(rs)
		return less(lhs, rhs)
	})

	f, err := m.openOutputFile(shardIdx)
	x.Check(err)

	defer func() {
		x.Check(f.Sync())
		x.Check(f.Close())
	}()

	w := s2.NewWriter(f)
	defer func() {
		x.Check(w.Close())
	}()

	// Create partition keys for the map file.
	header := &pb.MapHeader{
		PartitionKeys: [][]byte{},
	}

	var bufSize int64
	if err := cbuf.SliceIterate(func(slice []byte) error {
		me := MapEntry(slice)
		bufSize += int64(4 + len(me))
		if bufSize < m.opt.PartitionBufSize {
			return nil
		}
		sz := len(header.PartitionKeys)
		if sz > 0 && bytes.Equal(me.Key(), header.PartitionKeys[sz-1]) {
			// We already have this key.
			return nil
		}
		header.PartitionKeys = append(header.PartitionKeys, me.Key())
		bufSize = 0
		return nil
	}); err != nil {
		glog.Errorf("error while iterating over buf: %v", err)
		x.Check(err)
	}

	// Write the header to the map file.s
	headerBuf, err := proto.Marshal(header)
	x.Check(err)
	lenBuf := make([]byte, 4)
	binary.BigEndian.PutUint32(lenBuf, uint32(len(headerBuf)))
	x.Check2(w.Write(lenBuf))