Merge branch 'master' of github.com:pingcap/tidb into issue41434

br/pkg/storage/BUILD.bazel

@@ -69,6 +69,7 @@ go_test(
     ],
     embed = [":storage"],
     flaky = True,
+    shard_count = 41,
     deps = [
         "//br/pkg/mock",
         "@com_github_aws_aws_sdk_go//aws",

br/pkg/storage/s3_test.go

@@ -288,6 +288,9 @@ func TestS3Storage(t *testing.T) {
 		sendCredential bool
 	}
 
+	require.NoError(t, os.Setenv("AWS_ACCESS_KEY_ID", "ab"))
+	require.NoError(t, os.Setenv("AWS_SECRET_ACCESS_KEY", "cd"))
+	require.NoError(t, os.Setenv("AWS_SESSION_TOKEN", "ef"))
 	s := createGetBucketRegionServer("us-west-2", 200, true)
 	defer s.Close()
 
@@ -411,7 +414,17 @@ func TestS3Storage(t *testing.T) {
 }
 
 func TestS3URI(t *testing.T) {
-	backend, err := ParseBackend("s3://bucket/prefix/", nil)
+	accessKey := "ab"
+	secretAccessKey := "cd"
+	sessionToken := "ef"
+	options := &BackendOptions{
+		S3: S3BackendOptions{
+			AccessKey:       accessKey,
+			SecretAccessKey: secretAccessKey,
+			SessionToken:    sessionToken,
+		},
+	}
+	backend, err := ParseBackend("s3://bucket/prefix/", options)
 	require.NoError(t, err)
 	storage, err := New(context.Background(), backend, &ExternalStorageOptions{})
 	require.NoError(t, err)

build/nogo_config.json

@@ -273,7 +273,8 @@
   "ineffassign": {
     "exclude_files": {
       "external/": "no need to vet third party code",
-      ".*_generated\\.go$": "ignore generated code"
+      ".*_generated\\.go$": "ignore generated code",
+      "/cgo/": "no need to vet cgo code"
     }
   },
   "inspect": {

ddl/callback.go

@@ -51,6 +51,8 @@ type Callback interface {
 	OnSchemaStateChanged(schemaVer int64)
 	// OnJobRunBefore is called before running job.
 	OnJobRunBefore(job *model.Job)
+	// OnJobRunAfter is called after running job.
+	OnJobRunAfter(job *model.Job)
 	// OnJobUpdated is called after the running job is updated.
 	OnJobUpdated(job *model.Job)
 	// OnWatched is called after watching owner is completed.
@@ -80,6 +82,11 @@ func (*BaseCallback) OnJobRunBefore(_ *model.Job) {
 	// Nothing to do.
 }
 
+// OnJobRunAfter implements Callback.OnJobRunAfter interface.
+func (*BaseCallback) OnJobRunAfter(_ *model.Job) {
+	// Nothing to do.
+}
+
 // OnJobUpdated implements Callback.OnJobUpdated interface.
 func (*BaseCallback) OnJobUpdated(job *model.Job) {
 	// Nothing to do.

ddl/column.go

@@ -822,7 +822,7 @@ func doReorgWorkForModifyColumn(w *worker, d *ddlCtx, t *meta.Meta, job *model.J
 		return false, ver, errors.Trace(err)
 	}
 	reorgInfo, err := getReorgInfo(d.jobContext(job.ID), d, rh, job, dbInfo, tbl, BuildElements(changingCol, changingIdxs), false)
-	if err != nil || reorgInfo.first {
+	if err != nil || reorgInfo == nil || reorgInfo.first {
 		// If we run reorg firstly, we should update the job snapshot version
 		// and then run the reorg next time.
 		return false, ver, errors.Trace(err)

ddl/ddl.go

@@ -526,6 +526,13 @@ func (dc *ddlCtx) getReorgCtx(jobID int64) *reorgCtx {
 }
 
 func (dc *ddlCtx) newReorgCtx(jobID int64, startKey []byte, currElement *meta.Element, rowCount int64) *reorgCtx {
+	dc.reorgCtx.Lock()
+	defer dc.reorgCtx.Unlock()
+	existedRC, ok := dc.reorgCtx.reorgCtxMap[jobID]
+	if ok {
+		existedRC.references.Add(1)
+		return existedRC
+	}
 	rc := &reorgCtx{}
 	rc.doneCh = make(chan error, 1)
 	// initial reorgCtx
@@ -535,22 +542,10 @@ func (dc *ddlCtx) newReorgCtx(jobID int64, startKey []byte, currElement *meta.El
 	rc.mu.warnings = make(map[errors.ErrorID]*terror.Error)
 	rc.mu.warningsCount = make(map[errors.ErrorID]int64)
 	rc.references.Add(1)
-	dc.reorgCtx.Lock()
-	defer dc.reorgCtx.Unlock()
 	dc.reorgCtx.reorgCtxMap[jobID] = rc
 	return rc
 }
 
-func (dc *ddlCtx) setReorgCtxForBackfill(bfJob *BackfillJob) {
-	rc := dc.getReorgCtx(bfJob.JobID)
-	if rc == nil {
-		ele := &meta.Element{ID: bfJob.EleID, TypeKey: bfJob.EleKey}
-		dc.newReorgCtx(bfJob.JobID, bfJob.Meta.StartKey, ele, bfJob.Meta.RowCount)
-	} else {
-		rc.references.Add(1)
-	}
-}
-
 func (dc *ddlCtx) removeReorgCtx(jobID int64) {
 	dc.reorgCtx.Lock()
 	defer dc.reorgCtx.Unlock()

ddl/ddl_test.go

@@ -44,7 +44,6 @@ type DDLForTest interface {
 	// SetInterceptor sets the interceptor.
 	SetInterceptor(h Interceptor)
 	NewReorgCtx(jobID int64, startKey []byte, currElement *meta.Element, rowCount int64) *reorgCtx
-	SetReorgCtxForBackfill(bfJob *BackfillJob)
 	GetReorgCtx(jobID int64) *reorgCtx
 	RemoveReorgCtx(id int64)
 }
@@ -67,11 +66,6 @@ func (d *ddl) NewReorgCtx(jobID int64, startKey []byte, currElement *meta.Elemen
 	return d.newReorgCtx(jobID, startKey, currElement, rowCount)
 }
 
-// SetReorgCtxForBackfill exports for testing.
-func (d *ddl) SetReorgCtxForBackfill(bfJob *BackfillJob) {
-	d.setReorgCtxForBackfill(bfJob)
-}
-
 // GetReorgCtx exports for testing.
 func (d *ddl) GetReorgCtx(jobID int64) *reorgCtx {
 	return d.getReorgCtx(jobID)

ddl/ddl_worker.go

@@ -770,6 +770,10 @@ func (w *worker) HandleDDLJobTable(d *ddlCtx, job *model.Job) (int64, error) {
 	// and retry later if the job is not cancelled.
 	schemaVer, runJobErr = w.runDDLJob(d, t, job)
 
+	d.mu.RLock()
+	d.mu.hook.OnJobRunAfter(job)
+	d.mu.RUnlock()
+
 	if job.IsCancelled() {
 		defer d.unlockSchemaVersion(job.ID)
 		w.sess.reset()

ddl/ddl_worker_test.go

@@ -317,10 +317,10 @@ func TestUsingReorgCtx(t *testing.T) {
 	wg := util.WaitGroupWrapper{}
 	wg.Run(func() {
 		jobID := int64(1)
-		m := &model.BackfillMeta{StartKey: []byte("skey"), RowCount: 1}
-		bfJob := &ddl.BackfillJob{JobID: jobID, EleID: 1, EleKey: nil, Meta: m}
-		for i := 0; i < 100; i++ {
-			d.(ddl.DDLForTest).SetReorgCtxForBackfill(bfJob)
+		startKey := []byte("skey")
+		ele := &meta.Element{ID: 1, TypeKey: nil}
+		for i := 0; i < 500; i++ {
+			d.(ddl.DDLForTest).NewReorgCtx(jobID, startKey, ele, 0)
 			d.(ddl.DDLForTest).GetReorgCtx(jobID).IsReorgCanceled()
 			d.(ddl.DDLForTest).RemoveReorgCtx(jobID)
 		}
@@ -329,7 +329,7 @@ func TestUsingReorgCtx(t *testing.T) {
 		jobID := int64(1)
 		startKey := []byte("skey")
 		ele := &meta.Element{ID: 1, TypeKey: nil}
-		for i := 0; i < 100; i++ {
+		for i := 0; i < 500; i++ {
 			d.(ddl.DDLForTest).NewReorgCtx(jobID, startKey, ele, 0)
 			d.(ddl.DDLForTest).GetReorgCtx(jobID).IsReorgCanceled()
 			d.(ddl.DDLForTest).RemoveReorgCtx(jobID)

ddl/index.go

@@ -983,7 +983,7 @@ func runReorgJobAndHandleErr(w *worker, d *ddlCtx, t *meta.Meta, job *model.Job,
 		return false, ver, errors.Trace(err)
 	}
 	reorgInfo, err := getReorgInfo(d.jobContext(job.ID), d, rh, job, dbInfo, tbl, elements, mergingTmpIdx)
-	if err != nil || reorgInfo.first {
+	if err != nil || reorgInfo == nil || reorgInfo.first {
 		// If we run reorg firstly, we should update the job snapshot version
 		// and then run the reorg next time.
 		return false, ver, errors.Trace(err)

ddl/indexmergetest/merge_test.go

@@ -15,6 +15,7 @@
 package indexmergetest
 
 import (
+	"strconv"
 	"testing"
 	"time"
 
@@ -858,3 +859,41 @@ func TestAddIndexMultipleDelete(t *testing.T) {
 	tk.MustQuery("select * from t;").Check(testkit.Rows())
 	require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/ddl/mockDMLExecution"))
 }
+
+func TestAddIndexDuplicateAndWriteConflict(t *testing.T) {
+	store, dom := testkit.CreateMockStoreAndDomain(t)
+
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("use test")
+	tk.MustExec("create table t(id int primary key, b int);")
+	tk.MustExec("insert into t values (1, 1);")
+
+	tk1 := testkit.NewTestKit(t, store)
+	tk1.MustExec("use test")
+
+	d := dom.DDL()
+	originalCallback := d.GetHook()
+	defer d.SetHook(originalCallback)
+	callback := &callback.TestDDLCallback{}
+	var runCancel bool
+	callback.OnJobRunAfterExported = func(job *model.Job) {
+		if t.Failed() || runCancel {
+			return
+		}
+		switch job.SchemaState {
+		case model.StateWriteOnly:
+			_, err := tk1.Exec("insert into t values (2, 1);")
+			assert.NoError(t, err)
+		}
+		if job.State == model.JobStateRollingback {
+			_, err := tk1.Exec("admin cancel ddl jobs " + strconv.FormatInt(job.ID, 10))
+			assert.NoError(t, err)
+			runCancel = true
+		}
+	}
+	d.SetHook(callback)
+
+	tk.MustGetErrCode("alter table t add unique index idx(b);", errno.ErrCancelledDDLJob)
+	tk.MustExec("admin check table t;")
+	tk.MustQuery("select * from t;").Check(testkit.Rows("1 1", "2 1"))
+}

ddl/internal/callback/callback.go

@@ -51,6 +51,7 @@ type TestDDLCallback struct {
 
 	onJobRunBefore          func(*model.Job)
 	OnJobRunBeforeExported  func(*model.Job)
+	OnJobRunAfterExported   func(*model.Job)
 	onJobUpdated            func(*model.Job)
 	OnJobUpdatedExported    atomic.Pointer[func(*model.Job)]
 	onWatched               func(ctx context.Context)
@@ -103,6 +104,17 @@ func (tc *TestDDLCallback) OnJobRunBefore(job *model.Job) {
 	tc.BaseCallback.OnJobRunBefore(job)
 }
 
+// OnJobRunAfter is used to run the user customized logic of `OnJobRunAfter` first.
+func (tc *TestDDLCallback) OnJobRunAfter(job *model.Job) {
+	logutil.BgLogger().Info("on job run after", zap.String("job", job.String()))
+	if tc.OnJobRunAfterExported != nil {
+		tc.OnJobRunAfterExported(job)
+		return
+	}
+
+	tc.BaseCallback.OnJobRunAfter(job)
+}
+
 // OnJobUpdated is used to run the user customized logic of `OnJobUpdated` first.
 func (tc *TestDDLCallback) OnJobUpdated(job *model.Job) {
 	logutil.BgLogger().Info("on job updated", zap.String("job", job.String()))

ddl/job_table.go

@@ -402,7 +402,7 @@ func (d *ddl) loadBackfillJobAndRun() {
 		return
 	}
 	// TODO: Adjust how the non-owner uses ReorgCtx.
-	d.setReorgCtxForBackfill(bfJob)
+	d.newReorgCtx(bfJob.JobID, bfJob.Meta.StartKey, &meta.Element{ID: bfJob.EleID, TypeKey: bfJob.EleKey}, bfJob.Meta.RowCount)
 	d.wg.Run(func() {
 		defer func() {
 			tidbutil.Recover(metrics.LabelDistReorg, "runBackfillJobs", nil, false)

ddl/reorg.go

@@ -683,6 +683,9 @@ func getReorgInfo(ctx *JobContext, d *ddlCtx, rh *reorgHandler, job *model.Job,
 			if meta.ErrDDLReorgElementNotExist.Equal(err) {
 				job.SnapshotVer = 0
 				logutil.BgLogger().Warn("[ddl] get reorg info, the element does not exist", zap.String("job", job.String()))
+				if job.IsCancelling() {
+					return nil, nil
+				}
 			}
 			return &info, errors.Trace(err)
 		}