32b1dbd8d5
We import "github.com/pkg/errors" in the code, and actually put the "github.com/pingcap/errors" there in the vendor path. That's a dirty hack, and prevent the introduce of Go module.
353 lines
11 KiB
Go
353 lines
11 KiB
Go
// Copyright 2016 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 (
|
|
"fmt"
|
|
"strings"
|
|
|
|
"github.com/pingcap/errors"
|
|
"github.com/pingcap/parser/ast"
|
|
"github.com/pingcap/parser/auth"
|
|
"github.com/pingcap/parser/model"
|
|
"github.com/pingcap/parser/mysql"
|
|
"github.com/pingcap/parser/terror"
|
|
"github.com/pingcap/tidb/config"
|
|
"github.com/pingcap/tidb/domain"
|
|
"github.com/pingcap/tidb/infoschema"
|
|
"github.com/pingcap/tidb/sessionctx"
|
|
"github.com/pingcap/tidb/sessionctx/variable"
|
|
"github.com/pingcap/tidb/util/chunk"
|
|
"github.com/pingcap/tidb/util/sqlexec"
|
|
log "github.com/sirupsen/logrus"
|
|
"golang.org/x/net/context"
|
|
)
|
|
|
|
// SimpleExec represents simple statement executor.
|
|
// For statements do simple execution.
|
|
// includes `UseStmt`, 'SetStmt`, `DoStmt`,
|
|
// `BeginStmt`, `CommitStmt`, `RollbackStmt`.
|
|
// TODO: list all simple statements.
|
|
type SimpleExec struct {
|
|
baseExecutor
|
|
|
|
Statement ast.StmtNode
|
|
done bool
|
|
is infoschema.InfoSchema
|
|
}
|
|
|
|
// Next implements the Executor Next interface.
|
|
func (e *SimpleExec) Next(ctx context.Context, chk *chunk.Chunk) (err error) {
|
|
if e.done {
|
|
return nil
|
|
}
|
|
switch x := e.Statement.(type) {
|
|
case *ast.UseStmt:
|
|
err = e.executeUse(x)
|
|
case *ast.FlushStmt:
|
|
err = e.executeFlush(x)
|
|
case *ast.BeginStmt:
|
|
err = e.executeBegin(x)
|
|
case *ast.CommitStmt:
|
|
e.executeCommit(x)
|
|
case *ast.RollbackStmt:
|
|
err = e.executeRollback(x)
|
|
case *ast.CreateUserStmt:
|
|
err = e.executeCreateUser(x)
|
|
case *ast.AlterUserStmt:
|
|
err = e.executeAlterUser(x)
|
|
case *ast.DropUserStmt:
|
|
err = e.executeDropUser(x)
|
|
case *ast.SetPwdStmt:
|
|
err = e.executeSetPwd(x)
|
|
case *ast.KillStmt:
|
|
err = e.executeKillStmt(x)
|
|
case *ast.BinlogStmt:
|
|
// We just ignore it.
|
|
return nil
|
|
case *ast.DropStatsStmt:
|
|
err = e.executeDropStats(x)
|
|
}
|
|
e.done = true
|
|
return errors.Trace(err)
|
|
}
|
|
|
|
func (e *SimpleExec) executeUse(s *ast.UseStmt) error {
|
|
dbname := model.NewCIStr(s.DBName)
|
|
dbinfo, exists := e.is.SchemaByName(dbname)
|
|
if !exists {
|
|
return infoschema.ErrDatabaseNotExists.GenWithStackByArgs(dbname)
|
|
}
|
|
e.ctx.GetSessionVars().CurrentDB = dbname.O
|
|
// character_set_database is the character set used by the default database.
|
|
// The server sets this variable whenever the default database changes.
|
|
// See http://dev.mysql.com/doc/refman/5.7/en/server-system-variables.html#sysvar_character_set_database
|
|
sessionVars := e.ctx.GetSessionVars()
|
|
terror.Log(errors.Trace(sessionVars.SetSystemVar(variable.CharsetDatabase, dbinfo.Charset)))
|
|
terror.Log(errors.Trace(sessionVars.SetSystemVar(variable.CollationDatabase, dbinfo.Collate)))
|
|
return nil
|
|
}
|
|
|
|
func (e *SimpleExec) executeBegin(s *ast.BeginStmt) error {
|
|
// If BEGIN is the first statement in TxnCtx, we can reuse the existing transaction, without the
|
|
// need to call NewTxn, which commits the existing transaction and begins a new one.
|
|
txnCtx := e.ctx.GetSessionVars().TxnCtx
|
|
if txnCtx.Histroy != nil {
|
|
err := e.ctx.NewTxn()
|
|
if err != nil {
|
|
return errors.Trace(err)
|
|
}
|
|
}
|
|
// With START TRANSACTION, autocommit remains disabled until you end
|
|
// the transaction with COMMIT or ROLLBACK. The autocommit mode then
|
|
// reverts to its previous state.
|
|
e.ctx.GetSessionVars().SetStatusFlag(mysql.ServerStatusInTrans, true)
|
|
return nil
|
|
}
|
|
|
|
func (e *SimpleExec) executeCommit(s *ast.CommitStmt) {
|
|
e.ctx.GetSessionVars().SetStatusFlag(mysql.ServerStatusInTrans, false)
|
|
}
|
|
|
|
func (e *SimpleExec) executeRollback(s *ast.RollbackStmt) error {
|
|
sessVars := e.ctx.GetSessionVars()
|
|
log.Debugf("con:%d execute rollback statement", sessVars.ConnectionID)
|
|
sessVars.SetStatusFlag(mysql.ServerStatusInTrans, false)
|
|
if e.ctx.Txn().Valid() {
|
|
e.ctx.GetSessionVars().TxnCtx.ClearDelta()
|
|
return e.ctx.Txn().Rollback()
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (e *SimpleExec) executeCreateUser(s *ast.CreateUserStmt) error {
|
|
users := make([]string, 0, len(s.Specs))
|
|
for _, spec := range s.Specs {
|
|
exists, err1 := userExists(e.ctx, spec.User.Username, spec.User.Hostname)
|
|
if err1 != nil {
|
|
return errors.Trace(err1)
|
|
}
|
|
if exists {
|
|
if !s.IfNotExists {
|
|
return errors.New("Duplicate user")
|
|
}
|
|
continue
|
|
}
|
|
pwd, ok := spec.EncodedPassword()
|
|
if !ok {
|
|
return errors.Trace(ErrPasswordFormat)
|
|
}
|
|
user := fmt.Sprintf(`("%s", "%s", "%s")`, spec.User.Hostname, spec.User.Username, pwd)
|
|
users = append(users, user)
|
|
}
|
|
if len(users) == 0 {
|
|
return nil
|
|
}
|
|
sql := fmt.Sprintf(`INSERT INTO %s.%s (Host, User, Password) VALUES %s;`, mysql.SystemDB, mysql.UserTable, strings.Join(users, ", "))
|
|
_, err := e.ctx.(sqlexec.SQLExecutor).Execute(context.Background(), sql)
|
|
if err != nil {
|
|
return errors.Trace(err)
|
|
}
|
|
domain.GetDomain(e.ctx).NotifyUpdatePrivilege(e.ctx)
|
|
return errors.Trace(err)
|
|
}
|
|
|
|
func (e *SimpleExec) executeAlterUser(s *ast.AlterUserStmt) error {
|
|
if s.CurrentAuth != nil {
|
|
user := e.ctx.GetSessionVars().User
|
|
if user == nil {
|
|
return errors.New("Session user is empty")
|
|
}
|
|
spec := &ast.UserSpec{
|
|
User: user,
|
|
AuthOpt: s.CurrentAuth,
|
|
}
|
|
s.Specs = []*ast.UserSpec{spec}
|
|
}
|
|
|
|
failedUsers := make([]string, 0, len(s.Specs))
|
|
for _, spec := range s.Specs {
|
|
exists, err := userExists(e.ctx, spec.User.Username, spec.User.Hostname)
|
|
if err != nil {
|
|
return errors.Trace(err)
|
|
}
|
|
if !exists {
|
|
failedUsers = append(failedUsers, spec.User.String())
|
|
if s.IfExists {
|
|
// TODO: Make this error as a warning.
|
|
}
|
|
continue
|
|
}
|
|
pwd := ""
|
|
if spec.AuthOpt != nil {
|
|
if spec.AuthOpt.ByAuthString {
|
|
pwd = auth.EncodePassword(spec.AuthOpt.AuthString)
|
|
} else {
|
|
pwd = auth.EncodePassword(spec.AuthOpt.HashString)
|
|
}
|
|
}
|
|
sql := fmt.Sprintf(`UPDATE %s.%s SET Password = "%s" WHERE Host = "%s" and User = "%s";`,
|
|
mysql.SystemDB, mysql.UserTable, pwd, spec.User.Hostname, spec.User.Username)
|
|
_, _, err = e.ctx.(sqlexec.RestrictedSQLExecutor).ExecRestrictedSQL(e.ctx, sql)
|
|
if err != nil {
|
|
failedUsers = append(failedUsers, spec.User.String())
|
|
}
|
|
}
|
|
if len(failedUsers) > 0 {
|
|
// Commit the transaction even if we returns error
|
|
err := e.ctx.Txn().Commit(sessionctx.SetCommitCtx(context.Background(), e.ctx))
|
|
if err != nil {
|
|
return errors.Trace(err)
|
|
}
|
|
return ErrCannotUser.GenWithStackByArgs("ALTER USER", strings.Join(failedUsers, ","))
|
|
}
|
|
domain.GetDomain(e.ctx).NotifyUpdatePrivilege(e.ctx)
|
|
return nil
|
|
}
|
|
|
|
func (e *SimpleExec) executeDropUser(s *ast.DropUserStmt) error {
|
|
failedUsers := make([]string, 0, len(s.UserList))
|
|
for _, user := range s.UserList {
|
|
exists, err := userExists(e.ctx, user.Username, user.Hostname)
|
|
if err != nil {
|
|
return errors.Trace(err)
|
|
}
|
|
if !exists {
|
|
if !s.IfExists {
|
|
failedUsers = append(failedUsers, user.String())
|
|
}
|
|
continue
|
|
}
|
|
|
|
// begin a transaction to delete a user.
|
|
if _, err := e.ctx.(sqlexec.SQLExecutor).Execute(context.Background(), "begin"); err != nil {
|
|
return errors.Trace(err)
|
|
}
|
|
sql := fmt.Sprintf(`DELETE FROM %s.%s WHERE Host = "%s" and User = "%s";`, mysql.SystemDB, mysql.UserTable, user.Hostname, user.Username)
|
|
if _, err := e.ctx.(sqlexec.SQLExecutor).Execute(context.Background(), sql); err != nil {
|
|
failedUsers = append(failedUsers, user.String())
|
|
if _, err := e.ctx.(sqlexec.SQLExecutor).Execute(context.Background(), "rollback"); err != nil {
|
|
return errors.Trace(err)
|
|
}
|
|
continue
|
|
}
|
|
|
|
// delete privileges from mysql.db
|
|
sql = fmt.Sprintf(`DELETE FROM %s.%s WHERE Host = "%s" and User = "%s";`, mysql.SystemDB, mysql.DBTable, user.Hostname, user.Username)
|
|
if _, err := e.ctx.(sqlexec.SQLExecutor).Execute(context.Background(), sql); err != nil {
|
|
failedUsers = append(failedUsers, user.String())
|
|
if _, err := e.ctx.(sqlexec.SQLExecutor).Execute(context.Background(), "rollback"); err != nil {
|
|
return errors.Trace(err)
|
|
}
|
|
continue
|
|
}
|
|
|
|
// delete privileges from mysql.tables_priv
|
|
sql = fmt.Sprintf(`DELETE FROM %s.%s WHERE Host = "%s" and User = "%s";`, mysql.SystemDB, mysql.TablePrivTable, user.Hostname, user.Username)
|
|
if _, err := e.ctx.(sqlexec.SQLExecutor).Execute(context.Background(), sql); err != nil {
|
|
failedUsers = append(failedUsers, user.String())
|
|
if _, err := e.ctx.(sqlexec.SQLExecutor).Execute(context.Background(), "rollback"); err != nil {
|
|
return errors.Trace(err)
|
|
}
|
|
continue
|
|
}
|
|
|
|
//TODO: need delete columns_priv once we implement columns_priv functionality.
|
|
if _, err := e.ctx.(sqlexec.SQLExecutor).Execute(context.Background(), "commit"); err != nil {
|
|
failedUsers = append(failedUsers, user.String())
|
|
}
|
|
}
|
|
if len(failedUsers) > 0 {
|
|
return ErrCannotUser.GenWithStackByArgs("DROP USER", strings.Join(failedUsers, ","))
|
|
}
|
|
domain.GetDomain(e.ctx).NotifyUpdatePrivilege(e.ctx)
|
|
return nil
|
|
}
|
|
|
|
func userExists(ctx sessionctx.Context, name string, host string) (bool, error) {
|
|
sql := fmt.Sprintf(`SELECT * FROM %s.%s WHERE User="%s" AND Host="%s";`, mysql.SystemDB, mysql.UserTable, name, host)
|
|
rows, _, err := ctx.(sqlexec.RestrictedSQLExecutor).ExecRestrictedSQL(ctx, sql)
|
|
if err != nil {
|
|
return false, errors.Trace(err)
|
|
}
|
|
return len(rows) > 0, nil
|
|
}
|
|
|
|
func (e *SimpleExec) executeSetPwd(s *ast.SetPwdStmt) error {
|
|
if s.User == nil {
|
|
vars := e.ctx.GetSessionVars()
|
|
s.User = vars.User
|
|
if s.User == nil {
|
|
return errors.New("Session error is empty")
|
|
}
|
|
}
|
|
exists, err := userExists(e.ctx, s.User.Username, s.User.Hostname)
|
|
if err != nil {
|
|
return errors.Trace(err)
|
|
}
|
|
if !exists {
|
|
return errors.Trace(ErrPasswordNoMatch)
|
|
}
|
|
|
|
// update mysql.user
|
|
sql := fmt.Sprintf(`UPDATE %s.%s SET password="%s" WHERE User="%s" AND Host="%s";`, mysql.SystemDB, mysql.UserTable, auth.EncodePassword(s.Password), s.User.Username, s.User.Hostname)
|
|
_, _, err = e.ctx.(sqlexec.RestrictedSQLExecutor).ExecRestrictedSQL(e.ctx, sql)
|
|
domain.GetDomain(e.ctx).NotifyUpdatePrivilege(e.ctx)
|
|
return errors.Trace(err)
|
|
}
|
|
|
|
func (e *SimpleExec) executeKillStmt(s *ast.KillStmt) error {
|
|
conf := config.GetGlobalConfig()
|
|
if s.TiDBExtension || conf.CompatibleKillQuery {
|
|
sm := e.ctx.GetSessionManager()
|
|
if sm == nil {
|
|
return nil
|
|
}
|
|
sm.Kill(s.ConnectionID, s.Query)
|
|
} else {
|
|
err := errors.New("Invalid operation. Please use 'KILL TIDB [CONNECTION | QUERY] connectionID' instead")
|
|
e.ctx.GetSessionVars().StmtCtx.AppendWarning(err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (e *SimpleExec) executeFlush(s *ast.FlushStmt) error {
|
|
switch s.Tp {
|
|
case ast.FlushTables:
|
|
if s.ReadLock {
|
|
return errors.New("FLUSH TABLES WITH READ LOCK is not supported. Please use @@tidb_snapshot")
|
|
}
|
|
case ast.FlushPrivileges:
|
|
dom := domain.GetDomain(e.ctx)
|
|
sysSessionPool := dom.SysSessionPool()
|
|
ctx, err := sysSessionPool.Get()
|
|
if err != nil {
|
|
return errors.Trace(err)
|
|
}
|
|
defer sysSessionPool.Put(ctx)
|
|
err = dom.PrivilegeHandle().Update(ctx.(sessionctx.Context))
|
|
return errors.Trace(err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (e *SimpleExec) executeDropStats(s *ast.DropStatsStmt) error {
|
|
h := domain.GetDomain(e.ctx).StatsHandle()
|
|
err := h.DeleteTableStatsFromKV(s.Table.TableInfo.ID)
|
|
if err != nil {
|
|
return errors.Trace(err)
|
|
}
|
|
return errors.Trace(h.Update(GetInfoSchema(e.ctx)))
|
|
}
|