// Copyright 2023 PingCAP, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package executor

import (
	"context"
	"strconv"
	"strings"
	"sync/atomic"

	"github.com/pingcap/errors"
	"github.com/pingcap/tidb/ddl/schematracker"
	"github.com/pingcap/tidb/infoschema"
	"github.com/pingcap/tidb/kv"
	"github.com/pingcap/tidb/parser/ast"
	"github.com/pingcap/tidb/parser/model"
	"github.com/pingcap/tidb/parser/mysql"
	plannercore "github.com/pingcap/tidb/planner/core"
	"github.com/pingcap/tidb/sessionctx/stmtctx"
	"github.com/pingcap/tidb/sessionctx/variable"
	"github.com/pingcap/tidb/sessiontxn"
	"github.com/pingcap/tidb/types"
	"github.com/pingcap/tidb/util/chunk"
	"github.com/pingcap/tidb/util/collate"
	"github.com/pingcap/tidb/util/logutil"
	"github.com/pingcap/tidb/util/sqlexec"
	"go.uber.org/zap"
)

// ProcedureExec create \call \drop procedure exec plan.
type ProcedureExec struct {
	baseExecutor
	done                 bool
	isExit               bool
	hasWarning           bool
	exitBlockNum         int
	totBlockNum          int
	handlerBlock         int
	is                   infoschema.InfoSchema
	Statement            ast.StmtNode
	oldSqlMod            string
	procedureInfo        *plannercore.ProcedurebodyInfo
	Plan                 *plannercore.ProcedurePlan
	warnings             []error
	outParam             map[string]string
	cursList             map[string]*procedurceCurInfo
	cursHistring         []*procedurceCurInfo
	procedureErrorCode   []map[uint64]*plannercore.ProcedureErrcodeCon
	procedureErrorState  []map[string]*plannercore.ProcedureErrcodeState
	procedureErrorStatus [][]*plannercore.ProcedureErrStatusCon
}

// cacheFetchRes store the results of partial fetch.
type cacheFetchRes struct {
	res    []chunk.Row
	id     int
	maxLen int
}

// procedurceCurInfo store procedure cursor info.
type procedurceCurInfo struct {
	curName    string
	selectStmt ast.StmtNode      // select ast
	rs         sqlexec.RecordSet // resultset
	req        *chunk.Chunk
	ctx        context.Context
	cache      *cacheFetchRes // partail result
}

type errorOperation struct {
	operate plannercore.ProcedureBaseBody
	handle  int
}

// blockVariables procedure block variables.
type blockVariables struct {
	nameList []string
	cursList []string
}

// buildCreateProcedure Create stored procedure create executor.
func (b *executorBuilder) buildCreateProcedure(v *plannercore.CreateProcedure) Executor {
	base := newBaseExecutor(b.ctx, v.Schema(), v.ID())
	base.initCap = chunk.ZeroCapacity
	e := &ProcedureExec{
		baseExecutor: base,
		Statement:    v.ProcedureInfo,
		is:           b.is,
		done:         false,
		outParam:     make(map[string]string, 10),
		cursList:     make(map[string]*procedurceCurInfo, 10),
		cursHistring: make([]*procedurceCurInfo, 0, 10),
	}
	return e
}

// buildCreateProcedure create stored procedure drop executor.
func (b *executorBuilder) buildDropProcedure(v *plannercore.DropProcedure) Executor {
	base := newBaseExecutor(b.ctx, v.Schema(), v.ID())
	base.initCap = chunk.ZeroCapacity
	e := &ProcedureExec{
		baseExecutor: base,
		Statement:    v.Procedure,
		is:           b.is,
		done:         false,
	}
	return e
}

func (b *executorBuilder) buildAlterProcedure(v *plannercore.AlterProcedure) Executor {
	base := newBaseExecutor(b.ctx, v.Schema(), v.ID())
	base.initCap = chunk.ZeroCapacity
	e := &ProcedureExec{
		baseExecutor: base,
		Statement:    v.Procedure,
		is:           b.is,
		done:         false,
	}
	return e
}

// autoNewTxn Check if it is an implicit transaction.
func (e *ProcedureExec) autoNewTxn() bool {
	switch e.Statement.(type) {
	case *ast.ProcedureInfo:
		return true
	case *ast.DropProcedureStmt:
		return true
	case *ast.AlterProcedureStmt:
		return true
	}
	return false
}

// Close ProcedureExec.
func (e *ProcedureExec) Close() error {
	e.outParam = nil
	e.cursList = nil
	e.cursHistring = nil
	e.procedureErrorCode = nil
	e.procedureErrorState = nil
	e.procedureErrorStatus = nil
	//e.procedureErrorStatusHistory = nil
	err := e.base().Close()
	if err != nil {
		return err
	}
	return nil
}

func (e *ProcedureExec) Next(ctx context.Context, req *chunk.Chunk) (err error) {
	if e.done {
		return nil
	}
	// implicit commit
	if e.autoNewTxn() {
		if err = sessiontxn.NewTxnInStmt(ctx, e.ctx); err != nil {
			return err
		}
		defer func() { e.ctx.GetSessionVars().SetInTxn(false) }()
	}
	switch x := e.Statement.(type) {
	case *ast.ProcedureInfo:
		err = e.createProcedure(ctx, x)
	case *ast.DropProcedureStmt:
		err = e.dropProcedure(ctx, x)
	case *ast.CallStmt:
		err = e.callProcedure(ctx, x)
	case *ast.AlterProcedureStmt:
		err = e.alterProcedure(ctx, x)
	}

	e.done = true
	return err
}

// procedureExistsInternal Query whether the stored procedure exists.
func procedureExistsInternal(ctx context.Context, sqlExecutor sqlexec.SQLExecutor, name string, db string) (bool, error) {
	sql := new(strings.Builder)
	sqlexec.MustFormatSQL(sql, `SELECT * FROM %n.%n WHERE route_schema=%? AND name=%? AND type= 'PROCEDURE' FOR UPDATE;`, mysql.SystemDB, mysql.Routines, db, name)
	recordSet, err := sqlExecutor.ExecuteInternal(ctx, sql.String())
	if err != nil {
		return false, err
	}
	req := recordSet.NewChunk(nil)
	err = recordSet.Next(ctx, req)
	var rows int = 0
	if err == nil {
		rows = req.NumRows()
	}
	errClose := recordSet.Close()
	if errClose != nil {
		return false, errClose
	}
	return rows > 0, err
}

