431 lines
12 KiB
Go
431 lines
12 KiB
Go
|
// Copyright 2015 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 ddl
|
||
|
|
|
||
|
|
import (
|
||
|
|
"github.com/juju/errors"
|
||
|
|
"github.com/ngaut/log"
|
||
|
|
"github.com/pingcap/tidb/ast"
|
||
|
|
"github.com/pingcap/tidb/column"
|
||
|
|
"github.com/pingcap/tidb/kv"
|
||
|
|
"github.com/pingcap/tidb/meta"
|
||
|
|
"github.com/pingcap/tidb/model"
|
||
|
|
"github.com/pingcap/tidb/table"
|
||
|
|
"github.com/pingcap/tidb/table/tables"
|
||
|
|
"github.com/pingcap/tidb/terror"
|
||
|
|
)
|
||
|
|
|
||
|
|
func (d *ddl) adjustColumnOffset(columns []*model.ColumnInfo, indices []*model.IndexInfo, offset int, added bool) {
|
||
|
|
offsetChanged := make(map[int]int)
|
||
|
|
if added {
|
||
|
|
for i := offset + 1; i < len(columns); i++ {
|
||
|
|
offsetChanged[columns[i].Offset] = i
|
||
|
|
columns[i].Offset = i
|
||
|
|
}
|
||
|
|
columns[offset].Offset = offset
|
||
|
|
} else {
|
||
|
|
for i := offset + 1; i < len(columns); i++ {
|
||
|
|
offsetChanged[columns[i].Offset] = i - 1
|
||
|
|
columns[i].Offset = i - 1
|
||
|
|
}
|
||
|
|
columns[offset].Offset = len(columns) - 1
|
||
|
|
}
|
||
|
|
|
||
|
|
// TODO: index can't cover the add/remove column with offset now, we may check this later.
|
||
|
|
|
||
|
|
// Update index column offset info.
|
||
|
|
for _, idx := range indices {
|
||
|
|
for _, col := range idx.Columns {
|
||
|
|
newOffset, ok := offsetChanged[col.Offset]
|
||
|
|
if ok {
|
||
|
|
col.Offset = newOffset
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
func (d *ddl) addColumn(tblInfo *model.TableInfo, colInfo *model.ColumnInfo, pos *ast.ColumnPosition) (*model.ColumnInfo, int, error) {
|
||
|
|
// Check column name duplicate.
|
||
|
|
cols := tblInfo.Columns
|
||
|
|
position := len(cols)
|
||
|
|
|
||
|
|
// Get column position.
|
||
|
|
if pos.Tp == ast.ColumnPositionFirst {
|
||
|
|
position = 0
|
||
|
|
} else if pos.Tp == ast.ColumnPositionAfter {
|
||
|
|
c := findCol(cols, pos.RelativeColumn.Name.L)
|
||
|
|
if c == nil {
|
||
|
|
return nil, 0, errors.Errorf("No such column: %v", pos.RelativeColumn)
|
||
|
|
}
|
||
|
|
|
||
|
|
// Insert position is after the mentioned column.
|
||
|
|
position = c.Offset + 1
|
||
|
|
}
|
||
|
|
|
||
|
|
colInfo.State = model.StateNone
|
||
|
|
// To support add column asynchronous, we should mark its offset as the last column.
|
||
|
|
// So that we can use origin column offset to get value from row.
|
||
|
|
colInfo.Offset = len(cols)
|
||
|
|
|
||
|
|
// Insert col into the right place of the column list.
|
||
|
|
newCols := make([]*model.ColumnInfo, 0, len(cols)+1)
|
||
|
|
newCols = append(newCols, cols[:position]...)
|
||
|
|
newCols = append(newCols, colInfo)
|
||
|
|
newCols = append(newCols, cols[position:]...)
|
||
|
|
|
||
|
|
tblInfo.Columns = newCols
|
||
|
|
return colInfo, position, nil
|
||
|
|
}
|
||
|
|
|
||
|
|
func (d *ddl) onAddColumn(t *meta.Meta, job *model.Job) error {
|
||
|
|
schemaID := job.SchemaID
|
||
|
|
tblInfo, err := d.getTableInfo(t, job)
|
||
|
|
if err != nil {
|
||
|
|
return errors.Trace(err)
|
||
|
|
}
|
||
|
|
|
||
|
|
col := &model.ColumnInfo{}
|
||
|
|
pos := &ast.ColumnPosition{}
|
||
|
|
offset := 0
|
||
|
|
err = job.DecodeArgs(col, pos, &offset)
|
||
|
|
if err != nil {
|
||
|
|
job.State = model.JobCancelled
|
||
|
|
return errors.Trace(err)
|
||
|
|
}
|
||
|
|
|
||
|
|
columnInfo := findCol(tblInfo.Columns, col.Name.L)
|
||
|
|
if columnInfo != nil {
|
||
|
|
if columnInfo.State == model.StatePublic {
|
||
|
|
// we already have a column with same column name
|
||
|
|
job.State = model.JobCancelled
|
||
|
|
return errors.Errorf("ADD COLUMN: column already exist %s", col.Name.L)
|
||
|
|
}
|
||
|
|
} else {
|
||
|
|
columnInfo, offset, err = d.addColumn(tblInfo, col, pos)
|
||
|
|
if err != nil {
|
||
|
|
job.State = model.JobCancelled
|
||
|
|
return errors.Trace(err)
|
||
|
|
}
|
||
|
|
|
||
|
|
// Set offset arg to job.
|
||
|
|
if offset != 0 {
|
||
|
|
job.Args = []interface{}{columnInfo, pos, offset}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
_, err = t.GenSchemaVersion()
|
||
|
|
if err != nil {
|
||
|
|
return errors.Trace(err)
|
||
|
|
}
|
||
|
|
|
||
|
|
switch columnInfo.State {
|
||
|
|
case model.StateNone:
|
||
|
|
// none -> delete only
|
||
|
|
job.SchemaState = model.StateDeleteOnly
|
||
|
|
columnInfo.State = model.StateDeleteOnly
|
||
|
|
err = t.UpdateTable(schemaID, tblInfo)
|
||
|
|
return errors.Trace(err)
|
||
|
|
case model.StateDeleteOnly:
|
||
|
|
// delete only -> write only
|
||
|
|
job.SchemaState = model.StateWriteOnly
|
||
|
|
columnInfo.State = model.StateWriteOnly
|
||
|
|
err = t.UpdateTable(schemaID, tblInfo)
|
||
|
|
return errors.Trace(err)
|
||
|
|
case model.StateWriteOnly:
|
||
|
|
// write only -> reorganization
|
||
|
|
job.SchemaState = model.StateWriteReorganization
|
||
|
|
columnInfo.State = model.StateWriteReorganization
|
||
|
|
// initialize SnapshotVer to 0 for later reorganization check.
|
||
|
|
job.SnapshotVer = 0
|
||
|
|
err = t.UpdateTable(schemaID, tblInfo)
|
||
|
|
return errors.Trace(err)
|
||
|
|
case model.StateWriteReorganization:
|
||
|
|
// reorganization -> public
|
||
|
|
// get the current version for reorganization if we don't have
|
||
|
|
reorgInfo, err := d.getReorgInfo(t, job)
|
||
|
|
if err != nil || reorgInfo.first {
|
||
|
|
// if we run reorg firstly, we should update the job snapshot version
|
||
|
|
// and then run the reorg next time.
|
||
|
|
return errors.Trace(err)
|
||
|
|
}
|
||
|
|
|
||
|
|
tbl, err := d.getTable(schemaID, tblInfo)
|
||
|
|
if err != nil {
|
||
|
|
return errors.Trace(err)
|
||
|
|
}
|
||
|
|
|
||
|
|
err = d.runReorgJob(func() error {
|
||
|
|
return d.backfillColumn(tbl, columnInfo, reorgInfo)
|
||
|
|
})
|
||
|
|
|
||
|
|
if terror.ErrorEqual(err, errWaitReorgTimeout) {
|
||
|
|
// if timeout, we should return, check for the owner and re-wait job done.
|
||
|
|
return nil
|
||
|
|
}
|
||
|
|
if err != nil {
|
||
|
|
return errors.Trace(err)
|
||
|
|
}
|
||
|
|
|
||
|
|
// Adjust column offset.
|
||
|
|
d.adjustColumnOffset(tblInfo.Columns, tblInfo.Indices, offset, true)
|
||
|
|
|
||
|
|
columnInfo.State = model.StatePublic
|
||
|
|
|
||
|
|
if err = t.UpdateTable(schemaID, tblInfo); err != nil {
|
||
|
|
return errors.Trace(err)
|
||
|
|
}
|
||
|
|
|
||
|
|
// finish this job
|
||
|
|
job.SchemaState = model.StatePublic
|
||
|
|
job.State = model.JobDone
|
||
|
|
return nil
|
||
|
|
default:
|
||
|
|
return errors.Errorf("invalid column state %v", columnInfo.State)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
func (d *ddl) onDropColumn(t *meta.Meta, job *model.Job) error {
|
||
|
|
schemaID := job.SchemaID
|
||
|
|
tblInfo, err := d.getTableInfo(t, job)
|
||
|
|
if err != nil {
|
||
|
|
return errors.Trace(err)
|
||
|
|
}
|
||
|
|
|
||
|
|
var colName model.CIStr
|
||
|
|
err = job.DecodeArgs(&colName)
|
||
|
|
if err != nil {
|
||
|
|
job.State = model.JobCancelled
|
||
|
|
return errors.Trace(err)
|
||
|
|
}
|
||
|
|
|
||
|
|
colInfo := findCol(tblInfo.Columns, colName.L)
|
||
|
|
if colInfo == nil {
|
||
|
|
job.State = model.JobCancelled
|
||
|
|
return errors.Errorf("column %s doesn't exist", colName)
|
||
|
|
}
|
||
|
|
|
||
|
|
if len(tblInfo.Columns) == 1 {
|
||
|
|
job.State = model.JobCancelled
|
||
|
|
return errors.Errorf("can't drop only column %s in table %s", colName, tblInfo.Name)
|
||
|
|
}
|
||
|
|
|
||
|
|
// we don't support drop column with index covered now.
|
||
|
|
// we must drop the index first, then drop the column.
|
||
|
|
for _, indexInfo := range tblInfo.Indices {
|
||
|
|
for _, col := range indexInfo.Columns {
|
||
|
|
if col.Name.L == colName.L {
|
||
|
|
job.State = model.JobCancelled
|
||
|
|
return errors.Errorf("can't drop column %s with index %s covered now", colName, indexInfo.Name)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
_, err = t.GenSchemaVersion()
|
||
|
|
if err != nil {
|
||
|
|
return errors.Trace(err)
|
||
|
|
}
|
||
|
|
|
||
|
|
switch colInfo.State {
|
||
|
|
case model.StatePublic:
|
||
|
|
// public -> write only
|
||
|
|
job.SchemaState = model.StateWriteOnly
|
||
|
|
colInfo.State = model.StateWriteOnly
|
||
|
|
|
||
|
|
// set this column's offset to the last and reset all following columns' offset
|
||
|
|
d.adjustColumnOffset(tblInfo.Columns, tblInfo.Indices, colInfo.Offset, false)
|
||
|
|
|
||
|
|
err = t.UpdateTable(schemaID, tblInfo)
|
||
|
|
return errors.Trace(err)
|
||
|
|
case model.StateWriteOnly:
|
||
|
|
// write only -> delete only
|
||
|
|
job.SchemaState = model.StateDeleteOnly
|
||
|
|
colInfo.State = model.StateDeleteOnly
|
||
|
|
err = t.UpdateTable(schemaID, tblInfo)
|
||
|
|
return errors.Trace(err)
|
||
|
|
case model.StateDeleteOnly:
|
||
|
|
// delete only -> reorganization
|
||
|
|
job.SchemaState = model.StateDeleteReorganization
|
||
|
|
colInfo.State = model.StateDeleteReorganization
|
||
|
|
// initialize SnapshotVer to 0 for later reorganization check.
|
||
|
|
job.SnapshotVer = 0
|
||
|
|
err = t.UpdateTable(schemaID, tblInfo)
|
||
|
|
return errors.Trace(err)
|
||
|
|
case model.StateDeleteReorganization:
|
||
|
|
// reorganization -> absent
|
||
|
|
reorgInfo, err := d.getReorgInfo(t, job)
|
||
|
|
if err != nil || reorgInfo.first {
|
||
|
|
// if we run reorg firstly, we should update the job snapshot version
|
||
|
|
// and then run the reorg next time.
|
||
|
|
return errors.Trace(err)
|
||
|
|
}
|
||
|
|
|
||
|
|
tbl, err := d.getTable(schemaID, tblInfo)
|
||
|
|
if err != nil {
|
||
|
|
return errors.Trace(err)
|
||
|
|
}
|
||
|
|
|
||
|
|
err = d.runReorgJob(func() error {
|
||
|
|
return d.dropTableColumn(tbl, colInfo, reorgInfo)
|
||
|
|
})
|
||
|
|
|
||
|
|
if terror.ErrorEqual(err, errWaitReorgTimeout) {
|
||
|
|
// if timeout, we should return, check for the owner and re-wait job done.
|
||
|
|
return nil
|
||
|
|
}
|
||
|
|
if err != nil {
|
||
|
|
return errors.Trace(err)
|
||
|
|
}
|
||
|
|
|
||
|
|
// all reorganization jobs done, drop this column
|
||
|
|
newColumns := make([]*model.ColumnInfo, 0, len(tblInfo.Columns))
|
||
|
|
for _, col := range tblInfo.Columns {
|
||
|
|
if col.Name.L != colName.L {
|
||
|
|
newColumns = append(newColumns, col)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
tblInfo.Columns = newColumns
|
||
|
|
if err = t.UpdateTable(schemaID, tblInfo); err != nil {
|
||
|
|
return errors.Trace(err)
|
||
|
|
}
|
||
|
|
|
||
|
|
// finish this job
|
||
|
|
job.SchemaState = model.StateNone
|
||
|
|
job.State = model.JobDone
|
||
|
|
return nil
|
||
|
|
default:
|
||
|
|
return errors.Errorf("invalid table state %v", tblInfo.State)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
// How to backfill column data in reorganization state?
|
||
|
|
// 1. Generate a snapshot with special version.
|
||
|
|
// 2. Traverse the snapshot, get every row in the table.
|
||
|
|
// 3. For one row, if the row has been already deleted, skip to next row.
|
||
|
|
// 4. If not deleted, check whether column data has existed, if existed, skip to next row.
|
||
|
|
// 5. If column data doesn't exist, backfill the column with default value and then continue to handle next row.
|
||
|
|
func (d *ddl) backfillColumn(t table.Table, columnInfo *model.ColumnInfo, reorgInfo *reorgInfo) error {
|
||
|
|
seekHandle := reorgInfo.Handle
|
||
|
|
version := reorgInfo.SnapshotVer
|
||
|
|
|
||
|
|
for {
|
||
|
|
handles, err := d.getSnapshotRows(t, version, seekHandle)
|
||
|
|
if err != nil {
|
||
|
|
return errors.Trace(err)
|
||
|
|
} else if len(handles) == 0 {
|
||
|
|
return nil
|
||
|
|
}
|
||
|
|
|
||
|
|
seekHandle = handles[len(handles)-1] + 1
|
||
|
|
err = d.backfillColumnData(t, columnInfo, handles, reorgInfo)
|
||
|
|
if err != nil {
|
||
|
|
return errors.Trace(err)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
func (d *ddl) backfillColumnData(t table.Table, columnInfo *model.ColumnInfo, handles []int64, reorgInfo *reorgInfo) error {
|
||
|
|
for _, handle := range handles {
|
||
|
|
log.Info("[ddl] backfill column...", handle)
|
||
|
|
|
||
|
|
err := kv.RunInNewTxn(d.store, true, func(txn kv.Transaction) error {
|
||
|
|
if err := d.isReorgRunnable(txn); err != nil {
|
||
|
|
return errors.Trace(err)
|
||
|
|
}
|
||
|
|
|
||
|
|
// First check if row exists.
|
||
|
|
exist, err := checkRowExist(txn, t, handle)
|
||
|
|
if err != nil {
|
||
|
|
return errors.Trace(err)
|
||
|
|
} else if !exist {
|
||
|
|
// If row doesn't exist, skip it.
|
||
|
|
return nil
|
||
|
|
}
|
||
|
|
|
||
|
|
backfillKey := t.RecordKey(handle, &column.Col{ColumnInfo: *columnInfo})
|
||
|
|
backfillValue, err := txn.Get(backfillKey)
|
||
|
|
if err != nil && !kv.IsErrNotFound(err) {
|
||
|
|
return errors.Trace(err)
|
||
|
|
}
|
||
|
|
if backfillValue != nil {
|
||
|
|
return nil
|
||
|
|
}
|
||
|
|
|
||
|
|
value, _, err := table.GetColDefaultValue(nil, columnInfo)
|
||
|
|
if err != nil {
|
||
|
|
return errors.Trace(err)
|
||
|
|
}
|
||
|
|
|
||
|
|
// must convert to the column field type.
|
||
|
|
v, err := value.ConvertTo(&columnInfo.FieldType)
|
||
|
|
if err != nil {
|
||
|
|
return errors.Trace(err)
|
||
|
|
}
|
||
|
|
|
||
|
|
err = lockRow(txn, t, handle)
|
||
|
|
if err != nil {
|
||
|
|
return errors.Trace(err)
|
||
|
|
}
|
||
|
|
|
||
|
|
err = tables.SetColValue(txn, backfillKey, v)
|
||
|
|
if err != nil {
|
||
|
|
return errors.Trace(err)
|
||
|
|
}
|
||
|
|
|
||
|
|
return errors.Trace(reorgInfo.UpdateHandle(txn, handle))
|
||
|
|
})
|
||
|
|
|
||
|
|
if err != nil {
|
||
|
|
return errors.Trace(err)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
return nil
|
||
|
|
}
|
||
|
|
|
||
|
|
func (d *ddl) dropTableColumn(t table.Table, colInfo *model.ColumnInfo, reorgInfo *reorgInfo) error {
|
||
|
|
version := reorgInfo.SnapshotVer
|
||
|
|
seekHandle := reorgInfo.Handle
|
||
|
|
|
||
|
|
col := &column.Col{ColumnInfo: *colInfo}
|
||
|
|
for {
|
||
|
|
handles, err := d.getSnapshotRows(t, version, seekHandle)
|
||
|
|
if err != nil {
|
||
|
|
return errors.Trace(err)
|
||
|
|
} else if len(handles) == 0 {
|
||
|
|
return nil
|
||
|
|
}
|
||
|
|
|
||
|
|
seekHandle = handles[len(handles)-1] + 1
|
||
|
|
|
||
|
|
err = kv.RunInNewTxn(d.store, true, func(txn kv.Transaction) error {
|
||
|
|
if err1 := d.isReorgRunnable(txn); err1 != nil {
|
||
|
|
return errors.Trace(err1)
|
||
|
|
}
|
||
|
|
|
||
|
|
var h int64
|
||
|
|
for _, h = range handles {
|
||
|
|
key := t.RecordKey(h, col)
|
||
|
|
err1 := txn.Delete(key)
|
||
|
|
if err1 != nil && !terror.ErrorEqual(err1, kv.ErrNotExist) {
|
||
|
|
return errors.Trace(err1)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
return errors.Trace(reorgInfo.UpdateHandle(txn, h))
|
||
|
|
})
|
||
|
|
if err != nil {
|
||
|
|
return errors.Trace(err)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|