Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

ddl: migrate part of ddl package code from Execute/ExecRestricted to safe API (1) #22670

Merged
merged 17 commits into from
Feb 5, 2021
Merged
Show file tree
Hide file tree
Changes from 11 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 22 additions & 5 deletions ddl/column.go
Original file line number Diff line number Diff line change
Expand Up @@ -939,7 +939,12 @@ func (w *worker) doModifyColumnTypeWithData(
}
defer w.sessPool.put(ctx)

_, _, err = ctx.(sqlexec.RestrictedSQLExecutor).ExecRestrictedSQL(valStr)
stmt, err := ctx.(sqlexec.RestrictedSQLExecutor).ParseWithParams(context.Background(), valStr)
if err != nil {
job.State = model.JobStateCancelled
failpoint.Return(ver, err)
}
_, _, err = ctx.(sqlexec.RestrictedSQLExecutor).ExecRestrictedStmt(context.Background(), stmt)
if err != nil {
job.State = model.JobStateCancelled
failpoint.Return(ver, err)
Expand Down Expand Up @@ -1526,15 +1531,27 @@ func checkAndApplyNewAutoRandomBits(job *model.Job, t *meta.Meta, tblInfo *model
// `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 := ""
paramsList := make([]interface{}, 0, 2+len(oldCols))
paramsList = append(paramsList, schema.L, table.L)
for i, col := range oldCols {
if i == 0 {
colsStr += "`" + col.Name.L + "` is null"
colsStr += "%n is null"
paramsList = append(paramsList, col.Name.L)
} else {
colsStr += " or `" + col.Name.L + "` is null"
colsStr += " or %n is null"
paramsList = append(paramsList, 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(sql)
var buf strings.Builder
_, err := fmt.Fprintf(&buf, "select 1 from %%n.%%n where %s limit 1", colsStr)
AilinKid marked this conversation as resolved.
Show resolved Hide resolved
if err != nil {
return errors.Trace(err)
}
stmt, err := ctx.(sqlexec.RestrictedSQLExecutor).ParseWithParams(context.Background(), buf.String(), paramsList...)
if err != nil {
return errors.Trace(err)
}
rows, _, err := ctx.(sqlexec.RestrictedSQLExecutor).ExecRestrictedStmt(context.Background(), stmt)
if err != nil {
return errors.Trace(err)
}
Expand Down
4 changes: 2 additions & 2 deletions ddl/testutil/testutil.go
Original file line number Diff line number Diff line change
Expand Up @@ -40,13 +40,13 @@ func ExecMultiSQLInGoroutine(c *check.C, s kv.Storage, dbName string, multiSQL [
return
}
defer se.Close()
_, err = se.Execute(context.Background(), "use "+dbName)
_, err = se.ExecuteInternal(context.Background(), "use %n", dbName)
if err != nil {
done <- errors.Trace(err)
return
}
for _, sql := range multiSQL {
rs, err := se.Execute(context.Background(), sql)
AilinKid marked this conversation as resolved.
Show resolved Hide resolved
rs, err := se.ExecuteInternal(context.Background(), sql)
if err != nil {
done <- errors.Trace(err)
return
Expand Down
72 changes: 42 additions & 30 deletions ddl/util/util.go
Original file line number Diff line number Diff line change
Expand Up @@ -14,11 +14,10 @@
package util

import (
"bytes"
"context"
"encoding/hex"
"fmt"
"strconv"
"strings"

"github.com/pingcap/errors"
"github.com/pingcap/parser/terror"
Expand All @@ -32,12 +31,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`
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`
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 = %?`
completeDeleteMultiRangesSQL = `DELETE FROM mysql.gc_delete_range WHERE job_id = %%? AND element_id in (%s)`
AilinKid marked this conversation as resolved.
Show resolved Hide resolved
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 = %?`
)

// DelRangeTask is for run delete-range command in gc_worker.
Expand All @@ -62,16 +61,14 @@ 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)
if len(rss) > 0 {
defer terror.Call(rss[0].Close)
rs, err := ctx.(sqlexec.SQLExecutor).ExecuteInternal(context.TODO(), loadDeleteRangeSQL, table, safePoint)
if rs != nil {
defer terror.Call(rs.Close)
}
if err != nil {
return nil, errors.Trace(err)
}

rs := rss[0]
req := rs.NewChunk()
it := chunk.NewIterator4Chunk(req)
for {
Expand Down Expand Up @@ -106,8 +103,7 @@ 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)
_, err := ctx.(sqlexec.SQLExecutor).ExecuteInternal(context.TODO(), recordDoneDeletedRangeSQL, dr.JobID, dr.ElementID)
if err != nil {
return errors.Trace(err)
}
Expand All @@ -117,38 +113,42 @@ 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)
_, err := ctx.(sqlexec.SQLExecutor).ExecuteInternal(context.TODO(), completeDeleteRangeSQL, jobID, elementID)
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
idList := ""
paramIDs := make([]interface{}, 0, 1+len(elementIDs))
paramIDs = append(paramIDs, jobID)
for i, elementID := range elementIDs {
if i > 0 {
buf.WriteString(", ")
idList += ", "
}
buf.WriteString(strconv.FormatInt(elementID, 10))
idList += "%?"
paramIDs = append(paramIDs, elementID)
}
sql := fmt.Sprintf(completeDeleteMultiRangesSQL, jobID, buf.String())
_, err := ctx.(sqlexec.SQLExecutor).Execute(context.TODO(), sql)
var buf strings.Builder
_, err := fmt.Fprintf(&buf, completeDeleteMultiRangesSQL, idList)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Just let you know, we have fmt.Fprintf in our errcheck_ignore lists. And I agreed with that. fmt.Fprintf can not throw error on valid input, unless OOM. So you do not need to check that error.

BTW, You could use Fprintf to concat idList & nameList too.

if err != nil {
return errors.Trace(err)
}
_, err = ctx.(sqlexec.SQLExecutor).ExecuteInternal(context.TODO(), buf.String(), paramIDs...)
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)
_, err := ctx.(sqlexec.SQLExecutor).ExecuteInternal(context.TODO(), deleteDoneRecordSQL, dr.JobID, dr.ElementID)
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)
_, err := ctx.(sqlexec.SQLExecutor).ExecuteInternal(context.TODO(), updateDeleteRangeSQL, newStartKeyHex, dr.JobID, dr.ElementID, oldStartKeyHex)
return errors.Trace(err)
}

Expand All @@ -164,20 +164,32 @@ 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)"
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 := ""
paramNames := make([]interface{}, 0, len(varNames))
for i, name := range varNames {
if i > 0 {
nameList += ", "
}
nameList += fmt.Sprintf("'%s'", name)
nameList += "%?"
paramNames = append(paramNames, name)
}
var buf strings.Builder
_, err := fmt.Fprintf(&buf, loadGlobalVarsSQL, nameList)
if err != nil {
return errors.Trace(err)
}
stmt, err := sctx.ParseWithParams(context.Background(), buf.String(), paramNames...)
if err != nil {
return errors.Trace(err)
}
sql := fmt.Sprintf(loadGlobalVarsSQL, nameList)
rows, _, err := sctx.ExecRestrictedSQL(sql)
rows, _, err := sctx.ExecRestrictedStmt(context.Background(), stmt)
if err != nil {
return errors.Trace(err)
}
Expand Down