Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 15 additions & 3 deletions pkg/executor/join/hash_join_v1.go
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,9 @@ import (
"github.com/pingcap/tidb/pkg/util/memory"
)

// IsChildCloseCalledForTest is used for test
var IsChildCloseCalledForTest = false

var (
_ exec.Executor = &HashJoinV1Exec{}
_ exec.Executor = &NestedLoopApplyExec{}
Expand Down Expand Up @@ -114,7 +117,7 @@ type HashJoinV1Exec struct {
}

// Close implements the Executor Close interface.
func (e *HashJoinV1Exec) Close() error {
func (e *HashJoinV1Exec) Close() (err error) {
if e.closeCh != nil {
close(e.closeCh)
}
Expand All @@ -138,7 +141,11 @@ func (e *HashJoinV1Exec) Close() error {
channel.Clear(e.ProbeWorkers[i].joinChkResourceCh)
}
e.ProbeSideTupleFetcher.probeChkResourceCh = nil
terror.Call(e.RowContainer.Close)
util.WithRecovery(func() { err = e.RowContainer.Close() }, func(r any) {
if r != nil {
err = errors.Errorf("%v", r)
}
})
e.HashJoinCtxV1.SessCtx.GetSessionVars().MemTracker.UnbindActionFromHardLimit(e.RowContainer.ActionSpill())
e.waiterWg.Wait()
}
Expand All @@ -159,7 +166,12 @@ func (e *HashJoinV1Exec) Close() error {
if e.stats != nil {
defer e.Ctx().GetSessionVars().StmtCtx.RuntimeStatsColl.RegisterStats(e.ID(), e.stats)
}
err := e.BaseExecutor.Close()

IsChildCloseCalledForTest = true
childErr := e.BaseExecutor.Close()
if childErr != nil {
return childErr
}
return err
}

Expand Down
3 changes: 3 additions & 0 deletions pkg/executor/join/hash_table_v1.go
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,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"
Expand Down Expand Up @@ -532,6 +533,8 @@ func (c *hashRowContainer) Len() uint64 {
}

func (c *hashRowContainer) Close() error {
failpoint.Inject("issue60923", nil)

defer c.memTracker.Detach()
c.chkBuf = nil
return c.rowContainer.Close()
Expand Down
3 changes: 2 additions & 1 deletion pkg/executor/test/issuetest/BUILD.bazel
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@ go_test(
"main_test.go",
],
flaky = True,
shard_count = 24,
shard_count = 25,
deps = [
"//pkg/autoid_service",
"//pkg/config",
Expand All @@ -21,6 +21,7 @@ go_test(
"//pkg/parser/mysql",
"//pkg/session/types",
"//pkg/testkit",
"//pkg/testkit/testfailpoint",
"//pkg/util",
"//pkg/util/dbterror/exeerrors",
"//pkg/util/memory",
Expand Down
34 changes: 34 additions & 0 deletions pkg/executor/test/issuetest/executor_issue_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@
package issuetest_test

import (
"context"
"fmt"
"math/rand"
"strings"
Expand All @@ -32,6 +33,7 @@ import (
"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"
Expand Down Expand Up @@ -772,3 +774,35 @@ 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) x) from bbb;")
}
}

func TestIssue60923(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/issue60923", "panic")
tk.MustExec("set tidb_hash_join_version=legacy")

ctx := context.Background()
join.IsChildCloseCalledForTest = false
rs, _ := tk.ExecWithContext(context.Background(), "select * from t1 join (select col0, sum(col1) from t2 group by col0) as r on t1.col0 = r.col0;")
req := rs.NewChunk(nil)
for {
err := rs.Next(ctx, req)
require.NoError(t, err)
if req.NumRows() == 0 {
break
}
}
if rs != nil {
require.Error(t, rs.Close())
}
require.True(t, join.IsChildCloseCalledForTest)
}