diff --git a/pkg/ddl/executor.go b/pkg/ddl/executor.go
index 11acc18ae7f3a..87f725e265501 100644
--- a/pkg/ddl/executor.go
+++ b/pkg/ddl/executor.go
@@ -1988,6 +1988,8 @@ func (e *executor) multiSchemaChange(ctx sessionctx.Context, ti ast.Ident, info
 		logFn = logutil.DDLLogger().Fatal
 	}
 
+	// to do:(joccau)
+	// we need refactor this part to support V2 job version after refactor all of ddl types.
 	var involvingSchemaInfo []model.InvolvingSchemaInfo
 	for _, j := range subJobs {
 		switch j.Type {
@@ -2132,17 +2134,23 @@ func (e *executor) RebaseAutoID(ctx sessionctx.Context, ident ast.Ident, newBase
 		newBase = newBaseTemp
 	}
 	job := &model.Job{
+		Version:        model.JobVersion1,
 		SchemaID:       schema.ID,
 		TableID:        tbInfo.ID,
 		SchemaName:     schema.Name.L,
 		TableName:      tbInfo.Name.L,
 		Type:           actionType,
 		BinlogInfo:     &model.HistoryInfo{},
-		Args:           []any{newBase, force},
 		CDCWriteSource: ctx.GetSessionVars().CDCWriteSource,
 		SQLMode:        ctx.GetSessionVars().SQLMode,
 	}
-	err = e.DoDDLJob(ctx, job)
+	args := &model.RebaseAutoIDArgs{
+		NewBase: newBase,
+		Force:   force,
+	}
+	// need fill args, the job will be pushed subjob.
+	job.FillArgs(args)
+	err = e.doDDLJob2(ctx, job, args)
 	return errors.Trace(err)
 }
 
@@ -3525,18 +3533,19 @@ func (e *executor) AlterTableComment(ctx sessionctx.Context, ident ast.Ident, sp
 	}
 
 	job := &model.Job{
+		Version:        model.JobVersion1,
 		SchemaID:       schema.ID,
 		TableID:        tb.Meta().ID,
 		SchemaName:     schema.Name.L,
 		TableName:      tb.Meta().Name.L,
 		Type:           model.ActionModifyTableComment,
 		BinlogInfo:     &model.HistoryInfo{},
-		Args:           []any{spec.Comment},
 		CDCWriteSource: ctx.GetSessionVars().CDCWriteSource,
 		SQLMode:        ctx.GetSessionVars().SQLMode,
 	}
-
-	err = e.DoDDLJob(ctx, job)
+	args := &model.ModifyTableCommentArgs{Comment: spec.Comment}
+	job.FillArgs(args)
+	err = e.doDDLJob2(ctx, job, args)
 	return errors.Trace(err)
 }
 
@@ -3607,17 +3616,24 @@ func (e *executor) AlterTableCharsetAndCollate(ctx sessionctx.Context, ident ast
 	}
 
 	job := &model.Job{
+		Version:        model.JobVersion1,
 		SchemaID:       schema.ID,
 		TableID:        tb.Meta().ID,
 		SchemaName:     schema.Name.L,
 		TableName:      tb.Meta().Name.L,
 		Type:           model.ActionModifyTableCharsetAndCollate,
 		BinlogInfo:     &model.HistoryInfo{},
-		Args:           []any{toCharset, toCollate, needsOverwriteCols},
 		CDCWriteSource: ctx.GetSessionVars().CDCWriteSource,
 		SQLMode:        ctx.GetSessionVars().SQLMode,
 	}
-	err = e.DoDDLJob(ctx, job)
+
+	args := &model.ModifyTableCharsetAndCollateArgs{
+		ToCharset:          toCharset,
+		ToCollate:          toCollate,
+		NeedsOverwriteCols: needsOverwriteCols,
+	}
+	job.FillArgs(args)
+	err = e.doDDLJob2(ctx, job, args)
 	return errors.Trace(err)
 }
 