// getProcedureinfo read stored procedure content.
func getProcedureinfo(ctx context.Context, sqlExecutor sqlexec.SQLExecutor, name string, db string) (*plannercore.ProcedurebodyInfo, error) {
	sql := new(strings.Builder)
	//names = []string{"Procedure", "sql_mode", "Create Procedure", "character_set_client", "collation_connection", "Database Collation"}
	sqlexec.MustFormatSQL(sql, "select name, sql_mode ,definition_utf8,parameter_str,character_set_client, connection_collation, comment,")
	sqlexec.MustFormatSQL(sql, "schema_collation from %n.%n where route_schema = %?  and name = %? and type = 'PROCEDURE' ", mysql.SystemDB, mysql.Routines, db, name)
	recordSet, err := sqlExecutor.ExecuteInternal(ctx, sql.String())
	if err != nil {
		return nil, err
	}
	if recordSet != nil {
		defer recordSet.Close()
	}

	rows, err := sqlexec.DrainRecordSet(ctx, recordSet, 3)
	if err != nil {
		return nil, err
	}
	if len(rows) == 0 {
		return nil, ErrSpDoesNotExist.GenWithStackByArgs("PROCEDURE", name)
	}
	if len(rows) != 1 {
		return nil, errors.New("Multiple stored procedures found in table " + mysql.Routines)
	}
	procedurebodyInfo := &plannercore.ProcedurebodyInfo{}
	procedurebodyInfo.Name = rows[0].GetString(0)
	procedurebodyInfo.Procedurebody = " CREATE PROCEDURE `" + rows[0].GetString(0) + "`(" + rows[0].GetString(3) + ")"
	if len(rows[0].GetString(6)) != 0 {
		procedurebodyInfo.Procedurebody = procedurebodyInfo.Procedurebody + "\nCOMMENT '" + rows[0].GetString(6) + "'\n"
	}
	procedurebodyInfo.Procedurebody = procedurebodyInfo.Procedurebody + rows[0].GetString(2)
	procedurebodyInfo.SqlMode = rows[0].GetSet(1).String()
	procedurebodyInfo.CharacterSetClient = rows[0].GetString(4)
	procedurebodyInfo.CollationConnection = rows[0].GetString(5)
	procedurebodyInfo.ShemaCollation = rows[0].GetString(7)
	return procedurebodyInfo, nil
}

// getRowsProcedure read rows from table.
func (e *ShowExec) getRowsProcedure(ctx context.Context, sqlExecutor sqlexec.SQLExecutor, showType string) error {
	sql := new(strings.Builder)
	var (
		fieldPatternsLike collate.WildcardPattern
		fieldFilter       string
	)
	if e.Extractor != nil {
		fieldFilter = e.Extractor.Field()
		fieldPatternsLike = e.Extractor.FieldPatternLike()
	}
	//names = []string{"Db", "Name", "Type", "Definer", "Modified", "Created", "Security_type", "Comment", "character_set_client", "collation_connection", "Database Collation"}
	sqlexec.MustFormatSQL(sql, "select route_schema, name, type, definer ,last_altered,created,security_type, comment,")
	sqlexec.MustFormatSQL(sql, "character_set_client, connection_collation,schema_collation from %n.%n where type = %?", mysql.SystemDB, mysql.Routines, showType)
	recordSet, err := sqlExecutor.ExecuteInternal(ctx, sql.String())
	if err != nil {
		return err
	}
	if recordSet != nil {
		defer recordSet.Close()
	}

	rows, err := sqlexec.DrainRecordSet(ctx, recordSet, 1024)
	if err != nil {
		return err
	}

	for _, row := range rows {
		if fieldFilter != "" && row.GetString(1) != fieldFilter {
			continue
		} else if fieldPatternsLike != nil && !fieldPatternsLike.DoMatch(row.GetString(1)) {
			continue
		}
		e.appendRow([]interface{}{row.GetString(0), row.GetString(1), row.GetEnum(2).String(), row.GetString(3), row.GetTime(4), row.GetTime(5), row.GetEnum(6).String(),
			row.GetString(7), row.GetString(8), row.GetString(9), row.GetString(10)})
	}
	return nil
}

// checkProcedureVars check procedure variable validity.
func (e *ProcedureExec) checkProcedureVars(ctx context.Context, s *ast.ProcedureInfo) error {
	nameMap := make(map[string]struct{}, len(s.ProcedureParam))
	// init procedure in/out param.
	for _, param := range s.ProcedureParam {
		_, ok := nameMap[param.ParamName]
		if ok {
			return ErrSpDupParam.GenWithStackByArgs(param.ParamName)
		}
		nameMap[param.ParamName] = struct{}{}
		err := plannercore.NewEmptyVars(ctx, param.ParamName, param.ParamType, e.ctx.GetSessionVars())
		if err != nil {
			return err
		}
	}
	err := e.checkProcedureVal(ctx, s.ProcedureBody)
	return err
}

// checkProcedureVal check cursor\variable\error validity.
func (e *ProcedureExec) checkProcedureVal(ctx context.Context, stmt ast.StmtNode) (err error) {
	switch x := stmt.(type) {
	case *ast.ProcedureBlock:
		vars := x.ProcedureVars
		names := make([]string, 0, len(vars))
		curs := make([]string, 0, len(vars))
		errorCodes := make([]uint64, 0, 10)
		errorState := make([]string, 0, 10)
		errorStatus := make([]int, 0, 3)
		for _, varInfo := range vars {
			switch x := varInfo.(type) {
			case *ast.ProcedureCursor:
				if len(errorCodes) >= 1 || len(errorStatus) >= 1 {
					return errors.Trace(ErrSpCursorAfterHandler)
				}
				if ifInSlice(curs, x.CurName) {
					return ErrSpDupCurs.GenWithStackByArgs(x.CurName)
				}
				e.addEmptyCurs(x.CurName)
				curs = append(curs, x.CurName)

			case *ast.ProcedureDecl:
				// procedure variable must at begin.
				if len(errorCodes) >= 1 || len(errorStatus) >= 1 || len(curs) >= 1 {
					return errors.Trace(ErrSpVarcondAfterCurshndlr)
				}
				// The same block name cannot be repeated.
				for _, name := range x.DeclNames {
					err := plannercore.NewEmptyVars(ctx, name, x.DeclType, e.ctx.GetSessionVars())
					if err != nil {
						return err
					}
					if ifInSlice(names, name) {
						return ErrSpDupVar.GenWithStackByArgs(name)
					}
					names = append(names, name)
				}
			case *ast.ProcedureErrorControl:
				// error control must at end.
				err := e.checkProcedureVal(ctx, x.Operate)
				if err != nil {
					return err
				}
				// The same block name cannot be repeated.
				for _, node := range x.ErrorCon {
					switch v := node.(type) {
					case *ast.ProcedureErrorVal:
						for _, errorCode := range errorCodes {
							if errorCode == v.ErrorNum {
								return errors.Trace(ErrSpDupHandler)
							}
						}
						errorCodes = append(errorCodes, v.ErrorNum)
					case *ast.ProcedureErrorCon:
						for _, errorCode := range errorStatus {
							if errorCode == v.ErrorCon {
								return errors.Trace(ErrSpDupHandler)
							}
						}
						errorStatus = append(errorStatus, v.ErrorCon)
					case *ast.ProcedureErrorState:
						if !plannercore.IsSqlstateValid(v.CodeStatus) || plannercore.IsSqlStateCompletion(v.CodeStatus) {
							return ErrSpBadSQLstate.GenWithStackByArgs(v.CodeStatus)
						}
						for _, errorState := range errorState {
							if errorState == v.CodeStatus {
								return errors.Trace(ErrSpDupHandler)
							}
						}
						errorState = append(errorState, v.CodeStatus)
					default:
						return errors.Errorf("Procedure error control unsupport node %T", node)
					}
				}

			default:
				return errors.Errorf("Call procedure unsupport node %T", varInfo)
			}
		}
		stmts := x.ProcedureProcStmts
		for _, stmt := range stmts {
			err := e.checkProcedureVal(ctx, stmt)
			if err != nil {
				return err
			}
		}
		defer func() {
			for id := len(names) - 1; id >= 0; id-- {
				if err == nil {
					err = e.ctx.GetSessionVars().DeleteProcedureVariable(names[id], true)
				}
			}
			for id := len(curs) - 1; id >= 0; id-- {
				if err == nil {
					e.delCurs(curs[id])
				}
			}
		}()
	case *ast.SetStmt:
		// check variable if exists.
		for _, vari := range x.Variables {
			if !vari.IsSystem {
				continue
			}
			if e.ctx.GetSessionVars().CheckProcedureVariable(vari.Name) {
				continue
			}
			sysVar := variable.GetSysVar(vari.Name)
			if sysVar == nil {
				if variable.IsRemovedSysVar(vari.Name) {
					continue // removed vars permit parse-but-ignore
				}
				return variable.ErrUnknownSystemVar.GenWithStackByArgs(vari.Name)
			}
		}
	case *ast.ProcedureOpenCur:
		_, ok := e.cursList[x.CurName]
		if !ok {
			return ErrSpCursorMismatch.GenWithStackByArgs(x.CurName)
		}
	case *ast.ProcedureCloseCur:
		_, ok := e.cursList[x.CurName]
		if !ok {
			return ErrSpCursorMismatch.GenWithStackByArgs(x.CurName)
		}
	case *ast.ProcedureFetchInto:
		_, ok := e.cursList[x.CurName]
		if !ok {
			return ErrSpCursorMismatch.GenWithStackByArgs(x.CurName)
		}
		for _, varName := range x.Variables {
			if !e.ctx.GetSessionVars().CheckProcedureVariable(varName) {
				return ErrSpUndeclaredVar.GenWithStackByArgs(varName)
			}
		}
	}
	return nil
}

