ddl: use JobArgs in job wrapper and use doDDLJob2 for some job type

pkg/ddl/ddl_test.go

@@ -219,7 +219,12 @@ func TestBuildJobDependence(t *testing.T) {
 	job6 := &model.Job{ID: 6, TableID: 1, Version: model.JobVersion1, Type: model.ActionDropTable}
 	job7 := &model.Job{ID: 7, TableID: 2, Version: model.JobVersion1, Type: model.ActionModifyColumn}
 	job9 := &model.Job{ID: 9, SchemaID: 111, Version: model.JobVersion1, Type: model.ActionDropSchema}
-	job11 := &model.Job{ID: 11, TableID: 2, Version: model.JobVersion1, Type: model.ActionRenameTable, Args: []any{int64(111), "old db name"}}
+	job11 := &model.Job{ID: 11, TableID: 2, Version: model.JobVersion1, Type: model.ActionRenameTable}
+	job11.FillArgs(&model.RenameTableArgs{
+		OldSchemaID:   111,
+		NewTableName:  pmodel.NewCIStr("new table name"),
+		OldSchemaName: pmodel.NewCIStr("old db name"),
+	})
 	err := kv.RunInNewTxn(ctx, store, false, func(ctx context.Context, txn kv.Transaction) error {
 		m := meta.NewMutator(txn)
 		require.NoError(t, m.EnQueueDDLJob(job1))

pkg/ddl/executor.go

@@ -310,17 +310,17 @@ func (e *executor) CreateSchemaWithInfo(
 		}},
 		SQLMode: ctx.GetSessionVars().SQLMode,
 	}
-	job.FillArgs(&model.CreateSchemaArgs{
+	args := &model.CreateSchemaArgs{
 		DBInfo: dbInfo,
-	})
+	}
 	if ref := dbInfo.PlacementPolicyRef; ref != nil {
 		job.InvolvingSchemaInfo = append(job.InvolvingSchemaInfo, model.InvolvingSchemaInfo{
 			Policy: ref.Name.L,
 			Mode:   model.SharedInvolving,
 		})
 	}
 
-	err := e.DoDDLJob(ctx, job)
+	err := e.doDDLJob2(ctx, job, args)
 
 	if infoschema.ErrDatabaseExists.Equal(err) && onExist == OnExistIgnore {
 		ctx.GetSessionVars().StmtCtx.AppendNote(err)
@@ -361,11 +361,11 @@ func (e *executor) ModifySchemaCharsetAndCollate(ctx sessionctx.Context, stmt *a
 		}},
 		SQLMode: ctx.GetSessionVars().SQLMode,
 	}
-	job.FillArgs(&model.ModifySchemaArgs{
+	args := &model.ModifySchemaArgs{
 		ToCharset: toCharset,
 		ToCollate: toCollate,
-	})
-	err = e.DoDDLJob(ctx, job)
+	}
+	err = e.doDDLJob2(ctx, job, args)
 	return errors.Trace(err)
 }
 
@@ -400,15 +400,15 @@ func (e *executor) ModifySchemaDefaultPlacement(ctx sessionctx.Context, stmt *as
 		}},
 		SQLMode: ctx.GetSessionVars().SQLMode,
 	}
-	job.FillArgs(&model.ModifySchemaArgs{PolicyRef: placementPolicyRef})
+	args := &model.ModifySchemaArgs{PolicyRef: placementPolicyRef}
 
 	if placementPolicyRef != nil {
 		job.InvolvingSchemaInfo = append(job.InvolvingSchemaInfo, model.InvolvingSchemaInfo{
 			Policy: placementPolicyRef.Name.L,
 			Mode:   model.SharedInvolving,
 		})
 	}
-	err = e.DoDDLJob(ctx, job)
+	err = e.doDDLJob2(ctx, job, args)
 	return errors.Trace(err)
 }
 
