diff --git a/logservice/schemastore/persist_storage_ddl_handlers.go b/logservice/schemastore/persist_storage_ddl_handlers.go
index 7296c2d550..2b917737d9 100644
--- a/logservice/schemastore/persist_storage_ddl_handlers.go
+++ b/logservice/schemastore/persist_storage_ddl_handlers.go
@@ -520,7 +520,7 @@ func getSchemaID(tableMap map[int64]*BasicTableInfo, tableID int64) int64 {
 // schemaName should be "Name.O"
 func findSchemaIDByName(databaseMap map[int64]*BasicDatabaseInfo, schemaName string) (int64, bool) {
 	for id, info := range databaseMap {
-		if info.Name == schemaName {
+		if strings.EqualFold(info.Name, schemaName) {
 			return id, true
 		}
 	}
@@ -530,7 +530,7 @@ func findSchemaIDByName(databaseMap map[int64]*BasicDatabaseInfo, schemaName str
 // tableName should be "Name.O"
 func findTableIDByName(tableMap map[int64]*BasicTableInfo, schemaID int64, tableName string) (int64, bool) {
 	for id, info := range tableMap {
-		if info.SchemaID == schemaID && info.Name == tableName {
+		if info.SchemaID == schemaID && strings.EqualFold(info.Name, tableName) {
 			return id, true
 		}
 	}
@@ -600,7 +600,7 @@ func buildPersistedDDLEventForDropView(args buildPersistedDDLEventFuncArgs) Pers
 	// We don't store the relationship: view_id -> table_name, get table name from args.job
 	event.TableName = args.job.TableName
 	// The query in job maybe "DROP VIEW test1.view1, test2.view2", we need rebuild it here.
-	event.Query = fmt.Sprintf("DROP VIEW `%s`.`%s`", event.SchemaName, event.TableName)
+	event.Query = fmt.Sprintf("DROP VIEW %s", common.QuoteSchema(event.SchemaName, event.TableName))
 	return event
 }
 
@@ -662,7 +662,7 @@ func buildPersistedDDLEventForDropTable(args buildPersistedDDLEventFuncArgs) Per
 	event.SchemaName = getSchemaName(args.databaseMap, event.SchemaID)
 	event.TableName = getTableName(args.tableMap, event.TableID)
 	// The query in job maybe "DROP TABLE test1.table1, test2.table2", we need rebuild it here.
-	event.Query = fmt.Sprintf("DROP TABLE `%s`.`%s`", event.SchemaName, event.TableName)
+	event.Query = fmt.Sprintf("DROP TABLE %s", common.QuoteSchema(event.SchemaName, event.TableName))
 	return event
 }
 
@@ -723,6 +723,11 @@ func buildPersistedDDLEventForRenameTable(args buildPersistedDDLEventFuncArgs) P
 		//
 		// InvolvingSchemaInfo returns the schema info involved in the job.
 		// The value should be stored in lower case.
+		//
+		// InvolvingSchemaInfo may store normalized lower-case names,
+		// while the original query can keep user-provided identifier case.
+		// Prefer names parsed from the original query whenever possible.
+		// See https://github.com/pingcap/ticdc/pull/2218 for background.
 		oldSchemaName := args.job.InvolvingSchemaInfo[0].Database
 		oldTableName := args.job.InvolvingSchemaInfo[0].Table
 		stmt, err := parser.New().ParseOneStmt(args.job.Query, "", "")
@@ -744,9 +749,9 @@ func buildPersistedDDLEventForRenameTable(args buildPersistedDDLEventFuncArgs) P
 				log.Error("unknown stmt type", zap.String("query", args.job.Query), zap.Any("stmt", stmt))
 			}
 		}
-		event.Query = fmt.Sprintf("RENAME TABLE `%s`.`%s` TO `%s`.`%s`",
-			oldSchemaName, oldTableName,
-			event.SchemaName, event.TableName)
+		event.Query = fmt.Sprintf("RENAME TABLE %s TO %s",
+			common.QuoteSchema(oldSchemaName, oldTableName),
+			common.QuoteSchema(event.SchemaName, event.TableName))
 	}
 	return event
 }