// createProcedure Save stored procedure content.
func (e *ProcedureExec) createProcedure(ctx context.Context, s *ast.ProcedureInfo) error {
	e.ctx.GetSessionVars().SetInCallProcedure()
	e.ctx.GetSessionVars().NewProcedureVariables()
	defer func() {
		e.ctx.GetSessionVars().OutCallProcedure()
		e.ctx.GetSessionVars().ClearProcedureVariable()
	}()

	procedurceName := s.ProcedureName.Name.String()
	procedurceSchema := s.ProcedureName.Schema
	dbInfo, ok := e.is.SchemaByName(procedurceSchema)
	if !ok {
		return ErrBadDB.GenWithStackByArgs(procedurceSchema)
	}

	err := e.checkProcedureVars(ctx, s)
	if err != nil {
		return err
	}

	var comment string
	for _, characteristic := range s.Characteristics {
		switch x := characteristic.(type) {
		case *ast.ProcedureComment:
			comment = x.Comment
		default:
			errors.Errorf("Unsupported procedure characteristic type %T", characteristic)
		}
	}
	parameterStr := s.ProcedureParamStr
	bodyStr := s.ProcedureBody.Text()

	sqlMod, ok := e.ctx.GetSessionVars().GetSystemVar(variable.SQLModeVar)
	if !ok {
		return errors.New("unknown system var " + variable.SQLModeVar)
	}
	chs, ok := e.ctx.GetSessionVars().GetSystemVar(variable.CharacterSetClient)
	if !ok {
		return errors.New("unknown system var " + variable.CharacterSetClient)
	}
	var userInfo string
	//get create user info.
	if e.ctx.GetSessionVars().User != nil {
		u := e.ctx.GetSessionVars().User.AuthUsername
		h := e.ctx.GetSessionVars().User.AuthHostname
		userInfo = u + "@" + h
	}

	_, sessionCollation := e.ctx.GetSessionVars().GetCharsetInfo()
	sql := new(strings.Builder)
	sqlexec.MustFormatSQL(sql, "insert into mysql.routines (route_schema, name, type, definition, definition_utf8, parameter_str,")
	sqlexec.MustFormatSQL(sql, "is_deterministic, sql_data_access, security_type, definer, sql_mode, character_set_client, connection_collation, schema_collation, created, last_altered, comment, ")
	sqlexec.MustFormatSQL(sql, " external_language) values (%?, %?, 'PROCEDURE', %?, %?, %?, 1, 'CONTAINS SQL', 'DEFINER', %?, %?, %?, %?, %?, now(), now(),  %?, 'SQL') ;", procedurceSchema.String(), procedurceName,
		bodyStr, bodyStr, parameterStr, userInfo, sqlMod, chs, sessionCollation, dbInfo.Collate, comment)
	sysSession, err := e.getSysSession()
	if err != nil {
		return err
	}
	internalCtx := kv.WithInternalSourceType(context.Background(), kv.InternalTxnProcedure)
	defer e.releaseSysSession(internalCtx, sysSession)
	sqlExecutor := sysSession.(sqlexec.SQLExecutor)
	if _, err := sqlExecutor.ExecuteInternal(internalCtx, "BEGIN PESSIMISTIC"); err != nil {
		return err
	}
	// check if procedure exists.
	exists, err := procedureExistsInternal(internalCtx, sqlExecutor, procedurceName, procedurceSchema.L)
	if err != nil {
		return err
	}
	if exists {
		err = ErrSpAlreadyExists.GenWithStackByArgs("PROCEDURE", procedurceName)
		if s.IfNotExists {
			e.ctx.GetSessionVars().StmtCtx.AppendNote(err)
			return nil
		}
		return err
	}
	// insert procedure.
	if _, err := sqlExecutor.ExecuteInternal(internalCtx, sql.String()); err != nil {
		return err
	}
	if _, err := sqlExecutor.ExecuteInternal(internalCtx, "commit"); err != nil {
		return err
	}
	return nil
}

// fetchShowCreateProcdure query stored procedure.
func (e *ShowExec) fetchShowCreateProcdure(ctx context.Context) error {
	if e.Procedure.Schema.O == "" {
		e.Procedure.Schema = model.NewCIStr(e.ctx.GetSessionVars().CurrentDB)
	}
	_, ok := e.is.SchemaByName(e.Procedure.Schema)
	if !ok {
		return infoschema.ErrDatabaseNotExists.GenWithStackByArgs(e.DBName.O)
	}
	internalCtx := kv.WithInternalSourceType(context.Background(), kv.InternalTxnProcedure)
	sysSession, err := e.getSysSession()
	if err != nil {
		return err
	}
	defer e.releaseSysSession(internalCtx, sysSession)
	sqlExecutor := sysSession.(sqlexec.SQLExecutor)
	procedureInfo, err := getProcedureinfo(internalCtx, sqlExecutor, e.Procedure.Name.String(), e.Procedure.Schema.O)
	if err != nil {
		return err
	}
	//names = []string{"Procedure", "sql_mode", "Create Procedure", "character_set_client", "collation_connection", "Database Collation"}
	e.appendRow([]interface{}{procedureInfo.Name, procedureInfo.SqlMode, procedureInfo.Procedurebody, procedureInfo.CharacterSetClient,
		procedureInfo.CollationConnection, procedureInfo.ShemaCollation})
	return nil
}

