executor/join : use shallow copy for join.

executor/joiner.go

@@ -91,7 +91,7 @@ func newJoiner(ctx sessionctx.Context, joinType plan.JoinType,
 	colTypes := make([]*types.FieldType, 0, len(lhsColTypes)+len(rhsColTypes))
 	colTypes = append(colTypes, lhsColTypes...)
 	colTypes = append(colTypes, rhsColTypes...)
-	base.chk = chunk.NewChunkWithCapacity(colTypes, ctx.GetSessionVars().MaxChunkSize)
+	base.mutRow = chunk.MutRowFromTypes(colTypes)
 	base.selected = make([]bool, 0, chunk.InitialCapacity)
 	if joinType == plan.LeftOuterJoin || joinType == plan.RightOuterJoin {
 		innerColTypes := lhsColTypes
@@ -124,7 +124,7 @@ type baseJoiner struct {
 	conditions   []expression.Expression
 	defaultInner chunk.Row
 	outerIsRight bool
-	chk          *chunk.Chunk
+	mutRow       chunk.MutRow
 	selected     []bool
 	maxChunkSize int
 }
@@ -142,6 +142,16 @@ func (j *baseJoiner) makeJoinRowToChunk(chk *chunk.Chunk, lhs, rhs chunk.Row) {
 	chk.AppendPartialRow(lhs.Len(), rhs)
 }
 
+// makeShadowJoinRow combines inner, outer row into one join row and append to the chunk.
+// The result of join row will always append to the dst chunk's first row.
+func makeShadowJoinRow(isRight bool, inner, outer chunk.Row, mutRow chunk.MutRow) {
+	if !isRight {
+		inner, outer = outer, inner
+	}
+	chunk.ShadowCopyPartialRow(0, inner, mutRow)
+	chunk.ShadowCopyPartialRow(inner.Len(), outer, mutRow)
+}
+
 func (j *baseJoiner) filter(input, output *chunk.Chunk) (matched bool, err error) {
 	j.selected, err = expression.VectorizedFilter(j.ctx, j.conditions, chunk.NewIterator4Chunk(input), j.selected)
 	if err != nil {
@@ -173,14 +183,9 @@ func (j *semiJoiner) tryToMatch(outer chunk.Row, inners chunk.Iterator, chk *chu
 	}
 
 	for inner := inners.Current(); inner != inners.End(); inner = inners.Next() {
-		j.chk.Reset()
-		if j.outerIsRight {
-			j.makeJoinRowToChunk(j.chk, inner, outer)
-		} else {
-			j.makeJoinRowToChunk(j.chk, outer, inner)
-		}
+		makeShadowJoinRow(j.outerIsRight, inner, outer, j.mutRow)
 
-		matched, err = expression.EvalBool(j.ctx, j.conditions, j.chk.GetRow(0))
+		matched, err = expression.EvalBool(j.ctx, j.conditions, j.mutRow.ToRow())
 		if err != nil {
 			return false, errors.Trace(err)
 		}
@@ -212,14 +217,9 @@ func (j *antiSemiJoiner) tryToMatch(outer chunk.Row, inners chunk.Iterator, chk
 	}
 
 	for inner := inners.Current(); inner != inners.End(); inner = inners.Next() {
-		j.chk.Reset()
-		if j.outerIsRight {
-			j.makeJoinRowToChunk(j.chk, inner, outer)
-		} else {
-			j.makeJoinRowToChunk(j.chk, outer, inner)
-		}
+		makeShadowJoinRow(j.outerIsRight, inner, outer, j.mutRow)
 
-		matched, err = expression.EvalBool(j.ctx, j.conditions, j.chk.GetRow(0))
+		matched, err = expression.EvalBool(j.ctx, j.conditions, j.mutRow.ToRow())
 		if err != nil {
 			return false, errors.Trace(err)
 		}
@@ -252,10 +252,9 @@ func (j *leftOuterSemiJoiner) tryToMatch(outer chunk.Row, inners chunk.Iterator,
 	}
 
 	for inner := inners.Current(); inner != inners.End(); inner = inners.Next() {
-		j.chk.Reset()
-		j.makeJoinRowToChunk(j.chk, outer, inner)
+		makeShadowJoinRow(false, inner, outer, j.mutRow)
 
-		matched, err = expression.EvalBool(j.ctx, j.conditions, j.chk.GetRow(0))
+		matched, err = expression.EvalBool(j.ctx, j.conditions, j.mutRow.ToRow())
 		if err != nil {
 			return false, errors.Trace(err)
 		}
@@ -295,10 +294,9 @@ func (j *antiLeftOuterSemiJoiner) tryToMatch(outer chunk.Row, inners chunk.Itera
 	}
 
 	for inner := inners.Current(); inner != inners.End(); inner = inners.Next() {
-		j.chk.Reset()
-		j.makeJoinRowToChunk(j.chk, outer, inner)
-		matched, err := expression.EvalBool(j.ctx, j.conditions, j.chk.GetRow(0))
+		makeShadowJoinRow(false, inner, outer, j.mutRow)
 
+		matched, err := expression.EvalBool(j.ctx, j.conditions, j.mutRow.ToRow())
 		if err != nil {
 			return false, errors.Trace(err)
 		}
@@ -331,24 +329,8 @@ func (j *leftOuterJoiner) tryToMatch(outer chunk.Row, inners chunk.Iterator, chk
 		return false, nil
 	}
 
-	j.chk.Reset()
-	chkForJoin := j.chk
-	if len(j.conditions) == 0 {
-		chkForJoin = chk
-	}
-
 	numToAppend := j.maxChunkSize - chk.NumRows()
-	for ; inners.Current() != inners.End() && numToAppend > 0; numToAppend-- {
-		j.makeJoinRowToChunk(chkForJoin, outer, inners.Current())
-		inners.Next()
-	}
-	if len(j.conditions) == 0 {
-		return true, nil
-	}
-
-	// reach here, chkForJoin is j.chk
-	matched, err := j.filter(chkForJoin, chk)
-	return matched, errors.Trace(err)
+	return tryToMatchInerAndOuter(j.ctx, false, outer, inners, j.conditions, j.mutRow, chk, numToAppend)
 }
 
 func (j *leftOuterJoiner) onMissMatch(outer chunk.Row, chk *chunk.Chunk) {
@@ -366,24 +348,8 @@ func (j *rightOuterJoiner) tryToMatch(outer chunk.Row, inners chunk.Iterator, ch
 		return false, nil
 	}
 
-	j.chk.Reset()
-	chkForJoin := j.chk
-	if len(j.conditions) == 0 {
-		chkForJoin = chk
-	}
-
 	numToAppend := j.maxChunkSize - chk.NumRows()
-	for ; inners.Current() != inners.End() && numToAppend > 0; numToAppend-- {
-		j.makeJoinRowToChunk(chkForJoin, inners.Current(), outer)
-		inners.Next()
-	}
-	if len(j.conditions) == 0 {
-		return true, nil
-	}
-
-	// reach here, chkForJoin is j.chk
-	matched, err := j.filter(chkForJoin, chk)
-	return matched, errors.Trace(err)
+	return tryToMatchInerAndOuter(j.ctx, true, outer, inners, j.conditions, j.mutRow, chk, numToAppend)
 }
 
 func (j *rightOuterJoiner) onMissMatch(outer chunk.Row, chk *chunk.Chunk) {
@@ -400,26 +366,26 @@ func (j *innerJoiner) tryToMatch(outer chunk.Row, inners chunk.Iterator, chk *ch
 	if inners.Len() == 0 {
 		return false, nil
 	}
-	j.chk.Reset()
-	chkForJoin := j.chk
-	if len(j.conditions) == 0 {
-		chkForJoin = chk
-	}
-	inner, numToAppend := inners.Current(), j.maxChunkSize-chk.NumRows()
-	for ; inner != inners.End() && numToAppend > 0; inner, numToAppend = inners.Next(), numToAppend-1 {
-		if j.outerIsRight {
-			j.makeJoinRowToChunk(chkForJoin, inner, outer)
-		} else {
-			j.makeJoinRowToChunk(chkForJoin, outer, inner)
+
+	numToAppend := j.maxChunkSize - chk.NumRows()
+	return tryToMatchInerAndOuter(j.ctx, j.outerIsRight, outer, inners, j.conditions, j.mutRow, chk, numToAppend)
+}
+
+func tryToMatchInerAndOuter(ctx sessionctx.Context, isRight bool, outer chunk.Row, inners chunk.Iterator, conditions []expression.Expression, mutRow chunk.MutRow, outChk *chunk.Chunk, numToAppend int) (bool, error) {
+	match := false
+	for inner := inners.Current(); inner != inners.End() && numToAppend > 0; inner, numToAppend = inners.Next(), numToAppend-1 {
+		makeShadowJoinRow(isRight, inner, outer, mutRow)
+
+		matched, err := expression.VectorizedFilterOneRow(ctx, conditions, mutRow.ToRow())
+		if err != nil {
+			return false, errors.Trace(err)
+		}
+		if matched {
+			match = true
+			outChk.AppendRow(mutRow.ToRow())
 		}
 	}
-	if len(j.conditions) == 0 {
-		return true, nil
-	}
-
-	// reach here, chkForJoin is j.chk
-	matched, err := j.filter(chkForJoin, chk)
-	return matched, errors.Trace(err)
+	return match, nil
 }
 
 func (j *innerJoiner) onMissMatch(outer chunk.Row, chk *chunk.Chunk) {

expression/chunk_executor.go

@@ -237,15 +237,12 @@ func VectorizedFilter(ctx sessionctx.Context, filters []Expression, iterator *ch
 		selected = append(selected, true)
 	}
 	for _, filter := range filters {
-		isIntType := true
-		if filter.GetType().EvalType() != types.ETInt {
-			isIntType = false
-		}
+		isTypeInt := filter.GetType().EvalType() == types.ETInt
 		for row := iterator.Begin(); row != iterator.End(); row = iterator.Next() {
 			if !selected[row.Idx()] {
 				continue
 			}
-			if isIntType {
+			if isTypeInt {
 				filterResult, isNull, err := filter.EvalInt(ctx, row)
 				if err != nil {
 					return nil, errors.Trace(err)
@@ -263,3 +260,28 @@ func VectorizedFilter(ctx sessionctx.Context, filters []Expression, iterator *ch
 	}
 	return selected, nil
 }
+
+// VectorizedFilterOneRow applies a list of filters to a row.
+func VectorizedFilterOneRow(ctx sessionctx.Context, filters []Expression, row chunk.Row) (bool, error) {
+	selected := true
+	for _, filter := range filters {
+		isTypeInt := filter.GetType().EvalType() == types.ETInt
+		if isTypeInt {
+			filterResult, isNull, err := filter.EvalInt(ctx, row)
+			if err != nil {
+				return false, errors.Trace(err)
+			}
+			selected = selected && !isNull && (filterResult != 0)
+		} else {
+			// TODO: should rewrite the filter to `cast(expr as SIGNED) != 0` and always use `EvalInt`.
+			bVal, err := EvalBool(ctx, []Expression{filter}, row)
+			if err != nil {
+				return false, errors.Trace(err)
+			}
+			selected = selected && bVal
+		}
+
+	}
+	return selected, nil
+
+}