Merge branch 'master' into model2-ut

ddl/callback.go

@@ -48,7 +48,7 @@ type Callback interface {
  // OnChanged is called after a ddl statement is finished.
  OnChanged(err error) error
  // OnSchemaStateChanged is called after a schema state is changed.
- OnSchemaStateChanged()
+ OnSchemaStateChanged(schemaVer int64)
  // OnJobRunBefore is called before running job.
  OnJobRunBefore(job *model.Job)
  // OnJobUpdated is called after the running job is updated.
@@ -71,7 +71,7 @@ func (*BaseCallback) OnChanged(err error) error {
 }
 
 // OnSchemaStateChanged implements Callback interface.
-func (*BaseCallback) OnSchemaStateChanged() {
+func (*BaseCallback) OnSchemaStateChanged(schemaVer int64) {
  // Nothing to do.
 }
 
@@ -129,7 +129,7 @@ func (c *DefaultCallback) OnChanged(err error) error {
 }
 
 // OnSchemaStateChanged overrides the ddl Callback interface.
-func (c *DefaultCallback) OnSchemaStateChanged() {
+func (c *DefaultCallback) OnSchemaStateChanged(schemaVer int64) {
  err := c.do.Reload()
  if err != nil {
  logutil.BgLogger().Error("domain callback failed on schema state changed", zap.Error(err))
@@ -166,7 +166,7 @@ func (c *ctcCallback) OnChanged(err error) error {
 }
 
 // OnSchemaStateChanged overrides the ddl Callback interface.
-func (c *ctcCallback) OnSchemaStateChanged() {
+func (c *ctcCallback) OnSchemaStateChanged(retVer int64) {
  err := c.do.Reload()
  if err != nil {
  logutil.BgLogger().Error("domain callback failed on schema state changed", zap.Error(err))

ddl/callback_test.go

@@ -48,13 +48,14 @@ type TestDDLCallback struct {
  // domain to reload schema before your ddl stepping into the next state change.
  Do DomainReloader
 
- onJobRunBefore func(*model.Job)
- OnJobRunBeforeExported func(*model.Job)
- onJobUpdated func(*model.Job)
- OnJobUpdatedExported atomic.Pointer[func(*model.Job)]
- onWatched func(ctx context.Context)
- OnGetJobBeforeExported func(string)
- OnGetJobAfterExported func(string, *model.Job)
+ onJobRunBefore func(*model.Job)
+ OnJobRunBeforeExported func(*model.Job)
+ onJobUpdated func(*model.Job)
+ OnJobUpdatedExported atomic.Pointer[func(*model.Job)]
+ onWatched func(ctx context.Context)
+ OnGetJobBeforeExported func(string)
+ OnGetJobAfterExported func(string, *model.Job)
+ OnJobSchemaStateChanged func(int64)
 }
 
 // OnChanged mock the same behavior with the main DDL hook.
@@ -73,12 +74,17 @@ func (tc *TestDDLCallback) OnChanged(err error) error {
 }
 
 // OnSchemaStateChanged mock the same behavior with the main ddl hook.
-func (tc *TestDDLCallback) OnSchemaStateChanged() {
+func (tc *TestDDLCallback) OnSchemaStateChanged(schemaVer int64) {
  if tc.Do != nil {
  if err := tc.Do.Reload(); err != nil {
  logutil.BgLogger().Warn("reload failed on schema state changed", zap.Error(err))
  }
  }
+
+ if tc.OnJobSchemaStateChanged != nil {
+ tc.OnJobSchemaStateChanged(schemaVer)
+ return
+ }
 }
 
 // OnJobRunBefore is used to run the user customized logic of `onJobRunBefore` first.

ddl/ddl_worker.go

@@ -1028,7 +1028,7 @@ func (w *worker) handleDDLJobQueue(d *ddlCtx) error {
  if RunInGoTest {
  // d.mu.hook is initialed from domain / test callback, which will force the owner host update schema diff synchronously.
  d.mu.RLock()
- d.mu.hook.OnSchemaStateChanged()
+ d.mu.hook.OnSchemaStateChanged(schemaVer)
  d.mu.RUnlock()
  }
 

ddl/job_table.go

@@ -293,7 +293,7 @@ func (d *ddl) delivery2worker(wk *worker, pool *workerPool, job *model.Job) {
  if RunInGoTest {
  // d.mu.hook is initialed from domain / test callback, which will force the owner host update schema diff synchronously.
  d.mu.RLock()
- d.mu.hook.OnSchemaStateChanged()
+ d.mu.hook.OnSchemaStateChanged(schemaVer)
  d.mu.RUnlock()
  }
 

ddl/multi_schema_change.go

@@ -118,11 +118,13 @@ func onMultiSchemaChange(w *worker, d *ddlCtx, t *meta.Meta, job *model.Job) (ve
  proxyJob := sub.ToProxyJob(job)
  if schemaVersionGenerated {
  proxyJob.MultiSchemaInfo.SkipVersion = true
- } else {
+ }
+ proxyJobVer, err := w.runDDLJob(d, t, &proxyJob)
+ if !schemaVersionGenerated && proxyJobVer != 0 {
  schemaVersionGenerated = true
+ ver = proxyJobVer
  }
- ver, err = w.runDDLJob(d, t, &proxyJob)
- sub.FromProxyJob(&proxyJob, ver)
+ sub.FromProxyJob(&proxyJob, proxyJobVer)
  if err != nil || proxyJob.Error != nil {
  for j := i - 1; j >= 0; j-- {
  job.MultiSchemaInfo.SubJobs[j] = &subJobs[j]
@@ -376,8 +378,8 @@ func finishMultiSchemaJob(job *model.Job, t *meta.Meta) (ver int64, err error) {
  }
  tblInfo, err := t.GetTable(job.SchemaID, job.TableID)
  if err != nil {
- return ver, err
+ return 0, err
  }
  job.FinishTableJob(model.JobStateDone, model.StateNone, ver, tblInfo)
- return ver, err
+ return 0, err
 }

ddl/multi_schema_change_test.go

@@ -1141,6 +1141,33 @@ func TestMultiSchemaChangeUnsupportedType(t *testing.T) {
  "[ddl:8200]Unsupported multi schema change for modify auto id cache")
 }
 
+func TestMultiSchemaChangeSchemaVersion(t *testing.T) {
+ store, dom := testkit.CreateMockStoreAndDomain(t)
+ tk := testkit.NewTestKit(t, store)
+ tk.MustExec("use test;")
+ tk.MustExec("create table t(a int, b int, c int, d int)")
+ tk.MustExec("insert into t values (1,2,3,4)")
+
+ schemaVerMap := map[int64]struct{}{}
+
+ originHook := dom.DDL().GetHook()
+ hook := &ddl.TestDDLCallback{Do: dom}
+ hook.OnJobSchemaStateChanged = func(schemaVer int64) {
+ if schemaVer != 0 {
+ // No same return schemaVer during multi-schema change
+ _, ok := schemaVerMap[schemaVer]
+ assert.False(t, ok)
+ schemaVerMap[schemaVer] = struct{}{}
+ }
+ }
+ dom.DDL().SetHook(hook)
+ tk.MustExec("alter table t drop column b, drop column c")
+ tk.MustExec("alter table t add column b int, add column c int")
+ tk.MustExec("alter table t add index k(b), add column e int")
+ tk.MustExec("alter table t alter index k invisible, drop column e")
+ dom.DDL().SetHook(originHook)
+}
+
 func TestMultiSchemaChangeMixedWithUpdate(t *testing.T) {
  store, dom := testkit.CreateMockStoreAndDomain(t)
  tk := testkit.NewTestKit(t, store)

util/gctuner/tuner_test.go

@@ -25,7 +25,7 @@ var testHeap []byte
 
 func TestTuner(t *testing.T) {
  EnableGOGCTuner.Store(true)
- memLimit := uint64(100 * 1024 * 1024) //100 MB
+ memLimit := uint64(1000 * 1024 * 1024) //1000 MB
  threshold := memLimit / 2
  tn := newTuner(threshold)
  require.Equal(t, threshold, tn.threshold.Load())
@@ -44,17 +44,17 @@ func TestTuner(t *testing.T) {
  testHeap = make([]byte, threshold/4)
  for i := 0; i < 100; i++ {
  runtime.GC()
- require.GreaterOrEqual(t, tn.getGCPercent(), uint32(100))
- require.LessOrEqual(t, tn.getGCPercent(), uint32(500))
+ require.GreaterOrEqual(t, tn.getGCPercent(), MaxGCPercent/2)
+ require.LessOrEqual(t, tn.getGCPercent(), MaxGCPercent)
  }
 
  // 1/2 threshold
  testHeap = make([]byte, threshold/2)
  runtime.GC()
  for i := 0; i < 100; i++ {
  runtime.GC()
- require.GreaterOrEqual(t, tn.getGCPercent(), uint32(50))
- require.LessOrEqual(t, tn.getGCPercent(), uint32(100))
+ require.GreaterOrEqual(t, tn.getGCPercent(), MinGCPercent)
+ require.LessOrEqual(t, tn.getGCPercent(), MaxGCPercent/2)
  }
 
  // 3/4 threshold