executor: fix goroutine leak for EvalSubqueryFirstRow (#48133)

executor/adapter.go

@@ -331,7 +331,7 @@ func (a *ExecStmt) PointGet(ctx context.Context) (*recordSet, error) {
 		}
 	}
 
-	if err = pointExecutor.Open(ctx); err != nil {
+	if err = exec.Open(ctx, pointExecutor); err != nil {
 		terror.Call(pointExecutor.Close)
 		return nil, err
 	}
@@ -672,7 +672,7 @@ func (a *ExecStmt) handleForeignKeyCascade(ctx context.Context, fkc *FKCascadeEx
 		if err != nil || e == nil {
 			return err
 		}
-		if err := e.Open(ctx); err != nil {
+		if err := exec.Open(ctx, e); err != nil {
 			terror.Call(e.Close)
 			return err
 		}
@@ -1144,7 +1144,7 @@ func (a *ExecStmt) openExecutor(ctx context.Context, e Executor) (err error) {
 		}
 	}()
 	start := time.Now()
-	err = e.Open(ctx)
+	err = exec.Open(ctx, e)
 	a.phaseOpenDurations[0] += time.Since(start)
 	return err
 }

executor/admin.go

@@ -212,7 +212,11 @@ func (e *RecoverIndexExec) columnsTypes() []*types.FieldType {
 
 // Open implements the Executor Open interface.
 func (e *RecoverIndexExec) Open(ctx context.Context) error {
+<<<<<<< HEAD:executor/admin.go
 	if err := e.baseExecutor.Open(ctx); err != nil {
+=======
+	if err := exec.Open(ctx, e.BaseExecutor.Base()); err != nil {
+>>>>>>> 6e8df186f51 (executor: fix goroutine leak for EvalSubqueryFirstRow (#48133)):pkg/executor/admin.go
 		return err
 	}
 

executor/brie_test.go

@@ -83,7 +83,7 @@ func TestFetchShowBRIE(t *testing.T) {
 		baseExecutor: newBaseExecutor(sctx, schema, 0),
 		Tp:           ast.ShowBackups,
 	}
-	require.NoError(t, e.Open(ctx))
+	require.NoError(t, exec.Open(ctx, e))
 
 	tp := mysql.TypeDatetime
 	lateTime := types.NewTime(types.FromGoTime(time.Now().Add(-outdatedDuration.Duration+1)), tp, 0)

executor/builder.go

@@ -4443,6 +4443,33 @@ func (builder *dataReaderBuilder) buildIndexReaderForIndexJoin(ctx context.Conte
 		return e, err
 	}
 
+<<<<<<< HEAD:executor/builder.go
+=======
+	is := v.IndexPlans[0].(*plannercore.PhysicalIndexScan)
+	if is.Index.Global {
+		tmp, ok := builder.is.TableByID(tbInfo.ID)
+		if !ok {
+			return nil, infoschema.ErrTableNotExists
+		}
+		tbl, ok := tmp.(table.PartitionedTable)
+		if !ok {
+			return nil, exeerrors.ErrBuildExecutor
+		}
+		e.partitionIDMap, err = getPartitionIdsAfterPruning(builder.ctx, tbl, &v.PartitionInfo)
+		if err != nil {
+			return nil, err
+		}
+
+		if e.ranges, err = buildRangesForIndexJoin(e.Ctx(), lookUpContents, indexRanges, keyOff2IdxOff, cwc); err != nil {
+			return nil, err
+		}
+		if err := exec.Open(ctx, e); err != nil {
+			return nil, err
+		}
+		return e, nil
+	}
+
+>>>>>>> 6e8df186f51 (executor: fix goroutine leak for EvalSubqueryFirstRow (#48133)):pkg/executor/builder.go
 	tbl, _ := builder.executorBuilder.is.TableByID(tbInfo.ID)
 	usedPartition, canPrune, contentPos, err := builder.prunePartitionForInnerExecutor(tbl, e.Schema(), &v.PartitionInfo, lookUpContents)
 	if err != nil {
@@ -4463,13 +4490,18 @@ func (builder *dataReaderBuilder) buildIndexReaderForIndexJoin(ctx context.Conte
 				return nil, err
 			}
 		}
-		if err := e.Open(ctx); err != nil {
+		if err := exec.Open(ctx, e); err != nil {
 			return nil, err
 		}
 		return e, nil
 	}
+<<<<<<< HEAD:executor/builder.go
 	ret := &TableDualExec{baseExecutor: *e.base()}
 	err = ret.Open(ctx)
+=======
+	ret := &TableDualExec{BaseExecutor: *e.Base()}
+	err = exec.Open(ctx, ret)
+>>>>>>> 6e8df186f51 (executor: fix goroutine leak for EvalSubqueryFirstRow (#48133)):pkg/executor/builder.go
 	return ret, err
 }
 
@@ -4489,6 +4521,35 @@ func (builder *dataReaderBuilder) buildIndexLookUpReaderForIndexJoin(ctx context
 		err = e.open(ctx)
 		return e, err
 	}
+<<<<<<< HEAD:executor/builder.go
+=======
+
+	is := v.IndexPlans[0].(*plannercore.PhysicalIndexScan)
+	ts := v.TablePlans[0].(*plannercore.PhysicalTableScan)
+	if is.Index.Global {
+		tmp, ok := builder.is.TableByID(ts.Table.ID)
+		if !ok {
+			return nil, infoschema.ErrTableNotExists
+		}
+		tbl, ok := tmp.(table.PartitionedTable)
+		if !ok {
+			return nil, exeerrors.ErrBuildExecutor
+		}
+		e.partitionIDMap, err = getPartitionIdsAfterPruning(builder.ctx, tbl, &v.PartitionInfo)
+		if err != nil {
+			return nil, err
+		}
+		e.ranges, err = buildRangesForIndexJoin(e.Ctx(), lookUpContents, indexRanges, keyOff2IdxOff, cwc)
+		if err != nil {
+			return nil, err
+		}
+		if err := exec.Open(ctx, e); err != nil {
+			return nil, err
+		}
+		return e, err
+	}
+
+>>>>>>> 6e8df186f51 (executor: fix goroutine leak for EvalSubqueryFirstRow (#48133)):pkg/executor/builder.go
 	tbl, _ := builder.executorBuilder.is.TableByID(tbInfo.ID)
 	usedPartition, canPrune, contentPos, err := builder.prunePartitionForInnerExecutor(tbl, e.Schema(), &v.PartitionInfo, lookUpContents)
 	if err != nil {
@@ -4511,13 +4572,18 @@ func (builder *dataReaderBuilder) buildIndexLookUpReaderForIndexJoin(ctx context
 			}
 		}
 		e.partitionTableMode = true
-		if err := e.Open(ctx); err != nil {
+		if err := exec.Open(ctx, e); err != nil {
 			return nil, err
 		}
 		return e, err
 	}
+<<<<<<< HEAD:executor/builder.go
 	ret := &TableDualExec{baseExecutor: *e.base()}
 	err = ret.Open(ctx)
+=======
+	ret := &TableDualExec{BaseExecutor: *e.Base()}
+	err = exec.Open(ctx, ret)
+>>>>>>> 6e8df186f51 (executor: fix goroutine leak for EvalSubqueryFirstRow (#48133)):pkg/executor/builder.go
 	return ret, err
 }
 

executor/coprocessor.go

@@ -54,7 +54,7 @@ func (h *CoprocessorDAGHandler) HandleRequest(ctx context.Context, req *coproces
 		return h.buildErrorResponse(err)
 	}
 
-	err = e.Open(ctx)
+	err = exec.Open(ctx, e)
 	if err != nil {
 		return h.buildErrorResponse(err)
 	}
@@ -94,7 +94,7 @@ func (h *CoprocessorDAGHandler) HandleStreamRequest(ctx context.Context, req *co
 		return stream.Send(h.buildErrorResponse(err))
 	}
 
-	err = e.Open(ctx)
+	err = exec.Open(ctx, e)
 	if err != nil {
 		return stream.Send(h.buildErrorResponse(err))
 	}

executor/cte.go

@@ -172,7 +172,7 @@ func (p *cteProducer) openProducer(ctx context.Context, cteExec *CTEExec) (err e
 	if p.seedExec == nil {
 		return errors.New("seedExec for CTEExec is nil")
 	}
-	if err = p.seedExec.Open(ctx); err != nil {
+	if err = exec.Open(ctx, p.seedExec); err != nil {
 		return err
 	}
 
@@ -186,7 +186,7 @@ func (p *cteProducer) openProducer(ctx context.Context, cteExec *CTEExec) (err e
 	p.diskTracker.AttachTo(p.ctx.GetSessionVars().StmtCtx.DiskTracker)
 
 	if p.recursiveExec != nil {
-		if err = p.recursiveExec.Open(ctx); err != nil {
+		if err = exec.Open(ctx, p.recursiveExec); err != nil {
 			return err
 		}
 		// For non-recursive CTE, the result will be put into resTbl directly.
@@ -414,7 +414,7 @@ func (p *cteProducer) computeRecursivePart(ctx context.Context) (err error) {
 			if err = p.recursiveExec.Close(); err != nil {
 				return
 			}
-			if err = p.recursiveExec.Open(ctx); err != nil {
+			if err = exec.Open(ctx, p.recursiveExec); err != nil {
 				return
 			}
 		} else {

executor/delete.go

@@ -283,7 +283,11 @@ func (e *DeleteExec) Open(ctx context.Context) error {
 	e.memTracker = memory.NewTracker(e.id, -1)
 	e.memTracker.AttachTo(e.ctx.GetSessionVars().StmtCtx.MemTracker)
 
+<<<<<<< HEAD:executor/delete.go
 	return e.children[0].Open(ctx)
+=======
+	return exec.Open(ctx, e.Children(0))
+>>>>>>> 6e8df186f51 (executor: fix goroutine leak for EvalSubqueryFirstRow (#48133)):pkg/executor/delete.go
 }
 
 // GetFKChecks implements WithForeignKeyTrigger interface.

executor/executor.go

@@ -892,7 +892,7 @@ func (e *CheckTableExec) Open(ctx context.Context) error {
 		return err
 	}
 	for _, src := range e.srcs {
-		if err := src.Open(ctx); err != nil {
+		if err := exec.Open(ctx, src); err != nil {
 			return errors.Trace(err)
 		}
 	}
@@ -1462,8 +1462,13 @@ func init() {
 		if e.err != nil {
 			return nil, e.err
 		}
+<<<<<<< HEAD:executor/executor.go
 		err := exec.Open(ctx)
 		defer terror.Call(exec.Close)
+=======
+		err := exec.Open(ctx, executor)
+		defer terror.Call(executor.Close)
+>>>>>>> 6e8df186f51 (executor: fix goroutine leak for EvalSubqueryFirstRow (#48133)):pkg/executor/executor.go
 		if err != nil {
 			return nil, err
 		}
@@ -1853,7 +1858,11 @@ func (e *UnionExec) resultPuller(ctx context.Context, workerID int) {
 			e.mu.maxOpenedChildID = childID
 		}
 		e.mu.Unlock()
+<<<<<<< HEAD:executor/executor.go
 		if err := e.children[childID].Open(ctx); err != nil {
+=======
+		if err := exec.Open(ctx, e.Children(childID)); err != nil {
+>>>>>>> 6e8df186f51 (executor: fix goroutine leak for EvalSubqueryFirstRow (#48133)):pkg/executor/executor.go
 			result.err = err
 			e.stopFetchData.Store(true)
 			e.resultPool <- result

executor/explain.go

@@ -49,7 +49,7 @@ type ExplainExec struct {
 // Open implements the Executor Open interface.
 func (e *ExplainExec) Open(ctx context.Context) error {
 	if e.analyzeExec != nil {
-		return e.analyzeExec.Open(ctx)
+		return exec.Open(ctx, e.analyzeExec)
 	}
 	return nil
 }

executor/index_lookup_hash_join.go

@@ -124,7 +124,11 @@ type indexHashJoinTask struct {
 
 // Open implements the IndexNestedLoopHashJoin Executor interface.
 func (e *IndexNestedLoopHashJoin) Open(ctx context.Context) error {
+<<<<<<< HEAD:executor/index_lookup_hash_join.go
 	err := e.children[0].Open(ctx)
+=======
+	err := exec.Open(ctx, e.Children(0))
+>>>>>>> 6e8df186f51 (executor: fix goroutine leak for EvalSubqueryFirstRow (#48133)):pkg/executor/index_lookup_hash_join.go
 	if err != nil {
 		return err
 	}

executor/index_lookup_join.go

@@ -161,7 +161,11 @@ type innerWorker struct {
 
 // Open implements the Executor interface.
 func (e *IndexLookUpJoin) Open(ctx context.Context) error {
+<<<<<<< HEAD:executor/index_lookup_join.go
 	err := e.children[0].Open(ctx)
+=======
+	err := exec.Open(ctx, e.Children(0))
+>>>>>>> 6e8df186f51 (executor: fix goroutine leak for EvalSubqueryFirstRow (#48133)):pkg/executor/index_lookup_join.go
 	if err != nil {
 		return err
 	}

executor/index_lookup_merge_join.go

@@ -157,7 +157,11 @@ type indexMergeJoinResult struct {
 
 // Open implements the Executor interface
 func (e *IndexLookUpMergeJoin) Open(ctx context.Context) error {
+<<<<<<< HEAD:executor/index_lookup_merge_join.go
 	err := e.children[0].Open(ctx)
+=======
+	err := exec.Open(ctx, e.Children(0))
+>>>>>>> 6e8df186f51 (executor: fix goroutine leak for EvalSubqueryFirstRow (#48133)):pkg/executor/index_lookup_merge_join.go
 	if err != nil {
 		return err
 	}

executor/index_merge_reader.go

@@ -510,7 +510,7 @@ func (e *IndexMergeReaderExecutor) startPartialTableWorker(ctx context.Context,
 
 					// init partialTableReader and partialTableWorker again for the next table
 					partialTableReader.table = tbl
-					if err = partialTableReader.Open(ctx); err != nil {
+					if err = exec.Open(ctx, partialTableReader); err != nil {
 						logutil.Logger(ctx).Error("open Select result failed:", zap.Error(err))
 						syncErr(ctx, e.finished, fetchCh, err)
 						break

executor/insert.go

@@ -23,6 +23,7 @@ import (
 
 	"github.com/opentracing/opentracing-go"
 	"github.com/pingcap/errors"
+<<<<<<< HEAD:executor/insert.go
 	"github.com/pingcap/tidb/errno"
 	"github.com/pingcap/tidb/expression"
 	"github.com/pingcap/tidb/kv"
@@ -38,6 +39,25 @@ import (
 	"github.com/pingcap/tidb/util/logutil"
 	"github.com/pingcap/tidb/util/memory"
 	"github.com/pingcap/tidb/util/stringutil"
+=======
+	"github.com/pingcap/tidb/pkg/errno"
+	"github.com/pingcap/tidb/pkg/executor/internal/exec"
+	"github.com/pingcap/tidb/pkg/expression"
+	"github.com/pingcap/tidb/pkg/kv"
+	"github.com/pingcap/tidb/pkg/meta/autoid"
+	"github.com/pingcap/tidb/pkg/parser/model"
+	"github.com/pingcap/tidb/pkg/parser/mysql"
+	"github.com/pingcap/tidb/pkg/parser/terror"
+	"github.com/pingcap/tidb/pkg/table"
+	"github.com/pingcap/tidb/pkg/table/tables"
+	"github.com/pingcap/tidb/pkg/tablecodec"
+	"github.com/pingcap/tidb/pkg/types"
+	"github.com/pingcap/tidb/pkg/util/chunk"
+	"github.com/pingcap/tidb/pkg/util/logutil"
+	"github.com/pingcap/tidb/pkg/util/memory"
+	"github.com/pingcap/tidb/pkg/util/stringutil"
+	"github.com/pingcap/tidb/pkg/util/tracing"
+>>>>>>> 6e8df186f51 (executor: fix goroutine leak for EvalSubqueryFirstRow (#48133)):pkg/executor/insert.go
 	"go.uber.org/zap"
 )
 
@@ -349,7 +369,7 @@ func (e *InsertExec) Open(ctx context.Context) error {
 		e.initEvalBuffer4Dup()
 	}
 	if e.SelectExec != nil {
-		return e.SelectExec.Open(ctx)
+		return exec.Open(ctx, e.SelectExec)
 	}
 	if !e.allAssignmentsAreConstant {
 		e.initEvalBuffer()

executor/issuetest/BUILD.bazel

@@ -7,7 +7,11 @@ go_test(
         "main_test.go",
     ],
     flaky = True,
+<<<<<<< HEAD:executor/issuetest/BUILD.bazel
     shard_count = 50,
+=======
+    shard_count = 16,
+>>>>>>> 6e8df186f51 (executor: fix goroutine leak for EvalSubqueryFirstRow (#48133)):pkg/executor/test/issuetest/BUILD.bazel
     deps = [
         "//config",
         "//kv",

executor/join.go

@@ -1310,7 +1310,7 @@ func (e *NestedLoopApplyExec) Close() error {
 
 // Open implements the Executor interface.
 func (e *NestedLoopApplyExec) Open(ctx context.Context) error {
-	err := e.outerExec.Open(ctx)
+	err := exec.Open(ctx, e.outerExec)
 	if err != nil {
 		return err
 	}
@@ -1411,7 +1411,7 @@ func (e *NestedLoopApplyExec) fetchSelectedOuterRow(ctx context.Context, chk *ch
 
 // fetchAllInners reads all data from the inner table and stores them in a List.
 func (e *NestedLoopApplyExec) fetchAllInners(ctx context.Context) error {
-	err := e.innerExec.Open(ctx)
+	err := exec.Open(ctx, e.innerExec)
 	defer terror.Call(e.innerExec.Close)
 	if err != nil {
 		return err

executor/parallel_apply.go

@@ -88,7 +88,7 @@ type ParallelNestedLoopApplyExec struct {
 
 // Open implements the Executor interface.
 func (e *ParallelNestedLoopApplyExec) Open(ctx context.Context) error {
-	err := e.outerExec.Open(ctx)
+	err := exec.Open(ctx, e.outerExec)
 	if err != nil {
 		return err
 	}
@@ -299,7 +299,7 @@ func (e *ParallelNestedLoopApplyExec) fetchAllInners(ctx context.Context, id int
 		}
 	}
 
-	err = e.innerExecs[id].Open(ctx)
+	err = exec.Open(ctx, e.innerExecs[id])
 	defer terror.Call(e.innerExecs[id].Close)
 	if err != nil {
 		return err