// 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/hanchuanchuan/goInception/ast" "github.com/hanchuanchuan/goInception/config" "github.com/hanchuanchuan/goInception/domain" "github.com/hanchuanchuan/goInception/infoschema" "github.com/hanchuanchuan/goInception/model" "github.com/hanchuanchuan/goInception/mysql" "github.com/hanchuanchuan/goInception/sessionctx" "github.com/hanchuanchuan/goInception/sessionctx/variable" "github.com/hanchuanchuan/goInception/terror" "github.com/hanchuanchuan/goInception/util/auth" "github.com/hanchuanchuan/goInception/util/chunk" "github.com/hanchuanchuan/goInception/util/sqlexec" "github.com/pingcap/errors" 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))) }