schemastore: fix old table metadata for rename table

logservice/schemastore/persist_storage_ddl_handlers.go

@@ -795,45 +795,60 @@ func buildPersistedDDLEventForTruncateTable(args buildPersistedDDLEventFuncArgs)
 
 func buildPersistedDDLEventForRenameTable(args buildPersistedDDLEventFuncArgs) PersistedDDLEvent {
 	event := buildPersistedDDLEventCommon(args)
-	// Note: schema id/schema name/table name may be changed or not
-	// table id does not change, we use it to get the table's prev schema id/name and table name
-	event.ExtraSchemaID = getSchemaID(args.tableMap, event.TableID)
-	// TODO: check how ExtraTableName will be used later
-	event.ExtraTableName = getTableName(args.tableMap, event.TableID)
-	event.ExtraSchemaName = getSchemaName(args.databaseMap, event.ExtraSchemaID)
 	event.SchemaName = getSchemaName(args.databaseMap, event.SchemaID)
-	// get the table's current table name from the ddl job
 	event.TableName = event.TableInfo.Name.O
 
-	// The old schema/table names cannot rely on ExtraSchemaName/ExtraTableName,
-	// because the snapshot used by schema store may already reflect the post-rename state.
-	// Example (after https://github.com/pingcap/tidb/pull/43341):
-	//   table `test.t`, DDL `rename table t to test2.t;`, commit ts = 100
-	//   snapshot at ts = 99 already shows `t` under `test2`
-	//   => event.ExtraSchemaName becomes `test2`, which is wrong for the old name
-	// SchemaStore can still use ExtraSchemaID to update internal state,
-	// but the emitted event.Query must carry the correct old names.
-	// Rebuild them with the following precedence:
-	// 1. InvolvingSchemaInfo provides a fallback old schema/table pair, but names may be normalized.
-	// 2. RenameTableArgs.OldSchemaName overrides the fallback when available.
-	//    It is reliable in TiDB >= v8.5, but can be missing in older versions.
-	// 3. The original query (if it specifies old schema) has the highest priority for identifier case.
-	// 4. If the query omits old schema and ExtraSchemaID differs from SchemaID, use ExtraSchemaID to
-	//    recover the old schema name from the schema store.
+	// Why the old table identity must be recovered instead of being read directly from tableMap:
+	// suppose `RENAME TABLE test.t1 TO test2.t2` commits at ts=100 and SchemaStore starts
+	// from ts=99. SchemaStore first loads a TiDB metadata snapshot at ts=99, then replays
+	// DDL jobs after that snapshot. Since https://github.com/pingcap/tidb/pull/43341, the
+	// snapshot at ts=99 may already contain the post-rename table `test2.t2`. Therefore,
+	// tableMap[event.TableID] is not guaranteed to describe the table before this DDL.
+	// Using it directly would make ExtraSchemaID/ExtraSchemaName/ExtraTableName describe
+	// the new table, although these fields are consumed as the old table identity by DDL
+	// filtering, barrier construction, table-name changes, and cross-schema updates.
+	//
+	// Recover the old identity from independent fields in the DDL job:
+	//  1. InvolvingSchemaInfo[0] contains the old schema/table names. TiDB deliberately
+	//     stores these values in lower case (Schema.Name.L and Table.Name.L), so they are
+	//     useful for lookup but do not preserve the original identifier capitalization.
+	//  2. RenameTableArgs.OldSchemaID identifies the old schema. OldSchemaName preserves
+	//     its original capitalization. TiDB has included both fields in rename-table job
+	//     args since v5.3.0, so they are available in every supported TiDB version (v7.5.0+).
+	//  3. TiDB always records the original rename SQL in job.Query. It always contains the
+	//     old table name, but the old schema name is optional. Prefer names parsed from SQL
+	//     because they preserve the identifier capitalization written by the user.
+	//  4. Complete the old schema identity with databaseMap: look up the schema ID when only
+	//     its name is known, or look up its name when only the ID is known. If the original
+	//     SQL explicitly specifies the old schema name, preserve that spelling.
+	//
+	// Rename jobs from supported TiDB versions provide enough information to recover all three
+	// old identity fields. tableMap is only a defensive fallback for a malformed or unsupported
+	// job whose args cannot be decoded or whose SQL cannot be parsed. If this fallback is used
+	// with a post-rename snapshot, the Extra fields may describe the new table and cause incorrect
+	// filter or table-name-store behavior.
+	oldSchemaID := int64(0)
 	oldSchemaName := ""
 	oldTableName := ""
 	oldSchemaSource := "unknown"
+
+	// Start with the lower-case old schema/table names recorded for DDL dependency checks.
 	if len(args.job.InvolvingSchemaInfo) > 0 {
 		oldSchemaName = args.job.InvolvingSchemaInfo[0].Database
 		oldTableName = args.job.InvolvingSchemaInfo[0].Table
 		if oldSchemaName != "" {
 			oldSchemaSource = "involving_schema_info"
 		}
 	}
+
+	// Recover the authoritative old schema ID and its case-preserving name from job args.
 	if args.job.Version == model.JobVersion1 || args.job.Version == model.JobVersion2 {
 		if renameArgs, err := model.GetRenameTableArgs(args.job); err == nil {
+			oldSchemaID = renameArgs.OldSchemaID
 			if renameArgs.OldSchemaName.O != "" {
 				oldSchemaName = renameArgs.OldSchemaName.O
+			}
+			if oldSchemaID != 0 || renameArgs.OldSchemaName.O != "" {
 				oldSchemaSource = "rename_table_args"
 			}
 		} else {
@@ -843,26 +858,62 @@ func buildPersistedDDLEventForRenameTable(args buildPersistedDDLEventFuncArgs) P
 				zap.Error(err))
 		}
 	}
