infoschema: update raw args to make it drop correct table in `RENAME …

…TABLE` (#44585) (#44595)

close #43714

ddl/table.go

@@ -16,6 +16,7 @@ package ddl
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"strconv"
 	"sync/atomic"
@@ -1178,10 +1179,20 @@ func finishJobRenameTable(d *ddlCtx, t *meta.Meta, job *model.Job) (int64, error
 		job.State = model.JobStateCancelled
 		return 0, errors.Trace(err)
 	}
+	// Before updating the schema version, we need to reset the old schema ID to new schema ID, so that
+	// the table info can be dropped normally in `ApplyDiff`. This is because renaming table requires two
+	// schema versions to complete.
+	oldRawArgs := job.RawArgs
+	job.Args[0] = job.SchemaID
+	job.RawArgs, err = json.Marshal(job.Args)
+	if err != nil {
+		return 0, errors.Trace(err)
+	}
 	ver, err := updateSchemaVersion(d, t, job)
 	if err != nil {
 		return ver, errors.Trace(err)
 	}
+	job.RawArgs = oldRawArgs
 	job.FinishTableJob(model.JobStateDone, model.StatePublic, ver, tblInfo)
 	return ver, nil
 }
@@ -1202,10 +1213,21 @@ func finishJobRenameTables(d *ddlCtx, t *meta.Meta, job *model.Job,
 		}
 		tblInfos = append(tblInfos, tblInfo)
 	}
+	// Before updating the schema version, we need to reset the old schema ID to new schema ID, so that
+	// the table info can be dropped normally in `ApplyDiff`. This is because renaming table requires two
+	// schema versions to complete.
+	var err error
+	oldRawArgs := job.RawArgs
+	job.Args[0] = newSchemaIDs
+	job.RawArgs, err = json.Marshal(job.Args)
+	if err != nil {
+		return 0, errors.Trace(err)
+	}
 	ver, err := updateSchemaVersion(d, t, job)
 	if err != nil {
 		return ver, errors.Trace(err)
 	}
+	job.RawArgs = oldRawArgs
 	job.FinishMultipleTableJob(model.JobStateDone, model.StatePublic, ver, tblInfos)
 	return ver, nil
 }

infoschema/infoschema_test.go

@@ -822,3 +822,23 @@ func TestIssue42400(t *testing.T) {
 	tk.MustQuery("show create table information_schema.ddl_jobs").CheckContain("`QUERY` text")
 	tk.MustQuery("select length(query) from information_schema.ddl_jobs;") // No error
 }
+
+func TestInfoSchemaRenameTable(t *testing.T) {
+	store := testkit.CreateMockStore(t)
+
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("use test")
+	tk.MustExec("create table test.t1 (id int primary key, a text);")
+	tk.MustExec("insert test.t1 values(1,'334'),(4,'3443435'),(5,'fdf43t536653');")
+	tk.MustExec("rename table test.t1 to mysql.t1;")
+	tk.MustQuery("SELECT count(*) FROM information_schema.TABLES WHERE (TABLE_SCHEMA = 'mysql') AND (TABLE_NAME = 't1');").
+		Check(testkit.Rows("1"))
+
+	tk.MustExec("create table test.t2 (id int primary key, a text);")
+	tk.MustExec("insert test.t2 values(1,'334'),(4,'3443435'),(5,'fdf43t536653');")
+	tk.MustExec("create table test.t3 (id int primary key, a text);")
+	tk.MustExec("insert test.t3 values(1,'334'),(4,'3443435'),(5,'fdf43t536653');")
+	tk.MustExec("rename table test.t2 to mysql.t2, test.t3 to mysql.t3;")
+	tk.MustQuery("SELECT count(*) FROM information_schema.TABLES WHERE (TABLE_SCHEMA = 'mysql') AND (TABLE_NAME = 't3');").
+		Check(testkit.Rows("1"))
+}