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.
 
 

126 lines
3.7 KiB

// Copyright 2019 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 lock
import (
"github.com/pingcap/parser/model"
"github.com/pingcap/parser/mysql"
"github.com/pingcap/tidb/infoschema"
"github.com/pingcap/tidb/sessionctx"
"github.com/pingcap/tidb/table"
"github.com/pingcap/tidb/util"
)
// Checker uses to check tables lock.
type Checker struct {
ctx sessionctx.Context
is infoschema.InfoSchema
}
// NewChecker return new lock Checker.
func NewChecker(ctx sessionctx.Context, is infoschema.InfoSchema) *Checker {
return &Checker{ctx: ctx, is: is}
}
// CheckTableLock uses to check table lock.
func (c *Checker) CheckTableLock(db, table string, privilege mysql.PrivilegeType) error {
if db == "" && table == "" {
return nil
}
// System DB and memory DB are not support table lock.
if util.IsMemOrSysDB(db) {
return nil
}
// check operation on database.
if table == "" {
return c.CheckLockInDB(db, privilege)
}
switch privilege {
case mysql.ShowDBPriv, mysql.AllPrivMask:
// AllPrivMask only used in show create table statement now.
return nil
case mysql.CreatePriv, mysql.CreateViewPriv:
if c.ctx.HasLockedTables() {
// TODO: For `create table t_exists ...` statement, mysql will check out `t_exists` first, but in TiDB now,
// will return below error first.
return infoschema.ErrTableNotLocked.GenWithStackByArgs(table)
}
return nil
}
// TODO: try to remove this get for speed up.
tb, err := c.is.TableByName(model.NewCIStr(db), model.NewCIStr(table))
// Ignore this error for "drop table if not exists t1" when t1 doesn't exists.
if infoschema.ErrTableNotExists.Equal(err) {
return nil
}
if err != nil {
return err
}
if c.ctx.HasLockedTables() {
if locked, tp := c.ctx.CheckTableLocked(tb.Meta().ID); locked {
if checkLockTpMeetPrivilege(tp, privilege) {
return nil
}
return infoschema.ErrTableNotLockedForWrite.GenWithStackByArgs(tb.Meta().Name)
}
return infoschema.ErrTableNotLocked.GenWithStackByArgs(tb.Meta().Name)
}
if tb.Meta().Lock == nil {
return nil
}
if privilege == mysql.SelectPriv {
switch tb.Meta().Lock.Tp {
case model.TableLockRead, model.TableLockWriteLocal:
return nil
}
}
return infoschema.ErrTableLocked.GenWithStackByArgs(tb.Meta().Name.L, tb.Meta().Lock.Tp, tb.Meta().Lock.Sessions[0])
}
func checkLockTpMeetPrivilege(tp model.TableLockType, privilege mysql.PrivilegeType) bool {
switch tp {
case model.TableLockWrite, model.TableLockWriteLocal:
return true
case model.TableLockRead:
// ShowDBPriv, AllPrivMask,CreatePriv, CreateViewPriv already checked before.
// The other privilege in read lock was not allowed.
if privilege == mysql.SelectPriv {
return true
}
}
return false
}
// CheckLockInDB uses to check operation on database.
func (c *Checker) CheckLockInDB(db string, privilege mysql.PrivilegeType) error {
if c.ctx.HasLockedTables() {
switch privilege {
case mysql.CreatePriv, mysql.DropPriv, mysql.AlterPriv:
return table.ErrLockOrActiveTransaction.GenWithStackByArgs()
}
}
if privilege == mysql.CreatePriv {
return nil
}
tables := c.is.SchemaTables(model.NewCIStr(db))
for _, tbl := range tables {
err := c.CheckTableLock(db, tbl.Meta().Name.L, privilege)
if err != nil {
return err
}
}
return nil
}