@@ -764,11 +764,11 @@ func (e *executor) DropSchema(ctx sessionctx.Context, stmt *ast.DropDatabaseStmt
 		}},
 		SQLMode: ctx.GetSessionVars().SQLMode,
 	}
-	job.FillArgs(&model.DropSchemaArgs{
+	args := &model.DropSchemaArgs{
 		FKCheck: fkCheck,
-	})
+	}
 
-	err = e.DoDDLJob(ctx, job)
+	err = e.doDDLJob2(ctx, job, args)
 	if err != nil {
 		if infoschema.ErrDatabaseNotExists.Equal(err) {
 			if stmt.IfExists {

pkg/ddl/job_submitter.go

@@ -541,12 +541,7 @@ func assignGIDsForJobs(jobWs []*JobWrapper, ids []int64) {
 				}
 			}
 		case model.ActionCreateSchema:
-			var dbInfo *model.DBInfo
-			if jobW.Version == model.JobVersion1 {
-				dbInfo = jobW.Args[0].(*model.DBInfo)
-			} else {
-				dbInfo = jobW.Args[0].(*model.CreateSchemaArgs).DBInfo
-			}
+			dbInfo := jobW.JobArgs.(*model.CreateSchemaArgs).DBInfo
 			if !jobW.IDAllocated {
 				dbInfo.ID = alloc.next()
 			}

pkg/ddl/job_submitter_test.go

@@ -163,14 +163,13 @@ func TestCombinedIDAllocation(t *testing.T) {
 		)
 	}
 
-	genCreateDBJob := func() *model.Job {
+	genCreateDBJob := func(idAllocated bool) *ddl.JobWrapper {
 		info := &model.DBInfo{}
 		j := &model.Job{
 			Version: model.GetJobVerInUse(),
 			Type:    model.ActionCreateSchema,
 		}
-		j.FillArgs(&model.CreateSchemaArgs{DBInfo: info})
-		return j
+		return ddl.NewJobWrapperWithArgs(j, &model.CreateSchemaArgs{DBInfo: info}, idAllocated)
 	}
 
 	genRGroupJob := func(idAllocated bool) *ddl.JobWrapper {
@@ -279,11 +278,11 @@ func TestCombinedIDAllocation(t *testing.T) {
 			requiredIDCount: 1,
 		},
 		{
-			jobW:            ddl.NewJobWrapper(genCreateDBJob(), false),
+			jobW:            genCreateDBJob(false),
 			requiredIDCount: 2,
 		},
 		{
-			jobW:            ddl.NewJobWrapper(genCreateDBJob(), true),
+			jobW:            genCreateDBJob(true),
 			requiredIDCount: 1,
 		},
 		{

pkg/ddl/job_worker_test.go

@@ -50,13 +50,12 @@ func TestInvalidDDLJob(t *testing.T) {
 		TableID:             0,
 		Type:                model.ActionNone,
 		BinlogInfo:          &model.HistoryInfo{},
-		Args:                []any{},
 		InvolvingSchemaInfo: []model.InvolvingSchemaInfo{{}},
 	}
 	ctx := testNewContext(store)
 	ctx.SetValue(sessionctx.QueryString, "skip")
 	de := dom.DDLExecutor().(ddl.ExecutorForTest)
-	err := de.DoDDLJobWrapper(ctx, ddl.NewJobWrapper(job, true))
+	err := de.DoDDLJobWrapper(ctx, ddl.NewJobWrapperWithArgs(job, &model.EmptyArgs{}, true))
 	require.ErrorContains(t, err, "[ddl:8204]invalid ddl job type: none")
 }
 

pkg/ddl/restart_test.go

@@ -120,12 +120,11 @@ func TestSchemaResume(t *testing.T) {
 		Type:       model.ActionCreateSchema,
 		BinlogInfo: &model.HistoryInfo{},
 	}
-	job.FillArgs(&model.CreateSchemaArgs{DBInfo: dbInfo})
-	testRunInterruptedJob(t, store, dom, job, nil)
+	testRunInterruptedJob(t, store, dom, job, &model.CreateSchemaArgs{DBInfo: dbInfo})
 	testCheckSchemaState(t, store, dbInfo, model.StatePublic)
 
 	job = buildDropSchemaJob(dbInfo)
-	testRunInterruptedJob(t, store, dom, job, nil)
+	testRunInterruptedJob(t, store, dom, job, &model.DropSchemaArgs{FKCheck: true})
 	testCheckSchemaState(t, store, dbInfo, model.StateNone)
 }
 
@@ -139,7 +138,7 @@ func TestStat(t *testing.T) {
 
 	job := buildDropSchemaJob(dbInfo)
 	done := make(chan error, 1)
-	go runInterruptedJob(t, store, dom.DDLExecutor(), job, nil, done)
+	go runInterruptedJob(t, store, dom.DDLExecutor(), job, &model.DropSchemaArgs{FKCheck: true}, done)
 
 	ticker := time.NewTicker(dom.GetSchemaLease() * 1)
 	defer ticker.Stop()

pkg/ddl/schema_test.go

@@ -146,9 +146,8 @@ func testCreateSchema(t *testing.T, ctx sessionctx.Context, d ddl.ExecutorForTes
 			Table:    model.InvolvingAll,
 		}},
 	}
-	job.FillArgs(&model.CreateSchemaArgs{DBInfo: dbInfo})
 	ctx.SetValue(sessionctx.QueryString, "skip")
-	require.NoError(t, d.DoDDLJobWrapper(ctx, ddl.NewJobWrapper(job, true)))
+	require.NoError(t, d.DoDDLJobWrapper(ctx, ddl.NewJobWrapperWithArgs(job, &model.CreateSchemaArgs{DBInfo: dbInfo}, true)))
 
 	v := getSchemaVer(t, ctx)
 	dbInfo.State = model.StatePublic
@@ -168,14 +167,13 @@ func buildDropSchemaJob(dbInfo *model.DBInfo) *model.Job {
 			Table:    model.InvolvingAll,
 		}},
 	}
