From 0833914b100fde4df60123b3f77be46af3dfe8bb Mon Sep 17 00:00:00 2001 From: xufei Date: Wed, 1 Feb 2023 14:33:56 +0800 Subject: [PATCH] This is an automated cherry-pick of #40904 Signed-off-by: ti-chi-bot --- executor/index_merge_reader.go | 28 ++++++++++++++++++++++------ executor/index_merge_reader_test.go | 26 ++++++++++++++++++++++++++ 2 files changed, 48 insertions(+), 6 deletions(-) diff --git a/executor/index_merge_reader.go b/executor/index_merge_reader.go index a712e02f580bf..23f824acdbd21 100644 --- a/executor/index_merge_reader.go +++ b/executor/index_merge_reader.go @@ -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() @@ -890,12 +895,16 @@ 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 } @@ -903,17 +912,22 @@ func (w *indexMergeTableScanWorker) pickAndExecTask(ctx context.Context) (task * 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 @@ -921,7 +935,9 @@ func (w *indexMergeTableScanWorker) handlePickAndExecTaskPanic(ctx context.Conte err4Panic := errors.Errorf("panic in IndexMergeReaderExecutor indexMergeTableWorker: %v", r) logutil.Logger(ctx).Error(err4Panic.Error()) - task.doneCh <- err4Panic + if *task != nil { + (*task).doneCh <- err4Panic + } } } diff --git a/executor/index_merge_reader_test.go b/executor/index_merge_reader_test.go index 99670af4807f7..a8d6a3da46500 100644 --- a/executor/index_merge_reader_test.go +++ b/executor/index_merge_reader_test.go @@ -46,8 +46,34 @@ func (s *testSuite1) TestSingleTableRead(c *C) { tk.MustQuery("select /*+ use_index_merge(t1, t1a, t1b) */ sum(a) from t1 where a < 2 or b > 4").Check(testkit.Rows("6")) } +<<<<<<< HEAD func (s *testSuite1) TestJoin(c *C) { tk := testkit.NewTestKitWithInit(c, s.store) +======= +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 := testkit.CreateMockStore(t) + tk := testkit.NewTestKit(t, store) + tk.MustExec("use test") +>>>>>>> d6302c1144 (executor: Fix tidb crash on index merge reader (#40904)) 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)")