// fetchShowProcedureStatus implement SHOW PROCEDURE STATUS [LIKE 'pattern' | WHERE expr]
func (e *ShowExec) fetchShowProcedureStatus(ctx context.Context, showType string) error {
	internalCtx := kv.WithInternalSourceType(context.Background(), kv.InternalTxnProcedure)
	sysSession, err := e.getSysSession()
	if err != nil {
		return err
	}
	defer e.releaseSysSession(internalCtx, sysSession)
	sqlExecutor := sysSession.(sqlexec.SQLExecutor)
	err = e.getRowsProcedure(internalCtx, sqlExecutor, showType)
	if err != nil {
		return err
	}
	//names = []string{"Db", "Name", "Type", "Definer", "Modified", "Created", "Security_type", "Comment",
	//"character_set_client", "collation_connection", "Database Collation"}
	return nil
}

// dropProcedure delete stored procedure.
func (e *ProcedureExec) dropProcedure(ctx context.Context, s *ast.DropProcedureStmt) error {
	procedurceName := s.ProcedureName.Name.String()
	procedurceSchema := s.ProcedureName.Schema
	_, ok := e.is.SchemaByName(procedurceSchema)
	if !ok {
		return ErrBadDB.GenWithStackByArgs(procedurceSchema)
	}
	internalCtx := kv.WithInternalSourceType(context.Background(), kv.InternalTxnProcedure)
	sysSession, err := e.getSysSession()
	if err != nil {
		return err
	}
	defer e.releaseSysSession(internalCtx, sysSession)
	sqlExecutor := sysSession.(sqlexec.SQLExecutor)

	if _, err := sqlExecutor.ExecuteInternal(internalCtx, "BEGIN PESSIMISTIC"); err != nil {
		return err
	}
	exists, err := procedureExistsInternal(internalCtx, sqlExecutor, procedurceName, procedurceSchema.String())
	if err != nil {
		return err
	}
	if !exists {
		err = ErrSpDoesNotExist.GenWithStackByArgs("PROCEDURE", procedurceName)
		if s.IfExists {
			e.ctx.GetSessionVars().StmtCtx.AppendNote(err)
			return nil
		}
		return err
	}
	sql := new(strings.Builder)
	sqlexec.MustFormatSQL(sql, "delete from %n.%n where route_schema = %?  and name = %? and type = 'PROCEDURE' ", mysql.SystemDB,
		mysql.Routines, procedurceSchema.String(), procedurceName)
	if _, err := sqlExecutor.ExecuteInternal(internalCtx, sql.String()); err != nil {
		return err
	}
	if _, err := sqlExecutor.ExecuteInternal(internalCtx, "commit"); err != nil {
		return err
	}
	return nil
}

// buildCallProcedure generate the execution plan of the call.
func (b *executorBuilder) buildCallProcedure(v *plannercore.CallStmt) Executor {
	base := newBaseExecutor(b.ctx, v.Schema(), v.ID())
	base.initCap = chunk.ZeroCapacity
	e := &ProcedureExec{
		baseExecutor:         base,
		Statement:            v.Callstmt,
		isExit:               false,
		is:                   b.is,
		done:                 false,
		oldSqlMod:            v.OldSqlMod,
		procedureInfo:        v.ProcedureInfo,
		Plan:                 v.Plan,
		outParam:             make(map[string]string, 10),
		cursList:             make(map[string]*procedurceCurInfo, 10),
		cursHistring:         make([]*procedurceCurInfo, 0, 10),
		procedureErrorCode:   make([]map[uint64]*plannercore.ProcedureErrcodeCon, 0, 10),
		procedureErrorState:  make([]map[string]*plannercore.ProcedureErrcodeState, 0, 10),
		procedureErrorStatus: make([][]*plannercore.ProcedureErrStatusCon, 0, 10),
		handlerBlock:         0,
		exitBlockNum:         0,
		totBlockNum:          0,
	}
	return e
}

// initBlockVar enter the life cycle.
func (e *ProcedureExec) initBlockVar(ctx context.Context, blockVar *blockVariables, node *plannercore.ProcedureBlock) error {
	// Initialize sp internal variables
	procedureErrorCode := make(map[uint64]*plannercore.ProcedureErrcodeCon, 10)
	procedureErrorState := make(map[string]*plannercore.ProcedureErrcodeState, 10)
	procedureErrorStatus := make([]*plannercore.ProcedureErrStatusCon, ast.PROCEDUR_END)
	for _, pvar := range node.Vars {
		names, err := e.createVariable(ctx, pvar)
		if err != nil {
			return err
		}
		blockVar.nameList = append(blockVar.nameList, names...)
	}
	// Initialize sp internal cursor
	for _, cur := range node.Curs {
		e.addCurs(cur)
		blockVar.cursList = append(blockVar.cursList, cur.CurName)
	}
	// Initialize sp error code control.
	for _, errorCode := range node.ErrorCode {
		_, ok := procedureErrorCode[errorCode.Codes]
		if ok {
			return errors.Trace(ErrSpDupHandler)
		}
		procedureErrorCode[errorCode.Codes] = errorCode
	}
	e.addErrorCode(procedureErrorCode)

	for _, errorCode := range node.ErrorCodeState {
		_, ok := procedureErrorState[errorCode.CodeState]
		if ok {
			return errors.Trace(ErrSpDupHandler)
		}
		procedureErrorState[errorCode.CodeState] = errorCode
	}
	e.addCodeState(procedureErrorState)
	// Initialize sp error status control.
	for _, errorStatus := range node.ErrorStatus {
		if errorStatus.ErrStatus >= ast.PROCEDUR_END {
			return errors.Errorf("unkown error status %v", errorStatus.ErrStatus)
		}
		if procedureErrorStatus[errorStatus.ErrStatus] != nil {
			return errors.Trace(ErrSpDupHandler)
		}
		procedureErrorStatus[errorStatus.ErrStatus] = errorStatus
		//blockVar.statusList = append(blockVar.statusList, errorStatus.ErrStatus)
	}
	e.addErrorStatus(procedureErrorStatus)
	return nil
}

// releaseBlockVar block var leave life cycle at end of block.
func (e *ProcedureExec) releaseBlockVar(blockVar *blockVariables, err error) error {
	// detele sp internal variables
	for id := len(blockVar.nameList) - 1; id >= 0; id-- {
		if err == nil {
			err = e.ctx.GetSessionVars().DeleteProcedureVariable(blockVar.nameList[id], true)
		}
	}
	// delete sp internal cursor
	for id := len(blockVar.cursList) - 1; id >= 0; id-- {
		if err == nil {
			e.delCurs(blockVar.cursList[id])
		}
	}
	// delete sp error code control.
	if err == nil {
		e.delErrorCode()
		e.delErrorStatus()
		e.delCodeState()
	}
	return err
}

