Skip to content

Commit

Permalink
ddl: migrate part of ddl package code from Execute/ExecRestricted to …
Browse files Browse the repository at this point in the history 
…safe API (1) (#22670) (#22930)
  • Loading branch information
@ti-srebot
ti-srebot authored Mar 3, 2021
1 parent 29251c8 commit 6f96dde
Show file tree
Hide file tree
Showing 2 changed files with 31 additions and 33 deletions.
12 changes: 6 additions & 6 deletions ddl/column.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,6 @@
package ddl

import (
"fmt"
"strings"
"sync/atomic"
"time"
Expand Down Expand Up @@ -480,16 +479,17 @@ func (w *worker) doModifyColumn(t *meta.Meta, job *model.Job, newCol *model.Colu
// checkForNullValue ensure there are no null values of the column of this table.
// `isDataTruncated` indicates whether the new field and the old field type are the same, in order to be compatible with mysql.
func checkForNullValue(ctx sessionctx.Context, isDataTruncated bool, schema, table, newCol model.CIStr, oldCols ...*model.ColumnInfo) error {
colsStr := ""
var buf strings.Builder
sqlexec.MustFormatSQL(&buf, "select 1 from %n.%n where ", schema.L, table.L)
for i, col := range oldCols {
if i == 0 {
colsStr += "`" + col.Name.L + "` is null"
sqlexec.MustFormatSQL(&buf, "%n is null", col.Name.L)
} else {
colsStr += " or `" + col.Name.L + "` is null"
sqlexec.MustFormatSQL(&buf, " or %n is null", col.Name.L)
}
}
sql := fmt.Sprintf("select 1 from `%s`.`%s` where %s limit 1;", schema.L, table.L, colsStr)
rows, _, err := ctx.(sqlexec.RestrictedSQLExecutor).ExecRestrictedSQL(ctx, sql)
sqlexec.MustFormatSQL(&buf, " limit 1")
rows, _, err := ctx.(sqlexec.RestrictedSQLExecutor).ExecRestrictedSQL(ctx, buf.String())
if err != nil {
return errors.Trace(err)
}
Expand Down
52 changes: 25 additions & 27 deletions ddl/util/util.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@ package util
import (
"context"
"encoding/hex"
"fmt"
"strings"

"github.com/pingcap/errors"
"github.com/pingcap/parser/terror"
Expand All @@ -30,11 +30,12 @@ import (
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`
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`
loadDeleteRangeSQL = `SELECT HIGH_PRIORITY job_id, element_id, start_key, end_key FROM mysql.%n WHERE ts < %?`
recordDoneDeletedRangeSQL = `INSERT IGNORE INTO mysql.gc_delete_range_done SELECT * FROM mysql.gc_delete_range WHERE job_id = %? AND element_id = %?`
completeDeleteRangeSQL = `DELETE FROM mysql.gc_delete_range WHERE job_id = %? AND element_id = %?`
updateDeleteRangeSQL = `UPDATE mysql.gc_delete_range SET start_key = %? WHERE job_id = %? AND element_id = %? AND start_key = %?`
deleteDoneRecordSQL = `DELETE FROM mysql.gc_delete_range_done WHERE job_id = %? AND element_id = %?`
loadGlobalVars = `SELECT HIGH_PRIORITY variable_name, variable_value from mysql.global_variables where variable_name in (%?)`
)

// DelRangeTask is for run delete-range command in gc_worker.
Expand All @@ -59,8 +60,9 @@ func LoadDoneDeleteRanges(ctx sessionctx.Context, safePoint uint64) (ranges []De
}

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)
var buf strings.Builder
sqlexec.MustFormatSQL(&buf, loadDeleteRangeSQL, table, safePoint)
rss, err := ctx.(sqlexec.SQLExecutor).Execute(context.TODO(), buf.String())
if len(rss) > 0 {
defer terror.Call(rss[0].Close)
}
Expand Down Expand Up @@ -103,8 +105,9 @@ func loadDeleteRangesFromTable(ctx sessionctx.Context, table string, safePoint u
// 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)
var buf strings.Builder
sqlexec.MustFormatSQL(&buf, recordDoneDeletedRangeSQL, dr.JobID, dr.ElementID)
_, err := ctx.(sqlexec.SQLExecutor).Execute(context.TODO(), buf.String())
if err != nil {
return errors.Trace(err)
}
Expand All @@ -114,24 +117,27 @@ func CompleteDeleteRange(ctx sessionctx.Context, dr DelRangeTask) error {

// 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)
var buf strings.Builder
sqlexec.MustFormatSQL(&buf, completeDeleteRangeSQL, jobID, elementID)
_, err := ctx.(sqlexec.SQLExecutor).Execute(context.TODO(), buf.String())
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)
var buf strings.Builder
sqlexec.MustFormatSQL(&buf, deleteDoneRecordSQL, dr.JobID, dr.ElementID)
_, err := ctx.(sqlexec.SQLExecutor).Execute(context.TODO(), buf.String())
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)
var buf strings.Builder
sqlexec.MustFormatSQL(&buf, updateDeleteRangeSQL, newStartKeyHex, dr.JobID, dr.ElementID, oldStartKeyHex)
_, err := ctx.(sqlexec.SQLExecutor).Execute(context.TODO(), buf.String())
return errors.Trace(err)
}

Expand All @@ -145,20 +151,12 @@ 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(ctx, sql)
var buf strings.Builder
sqlexec.MustFormatSQL(&buf, loadGlobalVars, varNames)
rows, _, err := sctx.ExecRestrictedSQL(ctx, buf.String())
if err != nil {
return errors.Trace(err)
}
Expand Down

0 comments on commit 6f96dde

Please sign in to comment.