executor: add batch copy to inner join, left and right outer join.

executor/joiner.go

@@ -158,19 +158,18 @@ func (j *baseJoiner) makeShallowJoinRow(isRightJoin bool, inner, outer chunk.Row
 	j.shallowRow.ShallowCopyPartialRow(inner.Len(), outer)
 }
 
-func (j *baseJoiner) filter(input, output *chunk.Chunk) (matched bool, err error) {
+func (j *baseJoiner) filter(input, output *chunk.Chunk, outerColsLen int) (bool, error) {
+	var err error
 	j.selected, err = expression.VectorizedFilter(j.ctx, j.conditions, chunk.NewIterator4Chunk(input), j.selected)
 	if err != nil {
 		return false, errors.Trace(err)
 	}
-	for i := 0; i < len(j.selected); i++ {
-		if !j.selected[i] {
-			continue
-		}
-		matched = true
-		output.AppendRow(input.GetRow(i))
+	// batch copy selected row to output chunk
+	innerColOffset, outerColOffset := 0, input.NumCols()-outerColsLen
+	if !j.outerIsRight {
+		innerColOffset, outerColOffset = outerColsLen, 0
 	}
-	return matched, nil
+	return chunk.CopySelectedJoinRows(input, innerColOffset, outerColOffset, j.selected, output), nil
 }
 
 type semiJoiner struct {
@@ -350,8 +349,11 @@ func (j *leftOuterJoiner) tryToMatch(outer chunk.Row, inners chunk.Iterator, chk
 	}
 
 	// reach here, chkForJoin is j.chk
-	matched, err := j.filter(chkForJoin, chk)
-	return matched, errors.Trace(err)
+	matched, err := j.filter(chkForJoin, chk, outer.Len())
+	if err != nil {
+		return false, errors.Trace(err)
+	}
+	return matched, nil
 }
 
 func (j *leftOuterJoiner) onMissMatch(outer chunk.Row, chk *chunk.Chunk) {
@@ -384,9 +386,11 @@ func (j *rightOuterJoiner) tryToMatch(outer chunk.Row, inners chunk.Iterator, ch
 		return true, nil
 	}
 
-	// reach here, chkForJoin is j.chk
-	matched, err := j.filter(chkForJoin, chk)
-	return matched, errors.Trace(err)
+	matched, err := j.filter(chkForJoin, chk, outer.Len())
+	if err != nil {
+		return false, errors.Trace(err)
+	}
+	return matched, nil
 }
 
 func (j *rightOuterJoiner) onMissMatch(outer chunk.Row, chk *chunk.Chunk) {
@@ -421,8 +425,12 @@ func (j *innerJoiner) tryToMatch(outer chunk.Row, inners chunk.Iterator, chk *ch
 	}
 
 	// reach here, chkForJoin is j.chk
-	matched, err := j.filter(chkForJoin, chk)
-	return matched, errors.Trace(err)
+	matched, err := j.filter(chkForJoin, chk, outer.Len())
+	if err != nil {
+		return false, errors.Trace(err)
+	}
+	return matched, nil
+
 }
 
 func (j *innerJoiner) onMissMatch(outer chunk.Row, chk *chunk.Chunk) {

util/chunk/chunk_util.go

@@ -0,0 +1,99 @@
+// Copyright 2018 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package chunk
+
+// CopySelectedJoinRows uses for join to batch copy inner rows and outer row to chunk.
+// This function optimize for join. To be exact, `copyOuterRows` optimizes copy outer row to `dst` chunk.
+// Because the outer row in join is always same. so we can use batch copy for outer row data.
+func CopySelectedJoinRows(src *Chunk, innerColOffset, outerColOffset int, selected []bool, dst *Chunk) bool {
+	if src.NumRows() == 0 {
+		return false
+	}
+
+	numSelected := copySelectedInnerRows(innerColOffset, outerColOffset, src, selected, dst)
+	copyOuterRows(innerColOffset, outerColOffset, src, numSelected, dst)
+	dst.numVirtualRows += numSelected
+	return numSelected > 0
+}
+
+// copySelectedInnerRows appends different inner rows to the chunk.
+func copySelectedInnerRows(innerColOffset, outerColOffset int, src *Chunk, selected []bool, dst *Chunk) int {
+	oldLen := dst.columns[innerColOffset].length
+	var srcCols []*column
+	if innerColOffset == 0 {
+		srcCols = src.columns[:outerColOffset]
+	} else {
+		srcCols = src.columns[innerColOffset:]
+	}
+	for j, srcCol := range srcCols {
+		dstCol := dst.columns[innerColOffset+j]
+		if srcCol.isFixed() {
+			for i := 0; i < len(selected); i++ {
+				if !selected[i] {
+					continue
+				}
+				dstCol.appendNullBitmap(!srcCol.isNull(i))
+				dstCol.length++
+
+				elemLen := len(srcCol.elemBuf)
+				offset := i * elemLen
+				dstCol.data = append(dstCol.data, srcCol.data[offset:offset+elemLen]...)
+			}
+		} else {
+			for i := 0; i < len(selected); i++ {
+				if !selected[i] {
+					continue
+				}
+				dstCol.appendNullBitmap(!srcCol.isNull(i))
+				dstCol.length++
+
+				start, end := srcCol.offsets[i], srcCol.offsets[i+1]
+				dstCol.data = append(dstCol.data, srcCol.data[start:end]...)
+				dstCol.offsets = append(dstCol.offsets, int32(len(dstCol.data)))
+			}
+		}
+	}
+	return dst.columns[innerColOffset].length - oldLen
+}
+
+// copyOuterRows appends same outer row to the chunk with `numRows` times.
+func copyOuterRows(innerColOffset, outerColOffset int, src *Chunk, numRows int, dst *Chunk) {
+	row := src.GetRow(0)
+	var srcCols []*column
+	if innerColOffset == 0 {
+		srcCols = src.columns[outerColOffset:]
+	} else {
+		srcCols = src.columns[:innerColOffset]
+	}
+	for i, srcCol := range srcCols {
+		dstCol := dst.columns[outerColOffset+i]
+		dstCol.appendMultiSameNullBitmap(!srcCol.isNull(row.idx), numRows)
+		dstCol.length += numRows
+		if srcCol.isFixed() {
+			elemLen := len(srcCol.elemBuf)
+			start := row.idx * elemLen
+			end := start + numRows*elemLen
+			dstCol.data = append(dstCol.data, srcCol.data[start:end]...)
+		} else {
+			start, end := srcCol.offsets[row.idx], srcCol.offsets[row.idx+numRows]
+			dstCol.data = append(dstCol.data, srcCol.data[start:end]...)
+			offsets := dstCol.offsets
+			l := srcCol.offsets[row.idx+1] - srcCol.offsets[row.idx]
+			for j := 0; j < numRows; j++ {
+				offsets = append(offsets, int32(offsets[len(offsets)-1]+l))
+			}
+			dstCol.offsets = offsets
+		}
+	}
+}

util/chunk/chunk_util_test.go

@@ -0,0 +1,83 @@
+// Copyright 2018 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package chunk
+
+import (
+	"reflect"
+	"testing"
+
+	"github.com/pingcap/tidb/types"
+)
+
+func getChk() (*Chunk, *Chunk, []bool) {
+	numRows := 1024
+	srcChk := newChunkWithInitCap(numRows, 0, 0, 8, 8, 16, 0)
+	selected := make([]bool, numRows)
+	var row Row
+	for j := 0; j < numRows; j++ {
+		if j%7 == 0 {
+			row = MutRowFromValues("abc", "abcdefg", nil, 123, types.ZeroDatetime, "abcdefg").ToRow()
+		} else {
+			row = MutRowFromValues("abc", "abcdefg", j, 123, types.ZeroDatetime, "abcdefg").ToRow()
+			selected[j] = true
+		}
+		srcChk.AppendPartialRow(0, row)
+	}
+	dstChk := newChunkWithInitCap(numRows, 0, 0, 8, 8, 16, 0)
+	return srcChk, dstChk, selected
+}
+
+func TestBatchCopyJoinRowToChunk(t *testing.T) {
+	srcChk, dstChk, selected := getChk()
+	numRows := srcChk.NumRows()
+	for i := 0; i < numRows; i++ {
+		if !selected[i] {
+			continue
+		}
+		dstChk.AppendRow(srcChk.GetRow(i))
+	}
+	// batch copy
+	dstChk2 := newChunkWithInitCap(numRows, 0, 0, 8, 8, 16, 0)
+	CopySelectedJoinRows(srcChk, 0, 3, selected, dstChk2)
+
+	if !reflect.DeepEqual(dstChk, dstChk2) {
+		t.Fatal()
+	}
+}
+
+func BenchmarkChunkBatchCopyJoinRow(b *testing.B) {
+	b.ReportAllocs()
+	srcChk, dstChk, selected := getChk()
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		dstChk.Reset()
+		CopySelectedJoinRows(srcChk, 0, 3, selected, dstChk)
+	}
+}
+
+func BenchmarkChunkAppendRow(b *testing.B) {
+	b.ReportAllocs()
+	srcChk, dstChk, selected := getChk()
+	numRows := srcChk.NumRows()
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		dstChk.Reset()
+		for j := 0; j < numRows; j++ {
+			if !selected[j] {
+				continue
+			}
+			dstChk.AppendRow(srcChk.GetRow(j))
+		}
+	}
+}

util/chunk/column.go

@@ -85,6 +85,34 @@ func (c *column) appendNullBitmap(on bool) {
 	}
 }
 
+func (c *column) appendMultiSameNullBitmap(on bool, num int) {
+	l := ((c.length + num - 1) >> 3) - len(c.nullBitmap)
+	for i := 0; i <= l; i++ {
+		c.nullBitmap = append(c.nullBitmap, 0)
+	}
+	if on {
+		idx := c.length >> 3
+		pos := uint(c.length) & 7
+		for num > 0 {
+			if pos == 0 && num > 8 {
+				c.nullBitmap[idx] = 0xff
+				idx++
+				num = num - 8
+			} else {
+				c.nullBitmap[idx] |= byte(1 << pos)
+				pos++
+				num--
+				if pos == 8 {
+					pos = 0
+					idx++
+				}
+			}
+		}
+	} else {
+		c.nullCount += num
+	}
+}
+
 func (c *column) appendNull() {
 	c.appendNullBitmap(false)
 	if c.isFixed() {