Skip to content

Commit 1946f92

Browse files
D3HunterD3Hunter
authored
ddl: move insert job to table to job submitter (#56542)
ref #54436
1 parent a56674c commit 1946f92

File tree

2 files changed

+95
-99
lines changed

2 files changed

+95
-99
lines changed

‎pkg/ddl/job_scheduler.go

Lines changed: 1 addition & 99 deletions
Original file line numberDiff line numberDiff line change
@@ -15,13 +15,11 @@
1515
package ddl
1616

1717
import (
18-
"bytes"
1918
"context"
2019
"encoding/hex"
2120
"encoding/json"
2221
"fmt"
2322
"runtime"
24-
"slices"
2523
"strconv"
2624
"strings"
2725
"sync/atomic"
@@ -658,102 +656,6 @@ func (d *ddl) getTableByTxn(r autoid.Requirement, schemaID, tableID int64) (*mod
658656
return dbInfo, tbl, err
659657
}
660658

661-
const (
662-
addDDLJobSQL = "insert into mysql.tidb_ddl_job(job_id, reorg, schema_ids, table_ids, job_meta, type, processing) values"
663-
updateDDLJobSQL = "update mysql.tidb_ddl_job set job_meta = %s where job_id = %d"
664-
)
665-
666-
func insertDDLJobs2Table(ctx context.Context, se *sess.Session, jobWs ...*JobWrapper) error {
667-
failpoint.Inject("mockAddBatchDDLJobsErr", func(val failpoint.Value) {
668-
if val.(bool) {
669-
failpoint.Return(errors.Errorf("mockAddBatchDDLJobsErr"))
670-
}
671-
})
672-
if len(jobWs) == 0 {
673-
return nil
674-
}
675-
var sql bytes.Buffer
676-
sql.WriteString(addDDLJobSQL)
677-
for i, jobW := range jobWs {
678-
// TODO remove this check when all job type pass args in this way.
679-
if jobW.JobArgs != nil {
680-
jobW.FillArgs(jobW.JobArgs)
681-
}
682-
injectModifyJobArgFailPoint(jobWs)
683-
b, err := jobW.Encode(true)
684-
if err != nil {
685-
return err
686-
}
687-
if i != 0 {
688-
sql.WriteString(",")
689-
}
690-
fmt.Fprintf(&sql, "(%d, %t, %s, %s, %s, %d, %t)", jobW.ID, jobW.MayNeedReorg(),
691-
strconv.Quote(job2SchemaIDs(jobW)), strconv.Quote(job2TableIDs(jobW)),
692-
util.WrapKey2String(b), jobW.Type, jobW.Started())
693-
}
694-
se.GetSessionVars().SetDiskFullOpt(kvrpcpb.DiskFullOpt_AllowedOnAlmostFull)
695-
_, err := se.Execute(ctx, sql.String(), "insert_job")
696-
logutil.DDLLogger().Debug("add job to mysql.tidb_ddl_job table", zap.String("sql", sql.String()))
697-
return errors.Trace(err)
698-
}
699-
700-
func makeStringForIDs(ids []int64) string {
701-
set := make(map[int64]struct{}, len(ids))
702-
for _, id := range ids {
703-
set[id] = struct{}{}
704-
}
705-
706-
s := make([]string, 0, len(set))
707-
for id := range set {
708-
s = append(s, strconv.FormatInt(id, 10))
709-
}
710-
slices.Sort(s)
711-
return strings.Join(s, ",")
712-
}
713-
714-
func job2SchemaIDs(jobW *JobWrapper) string {
715-
switch jobW.Type {
716-
case model.ActionRenameTables:
717-
var ids []int64
718-
arg := jobW.JobArgs.(*model.RenameTablesArgs)
719-
ids = make([]int64, 0, len(arg.RenameTableInfos)*2)
720-
for _, info := range arg.RenameTableInfos {
721-
ids = append(ids, info.OldSchemaID, info.NewSchemaID)
722-
}
723-
return makeStringForIDs(ids)
724-
case model.ActionRenameTable:
725-
oldSchemaID := jobW.JobArgs.(*model.RenameTableArgs).OldSchemaID
726-
ids := []int64{oldSchemaID, jobW.SchemaID}
727-
return makeStringForIDs(ids)
728-
case model.ActionExchangeTablePartition:
729-
args := jobW.JobArgs.(*model.ExchangeTablePartitionArgs)
730-
return makeStringForIDs([]int64{jobW.SchemaID, args.PTSchemaID})
731-
default:
732-
return strconv.FormatInt(jobW.SchemaID, 10)
733-
}
734-
}
735-
736-
func job2TableIDs(jobW *JobWrapper) string {
737-
switch jobW.Type {
738-
case model.ActionRenameTables:
739-
var ids []int64
740-
arg := jobW.JobArgs.(*model.RenameTablesArgs)
741-
ids = make([]int64, 0, len(arg.RenameTableInfos))
742-
for _, info := range arg.RenameTableInfos {
743-
ids = append(ids, info.TableID)
744-
}
745-
return makeStringForIDs(ids)
746-
case model.ActionExchangeTablePartition:
747-
args := jobW.JobArgs.(*model.ExchangeTablePartitionArgs)
748-
return makeStringForIDs([]int64{jobW.TableID, args.PTTableID})
749-
case model.ActionTruncateTable:
750-
newTableID := jobW.JobArgs.(*model.TruncateTableArgs).NewTableID
751-
return strconv.FormatInt(jobW.TableID, 10) + "," + strconv.FormatInt(newTableID, 10)
752-
default:
753-
return strconv.FormatInt(jobW.TableID, 10)
754-
}
755-
}
756-
757659
func updateDDLJob2Table(
758660
ctx context.Context,
759661
se *sess.Session,
@@ -764,7 +666,7 @@ func updateDDLJob2Table(
764666
if err != nil {
765667
return err
766668
}
767-
sql := fmt.Sprintf(updateDDLJobSQL, util.WrapKey2String(b), job.ID)
669+
sql := fmt.Sprintf("update mysql.tidb_ddl_job set job_meta = %s where job_id = %d", util.WrapKey2String(b), job.ID)
768670
_, err = se.Execute(ctx, sql, "update_job")
769671
return errors.Trace(err)
770672
}

‎pkg/ddl/job_submitter.go

Lines changed: 94 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,9 +15,12 @@
1515
package ddl
1616

1717
import (
18+
"bytes"
1819
"context"
1920
"fmt"
2021
"math"
22+
"slices"
23+
"strconv"
2124
"strings"
2225
"time"
2326

@@ -677,6 +680,97 @@ func lockGlobalIDKey(ctx context.Context, ddlSe *sess.Session, txn kv.Transactio
677680
return forUpdateTs, err
678681
}
679682

683+
func insertDDLJobs2Table(ctx context.Context, se *sess.Session, jobWs ...*JobWrapper) error {
684+
failpoint.Inject("mockAddBatchDDLJobsErr", func(val failpoint.Value) {
685+
if val.(bool) {
686+
failpoint.Return(errors.Errorf("mockAddBatchDDLJobsErr"))
687+
}
688+
})
689+
if len(jobWs) == 0 {
690+
return nil
691+
}
692+
var sql bytes.Buffer
693+
sql.WriteString("insert into mysql.tidb_ddl_job(job_id, reorg, schema_ids, table_ids, job_meta, type, processing) values")
694+
for i, jobW := range jobWs {
695+
// TODO remove this check when all job type pass args in this way.
696+
if jobW.JobArgs != nil {
697+
jobW.FillArgs(jobW.JobArgs)
698+
}
699+
injectModifyJobArgFailPoint(jobWs)
700+
b, err := jobW.Encode(true)
701+
if err != nil {
702+
return err
703+
}
704+
if i != 0 {
705+
sql.WriteString(",")
706+
}
707+
fmt.Fprintf(&sql, "(%d, %t, %s, %s, %s, %d, %t)", jobW.ID, jobW.MayNeedReorg(),
708+
strconv.Quote(job2SchemaIDs(jobW)), strconv.Quote(job2TableIDs(jobW)),
709+
ddlutil.WrapKey2String(b), jobW.Type, jobW.Started())
710+
}
711+
se.GetSessionVars().SetDiskFullOpt(kvrpcpb.DiskFullOpt_AllowedOnAlmostFull)
712+
_, err := se.Execute(ctx, sql.String(), "insert_job")
713+
logutil.DDLLogger().Debug("add job to mysql.tidb_ddl_job table", zap.String("sql", sql.String()))
714+
return errors.Trace(err)
715+
}
716+
717+
func makeStringForIDs(ids []int64) string {
718+
set := make(map[int64]struct{}, len(ids))
719+
for _, id := range ids {
720+
set[id] = struct{}{}
721+
}
722+
723+
s := make([]string, 0, len(set))
724+
for id := range set {
725+
s = append(s, strconv.FormatInt(id, 10))
726+
}
727+
slices.Sort(s)
728+
return strings.Join(s, ",")
729+
}
730+
731+
func job2SchemaIDs(jobW *JobWrapper) string {
732+
switch jobW.Type {
733+
case model.ActionRenameTables:
734+
var ids []int64
735+
arg := jobW.JobArgs.(*model.RenameTablesArgs)
736+
ids = make([]int64, 0, len(arg.RenameTableInfos)*2)
737+
for _, info := range arg.RenameTableInfos {
738+
ids = append(ids, info.OldSchemaID, info.NewSchemaID)
739+
}
740+
return makeStringForIDs(ids)
741+
case model.ActionRenameTable:
742+
oldSchemaID := jobW.JobArgs.(*model.RenameTableArgs).OldSchemaID
743+
ids := []int64{oldSchemaID, jobW.SchemaID}
744+
return makeStringForIDs(ids)
745+
case model.ActionExchangeTablePartition:
746+
args := jobW.JobArgs.(*model.ExchangeTablePartitionArgs)
747+
return makeStringForIDs([]int64{jobW.SchemaID, args.PTSchemaID})
748+
default:
749+
return strconv.FormatInt(jobW.SchemaID, 10)
750+
}
751+
}
752+
753+
func job2TableIDs(jobW *JobWrapper) string {
754+
switch jobW.Type {
755+
case model.ActionRenameTables:
756+
var ids []int64
757+
arg := jobW.JobArgs.(*model.RenameTablesArgs)
758+
ids = make([]int64, 0, len(arg.RenameTableInfos))
759+
for _, info := range arg.RenameTableInfos {
760+
ids = append(ids, info.TableID)
761+
}
762+
return makeStringForIDs(ids)
763+
case model.ActionExchangeTablePartition:
764+
args := jobW.JobArgs.(*model.ExchangeTablePartitionArgs)
765+
return makeStringForIDs([]int64{jobW.TableID, args.PTTableID})
766+
case model.ActionTruncateTable:
767+
newTableID := jobW.JobArgs.(*model.TruncateTableArgs).NewTableID
768+
return strconv.FormatInt(jobW.TableID, 10) + "," + strconv.FormatInt(newTableID, 10)
769+
default:
770+
return strconv.FormatInt(jobW.TableID, 10)
771+
}
772+
}
773+
680774
// TODO this failpoint is only checking how job scheduler handle
681775
// corrupted job args, we should test it there by UT, not here.
682776
func injectModifyJobArgFailPoint(jobWs []*JobWrapper) {

0 commit comments

Comments
 (0)