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

executor/joiner.go

@@ -158,19 +158,12 @@ 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) (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))
+		return errors.Trace(err)
 	}
-	return matched, nil
+	return nil
 }
 
 type semiJoiner struct {
@@ -350,8 +343,12 @@ 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)
+	err := j.filter(chkForJoin, chk)
+	if err != nil {
+		return false, errors.Trace(err)
+	}
+	matched := chk.BatchCopyJoinRowToChunk(false, chkForJoin, outer, j.selected)
+	return matched, nil
 }
 
 func (j *leftOuterJoiner) onMissMatch(outer chunk.Row, chk *chunk.Chunk) {
@@ -384,9 +381,12 @@ 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)
+	err := j.filter(chkForJoin, chk)
+	if err != nil {
+		return false, errors.Trace(err)
+	}
+	matched := chk.BatchCopyJoinRowToChunk(true, chkForJoin, outer, j.selected)
+	return matched, nil
 }
 
 func (j *rightOuterJoiner) onMissMatch(outer chunk.Row, chk *chunk.Chunk) {
@@ -421,8 +421,13 @@ 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)
+	err := j.filter(chkForJoin, chk)
+	if err != nil {
+		return false, errors.Trace(err)
+	}
+	matched := chk.BatchCopyJoinRowToChunk(j.outerIsRight, chkForJoin, outer, j.selected)
+	return matched, nil
+
 }
 
 func (j *innerJoiner) onMissMatch(outer chunk.Row, chk *chunk.Chunk) {

util/chunk/chunk.go

@@ -232,6 +232,77 @@ func (c *Chunk) AppendPartialRow(colIdx int, row Row) {
 	}
 }
 
+// BatchCopyJoinRowToChunk uses for join to batch copy inner rows and outer row to chunk.
+func (c *Chunk) BatchCopyJoinRowToChunk(isRight bool, chkForJoin *Chunk, outer Row, selected []bool) bool {
+	if chkForJoin.NumRows() == 0 {
+		return false
+	}
+
+	innerIdx, outerIdx := 0, chkForJoin.NumCols()-outer.Len()
+	if !isRight {
+		innerIdx, outerIdx = outer.Len(), 0
+	}
+	rowNum := appendPartialRows(innerIdx, outerIdx, chkForJoin, c, selected)
+	appendPartialSameRows(outerIdx, outer, rowNum, c)
+	c.numVirtualRows += rowNum
+	return rowNum > 0
+}
+
+// appendPartialRows appends multiple different rows to the chunk.
+func appendPartialRows(colIdx, outerIdx int, chkForJoin, chk *Chunk, selected []bool) int {
+	oldLen := chk.columns[colIdx].length
+	var columns []*column
+	if colIdx == 0 {
+		columns = chkForJoin.columns[:outerIdx]
+	} else {
+		columns = chkForJoin.columns[colIdx:]
+	}
+	for j, rowCol := range columns {
+		chkCol := chk.columns[colIdx+j]
+		for i := 0; i < len(selected); i++ {
+			if !selected[i] {
+				continue
+			}
+			chkCol.appendNullBitmap(!rowCol.isNull(i))
+			chkCol.length++
+
+			if rowCol.isFixed() {
+				elemLen := len(rowCol.elemBuf)
+				offset := i * elemLen
+				chkCol.data = append(chkCol.data, rowCol.data[offset:offset+elemLen]...)
+			} else {
+				start, end := rowCol.offsets[i], rowCol.offsets[i+1]
+				chkCol.data = append(chkCol.data, rowCol.data[start:end]...)
+				chkCol.offsets = append(chkCol.offsets, int32(len(chkCol.data)))
+			}
+		}
+	}
+	return chk.columns[colIdx].length - oldLen
+}
+
+// appendPartialSameRows appends same row to the chunk with `rowNum` times.
+func appendPartialSameRows(colIdx int, row Row, rowNum int, c *Chunk) {
+	for i, rowCol := range row.c.columns {
+		chkCol := c.columns[colIdx+i]
+		chkCol.appendMultiSameNullBitmap(!rowCol.isNull(row.idx), rowNum)
+		chkCol.length += rowNum
+		if rowCol.isFixed() {
+			elemLen := len(rowCol.elemBuf)
+			start := row.idx * elemLen
+			end := start + elemLen
+			for j := 0; j < rowNum; j++ {
+				chkCol.data = append(chkCol.data, rowCol.data[start:end]...)
+			}
+		} else {
+			start, end := rowCol.offsets[row.idx], rowCol.offsets[row.idx+1]
+			for j := 0; j < rowNum; j++ {
+				chkCol.data = append(chkCol.data, rowCol.data[start:end]...)
+				chkCol.offsets = append(chkCol.offsets, int32(len(chkCol.data)))
+			}
+		}
+	}
+}
+
 // Append appends rows in [begin, end) in another Chunk to a Chunk.
 func (c *Chunk) Append(other *Chunk, begin, end int) {
 	for colID, src := range other.columns {

util/chunk/chunk_test.go

@@ -17,6 +17,7 @@ import (
 	"bytes"
 	"fmt"
 	"math"
+	"reflect"
 	"strconv"
 	"testing"
 	"time"
@@ -657,6 +658,70 @@ func BenchmarkChunkMemoryUsage(b *testing.B) {
 	}
 }
 
+func getChk() (*Chunk, *Chunk, Row, int, []bool) {
+	numRows := 1024
+	srcChk := newChunkWithInitCap(numRows, 0, 0, 8, 8, 16)
+	srcChk.Reset()
+	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).ToRow()
+		} else {
+			row = MutRowFromValues("abc", "abcdefg", j, 123, types.ZeroDatetime).ToRow()
+		}
+		srcChk.AppendPartialRow(0, row)
+		selected[j] = true
+	}
+	dstChk := newChunkWithInitCap(numRows, 0, 0, 8, 8, 16)
+	outerRow := MutRowFromValues(123, types.ZeroDatetime).ToRow()
+
+	return srcChk, dstChk, outerRow, numRows, selected
+}
+
+func TestBatchCopyJoinRowToChunk(t *testing.T) {
+	srcChk, dstChk, outerRow, numRows, selected := getChk()
+	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)
+	dstChk2.BatchCopyJoinRowToChunk(true, srcChk, outerRow, selected)
+
+	if !reflect.DeepEqual(dstChk, dstChk2) {
+		t.Fatal()
+	}
+}
+
+func BenchmarkChunkBatchCopyJoinRow(b *testing.B) {
+	b.ReportAllocs()
+	srcChk, dstChk, outerRow, _, selected := getChk()
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		dstChk.Reset()
+		dstChk.BatchCopyJoinRowToChunk(false, srcChk, outerRow, selected)
+	}
+}
+
+func BenchmarkChunkAppendRow(b *testing.B) {
+	b.ReportAllocs()
+	srcChk, dstChk, _, numRows, selected := getChk()
+	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))
+		}
+	}
+}
+
 type seqNumberGenerateExec struct {
 	seq          int
 	genCountSize int

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() {