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

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