executor: disable probe short path in hash join (pingcap#53079)

close pingcap#51998
RidRisR · May 23, 2024 · a79ac6b · a79ac6b
1 parent d1547ff
commit a79ac6b
Show file tree

Hide file tree

Showing 3 changed files with 26 additions and 4 deletions.
diff --git a/pkg/executor/join/join.go b/pkg/executor/join/join.go
@@ -272,9 +272,6 @@ func (fetcher *ProbeSideTupleFetcher) fetchProbeSideChunks(ctx context.Context,
 					probeSideResult.Reset()
 				}
 			})
-			if probeSideResult.NumRows() == 0 && !fetcher.UseOuterToBuild {
-				fetcher.finished.Store(true)
-			}
 			emptyBuild, buildErr := fetcher.wait4BuildSide()
 			if buildErr != nil {
 				fetcher.joinResultCh <- &hashjoinWorkerResult{
@@ -332,12 +329,22 @@ func (w *BuildWorker) fetchBuildSideRows(ctx context.Context, chkCh chan<- *chun
 		}
 	})
 	sessVars := w.HashJoinCtx.SessCtx.GetSessionVars()
+	failpoint.Inject("issue51998", func(val failpoint.Value) {
+		if val.(bool) {
+			time.Sleep(2 * time.Second)
+		}
+	})
 	for {
 		if w.HashJoinCtx.finished.Load() {
 			return
 		}
 		chk := w.HashJoinCtx.ChunkAllocPool.Alloc(w.BuildSideExec.RetFieldTypes(), sessVars.MaxChunkSize, sessVars.MaxChunkSize)
 		err = exec.Next(ctx, w.BuildSideExec, chk)
+		failpoint.Inject("issue51998", func(val failpoint.Value) {
+			if val.(bool) {
+				err = errors.Errorf("issue51998 build return error")
+			}
+		})
 		if err != nil {
 			errCh <- errors.Trace(err)
 			return

diff --git a/pkg/executor/test/issuetest/BUILD.bazel b/pkg/executor/test/issuetest/BUILD.bazel
@@ -8,7 +8,7 @@ go_test(
         "main_test.go",
     ],
     flaky = True,
-    shard_count = 19,
+    shard_count = 20,
     deps = [
         "//pkg/autoid_service",
         "//pkg/config",

diff --git a/pkg/executor/test/issuetest/executor_issue_test.go b/pkg/executor/test/issuetest/executor_issue_test.go
@@ -183,6 +183,21 @@ func TestIssue30289(t *testing.T) {
 	require.EqualError(t, err, "issue30289 build return error")
 }
 
+func TestIssue51998(t *testing.T) {
+	fpName := "github.com/pingcap/tidb/pkg/executor/join/issue51998"
+	store := testkit.CreateMockStore(t)
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("use test")
+	tk.MustExec("drop table if exists t")
+	tk.MustExec("create table t(a int)")
+	require.NoError(t, failpoint.Enable(fpName, `return(true)`))
+	defer func() {
+		require.NoError(t, failpoint.Disable(fpName))
+	}()
+	err := tk.QueryToErr("select /*+ hash_join(t1) */ * from t t1 join t t2 on t1.a=t2.a")
+	require.EqualError(t, err, "issue51998 build return error")
+}
+
 func TestIssue29498(t *testing.T) {
 	store := testkit.CreateMockStore(t)
 	tk := testkit.NewTestKit(t, store)