// parseNode handle various types of procedure sql.
func (e *ProcedureExec) parseNode(ctx context.Context, node plannercore.ProcedureBaseBody, errorCon bool) (err error) {
	if node == nil || e.isExit {
		return nil
	}
	vars := e.ctx.GetSessionVars()
	// killed
	if atomic.LoadUint32(&vars.Killed) == 1 {
		return errors.Trace(ErrQueryInterrupted)
	}

	blockVar := &blockVariables{}
	defer func() {
		if err != nil || e.hasWarning {
			if errorCon {
				e.handlerBlock = e.totBlockNum
			}
			err = e.errorControl(ctx, err)
		}
	}()
	switch x := node.(type) {
	// block
	case *plannercore.ProcedureBlock:
		err = e.initBlockVar(ctx, blockVar, x)
		if err != nil {
			return err
		}
		defer func() {
			err = e.releaseBlockVar(blockVar, err)
		}()
		for _, stmt := range x.ProcedureProcStmts {
			err = e.parseNode(ctx, stmt, errorCon)
			if err != nil {
				return err
			}
		}

	// procedure if block
	case *plannercore.ProcedureIfBody:
		datum, err := e.getExprValue(ctx, x.Condition, types.NewFieldType(mysql.TypeTiny))
		if err != nil {
			return err
		}
		if !datum.IsNull() {
			isTrue, err := datum.ToBool(e.ctx.GetSessionVars().StmtCtx)
			if err != nil {
				return err
			}
			if isTrue != 0 {
				for _, stmt := range x.ExecutionBranch1 {
					err := e.parseNode(ctx, stmt, errorCon)
					if err != nil {
						return err
					}
				}
				return nil
			}
		}
		err = e.parseNode(ctx, x.ExecutionBranch2, errorCon)
		if err != nil {
			return err
		}

	// procedure else block
	case *plannercore.ProcedureElseBody:
		for _, stmt := range x.ExecutionBranch {
			err := e.parseNode(ctx, stmt, errorCon)
			if err != nil {
				return err
			}
		}

	// handle base sql
	case *plannercore.ProcedureSQL:
		err := e.execNode(ctx, node.(*plannercore.ProcedureSQL).Stmt)
		if err != nil {
			return err
		}
	// handle while block
	case *plannercore.ProcedureWhileBody:
		for {
			datum, err := e.getExprValue(ctx, x.Condition, types.NewFieldType(mysql.TypeTiny))
			if err != nil {
				return err
			}
			if datum.IsNull() {
				break
			}
			isTrue, err := datum.ToBool(e.ctx.GetSessionVars().StmtCtx)
			if err != nil {
				return err
			}
			if isTrue == 0 {
				break
			}
			for _, stmt := range x.Body {
				err := e.parseNode(ctx, stmt, errorCon)
				if err != nil {
					return err
				}
			}
		}

	case *plannercore.ProcedureOpenCur:
		return e.openCurs(ctx, x.CurName)

	case *plannercore.ProcedureCloseCur:
		return e.closeCurs(x.CurName)

	case *plannercore.ProcedureFetchInto:
		return e.fetchInto(ctx, x)
	//to do:add logical structure.
	default:
		return errors.Errorf("Call procedure unsupport node %T", x)
	}
	return nil
}

// createVariable Create stored procedure internal variable.
func (e *ProcedureExec) createVariable(ctx context.Context, pval *plannercore.ProcedurebodyVal) ([]string, error) {
	var datum types.Datum
	var err error
	if pval.DeclDefault != nil {
		// get default variable.
		datum, err = e.getExprValue(ctx, pval.DeclDefault, pval.DeclType)
		if err != nil {
			return nil, err
		}
	}

	for _, name := range pval.DeclNames {
		if pval.DeclDefault == nil {
			// create variable without default variable.
			err = plannercore.NewEmptyVars(ctx, name, pval.DeclType, e.ctx.GetSessionVars())
			if err != nil {
				return nil, err
			}
		} else {
			// create variable with default variable.
			err = e.ctx.GetSessionVars().NewProcedureVariable(name, datum, pval.DeclType)
			if err != nil {
				return nil, err
			}
		}
	}
	return pval.DeclNames, nil
}

// execWithResult Execute sql with results.
func (e *ProcedureExec) execWithResult(ctx context.Context, node ast.StmtNode) error {
	err := e.ctx.GetSessionExec().MultiHanldeNodeWithResult(ctx, node)
	if err != nil {
		return err
	}
	return nil
}

// execDefalutStmt Execute sql without results.
func (e *ProcedureExec) execDefalutStmt(ctx context.Context, node ast.StmtNode) error {
	err := e.ctx.GetSessionExec().MultiHanldeNode(ctx, node)
	if err != nil {
		return err
	}
	return nil
}

// execNode implement node execution.
func (e *ProcedureExec) execNode(ctx context.Context, node ast.StmtNode) error {
	ast.SetFlag(node)
	switch x := node.(type) {
	case *ast.SelectStmt:
		if x.SelectIntoOpt != nil && x.SelectIntoOpt.Tp == ast.SelectIntoOutfile {
			err := e.execDefalutStmt(ctx, node)
			if err != nil {
				return err
			}

		} else {
			err := e.execWithResult(ctx, node)
			if err != nil {
				return err
			}
		}

	case *ast.ExplainStmt, *ast.SetOprStmt:
		err := e.execWithResult(ctx, node)
		if err != nil {
			return err
		}

	default:
		err := e.execDefalutStmt(ctx, node)
		if err != nil {
			return err
		}
	}
	sc := e.ctx.GetSessionVars().StmtCtx
	if len(sc.GetWarnings()) > 0 {
		e.hasWarning = true
	} else {
		e.hasWarning = false
	}
	return nil
}

// inParam handle input parameters.
func (e *ProcedureExec) inParam(ctx context.Context, param *plannercore.ProcedureParameterVal, inName string) error {
	// read data from user variable by name.
	datum, ok := e.ctx.GetSessionVars().GetUserVarVal(inName)
	if !ok {
		datum = types.NewDatum("")
		datum.SetNull()
	}
	newdatum, err := datum.Clone().ConvertTo(e.ctx.GetSessionVars().StmtCtx, param.DeclType)
	if err != nil {
		return err
	}
	err = e.ctx.GetSessionVars().NewProcedureVariable(param.DeclName, newdatum, param.DeclType)
	if err != nil {
		return err
	}
	return nil
}

// callParam handle store procedure parameters.
func (e *ProcedureExec) callParam(ctx context.Context, s *ast.CallStmt) error {
	for i, param := range e.Plan.ProcedureParam {
		switch s.Procedure.Args[i].(type) {
		// input variable is @name
		case *ast.VariableExpr:
			name := s.Procedure.Args[i].(*ast.VariableExpr).Name
			name = strings.ToLower(name)
			if (param.ParamType == ast.MODE_IN) || (param.ParamType == ast.MODE_INOUT) {
				err := e.inParam(ctx, param, name)
				if err != nil {
					return err
				}
			}
			if param.ParamType == ast.MODE_INOUT {
				_, ok := e.outParam[param.DeclName]
				if ok {
					return ErrSpDupParam.GenWithStackByArgs(param.DeclName)
				}
				e.outParam[param.DeclName] = name
			}
			if param.ParamType == ast.MODE_OUT {
				err := plannercore.NewEmptyVars(ctx, param.DeclName, param.DeclType, e.ctx.GetSessionVars())
				if err != nil {
					return err
				}
				e.outParam[param.DeclName] = name
			}

		// to do: support pseudo-variable in BEFORE trigger
		default:
			datum, err := e.getExprValue(ctx, param.DeclInput, param.DeclType)
			if err != nil {
				return err
			}
			if param.ParamType == ast.MODE_IN {
				err = e.ctx.GetSessionVars().NewProcedureVariable(param.DeclName, datum, param.DeclType)
				if err != nil {
					return err
				}
			}
			// out variable must @name
			if (param.ParamType == ast.MODE_OUT) || (param.ParamType == ast.MODE_INOUT) {
				return ErrSpNotVarArg.GenWithStackByArgs(i+1, s.Procedure.Schema.String()+"."+s.Procedure.FnName.String())
			}
		}
	}
	return nil
}