@@ -4955,13 +4971,13 @@ func (e *executor) CreateForeignKey(ctx sessionctx.Context, ti ast.Ident, fkName
 	}
 
 	job := &model.Job{
+		Version:        model.JobVersion1,
 		SchemaID:       schema.ID,
 		TableID:        t.Meta().ID,
 		SchemaName:     schema.Name.L,
 		TableName:      t.Meta().Name.L,
 		Type:           model.ActionAddForeignKey,
 		BinlogInfo:     &model.HistoryInfo{},
-		Args:           []any{fkInfo, fkCheck},
 		CDCWriteSource: ctx.GetSessionVars().CDCWriteSource,
 		InvolvingSchemaInfo: []model.InvolvingSchemaInfo{
 			{
@@ -4976,8 +4992,12 @@ func (e *executor) CreateForeignKey(ctx sessionctx.Context, ti ast.Ident, fkName
 		},
 		SQLMode: ctx.GetSessionVars().SQLMode,
 	}
-
-	err = e.DoDDLJob(ctx, job)
+	args := &model.AddForeignKeyArgs{
+		FkInfo:  fkInfo,
+		FkCheck: fkCheck,
+	}
+	job.FillArgs(args)
+	err = e.doDDLJob2(ctx, job, args)
 	return errors.Trace(err)
 }
 
@@ -4994,6 +5014,7 @@ func (e *executor) DropForeignKey(ctx sessionctx.Context, ti ast.Ident, fkName p
 	}
 
 	job := &model.Job{
+		Version:        model.JobVersion1,
 		SchemaID:       schema.ID,
 		TableID:        t.Meta().ID,
 		SchemaName:     schema.Name.L,
@@ -5001,12 +5022,12 @@ func (e *executor) DropForeignKey(ctx sessionctx.Context, ti ast.Ident, fkName p
 		TableName:      t.Meta().Name.L,
 		Type:           model.ActionDropForeignKey,
 		BinlogInfo:     &model.HistoryInfo{},
-		Args:           []any{fkName},
 		CDCWriteSource: ctx.GetSessionVars().CDCWriteSource,
 		SQLMode:        ctx.GetSessionVars().SQLMode,
 	}
-
-	err = e.DoDDLJob(ctx, job)
+	args := &model.DropForeignKeyArgs{FkName: fkName}
+	job.FillArgs(args)
+	err = e.doDDLJob2(ctx, job, args)
 	return errors.Trace(err)
 }
 
@@ -5614,18 +5635,22 @@ func (e *executor) AlterIndexVisibility(ctx sessionctx.Context, ident ast.Ident,
 	}
 
 	job := &model.Job{
+		Version:        model.JobVersion1,
 		SchemaID:       schema.ID,
 		TableID:        tb.Meta().ID,
 		SchemaName:     schema.Name.L,
 		TableName:      tb.Meta().Name.L,
 		Type:           model.ActionAlterIndexVisibility,
 		BinlogInfo:     &model.HistoryInfo{},
-		Args:           []any{indexName, invisible},
 		CDCWriteSource: ctx.GetSessionVars().CDCWriteSource,
 		SQLMode:        ctx.GetSessionVars().SQLMode,
 	}
-
-	err = e.DoDDLJob(ctx, job)
+	args := &model.AlterIndexVisibilityArgs{
+		IndexName: indexName,
+		Invisible: invisible,
+	}
+	job.FillArgs(args)
+	err = e.doDDLJob2(ctx, job, args)
 	return errors.Trace(err)
 }
 
diff --git a/pkg/ddl/foreign_key.go b/pkg/ddl/foreign_key.go
index c340c5cd2f684..9b2f3f47aeb55 100644
--- a/pkg/ddl/foreign_key.go
+++ b/pkg/ddl/foreign_key.go
@@ -39,26 +39,26 @@ func (w *worker) onCreateForeignKey(jobCtx *jobContext, t *meta.Meta, job *model
 		return ver, errors.Trace(err)
 	}
 
