This is an automated cherry-pick of #57282

Signed-off-by: ti-chi-bot <ti-community-prow-bot@tidb.io>

resolve conflict

Signed-off-by: you06 <you1474600@gmail.com>

fix pr

Signed-off-by: you06 <you1474600@gmail.com>

DEPS.bzl

@@ -6807,13 +6807,13 @@ def go_deps():
         name = "com_github_tikv_client_go_v2",
         build_file_proto_mode = "disable_global",
         importpath = "github.com/tikv/client-go/v2",
-        sha256 = "6701afd9ef373b22010ff1c3aeb91fca8a6165341c6a38dd31a00ed10d24f314",
-        strip_prefix = "github.com/tikv/client-go/v2@v2.0.8-0.20240913090512-3777c384feb1",
+        sha256 = "1fd861359a159d20435b7e4ed81f520b7235941aa2c5e059e05f338f1f781664",
+        strip_prefix = "github.com/tikv/client-go/v2@v2.0.8-0.20241125064226-08d0b3b826b3",
         urls = [
-            "http://bazel-cache.pingcap.net:8080/gomod/github.com/tikv/client-go/v2/com_github_tikv_client_go_v2-v2.0.8-0.20240913090512-3777c384feb1.zip",
-            "http://ats.apps.svc/gomod/github.com/tikv/client-go/v2/com_github_tikv_client_go_v2-v2.0.8-0.20240913090512-3777c384feb1.zip",
-            "https://cache.hawkingrei.com/gomod/github.com/tikv/client-go/v2/com_github_tikv_client_go_v2-v2.0.8-0.20240913090512-3777c384feb1.zip",
-            "https://storage.googleapis.com/pingcapmirror/gomod/github.com/tikv/client-go/v2/com_github_tikv_client_go_v2-v2.0.8-0.20240913090512-3777c384feb1.zip",
+            "http://bazel-cache.pingcap.net:8080/gomod/github.com/tikv/client-go/v2/com_github_tikv_client_go_v2-v2.0.8-0.20241125064226-08d0b3b826b3.zip",
+            "http://ats.apps.svc/gomod/github.com/tikv/client-go/v2/com_github_tikv_client_go_v2-v2.0.8-0.20241125064226-08d0b3b826b3.zip",
+            "https://cache.hawkingrei.com/gomod/github.com/tikv/client-go/v2/com_github_tikv_client_go_v2-v2.0.8-0.20241125064226-08d0b3b826b3.zip",
+            "https://storage.googleapis.com/pingcapmirror/gomod/github.com/tikv/client-go/v2/com_github_tikv_client_go_v2-v2.0.8-0.20241125064226-08d0b3b826b3.zip",
         ],
     )
     go_repository(

go.mod

@@ -106,7 +106,7 @@ require (
 	github.com/tdakkota/asciicheck v0.2.0
 	github.com/tiancaiamao/appdash v0.0.0-20181126055449-889f96f722a2
 	github.com/tidwall/btree v1.7.0
-	github.com/tikv/client-go/v2 v2.0.8-0.20240913090512-3777c384feb1
+	github.com/tikv/client-go/v2 v2.0.8-0.20241125064226-08d0b3b826b3
 	github.com/tikv/pd/client v0.0.0-20240806105739-10ecdbe92b55
 	github.com/timakin/bodyclose v0.0.0-20240125160201-f835fa56326a
 	github.com/twmb/murmur3 v1.1.6

go.sum

@@ -783,8 +783,8 @@ github.com/tiancaiamao/gp v0.0.0-20221230034425-4025bc8a4d4a h1:J/YdBZ46WKpXsxsW
 github.com/tiancaiamao/gp v0.0.0-20221230034425-4025bc8a4d4a/go.mod h1:h4xBhSNtOeEosLJ4P7JyKXX7Cabg7AVkWCK5gV2vOrM=
 github.com/tidwall/btree v1.7.0 h1:L1fkJH/AuEh5zBnnBbmTwQ5Lt+bRJ5A8EWecslvo9iI=
 github.com/tidwall/btree v1.7.0/go.mod h1:twD9XRA5jj9VUQGELzDO4HPQTNJsoWWfYEL+EUQ2cKY=
-github.com/tikv/client-go/v2 v2.0.8-0.20240913090512-3777c384feb1 h1:KmLTcRXq+sMZ5tKKLLX6H+bs1fUQB6uYaRg52Wkt6n0=
-github.com/tikv/client-go/v2 v2.0.8-0.20240913090512-3777c384feb1/go.mod h1:+vXk4Aex17GnI8gfSMPxrL0SQLbBYgP3Db4FvHiImwM=
+github.com/tikv/client-go/v2 v2.0.8-0.20241125064226-08d0b3b826b3 h1:M15i12ypNUsxuOPJ/zcCxajncJSs4yz0+mYB2RSJS0w=
+github.com/tikv/client-go/v2 v2.0.8-0.20241125064226-08d0b3b826b3/go.mod h1:+vXk4Aex17GnI8gfSMPxrL0SQLbBYgP3Db4FvHiImwM=
 github.com/tikv/pd/client v0.0.0-20240806105739-10ecdbe92b55 h1:+1unfy0TcJJtud3d7BuYsvNG6tPVuXIH+WiIFhOx1Kc=
 github.com/tikv/pd/client v0.0.0-20240806105739-10ecdbe92b55/go.mod h1:1zqLOMhnkZIpBLj2oXOO2bWvtXhb12OmYr+cPkjQ6tI=
 github.com/timakin/bodyclose v0.0.0-20240125160201-f835fa56326a h1:A6uKudFIfAEpoPdaal3aSqGxBzLyU8TqyXImLwo6dIo=

pkg/executor/adapter.go

@@ -603,7 +603,7 @@ func (a *ExecStmt) handleStmtForeignKeyTrigger(ctx context.Context, e exec.Execu
 	if stmtCtx.ForeignKeyTriggerCtx.HasFKCascades {
 		// If the ExecStmt has foreign key cascade to be executed, we need call `StmtCommit` to commit the ExecStmt itself
 		// change first.
-		// Since `UnionScanExec` use `SnapshotIter` and `SnapshotGetter` to read txn mem-buffer, if we don't  do `StmtCommit`,
+		// Since `UnionScanExec` use `SnapshotIter` and `SnapshotGetter` to read txn mem-buffer, if we don't do `StmtCommit`,
 		// then the fk cascade executor can't read the mem-buffer changed by the ExecStmt.
 		a.Ctx.StmtCommit(ctx)
 	}

pkg/executor/union_scan_test.go

@@ -295,6 +295,21 @@ func TestIssue32422(t *testing.T) {
 	tk.MustExec("rollback")
 }
 
+func TestSnapshotWithConcurrentWrite(t *testing.T) {
+	store := testkit.CreateMockStore(t)
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("use test")
+	tk.MustExec("create table t1 (id int auto_increment key, b int, index(b));")
+
+	tk.MustExec("begin")
+	tk.MustExec("insert into t1 (b) values (1),(2),(3),(4),(5),(6),(7),(8);")
+	for j := 0; j < 16; j++ {
+		tk.MustExec("insert into t1 (b) select /*+ use_index(t1, b) */ id from t1;")
+	}
+	tk.MustQuery("select count(1) from t1").Check(testkit.Rows("524288")) // 8 * 2^16 rows
+	tk.MustExec("rollback")
+}
+
 func BenchmarkUnionScanRead(b *testing.B) {
 	store := testkit.CreateMockStore(b)
 

pkg/session/test/txn/BUILD.bazel

@@ -9,7 +9,7 @@ go_test(
     ],
     flaky = True,
     race = "on",
-    shard_count = 8,
+    shard_count = 10,
     deps = [
         "//pkg/config",
         "//pkg/kv",

pkg/session/test/txn/txn_test.go

@@ -508,3 +508,74 @@ func TestInTrans(t *testing.T) {
 	tk.MustExec("rollback")
 	require.False(t, txn.Valid())
 }
+
+func TestMemBufferSnapshotRead(t *testing.T) {
+	store := testkit.CreateMockStore(t)
+
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("use test")
+
+	tk.MustExec("drop table if exists t;")
+	tk.MustExec("create table t(a int primary key, b int, index i(b));")
+
+	tk.MustExec("set session tidb_distsql_scan_concurrency = 1;")
+	tk.MustExec("set session tidb_index_lookup_join_concurrency = 1;")
+	tk.MustExec("set session tidb_projection_concurrency=1;")
+	tk.MustExec("set session tidb_init_chunk_size=1;")
+	tk.MustExec("set session tidb_max_chunk_size=40;")
+	tk.MustExec("set session tidb_index_join_batch_size = 10")
+
+	tk.MustExec("begin;")
+	// write (0, 0), (1, 1), ... ,(100, 100) into membuffer
+	var sb strings.Builder
+	sb.WriteString("insert into t values ")
+	for i := 0; i <= 100; i++ {
+		if i > 0 {
+			sb.WriteString(", ")
+		}
+		sb.WriteString(fmt.Sprintf("(%d, %d)", i, i))
+	}
+	tk.MustExec(sb.String())
+
+	// insert on duplicate key statement should update the table to (0, 100), (1, 99), ... (100, 0)
+	// This statement will create UnionScan dynamically during execution, and some UnionScan will see staging data(should be bypassed),
+	// so it relies on correct snapshot read to get the expected result.
+	tk.MustExec("insert into t (select /*+ INL_JOIN(t1) */ 100 - t1.a as a, t1.b from t t1, (select a, b from t) t2 where t1.b = t2.b) on duplicate key update b = values(b)")
+
+	require.Empty(t, tk.MustQuery("select a, b from t where a + b != 100;").Rows())
+	tk.MustExec("commit;")
+	require.Empty(t, tk.MustQuery("select a, b from t where a + b != 100;").Rows())
+
+	tk.MustExec("set session tidb_distsql_scan_concurrency = default;")
+	tk.MustExec("set session tidb_index_lookup_join_concurrency = default;")
+	tk.MustExec("set session tidb_projection_concurrency=default;")
+	tk.MustExec("set session tidb_init_chunk_size=default;")
+	tk.MustExec("set session tidb_max_chunk_size=default;")
+	tk.MustExec("set session tidb_index_join_batch_size = default")
+}
+
+func TestMemBufferCleanupMemoryLeak(t *testing.T) {
+	// Test if cleanup memory will cause a memory leak.
+	// When an in-txn statement fails, TiDB cleans up the mutations from this statement.
+	// If there's a memory leak, the memory usage could increase uncontrollably with retries.
+	store := testkit.CreateMockStore(t)
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("use test")
+	tk.MustExec("create table t(a varchar(255) primary key)")
+	key1 := strings.Repeat("a", 255)
+	key2 := strings.Repeat("b", 255)
+	tk.MustExec(`set global tidb_mem_oom_action='cancel'`)
+	tk.MustExec("set session tidb_mem_quota_query=10240")
+	tk.MustExec("begin")
+	tk.MustExec("insert into t values(?)", key2)
+	for i := 0; i < 100; i++ {
+		// The insert statement will fail because of the duplicate key error.
+		err := tk.ExecToErr("insert into t values(?), (?)", key1, key2)
+		require.Error(t, err)
+		if strings.Contains(err.Error(), "Duplicate") {
+			continue
+		}
+		require.NoError(t, err)
+	}
+	tk.MustExec("commit")
+}

tests/integrationtest/r/executor/ddl.result

@@ -379,7 +379,10 @@ create or replace definer='root'@'localhost' view v_nested as select * from v_ne
 Error 1146 (42S02): Table 'executor__ddl.v_nested' doesn't exist
 drop table test_v_nested;
 drop view v_nested, v_nested2;
-create view v_stale as select * from source_table as of timestamp current_timestamp(3);
+select sleep(1);
+sleep(1)
+0
+create view v_stale as select * from source_table as of timestamp date_sub(current_timestamp(3), interval 1 second);
 Error 1356 (HY000): View 'executor__ddl.v_stale' references invalid table(s) or column(s) or function(s) or definer/invoker of view lack rights to use them
 drop view if exists v1,v2;
 drop table if exists t1;

tests/integrationtest/t/executor/ddl.test

@@ -313,8 +313,9 @@ create or replace definer='root'@'localhost' view v_nested as select * from v_ne
 drop table test_v_nested;
 drop view v_nested, v_nested2;
 ## Refer https://github.com/pingcap/tidb/issues/25876
+select sleep(1);
 -- error 1356
-create view v_stale as select * from source_table as of timestamp current_timestamp(3);
+create view v_stale as select * from source_table as of timestamp date_sub(current_timestamp(3), interval 1 second);
 ## Refer https://github.com/pingcap/tidb/issues/32682
 drop view if exists v1,v2;
 drop table if exists t1;