fix a bug that process's cancel not match to its context (#17348)

修复了一个没有给process的cancel方法赋值，以及cancel方法与传入的ctx不对应导致的pipeline无法被正常杀死的问题。

Approved by: @XuPeng-SH, @ouyuanning, @badboynt1

pkg/sql/compile/remoterunServer.go

@@ -383,6 +383,10 @@ func (receiver *messageReceiverOnServer) newCompile() (*Compile, error) {
 		cnInfo.hakeeper,
 		cnInfo.udfService,
 		cnInfo.aicm)
+	if proc.Cancel != nil {
+		proc.Cancel()
+	}
+	proc.Ctx = runningCtx
 	proc.Cancel = runningCancel
 	proc.Base.UnixTime = pHelper.unixTime
 	proc.Base.Id = pHelper.id

pkg/vm/process/process.go

@@ -63,12 +63,15 @@ func New(
 	hakeeper logservice.CNHAKeeperClient,
 	udfService udf.Service,
 	aicm *defines.AutoIncrCacheManager) *Process {
+
+	// process should always have a cancel context to help we close this query.
+	processCtx, processCancel := context.WithCancel(ctx)
+
 	baseProcess := &BaseProcess{
 		mp:           m,
-		Ctx:          ctx,
 		TxnClient:    txnClient,
 		FileService:  fileService,
-		IncrService:  incrservice.GetAutoIncrementService(ctx),
+		IncrService:  incrservice.GetAutoIncrementService(processCtx),
 		UnixTime:     time.Now().UnixNano(),
 		LastInsertID: new(uint64),
 		LockService:  lockService,
@@ -85,8 +88,9 @@ func New(
 		TxnOperator:    txnOperator,
 	}
 	return &Process{
-		Ctx:  ctx,
-		Base: baseProcess,
+		Ctx:    ctx,
+		Cancel: processCancel,
+		Base:   baseProcess,
 	}
 }
 

pkg/vm/process/types.go

@@ -339,8 +339,6 @@ type BaseProcess struct {
 	TxnClient           client.TxnClient
 	AnalInfos           []*AnalyzeInfo
 	SessionInfo         SessionInfo
-	Ctx                 context.Context
-	Cancel              context.CancelFunc
 	FileService         fileservice.FileService
 	LockService         lockservice.LockService
 	IncrService         incrservice.AutoIncrementService