-	var fkInfo model.FKInfo
-	var fkCheck bool
-	err = job.DecodeArgs(&fkInfo, &fkCheck)
+	args, err := model.GetAddForeignKeyArgs(job)
 	if err != nil {
 		job.State = model.JobStateCancelled
 		return ver, errors.Trace(err)
 	}
+	fkInfo, fkCheck := args.FkInfo, args.FkCheck
+
 	if job.IsRollingback() {
 		return dropForeignKey(jobCtx, t, job, tblInfo, fkInfo.Name)
 	}
 	switch job.SchemaState {
 	case model.StateNone:
-		err = checkAddForeignKeyValidInOwner(jobCtx.infoCache, job.SchemaName, tblInfo, &fkInfo, fkCheck)
+		err = checkAddForeignKeyValidInOwner(jobCtx.infoCache, job.SchemaName, tblInfo, fkInfo, fkCheck)
 		if err != nil {
 			job.State = model.JobStateCancelled
 			return ver, err
 		}
 		fkInfo.State = model.StateWriteOnly
 		fkInfo.ID = allocateFKIndexID(tblInfo)
-		tblInfo.ForeignKeys = append(tblInfo.ForeignKeys, &fkInfo)
+		tblInfo.ForeignKeys = append(tblInfo.ForeignKeys, fkInfo)
 		ver, err = updateVersionAndTableInfo(jobCtx, t, job, tblInfo, true)
 		if err != nil {
 			return ver, errors.Trace(err)
@@ -66,7 +66,7 @@ func (w *worker) onCreateForeignKey(jobCtx *jobContext, t *meta.Meta, job *model
 		job.SchemaState = model.StateWriteOnly
 		return ver, nil
 	case model.StateWriteOnly:
-		err = checkForeignKeyConstrain(w, job.SchemaName, tblInfo.Name.L, &fkInfo, fkCheck)
+		err = checkForeignKeyConstrain(w, job.SchemaName, tblInfo.Name.L, fkInfo, fkCheck)
 		if err != nil {
 			job.State = model.JobStateRollingback
 			return ver, err
@@ -99,13 +99,12 @@ func onDropForeignKey(jobCtx *jobContext, t *meta.Meta, job *model.Job) (ver int
 		return ver, errors.Trace(err)
 	}
 
-	var fkName pmodel.CIStr
-	err = job.DecodeArgs(&fkName)
+	args, err := model.GetDropForeignKeyArgs(job)
 	if err != nil {
 		job.State = model.JobStateCancelled
 		return ver, errors.Trace(err)
 	}
-	return dropForeignKey(jobCtx, t, job, tblInfo, fkName)
+	return dropForeignKey(jobCtx, t, job, tblInfo, args.FkName)
 }
 
 func dropForeignKey(jobCtx *jobContext, t *meta.Meta, job *model.Job, tblInfo *model.TableInfo, fkName pmodel.CIStr) (ver int64, err error) {
diff --git a/pkg/ddl/foreign_key_test.go b/pkg/ddl/foreign_key_test.go
index 2a7d2227e6df2..a481525bdd0bd 100644
--- a/pkg/ddl/foreign_key_test.go
+++ b/pkg/ddl/foreign_key_test.go
@@ -58,34 +58,39 @@ func testCreateForeignKey(t *testing.T, d ddl.ExecutorForTest, ctx sessionctx.Co
 	}
 
 	job := &model.Job{
+		Version:    model.GetJobVerInUse(),
 		SchemaID:   dbInfo.ID,
 		SchemaName: dbInfo.Name.L,
 		TableID:    tblInfo.ID,
 		TableName:  tblInfo.Name.L,
 		Type:       model.ActionAddForeignKey,
 		BinlogInfo: &model.HistoryInfo{},
-		Args:       []any{fkInfo},
 	}
 	err := sessiontxn.NewTxn(context.Background(), ctx)
 	require.NoError(t, err)
 	ctx.SetValue(sessionctx.QueryString, "skip")
-	err = d.DoDDLJobWrapper(ctx, ddl.NewJobWrapper(job, true))
+
+	args := &model.AddForeignKeyArgs{FkInfo: fkInfo}
+	job.FillArgs(args)
+	err = d.DoDDLJobWrapper(ctx, ddl.NewJobWrapperWithArgs(job, args, true))
 	require.NoError(t, err)
 	return job
 }
 
 func testDropForeignKey(t *testing.T, ctx sessionctx.Context, d ddl.ExecutorForTest, dbInfo *model.DBInfo, tblInfo *model.TableInfo, foreignKeyName string) *model.Job {
 	job := &model.Job{
+		Version:    model.GetJobVerInUse(),
 		SchemaID:   dbInfo.ID,
 		SchemaName: dbInfo.Name.L,
 		TableID:    tblInfo.ID,
 		TableName:  tblInfo.Name.L,
 		Type:       model.ActionDropForeignKey,
 		BinlogInfo: &model.HistoryInfo{},
-		Args:       []any{pmodel.NewCIStr(foreignKeyName)},
 	}
 	ctx.SetValue(sessionctx.QueryString, "skip")
-	err := d.DoDDLJobWrapper(ctx, ddl.NewJobWrapper(job, true))
+	args := &model.DropForeignKeyArgs{FkName: pmodel.NewCIStr(foreignKeyName)}
+	job.FillArgs(args)
+	err := d.DoDDLJobWrapper(ctx, ddl.NewJobWrapperWithArgs(job, args, true))
 	require.NoError(t, err)
 	v := getSchemaVer(t, ctx)
 	checkHistoryJobArgs(t, ctx, job.ID, &historyJobArgs{ver: v, tbl: tblInfo})
diff --git a/pkg/ddl/index.go b/pkg/ddl/index.go
index 2f292703bff7d..44d13b4254305 100644
--- a/pkg/ddl/index.go
+++ b/pkg/ddl/index.go
@@ -1327,10 +1327,12 @@ func checkAlterIndexVisibility(t *meta.Meta, job *model.Job) (*model.TableInfo,
 		return nil, indexName, invisible, errors.Trace(err)
 	}
 
-	if err := job.DecodeArgs(&indexName, &invisible); err != nil {
+	args, err := model.GetAlterIndexVisibilityArgs(job)
+	if err != nil {
 		job.State = model.JobStateCancelled
 		return nil, indexName, invisible, errors.Trace(err)
 	}
+	indexName, invisible = args.IndexName, args.Invisible
 
 	skip, err := validateAlterIndexVisibility(nil, indexName, invisible, tblInfo)
 	if err != nil {
diff --git a/pkg/ddl/multi_schema_change.go b/pkg/ddl/multi_schema_change.go
index c9500f2d437eb..0d8bb4e79f3cb 100644
--- a/pkg/ddl/multi_schema_change.go
+++ b/pkg/ddl/multi_schema_change.go
@@ -251,11 +251,11 @@ func fillMultiSchemaInfo(info *model.MultiSchemaInfo, job *JobWrapper) error {
 		col := job.Args[0].(*table.Column)
 		info.ModifyColumns = append(info.ModifyColumns, col.Name)
 	case model.ActionAlterIndexVisibility:
-		idxName := job.Args[0].(pmodel.CIStr)
+		idxName := job.JobArgs.(*model.AlterIndexVisibilityArgs).IndexName
 		info.AlterIndexes = append(info.AlterIndexes, idxName)
 	case model.ActionRebaseAutoID, model.ActionModifyTableComment, model.ActionModifyTableCharsetAndCollate:
 	case model.ActionAddForeignKey:
-		fkInfo := job.Args[0].(*model.FKInfo)
+		fkInfo := job.JobArgs.(*model.AddForeignKeyArgs).FkInfo
 		info.AddForeignKeys = append(info.AddForeignKeys, model.AddForeignKeyInfo{
 			Name: fkInfo.Name,
 			Cols: fkInfo.Cols,
diff --git a/pkg/ddl/table.go b/pkg/ddl/table.go
index 58c76b4f6df50..1913b7577c5cc 100644
--- a/pkg/ddl/table.go
+++ b/pkg/ddl/table.go
@@ -592,17 +592,15 @@ func onRebaseAutoRandomType(jobCtx *jobContext, t *meta.Meta, job *model.Job) (v
 }
 
 func onRebaseAutoID(jobCtx *jobContext, t *meta.Meta, job *model.Job, tp autoid.AllocatorType) (ver int64, _ error) {
-	schemaID := job.SchemaID
-	var (
-		newBase int64
-		force   bool
-	)
-	err := job.DecodeArgs(&newBase, &force)
+	args, err := model.GetRebaseAutoIDArgs(job)
 	if err != nil {
 		job.State = model.JobStateCancelled
 		return ver, errors.Trace(err)
 	}
 
+	schemaID := job.SchemaID
+	newBase, force := args.NewBase, args.Force
+
 	if job.MultiSchemaInfo != nil && job.MultiSchemaInfo.Revertible {
 		job.MarkNonRevertible()
 		return ver, nil
@@ -987,8 +985,8 @@ func finishJobRenameTables(jobCtx *jobContext, t *meta.Meta, job *model.Job, arg
 }
 
 func onModifyTableComment(jobCtx *jobContext, t *meta.Meta, job *model.Job) (ver int64, _ error) {
-	var comment string
-	if err := job.DecodeArgs(&comment); err != nil {
+	args, err := model.GetModifyTableCommentArgs(job)
+	if err != nil {
 		job.State = model.JobStateCancelled
 		return ver, errors.Trace(err)
 	}
@@ -1003,7 +1001,7 @@ func onModifyTableComment(jobCtx *jobContext, t *meta.Meta, job *model.Job) (ver
 		return ver, nil
 	}
 
-	tblInfo.Comment = comment
+	tblInfo.Comment = args.Comment
 	ver, err = updateVersionAndTableInfo(jobCtx, t, job, tblInfo, true)
 	if err != nil {
 		return ver, errors.Trace(err)
@@ -1013,12 +1011,12 @@ func onModifyTableComment(jobCtx *jobContext, t *meta.Meta, job *model.Job) (ver
 }
 
 func onModifyTableCharsetAndCollate(jobCtx *jobContext, t *meta.Meta, job *model.Job) (ver int64, _ error) {
-	var toCharset, toCollate string
-	var needsOverwriteCols bool
-	if err := job.DecodeArgs(&toCharset, &toCollate, &needsOverwriteCols); err != nil {
+	args, err := model.GetModifyTableCharsetAndCollateArgs(job)
+	if err != nil {
 		job.State = model.JobStateCancelled
 		return ver, errors.Trace(err)
 	}
+	toCharset, toCollate, needsOverwriteCols := args.ToCharset, args.ToCollate, args.NeedsOverwriteCols
 
 	dbInfo, err := checkSchemaExistAndCancelNotExistJob(t, job)
 	if err != nil {
diff --git a/pkg/meta/model/BUILD.bazel b/pkg/meta/model/BUILD.bazel
index 60a6099325eed..08d9171be4f05 100644
--- a/pkg/meta/model/BUILD.bazel
+++ b/pkg/meta/model/BUILD.bazel
@@ -45,7 +45,7 @@ go_test(
     ],
     embed = [":model"],
     flaky = True,
-    shard_count = 41,
+    shard_count = 46,
     deps = [
         "//pkg/parser/ast",
         "//pkg/parser/charset",
diff --git a/pkg/meta/model/job.go b/pkg/meta/model/job.go
index edcebe755b4a1..2f244542b24ca 100644
--- a/pkg/meta/model/job.go
+++ b/pkg/meta/model/job.go
@@ -482,6 +482,7 @@ func (job *Job) GetWarnings() (map[errors.ErrorID]*terror.Error, map[errors.Erro
 
 // FillArgs fills args for new job.
 func (job *Job) FillArgs(args JobArgs) {
+	intest.Assert(job.Version == JobVersion1 || job.Version == JobVersion2, "job version is invalid")
 	args.fillJob(job)
 }
 
diff --git a/pkg/meta/model/job_args.go b/pkg/meta/model/job_args.go
index 866794382d71c..91dee5bd15b73 100644
--- a/pkg/meta/model/job_args.go
+++ b/pkg/meta/model/job_args.go
@@ -687,6 +687,191 @@ func GetResourceGroupArgs(job *Job) (*ResourceGroupArgs, error) {
 	return getOrDecodeArgsV2[*ResourceGroupArgs](job)
 }
 
+// RebaseAutoIDArgs is the arguments for ActionRebaseAutoID DDL.
+// It is also for ActionRebaseAutoRandomBase.
+type RebaseAutoIDArgs struct {
+	NewBase int64 `json:"new_base,omitempty"`
+	Force   bool  `json:"force,omitempty"`
+}
+
+func (a *RebaseAutoIDArgs) fillJob(job *Job) {
+	if job.Version == JobVersion1 {
+		job.Args = []any{a.NewBase, a.Force}
+	} else {
+		job.Args = []any{a}
+	}
+}
+
+// GetRebaseAutoIDArgs the args for ActionRebaseAutoID/ActionRebaseAutoRandomBase ddl.
+func GetRebaseAutoIDArgs(job *Job) (*RebaseAutoIDArgs, error) {
+	var (
+		newBase int64
+		force   bool
+	)
+
+	if job.Version == JobVersion1 {
+		if err := job.DecodeArgs(&newBase, &force); err != nil {
+			return nil, errors.Trace(err)
+		}
+		return &RebaseAutoIDArgs{
+			NewBase: newBase,
+			Force:   force,
+		}, nil
+	}
+
+	// for version V2
+	return getOrDecodeArgsV2[*RebaseAutoIDArgs](job)
+}
+
+// ModifyTableCommentArgs is the arguments for ActionModifyTableComment ddl.
+type ModifyTableCommentArgs struct {
+	Comment string `json:"comment,omitempty"`
+}
+
+func (a *ModifyTableCommentArgs) fillJob(job *Job) {
+	if job.Version == JobVersion1 {
+		job.Args = []any{a.Comment}
+	} else {
+		job.Args = []any{a}
+	}
+}
+
+// GetModifyTableCommentArgs gets the args for ActionModifyTableComment.
+func GetModifyTableCommentArgs(job *Job) (*ModifyTableCommentArgs, error) {
+	if job.Version == JobVersion1 {
+		var comment string
+		if err := job.DecodeArgs(&comment); err != nil {
+			return nil, errors.Trace(err)
+		}
+		return &ModifyTableCommentArgs{
+			Comment: comment,
+		}, nil
+	}
+
+	return getOrDecodeArgsV2[*ModifyTableCommentArgs](job)
+}
+
+// ModifyTableCharsetAndCollateArgs is the arguments for ActionModifyTableCharsetAndCollate ddl.
+type ModifyTableCharsetAndCollateArgs struct {
+	ToCharset          string `json:"to_charset,omitempty"`
+	ToCollate          string `json:"to_collate,omitempty"`
+	NeedsOverwriteCols bool   `json:"needs_overwrite_cols,omitempty"`
+}
+
+func (a *ModifyTableCharsetAndCollateArgs) fillJob(job *Job) {
+	if job.Version == JobVersion1 {
+		job.Args = []any{a.ToCharset, a.ToCollate, a.NeedsOverwriteCols}
+	} else {
+		job.Args = []any{a}
+	}
+}
+
+// GetModifyTableCharsetAndCollateArgs gets the args for ActionModifyTableCharsetAndCollate ddl.
+func GetModifyTableCharsetAndCollateArgs(job *Job) (*ModifyTableCharsetAndCollateArgs, error) {
+	if job.Version == JobVersion1 {
+		args := &ModifyTableCharsetAndCollateArgs{}
+		err := job.DecodeArgs(&args.ToCharset, &args.ToCollate, &args.NeedsOverwriteCols)
+		if err != nil {
+			return nil, errors.Trace(err)
+		}
+		return args, nil
+	}
+
+	return getOrDecodeArgsV2[*ModifyTableCharsetAndCollateArgs](job)
+}
+
+// AlterIndexVisibilityArgs is the arguments for ActionAlterIndexVisibility ddl.
+type AlterIndexVisibilityArgs struct {
+	IndexName pmodel.CIStr `json:"index_name,omitempty"`
+	Invisible bool         `json:"invisible,omitempty"`
+}
+
+func (a *AlterIndexVisibilityArgs) fillJob(job *Job) {
+	if job.Version == JobVersion1 {
+		job.Args = []any{a.IndexName, a.Invisible}
+	} else {
+		job.Args = []any{a}
+	}
+}
+
+// GetAlterIndexVisibilityArgs gets the args for AlterIndexVisibility ddl.
+func GetAlterIndexVisibilityArgs(job *Job) (*AlterIndexVisibilityArgs, error) {
+	if job.Version == JobVersion1 {
+		var (
+			indexName pmodel.CIStr
+			invisible bool
+		)
+		if err := job.DecodeArgs(&indexName, &invisible); err != nil {
+			return nil, errors.Trace(err)
+		}
+		return &AlterIndexVisibilityArgs{
+			IndexName: indexName,
+			Invisible: invisible,
+		}, nil
+	}
+
+	return getOrDecodeArgsV2[*AlterIndexVisibilityArgs](job)
+}
+
+// AddForeignKeyArgs is the arguments for ActionAddForeignKey ddl.
+type AddForeignKeyArgs struct {
+	FkInfo  *FKInfo `json:"fk_info,omitempty"`
+	FkCheck bool    `json:"fk_check,omitempty"`
+}
+
+func (a *AddForeignKeyArgs) fillJob(job *Job) {
+	if job.Version == JobVersion1 {
+		job.Args = []any{a.FkInfo, a.FkCheck}
+	} else {
+		job.Args = []any{a}
+	}
+}
+
+// GetAddForeignKeyArgs get the args for AddForeignKey ddl.
+func GetAddForeignKeyArgs(job *Job) (*AddForeignKeyArgs, error) {
+	if job.Version == JobVersion1 {
+		var (
+			fkInfo  *FKInfo
+			fkCheck bool
+		)
+		if err := job.DecodeArgs(&fkInfo, &fkCheck); err != nil {
+			return nil, errors.Trace(err)
+		}
+		return &AddForeignKeyArgs{
+			FkInfo:  fkInfo,
+			FkCheck: fkCheck,
+		}, nil
+	}
+
+	return getOrDecodeArgsV2[*AddForeignKeyArgs](job)
+}
+
+// DropForeignKeyArgs is the arguments for DropForeignKey ddl.
+type DropForeignKeyArgs struct {
+	FkName pmodel.CIStr `json:"fk_name,omitempty"`
+}
+
+func (a *DropForeignKeyArgs) fillJob(job *Job) {
+	if job.Version == JobVersion1 {
+		job.Args = []any{a.FkName}
+	} else {
+		job.Args = []any{a}
+	}
+}
+
+// GetDropForeignKeyArgs gets the args for DropForeignKey ddl.
+func GetDropForeignKeyArgs(job *Job) (*DropForeignKeyArgs, error) {
+	if job.Version == JobVersion1 {
+		var fkName pmodel.CIStr
+		if err := job.DecodeArgs(&fkName); err != nil {
+			return nil, errors.Trace(err)
+		}
+		return &DropForeignKeyArgs{FkName: fkName}, nil
+	}
+
+	return getOrDecodeArgsV2[*DropForeignKeyArgs](job)
+}
+
 // DropColumnArgs is the arguments of dropping column job.
 type DropColumnArgs struct {
 	ColName  pmodel.CIStr `json:"column_name,omitempty"`
diff --git a/pkg/meta/model/job_args_test.go b/pkg/meta/model/job_args_test.go
index 27f741a746c57..d17c0abca8c33 100644
--- a/pkg/meta/model/job_args_test.go
+++ b/pkg/meta/model/job_args_test.go
@@ -499,6 +499,83 @@ func TestDropColumnArgs(t *testing.T) {
 	}
 }
 
+func TestGetRebaseAutoIDArgs(t *testing.T) {
+	inArgs := &RebaseAutoIDArgs{
+		NewBase: 9527,
+		Force:   true,
+	}
+	for _, tp := range []ActionType{ActionRebaseAutoID, ActionRebaseAutoRandomBase} {
+		for _, v := range []JobVersion{JobVersion1, JobVersion2} {
+			j2 := &Job{}
+			require.NoError(t, j2.Decode(getJobBytes(t, inArgs, v, tp)))
+			args, err := GetRebaseAutoIDArgs(j2)
+			require.NoError(t, err)
+			require.Equal(t, inArgs, args)
+		}
+	}
+}
+
+func TestGetModifyTableCommentArgs(t *testing.T) {
+	inArgs := &ModifyTableCommentArgs{
+		Comment: "TiDB is great",
+	}
+
+	for _, v := range []JobVersion{JobVersion1, JobVersion2} {
+		j2 := &Job{}
+		require.NoError(t, j2.Decode(getJobBytes(t, inArgs, v, ActionModifyTableComment)))
+		args, err := GetModifyTableCommentArgs(j2)
+		require.NoError(t, err)
+		require.Equal(t, inArgs, args)
+	}
+}
+
+func TestGetAlterIndexVisibilityArgs(t *testing.T) {
+	inArgs := &AlterIndexVisibilityArgs{
+		IndexName: model.NewCIStr("index-name"),
+		Invisible: true,
+	}
+
+	for _, v := range []JobVersion{JobVersion1, JobVersion2} {
+		j2 := &Job{}
+		require.NoError(t, j2.Decode(getJobBytes(t, inArgs, v, ActionAlterIndexVisibility)))
+		args, err := GetAlterIndexVisibilityArgs(j2)
+		require.NoError(t, err)
+		require.Equal(t, inArgs, args)
+	}
+}
+
+func TestGetAddForeignKeyArgs(t *testing.T) {
+	inArgs := &AddForeignKeyArgs{
+		FkInfo: &FKInfo{
+			ID:   7527,
+			Name: model.NewCIStr("fk-name"),
+		},
+		FkCheck: true,
+	}
+
+	for _, v := range []JobVersion{JobVersion1, JobVersion2} {
+		j2 := &Job{}
+		require.NoError(t, j2.Decode(getJobBytes(t, inArgs, v, ActionAddForeignKey)))
+		args, err := GetAddForeignKeyArgs(j2)
+		require.NoError(t, err)
+		require.Equal(t, inArgs, args)
+	}
+}
+
+func TestGetDropForeignKeyArgs(t *testing.T) {
+	inArgs := &DropForeignKeyArgs{
+		FkName: model.NewCIStr("fk-name"),
+	}
+
+	for _, v := range []JobVersion{JobVersion1, JobVersion2} {
+		j2 := &Job{}
+		require.NoError(t, j2.Decode(getJobBytes(t, inArgs, v, ActionDropForeignKey)))
+		args, err := GetDropForeignKeyArgs(j2)
+		require.NoError(t, j2.Decode(getJobBytes(t, inArgs, v, ActionDropColumn)))
+		require.NoError(t, err)
+		require.Equal(t, inArgs, args)
+	}
+}
 func TestAddCheckConstraintArgs(t *testing.T) {
 	Constraint :=
 		&ConstraintInfo{