// outParams handle out parameters.
func (e *ProcedureExec) outParams() error {
	for key, v := range e.outParam {
		varType, datum, notFind, err := e.ctx.GetSessionVars().GetProcedureVariable(key)
		if err != nil {
			return err
		}
		if notFind {
			continue
		}
		if datum.IsNull() {
			e.ctx.GetSessionVars().UnsetUserVar(v)
		} else {
			e.ctx.GetSessionVars().SetUserVarVal(v, datum)
			e.ctx.GetSessionVars().SetUserVarType(v, varType)
		}

	}
	return nil
}

func (e *ProcedureExec) alterProcedure(ctx context.Context, s *ast.AlterProcedureStmt) error {
	procedurceName := s.ProcedureName.Name.String()
	procedurceSchema := s.ProcedureName.Schema
	internalCtx := kv.WithInternalSourceType(context.Background(), kv.InternalTxnProcedure)
	sysSession, err := e.getSysSession()
	if err != nil {
		return err
	}
	defer e.releaseSysSession(internalCtx, sysSession)
	sqlExecutor := sysSession.(sqlexec.SQLExecutor)
	if _, err := sqlExecutor.ExecuteInternal(internalCtx, "BEGIN PESSIMISTIC"); err != nil {
		return err
	}
	// check if procedure exists.
	exists, err := procedureExistsInternal(internalCtx, sqlExecutor, procedurceName, procedurceSchema.L)
	if err != nil {
		return err
	}
	if !exists {
		err = ErrSpDoesNotExist.GenWithStackByArgs("PROCEDURE", procedurceName)
		return err
	}
	type alterField struct {
		expr  string
		value any
	}
	var fields []*alterField
	for _, characteristic := range s.Characteristics {
		switch x := characteristic.(type) {
		case *ast.ProcedureComment:
			fields = append(fields, &alterField{"comment = %? ", x.Comment})
		default:
			errors.Errorf("Unsupported procedure characteristic type %T", characteristic)
		}
	}
	sql := new(strings.Builder)
	sqlexec.MustFormatSQL(sql, "update %n.%n set last_altered = now(),", mysql.SystemDB, mysql.Routines)
	for i, field := range fields {
		sqlexec.MustFormatSQL(sql, field.expr, field.value)
		if i < len(fields)-1 {
			sqlexec.MustFormatSQL(sql, ",")
		}
	}

	sqlexec.MustFormatSQL(sql, " where route_schema = %?  and name = %? and type = 'PROCEDURE' ", procedurceSchema.String(), procedurceName)
	if _, err := sqlExecutor.ExecuteInternal(internalCtx, sql.String()); err != nil {
		return err
	}
	if _, err := sqlExecutor.ExecuteInternal(internalCtx, "commit"); err != nil {
		return err
	}
	return nil
}

// callProcedure implement call.
func (e *ProcedureExec) callProcedure(ctx context.Context, s *ast.CallStmt) error {
	defer func() {
		restoreErr := e.ctx.GetSessionVars().SetSystemVar(variable.SQLModeVar, e.oldSqlMod)
		if restoreErr != nil {
			logutil.BgLogger().Error("call recover sql_mode failed. err:", zap.Error(restoreErr))
		}
	}()

	mutliStateModeSave := variable.GetSysVar(variable.TiDBMultiStatementMode)
	variable.SetSysVar(variable.TiDBMultiStatementMode, "ON")
	clientCapabilitySave := e.ctx.GetSessionVars().ClientCapability
	e.ctx.GetSessionVars().ClientCapability = (e.ctx.GetSessionVars().ClientCapability | mysql.ClientMultiStatements)
	defer func() {
		e.ctx.GetSessionVars().ClientCapability = clientCapabilitySave
		variable.SetSysVar(variable.TiDBMultiStatementMode, mutliStateModeSave.Value)
		e.ctx.GetSessionVars().OutCallProcedure()
		e.ctx.GetSessionVars().ClearProcedureVariable()
	}()
	// sp use routines database as default database.
	sysvar := variable.GetSysVar(variable.LowerCaseTableNames)
	val, err := strconv.Atoi(sysvar.Value)
	if err != nil {
		return err
	}
	is := schematracker.NewSchemaTracker(val)
	key := is.InfoStore.CiStr2Key(s.Procedure.Schema)
	if key != e.ctx.GetSessionVars().CurrentDB {
		oldDB := e.ctx.GetSessionVars().CurrentDB
		defer func() {
			e.ctx.GetSessionVars().CurrentDB = oldDB
		}()
		e.ctx.GetSessionVars().CurrentDB = key
	}

	// set error handle.
	sc := e.ctx.GetSessionVars().StmtCtx
	vars := e.ctx.GetSessionVars()
	sc.TruncateAsWarning = !vars.StrictSQLMode
	sc.DividedByZeroAsWarning = !vars.StrictSQLMode
	sc.AllowInvalidDate = vars.SQLMode.HasAllowInvalidDatesMode()
	sc.IgnoreZeroInDate = !vars.SQLMode.HasNoZeroInDateMode() || !vars.SQLMode.HasNoZeroDateMode() || !vars.StrictSQLMode || sc.AllowInvalidDate
	// handle in\out variable.
	err = e.callParam(ctx, s)
	if err != nil {
		return err
	}
	switch x := e.Plan.Procedurebody.(type) {
	case *plannercore.ProcedureBlock, *plannercore.ProcedureWhileBody, *plannercore.ProcedureIfBody, *plannercore.ProcedureSQL:
		err := e.parseNode(ctx, x, true)
		if err != nil {
			return err
		}
	default:
		return errors.Errorf("Call procedure unsupport node %T", x)
	}
	// handle out variable.
	err = e.outParams()
	if err != nil {
		return err
	}
	e.applyWarnings()
	return nil
}

