package service

import (
	"context"
	"database/sql"
	"database/sql/driver"
	"errors"
	"fmt"
	"strconv"
	"strings"
	"time"

	"dataservices/commons/contextvar"
	comErrs "dataservices/commons/errs"
	"dataservices/commons/logger"
	"dataservices/commons/metrics"
	"dataservices/commons/sessioncache"
	"dataservices/commons/utils"
	"dataservices/dbhubsdk"
	dto "dataservices/internal/dto/dataapi"
	"dataservices/internal/entity"

	"github.com/pingcap/tidb/pkg/parser"
	"github.com/pingcap/tidb/pkg/parser/ast"
	"github.com/pingcap/tidb/pkg/parser/test_driver"
	"github.com/tidbcloud/dataservices/commons/bizerr"
	"github.com/tidbcloud/dataservices/idl/pbgen/proto/dataapi"
	mysql "github.com/tidbcloud/go-mysql-driver"
	"github.com/tidbcloud/tiinf/errs"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
)

const maxRetries = 3

var errInvalidConn = errs.NewInternalDB("invalid connection", nil)

// errors.New("invalid connection")

func CreateSession(ctx context.Context, in *dataapi.CreateSessionReq) (sessionId string, err error) {
	sessionItem, err := createSessionWithCmds(ctx, in, "", nil)
	if err != nil {
		return "", err
	}
	return sessionItem.SessionID, nil
}

func createSessionWithCmds(ctx context.Context, in *dataapi.CreateSessionReq, sessionID string, cmds []sessioncache.Cmd) (sessionItem *sessioncache.SessionItem, err error) {
	defer func() {
		ok := strconv.FormatBool(err == nil)
		metrics.SessionCreateCounter.WithLabelValues(strconv.FormatUint(in.ProjectId, 10), strconv.FormatUint(in.ClusterId, 10), ok).Inc()
	}()

	// build db connection
	connectInput := &entity.ConnectionInput{
		ProjectId:  in.ProjectId,
		ClusterId:  in.ClusterId,
		Database:   in.Database,
		ExpireTime: time.Duration(in.ExpireSec) * time.Second,
		Credential: in.Credential,
	}
	sqlDB, err := buildConnection(ctx, connectInput, cmds)
	if err != nil {
		return nil, err
	}

	// save session to local
	crtID := getCrtID(in.Credential)
	sessionItem = sessioncache.CreateSession(sessionID,
		sqlDB, in.Database,
		crtID, in.ClusterId, in.ExpireSec, cmds,
		func(ctx context.Context, database string, cmds []sessioncache.Cmd) (*sql.DB, error) {
			connectInput.Database = database
			connectInput.Force = true
			return buildConnection(ctx, connectInput, cmds)
		})

	// save session to remote
	if err := sessionItem.Save(ctx); err != nil {
		logger.Error(ctx, "createSessionWithCmds save session to redis failed", zap.String("session_id", sessionID), zap.Error(err))
	}
	return sessionItem, nil
}

func CloseSession(ctx context.Context, in *dataapi.CloseSessionReq) error {

	sessionItem, err := getLocalSession(ctx, in.SessionId, in.ClusterId, in.Credential)
	if err != nil {
		logger.LogWithError(ctx, err, "get local session failed, try to convert from request")
		if in.SessionItem == nil {
			return comErrs.ErrInvalidSession
		}
		sessionItem = convertSessionItem(in.SessionItem)
		if err = checkSessionValid(ctx, sessionItem, in.ClusterId, in.Credential); err != nil {
			return err
		}
	}

	if err := sessionItem.Delete(ctx); err != nil {
		return err
	}
	if err := sessioncache.CloseSession(in.SessionId); err != nil {
		logger.Warn(ctx, "close local session failed", zap.String("session_id", in.SessionId), zap.Error(err))
		return comErrs.ErrInvalidSession
	}
	logger.Info(ctx, "close session succeed", zap.Any("session", sessionItem))
	return nil
}

