Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

executor: Fix tidb crash on index merge reader (#40904) #40982

Closed
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
24 changes: 16 additions & 8 deletions executor/index_merge_reader.go
Original file line number Diff line number Diff line change
Expand Up @@ -585,8 +585,13 @@ func (e *IndexMergeReaderExecutor) startIndexMergeTableScanWorker(ctx context.Co
defer trace.StartRegion(ctx, "IndexMergeTableScanWorker").End()
var task *lookupTableTask
util.WithRecovery(
func() { task = worker.pickAndExecTask(ctx1) },
worker.handlePickAndExecTaskPanic(ctx1, task),
// Note we use the address of `task` as the argument of both `pickAndExecTask` and `handlePickAndExecTaskPanic`
// because `task` is expected to be assigned in `pickAndExecTask`, and this assignment should also be visible
// in `handlePickAndExecTaskPanic` since it will get `doneCh` from `task`. Golang always pass argument by value,
// so if we don't use the address of `task` as the argument, the assignment to `task` in `pickAndExecTask` is
// not visible in `handlePickAndExecTaskPanic`
func() { worker.pickAndExecTask(ctx1, &task) },
worker.handlePickAndExecTaskPanic(ctx1, &task),
)
cancel()
e.tblWorkerWg.Done()
Expand Down Expand Up @@ -890,38 +895,41 @@ type indexMergeTableScanWorker struct {
memTracker *memory.Tracker
}

func (w *indexMergeTableScanWorker) pickAndExecTask(ctx context.Context) (task *lookupTableTask) {
func (w *indexMergeTableScanWorker) pickAndExecTask(ctx context.Context, task **lookupTableTask) {
var ok bool
for {
waitStart := time.Now()
select {
case task, ok = <-w.workCh:
case *task, ok = <-w.workCh:
if !ok {
return
}
case <-w.finished:
return
}
execStart := time.Now()
err := w.executeTask(ctx, task)
err := w.executeTask(ctx, *task)
if w.stats != nil {
atomic.AddInt64(&w.stats.WaitTime, int64(execStart.Sub(waitStart)))
atomic.AddInt64(&w.stats.FetchRow, int64(time.Since(execStart)))
atomic.AddInt64(&w.stats.TableTaskNum, 1)
}
task.doneCh <- err
failpoint.Inject("testIndexMergePickAndExecTaskPanic", nil)
(*task).doneCh <- err
}
}

func (w *indexMergeTableScanWorker) handlePickAndExecTaskPanic(ctx context.Context, task *lookupTableTask) func(r interface{}) {
func (w *indexMergeTableScanWorker) handlePickAndExecTaskPanic(ctx context.Context, task **lookupTableTask) func(r interface{}) {
return func(r interface{}) {
if r == nil {
return
}

err4Panic := errors.Errorf("panic in IndexMergeReaderExecutor indexMergeTableWorker: %v", r)
logutil.Logger(ctx).Error(err4Panic.Error())
task.doneCh <- err4Panic
if *task != nil {
(*task).doneCh <- err4Panic
}
}
}

Expand Down
18 changes: 18 additions & 0 deletions executor/index_merge_reader_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,24 @@ func (s *testSuite1) TestJoin(c *C) {
tk.MustQuery("select /*+ use_index_merge(t1, t1a, t1b) */ sum(t1.a) from t1 join t2 on t1.id = t2.id where t1.a < 2 or t1.b > 5").Check(testkit.Rows("1"))
}

func TestIndexMergePickAndExecTaskPanic(t *testing.T) {
tk := testkit.NewTestKitWithInit(c, s.store)
tk.MustExec("use test")
tk.MustExec("drop table if exists t1, t2")
tk.MustExec("create table t1(id int primary key, a int, b int, c int, d int)")
tk.MustExec("create index t1a on t1(a)")
tk.MustExec("create index t1b on t1(b)")
tk.MustExec("insert into t1 values(1,1,1,1,1),(2,2,2,2,2),(3,3,3,3,3),(4,4,4,4,4),(5,5,5,5,5)")
tk.MustQuery("select /*+ use_index_merge(t1, primary, t1a) */ * from t1 where id < 2 or a > 4 order by id").Check(testkit.Rows("1 1 1 1 1",
"5 5 5 5 5"))
require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/executor/testIndexMergePickAndExecTaskPanic", "panic(\"pickAndExecTaskPanic\")"))
defer func() {
require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/executor/testIndexMergePickAndExecTaskPanic"))
}()
err := tk.QueryToErr("select /*+ use_index_merge(t1, primary, t1a) */ * from t1 where id < 2 or a > 4 order by id")
require.Contains(t, err.Error(), "pickAndExecTaskPanic")
}

func (s *testSuite1) TestIndexMergeReaderAndGeneratedColumn(c *C) {
tk := testkit.NewTestKitWithInit(c, s.store)
tk.MustExec("use test")
Expand Down