executor: support hash join spill

pkg/executor/internal/testutil/testutil.go

@@ -48,7 +48,13 @@ type MockDataSourceParameters struct {
 	GenDataFunc func(row int, typ *types.FieldType) any
 	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
+
+	Rows   int
+	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/join/joinversion",
         "//pkg/executor/unionexec",
@@ -50,6 +54,7 @@ go_library(
         "//pkg/util/disk",
         "//pkg/util/execdetails",
         "//pkg/util/hack",
+        "//pkg/util/intest",
         "//pkg/util/logutil",
         "//pkg/util/memory",
         "//pkg/util/mvmap",
@@ -59,6 +64,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",
     ],
 )
@@ -74,12 +80,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",