func GetSession(ctx context.Context, in *dataapi.BasicEndpointReqV2) (sessionItem *sessioncache.SessionItem, err errs.Error) {

	start := time.Now()
	// logger := contextvar.Logger(ctx)
	defer func() {
		logger.LogWithError(ctx, err, "GetSession completed", zap.Duration("cost", time.Since(start)))
		if err != nil && !utils.IsTestUser(in.Credential.AccountId) {
			metrics.SessionNotFound.Inc()
		}
	}()
	if in.SessionId == "" {
		connectInput := &entity.ConnectionInput{
			ProjectId:        in.ProjectId,
			ClusterId:        in.ClusterId,
			Database:         in.Database,
			Credential:       in.Credential,
			ConnectionSource: in.Source,
		}
		connPool, err := dbhubsdk.GetConnPool(ctx, connectInput)
		if err != nil {
			logger.LogWithError(ctx, err, "DS-Region-Dataapi-grpc-ExecSql get connection pool failed.")
			if _, ok := err.(errs.Error); !ok {
				err = errs.NewInternalServerError("get conn pool failed", err)
			}
			return nil, err.(errs.Error)
		}
		return sessioncache.CreateOnceSession(connPool, in.ClusterId), nil
	}

	if sessioncache.IsEndpointSessionID(in.SessionId) {
		connectInput := &entity.ConnectionInput{
			ProjectId:  in.ProjectId,
			ClusterId:  in.ClusterId,
			Database:   in.Database,
			Credential: in.Credential,
		}
		connPool, err := dbhubsdk.GetConnPool(ctx, connectInput)
		if err != nil {
			logger.LogWithError(ctx, err, "DS-Region-Dataapi-grpc-ExecSql get connection pool failed.")
			if _, ok := err.(errs.Error); !ok {
				err = errs.NewInternalServerError("get conn pool failed", err)
			}
			return nil, err.(errs.Error)
		}
		return sessioncache.CreateEndpointSession(in.SessionId, connPool, in.ClusterId), nil
	}

	sessionItem, err = getLocalSession(ctx, in.SessionId, in.ClusterId, in.Credential)
	if err == nil {
		return sessionItem, nil
	}
	logger.LogWithError(ctx, err, "get session by id failed, try rebuild", zap.String("session_id", in.SessionId))

	if in.SessionItem == nil {
		return nil, comErrs.ErrInvalidSession
	}
	sessionItem = convertSessionItem(in.SessionItem)
	if err := checkSessionValid(ctx, sessionItem, in.ClusterId, in.Credential); err != nil {
		return nil, err
	}

	database := in.Database
	if database == "" {
		database = sessionItem.Database
	}
	sessionItem, cerr := createSessionWithCmds(ctx, &dataapi.CreateSessionReq{
		ProjectId:  in.ProjectId,
		ClusterId:  in.ClusterId,
		Database:   database,
		ExpireSec:  sessionItem.ExpireSec,
		Credential: in.Credential,
	}, in.SessionId, sessionItem.Cmds)
	if cerr != nil {
		metrics.SessionRebuildCounter.WithLabelValues("false").Inc()
		if err, ok := cerr.(errs.Error); ok {
			return nil, err
		} else {
			return nil, errs.NewInternalServerError("create session failed", cerr)
		}
	}
	metrics.SessionRebuildCounter.WithLabelValues("true").Inc()
	return sessionItem, nil
}

func getLocalSession(ctx context.Context, sessionId string, clusterID uint64, crtInfo *dataapi.Credential) (*sessioncache.SessionItem, errs.Error) {
	if sessionId == "" {
		metrics.SessionIdEmpty.With(nil).Inc()
		return nil, errs.NewBadRequest("Session id is required", nil)
	}
	sessionItem, err := sessioncache.GetSession(sessionId)
	if err != nil {
		return nil, comErrs.ErrInvalidSession
	}
	if err := checkSessionValid(ctx, sessionItem, clusterID, crtInfo); err != nil {
		return nil, err
	}
	// refresh session expire time
	sessioncache.RefreshSession(sessionId)
	return sessionItem, nil
}

func checkSessionValid(ctx context.Context, sessionItem *sessioncache.SessionItem, clusterID uint64, crtInfo *dataapi.Credential) errs.Error {
	logger := contextvar.Logger(ctx)
	if sessionItem.ClusterID != clusterID {
		logger.Warn("session cluster not match", zap.Uint64("session_cluster", sessionItem.ClusterID), zap.Uint64("request_cluster", clusterID))
		return errs.NewBadRequest("session not found", errors.New("session cluster id not match"))
	}
	crtID := getCrtID(crtInfo)
	if crtID != sessionItem.CrtID {
		logger.Warn("session credential id not match", zap.String("session_crtid", sessionItem.CrtID), zap.String("request_crtid", crtID))
		return errs.NewBadRequest("session not found", errors.New("session credential id not match"))
	}
	if sessionItem.EndSec <= uint64(time.Now().Unix()) {
		logger.Warn("session has expired", zap.Any("session", sessionItem))
		return errs.NewBadRequest("session has expired", nil)
	}
	return nil
}