-	j.FillArgs(&model.DropSchemaArgs{FKCheck: true})
 	return j
 }
 
 func testDropSchema(t *testing.T, ctx sessionctx.Context, d ddl.ExecutorForTest, dbInfo *model.DBInfo) (*model.Job, int64) {
 	job := buildDropSchemaJob(dbInfo)
 	ctx.SetValue(sessionctx.QueryString, "skip")
-	err := d.DoDDLJobWrapper(ctx, ddl.NewJobWrapper(job, true))
+	err := d.DoDDLJobWrapper(ctx, ddl.NewJobWrapperWithArgs(job, &model.DropSchemaArgs{FKCheck: true}, true))
 	require.NoError(t, err)
 	ver := getSchemaVer(t, ctx)
 	return job, ver
@@ -282,7 +280,7 @@ func TestSchema(t *testing.T) {
 	}
 	ctx := testkit.NewTestKit(t, store).Session()
 	ctx.SetValue(sessionctx.QueryString, "skip")
-	err = de.DoDDLJobWrapper(ctx, ddl.NewJobWrapper(job, true))
+	err = de.DoDDLJobWrapper(ctx, ddl.NewJobWrapperWithArgs(job, &model.DropSchemaArgs{}, true))
 	require.True(t, terror.ErrorEqual(err, infoschema.ErrDatabaseDropExists), "err %v", err)
 
 	// Drop a database without a table.
@@ -339,8 +337,7 @@ func TestSchemaWaitJob(t *testing.T) {
 	schemaID := genIDs[0]
 	doDDLJobErr(t, schemaID, 0, "test_schema", "", model.ActionCreateSchema,
 		testkit.NewTestKit(t, store).Session(), det2, store, func(job *model.Job) model.JobArgs {
-			job.FillArgs(&model.CreateSchemaArgs{DBInfo: dbInfo})
-			return nil
+			return &model.CreateSchemaArgs{DBInfo: dbInfo}
 		})
 }