executor: replace Call with CallWithRecover in the close of hash join v1 (#61868) (#62341)

close #60926

Author: ti-chi-bot

pkg/executor/hash_table.go

@@ -23,6 +23,7 @@ import (
 	"unsafe"
 
 	"github.com/pingcap/errors"
+	"github.com/pingcap/failpoint"
 	"github.com/pingcap/tidb/pkg/sessionctx"
 	"github.com/pingcap/tidb/pkg/sessionctx/stmtctx"
 	"github.com/pingcap/tidb/pkg/types"
@@ -546,6 +547,8 @@ func (c *hashRowContainer) Len() uint64 {
 }
 
 func (c *hashRowContainer) Close() error {
+	failpoint.Inject("issue60926", nil)
+
 	defer c.memTracker.Detach()
 	c.chkBuf = nil
 	return c.rowContainer.Close()

pkg/executor/join.go

@@ -42,9 +42,14 @@ import (
 	"github.com/pingcap/tidb/pkg/util/dbterror/exeerrors"
 	"github.com/pingcap/tidb/pkg/util/disk"
 	"github.com/pingcap/tidb/pkg/util/execdetails"
+	"github.com/pingcap/tidb/pkg/util/logutil"
 	"github.com/pingcap/tidb/pkg/util/memory"
+	"go.uber.org/zap"
 )
 
+// IsChildCloseCalledForTest is used for test
+var IsChildCloseCalledForTest atomic.Bool
+
 var (
 	_ exec.Executor = &HashJoinExec{}
 	_ exec.Executor = &NestedLoopApplyExec{}
@@ -175,7 +180,14 @@ func (e *HashJoinExec) Close() error {
 			channel.Clear(e.probeWorkers[i].joinChkResourceCh)
 		}
 		e.probeSideTupleFetcher.probeChkResourceCh = nil
-		terror.Call(e.rowContainer.Close)
+		util.WithRecovery(func() {
+			err := e.rowContainer.Close()
+			if err != nil {
+				logutil.BgLogger().Error("RowContainer encounters error",
+					zap.Error(err),
+					zap.Stack("stack trace"))
+			}
+		}, nil)
 		e.hashJoinCtx.sessCtx.GetSessionVars().MemTracker.UnbindActionFromHardLimit(e.rowContainer.ActionSpill())
 		e.waiterWg.Wait()
 	}
@@ -196,8 +208,9 @@ func (e *HashJoinExec) Close() error {
 	if e.stats != nil {
 		defer e.Ctx().GetSessionVars().StmtCtx.RuntimeStatsColl.RegisterStats(e.ID(), e.stats)
 	}
-	err := e.BaseExecutor.Close()
-	return err
+
+	IsChildCloseCalledForTest.Store(true)
+	return e.BaseExecutor.Close()
 }
 
 // Open implements the Executor Open interface.

pkg/executor/test/issuetest/BUILD.bazel

@@ -8,17 +8,19 @@ go_test(
         "main_test.go",
     ],
     flaky = True,
-    shard_count = 22,
+    shard_count = 23,
     deps = [
         "//pkg/autoid_service",
         "//pkg/config",
+        "//pkg/executor",
         "//pkg/kv",
         "//pkg/meta/autoid",
         "//pkg/parser/auth",
         "//pkg/parser/charset",
         "//pkg/parser/mysql",
         "//pkg/session/types",
         "//pkg/testkit",
+        "//pkg/testkit/testfailpoint",
         "//pkg/util",
         "//pkg/util/dbterror/exeerrors",
         "//pkg/util/memory",

pkg/executor/test/issuetest/executor_issue_test.go

@@ -24,12 +24,14 @@ import (
 	"github.com/pingcap/failpoint"
 	_ "github.com/pingcap/tidb/pkg/autoid_service"
 	"github.com/pingcap/tidb/pkg/config"
+	"github.com/pingcap/tidb/pkg/executor"
 	"github.com/pingcap/tidb/pkg/kv"
 	"github.com/pingcap/tidb/pkg/parser/auth"
 	"github.com/pingcap/tidb/pkg/parser/charset"
 	"github.com/pingcap/tidb/pkg/parser/mysql"
 	sessiontypes "github.com/pingcap/tidb/pkg/session/types"
 	"github.com/pingcap/tidb/pkg/testkit"
+	"github.com/pingcap/tidb/pkg/testkit/testfailpoint"
 	"github.com/pingcap/tidb/pkg/util"
 	"github.com/pingcap/tidb/pkg/util/dbterror/exeerrors"
 	"github.com/pingcap/tidb/pkg/util/memory"
@@ -712,3 +714,21 @@ func TestIssue55881(t *testing.T) {
 			"(select max(value) from (select * from cte union all select * from cte union all select * from aaa where aaa.id > bbb.id)) from bbb;")
 	}
 }
+
+func TestIssue60926(t *testing.T) {
+	store := testkit.CreateMockStore(t)
+	tk := testkit.NewTestKit(t, store)
+
+	tk.MustExec("use test")
+	tk.MustExec("drop table if exists t1")
+	tk.MustExec("drop table if exists t2")
+	tk.MustExec("create table t1 (col0 int, col1 int);")
+	tk.MustExec("create table t2 (col0 int, col1 int);")
+	tk.MustExec("insert into t1 values (0, 10), (1, 10), (2, 10), (3, 10), (4, 10), (5, 10), (6, 10), (7, 10), (8, 10), (9, 10), (10, 10);")
+	tk.MustExec("insert into t2 values (0, 5), (0, 5), (1, 5), (2, 5), (2, 5), (3, 5), (4, 5), (5, 5), (5, 5), (6, 5), (7, 5), (8, 5), (8, 5), (9, 5), (9, 5), (10, 5);")
+
+	testfailpoint.Enable(t, "github.com/pingcap/tidb/pkg/executor/join/issue60926", "panic")
+	executor.IsChildCloseCalledForTest.Store(false)
+	tk.MustQuery("select * from t1 join (select col0, sum(col1) from t2 group by col0) as r on t1.col0 = r.col0;")
+	require.True(t, executor.IsChildCloseCalledForTest.Load())
+}