-	queryProvidedOldSchema := false
-	if queryInfo, parsed := parseRenameTableQueryInfo(args.job.Query); parsed {
+
+	// Recover the old table name from the normalized SQL, and prefer its schema spelling
+	// when it identifies the same schema as the DDL job args. event.Query has already been
+	// parsed with job.SQLMode and restored by buildPersistedDDLEventCommon.
+	queryInfo, parsed := parseRenameTableQueryInfo(event.Query)
+	if parsed {
 		if queryInfo.oldTableName != "" {
 			oldTableName = queryInfo.oldTableName
 		}
 		if queryInfo.oldSchemaName != "" {
-			oldSchemaName = queryInfo.oldSchemaName
-			queryProvidedOldSchema = true
-			oldSchemaSource = "query"
+			queryOldSchemaID, _ := findSchemaIDByName(args.databaseMap, queryInfo.oldSchemaName)
+			if oldSchemaID != 0 && oldSchemaID != queryOldSchemaID {
+				log.Warn("rename table old schema is inconsistent between job args and query",
+					zap.Int64("jobID", args.job.ID),
+					zap.Int64("argsOldSchemaID", oldSchemaID),
+					zap.Int64("queryOldSchemaID", queryOldSchemaID),
+					zap.String("queryOldSchemaName", queryInfo.oldSchemaName),
+					zap.String("query", event.Query))
+				oldSchemaName = getSchemaName(args.databaseMap, oldSchemaID)
+			} else {
+				oldSchemaID = queryOldSchemaID
+				oldSchemaName = queryInfo.oldSchemaName
+				oldSchemaSource = "query"
+			}
 		}
 	}
-	// ExtraSchemaID can be incorrect due to snapshot timing, so only use it if the query
-	// does not specify the old schema.
-	if !queryProvidedOldSchema && event.ExtraSchemaID != 0 && event.ExtraSchemaID != event.SchemaID {
-		if extraName := getSchemaName(args.databaseMap, event.ExtraSchemaID); extraName != "" {
-			oldSchemaName = extraName
-			oldSchemaSource = "extra_schema_id"
+
+	// Complete a missing old schema ID or name through databaseMap.
+	if oldSchemaID == 0 && oldSchemaName != "" {
+		oldSchemaID, _ = findSchemaIDByName(args.databaseMap, oldSchemaName)
+	}
+	if queryInfo.oldSchemaName == "" {
+		if oldSchemaID != 0 {
+			// SQL does not provide the old schema spelling. Use databaseMap to replace
+			// the lower-case InvolvingSchemaInfo name with its original capitalization.
+			oldSchemaName = getSchemaName(args.databaseMap, oldSchemaID)
 		}
 	}
-	if oldSchemaName != "" && oldTableName != "" {
+
+	// TiDB v7.5+ rename jobs provide the old schema through job args and the old table
+	// through SQL. For malformed or unsupported jobs, fall back to the complete snapshot
+	// identity instead of combining fields from different sources.
+	if oldSchemaID == 0 || oldSchemaName == "" || oldTableName == "" {
+		oldSchemaID = getSchemaID(args.tableMap, event.TableID)
+		oldSchemaName = getSchemaName(args.databaseMap, oldSchemaID)
+		oldTableName = getTableName(args.tableMap, event.TableID)
+		oldSchemaSource = "table_map_fallback"
+	}
+
+	// Persist the recovered old identity for downstream filtering and coordination.
+	event.ExtraSchemaID = oldSchemaID
+	event.ExtraSchemaName = oldSchemaName
+	event.ExtraTableName = oldTableName
+
+	// Keep the executable query consistent with the recovered structured metadata.
+	if event.ExtraSchemaName != "" && event.ExtraTableName != "" {
 		log.Info("rebuild rename table query",
 			zap.Int64("jobID", event.ID),
 			zap.String("query", event.Query),
@@ -872,11 +923,9 @@ func buildPersistedDDLEventForRenameTable(args buildPersistedDDLEventFuncArgs) P
 			zap.Int64("extraSchemaID", event.ExtraSchemaID),
 			zap.String("extraSchemaName", event.ExtraSchemaName),
 			zap.String("extraTableName", event.ExtraTableName),
-			zap.String("oldSchemaName", oldSchemaName),
-			zap.String("oldTableName", oldTableName),
 			zap.String("oldSchemaSource", oldSchemaSource))
 		event.Query = fmt.Sprintf("RENAME TABLE %s TO %s",
-			common.QuoteSchema(oldSchemaName, oldTableName),
+			common.QuoteSchema(event.ExtraSchemaName, event.ExtraTableName),
 			common.QuoteSchema(event.SchemaName, event.TableName))
 	}
 	return event

logservice/schemastore/persist_storage_test.go

@@ -29,6 +29,7 @@ import (
 	"github.com/pingcap/tidb/pkg/meta/model"
 	"github.com/pingcap/tidb/pkg/parser/ast"
 	"github.com/pingcap/tidb/pkg/parser/charset"
+	"github.com/pingcap/tidb/pkg/parser/mysql"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/zap"
@@ -3197,6 +3198,171 @@ func TestRenameTable(t *testing.T) {
 	assert.Equal(t, "RENAME TABLE `SalesDB`.`t1` TO `ArchiveDB`.`t1`", ddl.Query)
 }
 
+func TestRenameTableRepairsOldTableMetadata(t *testing.T) {
+	t.Run("same schema", func(t *testing.T) {
+		job := buildRenameTableJobForTest(100, 101, "t2", 100, &model.InvolvingSchemaInfo{
+			Database: "test",
+			Table:    "t1",
+		})
+		job.Query = "RENAME TABLE t1 TO t2"
+		rawEvent := buildPersistedDDLEventForRenameTable(buildPersistedDDLEventFuncArgs{
+			job: job,
+			databaseMap: map[int64]*BasicDatabaseInfo{
+				100: {Name: "test", Tables: map[int64]bool{101: true}},
+			},
+			// Simulate a snapshot that already contains the post-rename table name.
+			tableMap: map[int64]*BasicTableInfo{
+				101: {SchemaID: 100, Name: "t2"},
+			},
+		})
+
+		require.Equal(t, "RENAME TABLE `test`.`t1` TO `test`.`t2`", rawEvent.Query)
+		require.Equal(t, int64(100), rawEvent.ExtraSchemaID)
+		require.Equal(t, "test", rawEvent.ExtraSchemaName)
+		require.Equal(t, "t1", rawEvent.ExtraTableName)
+
+		ddlEvent, ok, err := buildDDLEventForRenameTable(&rawEvent, nil, 0)
+		require.NoError(t, err)
+		require.True(t, ok)
+		require.Equal(t, []commonEvent.SchemaTableName{{SchemaName: "test", TableName: "t1"}}, ddlEvent.BlockedTableNames)
+		require.Equal(t, &commonEvent.TableNameChange{
+			AddName:  []commonEvent.SchemaTableName{{SchemaName: "test", TableName: "t2"}},
+			DropName: []commonEvent.SchemaTableName{{SchemaName: "test", TableName: "t1"}},
+		}, ddlEvent.TableNameChange)
+	})
+
+	t.Run("cross schema with old TiDB job args", func(t *testing.T) {
+		job := buildRenameTableJobForTest(100, 101, "t1", 100, nil)
+		job.Version = model.JobVersion1
+		job.FillArgs(&model.RenameTableArgs{
+			OldSchemaID:  200,
+			NewTableName: ast.NewCIStr("t1"),
+		})
+		_, err := job.Encode(true)
+		require.NoError(t, err)
+		job.Query = "RENAME TABLE t1 TO target_db.t1"
+		rawEvent := buildPersistedDDLEventForRenameTable(buildPersistedDDLEventFuncArgs{
+			job: job,
+			databaseMap: map[int64]*BasicDatabaseInfo{
+				100: {Name: "target_db", Tables: map[int64]bool{101: true}},
+				200: {Name: "source_db", Tables: map[int64]bool{}},
+			},
+			// Simulate a snapshot that has already moved the table to the new schema.
+			tableMap: map[int64]*BasicTableInfo{
+				101: {SchemaID: 100, Name: "t1"},
+			},
+		})
+
+		require.Equal(t, "RENAME TABLE `source_db`.`t1` TO `target_db`.`t1`", rawEvent.Query)
+		require.Equal(t, int64(200), rawEvent.ExtraSchemaID)
+		require.Equal(t, "source_db", rawEvent.ExtraSchemaName)
+		require.Equal(t, "t1", rawEvent.ExtraTableName)
+
+		ddlEvent, ok, err := buildDDLEventForRenameTable(&rawEvent, nil, 0)
+		require.NoError(t, err)
+		require.True(t, ok)
+		require.Equal(t, []commonEvent.SchemaIDChange{{
+			TableID:     101,
+			OldSchemaID: 200,
+			NewSchemaID: 100,
+		}}, ddlEvent.UpdatedSchemas)
+		require.Equal(t, &commonEvent.TableNameChange{
+			AddName:  []commonEvent.SchemaTableName{{SchemaName: "target_db", TableName: "t1"}},
+			DropName: []commonEvent.SchemaTableName{{SchemaName: "source_db", TableName: "t1"}},
+		}, ddlEvent.TableNameChange)
+
+		ddlEvent, ok, err = buildDDLEventForRenameTable(
+			&rawEvent, buildTableFilterByNameForTest("source_db", "*"), 0)
+		require.NoError(t, err)
+		require.True(t, ok)
+		require.Equal(t, &commonEvent.InfluencedTables{
+			InfluenceType: commonEvent.InfluenceTypeNormal,
+			TableIDs:      []int64{101},
+		}, ddlEvent.NeedDroppedTables)
+		require.Equal(t, &commonEvent.TableNameChange{
+			DropName: []commonEvent.SchemaTableName{{SchemaName: "source_db", TableName: "t1"}},
+		}, ddlEvent.TableNameChange)
+	})
+
+	t.Run("parse normalized query with ANSI quotes", func(t *testing.T) {
+		job := buildRenameTableJobForTest(100, 101, "NewTable", 100, &model.InvolvingSchemaInfo{
+			Database: "sourcedb",
+			Table:    "oldtable",
+		})
+		job.Version = model.JobVersion2
+		job.SQLMode = mysql.ModeANSIQuotes
+		job.FillArgs(&model.RenameTableArgs{
+			OldSchemaID:   200,
+			OldSchemaName: ast.NewCIStr("SourceDB"),
+			NewTableName:  ast.NewCIStr("NewTable"),
+		})
+		job.Query = `RENAME TABLE "SourceDB"."OldTable" TO "TargetDB"."NewTable"`
+
+		rawEvent := buildPersistedDDLEventForRenameTable(buildPersistedDDLEventFuncArgs{
+			job: job,
+			databaseMap: map[int64]*BasicDatabaseInfo{
+				100: {Name: "TargetDB", Tables: map[int64]bool{101: true}},
+				200: {Name: "SourceDB", Tables: map[int64]bool{}},
+			},
+			tableMap: map[int64]*BasicTableInfo{
+				101: {SchemaID: 100, Name: "NewTable"},
+			},
+		})
+
+		require.Equal(t, int64(200), rawEvent.ExtraSchemaID)
+		require.Equal(t, "SourceDB", rawEvent.ExtraSchemaName)
+		require.Equal(t, "OldTable", rawEvent.ExtraTableName)
+		require.Equal(t, "RENAME TABLE `SourceDB`.`OldTable` TO `TargetDB`.`NewTable`", rawEvent.Query)
+	})
+
+	t.Run("prefer job args when query schema ID is inconsistent", func(t *testing.T) {
+		job := buildRenameTableJobForTest(100, 101, "target_t", 100, nil)
+		job.Version = model.JobVersion2
+		job.FillArgs(&model.RenameTableArgs{
+			OldSchemaID:   200,
+			OldSchemaName: ast.NewCIStr("source_db"),
+			NewTableName:  ast.NewCIStr("target_t"),
+		})
+		job.Query = "RENAME TABLE wrong_db.source_t TO target_db.target_t"
+
+		rawEvent := buildPersistedDDLEventForRenameTable(buildPersistedDDLEventFuncArgs{
+			job: job,
+			databaseMap: map[int64]*BasicDatabaseInfo{
+				100: {Name: "target_db", Tables: map[int64]bool{101: true}},
+				200: {Name: "source_db", Tables: map[int64]bool{}},
+				300: {Name: "wrong_db", Tables: map[int64]bool{}},
+			},
+			// The snapshot contains the post-rename identity and must not be mixed in.
+			tableMap: map[int64]*BasicTableInfo{
+				101: {SchemaID: 100, Name: "target_t"},
+			},
+		})
+
+		require.Equal(t, int64(200), rawEvent.ExtraSchemaID)
+		require.Equal(t, "source_db", rawEvent.ExtraSchemaName)
+		require.Equal(t, "source_t", rawEvent.ExtraTableName)
+		require.Equal(t, "RENAME TABLE `source_db`.`source_t` TO `target_db`.`target_t`", rawEvent.Query)
+	})
+
+	t.Run("fall back to complete snapshot identity", func(t *testing.T) {
+		job := buildRenameTableJobForTest(100, 101, "target_t", 100, nil)
+
+		rawEvent := buildPersistedDDLEventForRenameTable(buildPersistedDDLEventFuncArgs{
+			job: job,
+			databaseMap: map[int64]*BasicDatabaseInfo{
+				100: {Name: "snapshot_db", Tables: map[int64]bool{101: true}},
+			},
+			tableMap: map[int64]*BasicTableInfo{
+				101: {SchemaID: 100, Name: "snapshot_t"},
+			},
+		})
+
+		require.Equal(t, int64(100), rawEvent.ExtraSchemaID)
+		require.Equal(t, "snapshot_db", rawEvent.ExtraSchemaName)
+		require.Equal(t, "snapshot_t", rawEvent.ExtraTableName)
+	})
+}
+
 func TestBuildPersistedDDLEventForRenameTablesFallbackOldTableName(t *testing.T) {
 	job := buildRenameTablesJobForTest(
 		[]int64{100, 100},
@@ -3397,9 +3563,10 @@ func TestBuildPersistedDDLEventEscapesIdentifiers(t *testing.T) {
 			job: job,
 			databaseMap: map[int64]*BasicDatabaseInfo{
 				100: {Name: "target`db", Tables: map[int64]bool{101: true}},
+				200: {Name: "source`db", Tables: map[int64]bool{}},
 			},
 			tableMap: map[int64]*BasicTableInfo{
-				101: {SchemaID: 100, Name: "source`t"},
+				101: {SchemaID: 200, Name: "source`t"},
 			},
 		})