ddl: roll back txn before handling ErrEntryTooLarge (#43867) (#43878)

close #43725

ddl/db_test.go

@@ -979,6 +979,7 @@ func TestDDLJobErrorCount(t *testing.T) {
 	require.NotNil(t, historyJob)
 	require.Equal(t, int64(1), historyJob.ErrorCount)
 	require.True(t, kv.ErrEntryTooLarge.Equal(historyJob.Error))
+	tk.MustQuery("select * from ddl_error_table;").Check(testkit.Rows())
 }
 
 func TestCommitTxnWithIndexChange(t *testing.T) {

ddl/ddl_worker.go

@@ -451,12 +451,16 @@ func (w *worker) getFirstDDLJob(t *meta.Meta) (*model.Job, error) {
 }
 
 // handleUpdateJobError handles the too large DDL job.
-func (w *worker) handleUpdateJobError(t *meta.Meta, job *model.Job, err error) error {
+func (w *worker) handleUpdateJobError(t *meta.Meta, job *model.Job, err error, txn kv.Transaction) error {
 	if err == nil {
 		return nil
 	}
 	if kv.ErrEntryTooLarge.Equal(err) {
 		logutil.Logger(w.logCtx).Warn("[ddl] update DDL job failed", zap.String("job", job.String()), zap.Error(err))
+		err1 := w.rollbackOrReset(txn)
+		if err1 != nil {
+			return errors.Trace(err1)
+		}
 		// Reduce this txn entry size.
 		job.BinlogInfo.Clean()
 		job.Error = toTError(err)
@@ -468,6 +472,15 @@ func (w *worker) handleUpdateJobError(t *meta.Meta, job *model.Job, err error) e
 	return errors.Trace(err)
 }
 
+func (w *worker) rollbackOrReset(txn kv.Transaction) error {
+	if w.concurrentDDL {
+		w.sess.rollback()
+		return w.sess.begin()
+	}
+	txn.Reset()
+	return nil
+}
+
 // updateDDLJob updates the DDL job information.
 // Every time we enter another state except final state, we must call this function.
 func (w *worker) updateDDLJob(t *meta.Meta, job *model.Job, meetErr bool) error {
@@ -865,7 +878,7 @@ func (w *worker) HandleDDLJobTable(d *ddlCtx, job *model.Job) (int64, error) {
 		return 0, err
 	}
 	err = w.updateDDLJob(t, job, runJobErr != nil)
-	if err = w.handleUpdateJobError(t, job, err); err != nil {
+	if err = w.handleUpdateJobError(t, job, err, txn); err != nil {
 		w.sess.rollback()
 		d.unlockSchemaVersion(job.ID)
 		return 0, err
@@ -1004,7 +1017,7 @@ func (w *worker) handleDDLJobQueue(d *ddlCtx) error {
 				schemaVer = 0
 			}
 			err = w.updateDDLJob(t, job, runJobErr != nil)
-			if err = w.handleUpdateJobError(t, job, err); err != nil {
+			if err = w.handleUpdateJobError(t, job, err, txn); err != nil {
 				return errors.Trace(err)
 			}
 			writeBinlog(d.binlogCli, txn, job)