// getExprValue get expr value by execute select expr.
// (*PlanBuilder).rewrite expr ==> expr.Eval() cannot handle a in (select * from t1)
// To Do: Get results from logical plan.
func (e *ProcedureExec) getExprValue(ctx context.Context, node ast.ExprNode, tp *types.FieldType) (types.Datum, error) {
	if node == nil {
		datum := types.NewDatum("")
		return datum.Clone().ConvertTo(e.ctx.GetSessionVars().StmtCtx, tp)
	}
	ast.SetFlag(node)
	ctx = kv.WithInternalSourceType(ctx, kv.InternalTxnProcedure)
	stmt := &ast.SelectStmt{
		Fields: &ast.FieldList{Fields: []*ast.SelectField{{Expr: node}}},
	}

	exec := e.ctx.(sqlexec.SQLExecutor)
	//execute as select expr.
	rs, err := exec.ExecuteStmt(ctx, stmt)
	if err != nil {
		return types.NewDatum(nil), err
	}
	defer func() {
		if rs == nil {
			return
		}
		if closeErr := rs.Close(); closeErr != nil {
			err = closeErr
		}
	}()
	if rs == nil {
		return types.NewDatum(nil), errors.Errorf("Call expr node %T can not get value", node)
	}
	rows, err := sqlexec.DrainRecordSet(ctx, rs, 3)
	if err != nil {
		return types.NewDatum(nil), err
	}
	if len(rows) > 1 {
		return types.NewDatum(nil), ErrOperandColumns.GenWithStackByArgs(1)
	}
	// reset error ignore status,ensure that type conversion failures can be reported normally.
	e.resetStatus()
	for _, row := range rows {
		if row.Len() > 1 {
			return types.NewDatum(nil), ErrSubqueryMoreThan1Row.GenWithStackByArgs(1)
		}
		if row.Len() == 1 {
			if row.IsNull(0) {
				break
			}
			r := row.GetDatum(0, &rs.Fields()[0].Column.FieldType)
			return r.Clone().ConvertTo(e.ctx.GetSessionVars().StmtCtx, tp)
		}
	}
	return types.NewDatum(nil), nil
}

// resetStatus reset error control by sql mode.
func (e *ProcedureExec) resetStatus() {
	sc := e.ctx.GetSessionVars().StmtCtx
	vars := e.ctx.GetSessionVars()
	sc.BadNullAsWarning = !vars.StrictSQLMode
	sc.TruncateAsWarning = !vars.StrictSQLMode
	sc.DividedByZeroAsWarning = !vars.StrictSQLMode
	sc.AllowInvalidDate = vars.SQLMode.HasAllowInvalidDatesMode()
	sc.IgnoreTruncate = false
	sc.IgnoreZeroInDate = !vars.SQLMode.HasNoZeroInDateMode() || !vars.SQLMode.HasNoZeroDateMode() || !vars.StrictSQLMode || sc.AllowInvalidDate
}

func ifInSlice(items []string, item string) bool {
	for _, eachititem := range items {
		if eachititem == item {
			return true
		}
	}
	return false
}

// addEmptyCurs use for check cursor validity.
func (e *ProcedureExec) addEmptyCurs(name string) {
	curInfo, ok := e.cursList[name]
	if ok {
		e.cursHistring = append(e.cursHistring, curInfo)
	}
	e.cursList[name] = &procedurceCurInfo{curName: name}
}

// addCurs Really create cursor.
func (e *ProcedureExec) addCurs(cur *plannercore.ProcedureCur) {
	block := &procedurceCurInfo{
		curName:    cur.CurName,
		selectStmt: cur.Curstmt,
	}
	curInfo, ok := e.cursList[cur.CurName]
	if ok {
		// cursor coverage.
		e.cursHistring = append(e.cursHistring, curInfo)
	}
	e.cursList[cur.CurName] = block
}

// delCurs out of cursor scope，delete cursor by name.
func (e *ProcedureExec) delCurs(name string) {
	curInfo, ok := e.cursList[name]
	if ok {
		if curInfo.rs != nil {
			_ = curInfo.rs.Close()
		}
		curInfo.ctx = nil
		curInfo.req = nil
		curInfo.cache = nil
	}
	// check if exists covered cursor.
	if len(e.cursHistring) > 0 && e.cursHistring[len(e.cursHistring)-1].curName == name {
		e.cursList[name] = e.cursHistring[len(e.cursHistring)-1]
		e.cursHistring = e.cursHistring[:len(e.cursHistring)-1]
	} else {
		delete(e.cursList, name)
	}
}

// openCurs implement open cursor.
func (e *ProcedureExec) openCurs(ctx context.Context, name string) error {
	curInfo, ok := e.cursList[name]
	if !ok {
		return ErrSpCursorMismatch.GenWithStackByArgs(name)
	}
	if curInfo.rs != nil {
		return ErrSpCursorAlreadyOpen.GenWithStackByArgs(name)
	}
	ast.SetFlag(curInfo.selectStmt)
	curInfo.ctx = kv.WithInternalSourceType(ctx, kv.InternalTxnProcedure)
	exec := e.ctx.(sqlexec.SQLExecutor)
	rs, err := exec.ExecuteStmt(curInfo.ctx, curInfo.selectStmt)
	if err != nil {
		return err
	}
	if rs == nil {
		errors.Errorf("Open cursor failed without resultset, SQL:%s ", curInfo.selectStmt.Text())
	}
	// todo:Waiting for the user-defined variable bug to be fixed can be changed to streaming.
	// https://github.com/pingcap/tidb/issues/41891
	rows, err := sqlexec.DrainRecordSet(ctx, rs, e.base().maxChunkSize)
	if err != nil {
		return err
	}
	curInfo.cache = &cacheFetchRes{res: rows, id: 0, maxLen: len(rows)}
	sc := e.ctx.GetSessionVars().StmtCtx
	if len(sc.GetWarnings()) > 0 {
		e.hasWarning = true
	} else {
		e.hasWarning = false
	}
	req := rs.NewChunk(nil)
	req = chunk.Renew(req, 32)
	curInfo.req = req
	curInfo.rs = rs
	return nil
}

// closeCurs implement close cursor.
func (e *ProcedureExec) closeCurs(name string) error {
	curInfo, ok := e.cursList[name]
	if !ok {
		return ErrSpCursorMismatch.GenWithStackByArgs(name)
	}
	curInfo.ctx = nil
	curInfo.req = nil
	if curInfo.rs == nil {
		return errors.Trace(ErrSpCursorNotOpen)
	}
	err := curInfo.rs.Close()
	if err != nil {
		return err
	}
	curInfo.rs = nil
	curInfo.cache = nil
	return nil
}

// fetchInto implement fetch into .
func (e *ProcedureExec) fetchInto(_ context.Context, command *plannercore.ProcedureFetchInto) error {
	curInfo, ok := e.cursList[command.CurName]
	if !ok {
		return ErrSpCursorMismatch.GenWithStackByArgs(command.CurName)
	}
	if curInfo.rs == nil {
		return errors.Trace(ErrSpCursorNotOpen)
	}
	// expected to be streamed query, now query is done on open.
	if curInfo.needUpdateCache() {
		err := curInfo.rs.Next(curInfo.ctx, curInfo.req)
		if err != nil {
			return err
		}
		if curInfo.req.NumRows() == 0 {
			return errors.Trace(ErrSpFetchNoData)
		}
		iter := chunk.NewIterator4Chunk(curInfo.req)
		rows := make([]chunk.Row, 0, curInfo.req.NumRows())
		for r := iter.Begin(); r != iter.End(); r = iter.Next() {
			if r.Len() != len(command.Variables) {
				return errors.Trace(ErrSpWrongNoOfFetchArgs)
			}
			rows = append(rows, r)
		}
		curInfo.req = chunk.Renew(curInfo.req, 32)
		curInfo.cache = &cacheFetchRes{res: rows, id: 0, maxLen: len(rows)}
	}
	vars := e.ctx.GetSessionVars()
	defer func() {
		curInfo.nextRows()
	}()
	// set data into procedure variable.
	if curInfo.getRows().Len() != len(command.Variables) {
		return errors.Trace(ErrSpWrongNoOfFetchArgs)
	}
	for i := 0; i < curInfo.getRows().Len(); i++ {
		datum := curInfo.getRows().GetDatum(i, &curInfo.rs.Fields()[i].Column.FieldType)
		err := vars.UpdateProcedureVariable(command.Variables[i], datum)
		if err != nil {
			return err
		}
	}
	return nil
}

