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.
683 lines
25 KiB
683 lines
25 KiB
// Copyright 2020 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 session
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/hex"
|
|
"fmt"
|
|
"reflect"
|
|
|
|
"github.com/pingcap/errors"
|
|
pb "github.com/pingcap/kvproto/pkg/kvrpcpb"
|
|
"github.com/pingcap/parser/model"
|
|
"github.com/pingcap/tidb/ddl"
|
|
"github.com/pingcap/tidb/executor"
|
|
"github.com/pingcap/tidb/expression"
|
|
"github.com/pingcap/tidb/infoschema"
|
|
"github.com/pingcap/tidb/kv"
|
|
"github.com/pingcap/tidb/sessionctx"
|
|
"github.com/pingcap/tidb/store/tikv"
|
|
"github.com/pingcap/tidb/table"
|
|
"github.com/pingcap/tidb/table/tables"
|
|
"github.com/pingcap/tidb/tablecodec"
|
|
"github.com/pingcap/tidb/types"
|
|
"github.com/pingcap/tidb/util/chunk"
|
|
"github.com/pingcap/tidb/util/logutil"
|
|
"github.com/pingcap/tidb/util/rowcodec"
|
|
"go.uber.org/zap"
|
|
)
|
|
|
|
const amendableType = nonMemAmendType | memBufAmendType
|
|
const nonMemAmendType = (1 << model.ActionAddColumn) | (1 << model.ActionDropColumn) | (1 << model.ActionDropIndex)
|
|
const memBufAmendType = uint64(1<<model.ActionAddIndex) | (1 << model.ActionModifyColumn)
|
|
|
|
// Amend operation types.
|
|
const (
|
|
AmendNone int = iota
|
|
|
|
// For add index.
|
|
AmendNeedAddDelete
|
|
AmendNeedAddDeleteAndInsert
|
|
AmendNeedAddInsert
|
|
)
|
|
|
|
// ConstOpAddIndex is the possible ddl state changes, and related amend action types.
|
|
var ConstOpAddIndex = map[model.SchemaState]map[model.SchemaState]int{
|
|
model.StateNone: {
|
|
model.StateDeleteOnly: AmendNeedAddDelete,
|
|
model.StateWriteOnly: AmendNeedAddDeleteAndInsert,
|
|
model.StateWriteReorganization: AmendNeedAddDeleteAndInsert,
|
|
model.StatePublic: AmendNeedAddDeleteAndInsert,
|
|
},
|
|
model.StateDeleteOnly: {
|
|
model.StateWriteOnly: AmendNeedAddInsert,
|
|
model.StateWriteReorganization: AmendNeedAddInsert,
|
|
model.StatePublic: AmendNeedAddInsert,
|
|
},
|
|
model.StateWriteOnly: {
|
|
model.StateWriteReorganization: AmendNone,
|
|
model.StatePublic: AmendNone,
|
|
},
|
|
model.StateWriteReorganization: {
|
|
model.StatePublic: AmendNone,
|
|
},
|
|
}
|
|
|
|
type schemaAndDecoder struct {
|
|
schema *expression.Schema
|
|
decoder *rowcodec.ChunkDecoder
|
|
}
|
|
|
|
// amendCollector collects all amend operations.
|
|
type amendCollector struct {
|
|
tblAmendOpMap map[int64][]amendOp
|
|
}
|
|
|
|
func newAmendCollector() *amendCollector {
|
|
res := &amendCollector{
|
|
tblAmendOpMap: make(map[int64][]amendOp),
|
|
}
|
|
return res
|
|
}
|
|
|
|
func findIndexByID(tbl table.Table, ID int64) table.Index {
|
|
for _, indexInfo := range tbl.Indices() {
|
|
if indexInfo.Meta().ID == ID {
|
|
return indexInfo
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func findColByID(tbl table.Table, colID int64) *table.Column {
|
|
for _, colInfo := range tbl.Cols() {
|
|
if colInfo.ID == colID {
|
|
return colInfo
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func addIndexNeedRemoveOp(amendOp int) bool {
|
|
if amendOp == AmendNeedAddDelete || amendOp == AmendNeedAddDeleteAndInsert {
|
|
return true
|
|
}
|
|
return false
|
|
}
|
|
|
|
func addIndexNeedAddOp(amendOp int) bool {
|
|
if amendOp == AmendNeedAddDeleteAndInsert || amendOp == AmendNeedAddInsert {
|
|
return true
|
|
}
|
|
return false
|
|
}
|
|
|
|
func (a *amendCollector) keyHasAmendOp(key []byte) bool {
|
|
tblID := tablecodec.DecodeTableID(key)
|
|
ops := a.tblAmendOpMap[tblID]
|
|
return len(ops) > 0
|
|
}
|
|
|
|
func needCollectIndexOps(actionType uint64) bool {
|
|
return actionType&(1<<model.ActionAddIndex) != 0
|
|
}
|
|
|
|
func needCollectModifyColOps(actionType uint64) bool {
|
|
return actionType&(1<<model.ActionModifyColumn) != 0
|
|
}
|
|
|
|
func fieldTypeDeepEquals(ft1 *types.FieldType, ft2 *types.FieldType) bool {
|
|
if ft1.Tp == ft2.Tp &&
|
|
ft1.Flag == ft2.Flag &&
|
|
ft1.Flen == ft2.Flen &&
|
|
ft1.Decimal == ft2.Decimal &&
|
|
ft1.Charset == ft2.Charset &&
|
|
ft1.Collate == ft2.Collate &&
|
|
len(ft1.Elems) == len(ft2.Elems) {
|
|
for i, elem := range ft1.Elems {
|
|
if elem != ft2.Elems[i] {
|
|
return false
|
|
}
|
|
}
|
|
return true
|
|
}
|
|
return false
|
|
}
|
|
|
|
// colChangeAmendable checks whether the column change is amendable, now only increasing column field
|
|
// length is allowed for committing concurrent pessimistic transactions.
|
|
func colChangeAmendable(colAtStart *model.ColumnInfo, colAtCommit *model.ColumnInfo) error {
|
|
// Modifying a stored generated column is not allowed by DDL, the generated related fields are not considered.
|
|
if !fieldTypeDeepEquals(&colAtStart.FieldType, &colAtCommit.FieldType) {
|
|
if colAtStart.FieldType.Flag != colAtCommit.FieldType.Flag {
|
|
return errors.Trace(errors.Errorf("flag is not matched for column=%v, from=%v to=%v",
|
|
colAtCommit.Name.String(), colAtStart.FieldType.Flag, colAtCommit.FieldType.Flag))
|
|
}
|
|
if colAtStart.Charset != colAtCommit.Charset || colAtStart.Collate != colAtCommit.Collate {
|
|
return errors.Trace(errors.Errorf("charset or collate is not matched for column=%v", colAtCommit.Name.String()))
|
|
}
|
|
err := ddl.CheckModifyTypeCompatible(&colAtStart.FieldType, &colAtCommit.FieldType)
|
|
if err != nil {
|
|
return errors.Trace(err)
|
|
}
|
|
}
|
|
// TODO Default value change is not supported.
|
|
if !reflect.DeepEqual(colAtStart.DefaultValue, colAtCommit.DefaultValue) {
|
|
return errors.Trace(errors.Errorf("default value is not matched for column=%v, from=%v to=%v",
|
|
colAtCommit.Name.String(), colAtStart.DefaultValue, colAtCommit.DefaultValue))
|
|
}
|
|
if !bytes.Equal(colAtStart.DefaultValueBit, colAtCommit.DefaultValueBit) {
|
|
return errors.Trace(errors.Errorf("default value bits is not matched for column=%v, from=%v to=%v",
|
|
colAtCommit.Name.String(), colAtStart.DefaultValueBit, colAtCommit.DefaultValueBit))
|
|
}
|
|
if colAtStart.Version != colAtCommit.Version {
|
|
return errors.Trace(errors.Errorf("column version is not matched for column=%v, from=%v to=%v",
|
|
colAtCommit.Name.String(), colAtStart.Version, colAtCommit.Version))
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// collectModifyColAmendOps is used to check if there is column change from nullable to not null by now.
|
|
// TODO allow column change from nullable to not null, and generate keys check operation.
|
|
func (a *amendCollector) collectModifyColAmendOps(tblAtStart, tblAtCommit table.Table) ([]amendOp, error) {
|
|
for _, colAtCommit := range tblAtCommit.Cols() {
|
|
colAtStart := findColByID(tblAtStart, colAtCommit.ID)
|
|
if colAtStart != nil {
|
|
err := colChangeAmendable(colAtStart.ColumnInfo, colAtCommit.ColumnInfo)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
}
|
|
return nil, nil
|
|
}
|
|
|
|
func (a *amendCollector) collectIndexAmendOps(sctx sessionctx.Context, tblAtStart, tblAtCommit table.Table) ([]amendOp, error) {
|
|
res := make([]amendOp, 0, 4)
|
|
// Check index having state change, collect index column info.
|
|
for _, idxInfoAtCommit := range tblAtCommit.Indices() {
|
|
idxInfoAtStart := findIndexByID(tblAtStart, idxInfoAtCommit.Meta().ID)
|
|
// Try to find index state change.
|
|
var amendOpType int
|
|
if idxInfoAtStart == nil {
|
|
amendOpType = ConstOpAddIndex[model.StateNone][idxInfoAtCommit.Meta().State]
|
|
} else if idxInfoAtCommit.Meta().State > idxInfoAtStart.Meta().State {
|
|
amendOpType = ConstOpAddIndex[idxInfoAtStart.Meta().State][idxInfoAtCommit.Meta().State]
|
|
}
|
|
if amendOpType != AmendNone {
|
|
opInfo := &amendOperationAddIndexInfo{}
|
|
opInfo.AmendOpType = amendOpType
|
|
opInfo.tblInfoAtStart = tblAtStart
|
|
opInfo.tblInfoAtCommit = tblAtCommit
|
|
opInfo.indexInfoAtStart = idxInfoAtStart
|
|
opInfo.indexInfoAtCommit = idxInfoAtCommit
|
|
for _, idxCol := range idxInfoAtCommit.Meta().Columns {
|
|
colID := tblAtCommit.Meta().Columns[idxCol.Offset].ID
|
|
oldColInfo := findColByID(tblAtStart, colID)
|
|
// TODO: now index column MUST be found in old table columns, generated column is not supported.
|
|
if oldColInfo == nil || oldColInfo.IsGenerated() || oldColInfo.Hidden {
|
|
return nil, errors.Trace(errors.Errorf("amend index column=%v id=%v is not found or generated in table=%v",
|
|
idxCol.Name, colID, tblAtCommit.Meta().Name.String()))
|
|
}
|
|
opInfo.relatedOldIdxCols = append(opInfo.relatedOldIdxCols, oldColInfo)
|
|
}
|
|
opInfo.schemaAndDecoder = newSchemaAndDecoder(sctx, tblAtStart.Meta())
|
|
fieldTypes := make([]*types.FieldType, 0, len(tblAtStart.Meta().Columns))
|
|
for _, col := range tblAtStart.Meta().Columns {
|
|
fieldTypes = append(fieldTypes, &col.FieldType)
|
|
}
|
|
opInfo.chk = chunk.NewChunkWithCapacity(fieldTypes, 4)
|
|
addNewIndexOp := &amendOperationAddIndex{
|
|
info: opInfo,
|
|
insertedNewIndexKeys: make(map[string]struct{}),
|
|
deletedOldIndexKeys: make(map[string]struct{}),
|
|
}
|
|
res = append(res, addNewIndexOp)
|
|
}
|
|
}
|
|
return res, nil
|
|
}
|
|
|
|
// collectTblAmendOps collects amend operations for each table using the schema diff between startTS and commitTS.
|
|
func (a *amendCollector) collectTblAmendOps(sctx sessionctx.Context, phyTblID int64,
|
|
tblInfoAtStart, tblInfoAtCommit table.Table, actionType uint64) error {
|
|
if _, ok := a.tblAmendOpMap[phyTblID]; !ok {
|
|
a.tblAmendOpMap[phyTblID] = make([]amendOp, 0, 4)
|
|
}
|
|
if needCollectModifyColOps(actionType) {
|
|
_, err := a.collectModifyColAmendOps(tblInfoAtStart, tblInfoAtCommit)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
if needCollectIndexOps(actionType) {
|
|
// TODO: currently only "add index" is considered.
|
|
ops, err := a.collectIndexAmendOps(sctx, tblInfoAtStart, tblInfoAtCommit)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
a.tblAmendOpMap[phyTblID] = append(a.tblAmendOpMap[phyTblID], ops...)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// mayGenDelIndexRowKeyOp returns if the row key op could generate Op_Del index key mutations.
|
|
func mayGenDelIndexRowKeyOp(keyOp pb.Op) bool {
|
|
return keyOp == pb.Op_Del || keyOp == pb.Op_Put
|
|
}
|
|
|
|
// mayGenPutIndexRowKeyOp returns if the row key op could generate Op_Put/Op_Insert index key mutations.
|
|
func mayGenPutIndexRowKeyOp(keyOp pb.Op) bool {
|
|
return keyOp == pb.Op_Put || keyOp == pb.Op_Insert
|
|
}
|
|
|
|
// amendOp is an amend operation for a specific schema change, new mutations will be generated using input ones.
|
|
type amendOp interface {
|
|
genMutations(ctx context.Context, sctx sessionctx.Context, commitMutations tikv.CommitterMutations, kvMap *rowKvMap,
|
|
resultMutations *tikv.CommitterMutations) error
|
|
}
|
|
|
|
// amendOperationAddIndex represents one amend operation related to a specific add index change.
|
|
type amendOperationAddIndexInfo struct {
|
|
AmendOpType int
|
|
tblInfoAtStart table.Table
|
|
tblInfoAtCommit table.Table
|
|
indexInfoAtStart table.Index
|
|
indexInfoAtCommit table.Index
|
|
relatedOldIdxCols []*table.Column
|
|
|
|
schemaAndDecoder *schemaAndDecoder
|
|
chk *chunk.Chunk
|
|
}
|
|
|
|
// amendOperationAddIndex represents the add operation will be performed on new key values for add index amend.
|
|
type amendOperationAddIndex struct {
|
|
info *amendOperationAddIndexInfo
|
|
|
|
// insertedNewIndexKeys is used to check duplicates for new index generated by unique key.
|
|
insertedNewIndexKeys map[string]struct{}
|
|
// deletedOldIndexKeys is used to check duplicates for deleted old index keys.
|
|
deletedOldIndexKeys map[string]struct{}
|
|
}
|
|
|
|
func (a *amendOperationAddIndexInfo) String() string {
|
|
var colStr string
|
|
colStr += "["
|
|
for _, colInfo := range a.relatedOldIdxCols {
|
|
colStr += fmt.Sprintf(" %s ", colInfo.Name)
|
|
}
|
|
colStr += "]"
|
|
res := fmt.Sprintf("AmenedOpType=%d phyTblID=%d idxID=%d columns=%v", a.AmendOpType, a.indexInfoAtCommit.Meta().ID,
|
|
a.indexInfoAtCommit.Meta().ID, colStr)
|
|
return res
|
|
}
|
|
|
|
func (a *amendOperationAddIndex) genMutations(ctx context.Context, sctx sessionctx.Context, commitMutations tikv.CommitterMutations,
|
|
kvMap *rowKvMap, resAddMutations *tikv.CommitterMutations) error {
|
|
// There should be no duplicate keys in deletedOldIndexKeys and insertedNewIndexKeys.
|
|
deletedMutations := tikv.NewCommiterMutations(32)
|
|
insertedMutations := tikv.NewCommiterMutations(32)
|
|
for i, key := range commitMutations.GetKeys() {
|
|
if tablecodec.IsIndexKey(key) || tablecodec.DecodeTableID(key) != a.info.tblInfoAtCommit.Meta().ID {
|
|
continue
|
|
}
|
|
var newIdxMutation *tikv.Mutation
|
|
var oldIdxMutation *tikv.Mutation
|
|
var err error
|
|
keyOp := commitMutations.GetOps()[i]
|
|
if addIndexNeedRemoveOp(a.info.AmendOpType) {
|
|
if mayGenDelIndexRowKeyOp(keyOp) {
|
|
oldIdxMutation, err = a.genOldIdxKey(ctx, sctx, key, kvMap.oldRowKvMap)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
if addIndexNeedAddOp(a.info.AmendOpType) {
|
|
if mayGenPutIndexRowKeyOp(keyOp) {
|
|
newIdxMutation, err = a.genNewIdxKey(ctx, sctx, key, kvMap.newRowKvMap)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
skipMerge := false
|
|
if a.info.AmendOpType == AmendNeedAddDeleteAndInsert {
|
|
// If the old index key is the same with new index key, then the index related row value
|
|
// is not changed in this row, we don't need to add or remove index keys for this row.
|
|
if oldIdxMutation != nil && newIdxMutation != nil {
|
|
if bytes.Equal(oldIdxMutation.Key, newIdxMutation.Key) {
|
|
skipMerge = true
|
|
}
|
|
}
|
|
}
|
|
if !skipMerge {
|
|
if oldIdxMutation != nil {
|
|
deletedMutations.AppendMutation(*oldIdxMutation)
|
|
}
|
|
if newIdxMutation != nil {
|
|
insertedMutations.AppendMutation(*newIdxMutation)
|
|
}
|
|
}
|
|
}
|
|
// For unique index, there may be conflicts on the same unique index key from different rows.Consider a update statement,
|
|
// "Op_Del" on row_key = 3, row_val = 4, the "Op_Del" unique_key_4 -> nil will be generated.
|
|
// "Op_Put" on row_key = 0, row_val = 4, the "Op_Insert" unique_key_4 -> 0 will be generated.
|
|
// The "Op_Insert" should cover the "Op_Del" otherwise the new put row value will not have a correspond index value.
|
|
if a.info.indexInfoAtCommit.Meta().Unique {
|
|
for i := 0; i < len(deletedMutations.GetKeys()); i++ {
|
|
key := deletedMutations.GetKeys()[i]
|
|
if _, ok := a.insertedNewIndexKeys[string(key)]; !ok {
|
|
resAddMutations.Push(deletedMutations.GetOps()[i], key, deletedMutations.GetValues()[i], deletedMutations.GetPessimisticFlags()[i])
|
|
}
|
|
}
|
|
for i := 0; i < len(insertedMutations.GetKeys()); i++ {
|
|
key := insertedMutations.GetKeys()[i]
|
|
destKeyOp := pb.Op_Insert
|
|
if _, ok := a.deletedOldIndexKeys[string(key)]; ok {
|
|
destKeyOp = pb.Op_Put
|
|
}
|
|
resAddMutations.Push(destKeyOp, key, insertedMutations.GetValues()[i], insertedMutations.GetPessimisticFlags()[i])
|
|
}
|
|
} else {
|
|
resAddMutations.MergeMutations(deletedMutations)
|
|
resAddMutations.MergeMutations(insertedMutations)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (a *amendOperationAddIndexInfo) genIndexKeyValue(ctx context.Context, sctx sessionctx.Context, kvMap map[string][]byte,
|
|
key []byte, kvHandle int64, keyOnly bool) ([]byte, []byte, error) {
|
|
chk := a.chk
|
|
chk.Reset()
|
|
val, ok := kvMap[string(key)]
|
|
if !ok {
|
|
// The Op_Put may not exist in old value kv map.
|
|
if keyOnly {
|
|
return nil, nil, nil
|
|
}
|
|
return nil, nil, errors.Errorf("key=%v is not found in new row kv map", kv.Key(key).String())
|
|
}
|
|
err := executor.DecodeRowValToChunk(sctx, a.schemaAndDecoder.schema, a.tblInfoAtStart.Meta(), kvHandle, val, chk, a.schemaAndDecoder.decoder)
|
|
if err != nil {
|
|
logutil.Logger(ctx).Warn("amend decode value to chunk failed", zap.Error(err))
|
|
return nil, nil, errors.Trace(err)
|
|
}
|
|
idxVals := make([]types.Datum, 0, len(a.indexInfoAtCommit.Meta().Columns))
|
|
for _, oldCol := range a.relatedOldIdxCols {
|
|
idxVals = append(idxVals, chk.GetRow(0).GetDatum(oldCol.Offset, &oldCol.FieldType))
|
|
}
|
|
|
|
// Generate index key buf.
|
|
newIdxKey, distinct, err := tablecodec.GenIndexKey(sctx.GetSessionVars().StmtCtx,
|
|
a.tblInfoAtCommit.Meta(), a.indexInfoAtCommit.Meta(), a.tblInfoAtCommit.Meta().ID, idxVals, kvHandle, nil)
|
|
if err != nil {
|
|
logutil.Logger(ctx).Warn("amend generate index key failed", zap.Error(err))
|
|
return nil, nil, errors.Trace(err)
|
|
}
|
|
if keyOnly {
|
|
return newIdxKey, []byte{}, nil
|
|
}
|
|
|
|
// Generate index value buf.
|
|
containsNonBinaryString := tables.ContainsNonBinaryString(a.indexInfoAtCommit.Meta().Columns, a.tblInfoAtCommit.Meta().Columns)
|
|
newIdxVal, err := tablecodec.GenIndexValue(sctx.GetSessionVars().StmtCtx, a.tblInfoAtCommit.Meta(),
|
|
a.indexInfoAtCommit.Meta(), containsNonBinaryString, distinct, false, idxVals, kvHandle)
|
|
if err != nil {
|
|
logutil.Logger(ctx).Warn("amend generate index values failed", zap.Error(err))
|
|
return nil, nil, errors.Trace(err)
|
|
}
|
|
return newIdxKey, newIdxVal, nil
|
|
}
|
|
|
|
func (a *amendOperationAddIndex) genNewIdxKey(ctx context.Context, sctx sessionctx.Context, key []byte,
|
|
kvMap map[string][]byte) (*tikv.Mutation, error) {
|
|
kvHandle, err := tablecodec.DecodeRowKey(key)
|
|
if err != nil {
|
|
logutil.Logger(ctx).Error("decode key error", zap.String("key", hex.EncodeToString(key)), zap.Error(err))
|
|
return nil, errors.Trace(err)
|
|
}
|
|
|
|
newIdxKey, newIdxValue, err := a.info.genIndexKeyValue(ctx, sctx, kvMap, key, kvHandle, false)
|
|
if err != nil {
|
|
return nil, errors.Trace(err)
|
|
}
|
|
newIndexOp := pb.Op_Put
|
|
isPessimisticLock := false
|
|
if _, ok := a.insertedNewIndexKeys[string(newIdxKey)]; ok {
|
|
return nil, errors.Trace(errors.Errorf("amend process key same key=%v found for index=%v in table=%v",
|
|
newIdxKey, a.info.indexInfoAtCommit.Meta().Name, a.info.tblInfoAtCommit.Meta().Name))
|
|
}
|
|
if a.info.indexInfoAtCommit.Meta().Unique {
|
|
newIndexOp = pb.Op_Insert
|
|
isPessimisticLock = true
|
|
}
|
|
a.insertedNewIndexKeys[string(newIdxKey)] = struct{}{}
|
|
newMutation := &tikv.Mutation{KeyOp: newIndexOp, Key: newIdxKey, Value: newIdxValue, IsPessimisticLock: isPessimisticLock}
|
|
return newMutation, nil
|
|
}
|
|
|
|
func (a *amendOperationAddIndex) genOldIdxKey(ctx context.Context, sctx sessionctx.Context, key []byte,
|
|
oldValKvMap map[string][]byte) (*tikv.Mutation, error) {
|
|
kvHandle, err := tablecodec.DecodeRowKey(key)
|
|
if err != nil {
|
|
logutil.Logger(ctx).Error("decode key error", zap.String("key", hex.EncodeToString(key)), zap.Error(err))
|
|
return nil, errors.Trace(err)
|
|
}
|
|
// Generated delete index key value.
|
|
newIdxKey, emptyVal, err := a.info.genIndexKeyValue(ctx, sctx, oldValKvMap, key, kvHandle, true)
|
|
if err != nil {
|
|
return nil, errors.Trace(err)
|
|
}
|
|
// For Op_Put the key may not exist in old key value map.
|
|
if len(newIdxKey) > 0 {
|
|
isPessimisticLock := false
|
|
if _, ok := a.deletedOldIndexKeys[string(newIdxKey)]; ok {
|
|
return nil, errors.Trace(errors.Errorf("amend process key same key=%v found for index=%v in table=%v",
|
|
newIdxKey, a.info.indexInfoAtCommit.Meta().Name, a.info.tblInfoAtCommit.Meta().Name))
|
|
}
|
|
if a.info.indexInfoAtCommit.Meta().Unique {
|
|
isPessimisticLock = true
|
|
}
|
|
a.deletedOldIndexKeys[string(newIdxKey)] = struct{}{}
|
|
return &tikv.Mutation{KeyOp: pb.Op_Del, Key: newIdxKey, Value: emptyVal, IsPessimisticLock: isPessimisticLock}, nil
|
|
}
|
|
return nil, nil
|
|
}
|
|
|
|
// SchemaAmender is used to amend pessimistic transactions for schema change.
|
|
type SchemaAmender struct {
|
|
sess *session
|
|
}
|
|
|
|
// NewSchemaAmenderForTikvTxn creates a schema amender for tikvTxn type.
|
|
func NewSchemaAmenderForTikvTxn(sess *session) *SchemaAmender {
|
|
amender := &SchemaAmender{sess: sess}
|
|
return amender
|
|
}
|
|
|
|
func (s *SchemaAmender) getAmendableKeys(commitMutations tikv.CommitterMutations, info *amendCollector) ([]kv.Key, []kv.Key) {
|
|
addKeys := make([]kv.Key, 0, len(commitMutations.GetKeys()))
|
|
removeKeys := make([]kv.Key, 0, len(commitMutations.GetKeys()))
|
|
for i, byteKey := range commitMutations.GetKeys() {
|
|
if tablecodec.IsIndexKey(byteKey) || !info.keyHasAmendOp(byteKey) {
|
|
continue
|
|
}
|
|
keyOp := commitMutations.GetOps()[i]
|
|
if pb.Op_Put == keyOp {
|
|
addKeys = append(addKeys, byteKey)
|
|
removeKeys = append(removeKeys, byteKey)
|
|
} else if pb.Op_Insert == keyOp {
|
|
addKeys = append(addKeys, byteKey)
|
|
} else if pb.Op_Del == keyOp {
|
|
removeKeys = append(removeKeys, byteKey)
|
|
} else {
|
|
// Do nothing.
|
|
}
|
|
}
|
|
return addKeys, removeKeys
|
|
}
|
|
|
|
type rowKvMap struct {
|
|
oldRowKvMap map[string][]byte
|
|
newRowKvMap map[string][]byte
|
|
}
|
|
|
|
func (s *SchemaAmender) prepareKvMap(ctx context.Context, commitMutations tikv.CommitterMutations, info *amendCollector) (*rowKvMap, error) {
|
|
// Get keys need to be considered for the amend operation, currently only row keys.
|
|
addKeys, removeKeys := s.getAmendableKeys(commitMutations, info)
|
|
|
|
// BatchGet the new key values, the Op_Put and Op_Insert type keys in memory buffer.
|
|
txn, err := s.sess.Txn(true)
|
|
if err != nil {
|
|
return nil, errors.Trace(err)
|
|
}
|
|
newValKvMap, err := txn.BatchGet(ctx, addKeys)
|
|
if err != nil {
|
|
logutil.Logger(ctx).Warn("amend failed to batch get kv new keys", zap.Error(err))
|
|
return nil, errors.Trace(err)
|
|
}
|
|
if len(newValKvMap) != len(addKeys) {
|
|
logutil.Logger(ctx).Error("amend failed to batch get results invalid",
|
|
zap.Int("addKeys len", len(addKeys)), zap.Int("newValKvMap", len(newValKvMap)))
|
|
return nil, errors.Errorf("add keys has %v values but result kvMap has %v", len(addKeys), len(newValKvMap))
|
|
}
|
|
// BatchGet the old key values, the Op_Del and Op_Put types keys in storage using forUpdateTS, the Op_put type is for
|
|
// row update using the same row key, it may not exist.
|
|
snapshot, err := s.sess.GetStore().GetSnapshot(kv.Version{Ver: s.sess.sessionVars.TxnCtx.GetForUpdateTS()})
|
|
if err != nil {
|
|
logutil.Logger(ctx).Warn("amend failed to get snapshot using forUpdateTS", zap.Error(err))
|
|
return nil, errors.Trace(err)
|
|
}
|
|
oldValKvMap, err := snapshot.BatchGet(ctx, removeKeys)
|
|
if err != nil {
|
|
logutil.Logger(ctx).Warn("amend failed to batch get kv old keys", zap.Error(err))
|
|
return nil, errors.Trace(err)
|
|
}
|
|
|
|
res := &rowKvMap{
|
|
oldRowKvMap: oldValKvMap,
|
|
newRowKvMap: newValKvMap,
|
|
}
|
|
return res, nil
|
|
}
|
|
|
|
func (s *SchemaAmender) checkDupKeys(ctx context.Context, mutations *tikv.CommitterMutations) error {
|
|
// Check if there are duplicate key entries.
|
|
checkMap := make(map[string]pb.Op)
|
|
for i := 0; i < len(mutations.GetKeys()); i++ {
|
|
key := mutations.GetKeys()[i]
|
|
keyOp := mutations.GetOps()[i]
|
|
keyVal := mutations.GetValues()[i]
|
|
if foundOp, ok := checkMap[string(key)]; ok {
|
|
logutil.Logger(ctx).Error("duplicate key found in amend result mutations",
|
|
zap.Stringer("key", kv.Key(key)),
|
|
zap.Stringer("foundKeyOp", foundOp),
|
|
zap.Stringer("thisKeyOp", keyOp),
|
|
zap.Stringer("thisKeyValue", kv.Key(keyVal)))
|
|
return errors.Trace(errors.Errorf("duplicate key=%s is found in mutations", kv.Key(key).String()))
|
|
}
|
|
checkMap[string(key)] = keyOp
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// genAllAmendMutations generates CommitterMutations for all tables and related amend operations.
|
|
func (s *SchemaAmender) genAllAmendMutations(ctx context.Context, commitMutations tikv.CommitterMutations,
|
|
info *amendCollector) (*tikv.CommitterMutations, error) {
|
|
rowKvMap, err := s.prepareKvMap(ctx, commitMutations, info)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
// Do generate add/remove mutations processing each key.
|
|
resultNewMutations := tikv.NewCommiterMutations(32)
|
|
for _, amendOps := range info.tblAmendOpMap {
|
|
for _, curOp := range amendOps {
|
|
err := curOp.genMutations(ctx, s.sess, commitMutations, rowKvMap, &resultNewMutations)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
}
|
|
err = s.checkDupKeys(ctx, &resultNewMutations)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return &resultNewMutations, nil
|
|
}
|
|
|
|
// AmendTxn does check and generate amend mutations based on input infoSchema and mutations, mutations need to prewrite
|
|
// are returned, the input commitMutations will not be changed.
|
|
func (s *SchemaAmender) AmendTxn(ctx context.Context, startInfoSchema tikv.SchemaVer, change *tikv.RelatedSchemaChange,
|
|
commitMutations tikv.CommitterMutations) (*tikv.CommitterMutations, error) {
|
|
// Get info schema meta
|
|
infoSchemaAtStart := startInfoSchema.(infoschema.InfoSchema)
|
|
infoSchemaAtCheck := change.LatestInfoSchema.(infoschema.InfoSchema)
|
|
|
|
// Collect amend operations for each table by physical table ID.
|
|
var needAmendMem bool
|
|
amendCollector := newAmendCollector()
|
|
for i, tblID := range change.PhyTblIDS {
|
|
actionType := change.ActionTypes[i]
|
|
// Check amendable flags, return if not supported flags exist.
|
|
if actionType&(^amendableType) != 0 {
|
|
logutil.Logger(ctx).Info("amend action type not supported for txn", zap.Int64("tblID", tblID), zap.Uint64("actionType", actionType))
|
|
return nil, errors.Trace(table.ErrUnsupportedOp)
|
|
}
|
|
// Partition table is not supported now.
|
|
tblInfoAtStart, ok := infoSchemaAtStart.TableByID(tblID)
|
|
if !ok {
|
|
return nil, errors.Trace(errors.Errorf("tableID=%d is not found in infoSchema", tblID))
|
|
}
|
|
if tblInfoAtStart.Meta().Partition != nil {
|
|
logutil.Logger(ctx).Info("Amend for partition table is not supported",
|
|
zap.String("tableName", tblInfoAtStart.Meta().Name.String()), zap.Int64("tableID", tblID))
|
|
return nil, errors.Trace(table.ErrUnsupportedOp)
|
|
}
|
|
tblInfoAtCommit, ok := infoSchemaAtCheck.TableByID(tblID)
|
|
if !ok {
|
|
return nil, errors.Trace(errors.Errorf("tableID=%d is not found in infoSchema", tblID))
|
|
}
|
|
if actionType&(memBufAmendType) != 0 {
|
|
needAmendMem = true
|
|
err := amendCollector.collectTblAmendOps(s.sess, tblID, tblInfoAtStart, tblInfoAtCommit, actionType)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
}
|
|
// After amend operations collect, generate related new mutations based on input commitMutations
|
|
if needAmendMem {
|
|
return s.genAllAmendMutations(ctx, commitMutations, amendCollector)
|
|
}
|
|
return nil, nil
|
|
}
|
|
|
|
func newSchemaAndDecoder(ctx sessionctx.Context, tbl *model.TableInfo) *schemaAndDecoder {
|
|
schema := expression.NewSchema(make([]*expression.Column, 0, len(tbl.Columns))...)
|
|
for _, col := range tbl.Columns {
|
|
colExpr := &expression.Column{
|
|
RetType: &col.FieldType,
|
|
ID: col.ID,
|
|
}
|
|
if col.IsGenerated() && !col.GeneratedStored {
|
|
// This will not be used since generated column is rejected in collectIndexAmendOps.
|
|
colExpr.VirtualExpr = &expression.Constant{}
|
|
}
|
|
schema.Append(colExpr)
|
|
}
|
|
return &schemaAndDecoder{schema, executor.NewRowDecoder(ctx, schema, tbl)}
|
|
}
|
|
|