executor: solve bug of copy joined tuples after inline projection

executor/join_test.go

@@ -1888,3 +1888,36 @@ func (s *testSuiteJoinSerial) TestOuterMatchStatusIssue14742(c *C) {
 		"2 1",
 	))
 }
+
+func (s *testSuiteJoinSerial) TestInlineProjection4HashJoinIssue15316(c *C) {
+	// Two necessary factors to reproduce this issue:
+	// (1) taking HashLeftJoin, i.e., letting the probing tuple lay at the left side of joined tuples
+	// (2) the projection only contains a part of columns from the build side, i.e., pruning the same probe side
+	plannercore.ForcedHashLeftJoin4Test = true
+	defer func() { plannercore.ForcedHashLeftJoin4Test = false }()
+	tk := testkit.NewTestKit(c, s.store)
+	tk.MustExec("use test")
+	tk.MustExec("create table S (a int not null, b int, c int);")
+	tk.MustExec("create table T (a int not null, b int, c int);")
+	tk.MustExec("insert into S values (0,1,2),(0,1,null),(0,1,2);")
+	tk.MustExec("insert into T values (0,10,2),(0,10,null),(1,10,2);")
+	tk.MustQuery("select T.a,T.a,T.c from S join T on T.a = S.a where S.b<T.b order by T.a,T.c;").Check(testkit.Rows(
+		"0 0 <nil>",
+		"0 0 <nil>",
+		"0 0 <nil>",
+		"0 0 2",
+		"0 0 2",
+		"0 0 2",
+	))
+	// NOTE: the HashLeftJoin should be kept
+	tk.MustQuery("explain select T.a,T.a,T.c from S join T on T.a = S.a where S.b<T.b order by T.a,T.c;").Check(testkit.Rows(
+		"Sort_8 12487.50 root test.t.a:asc, test.t.c:asc",
+		"└─Projection_10 12487.50 root test.t.a, test.t.a, test.t.c",
+		"  └─HashLeftJoin_11 12487.50 root inner join, equal:[eq(test.s.a, test.t.a)], other cond:lt(test.s.b, test.t.b)",
+		"    ├─TableReader_17(Build) 9990.00 root data:Selection_16",
+		"    │ └─Selection_16 9990.00 cop[tikv] not(isnull(test.t.b))",
+		"    │   └─TableFullScan_15 10000.00 cop[tikv] table:T, keep order:false, stats:pseudo",
+		"    └─TableReader_14(Probe) 9990.00 root data:Selection_13",
+		"      └─Selection_13 9990.00 cop[tikv] not(isnull(test.s.b))",
+		"        └─TableFullScan_12 10000.00 cop[tikv] table:S, keep order:false, stats:pseudo"))
+}

executor/joiner.go

@@ -241,6 +241,7 @@ func (j *baseJoiner) filter(
 	}
 	// Batch copies selected rows to output chunk.
 	innerColOffset, outerColOffset := 0, input.NumCols()-outerColsLen
+	innerColLen, outerColLen := input.NumCols()-outerColsLen, outerColsLen
 	if !j.outerIsRight {
 		innerColOffset, outerColOffset = outerColsLen, 0
 	}
@@ -257,11 +258,14 @@ func (j *baseJoiner) filter(
 		input = input.Prune(used)
 
 		innerColOffset, outerColOffset = 0, len(lUsed)
+		innerColLen, outerColLen = len(lUsed), len(rUsed)
 		if !j.outerIsRight {
 			innerColOffset, outerColOffset = len(lUsed), 0
+			innerColLen, outerColLen = outerColLen, innerColLen
 		}
+
 	}
-	return chunk.CopySelectedJoinRowsWithSameOuterRows(input, innerColOffset, outerColOffset, j.selected, output)
+	return chunk.CopySelectedJoinRowsWithSameOuterRows(input, innerColOffset, innerColLen, outerColOffset, outerColLen, j.selected, output)
 }
 
 // filterAndCheckOuterRowStatus is used to filter the result constructed by

planner/core/exhaust_physical_plans.go