func needRetry(err error) bool {
	if errors.Is(err, dbhubsdk.ErrInvalidConn) ||
		errors.Is(err, driver.ErrBadConn) ||
		errors.Is(err, errInvalidConn) {
		return true
	}
	if errMySQL, ok := err.(*mysql.MySQLError); ok &&
		errMySQL.Number == 1045 &&
		strings.HasPrefix(errMySQL.Message, "Access denied") {
		return true
	}
	return false
}

func execOneSql(ctx context.Context, sessionItem *sessioncache.SessionItem, db *sql.Conn, sqlStr string, args []any, limits uint64) (*dbhubsdk.QueryRes, error) {

	logger := contextvar.Logger(ctx)
	res, err := dbhubsdk.Serverless.QueryByPrepare(ctx, db, sqlStr, args, limits)
	if needRetry(err) {
		return nil, errInvalidConn
	}
	if err != nil {
		if strings.Contains(err.Error(), "context deadline exceeded") ||
			strings.Contains(err.Error(), "context canceled") {
			return nil, errs.NewBadRequest("execute sql timeout", err)
		} else {
			return nil, errs.NewInternalDB("Failed to execute sql, err:"+err.Error(), nil)
		}
	}
	if res != nil && res.Code == 1045 && strings.HasPrefix(res.Message, "Access denied") {
		logger.Error("Access denied while executing dbhubsdk.Serverless.QueryByPrepare", zap.String("message", res.Message),
			zap.Uint64("cluster_id", sessionItem.ClusterID), zap.String("sql", sqlStr))
		return nil, errInvalidConn
	}

	if sessionItem.IsStandard() && res.Code == 0 {
		if res.UseDatabase != "" {
			sessioncache.ResetDBName(sessionItem.SessionID, res.UseDatabase)
		} else if res.DropDatabase != "" && res.DropDatabase == sessionItem.Database {
			sessioncache.ResetDBName(sessionItem.SessionID, "")
		}
	}
	FillSqlDatabase(res.Databases, sessionItem.Database)
	return res, nil
}

func GetDatabaseMeta(ctx context.Context, in *dataapi.GetDatabaseMetaReq) (*dbhubsdk.QueryRes, error) {
	sql := "select table_name, column_name from INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA='" + in.Database + "'"
	return ExecSimpleSql(ctx, &dataapi.BasicEndpointReqV2{
		ProjectId:  in.ProjectId,
		ClusterId:  in.ClusterId,
		Database:   "INFORMATION_SCHEMA",
		Credential: in.Credential,
		Sql:        sql,
	})
}

type PrepareStmt struct {
	Stmt ast.StmtNode
	Args []interface{}
}

func SplitSqlIntoStmts(sql string, args []interface{}) ([]PrepareStmt, errs.Error) {
	stmts, _, err := parser.New().ParseSQL(sql)
	if err != nil {
		// return nil, errs.NewBadRequest("Parsing sql failed, Please check if sql syntax is correct, "+err.Error(), nil)
		return nil, errs.NewError(bizerr.ErrInvalidUserSql, "Parsing sql failed, Please check if sql syntax is correct, "+err.Error(), nil)
	}
	result := make([]PrepareStmt, len(stmts))
	for i, stmt := range stmts {
		countParamVisitor := &CountParamVisitor{}
		stmt.Accept(countParamVisitor)
		if countParamVisitor.Count > len(args) {
			// return nil, errs.NewBadRequest(fmt.Sprintf("Parsing sql failed, sql parameters not match, need %v params, but only got %v", countParamVisitor.Count, len(args)), nil)
			return nil, errs.NewError(bizerr.ErrInvalidUserSql, fmt.Sprintf("Parsing sql failed, sql parameters not match, need %v params, but only got %v", countParamVisitor.Count, len(args)), nil)
		}
		result[i].Stmt = stmt
		if countParamVisitor.Count > 0 {
			result[i].Args = args[:countParamVisitor.Count]
			args = args[countParamVisitor.Count:]
		}
	}
	return result, nil
}

