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

ddl/column.go

@@ -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)
@@ -1526,15 +1531,23 @@ 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)
+	sql := "select 1 from %n.%n where " + colsStr + " limit 1;"
+	stmt, err := ctx.(sqlexec.RestrictedSQLExecutor).ParseWithParams(context.Background(), sql, paramsList...)
+	if err != nil {
+		return errors.Trace(err)
+	}
+	rows, _, err := ctx.(sqlexec.RestrictedSQLExecutor).ExecRestrictedStmt(context.Background(), stmt)
 	if err != nil {
 		return errors.Trace(err)
 	}

ddl/db_integration_test.go

@@ -87,7 +87,7 @@ func setupIntegrationSuite(s *testIntegrationSuite, c *C) {
 	se, err := session.CreateSession4Test(s.store)
 	c.Assert(err, IsNil)
 	s.ctx = se.(sessionctx.Context)
-	_, err = se.Execute(context.Background(), "create database test_db")
+	_, err = se.ExecuteInternal(context.Background(), "create database test_db")
 	c.Assert(err, IsNil)
 }
 

ddl/db_partition_test.go

@@ -3185,7 +3185,7 @@ func (s *testIntegrationSuite7) TestCommitWhenSchemaChange(c *C) {
 	defer func() {
 		atomic.StoreUint32(&session.SchemaChangedWithoutRetry, 0)
 	}()
-	_, err := tk.Se.Execute(context.Background(), "commit")
+	_, err := tk.Se.ExecuteInternal(context.Background(), "commit")
 	c.Assert(domain.ErrInfoSchemaChanged.Equal(err), IsTrue)
 
 	// Cover a bug that schema validator does not prevent transaction commit when
@@ -3203,7 +3203,7 @@ func (s *testIntegrationSuite7) TestCommitWhenSchemaChange(c *C) {
 	tk.MustExec("insert into nt values (1), (3), (5);")
 	tk2.MustExec("alter table pt exchange partition p1 with table nt;")
 	tk.MustExec("insert into nt values (7), (9);")
-	_, err = tk.Se.Execute(context.Background(), "commit")
+	_, err = tk.Se.ExecuteInternal(context.Background(), "commit")
 	c.Assert(domain.ErrInfoSchemaChanged.Equal(err), IsTrue)
 
 	tk.MustExec("admin check table pt")
@@ -3215,7 +3215,7 @@ func (s *testIntegrationSuite7) TestCommitWhenSchemaChange(c *C) {
 	tk.MustExec("insert into pt values (1), (3), (5);")
 	tk2.MustExec("alter table pt exchange partition p1 with table nt;")
 	tk.MustExec("insert into pt values (7), (9);")
-	_, err = tk.Se.Execute(context.Background(), "commit")
+	_, err = tk.Se.ExecuteInternal(context.Background(), "commit")
 	c.Assert(domain.ErrInfoSchemaChanged.Equal(err), IsTrue)
 
 	tk.MustExec("admin check table pt")

ddl/db_test.go

@@ -115,13 +115,13 @@ func setUpSuite(s *testDBSuite, c *C) {
 	s.s, err = session.CreateSession4Test(s.store)
 	c.Assert(err, IsNil)
 
-	_, err = s.s.Execute(context.Background(), "create database test_db")
+	_, err = s.s.ExecuteInternal(context.Background(), "create database test_db")
 	c.Assert(err, IsNil)
-	s.s.Execute(context.Background(), "set @@global.tidb_max_delta_schema_count= 4096")
+	s.s.ExecuteInternal(context.Background(), "set @@global.tidb_max_delta_schema_count= 4096")
 }
 
 func tearDownSuite(s *testDBSuite, c *C) {
-	s.s.Execute(context.Background(), "drop database if exists test_db")
+	s.s.ExecuteInternal(context.Background(), "drop database if exists test_db")
 	s.s.Close()
 	s.dom.Close()
 	s.store.Close()
@@ -277,12 +277,12 @@ func backgroundExec(s kv.Storage, sql string, done chan error) {
 		return
 	}
 	defer se.Close()
-	_, err = se.Execute(context.Background(), "use test_db")
+	_, err = se.ExecuteInternal(context.Background(), "use test_db")
 	if err != nil {
 		done <- errors.Trace(err)
 		return
 	}
-	_, err = se.Execute(context.Background(), sql)
+	_, err = se.ExecuteInternal(context.Background(), sql)
 	done <- errors.Trace(err)
 }
 
@@ -2048,9 +2048,9 @@ func (s *testDBSuite1) TestColumn(c *C) {
 func sessionExec(c *C, s kv.Storage, sql string) {
 	se, err := session.CreateSession4Test(s)
 	c.Assert(err, IsNil)
-	_, err = se.Execute(context.Background(), "use test_db")
+	_, err = se.ExecuteInternal(context.Background(), "use test_db")
 	c.Assert(err, IsNil)
-	rs, err := se.Execute(context.Background(), sql)
+	rs, err := se.ExecuteInternal(context.Background(), sql)
 	c.Assert(err, IsNil, Commentf("err:%v", errors.ErrorStack(err)))
 	c.Assert(rs, IsNil)
 	se.Close()
@@ -2420,7 +2420,7 @@ func (s *testDBSuite5) TestRenameColumn(c *C) {
 }
 
 func (s *testDBSuite7) TestSelectInViewFromAnotherDB(c *C) {
-	_, _ = s.s.Execute(context.Background(), "create database test_db2")
+	_, _ = s.s.ExecuteInternal(context.Background(), "create database test_db2")
 	tk := testkit.NewTestKit(c, s.store)
 	tk.MustExec("use " + s.schemaName)
 	tk.MustExec("create table t(a int)")
@@ -5770,12 +5770,12 @@ func (s *testDBSuite4) testParallelExecSQL(c *C, sql1, sql2 string, se1, se2 ses
 	}()
 	go func() {
 		defer wg.Done()
-		_, err1 = se1.Execute(context.Background(), sql1)
+		_, err1 = se1.ExecuteInternal(context.Background(), sql1)
 	}()
 	go func() {
 		defer wg.Done()
 		<-ch
-		_, err2 = se2.Execute(context.Background(), sql2)
+		_, err2 = se2.ExecuteInternal(context.Background(), sql2)
 	}()
 
 	wg.Wait()

ddl/ddl_worker_test.go

@@ -1446,5 +1446,5 @@ func (s *testDDLSuite) TestDDLPackageExecuteSQL(c *C) {
 	c.Assert(err, IsNil)
 	defer worker.sessPool.put(sess)
 	se := sess.(sqlexec.SQLExecutor)
-	_, _ = se.Execute(context.Background(), "create table t(a int);")
+	_, _ = se.ExecuteInternal(context.Background(), "create table t(a int);")
 }

ddl/failtest/fail_db_test.go

@@ -89,7 +89,7 @@ func (s *testFailDBSuite) SetUpSuite(c *C) {
 }
 
 func (s *testFailDBSuite) TearDownSuite(c *C) {
-	_, err := s.se.Execute(context.Background(), "drop database if exists test_db_state")
+	_, err := s.se.ExecuteInternal(context.Background(), "drop database if exists test_db_state")
 	c.Assert(err, IsNil)
 	s.se.Close()
 	s.dom.Close()

ddl/testutil/testutil.go

@@ -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)
+			rs, err := se.ExecuteInternal(context.Background(), sql)
 			if err != nil {
 				done <- errors.Trace(err)
 				return

ddl/util/util.go

@@ -32,12 +32,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 (%?)`
+	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.
@@ -62,16 +62,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 {
@@ -106,8 +104,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)
 	}
@@ -117,8 +114,7 @@ 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)
 }
 
@@ -131,24 +127,21 @@ func RemoveMultiFromGCDeleteRange(ctx sessionctx.Context, jobID int64, elementID
 		}
 		buf.WriteString(strconv.FormatInt(elementID, 10))
 	}
-	sql := fmt.Sprintf(completeDeleteMultiRangesSQL, jobID, buf.String())
-	_, err := ctx.(sqlexec.SQLExecutor).Execute(context.TODO(), sql)
+	_, err := ctx.(sqlexec.SQLExecutor).ExecuteInternal(context.TODO(), completeDeleteMultiRangesSQL, jobID, 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)
+	_, 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)
 }
 
@@ -164,7 +157,7 @@ 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 (%?)"
 
 // LoadGlobalVars loads global variable from mysql.global_variables.
 func LoadGlobalVars(ctx sessionctx.Context, varNames []string) error {
@@ -176,8 +169,11 @@ func LoadGlobalVars(ctx sessionctx.Context, varNames []string) error {
 			}
 			nameList += fmt.Sprintf("'%s'", name)
 		}
-		sql := fmt.Sprintf(loadGlobalVarsSQL, nameList)
-		rows, _, err := sctx.ExecRestrictedSQL(sql)
+		stmt, err := sctx.ParseWithParams(context.Background(), loadGlobalVarsSQL, nameList)
+		if err != nil {
+			return errors.Trace(err)
+		}
+		rows, _, err := sctx.ExecRestrictedStmt(context.Background(), stmt)
 		if err != nil {
 			return errors.Trace(err)
 		}