@@ -289,6 +289,10 @@ func (p *PhysicalMergeJoin) initCompareFuncs() {
 // TODO: use hint and remove this variable
 var ForceUseOuterBuild4Test = false
 
+// ForcedHashLeftJoin4Test is a test option to force using HashLeftJoin
+// TODO: use hint and remove this variable
+var ForcedHashLeftJoin4Test = false
+
 func (p *LogicalJoin) getHashJoins(prop *property.PhysicalProperty) []PhysicalPlan {
 	if !prop.IsEmpty() { // hash join doesn't promise any orders
 		return nil
@@ -312,8 +316,12 @@ func (p *LogicalJoin) getHashJoins(prop *property.PhysicalProperty) []PhysicalPl
 			joins = append(joins, p.getHashJoin(prop, 0, true))
 		}
 	case InnerJoin:
-		joins = append(joins, p.getHashJoin(prop, 1, false))
-		joins = append(joins, p.getHashJoin(prop, 0, false))
+		if ForcedHashLeftJoin4Test {
+			joins = append(joins, p.getHashJoin(prop, 1, false))
+		} else {
+			joins = append(joins, p.getHashJoin(prop, 1, false))
+			joins = append(joins, p.getHashJoin(prop, 0, false))
+		}
 	}
 	return joins
 }

util/chunk/chunk_util.go

