lock.go

// 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
}