Merge pull request #1011 from pingcap/shenli/fix-989

executor: Fix replace statement bug when handle is changed.

executor/executor_test.go

@@ -522,6 +522,16 @@ func (s *testSuite) TestReplace(c *C) {
 	replacePrimaryKeySQL = `replace into replace_test_5 set c1=1, c2=2;`
 	tk.MustExec(replacePrimaryKeySQL)
 	c.Assert(int64(tk.Se.AffectedRows()), Equals, int64(1))
+
+	// For Issue989
+	issue989SQL := `CREATE TABLE tIssue989 (a int, b int, PRIMARY KEY(a), UNIQUE KEY(b));`
+	tk.MustExec(issue989SQL)
+	issue989SQL = `insert into tIssue989 (a, b) values (1, 2);`
+	tk.MustExec(issue989SQL)
+	issue989SQL = `replace into tIssue989(a, b) values (111, 2);`
+	tk.MustExec(issue989SQL)
+	r := tk.MustQuery("select * from tIssue989;")
+	r.Check(testkit.Rows("111 2"))
 }
 
 func (s *testSuite) TestSelectWithoutFrom(c *C) {

executor/executor_write.go

@@ -830,6 +830,40 @@ func (e *ReplaceExec) Next() (*Row, error) {
 		// While the insertion fails because a duplicate-key error occurs for a primary key or unique index,
 		// a storage engine may perform the REPLACE as an update rather than a delete plus insert.
 		// See: http://dev.mysql.com/doc/refman/5.7/en/replace.html.
+
+		// If PKIsHandle && PK value is changed, we should delete the row and insert a new row.
+		if e.Table.Meta().PKIsHandle {
+			changeHandle := false
+			oldRow, err1 := e.Table.Row(e.ctx, h)
+			if err1 != nil {
+				return nil, errors.Trace(err1)
+			}
+			for i, col := range e.Table.Cols() {
+				if !col.IsPKHandleColumn(e.Table.Meta()) {
+					continue
+				}
+				// Check if PK value changes.
+				cmp, err2 := row[i].CompareDatum(oldRow[i])
+				if err2 != nil {
+					return nil, errors.Trace(err2)
+				}
+				if cmp != 0 {
+					changeHandle = true
+				}
+				break
+			}
+			if changeHandle {
+				// Remove old data and insert new data.
+				err = e.Table.RemoveRecord(e.ctx, h, oldRow)
+				if err != nil {
+					return nil, errors.Trace(err)
+				}
+				_, err = e.Table.AddRecord(e.ctx, row)
+				variable.GetSessionVars(e.ctx).AddAffectedRows(1)
+				continue
+			}
+		}
+		// Replace the current row data.
 		if err = e.replaceRow(h, row); err != nil {
 			return nil, errors.Trace(err)
 		}