ddl: fix rawArgs is overwritten (#5531)

ddl/ddl_worker.go

@@ -123,10 +123,16 @@ func (d *ddl) handleUpdateJobError(t *meta.Meta, job *model.Job, err error) erro
 
 // updateDDLJob updates the DDL job information.
 // Every time we enter another state except final state, we must call this function.
-func (d *ddl) updateDDLJob(t *meta.Meta, job *model.Job, updateTS uint64) error {
+func (d *ddl) updateDDLJob(t *meta.Meta, job *model.Job, updateTS uint64, meetErr bool) error {
 	job.LastUpdateTS = int64(updateTS)
-	err := t.UpdateDDLJob(0, job)
-	return errors.Trace(err)
+	updateRawArgs := true
+	// If there is an error when running job and the RawArgs hasn't been decoded by DecodeArgs,
+	// so we shouldn't replace RawArgs with the marshaling Args.
+	if meetErr && (job.RawArgs != nil && job.Args == nil) {
+		log.Infof("[ddl] update DDL Job %s shouldn't update raw args", job)
+		updateRawArgs = false
+	}
+	return errors.Trace(t.UpdateDDLJob(0, job, updateRawArgs))
 }
 
 // finishDDLJob deletes the finished DDL job in the ddl queue and puts it to history queue.
@@ -211,12 +217,12 @@ func (d *ddl) handleDDLJobQueue() error {
 
 			// If running job meets error, we will save this error in job Error
 			// and retry later if the job is not cancelled.
-			schemaVer = d.runDDLJob(t, job)
+			schemaVer, err = d.runDDLJob(t, job)
 			if job.IsCancelled() {
 				err = d.finishDDLJob(t, job)
 				return errors.Trace(err)
 			}
-			err = d.updateDDLJob(t, job, txn.StartTS())
+			err = d.updateDDLJob(t, job, txn.StartTS(), err != nil)
 			return errors.Trace(d.handleUpdateJobError(t, job, err))
 		})
 		if err != nil {
@@ -249,8 +255,8 @@ func chooseLeaseTime(t, max time.Duration) time.Duration {
 	return t
 }
 
-// runDDLJob runs a DDL job. It returns the current schema version in this transaction.
-func (d *ddl) runDDLJob(t *meta.Meta, job *model.Job) (ver int64) {
+// runDDLJob runs a DDL job. It returns the current schema version in this transaction and the error.
+func (d *ddl) runDDLJob(t *meta.Meta, job *model.Job) (ver int64, err error) {
 	log.Infof("[ddl] run DDL job %s", job)
 	if job.IsFinished() {
 		return
@@ -273,7 +279,6 @@ func (d *ddl) runDDLJob(t *meta.Meta, job *model.Job) (ver int64) {
 		job.State = model.JobRunning
 	}
 
-	var err error
 	switch job.Type {
 	case model.ActionCreateSchema:
 		ver, err = d.onCreateSchema(t, job)

inspectkv/inspectkv.go

@@ -101,7 +101,7 @@ func CancelJobs(txn kv.Transaction, ids []int64) ([]error, error) {
 				errs[i] = errors.Trace(err)
 				continue
 			}
-			err = t.UpdateDDLJob(int64(j), job)
+			err = t.UpdateDDLJob(int64(j), job, true)
 			if err != nil {
 				errs[i] = errors.Trace(err)
 			}

meta/meta.go

@@ -486,17 +486,20 @@ func (m *Meta) GetDDLJob(index int64) (*model.Job, error) {
 	return job, errors.Trace(err)
 }
 
-func (m *Meta) updateDDLJob(index int64, job *model.Job, key []byte) error {
-	b, err := job.Encode(true)
+// updateDDLJob updates the DDL job with index and key.
+// updateRawArgs is used to determine whether to update the raw args when encode the job.
+func (m *Meta) updateDDLJob(index int64, job *model.Job, key []byte, updateRawArgs bool) error {
+	b, err := job.Encode(updateRawArgs)
 	if err != nil {
 		return errors.Trace(err)
 	}
 	return m.txn.LSet(key, index, b)
 }
 
 // UpdateDDLJob updates the DDL job with index.
-func (m *Meta) UpdateDDLJob(index int64, job *model.Job) error {
-	return m.updateDDLJob(index, job, mDDLJobListKey)
+// updateRawArgs is used to determine whether to update the raw args when encode the job.
+func (m *Meta) UpdateDDLJob(index int64, job *model.Job, updateRawArgs bool) error {
+	return m.updateDDLJob(index, job, mDDLJobListKey, updateRawArgs)
 }
 
 // DDLJobQueueLen returns the DDL job queue length.

meta/meta_test.go

@@ -280,7 +280,7 @@ func (s *testSuite) TestDDL(c *C) {
 	c.Assert(err, IsNil)
 	c.Assert(v, IsNil)
 	job.ID = 2
-	err = t.UpdateDDLJob(0, job)
+	err = t.UpdateDDLJob(0, job, true)
 	c.Assert(err, IsNil)
 
 	err = t.UpdateDDLReorgHandle(job, 1)