*: fix lost index bug of insert on duplicate key update (#16672)

Fix issue #16669

When check the untouched index, we should also check the memory-buffer in the session too.

executor/union_scan_test.go

@@ -290,3 +290,41 @@ func (s *testSuite7) TestUnionScanForMemBufferReader(c *C) {
 	tk.MustQuery("select * from t1 use index(idx2);").Check(testkit.Rows("1 2 1"))
 	tk.MustExec("admin check table t1;")
 }
+
+func (s *testSuite7) TestForUpdateUntouchedIndex(c *C) {
+	tk := testkit.NewTestKit(c, s.store)
+	tk.MustExec("use test")
+	tk.MustExec("drop table if exists t")
+
+	checkFunc := func() {
+		tk.MustExec("begin")
+		tk.MustExec("insert into t values ('a', 1), ('b', 3), ('a', 2) on duplicate key update b = b + 1;")
+		tk.MustExec("commit")
+		tk.MustExec("admin check table t")
+
+		// Test for autocommit
+		tk.MustExec("set autocommit=0")
+		tk.MustExec("insert into t values ('a', 1), ('b', 3), ('a', 2) on duplicate key update b = b + 1;")
+		tk.MustExec("set autocommit=1")
+		tk.MustExec("admin check table t")
+	}
+
+	// Test for primary key.
+	tk.MustExec("create table t (a varchar(10) primary key,b int)")
+	checkFunc()
+
+	// Test for unique key.
+	tk.MustExec("drop table if exists t")
+	tk.MustExec("create table t (a varchar(10),b int, unique index(a))")
+	checkFunc()
+
+	// Test for on duplicate update also conflict too.
+	tk.MustExec("drop table if exists t")
+	tk.MustExec("create table t (a int,b int, unique index(a))")
+	tk.MustExec("begin")
+	_, err := tk.Exec("insert into t values (1, 1), (2, 2), (1, 3) on duplicate key update a = a + 1;")
+	c.Assert(err, NotNil)
+	c.Assert(err.Error(), Equals, "[kv:1062]Duplicate entry '2' for key 'a'")
+	tk.MustExec("commit")
+	tk.MustExec("admin check table t")
+}

kv/buffer_store.go

@@ -37,6 +37,14 @@ func NewBufferStore(r Retriever) *BufferStore {
 	}
 }
 
+// NewBufferStoreFrom creates a BufferStore from retriever and mem-buffer.
+func NewBufferStoreFrom(r Retriever, buf MemBuffer) *BufferStore {
+	return &BufferStore{
+		r:         r,
+		MemBuffer: buf,
+	}
+}
+
 // NewStagingBufferStore returns a BufferStore with buffer derived from the buffer.
 func NewStagingBufferStore(buf MemBuffer) *BufferStore {
 	return &BufferStore{

session/txn.go

@@ -307,6 +307,14 @@ func (st *TxnState) Get(ctx context.Context, k kv.Key) ([]byte, error) {
 	return val, nil
 }
 
+// GetMemBuffer overrides the Transaction interface.
+func (st *TxnState) GetMemBuffer() kv.MemBuffer {
+	if st.stmtBuf == nil || st.stmtBuf.Size() == 0 {
+		return st.Transaction.GetMemBuffer()
+	}
+	return kv.NewBufferStoreFrom(st.Transaction.GetMemBuffer(), st.stmtBuf)
+}
+
 // BatchGet overrides the Transaction interface.
 func (st *TxnState) BatchGet(ctx context.Context, keys []kv.Key) (map[string][]byte, error) {
 	bufferValues := make([][]byte, len(keys))