MySQL sink will generate duplicated replace if batch-replace-enable is disabled

[amyangfei]

What did you do?

• Create a changefeed with batch-replace-enable disabled
• Run insert or update in upstream
• MySQL sink will generate duplicated replace

What did you expect to see?

No response

What did you see instead?

For example

mysql root@127.0.0.1:test> insert into t2 values ();
Query OK, 1 row affected
Time: 0.005s

[2022/02/05 17:15:24.768 +08:00] [DEBUG] [mysql.go:722] ["prepare DMLs"] [rows="[{\"start-ts\":430978504702296065,\"commit-ts\":430978504702296066,\"row-id\":30004,\"table\":{\"db-name\":\"test\",\"tbl-name\":\"t2\",\"tbl-id\":57,\"is-partition\":false},\"table-info-version\":430978503889387522,\"replica-id\":0,\"columns\":[{\"name\":\"id\",\"type\":8,\"flag\":27,\"value\":30003}],\"pre-columns\":null}]"] [sqls="[\"REPLACE INTO `test`.`t2`(`id`) VALUES (?);\",\"REPLACE INTO `test`.`t2`(`id`) VALUES (?);\"]"] [values="[[30003],[30003]]"]
[2022/02/05 17:15:24.768 +08:00] [DEBUG] [mysql.go:566] ["exec row"] [sql="REPLACE INTO `test`.`t2`(`id`) VALUES (?);"] [args="[30003]"]
[2022/02/05 17:15:24.769 +08:00] [DEBUG] [mysql.go:566] ["exec row"] [sql="REPLACE INTO `test`.`t2`(`id`) VALUES (?);"] [args="[30003]"]
[2022/02/05 17:15:24.807 +08:00] [DEBUG] [mysql.go:593] ["Exec Rows succeeded"] [changefeed=test-cf] ["num of Rows"=1] [bucket=0]

Versions of the cluster

Upstream TiDB cluster version (execute SELECT tidb_version(); in a MySQL client):

master@2022/02/05

TiCDC version (execute cdc version):

master@596f558