@@ -787,8 +792,10 @@ func buildPersistedDDLEventForExchangePartition(args buildPersistedDDLEventFuncA
 		// Note that partition name should be parsed from original query, not the upperQuery.
 		partName := strings.TrimSpace(event.Query[idx1:idx2])
 		partName = strings.Replace(partName, "`", "", -1)
-		event.Query = fmt.Sprintf("ALTER TABLE `%s`.`%s` EXCHANGE PARTITION `%s` WITH TABLE `%s`.`%s`",
-			event.ExtraSchemaName, event.ExtraTableName, partName, event.SchemaName, event.TableName)
+		event.Query = fmt.Sprintf("ALTER TABLE %s EXCHANGE PARTITION %s WITH TABLE %s",
+			common.QuoteSchema(event.ExtraSchemaName, event.ExtraTableName),
+			common.QuoteName(partName),
+			common.QuoteSchema(event.SchemaName, event.TableName))
 
 		if strings.HasSuffix(upperQuery, "WITHOUT VALIDATION") {
 			event.Query += " WITHOUT VALIDATION"
@@ -800,6 +807,41 @@ func buildPersistedDDLEventForExchangePartition(args buildPersistedDDLEventFuncA
 	return event
 }
 
+type renameTableQueryInfo struct {
+	oldSchemaName string
+	oldTableName  string
+	newSchemaName string
+	newTableName  string
+}
+
+func parseRenameTablesQueryInfos(query string) ([]renameTableQueryInfo, bool) {
+	if query == "" {
+		return nil, false
+	}
+	stmt, err := parser.New().ParseOneStmt(query, "", "")
+	if err != nil {
+		log.Warn("parse rename tables query failed",
+			zap.String("query", query),
+			zap.Error(err))
+		return nil, false
+	}
+	renameStmt, ok := stmt.(*ast.RenameTableStmt)
+	if !ok {
+		return nil, false
+	}
+
+	queryInfos := make([]renameTableQueryInfo, 0, len(renameStmt.TableToTables))
+	for _, tableToTable := range renameStmt.TableToTables {
+		queryInfos = append(queryInfos, renameTableQueryInfo{
+			oldSchemaName: tableToTable.OldTable.Schema.O,
+			oldTableName:  tableToTable.OldTable.Name.O,
+			newSchemaName: tableToTable.NewTable.Schema.O,
+			newTableName:  tableToTable.NewTable.Name.O,
+		})
+	}
+	return queryInfos, true
+}
+
 func buildPersistedDDLEventForRenameTables(args buildPersistedDDLEventFuncArgs) PersistedDDLEvent {
 	// TODO: does rename tables has the same problem(finished ts is not the real commit ts) with rename table?
 	event := buildPersistedDDLEventCommon(args)
@@ -809,25 +851,71 @@ func buildPersistedDDLEventForRenameTables(args buildPersistedDDLEventFuncArgs)
 			zap.String("query", args.job.Query),
 			zap.Error(err))
 	}
-	if len(renameArgs.RenameTableInfos) != len(args.job.BinlogInfo.MultipleTableInfos) {
+	renameTableInfos := renameArgs.RenameTableInfos
+	multipleTableInfos := args.job.BinlogInfo.MultipleTableInfos
+	if len(renameTableInfos) != len(multipleTableInfos) {
 		log.Panic("should not happen",
-			zap.Int("renameArgsLen", len(renameArgs.RenameTableInfos)),
-			zap.Int("multipleTableInfosLen", len(args.job.BinlogInfo.MultipleTableInfos)))
+			zap.Int("renameArgsLen", len(renameTableInfos)),
+			zap.Int("multipleTableInfosLen", len(multipleTableInfos)),
+			zap.String("query", args.job.Query))
+	}
+
+	// RenameTableInfos may store normalized lower-case names,
+	// while the original query can keep user-provided identifier case.
+	// Prefer names parsed from the original query whenever possible.
+	// See https://github.com/pingcap/ticdc/pull/2218 for background.
+	queryInfos, queryParsed := parseRenameTablesQueryInfos(args.job.Query)
+	if queryParsed && len(queryInfos) != len(renameTableInfos) {
+		log.Panic("rename tables query info length is inconsistent with args",
+			zap.Int("queryInfosLen", len(queryInfos)),
+			zap.Int("renameArgsLen", len(renameTableInfos)),
+			zap.String("query", args.job.Query))
+	}
+
+	// TiDB <= v8.1 may emit empty old table names for RENAME TABLES args.
+	// See https://github.com/pingcap/tidb/pull/64421 for the upstream fix.
+	if !queryParsed && renameTableInfos[0].OldTableName.O == "" {
+		// TODO: return error instead of falling back to args once builder supports error propagation.
+		log.Warn("rename tables args miss old table name and query is unavailable, keep args as-is",
+			zap.Int("tableCount", len(renameTableInfos)),
+			zap.String("query", args.job.Query))
 	}
 
 	var querys []string
-	for _, info := range renameArgs.RenameTableInfos {
-		event.ExtraSchemaIDs = append(event.ExtraSchemaIDs, info.OldSchemaID)
-		event.ExtraSchemaNames = append(event.ExtraSchemaNames, info.OldSchemaName.O)
-		event.ExtraTableNames = append(event.ExtraTableNames, info.OldTableName.O)
+	for i, info := range renameTableInfos {
+		oldSchemaID := info.OldSchemaID
+		oldSchemaName := info.OldSchemaName.O
+		oldTableName := info.OldTableName.O
+		newSchemaName := getSchemaName(args.databaseMap, info.NewSchemaID)
+		newTableName := info.NewTableName.O
+		if queryParsed {
+			queryInfo := queryInfos[i]
+			if queryInfo.oldSchemaName != "" {
+				oldSchemaName = queryInfo.oldSchemaName
+			}
+			if queryInfo.oldTableName != "" {
+				oldTableName = queryInfo.oldTableName
+			}
+			if queryInfo.newSchemaName != "" {
+				newSchemaName = queryInfo.newSchemaName
+			}
+			if queryInfo.newTableName != "" {
+				newTableName = queryInfo.newTableName
+			}
+		}
+
+		event.ExtraSchemaIDs = append(event.ExtraSchemaIDs, oldSchemaID)
+		event.ExtraSchemaNames = append(event.ExtraSchemaNames, oldSchemaName)
+		event.ExtraTableNames = append(event.ExtraTableNames, oldTableName)
 		event.SchemaIDs = append(event.SchemaIDs, info.NewSchemaID)
-		SchemaName := getSchemaName(args.databaseMap, info.NewSchemaID)
-		event.SchemaNames = append(event.SchemaNames, SchemaName)
-		querys = append(querys, fmt.Sprintf("RENAME TABLE `%s`.`%s` TO `%s`.`%s`;", info.OldSchemaName.O, info.OldTableName.O, SchemaName, info.NewTableName.O))
+		event.SchemaNames = append(event.SchemaNames, newSchemaName)
+		querys = append(querys, fmt.Sprintf("RENAME TABLE %s TO %s;",
+			common.QuoteSchema(oldSchemaName, oldTableName),
+			common.QuoteSchema(newSchemaName, newTableName)))
 	}
 
 	event.Query = strings.Join(querys, "")
-	event.MultipleTableInfos = args.job.BinlogInfo.MultipleTableInfos
+	event.MultipleTableInfos = multipleTableInfos
 	// we have to reverse MultipleTableInfos to get correct schema name
 	// see https://github.com/pingcap/tidb/issues/63710
 	//
diff --git a/logservice/schemastore/persist_storage_test.go b/logservice/schemastore/persist_storage_test.go
index fd22723db9..bb52e1b5a3 100644
--- a/logservice/schemastore/persist_storage_test.go
+++ b/logservice/schemastore/persist_storage_test.go
@@ -3130,6 +3130,336 @@ func TestRenameTable(t *testing.T) {
 	assert.Equal(t, "RENAME TABLE `test`.`t1` TO `test`.`t2`", ddl.Query)
 }
 
+func TestBuildPersistedDDLEventForRenameTablesFallbackOldTableName(t *testing.T) {
+	job := buildRenameTablesJobForTest(
+		[]int64{100, 100},
+		[]int64{105, 105},
+		[]int64{200, 201},
+		[]string{"source_db", "source_db"},
+		[]string{"", ""},
+		[]string{"target_t1", "target_t2"},
+		1010,
+	)
+	job.Query = "RENAME TABLE `source_db`.`source_t1` TO `target_db`.`target_t1`, `source_db`.`source_t2` TO `target_db`.`target_t2`"
+
+	ddl := buildPersistedDDLEventForRenameTables(buildPersistedDDLEventFuncArgs{
+		job: job,
+		databaseMap: map[int64]*BasicDatabaseInfo{
+			100: {Name: "source_db", Tables: map[int64]bool{200: true, 201: true}},
+			105: {Name: "target_db", Tables: map[int64]bool{200: true, 201: true}},
+		},
+		tableMap: map[int64]*BasicTableInfo{
+			200: {SchemaID: 100, Name: "source_t1"},
+		},
+	})
+
+	assert.Equal(t,
+		"RENAME TABLE `source_db`.`source_t1` TO `target_db`.`target_t1`;"+
+			"RENAME TABLE `source_db`.`source_t2` TO `target_db`.`target_t2`;",
+		ddl.Query)
+	assert.Equal(t, []string{"source_t1", "source_t2"}, ddl.ExtraTableNames)
+	assert.Equal(t, []string{"source_db", "source_db"}, ddl.ExtraSchemaNames)
+	assert.Equal(t, []string{"target_db", "target_db"}, ddl.SchemaNames)
+}
+
+func TestBuildPersistedDDLEventForRenameTablesCyclicRename(t *testing.T) {
+	// Simulate: rename table a to c, b to a, c to b.
+	// Table c only exists as a temporary name inside this statement.
+	job := buildRenameTablesJobForTest(
+		[]int64{100, 100, 100},
+		[]int64{100, 100, 100},
+		[]int64{200, 201, 201},
+		[]string{"test", "test", "test"},
+		[]string{"", "", ""},
+		[]string{"c", "a", "b"},
+		1010,
+	)
+	job.Query = "RENAME TABLE `test`.`a` TO `test`.`c`, `test`.`b` TO `test`.`a`, `test`.`c` TO `test`.`b`"
+
+	ddl := buildPersistedDDLEventForRenameTables(buildPersistedDDLEventFuncArgs{
+		job: job,
+		databaseMap: map[int64]*BasicDatabaseInfo{
+			100: {Name: "test", Tables: map[int64]bool{200: true, 201: true}},
+		},
+		tableMap: map[int64]*BasicTableInfo{
+			200: {SchemaID: 100, Name: "a"},
+			201: {SchemaID: 100, Name: "b"},
+		},
+	})
+
+	assert.Equal(t,
+		"RENAME TABLE `test`.`a` TO `test`.`c`;"+
+			"RENAME TABLE `test`.`b` TO `test`.`a`;"+
+			"RENAME TABLE `test`.`c` TO `test`.`b`;",
+		ddl.Query)
+	assert.Equal(t, []string{"a", "b", "c"}, ddl.ExtraTableNames)
+}
+
+func TestBuildPersistedDDLEventForRenameTablesPreferQueryNames(t *testing.T) {
+	job := buildRenameTablesJobForTest(
+		[]int64{100, 100},
+		[]int64{105, 105},
+		[]int64{200, 201},
+		[]string{"source_db", "source_db"},
+		[]string{"source_t1_from_args", "source_t2_from_args"},
+		[]string{"target_t1", "target_t2"},
+		1010,
+	)
+	job.Query = "RENAME TABLE `source_db`.`source_t1_from_query` TO `target_db`.`target_t1`, `source_db`.`source_t2_from_query` TO `target_db`.`target_t2`"
+
+	ddl := buildPersistedDDLEventForRenameTables(buildPersistedDDLEventFuncArgs{
+		job: job,
+		databaseMap: map[int64]*BasicDatabaseInfo{
+			100: {Name: "source_db", Tables: map[int64]bool{200: true, 201: true}},
+			105: {Name: "target_db", Tables: map[int64]bool{200: true, 201: true}},
+		},
+		tableMap: map[int64]*BasicTableInfo{
+			200: {SchemaID: 100, Name: "source_t1_from_store"},
+			201: {SchemaID: 100, Name: "source_t2_from_store"},
+		},
+	})
+
+	assert.Equal(t, []string{"source_t1_from_query", "source_t2_from_query"}, ddl.ExtraTableNames)
+	assert.Equal(t, []string{"source_db", "source_db"}, ddl.ExtraSchemaNames)
+	assert.Equal(t,
+		"RENAME TABLE `source_db`.`source_t1_from_query` TO `target_db`.`target_t1`;"+
+			"RENAME TABLE `source_db`.`source_t2_from_query` TO `target_db`.`target_t2`;",
+		ddl.Query)
+}
+
+func TestBuildPersistedDDLEventForRenameTablesKeepArgsWhenQueryUnavailable(t *testing.T) {
+	job := buildRenameTablesJobForTest(
+		[]int64{100, 100},
+		[]int64{105, 105},
+		[]int64{200, 201},
+		[]string{"source_db", "source_db"},
+		[]string{"source_t1_from_args", "source_t2_from_args"},
+		[]string{"target_t1", "target_t2"},
+		1010,
+	)
+	job.Query = "RENAME TABLE"
+
+	require.NotPanics(t, func() {
+		ddl := buildPersistedDDLEventForRenameTables(buildPersistedDDLEventFuncArgs{
+			job: job,
+			databaseMap: map[int64]*BasicDatabaseInfo{
+				100: {Name: "source_db", Tables: map[int64]bool{200: true, 201: true}},
+				105: {Name: "target_db", Tables: map[int64]bool{200: true, 201: true}},
+			},
+			tableMap: map[int64]*BasicTableInfo{
+				200: {SchemaID: 100, Name: "source_t1_from_store"},
+				201: {SchemaID: 100, Name: "source_t2_from_store"},
+			},
+		})
+
+		assert.Equal(t, []string{"source_t1_from_args", "source_t2_from_args"}, ddl.ExtraTableNames)
+		assert.Equal(t, []string{"source_db", "source_db"}, ddl.ExtraSchemaNames)
+		assert.Equal(t,
+			"RENAME TABLE `source_db`.`source_t1_from_args` TO `target_db`.`target_t1`;"+
+				"RENAME TABLE `source_db`.`source_t2_from_args` TO `target_db`.`target_t2`;",
+			ddl.Query)
+	})
+}
+
+func TestBuildPersistedDDLEventForRenameTablesPanicOnQueryInfoLengthMismatch(t *testing.T) {
+	job := buildRenameTablesJobForTest(
+		[]int64{100, 100},
+		[]int64{105, 105},
+		[]int64{200, 201},
+		[]string{"source_db", "source_db"},
+		[]string{"source_t1", "source_t2"},
+		[]string{"target_t1", "target_t2"},
+		1010,
+	)
+	job.Query = "RENAME TABLE `source_db`.`source_t1` TO `target_db`.`target_t1`, `source_db`.`source_t2` TO `target_db`.`target_t2`, `source_db`.`source_t3` TO `target_db`.`target_t3`"
+
+	require.Panics(t, func() {
+		_ = buildPersistedDDLEventForRenameTables(buildPersistedDDLEventFuncArgs{
+			job: job,
+			databaseMap: map[int64]*BasicDatabaseInfo{
+				100: {Name: "source_db", Tables: map[int64]bool{200: true, 201: true}},
+				105: {Name: "target_db", Tables: map[int64]bool{200: true, 201: true}},
+			},
+			tableMap: map[int64]*BasicTableInfo{
+				200: {SchemaID: 100, Name: "source_t1"},
+				201: {SchemaID: 100, Name: "source_t2"},
+			},
+		})
+	})
+}
+
+func TestBuildPersistedDDLEventEscapesIdentifiers(t *testing.T) {
+	t.Run("rename tables", func(t *testing.T) {
+		job := buildRenameTablesJobForTest(
+			[]int64{100, 100},
+			[]int64{105, 105},
+			[]int64{200, 201},
+			[]string{"source`db", "source`db"},
+			[]string{"source`t1", "source`t2"},
+			[]string{"target`t1", "target`t2"},
+			1010,
+		)
+
+		ddl := buildPersistedDDLEventForRenameTables(buildPersistedDDLEventFuncArgs{
+			job: job,
+			databaseMap: map[int64]*BasicDatabaseInfo{
+				100: {Name: "source`db", Tables: map[int64]bool{200: true, 201: true}},
+				105: {Name: "target`db", Tables: map[int64]bool{200: true, 201: true}},
+			},
+			tableMap: map[int64]*BasicTableInfo{
+				200: {SchemaID: 100, Name: "source`t1"},
+				201: {SchemaID: 100, Name: "source`t2"},
+			},
+		})
+
+		assert.Equal(t,
+			"RENAME TABLE `source``db`.`source``t1` TO `target``db`.`target``t1`;"+
+				"RENAME TABLE `source``db`.`source``t2` TO `target``db`.`target``t2`;",
+			ddl.Query)
+	})
+
+	t.Run("rename table", func(t *testing.T) {
+		job := buildRenameTableJobForTest(100, 101, "target`t", 100, &model.InvolvingSchemaInfo{
+			Database: "source`db",
+			Table:    "source`t",
+		})
+		// Keep empty to force using InvolvingSchemaInfo as source name.
+		job.Query = ""
+
+		ddl := buildPersistedDDLEventForRenameTable(buildPersistedDDLEventFuncArgs{
+			job: job,
+			databaseMap: map[int64]*BasicDatabaseInfo{
+				100: {Name: "target`db", Tables: map[int64]bool{101: true}},
+			},
+			tableMap: map[int64]*BasicTableInfo{
+				101: {SchemaID: 100, Name: "source`t"},
+			},
+		})
+
+		assert.Equal(t, "RENAME TABLE `source``db`.`source``t` TO `target``db`.`target``t`", ddl.Query)
+	})
+
+	t.Run("drop table", func(t *testing.T) {
+		job := buildDropTableJobForTest(100, 200, 1000)
+		ddl := buildPersistedDDLEventForDropTable(buildPersistedDDLEventFuncArgs{
+			job: job,
+			databaseMap: map[int64]*BasicDatabaseInfo{
+				100: {Name: "schema`x", Tables: map[int64]bool{200: true}},
+			},
+			tableMap: map[int64]*BasicTableInfo{
+				200: {SchemaID: 100, Name: "table`x"},
+			},
+		})
+		assert.Equal(t, "DROP TABLE `schema``x`.`table``x`", ddl.Query)
+	})
+
+	t.Run("drop view", func(t *testing.T) {
+		job := buildDropViewJobForTest(100, 1000)
+		job.TableName = "view`x"
+		ddl := buildPersistedDDLEventForDropView(buildPersistedDDLEventFuncArgs{
+			job: job,
+			databaseMap: map[int64]*BasicDatabaseInfo{
+				100: {Name: "schema`x", Tables: map[int64]bool{}},
+			},
+		})
+		assert.Equal(t, "DROP VIEW `schema``x`.`view``x`", ddl.Query)
+	})
+
+	t.Run("exchange partition", func(t *testing.T) {
+		job := buildExchangePartitionJobForTest(100, 200, 300, "pt`x", []int64{301}, 1000)
+		job.Query = "ALTER TABLE `ignored`.`ignored` EXCHANGE PARTITION `p0` WITH TABLE `ignored2`.`ignored2` WITHOUT VALIDATION"
+
+		ddl := buildPersistedDDLEventForExchangePartition(buildPersistedDDLEventFuncArgs{
+			job: job,
+			databaseMap: map[int64]*BasicDatabaseInfo{
+				100: {Name: "normal`db", Tables: map[int64]bool{200: true}},
+				101: {Name: "part`db", Tables: map[int64]bool{300: true}},
+			},
+			tableMap: map[int64]*BasicTableInfo{
+				200: {SchemaID: 100, Name: "normal`t"},
+				300: {SchemaID: 101, Name: "pt`x"},
+			},
+			partitionMap: map[int64]BasicPartitionInfo{
+				300: {
+					301: nil,
+				},
+			},
+		})
+
+		assert.Equal(t,
+			"ALTER TABLE `part``db`.`pt``x` EXCHANGE PARTITION `p0` WITH TABLE `normal``db`.`normal``t` WITHOUT VALIDATION",
+			ddl.Query)
+	})
+}
+
+func TestParseRenameTablesQueryInfos(t *testing.T) {
+	cases := []struct {
+		name     string
+		query    string
+		parsed   bool
+		expected []renameTableQueryInfo
+	}{
+		{
+			name:   "multiple tables with schema",
+			query:  "RENAME TABLE `db1`.`t1` TO `db2`.`t2`, `db1`.`t3` TO `db2`.`t4`",
+			parsed: true,
+			expected: []renameTableQueryInfo{
+				{
+					oldSchemaName: "db1",
+					oldTableName:  "t1",
+					newSchemaName: "db2",
+					newTableName:  "t2",
+				},
+				{
+					oldSchemaName: "db1",
+					oldTableName:  "t3",
+					newSchemaName: "db2",
+					newTableName:  "t4",
+				},
+			},
+		},
+		{
+			name:   "without schema names",
+			query:  "RENAME TABLE `t1` TO `t2`",
+			parsed: true,
+			expected: []renameTableQueryInfo{
+				{
+					oldSchemaName: "",
+					oldTableName:  "t1",
+					newSchemaName: "",
+					newTableName:  "t2",
+				},
+			},
+		},
+		{
+			name:     "empty query",
+			query:    "",
+			parsed:   false,
+			expected: nil,
+		},
+		{
+			name:     "non rename statement",
+			query:    "CREATE TABLE t(a INT)",
+			parsed:   false,
+			expected: nil,
+		},
+		{
+			name:     "invalid sql",
+			query:    "RENAME TABLE",
+			parsed:   false,
+			expected: nil,
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			got, parsed := parseRenameTablesQueryInfos(tc.query)
+			assert.Equal(t, tc.parsed, parsed)
+			assert.Equal(t, tc.expected, got)
+		})
+	}
+}
+
 func TestBuildDDLEventForNewTableDDL_CreateTableLikeBlockedTableNames(t *testing.T) {
 	cases := []struct {
 		name     string
@@ -3189,6 +3519,18 @@ func TestBuildPersistedDDLEventForCreateTableLikeSetsReferTableID(t *testing.T)
 			partitionIDs:    []int64{111, 112},
 			expectedReferID: 101,
 		},
+		{
+			name:            "refer table name with different case",
+			query:           "CREATE TABLE `b` LIKE `A`",
+			partitionIDs:    nil,
+			expectedReferID: 101,
+		},
+		{
+			name:            "refer schema and table names with different case",
+			query:           "CREATE TABLE `b` LIKE `TeSt`.`A`",
+			partitionIDs:    nil,
+			expectedReferID: 101,
+		},
 	}
 
 	for _, tc := range cases {
diff --git a/tests/integration_tests/_utils/test_prepare b/tests/integration_tests/_utils/test_prepare
index 81121ee207..4d087151fd 100644
--- a/tests/integration_tests/_utils/test_prepare
+++ b/tests/integration_tests/_utils/test_prepare
@@ -93,3 +93,76 @@ else
 fi
 
 export TEST_OUT_DATA_DIR=test_out_data
+
+function get_tidb_release_version() {
+	local tidbHost=${1:-$UP_TIDB_HOST}
+	local tidbPort=${2:-$UP_TIDB_PORT}
+
+	mysql -uroot -h${tidbHost} -P${tidbPort} --default-character-set utf8mb4 -e "select tidb_version()\G" \
+		| sed -nE 's/.*Release Version:[[:space:]]*(v[0-9]+(\.[0-9]+){0,2}([-.+][0-9A-Za-z.-]+)?).*/\1/p' \
+		| head -n1
+}
+
+function normalize_tidb_semver_triplet() {
+	local rawVersion=${1#v}
+	rawVersion=${rawVersion%%-*}
+	rawVersion=${rawVersion%%+*}
+
+	local major minor patch
+	IFS='.' read -r major minor patch <<<"$rawVersion"
+	major=${major:-0}
+	minor=${minor:-0}
+	patch=${patch:-0}
+
+	if ! [[ "$major" =~ ^[0-9]+$ && "$minor" =~ ^[0-9]+$ && "$patch" =~ ^[0-9]+$ ]]; then
+		return 1
+	fi
+
+	echo "$major $minor $patch"
+}
+
+function tidb_version_less_than() {
+	local currentVersion=$1
+	local minVersion=$2
+
+	local currentTriplet minTriplet
+	currentTriplet=$(normalize_tidb_semver_triplet "$currentVersion") || return 1
+	minTriplet=$(normalize_tidb_semver_triplet "$minVersion") || return 1
+
+	local -a currentParts minParts
+	read -r -a currentParts <<<"$currentTriplet"
+	read -r -a minParts <<<"$minTriplet"
+
+	local idx currentPart minPart
+	for idx in 0 1 2; do
+		currentPart=$((10#${currentParts[$idx]}))
+		minPart=$((10#${minParts[$idx]}))
+		if ((currentPart < minPart)); then
+			return 0
+		fi
+		if ((currentPart > minPart)); then
+			return 1
+		fi
+	done
+
+	return 1
+}
+
+function skip_if_tidb_version_less_than() {
+	local minVersion=$1
+	local tidbHost=${2:-$UP_TIDB_HOST}
+	local tidbPort=${3:-$UP_TIDB_PORT}
+	local testName=${TEST_NAME:-unknown}
+
+	local tidbReleaseVersion
+	tidbReleaseVersion=$(get_tidb_release_version "$tidbHost" "$tidbPort" || true)
+	if [ -z "$tidbReleaseVersion" ]; then
+		echo "[$(date)] failed to parse TiDB release version from ${tidbHost}:${tidbPort}, continue test case ${testName}"
+		return
+	fi
+
+	if tidb_version_less_than "$tidbReleaseVersion" "$minVersion"; then
+		echo "[$(date)] <<<<<< skip test case ${testName}, TiDB version ${tidbReleaseVersion} is less than ${minVersion} >>>>>>"
+		exit 0
+	fi
+}
diff --git a/tests/integration_tests/vector/run.sh b/tests/integration_tests/vector/run.sh
index 4bec91c09d..4ddd2b6b05 100755
--- a/tests/integration_tests/vector/run.sh
+++ b/tests/integration_tests/vector/run.sh
@@ -12,6 +12,7 @@ function run() {
 	rm -rf $WORK_DIR && mkdir -p $WORK_DIR
 
 	start_tidb_cluster --workdir $WORK_DIR
+	skip_if_tidb_version_less_than "v8.5.0"
 
 	start_ts=$(run_cdc_cli_tso_query ${UP_PD_HOST_1} ${UP_PD_PORT_1})
 	run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY