Release 6.5.11 hotfix 20260410

config/config.go

@@ -501,9 +501,10 @@ type Instance struct {
 	PluginDir                  string     `toml:"plugin_dir" json:"plugin_dir"`
 	PluginLoad                 string     `toml:"plugin_load" json:"plugin_load"`
 	// MaxConnections is the maximum permitted number of simultaneous client connections.
-	MaxConnections    uint32     `toml:"max_connections" json:"max_connections"`
-	TiDBEnableDDL     AtomicBool `toml:"tidb_enable_ddl" json:"tidb_enable_ddl"`
-	TiDBRCReadCheckTS bool       `toml:"tidb_rc_read_check_ts" json:"tidb_rc_read_check_ts"`
+	MaxConnections       uint32     `toml:"max_connections" json:"max_connections"`
+	TiDBEnableDDL        AtomicBool `toml:"tidb_enable_ddl" json:"tidb_enable_ddl"`
+	TiDBEnableStatsOwner AtomicBool `toml:"tidb_enable_stats_owner" json:"tidb_enable_stats_owner"`
+	TiDBRCReadCheckTS    bool       `toml:"tidb_rc_read_check_ts" json:"tidb_rc_read_check_ts"`
 }
 
 func (l *Log) getDisableTimestamp() bool {
@@ -903,6 +904,7 @@ var defaultConf = Config{
 		PluginLoad:                  "",
 		MaxConnections:              0,
 		TiDBEnableDDL:               *NewAtomicBool(true),
+		TiDBEnableStatsOwner:        *NewAtomicBool(true),
 		TiDBRCReadCheckTS:           false,
 	},
 	Status: Status{

config/config_test.go

@@ -1041,7 +1041,7 @@ func TestConflictInstanceConfig(t *testing.T) {
 	_, err = f.WriteString("check-mb4-value-in-utf8 = true \nrun-ddl = true \n" +
 		"[log] \nenable-slow-log = true \n" +
 		"[performance] \nforce-priority = \"NO_PRIORITY\"\n" +
-		"[instance] \ntidb_check_mb4_value_in_utf8 = false \ntidb_enable_slow_log = false \ntidb_force_priority = \"LOW_PRIORITY\"\ntidb_enable_ddl = false")
+		"[instance] \ntidb_check_mb4_value_in_utf8 = false \ntidb_enable_slow_log = false \ntidb_force_priority = \"LOW_PRIORITY\"\ntidb_enable_ddl = false\ntidb_enable_stats_owner = false")
 	require.NoError(t, err)
 	require.NoError(t, f.Sync())
 	err = conf.Load(configFile)
@@ -1055,6 +1055,7 @@ func TestConflictInstanceConfig(t *testing.T) {
 	require.Equal(t, "LOW_PRIORITY", conf.Instance.ForcePriority)
 	require.Equal(t, true, conf.RunDDL)
 	require.Equal(t, false, conf.Instance.TiDBEnableDDL.Load())
+	require.Equal(t, false, conf.Instance.TiDBEnableStatsOwner.Load())
 	require.Equal(t, 0, len(DeprecatedOptions))
 	for _, conflictOption := range ConflictOptions {
 		expectedConflictOption, ok := expectedConflictOptions[conflictOption.SectionName]

ddl/BUILD.bazel

@@ -35,6 +35,7 @@ go_library(
         "multi_schema_change.go",
         "options.go",
         "partition.go",
+        "pkdb_force_merge.go",
         "placement_policy.go",
         "reorg.go",
         "rollingback.go",
@@ -186,6 +187,7 @@ go_test(
         "multi_schema_change_test.go",
         "options_test.go",
         "partition_test.go",
+        "pkdb_force_merge_test.go",
         "placement_policy_ddl_test.go",
         "placement_policy_test.go",
         "placement_sql_test.go",

ddl/db_partition_test.go

@@ -4912,3 +4912,47 @@ func TestIssue59047(t *testing.T) {
 	tk.MustExec(`alter table t add column d date not null`)
 	tk.MustExec(`update t set name = 'x'`)
 }
+
+// Test for issue 64176.
+func TestExchangeTiDBRowID(t *testing.T) {
+	store := testkit.CreateMockStore(t)
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("use test")
+	tk.MustExec(`create table t (a int, b int, primary key (a) nonclustered)`)
+	tk.MustExec(`create table tp (a int, b int, primary key (a) nonclustered) partition by hash (a) partitions 2`)
+
+	tk.MustExec(`insert into t values (2,2),(4,4),(6,6)`)
+	tk.MustExec(`insert into tp values (2,2),(4,4),(6,6)`)
+	tk.MustExec(`insert into t select a + 8, b + 8 from t`)
+
+	tk.MustQuery(`select *, _tidb_rowid from t`).Sort().Check(testkit.Rows(""+
+		"10 10 4",
+		"12 12 5",
+		"14 14 6",
+		"2 2 1",
+		"4 4 2",
+		"6 6 3"))
+	tk.MustQuery(`select *, _tidb_rowid from tp`).Sort().Check(testkit.Rows(""+
+		"2 2 1",
+		"4 4 2",
+		"6 6 3"))
+
+	tk.MustExec(`alter table tp exchange partition p0 with table t`)
+	tk.MustExec(`insert into t values (8,8)`)
+	// This failed before, since it will use _tidb_rowid = 4
+	tk.MustExec(`insert into tp values (8,8)`)
+
+	tk.MustQuery(`select *, _tidb_rowid from tp`).Sort().Check(testkit.Rows(""+
+		"10 10 4",
+		"12 12 5",
+		"14 14 6",
+		"2 2 1",
+		"4 4 2",
+		"6 6 3",
+		"8 8 5001"))
+	tk.MustQuery(`select *, _tidb_rowid from t`).Sort().Check(testkit.Rows(""+
+		"2 2 1",
+		"4 4 2",
+		"6 6 3",
+		"8 8 5001"))
+}

ddl/ddl.go

@@ -769,6 +769,9 @@ func (d *ddl) Start(ctxPool *pools.ResourcePool) error {
 	d.wg.Run(d.PollTiFlashRoutine)
 
 	ingest.InitGlobalLightningEnv()
+	d.ownerManager.SetRetireOwnerHook(func() {
+		d.runningJobs = newRunningJobs()
+	})
 
 	return nil
 }
@@ -1329,12 +1332,9 @@ func (d *ddl) wait4Switch(ctx context.Context) error {
 			return ctx.Err()
 		default:
 		}
-		d.runningJobs.RLock()
-		if len(d.runningJobs.ids) == 0 {
-			d.runningJobs.RUnlock()
+		if len(d.runningJobs.allIDs()) == 0 {
 			return nil
 		}
-		d.runningJobs.RUnlock()
 		time.Sleep(time.Second * 1)
 	}
 }

ddl/ddl_running_jobs.go

@@ -28,57 +28,96 @@ import (
 
 type runningJobs struct {
 	sync.RWMutex
-	ids           map[int64]struct{}
-	runningSchema map[string]map[string]struct{} // database -> table -> struct{}
-	runningJobIDs string
+	// processingIDs records the IDs of the jobs that are being processed by a worker.
+	processingIDs    map[int64]struct{}
+	processingIDsStr string
+
+	// unfinishedIDs records the IDs of the jobs that are not finished yet.
+	// It is not necessarily being processed by a worker.
+	unfinishedIDs    map[int64]struct{}
+	unfinishedSchema map[string]map[string]struct{} // database -> table -> struct{}
 }
 
 func newRunningJobs() *runningJobs {
 	return &runningJobs{
-		ids:           make(map[int64]struct{}),
-		runningSchema: make(map[string]map[string]struct{}),
+		processingIDs:    make(map[int64]struct{}),
+		unfinishedSchema: make(map[string]map[string]struct{}),
+		unfinishedIDs:    make(map[int64]struct{}),
 	}
 }
 
 func (j *runningJobs) add(job *model.Job) {
 	j.Lock()
 	defer j.Unlock()
-	j.ids[job.ID] = struct{}{}
+	j.processingIDs[job.ID] = struct{}{}
 	j.updateInternalRunningJobIDs()
+
+	if _, ok := j.unfinishedIDs[job.ID]; ok {
+		// Already exists, no need to add it again.
+		return
+	}
+	j.unfinishedIDs[job.ID] = struct{}{}
 	for _, info := range job.GetInvolvingSchemaInfo() {
-		if _, ok := j.runningSchema[info.Database]; !ok {
-			j.runningSchema[info.Database] = make(map[string]struct{})
+		if _, ok := j.unfinishedSchema[info.Database]; !ok {
+			j.unfinishedSchema[info.Database] = make(map[string]struct{})
 		}
-		j.runningSchema[info.Database][info.Table] = struct{}{}
+		j.unfinishedSchema[info.Database][info.Table] = struct{}{}
 	}
 }
 
 func (j *runningJobs) remove(job *model.Job) {
 	j.Lock()
 	defer j.Unlock()
-	delete(j.ids, job.ID)
+	delete(j.processingIDs, job.ID)
 	j.updateInternalRunningJobIDs()
-	for _, info := range job.GetInvolvingSchemaInfo() {
-		if db, ok := j.runningSchema[info.Database]; ok {
-			delete(db, info.Table)
+
+	if job.IsFinished() || job.IsSynced() {
+		delete(j.unfinishedIDs, job.ID)
+		for _, info := range job.GetInvolvingSchemaInfo() {
+			if db, ok := j.unfinishedSchema[info.Database]; ok {
+				delete(db, info.Table)
+			}
+			if len(j.unfinishedSchema[info.Database]) == 0 {
+				delete(j.unfinishedSchema, info.Database)
+			}
 		}
-		if len(j.runningSchema[info.Database]) == 0 {
-			delete(j.runningSchema, info.Database)
+	}
+}
+
+func (j *runningJobs) allIDs() string {
+	j.RLock()
+	defer j.RUnlock()
+	return j.processingIDsStr
+}
+
+func (j *runningJobs) updateInternalRunningJobIDs() {
+	var sb strings.Builder
+	i := 0
+	for id := range j.processingIDs {
+		sb.WriteString(strconv.Itoa(int(id)))
+		if i != len(j.processingIDs)-1 {
+			sb.WriteString(",")
 		}
+		i++
 	}
+	j.processingIDsStr = sb.String()
 }
 
 func (j *runningJobs) checkRunnable(job *model.Job) bool {
 	j.RLock()
 	defer j.RUnlock()
+	if _, ok := j.processingIDs[job.ID]; ok {
+		// Already processing by a worker. Skip running it again.
+		return false
+	}
 	for _, info := range job.GetInvolvingSchemaInfo() {
-		if _, ok := j.runningSchema[model.InvolvingAll]; ok {
+		if _, ok := j.unfinishedSchema[model.InvolvingAll]; ok {
 			return false
 		}
 		if info.Database == model.InvolvingNone {
 			continue
 		}
-		if tbls, ok := j.runningSchema[info.Database]; ok {
+		if tbls, ok := j.unfinishedSchema[info.Database]; ok {
 			if _, ok := tbls[model.InvolvingAll]; ok {
 				return false
 			}
@@ -92,22 +131,3 @@ func (j *runningJobs) checkRunnable(job *model.Job) bool {
 	}
 	return true
 }
-
-func (j *runningJobs) allIDs() string {
-	j.RLock()
-	defer j.RUnlock()
-	return j.runningJobIDs
-}
-
-func (j *runningJobs) updateInternalRunningJobIDs() {
-	var sb strings.Builder
-	i := 0
-	for id := range j.ids {
-		sb.WriteString(strconv.Itoa(int(id)))
-		if i != len(j.ids)-1 {
-			sb.WriteString(",")
-		}
-		i++
-	}
-	j.runningJobIDs = sb.String()
-}

ddl/ddl_running_jobs_test.go

@@ -95,16 +95,19 @@ func TestRunningJobs(t *testing.T) {
 	runnable = j.checkRunnable(mkJob(0, "db100.t100"))
 	require.False(t, runnable)
 
+	job5.State = model.JobStateDone
 	j.remove(job5)
 	require.Equal(t, "1,2,3,4", orderedAllIDs(j.allIDs()))
 	runnable = j.checkRunnable(mkJob(0, "db100.t100"))
 	require.True(t, runnable)
 
+	job3.State = model.JobStateDone
 	j.remove(job3)
 	require.Equal(t, "1,2,4", orderedAllIDs(j.allIDs()))
 	runnable = j.checkRunnable(mkJob(0, "db1.t100"))
 	require.True(t, runnable)
 
+	job1.State = model.JobStateDone
 	j.remove(job1)
 	require.Equal(t, "2,4", orderedAllIDs(j.allIDs()))
 	runnable = j.checkRunnable(mkJob(0, "db1.t1"))