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.
189 lines
6.6 KiB
189 lines
6.6 KiB
// Copyright 2017 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 util
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/hex"
|
|
"fmt"
|
|
"strconv"
|
|
|
|
"github.com/pingcap/errors"
|
|
"github.com/pingcap/parser/terror"
|
|
"github.com/pingcap/tidb/kv"
|
|
"github.com/pingcap/tidb/sessionctx"
|
|
"github.com/pingcap/tidb/sessionctx/variable"
|
|
"github.com/pingcap/tidb/util/chunk"
|
|
"github.com/pingcap/tidb/util/sqlexec"
|
|
)
|
|
|
|
const (
|
|
deleteRangesTable = `gc_delete_range`
|
|
doneDeleteRangesTable = `gc_delete_range_done`
|
|
loadDeleteRangeSQL = `SELECT HIGH_PRIORITY job_id, element_id, start_key, end_key FROM mysql.%s WHERE ts < %v`
|
|
recordDoneDeletedRangeSQL = `INSERT IGNORE INTO mysql.gc_delete_range_done SELECT * FROM mysql.gc_delete_range WHERE job_id = %d AND element_id = %d`
|
|
completeDeleteRangeSQL = `DELETE FROM mysql.gc_delete_range WHERE job_id = %d AND element_id = %d`
|
|
completeDeleteMultiRangesSQL = `DELETE FROM mysql.gc_delete_range WHERE job_id = %d AND element_id in (%v)`
|
|
updateDeleteRangeSQL = `UPDATE mysql.gc_delete_range SET start_key = "%s" WHERE job_id = %d AND element_id = %d AND start_key = "%s"`
|
|
deleteDoneRecordSQL = `DELETE FROM mysql.gc_delete_range_done WHERE job_id = %d AND element_id = %d`
|
|
)
|
|
|
|
// DelRangeTask is for run delete-range command in gc_worker.
|
|
type DelRangeTask struct {
|
|
JobID, ElementID int64
|
|
StartKey, EndKey kv.Key
|
|
}
|
|
|
|
// Range returns the range [start, end) to delete.
|
|
func (t DelRangeTask) Range() (kv.Key, kv.Key) {
|
|
return t.StartKey, t.EndKey
|
|
}
|
|
|
|
// LoadDeleteRanges loads delete range tasks from gc_delete_range table.
|
|
func LoadDeleteRanges(ctx sessionctx.Context, safePoint uint64) (ranges []DelRangeTask, _ error) {
|
|
return loadDeleteRangesFromTable(ctx, deleteRangesTable, safePoint)
|
|
}
|
|
|
|
// LoadDoneDeleteRanges loads deleted ranges from gc_delete_range_done table.
|
|
func LoadDoneDeleteRanges(ctx sessionctx.Context, safePoint uint64) (ranges []DelRangeTask, _ error) {
|
|
return loadDeleteRangesFromTable(ctx, doneDeleteRangesTable, safePoint)
|
|
}
|
|
|
|
func loadDeleteRangesFromTable(ctx sessionctx.Context, table string, safePoint uint64) (ranges []DelRangeTask, _ error) {
|
|
sql := fmt.Sprintf(loadDeleteRangeSQL, table, safePoint)
|
|
rss, err := ctx.(sqlexec.SQLExecutor).Execute(context.TODO(), sql)
|
|
if len(rss) > 0 {
|
|
defer terror.Call(rss[0].Close)
|
|
}
|
|
if err != nil {
|
|
return nil, errors.Trace(err)
|
|
}
|
|
|
|
rs := rss[0]
|
|
req := rs.NewChunk()
|
|
it := chunk.NewIterator4Chunk(req)
|
|
for {
|
|
err = rs.Next(context.TODO(), req)
|
|
if err != nil {
|
|
return nil, errors.Trace(err)
|
|
}
|
|
if req.NumRows() == 0 {
|
|
break
|
|
}
|
|
|
|
for row := it.Begin(); row != it.End(); row = it.Next() {
|
|
startKey, err := hex.DecodeString(row.GetString(2))
|
|
if err != nil {
|
|
return nil, errors.Trace(err)
|
|
}
|
|
endKey, err := hex.DecodeString(row.GetString(3))
|
|
if err != nil {
|
|
return nil, errors.Trace(err)
|
|
}
|
|
ranges = append(ranges, DelRangeTask{
|
|
JobID: row.GetInt64(0),
|
|
ElementID: row.GetInt64(1),
|
|
StartKey: startKey,
|
|
EndKey: endKey,
|
|
})
|
|
}
|
|
}
|
|
return ranges, nil
|
|
}
|
|
|
|
// CompleteDeleteRange moves a record from gc_delete_range table to gc_delete_range_done table.
|
|
// NOTE: This function WILL NOT start and run in a new transaction internally.
|
|
func CompleteDeleteRange(ctx sessionctx.Context, dr DelRangeTask) error {
|
|
sql := fmt.Sprintf(recordDoneDeletedRangeSQL, dr.JobID, dr.ElementID)
|
|
_, err := ctx.(sqlexec.SQLExecutor).Execute(context.TODO(), sql)
|
|
if err != nil {
|
|
return errors.Trace(err)
|
|
}
|
|
|
|
return RemoveFromGCDeleteRange(ctx, dr.JobID, dr.ElementID)
|
|
}
|
|
|
|
// RemoveFromGCDeleteRange is exported for ddl pkg to use.
|
|
func RemoveFromGCDeleteRange(ctx sessionctx.Context, jobID, elementID int64) error {
|
|
sql := fmt.Sprintf(completeDeleteRangeSQL, jobID, elementID)
|
|
_, err := ctx.(sqlexec.SQLExecutor).Execute(context.TODO(), sql)
|
|
return errors.Trace(err)
|
|
}
|
|
|
|
// RemoveMultiFromGCDeleteRange is exported for ddl pkg to use.
|
|
func RemoveMultiFromGCDeleteRange(ctx sessionctx.Context, jobID int64, elementIDs []int64) error {
|
|
var buf bytes.Buffer
|
|
for i, elementID := range elementIDs {
|
|
if i > 0 {
|
|
buf.WriteString(", ")
|
|
}
|
|
buf.WriteString(strconv.FormatInt(elementID, 10))
|
|
}
|
|
sql := fmt.Sprintf(completeDeleteMultiRangesSQL, jobID, buf.String())
|
|
_, err := ctx.(sqlexec.SQLExecutor).Execute(context.TODO(), sql)
|
|
return errors.Trace(err)
|
|
}
|
|
|
|
// DeleteDoneRecord removes a record from gc_delete_range_done table.
|
|
func DeleteDoneRecord(ctx sessionctx.Context, dr DelRangeTask) error {
|
|
sql := fmt.Sprintf(deleteDoneRecordSQL, dr.JobID, dr.ElementID)
|
|
_, err := ctx.(sqlexec.SQLExecutor).Execute(context.TODO(), sql)
|
|
return errors.Trace(err)
|
|
}
|
|
|
|
// UpdateDeleteRange is only for emulator.
|
|
func UpdateDeleteRange(ctx sessionctx.Context, dr DelRangeTask, newStartKey, oldStartKey kv.Key) error {
|
|
newStartKeyHex := hex.EncodeToString(newStartKey)
|
|
oldStartKeyHex := hex.EncodeToString(oldStartKey)
|
|
sql := fmt.Sprintf(updateDeleteRangeSQL, newStartKeyHex, dr.JobID, dr.ElementID, oldStartKeyHex)
|
|
_, err := ctx.(sqlexec.SQLExecutor).Execute(context.TODO(), sql)
|
|
return errors.Trace(err)
|
|
}
|
|
|
|
// LoadDDLReorgVars loads ddl reorg variable from mysql.global_variables.
|
|
func LoadDDLReorgVars(ctx sessionctx.Context) error {
|
|
return LoadGlobalVars(ctx, []string{variable.TiDBDDLReorgWorkerCount, variable.TiDBDDLReorgBatchSize})
|
|
}
|
|
|
|
// LoadDDLVars loads ddl variable from mysql.global_variables.
|
|
func LoadDDLVars(ctx sessionctx.Context) error {
|
|
return LoadGlobalVars(ctx, []string{variable.TiDBDDLErrorCountLimit})
|
|
}
|
|
|
|
const loadGlobalVarsSQL = "select HIGH_PRIORITY variable_name, variable_value from mysql.global_variables where variable_name in (%s)"
|
|
|
|
// LoadGlobalVars loads global variable from mysql.global_variables.
|
|
func LoadGlobalVars(ctx sessionctx.Context, varNames []string) error {
|
|
if sctx, ok := ctx.(sqlexec.RestrictedSQLExecutor); ok {
|
|
nameList := ""
|
|
for i, name := range varNames {
|
|
if i > 0 {
|
|
nameList += ", "
|
|
}
|
|
nameList += fmt.Sprintf("'%s'", name)
|
|
}
|
|
sql := fmt.Sprintf(loadGlobalVarsSQL, nameList)
|
|
rows, _, err := sctx.ExecRestrictedSQL(sql)
|
|
if err != nil {
|
|
return errors.Trace(err)
|
|
}
|
|
for _, row := range rows {
|
|
varName := row.GetString(0)
|
|
varValue := row.GetString(1)
|
|
variable.SetLocalSystemVar(varName, varValue)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|