@@ -75,30 +75,25 @@ func CopySelectedJoinRowsDirect(src *Chunk, selected []bool, dst *Chunk) (bool,
 // Return true if at least one joined row was selected.
 //
 // NOTE: All the outer rows in the source Chunk should be the same.
-func CopySelectedJoinRowsWithSameOuterRows(src *Chunk, innerColOffset, outerColOffset int, selected []bool, dst *Chunk) (bool, error) {
+func CopySelectedJoinRowsWithSameOuterRows(src *Chunk, innerColOffset, innerColLen, outerColOffset, outerColLen int, selected []bool, dst *Chunk) (bool, error) {
 	if src.NumRows() == 0 {
 		return false, nil
 	}
 	if src.sel != nil || dst.sel != nil {
 		return false, errors.New(msgErrSelNotNil)
 	}
 
-	numSelected := copySelectedInnerRows(innerColOffset, outerColOffset, src, selected, dst)
-	copyOuterRows(innerColOffset, outerColOffset, src, numSelected, dst)
+	numSelected := copySelectedInnerRows(innerColOffset, innerColLen, src, selected, dst)
+	copySameOuterRows(outerColOffset, outerColLen, src, numSelected, dst)
 	dst.numVirtualRows += numSelected
 	return numSelected > 0, nil
 }
 
 // copySelectedInnerRows copies the selected inner rows from the source Chunk
 // to the destination Chunk.
 // return the number of rows which is selected.
-func copySelectedInnerRows(innerColOffset, outerColOffset int, src *Chunk, selected []bool, dst *Chunk) int {
-	var srcCols []*Column
-	if innerColOffset == 0 {
-		srcCols = src.columns[:outerColOffset]
-	} else {
-		srcCols = src.columns[innerColOffset:]
-	}
+func copySelectedInnerRows(innerColOffset, innerColLen int, src *Chunk, selected []bool, dst *Chunk) int {
+	srcCols := src.columns[innerColOffset : innerColOffset+innerColLen]
 	if len(srcCols) == 0 {
 		numSelected := 0
 		for _, s := range selected {
@@ -140,19 +135,14 @@ func copySelectedInnerRows(innerColOffset, outerColOffset int, src *Chunk, selec
 	return dst.columns[innerColOffset].length - oldLen
 }
 
-// copyOuterRows copies the continuous 'numRows' outer rows in the source Chunk
+// copySameOuterRows copies the continuous 'numRows' outer rows in the source Chunk
 // to the destination Chunk.
-func copyOuterRows(innerColOffset, outerColOffset int, src *Chunk, numRows int, dst *Chunk) {
-	if numRows <= 0 {
+func copySameOuterRows(outerColOffset, outerColLen int, src *Chunk, numRows int, dst *Chunk) {
+	if numRows <= 0 || outerColLen <= 0 {
 		return
 	}
 	row := src.GetRow(0)
-	var srcCols []*Column
-	if innerColOffset == 0 {
-		srcCols = src.columns[outerColOffset:]
-	} else {
-		srcCols = src.columns[:innerColOffset]
-	}
+	srcCols := src.columns[outerColOffset : outerColOffset+outerColLen]
 	for i, srcCol := range srcCols {
 		dstCol := dst.columns[outerColOffset+i]
 		dstCol.appendMultiSameNullBitmap(!srcCol.IsNull(row.idx), numRows)

util/chunk/chunk_util_test.go

@@ -21,7 +21,7 @@ import (
 	"github.com/pingcap/tidb/types"
 )
 
-// getChk generate a chunk of data, isFirst3ColTheSame means the first three columns are the same.
+// getChk generate a chunk of data, isLast3ColTheSame means the last three columns are the same.
 func getChk(isLast3ColTheSame bool) (*Chunk, *Chunk, []bool) {
 	numRows := 1024
 	srcChk := newChunkWithInitCap(numRows, 0, 0, 8, 8, sizeTime, 0)
@@ -61,7 +61,34 @@ func TestCopySelectedJoinRows(t *testing.T) {
 	}
 	// batch copy
 	dstChk2 := newChunkWithInitCap(numRows, 0, 0, 8, 8, sizeTime, 0)
-	CopySelectedJoinRowsWithSameOuterRows(srcChk, 0, 3, selected, dstChk2)
+	CopySelectedJoinRowsWithSameOuterRows(srcChk, 0, 3, 3, 3, selected, dstChk2)
+
+	if !reflect.DeepEqual(dstChk, dstChk2) {
+		t.Fatal()
+	}
+	numSelected := 0
+	for i := range selected {
+		if selected[i] {
+			numSelected++
+		}
+	}
+	if dstChk2.numVirtualRows != numSelected || dstChk2.NumRows() != numSelected {
+		t.Fatal(dstChk2.numVirtualRows, dstChk2.NumRows(), numSelected)
+	}
+}
+
+func TestCopySelectedJoinRowsWithoutSameOuters(t *testing.T) {
+	srcChk, dstChk, selected := getChk(false)
+	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, sizeTime, 0)
+	CopySelectedJoinRowsWithSameOuterRows(srcChk, 0, 6, 0, 0, selected, dstChk2)
 
 	if !reflect.DeepEqual(dstChk, dstChk2) {
 		t.Fatal()
@@ -105,6 +132,7 @@ func TestCopySelectedJoinRowsDirect(t *testing.T) {
 }
 
 func TestCopySelectedVirtualNum(t *testing.T) {
+	// srcChk does not contain columns
 	srcChk := newChunk()
 	srcChk.TruncateTo(3)
 	dstChk := newChunk()
@@ -118,7 +146,7 @@ func TestCopySelectedVirtualNum(t *testing.T) {
 	}
 
 	dstChk = newChunk()
-	ok, err = CopySelectedJoinRowsWithSameOuterRows(srcChk, 0, 0, selected, dstChk)
+	ok, err = CopySelectedJoinRowsWithSameOuterRows(srcChk, 0, 0, 0, 0, selected, dstChk)
 	if err != nil || !ok {
 		t.Fatal(ok, err)
 	}
@@ -132,7 +160,7 @@ func TestCopySelectedVirtualNum(t *testing.T) {
 	srcChk.AppendInt64(0, 1)
 	srcChk.AppendInt64(0, 2)
 	dstChk = newChunkWithInitCap(0, 8)
-	ok, err = CopySelectedJoinRowsWithSameOuterRows(srcChk, 0, 1, selected, dstChk)
+	ok, err = CopySelectedJoinRowsWithSameOuterRows(srcChk, 0, 1, 1, 0, selected, dstChk)
 	if err != nil || !ok {
 		t.Fatal(ok, err)
 	}
@@ -149,7 +177,7 @@ func TestCopySelectedVirtualNum(t *testing.T) {
 	srcChk.AppendInt64(0, 3)
 	srcChk.AppendInt64(0, 3)
 	dstChk = newChunkWithInitCap(0, 8)
-	ok, err = CopySelectedJoinRowsWithSameOuterRows(srcChk, 1, 0, selected, dstChk)
+	ok, err = CopySelectedJoinRowsWithSameOuterRows(srcChk, 1, 0, 0, 1, selected, dstChk)
 	if err != nil || !ok {
 		t.Fatal(ok, err)
 	}
@@ -167,7 +195,7 @@ func BenchmarkCopySelectedJoinRows(b *testing.B) {
 	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
 		dstChk.Reset()
-		CopySelectedJoinRowsWithSameOuterRows(srcChk, 0, 3, selected, dstChk)
+		CopySelectedJoinRowsWithSameOuterRows(srcChk, 0, 3, 3, 3, selected, dstChk)
 	}
 }
 func BenchmarkCopySelectedJoinRowsDirect(b *testing.B) {