Add partition key based iterator #4841

poonai · 2020-02-24T09:00:06Z

This pr will remove the heap-based sorting, instead we'll batch every keys in memory in partition manner and we sort it

This change is

Docs Preview:

Signed-off-by: balaji <[email protected]>

…ures that we don't have too many pending requests.

…d speed.

…encoder. We need a way to parse key easily for comparisons though.

Signed-off-by: balaji <[email protected]>

…aji/new_reduce Signed-off-by: balaji <[email protected]>

Signed-off-by: balaji <[email protected]>

manishrjain

Looks alright to me. Have some comments. Will defer to @animesh2049 to do the final review and LGTM.

Reviewed 1 of 5 files at r2, 1 of 3 files at r3, 7 of 8 files at r5.
Reviewable status: all files reviewed, 15 unresolved discussions (waiting on @balajijinnah)

dgraph/cmd/bulk/key.go, line 1 at r5 (raw file):

package bulk

Add a license, but use the license that protobuf generated code has (or protobuf has).

dgraph/cmd/bulk/mapper.go, line 129 at r5 (raw file):

		PartitionKeys: [][]byte{},
	}
	shardPartioionNo := len(entries) / partitionKeyShard

spelling mistake

dgraph/cmd/bulk/mapper.go, line 132 at r5 (raw file):

	for i := range entries {
		if shardPartioionNo == 0 {
			// we have only less entries so no need for partition keys.

we have very few ...

dgraph/cmd/bulk/mapper.go, line 144 at r5 (raw file):

	lenBuf := make([]byte, 4)
	binary.BigEndian.PutUint32(lenBuf, uint32(len(headerBuf)))
	_, err = w.Write(lenBuf)

x.Check2(...)

dgraph/cmd/bulk/mapper.go, line 146 at r5 (raw file):

	_, err = w.Write(lenBuf)
	x.Check(err)
	_, err = w.Write(headerBuf)

x.Check2(...)

dgraph/cmd/bulk/reduce.go, line 89 at r5 (raw file):

			})

			// Start bactching for the given keys

spelling

dgraph/cmd/bulk/reduce.go, line 181 at r5 (raw file):

	// readKey reads the next map entry key.
	readKey := func() {

readKey := func() error {

...
return io.EOF
}

Don't need eof variable.

Also, rename to readMapEntry

dgraph/cmd/bulk/reduce.go, line 185 at r5 (raw file):

			return
		}
		if !prevKeyExist {

if prevKeyExist { return }

dgraph/cmd/bulk/reduce.go, line 202 at r5 (raw file):

			x.Check2(io.ReadFull(r, eBuf))

			key, err = GetKeyForMapEntry(eBuf)

return key, err

key, err = readMapEntry(...)

dgraph/cmd/bulk/reduce.go, line 217 at r5 (raw file):

		ie.partitionKey = pKey
		for {
			readKey()

if err := readKey(); err == io.EOF { break }

dgraph/cmd/bulk/reduce.go, line 231 at r5 (raw file):

		}
		mi.batchCh <- ie
		i++

Doesn't look like its used.

dgraph/cmd/bulk/reduce.go, line 237 at r5 (raw file):

	batch := make([][]byte, 0, batchAlloc)
	for {
		readKey()

fix this up as well.

dgraph/cmd/bulk/reduce.go, line 257 at r5 (raw file):

}

func (mi *mapIterator) Next() bool {

Next() *iteratorEntry

Remove current(). If nothing is there, then return nil.

dgraph/cmd/bulk/reduce.go, line 374 at r5 (raw file):

		go r.encode(encoderCh, encoderCloser)
	}
	// Start lisenting to write the badger list.

spelling mistake.

dgraph/cmd/bulk/reduce.go, line 380 at r5 (raw file):

	for i := 0; i < len(partitionKeys); i++ {
		batch := make([][]byte, 0, batchAlloc)
		for _, itr := range mapItrs {

Let's add the assert back.

Signed-off-by: balaji <[email protected]>

poonai

Reviewable status: 6 of 9 files reviewed, 15 unresolved discussions (waiting on @manishrjain)

dgraph/cmd/bulk/key.go, line 1 at r5 (raw file):