func ExecSql(ctx context.Context, in *dataapi.BasicEndpointReqV2) (res []*dataapi.ExecStmtResult, err error) {

	start := time.Now()
	defer func() {
		if errMySQL, ok := err.(*mysql.MySQLError); ok {
			code := errMySQL.Number
			message := errMySQL.Message
			if dbhubsdk.IsCausedByUser(code, message) {
				err = errs.NewError(bizerr.ErrInvalidUserSql, err.Error(), err)
			} else {
				err = errs.NewInternalDB(message, fmt.Errorf("mysql err code:%d", code))
			}
		}
		if !utils.IsTestUser(in.GetCredential().GetAccountId()) {
			ok := strconv.FormatBool(err == nil || comErrs.IsClientErr(err))
			metrics.ExecSqlRequestDuration.WithLabelValues(ok).Observe(time.Since(start).Seconds())
		}
		logger.LogWithError(ctx, err, "service.ExecSql completed", zap.Duration("cost", time.Since(start)))
	}()

	// get session
	var sessionItem *sessioncache.SessionItem
	sessionItem, err = GetSession(ctx, in)
	if err != nil {
		logger.LogWithError(ctx, err, "DS-Region-Dataapi-grpc-ExecSql get session failed.")
		return
	}
	defer func() {
		go sessionItem.RequestPost(utils.TraceContext(ctx))
	}()

	// use db if needed.
	logger.Info(ctx, "db info", zap.String("in.Database", in.Database), zap.String("sessionItem.Database", sessionItem.Database))
	var dbInit bool
	if in.Database != "" {
		if !sessionItem.IsStandard() || in.Database != sessionItem.Database {
			useSQL := fmt.Sprintf("use `%v`;", in.Database)
			in.Sql = useSQL + in.Sql
			dbInit = true
			tilog.Info(ctx, "db initial!", zap.String("sql", in.Sql))
		}
	}

	var resetSession bool
	for i := 0; i < maxRetries; i++ {
		err = sessionItem.Handle(ctx, i, func(ctx context.Context, conn *sql.Conn) (bool, error) {
			res, resetSession, err = execSQLWithSession(ctx, in, sessionItem, conn, dbInit)
			return resetSession, err
		})
		if err != nil {
			logger.Warn(ctx, "sessionItem.Handle failed", zap.Int("idx", i), zap.Error(err))
		}
		if !needRetry(err) {
			break
		}
	}
	return
}

func execSQLWithSession(ctx context.Context, in *dataapi.BasicEndpointReqV2, sessionItem *sessioncache.SessionItem, db *sql.Conn, dbInit bool) (stmtResults []*dataapi.ExecStmtResult, resetSession bool, err error) {

	args, err := buildSqlArgs(in.ArgValues)
	if err != nil {
		logger.LogWithError(ctx, err, "DS-Region-Dataapi-grpc-execSqlWithSession BuildSqlArgs failed.", zap.Any("args", in.Args))
		return
	}

	logger.Info(ctx, "real sql", zap.String("sql", in.Sql), zap.Any("args", args))
	stmts, err := SplitSqlIntoStmts(in.Sql, args)
	if err != nil {
		logger.LogWithError(ctx, err, "DS-Region-Dataapi-grpc-execSqlWithSession SplitSqlIntoStmts failed.", zap.String("sql", in.Sql))
		return
	}

	stmtResults = make([]*dataapi.ExecStmtResult, 0, len(stmts))
	var lastStmtResult *dataapi.ExecStmtResult
	for i, stmt := range stmts {
		stmtSQL := stmt.Stmt.OriginalText()
		stmtArgs := stmt.Args
		if i == len(stmts)-1 {
			stmtSQL, stmtArgs = PaginationConvert(ctx, stmt, in)
		}
		logger.Info(ctx, "real one sql", zap.String("sql", stmtSQL), zap.Any("args", stmtArgs))
		res, serr := execOneSql(ctx, sessionItem, db, stmtSQL, stmtArgs, in.LimitRows)
		if serr != nil {
			logger.Warn(ctx, "DS-Region-Dataapi-grpc-execSqlWithSession failed to execute one sql", zap.String("sql", stmtSQL), zap.Error(serr))
			if errors.Is(serr, errInvalidConn) && i > 0 {
				return nil, false, errs.NewInternalDB("Failed to execute sql, err:"+serr.Error(), nil)
			}
			return nil, false, serr
		}
		if res.Code == 0 && isChangeSessionStats(res.StmtType) {
			if sessionItem.IsStandard() {
				sessionItem.AppendCmd(sessioncache.Cmd{
					Type:    string(res.StmtType),
					Content: dto.BuildStmtSqlForHumanReadable(stmtSQL, stmtArgs),
				})
			} else {
				resetSession = true
			}
		}
		if in.ReturnAllStmtResult || i == len(stmts)-1 || res.Code != 0 {
			// return all stmt result or the last stmt result.
			stmtRes := dto.ConvertExecSqlExecStmtResult(stmtSQL, stmtArgs, res, in.LimitRows)
			// FillSqlDatabase(stmtRes.Databases, in.Database)
			stmtResults = append(stmtResults, stmtRes)
			lastStmtResult = stmtRes
			if res.Code != 0 {
				logLevel := zap.ErrorLevel
				if dbhubsdk.IsCausedByUser(res.Code, res.Message) {
					logLevel = zap.WarnLevel
				}
				logger.Log(ctx, logLevel, fmt.Sprintf("DS-Region-Dataapi-grpc-execSqlWithSession execute one sql failed with cluster error, errmsg: %s", res.Message),
					zap.String("sql", stmtSQL),
					zap.Uint16("errcode", res.Code),
					zap.String("errmsg", res.Message))

				// For system endpoints, if the first use db reports an error, skip it directly without reporting an error,
				if i == 0 && strings.HasPrefix(res.Message, "Unknown database") &&
					dbInit && in.EndpointType == entity.EndpointTypeSystemData {
					continue
				}
				return
			}
		}
	}

	// get warning message
	if lastStmtResult != nil && lastStmtResult.GetErrCode() == 0 && needWarning(in) {
		lastStmtResult.WarnCount, lastStmtResult.WarnMessages = getDBWarning(ctx, sessionItem, db)
	}
	return
}

type CountParamVisitor struct {
	Count int
}

func (c *CountParamVisitor) Enter(n ast.Node) (node ast.Node, skipChildren bool) {
	_, ok := n.(*test_driver.ParamMarkerExpr)
	if ok {
		c.Count++
	}
	return n, false
}
func (c *CountParamVisitor) Leave(n ast.Node) (node ast.Node, ok bool) {
	return n, true
}

func ExecSimpleSql(ctx context.Context, in *dataapi.BasicEndpointReqV2) (res *dbhubsdk.QueryRes, err error) {
	defaultExpire := time.Second * 600
	connOut, err := dbhubsdk.Serverless.BuildConnection(ctx, &entity.ConnectionInput{
		ProjectId:  in.ProjectId,
		ClusterId:  in.ClusterId,
		Database:   in.Database,
		ExpireTime: defaultExpire,
		Credential: in.Credential,
	})
	if err != nil {
		return nil, err
	}
	defer func() {
		if err := connOut.DB.Close(); err != nil {
			tilog.Info(ctx, "close connection failed", zap.Error(err))
		}
	}()

	err = dbhubsdk.HandleWithDBConn(ctx, connOut.DB, func(ctx context.Context, conn *sql.Conn) error {
		res, err = dbhubsdk.Serverless.QueryByPrepare(ctx, conn, in.Sql, nil, in.LimitRows)
		if err != nil {
			return errs.NewInternalServerError("Query sql failed, error: "+err.Error(), err)
		}
		return nil
	})
	return
}

func ParseOneStmt(ctx context.Context, sql string) (dbhubsdk.StmtType, error) {
	return dbhubsdk.GetStmtTypeBySql(ctx, sql)
}

func buildSqlArgs(argValues string) ([]any, error) {
	if argValues == "" {
		return nil, nil
	}
	var sqlArgValues []entity.SqlArgValue
	if err := utils.FromJsonWithNum(argValues, &sqlArgValues); err != nil {
		return nil, errs.NewBadRequest("Parse sql arguments failed, Please check if arguments is correct, "+err.Error(), err)
	}

	result := make([]interface{}, 0, len(sqlArgValues))
	for i := range sqlArgValues {
		if val, err := sqlArgValues[i].Value(); err != nil {
			return nil, errs.NewBadRequest("Parse sql arguments failed, Please check if arguments is correct, "+err.Error(), err)
		} else {
			result = append(result, val)
		}
	}
	return result, nil
}

func FillSqlDatabase(dbs []string, indb string) []string {
	for i, database := range dbs {
		if database == "" {
			dbs[i] = indb
			break
		}
	}
	return dbs
}

func getCrtID(credential *dataapi.Credential) string {
	if credential == nil {
		return ""
	}
	if credential.AccountId != "" {
		return credential.AccountId
	}
	return utils.Sha256(credential.String())
}

