session: migrate test-infra to testify for testSessionSerialSuite

session/schema_test.go

@@ -880,3 +880,179 @@ func TestSchemaCheckerTempTable(t *testing.T) {
 	err = tk1.ExecToErr(`commit;`)
 	require.True(t, terror.ErrorEqual(err, domain.ErrInfoSchemaChanged), fmt.Sprintf("err %v", err))
 }
+
+func TestGlobalAndLocalTxn(t *testing.T) {
+	skipIfWithRealTiKV(t)
+	// Because the PD config of check_dev_2 test is not compatible with local/global txn yet,
+	// so we will skip this test for now.
+	store, clean := testkit.CreateMockStore(t)
+	defer clean()
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("use test")
+	tk.MustExec("set global tidb_enable_local_txn = on;")
+	tk.MustExec("drop table if exists t1")
+	tk.MustExec("drop placement policy if exists p1")
+	tk.MustExec("drop placement policy if exists p2")
+	tk.MustExec("create placement policy p1 leader_constraints='[+zone=dc-1]'")
+	tk.MustExec("create placement policy p2 leader_constraints='[+zone=dc-2]'")
+	tk.MustExec(`create table t1 (c int)
+PARTITION BY RANGE (c) (
+	PARTITION p0 VALUES LESS THAN (100) placement policy p1,
+	PARTITION p1 VALUES LESS THAN (200) placement policy p2
+);`)
+	defer func() {
+		tk.MustExec("drop table if exists t1")
+		tk.MustExec("drop placement policy if exists p1")
+		tk.MustExec("drop placement policy if exists p2")
+	}()
+
+	// set txn_scope to global
+	tk.MustExec(fmt.Sprintf("set @@session.txn_scope = '%s';", kv.GlobalTxnScope))
+	result := tk.MustQuery("select @@txn_scope;")
+	result.Check(testkit.Rows(kv.GlobalTxnScope))
+
+	// test global txn auto commit
+	tk.MustExec("insert into t1 (c) values (1)") // write dc-1 with global scope
+	result = tk.MustQuery("select * from t1")    // read dc-1 and dc-2 with global scope
+	require.Equal(t, 1, len(result.Rows()))
+
+	// begin and commit with global txn scope
+	tk.MustExec("begin")
+	txn, err := tk.Session().Txn(true)
+	require.NoError(t, err)
+	require.Equal(t, kv.GlobalTxnScope, tk.Session().GetSessionVars().TxnCtx.TxnScope)
+	require.True(t, txn.Valid())
+	tk.MustExec("insert into t1 (c) values (1)") // write dc-1 with global scope
+	result = tk.MustQuery("select * from t1")    // read dc-1 and dc-2 with global scope
+	require.Equal(t, 2, len(result.Rows()))
+	require.True(t, txn.Valid())
+	tk.MustExec("commit")
+	result = tk.MustQuery("select * from t1")
+	require.Equal(t, 2, len(result.Rows()))
+
+	// begin and rollback with global txn scope
+	tk.MustExec("begin")
+	txn, err = tk.Session().Txn(true)
+	require.NoError(t, err)
+	require.Equal(t, kv.GlobalTxnScope, tk.Session().GetSessionVars().TxnCtx.TxnScope)
+	require.True(t, txn.Valid())
+	tk.MustExec("insert into t1 (c) values (101)") // write dc-2 with global scope
+	result = tk.MustQuery("select * from t1")      // read dc-1 and dc-2 with global scope
+	require.Equal(t, 3, len(result.Rows()))
+	require.True(t, txn.Valid())
+	tk.MustExec("rollback")
+	result = tk.MustQuery("select * from t1")
+	require.Equal(t, 2, len(result.Rows()))
+
+	timeBeforeWriting := time.Now()
+	tk.MustExec("insert into t1 (c) values (101)") // write dc-2 with global scope
+	result = tk.MustQuery("select * from t1")      // read dc-1 and dc-2 with global scope
+	require.Equal(t, 3, len(result.Rows()))
+
+	failpoint.Enable("tikvclient/injectTxnScope", `return("dc-1")`)
+	defer failpoint.Disable("tikvclient/injectTxnScope")
+	// set txn_scope to local
+	tk.MustExec("set @@session.txn_scope = 'local';")
+	result = tk.MustQuery("select @@txn_scope;")
+	result.Check(testkit.Rows("local"))
+
+	// test local txn auto commit
+	tk.MustExec("insert into t1 (c) values (1)")          // write dc-1 with dc-1 scope
+	result = tk.MustQuery("select * from t1 where c = 1") // point get dc-1 with dc-1 scope
+	require.Equal(t, 3, len(result.Rows()))
+	result = tk.MustQuery("select * from t1 where c < 100") // read dc-1 with dc-1 scope
+	require.Equal(t, 3, len(result.Rows()))
+
+	// begin and commit with dc-1 txn scope
+	tk.MustExec("begin")
+	txn, err = tk.Session().Txn(true)
+	require.NoError(t, err)
+	require.Equal(t, "dc-1", tk.Session().GetSessionVars().CheckAndGetTxnScope())
+	require.True(t, txn.Valid())
+	tk.MustExec("insert into t1 (c) values (1)")            // write dc-1 with dc-1 scope
+	result = tk.MustQuery("select * from t1 where c < 100") // read dc-1 with dc-1 scope
+	require.Equal(t, 4, len(result.Rows()))
+	require.True(t, txn.Valid())
+	tk.MustExec("commit")
+	result = tk.MustQuery("select * from t1 where c < 100")
+	require.Equal(t, 4, len(result.Rows()))
+
+	// begin and rollback with dc-1 txn scope
+	tk.MustExec("begin")
+	txn, err = tk.Session().Txn(true)
+	require.NoError(t, err)
+	require.Equal(t, "dc-1", tk.Session().GetSessionVars().CheckAndGetTxnScope())
+	require.True(t, txn.Valid())
+	tk.MustExec("insert into t1 (c) values (1)")            // write dc-1 with dc-1 scope
+	result = tk.MustQuery("select * from t1 where c < 100") // read dc-1 with dc-1 scope
+	require.Equal(t, 5, len(result.Rows()))
+	require.True(t, txn.Valid())
+	tk.MustExec("rollback")
+	result = tk.MustQuery("select * from t1 where c < 100")
+	require.Equal(t, 4, len(result.Rows()))
+
+	// test wrong scope local txn auto commit
+	_, err = tk.Exec("insert into t1 (c) values (101)") // write dc-2 with dc-1 scope
+	require.Error(t, err)
+	require.Regexp(t, ".*out of txn_scope.*", err)
+	err = tk.ExecToErr("select * from t1 where c = 101") // point get dc-2 with dc-1 scope
+	require.Error(t, err)
+	require.Regexp(t, ".*can not be read by.*", err)
+	err = tk.ExecToErr("select * from t1 where c > 100") // read dc-2 with dc-1 scope
+	require.Error(t, err)
+	require.Regexp(t, ".*can not be read by.*", err)
+	tk.MustExec("begin")
+	err = tk.ExecToErr("select * from t1 where c = 101") // point get dc-2 with dc-1 scope
+	require.Error(t, err)
+	require.Regexp(t, ".*can not be read by.*", err)
+	err = tk.ExecToErr("select * from t1 where c > 100") // read dc-2 with dc-1 scope
+	require.Error(t, err)
+	require.Regexp(t, ".*can not be read by.*", err)
+	tk.MustExec("commit")
+
+	// begin and commit reading & writing the data in dc-2 with dc-1 txn scope
+	tk.MustExec("begin")
+	txn, err = tk.Session().Txn(true)
+	require.NoError(t, err)
+	require.Equal(t, "dc-1", tk.Session().GetSessionVars().CheckAndGetTxnScope())
+	require.True(t, txn.Valid())
+	tk.MustExec("insert into t1 (c) values (101)")       // write dc-2 with dc-1 scope
+	err = tk.ExecToErr("select * from t1 where c > 100") // read dc-2 with dc-1 scope
+	require.Error(t, err)
+	require.Regexp(t, ".*can not be read by.*", err)
+	tk.MustExec("insert into t1 (c) values (99)")           // write dc-1 with dc-1 scope
+	result = tk.MustQuery("select * from t1 where c < 100") // read dc-1 with dc-1 scope
+	require.Equal(t, 5, len(result.Rows()))
+	require.True(t, txn.Valid())
+	_, err = tk.Exec("commit")
+	require.Error(t, err)
+	require.Regexp(t, ".*out of txn_scope.*", err)
+	// Won't read the value 99 because the previous commit failed
+	result = tk.MustQuery("select * from t1 where c < 100") // read dc-1 with dc-1 scope
+	require.Equal(t, 4, len(result.Rows()))
+
+	// Stale Read will ignore the cross-dc txn scope.
+	require.Equal(t, "dc-1", tk.Session().GetSessionVars().CheckAndGetTxnScope())
+	result = tk.MustQuery("select @@txn_scope;")
+	result.Check(testkit.Rows("local"))
+	err = tk.ExecToErr("select * from t1 where c > 100") // read dc-2 with dc-1 scope
+	require.Error(t, err)
+	require.Regexp(t, ".*can not be read by.*", err)
+	// Read dc-2 with Stale Read (in dc-1 scope)
+	timestamp := timeBeforeWriting.Format(time.RFC3339Nano)
+	// TODO: check the result of Stale Read when we figure out how to make the time precision more accurate.
+	tk.MustExec(fmt.Sprintf("select * from t1 AS OF TIMESTAMP '%s' where c = 101", timestamp))
+	tk.MustExec(fmt.Sprintf("select * from t1 AS OF TIMESTAMP '%s' where c > 100", timestamp))
+	tk.MustExec(fmt.Sprintf("START TRANSACTION READ ONLY AS OF TIMESTAMP '%s'", timestamp))
+	tk.MustExec("select * from t1 where c = 101")
+	tk.MustExec("select * from t1 where c > 100")
+	tk.MustExec("commit")
+	tk.MustExec("set @@tidb_replica_read='closest-replicas'")
+	tk.MustExec(fmt.Sprintf("select * from t1 AS OF TIMESTAMP '%s' where c > 100", timestamp))
+	tk.MustExec(fmt.Sprintf("START TRANSACTION READ ONLY AS OF TIMESTAMP '%s'", timestamp))
+	tk.MustExec("select * from t1 where c = 101")
+	tk.MustExec("select * from t1 where c > 100")
+	tk.MustExec("commit")
+
+	tk.MustExec("set global tidb_enable_local_txn = off;")
+}