sink(ticdc): fix duplicated replace when batch-replace is disabled (#…

…4502) close #4501
pingcap · Feb 7, 2022 · f710abb · f710abb
1 parent 596f558
commit f710abb
Show file tree

Hide file tree

Showing 2 changed files with 56 additions and 31 deletions.
diff --git a/cdc/sink/mysql.go b/cdc/sink/mysql.go
@@ -678,8 +678,6 @@ func (s *mysqlSink) prepareDMLs(rows []*model.RowChangedEvent, replicaID uint64,
 				}
 			} else {
 				query, args = prepareReplace(quoteTable, row.Columns, true /* appendPlaceHolder */, translateToInsert)
-				sqls = append(sqls, query)
-				values = append(values, args)
 				if query != "" {
 					sqls = append(sqls, query)
 					values = append(values, args)

diff --git a/cdc/sink/mysql_test.go b/cdc/sink/mysql_test.go
@@ -58,41 +58,68 @@ func TestPrepareDML(t *testing.T) {
 	testCases := []struct {
 		input    []*model.RowChangedEvent
 		expected *preparedDMLs
-	}{{
-		input:    []*model.RowChangedEvent{},
-		expected: &preparedDMLs{sqls: []string{}, values: [][]interface{}{}},
-	}, {
-		input: []*model.RowChangedEvent{
-			{
-				StartTs:  418658114257813514,
-				CommitTs: 418658114257813515,
-				Table:    &model.TableName{Schema: "common_1", Table: "uk_without_pk"},
-				PreColumns: []*model.Column{nil, {
-					Name:  "a1",
-					Type:  mysql.TypeLong,
-					Flag:  model.BinaryFlag | model.MultipleKeyFlag | model.HandleKeyFlag,
-					Value: 1,
-				}, {
-					Name:  "a3",
-					Type:  mysql.TypeLong,
-					Flag:  model.BinaryFlag | model.MultipleKeyFlag | model.HandleKeyFlag,
-					Value: 1,
-				}},
-				IndexColumns: [][]int{{1, 2}},
+	}{
+		{
+			input:    []*model.RowChangedEvent{},
+			expected: &preparedDMLs{sqls: []string{}, values: [][]interface{}{}},
+		}, {
+			input: []*model.RowChangedEvent{
+				{
+					StartTs:  418658114257813514,
+					CommitTs: 418658114257813515,
+					Table:    &model.TableName{Schema: "common_1", Table: "uk_without_pk"},
+					PreColumns: []*model.Column{nil, {
+						Name:  "a1",
+						Type:  mysql.TypeLong,
+						Flag:  model.BinaryFlag | model.MultipleKeyFlag | model.HandleKeyFlag,
+						Value: 1,
+					}, {
+						Name:  "a3",
+						Type:  mysql.TypeLong,
+						Flag:  model.BinaryFlag | model.MultipleKeyFlag | model.HandleKeyFlag,
+						Value: 1,
+					}},
+					IndexColumns: [][]int{{1, 2}},
+				},
+			},
+			expected: &preparedDMLs{
+				sqls:     []string{"DELETE FROM `common_1`.`uk_without_pk` WHERE `a1` = ? AND `a3` = ? LIMIT 1;"},
+				values:   [][]interface{}{{1, 1}},
+				rowCount: 1,
+			},
+		}, {
+			input: []*model.RowChangedEvent{
+				{
+					StartTs:  418658114257813516,
+					CommitTs: 418658114257813517,
+					Table:    &model.TableName{Schema: "common_1", Table: "uk_without_pk"},
+					Columns: []*model.Column{nil, {
+						Name:  "a1",
+						Type:  mysql.TypeLong,
+						Flag:  model.BinaryFlag | model.MultipleKeyFlag | model.HandleKeyFlag,
+						Value: 2,
+					}, {
+						Name:  "a3",
+						Type:  mysql.TypeLong,
+						Flag:  model.BinaryFlag | model.MultipleKeyFlag | model.HandleKeyFlag,
+						Value: 2,
+					}},
+					IndexColumns: [][]int{{1, 2}},
+				},
+			},
+			expected: &preparedDMLs{
+				sqls:     []string{"REPLACE INTO `common_1`.`uk_without_pk`(`a1`,`a3`) VALUES (?,?);"},
+				values:   [][]interface{}{{2, 2}},
+				rowCount: 1,
 			},
 		},
-		expected: &preparedDMLs{
-			sqls:     []string{"DELETE FROM `common_1`.`uk_without_pk` WHERE `a1` = ? AND `a3` = ? LIMIT 1;"},
-			values:   [][]interface{}{{1, 1}},
-			rowCount: 1,
-		},
-	}}
+	}
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 	ms := newMySQLSink4Test(ctx, t)
-	for i, tc := range testCases {
+	for _, tc := range testCases {
 		dmls := ms.prepareDMLs(tc.input, 0, 0)
-		require.Equal(t, tc.expected, dmls, tc.expected, fmt.Sprintf("%d", i))
+		require.Equal(t, tc.expected, dmls)
 	}
 }