func PaginationConvert(ctx context.Context, stmt PrepareStmt, in *dataapi.BasicEndpointReqV2) (stmtSql string, stmtArgs []interface{}) {
	stmtSql, stmtArgs = stmt.Stmt.OriginalText(), stmt.Args
	if in.Page == 0 || in.PageSize == 0 {
		return
	}
	if stmtType := dbhubsdk.GetStmtType(ctx, stmt.Stmt); stmtType != dbhubsdk.SELECT {
		tilog.Info(ctx, "the SQL statements does not support pagination", zap.String("stmt_type", string(stmtType)))
		return
	}

	poffset := (in.Page - 1) * in.PageSize
	stmtSql = strings.TrimSpace(stmtSql)
	stmtSql = strings.TrimSuffix(stmtSql, ";")
	stmtSql = fmt.Sprintf("SELECT * FROM (%s) AS a LIMIT %d,%d", stmtSql, poffset, in.PageSize)
	return
}

func convertSessionItem(sessionItem *dataapi.SessionItem) *sessioncache.SessionItem {
	var cmds []sessioncache.Cmd
	for _, cmd := range sessionItem.Cmds {
		cmds = append(cmds, sessioncache.Cmd{
			Type:    cmd.Type,
			Content: cmd.Content,
		})
	}
	return &sessioncache.SessionItem{
		SessionID:        sessionItem.SessionId,
		EndSec:           sessionItem.EndSec,
		Database:         sessionItem.Database,
		LastActiveTime:   sessionItem.LastActiveTime,
		CrtID:            sessionItem.CrtId,
		ClusterID:        sessionItem.ClusterId,
		ExpireSec:        sessionItem.ExpireSec,
		Cmds:             cmds,
		ExecutorEndpoint: sessionItem.ExecutorEndpoint,
	}
}

func buildConnection(ctx context.Context, in *entity.ConnectionInput, cmds []sessioncache.Cmd) (*sql.DB, error) {
	logger := contextvar.Logger(ctx)
	connOut, err := dbhubsdk.Serverless.BuildConnection(ctx, in)
	if err != nil {
		return nil, err
	}

	// recover session status by re execute status command
	if len(cmds) > 0 {
		startTime := time.Now()
		defer func() {
			logger.Info("recover session status complete", zap.Duration("cost", time.Since(startTime)))
		}()
		dbhubsdk.HandleWithDBConn(ctx, connOut.DB, func(ctx context.Context, conn *sql.Conn) error {
			for _, cmd := range cmds {
				if _, err := dbhubsdk.Serverless.QueryByPrepare(ctx, conn, cmd.Content, nil, 0); err != nil {
					logger.Warn("failed to recover session status with one commaned", zap.Any("command", cmd), zap.Error(err))
				}
			}
			return nil
		})
	}

	return connOut.DB, nil
}

func needWarning(in *dataapi.BasicEndpointReqV2) bool {
	return in.Source == entity.DataAPISourceChat2QueryConsole || in.Source == entity.DataAPISourceDataAPIConsole
}

func getDBWarning(ctx context.Context, sessionItem *sessioncache.SessionItem, db *sql.Conn) (warnCount int32, warnMessage []*dataapi.WarnMessage) {
	sql := "SHOW WARNINGS;"
	res, err := execOneSql(ctx, sessionItem, db, sql, nil, 10)
	if err != nil {
		tilog.Error(ctx, "DS-Region-Dataapi-grpc-execSqlWithSession getDBWarning failed", zap.String("sql", sql), zap.Error(err))
		return
	}

	warnCount = int32(res.RowNum)
	warnMessage = make([]*dataapi.WarnMessage, 0, len(res.Rows))
	for _, row := range res.Rows {
		if len(row) != 3 || row[1] == "" || row[2] == "" {
			continue
		}
		warnMessage = append(warnMessage, &dataapi.WarnMessage{
			Code:    row[1],
			Message: row[2],
		})
	}
	return
}

func isChangeSessionStats(stmtType dbhubsdk.StmtType) bool {
	return stmtType == dbhubsdk.SET ||
		stmtType == dbhubsdk.CREATETEMPTABLE ||
		stmtType == dbhubsdk.LOCKTABLE ||
		stmtType == dbhubsdk.BEGIN ||
		stmtType == dbhubsdk.COMMIT ||
		stmtType == dbhubsdk.ROLLBACK
}