ticdc: unquote exchange partition names when rebuilding ddl

cdc/model/sink.go

@@ -1130,10 +1130,12 @@ func (d *DDLEvent) FromJobWithArgs(
 
 		// Note that partition name should be parsed from original query, not the upperQuery.
 		partName := strings.TrimSpace(job.Query[idx1:idx2])
+		partName = quotes.UnquoteName(partName)
 		// The tableInfo is the partition table, preTableInfo is non partition table.
-		d.Query = fmt.Sprintf("ALTER TABLE `%s`.`%s` EXCHANGE PARTITION `%s` WITH TABLE `%s`.`%s`",
-			tableInfo.TableName.Schema, tableInfo.TableName.Table, partName,
-			preTableInfo.TableName.Schema, preTableInfo.TableName.Table)
+		d.Query = fmt.Sprintf("ALTER TABLE %s EXCHANGE PARTITION %s WITH TABLE %s",
+			quotes.QuoteSchema(tableInfo.TableName.Schema, tableInfo.TableName.Table),
+			quotes.QuoteName(partName),
+			quotes.QuoteSchema(preTableInfo.TableName.Schema, preTableInfo.TableName.Table))
 
 		if strings.HasSuffix(upperQuery, "WITHOUT VALIDATION") {
 			d.Query += " WITHOUT VALIDATION"

cdc/model/sink_test.go

@@ -401,6 +401,68 @@ func TestExchangeTablePartition(t *testing.T) {
 	require.Equal(t, event.Type, timodel.ActionExchangeTablePartition)
 }
 
+func TestExchangeTablePartitionQuotedName(t *testing.T) {
+	ft := types.NewFieldType(mysql.TypeUnspecified)
+	ft.SetFlag(mysql.PriKeyFlag | mysql.UniqueFlag)
+	job := &timodel.Job{
+		ID:         71,
+		TableID:    69,
+		SchemaName: "test1",
+		Type:       timodel.ActionExchangeTablePartition,
+		StartTS:    432853521879007233,
+		Query:      "alter table t1 exchange partition `p``0` with table t2 without validation",
+		BinlogInfo: &timodel.HistoryInfo{
+			FinishedTS: 432853521879007238,
+		},
+	}
+
+	preTableInfo := &TableInfo{
+		TableName: TableName{
+			Schema:  "test2",
+			Table:   "t2",
+			TableID: 67,
+		},
+		TableInfo: &timodel.TableInfo{
+			ID:   67,
+			Name: pmodel.CIStr{O: "t1"},
+			Columns: []*timodel.ColumnInfo{
+				{
+					ID:        1,
+					Name:      pmodel.CIStr{O: "id"},
+					FieldType: *ft,
+					State:     timodel.StatePublic,
+				},
+			},
+		},
+	}
+	tableInfo := &TableInfo{
+		TableName: TableName{
+			Schema:  "test1",
+			Table:   "t1",
+			TableID: 69,
+		},
+		TableInfo: &timodel.TableInfo{
+			ID:   69,
+			Name: pmodel.CIStr{O: "t10"},
+			Columns: []*timodel.ColumnInfo{
+				{
+					ID:        1,
+					Name:      pmodel.CIStr{O: "id"},
+					FieldType: *ft,
+					State:     timodel.StatePublic,
+				},
+			},
+		},
+	}
+
+	event := &DDLEvent{}
+	event.FromJob(job, preTableInfo, tableInfo)
+	require.Equal(t,
+		"ALTER TABLE `test1`.`t1` EXCHANGE PARTITION `p``0` WITH TABLE `test2`.`t2` WITHOUT VALIDATION",
+		event.Query)
+	require.Equal(t, event.Type, timodel.ActionExchangeTablePartition)
+}
+
 func TestSortRowChangedEvent(t *testing.T) {
 	events := []*RowChangedEvent{
 		{

pkg/quotes/quotes.go

@@ -28,6 +28,14 @@ func QuoteName(name string) string {
 	return "`" + EscapeName(name) + "`"
 }
 
+// UnquoteName removes one layer of MySQL identifier quoting and unescapes doubled backticks.
+func UnquoteName(name string) string {
+	if len(name) < 2 || name[0] != '`' || name[len(name)-1] != '`' {
+		return name
+	}
+	return strings.ReplaceAll(name[1:len(name)-1], "``", "`")
+}
+
 // EscapeName replaces all "`" in name with double "`"
 func EscapeName(name string) string {
 	return strings.Replace(name, "`", "``", -1)

pkg/quotes/quotes_test.go

@@ -72,3 +72,22 @@ func TestEscapeName(t *testing.T) {
 		require.Equal(t, testCase.expected, escaped)
 	}
 }
+
+func TestUnquoteName(t *testing.T) {
+	t.Parallel()
+
+	cases := []struct {
+		name     string
+		expected string
+	}{
+		{"tbl", "tbl"},
+		{"`tbl`", "tbl"},
+		{"`t``bl`", "t`bl"},
+		{"`t````bl`", "t``bl"},
+		{"", ""},
+	}
+	for _, testCase := range cases {
+		unquoted := UnquoteName(testCase.name)
+		require.Equal(t, testCase.expected, unquoted)
+	}
+}