Skip to content

Commit

Permalink
This is an automated cherry-pick of pingcap#40904
Browse files Browse the repository at this point in the history
Signed-off-by: ti-chi-bot <ti-community-prow-bot@tidb.io>
  • Loading branch information
windtalker authored and ti-chi-bot committed Feb 2, 2023
1 parent ac6560f commit c346e22
Show file tree
Hide file tree
Showing 2 changed files with 41 additions and 6 deletions.
28 changes: 22 additions & 6 deletions executor/index_merge_reader.go
Original file line number Diff line number Diff line change
Expand Up @@ -590,8 +590,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 @@ -895,38 +900,49 @@ type indexMergeTableScanWorker struct {
memTracker *memory.Tracker
}

<<<<<<< HEAD
func (w *indexMergeTableScanWorker) pickAndExecTask(ctx context.Context) (task *lookupTableTask) {
=======
func (w *indexMergeTableScanWorker) pickAndExecTask(ctx context.Context, task **indexMergeTableTask) {
>>>>>>> d6302c1144 (executor: Fix tidb crash on index merge reader (#40904))
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
}
}

<<<<<<< HEAD
func (w *indexMergeTableScanWorker) handlePickAndExecTaskPanic(ctx context.Context, task *lookupTableTask) func(r interface{}) {
=======
func (w *indexMergeTableScanWorker) handlePickAndExecTaskPanic(ctx context.Context, task **indexMergeTableTask) func(r interface{}) {
>>>>>>> d6302c1144 (executor: Fix tidb crash on index merge reader (#40904))
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
19 changes: 19 additions & 0 deletions executor/index_merge_reader_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,25 @@ func TestSingleTableRead(t *testing.T) {
tk.MustQuery("select /*+ use_index_merge(t1, t1a, t1b) */ sum(a) from t1 where a < 2 or b > 4").Check(testkit.Rows("6"))
}

func TestIndexMergePickAndExecTaskPanic(t *testing.T) {
store := testkit.CreateMockStore(t)
tk := testkit.NewTestKit(t, 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 TestJoin(t *testing.T) {
store, clean := testkit.CreateMockStore(t)
defer clean()
Expand Down

0 comments on commit c346e22

Please sign in to comment.