ddl: fix reorg handle not resumed after changing DDL owner

pkg/ddl/backfilling.go

@@ -510,6 +510,7 @@ func loadTableRanges(
 		zap.String("range start", hex.EncodeToString(ranges[0].StartKey)),
 		zap.String("range end", hex.EncodeToString(ranges[len(ranges)-1].EndKey)),
 		zap.Int("range count", len(ranges)))
+	failpoint.InjectCall("afterLoadTableRanges", len(ranges))
 	return ranges, nil
 }
 
@@ -845,12 +846,20 @@ func (dc *ddlCtx) writePhysicalTableRecord(
 	t table.PhysicalTable,
 	bfWorkerType backfillerType,
 	reorgInfo *reorgInfo,
-) error {
+) (err error) {
 	startKey, endKey := reorgInfo.StartKey, reorgInfo.EndKey
 
 	if err := dc.isReorgRunnable(reorgInfo.Job.ID, false); err != nil {
 		return errors.Trace(err)
 	}
+	defer func() {
+		if err != nil && ctx.Err() != nil {
+			// Prevent "not owner" error being overwritten by context.Canceled.
+			if !dc.isOwner() {
+				err = dbterror.ErrNotOwner
+			}
+		}
+	}()
 
 	failpoint.Inject("MockCaseWhenParseFailure", func(val failpoint.Value) {
 		//nolint:forcetypeassert
@@ -931,6 +940,7 @@ func (dc *ddlCtx) writePhysicalTableRecord(
 							zap.Int64("job ID", reorgInfo.ID),
 							zap.Error(err2))
 					}
+					failpoint.InjectCall("afterUpdateReorgMeta")
 				}
 			}
 		}

pkg/ddl/backfilling_scheduler.go

@@ -375,6 +375,7 @@ func newTaskIDAllocator() *taskIDAllocator {
 }
 
 func (a *taskIDAllocator) alloc() int {
+	ret := a.id
 	a.id++
-	return a.id
+	return ret
 }

pkg/ddl/column_modify_test.go

@@ -635,3 +635,36 @@ func TestColumnTypeChangeGenUniqueChangingName(t *testing.T) {
 
 	tk.MustExec("drop table if exists t")
 }
+
+func TestModifyColumnReorgCheckpoint(t *testing.T) {
+	store, dom := testkit.CreateMockStoreAndDomainWithSchemaLease(t, columnModifyLease)
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("use test")
+	tk2 := testkit.NewTestKit(t, store)
+	tk2.MustExec("use test")
+	tk.MustExec("set global tidb_ddl_reorg_worker_cnt = 1;")
+	tk.MustExec("create table t (a int primary key, b bigint);")
+	rowCnt := 10
+	for i := 0; i < rowCnt; i++ {
+		tk.MustExec(fmt.Sprintf("insert into t values (%d, %d)", i*10000, i*10000))
+	}
+	splitTableSQL := fmt.Sprintf("split table t between (0) and (%d*10000) regions %d;", rowCnt, rowCnt)
+	tk.MustQuery(splitTableSQL).Check(testkit.Rows(fmt.Sprintf("%d 1", rowCnt-1)))
+
+	retireOwner := false
+	testfailpoint.EnableCall(t, "github.com/pingcap/tidb/pkg/ddl/afterUpdateReorgMeta", func() {
+		if !retireOwner {
+			retireOwner = true
+			dom.DDL().OwnerManager().ResignOwner(context.Background())
+		}
+	})
+
+	rangeCnts := []int{}
+	testfailpoint.EnableCall(t, "github.com/pingcap/tidb/pkg/ddl/afterLoadTableRanges", func(rangeCnt int) {
+		rangeCnts = append(rangeCnts, rangeCnt)
+	})
+
+	tk.MustExec("alter table t modify column b int;")
+	require.Len(t, rangeCnts, 2)                // It should have two rounds for loading table ranges.
+	require.Less(t, rangeCnts[1], rangeCnts[0]) // Verify if the checkpoint is progressing.
+}