You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
331 lines
8.9 KiB
331 lines
8.9 KiB
// Copyright 2015 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,
|
|
// See the License for the specific language governing permissions and
|
|
// limitations under the License.
|
|
|
|
package executor
|
|
|
|
import (
|
|
"context"
|
|
"math"
|
|
"sort"
|
|
"time"
|
|
|
|
"github.com/pingcap/errors"
|
|
"github.com/pingcap/log"
|
|
"github.com/pingcap/parser"
|
|
"github.com/pingcap/parser/ast"
|
|
"github.com/pingcap/parser/mysql"
|
|
"github.com/pingcap/tidb/expression"
|
|
"github.com/pingcap/tidb/infoschema"
|
|
"github.com/pingcap/tidb/planner"
|
|
plannercore "github.com/pingcap/tidb/planner/core"
|
|
"github.com/pingcap/tidb/sessionctx"
|
|
"github.com/pingcap/tidb/types"
|
|
driver "github.com/pingcap/tidb/types/parser_driver"
|
|
"github.com/pingcap/tidb/util"
|
|
"github.com/pingcap/tidb/util/chunk"
|
|
"github.com/pingcap/tidb/util/hint"
|
|
"github.com/pingcap/tidb/util/sqlexec"
|
|
"go.uber.org/zap"
|
|
)
|
|
|
|
var (
|
|
_ Executor = &DeallocateExec{}
|
|
_ Executor = &ExecuteExec{}
|
|
_ Executor = &PrepareExec{}
|
|
)
|
|
|
|
type paramMarkerSorter struct {
|
|
markers []ast.ParamMarkerExpr
|
|
}
|
|
|
|
func (p *paramMarkerSorter) Len() int {
|
|
return len(p.markers)
|
|
}
|
|
|
|
func (p *paramMarkerSorter) Less(i, j int) bool {
|
|
return p.markers[i].(*driver.ParamMarkerExpr).Offset < p.markers[j].(*driver.ParamMarkerExpr).Offset
|
|
}
|
|
|
|
func (p *paramMarkerSorter) Swap(i, j int) {
|
|
p.markers[i], p.markers[j] = p.markers[j], p.markers[i]
|
|
}
|
|
|
|
type paramMarkerExtractor struct {
|
|
markers []ast.ParamMarkerExpr
|
|
}
|
|
|
|
func (e *paramMarkerExtractor) Enter(in ast.Node) (ast.Node, bool) {
|
|
return in, false
|
|
}
|
|
|
|
func (e *paramMarkerExtractor) Leave(in ast.Node) (ast.Node, bool) {
|
|
if x, ok := in.(*driver.ParamMarkerExpr); ok {
|
|
e.markers = append(e.markers, x)
|
|
}
|
|
return in, true
|
|
}
|
|
|
|
// PrepareExec represents a PREPARE executor.
|
|
type PrepareExec struct {
|
|
baseExecutor
|
|
|
|
is infoschema.InfoSchema
|
|
name string
|
|
sqlText string
|
|
|
|
ID uint32
|
|
ParamCount int
|
|
Fields []*ast.ResultField
|
|
}
|
|
|
|
// NewPrepareExec creates a new PrepareExec.
|
|
func NewPrepareExec(ctx sessionctx.Context, is infoschema.InfoSchema, sqlTxt string) *PrepareExec {
|
|
base := newBaseExecutor(ctx, nil, 0)
|
|
base.initCap = chunk.ZeroCapacity
|
|
return &PrepareExec{
|
|
baseExecutor: base,
|
|
is: is,
|
|
sqlText: sqlTxt,
|
|
}
|
|
}
|
|
|
|
// Next implements the Executor Next interface.
|
|
func (e *PrepareExec) Next(ctx context.Context, req *chunk.Chunk) error {
|
|
vars := e.ctx.GetSessionVars()
|
|
if e.ID != 0 {
|
|
// Must be the case when we retry a prepare.
|
|
// Make sure it is idempotent.
|
|
_, ok := vars.PreparedStmts[e.ID]
|
|
if ok {
|
|
return nil
|
|
}
|
|
}
|
|
charset, collation := vars.GetCharsetInfo()
|
|
var (
|
|
stmts []ast.StmtNode
|
|
err error
|
|
)
|
|
if sqlParser, ok := e.ctx.(sqlexec.SQLParser); ok {
|
|
stmts, err = sqlParser.ParseSQL(e.sqlText, charset, collation)
|
|
} else {
|
|
p := parser.New()
|
|
p.EnableWindowFunc(vars.EnableWindowFunction)
|
|
var warns []error
|
|
stmts, warns, err = p.Parse(e.sqlText, charset, collation)
|
|
for _, warn := range warns {
|
|
e.ctx.GetSessionVars().StmtCtx.AppendWarning(util.SyntaxWarn(warn))
|
|
}
|
|
}
|
|
if err != nil {
|
|
return util.SyntaxError(err)
|
|
}
|
|
if len(stmts) != 1 {
|
|
return ErrPrepareMulti
|
|
}
|
|
stmt := stmts[0]
|
|
|
|
err = ResetContextOfStmt(e.ctx, stmt)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
var extractor paramMarkerExtractor
|
|
stmt.Accept(&extractor)
|
|
|
|
// DDL Statements can not accept parameters
|
|
if _, ok := stmt.(ast.DDLNode); ok && len(extractor.markers) > 0 {
|
|
return ErrPrepareDDL
|
|
}
|
|
|
|
if _, ok := stmt.(*ast.LoadDataStmt); ok {
|
|
return ErrUnsupportedPs
|
|
}
|
|
|
|
// Prepare parameters should NOT over 2 bytes(MaxUint16)
|
|
// https://dev.mysql.com/doc/internals/en/com-stmt-prepare-response.html#packet-COM_STMT_PREPARE_OK.
|
|
if len(extractor.markers) > math.MaxUint16 {
|
|
return ErrPsManyParam
|
|
}
|
|
|
|
err = plannercore.Preprocess(e.ctx, stmt, e.is, plannercore.InPrepare)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
// The parameter markers are appended in visiting order, which may not
|
|
// be the same as the position order in the query string. We need to
|
|
// sort it by position.
|
|
sorter := ¶mMarkerSorter{markers: extractor.markers}
|
|
sort.Sort(sorter)
|
|
e.ParamCount = len(sorter.markers)
|
|
for i := 0; i < e.ParamCount; i++ {
|
|
sorter.markers[i].SetOrder(i)
|
|
}
|
|
prepared := &ast.Prepared{
|
|
Stmt: stmt,
|
|
StmtType: GetStmtLabel(stmt),
|
|
Params: sorter.markers,
|
|
SchemaVersion: e.is.SchemaMetaVersion(),
|
|
}
|
|
|
|
prepared.UseCache = plannercore.PreparedPlanCacheEnabled() && plannercore.Cacheable(stmt, e.is)
|
|
|
|
// We try to build the real statement of preparedStmt.
|
|
for i := range prepared.Params {
|
|
param := prepared.Params[i].(*driver.ParamMarkerExpr)
|
|
param.Datum.SetNull()
|
|
param.InExecute = false
|
|
}
|
|
var p plannercore.Plan
|
|
e.ctx.GetSessionVars().PlanID = 0
|
|
e.ctx.GetSessionVars().PlanColumnID = 0
|
|
destBuilder := plannercore.NewPlanBuilder(e.ctx, e.is, &hint.BlockHintProcessor{})
|
|
p, err = destBuilder.Build(ctx, stmt)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if _, ok := stmt.(*ast.SelectStmt); ok {
|
|
e.Fields = colNames2ResultFields(p.Schema(), p.OutputNames(), vars.CurrentDB)
|
|
}
|
|
if e.ID == 0 {
|
|
e.ID = vars.GetNextPreparedStmtID()
|
|
}
|
|
if e.name != "" {
|
|
vars.PreparedStmtNameToID[e.name] = e.ID
|
|
}
|
|
|
|
normalized, digest := parser.NormalizeDigest(prepared.Stmt.Text())
|
|
preparedObj := &plannercore.CachedPrepareStmt{
|
|
PreparedAst: prepared,
|
|
VisitInfos: destBuilder.GetVisitInfo(),
|
|
NormalizedSQL: normalized,
|
|
SQLDigest: digest,
|
|
}
|
|
return vars.AddPreparedStmt(e.ID, preparedObj)
|
|
}
|
|
|
|
// ExecuteExec represents an EXECUTE executor.
|
|
// It cannot be executed by itself, all it needs to do is to build
|
|
// another Executor from a prepared statement.
|
|
type ExecuteExec struct {
|
|
baseExecutor
|
|
|
|
is infoschema.InfoSchema
|
|
name string
|
|
usingVars []expression.Expression
|
|
stmtExec Executor
|
|
stmt ast.StmtNode
|
|
plan plannercore.Plan
|
|
id uint32
|
|
lowerPriority bool
|
|
outputNames []*types.FieldName
|
|
}
|
|
|
|
// Next implements the Executor Next interface.
|
|
func (e *ExecuteExec) Next(ctx context.Context, req *chunk.Chunk) error {
|
|
return nil
|
|
}
|
|
|
|
// Build builds a prepared statement into an executor.
|
|
// After Build, e.StmtExec will be used to do the real execution.
|
|
func (e *ExecuteExec) Build(b *executorBuilder) error {
|
|
ok, err := plannercore.IsPointGetWithPKOrUniqueKeyByAutoCommit(e.ctx, e.plan)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if ok {
|
|
err = e.ctx.InitTxnWithStartTS(math.MaxUint64)
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
stmtExec := b.build(e.plan)
|
|
if b.err != nil {
|
|
log.Warn("rebuild plan in EXECUTE statement failed", zap.String("labelName of PREPARE statement", e.name))
|
|
return errors.Trace(b.err)
|
|
}
|
|
e.stmtExec = stmtExec
|
|
if e.ctx.GetSessionVars().StmtCtx.Priority == mysql.NoPriority {
|
|
e.lowerPriority = needLowerPriority(e.plan)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// DeallocateExec represent a DEALLOCATE executor.
|
|
type DeallocateExec struct {
|
|
baseExecutor
|
|
|
|
Name string
|
|
}
|
|
|
|
// Next implements the Executor Next interface.
|
|
func (e *DeallocateExec) Next(ctx context.Context, req *chunk.Chunk) error {
|
|
vars := e.ctx.GetSessionVars()
|
|
id, ok := vars.PreparedStmtNameToID[e.Name]
|
|
if !ok {
|
|
return errors.Trace(plannercore.ErrStmtNotFound)
|
|
}
|
|
preparedPointer := vars.PreparedStmts[id]
|
|
preparedObj, ok := preparedPointer.(*plannercore.CachedPrepareStmt)
|
|
if !ok {
|
|
return errors.Errorf("invalid CachedPrepareStmt type")
|
|
}
|
|
prepared := preparedObj.PreparedAst
|
|
delete(vars.PreparedStmtNameToID, e.Name)
|
|
if plannercore.PreparedPlanCacheEnabled() {
|
|
e.ctx.PreparedPlanCache().Delete(plannercore.NewPSTMTPlanCacheKey(
|
|
vars, id, prepared.SchemaVersion,
|
|
))
|
|
}
|
|
vars.RemovePreparedStmt(id)
|
|
return nil
|
|
}
|
|
|
|
// CompileExecutePreparedStmt compiles a session Execute command to a stmt.Statement.
|
|
func CompileExecutePreparedStmt(ctx context.Context, sctx sessionctx.Context,
|
|
ID uint32, args []types.Datum) (sqlexec.Statement, error) {
|
|
startTime := time.Now()
|
|
defer func() {
|
|
sctx.GetSessionVars().DurationCompile = time.Since(startTime)
|
|
}()
|
|
execStmt := &ast.ExecuteStmt{ExecID: ID}
|
|
if err := ResetContextOfStmt(sctx, execStmt); err != nil {
|
|
return nil, err
|
|
}
|
|
execStmt.BinaryArgs = args
|
|
is := infoschema.GetInfoSchema(sctx)
|
|
execPlan, names, err := planner.Optimize(ctx, sctx, execStmt, is)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
stmt := &ExecStmt{
|
|
GoCtx: ctx,
|
|
InfoSchema: is,
|
|
Plan: execPlan,
|
|
StmtNode: execStmt,
|
|
Ctx: sctx,
|
|
OutputNames: names,
|
|
}
|
|
if preparedPointer, ok := sctx.GetSessionVars().PreparedStmts[ID]; ok {
|
|
preparedObj, ok := preparedPointer.(*plannercore.CachedPrepareStmt)
|
|
if !ok {
|
|
return nil, errors.Errorf("invalid CachedPrepareStmt type")
|
|
}
|
|
stmtCtx := sctx.GetSessionVars().StmtCtx
|
|
stmt.Text = preparedObj.PreparedAst.Stmt.Text()
|
|
stmtCtx.OriginalSQL = stmt.Text
|
|
stmtCtx.InitSQLDigest(preparedObj.NormalizedSQL, preparedObj.SQLDigest)
|
|
}
|
|
return stmt, nil
|
|
}
|
|
|