// addErrorCode implement HANDLER error code
func (e *ProcedureExec) addErrorCode(code map[uint64]*plannercore.ProcedureErrcodeCon) {
	e.totBlockNum++
	e.procedureErrorCode = append(e.procedureErrorCode, code)
}

// delErrorCode clear HANDLER error
func (e *ProcedureExec) delErrorCode() {
	e.procedureErrorCode[len(e.procedureErrorCode)-1] = nil
	e.procedureErrorCode = e.procedureErrorCode[:len(e.procedureErrorCode)-1]
	// check if exit finished.
	e.totBlockNum--
	if e.isExit {
		if e.exitBlockNum > e.totBlockNum {
			e.isExit = false
		}
	}
}

// addCodeState implement HANDLER error status
func (e *ProcedureExec) addCodeState(code map[string]*plannercore.ProcedureErrcodeState) {
	e.procedureErrorState = append(e.procedureErrorState, code)
}

// delCodeState clear HANDLER state
func (e *ProcedureExec) delCodeState() {
	e.procedureErrorState[len(e.procedureErrorState)-1] = nil
	e.procedureErrorState = e.procedureErrorState[:len(e.procedureErrorState)-1]
}

// addErrorStatus implement HANDLER status code
func (e *ProcedureExec) addErrorStatus(codeStatus []*plannercore.ProcedureErrStatusCon) {
	e.procedureErrorStatus = append(e.procedureErrorStatus, codeStatus)
}

// delErrorStatus clear HANDLER error status
func (e *ProcedureExec) delErrorStatus() {
	e.procedureErrorStatus[len(e.procedureErrorStatus)-1] = nil
	e.procedureErrorStatus = e.procedureErrorStatus[:len(e.procedureErrorStatus)-1]
}

func (e *ProcedureExec) findErrorOper(code int) (*errorOperation, bool, error) {
	if len(e.procedureErrorCode) != len(e.procedureErrorStatus) {
		return nil, false, errors.New("Procedure error list and error status list inconsistent")
	}
	num := e.handlerBlock
	var errStatus string
	errStatus, ok := mysql.MySQLState[uint16(code)]
	if !ok {
		errStatus = mysql.DefaultMySQLState
	}
	for i := num - 1; i >= 0; i-- {
		errOper, ok := e.procedureErrorCode[i][uint64(code)]
		if ok {
			oper := &errorOperation{
				operate: errOper.Operate,
				handle:  errOper.Handle,
			}
			e.exitBlockNum = i + 1
			return oper, ok, nil
		}
		stateOper, ok := e.procedureErrorState[i][errStatus]
		if ok {
			oper := &errorOperation{
				operate: stateOper.Operate,
				handle:  stateOper.Handle,
			}
			e.exitBlockNum = i + 1
			return oper, ok, nil
		}
		// SQLWARNING
		if errStatus[:2] == "01" || e.hasWarning {
			if e.procedureErrorStatus[i][ast.PROCEDUR_SQLWARNING] != nil {
				operater := e.procedureErrorStatus[i][ast.PROCEDUR_SQLWARNING]
				oper := &errorOperation{
					operate: operater.Operate,
					handle:  operater.Handle,
				}
				e.exitBlockNum = i + 1
				return oper, true, nil
			}
		} else if errStatus[:2] == "02" {
			// NOT_FOUND ERROR
			if e.procedureErrorStatus[i][ast.PROCEDUR_NOT_FOUND] != nil {
				operater := e.procedureErrorStatus[i][ast.PROCEDUR_NOT_FOUND]
				oper := &errorOperation{
					operate: operater.Operate,
					handle:  operater.Handle,
				}
				e.exitBlockNum = i + 1
				return oper, true, nil
			}
		} else if e.procedureErrorStatus[i][ast.PROCEDUR_SQLEXCEPTION] != nil {
			// other SQLSTATEs
			operater := e.procedureErrorStatus[i][ast.PROCEDUR_SQLEXCEPTION]
			oper := &errorOperation{
				operate: operater.Operate,
				handle:  operater.Handle,
			}
			e.exitBlockNum = i + 1
			return oper, true, nil
		}
	}
	return nil, false, nil
}

// errorControl handling procedure errors.
func (e *ProcedureExec) errorControl(ctx context.Context, err error) error {
	if err == nil && !e.hasWarning {
		return nil
	}
	if err != nil {
		e.hasWarning = false
		_, err := e.handleSingleError(ctx, err)
		return err
	}
	warnings := e.ctx.GetSessionVars().StmtCtx.GetWarnings()
	warnLen := len(warnings)
	for i := warnLen - 1; i >= 0; i-- {
		ok, err := e.handleSingleError(ctx, warnings[i].Err)
		if ok {
			return err
		}
	}
	return nil
}

func (e *ProcedureExec) handleSingleError(ctx context.Context, err error) (bool, error) {
	err = errors.Cause(err)
	newErr, ok := err.(*errors.Error)
	if !ok {
		return false, err
	}
	oper, ok, finderr := e.findErrorOper(int(newErr.Code()))
	if finderr != nil {
		return true, finderr
	}
	if ok {
		if e.hasWarning {
			sc := e.ctx.GetSessionVars().StmtCtx
			for _, warns := range sc.GetWarnings() {
				e.warnings = append(e.warnings, warns.Err)
			}
			newerror := make([]stmtctx.SQLWarn, 0, len(sc.GetWarnings()))
			sc.SetWarnings(newerror)
		}
		return true, e.handleError(ctx, oper.handle, oper.operate, err)
	}
	return false, err
}

// handleError ignore error and do error operate.
func (e *ProcedureExec) handleError(ctx context.Context, handle int, operate plannercore.ProcedureBaseBody, err error) error {
	if err != nil {
		e.warnings = append(e.warnings, err)
	}
	// If statement error , handle error from the previous block.
	e.handlerBlock = e.exitBlockNum - 1
	exitBlockNum := e.exitBlockNum
	err = e.parseNode(ctx, operate, false)
	if err != nil {
		return err
	}
	e.exitBlockNum = exitBlockNum
	if handle == ast.PROCEDUR_EXIT && e.totBlockNum >= e.exitBlockNum {
		e.isExit = true
	}
	return nil
}

func (e *ProcedureExec) applyWarnings() {
	sc := e.ctx.GetSessionVars().StmtCtx
	for _, warning := range e.warnings {
		sc.AppendError(warning)
	}
}

// needUpdateCache check cache is finished reading.
func (curInfo *procedurceCurInfo) needUpdateCache() bool {
	if curInfo.cache == nil || (curInfo.cache.id >= curInfo.cache.maxLen) {
		return true
	}
	return false
}

// getRows get current row.
func (curInfo *procedurceCurInfo) getRows() chunk.Row {
	return curInfo.cache.res[curInfo.cache.id]
}

// nextRows switch to next line.
func (curInfo *procedurceCurInfo) nextRows() {
	curInfo.cache.id = curInfo.cache.id + 1
}