kv: merge batch get results to fix insert ignore in a transaction block

executor/admin.go

@@ -361,7 +361,7 @@ func (e *RecoverIndexExec) batchMarkDup(txn kv.Transaction, rows []recoverRows)
 		distinctFlags[i] = distinct
 	}
 
-	values, err := txn.GetSnapshot().BatchGet(e.batchKeys)
+	values, err := kv.BatchGetValues(txn, e.batchKeys)
 	if err != nil {
 		return errors.Trace(err)
 	}
@@ -485,7 +485,7 @@ func (e *CleanupIndexExec) batchGetRecord(txn kv.Transaction) (map[string][]byte
 	for handle := range e.idxValues {
 		e.batchKeys = append(e.batchKeys, e.table.RecordKey(handle))
 	}
-	values, err := txn.GetSnapshot().BatchGet(e.batchKeys)
+	values, err := kv.BatchGetValues(txn, e.batchKeys)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}

executor/write.go

@@ -965,7 +965,7 @@ func batchMarkDupRows(ctx sessionctx.Context, t table.Table, rows [][]types.Datu
 			batchKeys = append(batchKeys, k.key)
 		}
 	}
-	values, err := ctx.Txn().GetSnapshot().BatchGet(batchKeys)
+	values, err := kv.BatchGetValues(ctx.Txn(), batchKeys)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}

executor/write_test.go

@@ -409,6 +409,40 @@ func (s *testSuite) TestInsertIgnore(c *C) {
 	c.Assert(err, IsNil)
 	r = tk.MustQuery("SHOW WARNINGS")
 	r.Check(testkit.Rows("Warning 1062 Duplicate entry '1' for key 'PRIMARY'"))
+
+	testSQL = `drop table if exists test;
+create table test (i int primary key, j int unique);
+begin;
+insert into test values (1,1);
+insert ignore into test values (2,1);
+commit;`
+	tk.MustExec(testSQL)
+	testSQL = `select * from test;`
+	r = tk.MustQuery(testSQL)
+	r.Check(testkit.Rows("1 1"))
+
+	testSQL = `delete from test;
+insert into test values (1, 1);
+begin;
+delete from test where i = 1;
+insert ignore into test values (2, 1);
+commit;`
+	tk.MustExec(testSQL)
+	testSQL = `select * from test;`
+	r = tk.MustQuery(testSQL)
+	r.Check(testkit.Rows("2 1"))
+
+	testSQL = `delete from test;
+insert into test values (1, 1);
+begin;
+update test set i = 2, j = 2 where i = 1;
+insert ignore into test values (1, 3);
+insert ignore into test values (2, 4);
+commit;`
+	tk.MustExec(testSQL)
+	testSQL = `select * from test order by i;`
+	r = tk.MustQuery(testSQL)
+	r.Check(testkit.Rows("1 3", "2 2"))
 }
 
 func (s *testSuite) TestReplace(c *C) {

kv/txn.go

@@ -88,3 +88,31 @@ func BackOff(attempts uint) int {
 	time.Sleep(sleep)
 	return int(sleep)
 }
+
+// BatchGetValues gets values in batch.
+// The values from buffer in transaction and the values from the storage node are merged together.
+func BatchGetValues(txn Transaction, keys []Key) (map[string][]byte, error) {
+	bufferValues := make(map[string][]byte, len(keys))
+	shrinkKeys := make([]Key, 0, len(keys))
+	for _, key := range keys {
+		val, err := txn.GetMemBuffer().Get(key)
+		if IsErrNotFound(err) {
+			shrinkKeys = append(shrinkKeys, key)
+			continue
+		}
+		if err != nil {
+			return nil, errors.Trace(err)
+		}
+		if len(val) != 0 {
+			bufferValues[string(key)] = val
+		}
+	}
+	storageValues, err := txn.GetSnapshot().BatchGet(shrinkKeys)
+	if err != nil {
+		return nil, errors.Trace(err)
+	}
+	for key, val := range bufferValues {
+		storageValues[string(key)] = val
+	}
+	return storageValues, nil
+}