executor: support hash join spill

pkg/executor/benchmark_test.go

@@ -732,7 +732,7 @@ func prepare4HashJoinV2(testCase *hashJoinTestCase, innerExec, outerExec exec.Ex
 	for i := 0; i < testCase.concurrency; i++ {
 		e.ProbeWorkers[i] = &join.ProbeWorkerV2{
 			HashJoinCtx: e.HashJoinCtxV2,
-			JoinProbe:   join.NewJoinProbe(e.HashJoinCtxV2, uint(i), testCase.joinType, probeKeysColIdx, joinedTypes, probeKeyTypes, false),
+			JoinProbe:   join.NewJoinProbe(e.HashJoinCtxV2, uint(i), testCase.joinType, probeKeysColIdx, joinedTypes, probeKeyTypes, false, nil),
 		}
 		e.ProbeWorkers[i].WorkerID = uint(i)
 		e.BuildWorkers[i] = join.NewJoinBuildWorkerV2(e.HashJoinCtxV2, uint(i), innerExec, joinKeysColIdx, innerTypes)

pkg/executor/builder.go

@@ -1654,7 +1654,7 @@ func (b *executorBuilder) buildHashJoinV2(v *plannercore.PhysicalHashJoin) exec.
 	for i := uint(0); i < e.Concurrency; i++ {
 		e.ProbeWorkers[i] = &join.ProbeWorkerV2{
 			HashJoinCtx: e.HashJoinCtxV2,
-			JoinProbe:   join.NewJoinProbe(e.HashJoinCtxV2, i, v.JoinType, probeKeyColIdx, joinedTypes, e.ProbeKeyTypes, e.RightAsBuildSide),
+			JoinProbe:   join.NewJoinProbe(e.HashJoinCtxV2, i, v.JoinType, probeKeyColIdx, joinedTypes, e.ProbeKeyTypes, e.RightAsBuildSide, e.ProbeSideTupleFetcher.ProbeSideExec.RetFieldTypes()),
 		}
 		e.ProbeWorkers[i].WorkerID = i
 

pkg/executor/internal/testutil/testutil.go

@@ -49,6 +49,12 @@ type MockDataSourceParameters struct {
 	Ndvs        []int
 	Orders      []bool
 	Rows        int
+
+	// Sometimes, user wants to manually provide test data
+	// and he can save provided test data at here.
+	Datums [][]any
+
+	HasSel bool
 }
 
 // MockDataSource mocks data source
@@ -73,6 +79,7 @@ func (mds *MockDataSource) GenColDatums(col int) (results []any) {
 		ndv = mds.P.Ndvs[col]
 	}
 	results = make([]any, 0, rows)
+
 	if ndv == 0 {
 		if mds.P.GenDataFunc == nil {
 			for i := 0; i < rows; i++ {
@@ -84,20 +91,28 @@ func (mds *MockDataSource) GenColDatums(col int) (results []any) {
 			}
 		}
 	} else {
-		datumSet := make(map[string]bool, ndv)
-		datums := make([]any, 0, ndv)
-		for len(datums) < ndv {
-			d := mds.RandDatum(typ)
-			str := fmt.Sprintf("%v", d)
-			if datumSet[str] {
-				continue
+		datums := make([]any, 0, max(ndv, 0))
+		if ndv == -1 {
+			if mds.P.Datums[col] == nil {
+				panic("need to provid data")
+			}
+
+			datums = mds.P.Datums[col]
+		} else {
+			datumSet := make(map[string]bool, ndv)
+			for len(datums) < ndv {
+				d := mds.RandDatum(typ)
+				str := fmt.Sprintf("%v", d)
+				if datumSet[str] {
+					continue
+				}
+				datumSet[str] = true
+				datums = append(datums, d)
 			}
-			datumSet[str] = true
-			datums = append(datums, d)
 		}
 
 		for i := 0; i < rows; i++ {
-			val, err := rand.Int(rand.Reader, big.NewInt(int64(ndv)))
+			val, err := rand.Int(rand.Reader, big.NewInt(int64(len(datums))))
 			if err != nil {
 				panic("Fail to generate int number")
 			}
@@ -272,6 +287,18 @@ func BuildMockDataSource(opt MockDataSourceParameters) *MockDataSource {
 			}
 		}
 	}
+
+	if opt.HasSel {
+		for _, chk := range m.GenData {
+			rowNum := chk.NumRows()
+			sel := make([]int, 0, rowNum/2)
+			for i := mathrand.Int31n(2); int(i) < rowNum; i += 2 {
+				sel = append(sel, int(i))
+			}
+			chk.SetSel(sel)
+		}
+	}
+
 	return m
 }
 

pkg/executor/join/BUILD.bazel

@@ -6,6 +6,9 @@ go_library(
         "base_join_probe.go",
         "concurrent_map.go",
         "hash_join_base.go",
+        "hash_join_spill.go",
+        "hash_join_spill_helper.go",
+        "hash_join_test_util.go",
         "hash_join_v1.go",
         "hash_join_v2.go",
         "hash_table_v1.go",
@@ -28,6 +31,7 @@ go_library(
         "//pkg/executor/aggregate",
         "//pkg/executor/internal/applycache",
         "//pkg/executor/internal/exec",
+        "//pkg/executor/internal/testutil",
         "//pkg/executor/internal/vecgroupchecker",
         "//pkg/executor/unionexec",
         "//pkg/expression",
@@ -58,6 +62,7 @@ go_library(
         "//pkg/util/syncutil",
         "@com_github_pingcap_errors//:errors",
         "@com_github_pingcap_failpoint//:failpoint",
+        "@com_github_stretchr_testify//require",
         "@org_uber_go_zap//:zap",
     ],
 )
@@ -73,12 +78,14 @@ go_test(
         "index_lookup_join_test.go",
         "index_lookup_merge_join_test.go",
         "inner_join_probe_test.go",
+        "inner_join_spill_test.go",
         "join_row_table_test.go",
         "join_stats_test.go",
         "join_table_meta_test.go",
         "joiner_test.go",
         "left_outer_join_probe_test.go",
         "merge_join_test.go",
+        "outer_join_spill_test.go",
         "right_outer_join_probe_test.go",
         "row_table_builder_test.go",
         "tagged_ptr_test.go",