realtikvtest: dedicated package for real tikv tests (#34578)

ref #28328

.github/workflows/integration-test-with-real-tikv.yml

@@ -33,3 +33,25 @@ jobs:
           export log_level=error
           make failpoint-enable
           go test ./tests/pessimistictest -v -with-real-tikv -timeout 20m
+
+  realtikv-test:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - uses: actions/setup-go@v2
+        with:
+          go-version: 1.18
+      - name: Setup TiKV cluster
+        run: |
+          # Disable pipelined pessimistic lock temporarily until tikv#11649 is resolved
+          echo -e "[pessimistic-txn]\npipelined = false\n" > tikv.toml
+          echo -e "[raftdb]\nmax-open-files = 20480\n" >> tikv.toml
+          echo -e "[rocksdb]\nmax-open-files = 20480\n" >> tikv.toml
+          curl --proto '=https' --tlsv1.2 -sSf https://tiup-mirrors.pingcap.com/install.sh | sh
+          $HOME/.tiup/bin/tiup playground nightly --mode tikv-slim --kv 3 --pd 3 --without-monitor --kv.config tikv.toml &
+          curl --retry-connrefused --retry 5 --retry-delay 5 http://127.0.0.1:2379
+      - name: Run Tests
+        run: |
+          export log_level=error
+          make failpoint-enable
+          go test ./tests/realtikvtest -v -with-real-tikv -timeout 30m

session/nontransactional_test.go

@@ -28,7 +28,7 @@ import (
 )
 
 func TestNonTransactionalDeleteSharding(t *testing.T) {
-	store, clean := createMockStoreAndSetup(t)
+	store, clean := testkit.CreateMockStore(t)
 	defer clean()
 	tk := testkit.NewTestKit(t, store)
 	tk.MustExec("set @@tidb_max_chunk_size=35")
@@ -70,7 +70,7 @@ func TestNonTransactionalDeleteSharding(t *testing.T) {
 }
 
 func TestNonTransactionalDeleteDryRun(t *testing.T) {
-	store, clean := createMockStoreAndSetup(t)
+	store, clean := testkit.CreateMockStore(t)
 	defer clean()
 	tk := testkit.NewTestKit(t, store)
 	tk.MustExec("set @@tidb_max_chunk_size=35")
@@ -89,7 +89,7 @@ func TestNonTransactionalDeleteDryRun(t *testing.T) {
 }
 
 func TestNonTransactionalDeleteErrorMessage(t *testing.T) {
-	store, clean := createMockStoreAndSetup(t)
+	store, clean := testkit.CreateMockStore(t)
 	defer clean()
 	tk := testkit.NewTestKit(t, store)
 	tk.MustExec("set @@tidb_max_chunk_size=35")
@@ -124,7 +124,7 @@ func TestNonTransactionalDeleteErrorMessage(t *testing.T) {
 }
 
 func TestNonTransactionalDeleteSplitOnTiDBRowID(t *testing.T) {
-	store, clean := createMockStoreAndSetup(t)
+	store, clean := testkit.CreateMockStore(t)
 	defer clean()
 	tk := testkit.NewTestKit(t, store)
 	tk.MustExec("set @@tidb_max_chunk_size=35")
@@ -138,7 +138,7 @@ func TestNonTransactionalDeleteSplitOnTiDBRowID(t *testing.T) {
 }
 
 func TestNonTransactionalDeleteNull(t *testing.T) {
-	store, clean := createMockStoreAndSetup(t)
+	store, clean := testkit.CreateMockStore(t)
 	defer clean()
 	tk := testkit.NewTestKit(t, store)
 	tk.MustExec("set @@tidb_max_chunk_size=35")
@@ -161,7 +161,7 @@ func TestNonTransactionalDeleteNull(t *testing.T) {
 }
 
 func TestNonTransactionalDeleteSmallBatch(t *testing.T) {
-	store, clean := createMockStoreAndSetup(t)
+	store, clean := testkit.CreateMockStore(t)
 	defer clean()
 	tk := testkit.NewTestKit(t, store)
 	tk.MustExec("set @@tidb_max_chunk_size=1024")
@@ -177,7 +177,7 @@ func TestNonTransactionalDeleteSmallBatch(t *testing.T) {
 }
 
 func TestNonTransactionalDeleteShardOnGeneratedColumn(t *testing.T) {
-	store, clean := createMockStoreAndSetup(t)
+	store, clean := testkit.CreateMockStore(t)
 	defer clean()
 	tk := testkit.NewTestKit(t, store)
 	tk.MustExec("set @@tidb_max_chunk_size=35")
@@ -191,7 +191,7 @@ func TestNonTransactionalDeleteShardOnGeneratedColumn(t *testing.T) {
 }
 
 func TestNonTransactionalDeleteAutoDetectShardColumn(t *testing.T) {
-	store, clean := createMockStoreAndSetup(t)
+	store, clean := testkit.CreateMockStore(t)
 	defer clean()
 	tk := testkit.NewTestKit(t, store)
 	tk.MustExec("set @@tidb_max_chunk_size=35")
@@ -229,7 +229,7 @@ func TestNonTransactionalDeleteAutoDetectShardColumn(t *testing.T) {
 }
 
 func TestNonTransactionalDeleteInvisibleIndex(t *testing.T) {
-	store, clean := createMockStoreAndSetup(t)
+	store, clean := testkit.CreateMockStore(t)
 	defer clean()
 	tk := testkit.NewTestKit(t, store)
 	tk.MustExec("set @@tidb_max_chunk_size=35")
@@ -249,7 +249,7 @@ func TestNonTransactionalDeleteInvisibleIndex(t *testing.T) {
 }
 
 func TestNonTransactionalDeleteIgnoreSelectLimit(t *testing.T) {
-	store, clean := createMockStoreAndSetup(t)
+	store, clean := testkit.CreateMockStore(t)
 	defer clean()
 	tk := testkit.NewTestKit(t, store)
 	tk.MustExec("set @@tidb_max_chunk_size=35")
@@ -264,7 +264,7 @@ func TestNonTransactionalDeleteIgnoreSelectLimit(t *testing.T) {
 }
 
 func TestNonTransactionalDeleteReadStaleness(t *testing.T) {
-	store, clean := createMockStoreAndSetup(t)
+	store, clean := testkit.CreateMockStore(t)
 	defer clean()
 	tk := testkit.NewTestKit(t, store)
 	tk.MustExec("set @@tidb_max_chunk_size=35")
@@ -280,7 +280,7 @@ func TestNonTransactionalDeleteReadStaleness(t *testing.T) {
 }
 
 func TestNonTransactionalDeleteCheckConstraint(t *testing.T) {
-	store, clean := createMockStoreAndSetup(t)
+	store, clean := testkit.CreateMockStore(t)
 	defer clean()
 	tk := testkit.NewTestKit(t, store)
 
@@ -348,7 +348,7 @@ func TestNonTransactionalDeleteCheckConstraint(t *testing.T) {
 }
 
 func TestNonTransactionalDeleteOptimizerHints(t *testing.T) {
-	store, clean := createMockStoreAndSetup(t)
+	store, clean := testkit.CreateMockStore(t)
 	defer clean()
 	tk := testkit.NewTestKit(t, store)
 	tk.MustExec("use test")
@@ -361,7 +361,7 @@ func TestNonTransactionalDeleteOptimizerHints(t *testing.T) {
 }
 
 func TestNonTransactionalDeleteMultiTables(t *testing.T) {
-	store, clean := createMockStoreAndSetup(t)
+	store, clean := testkit.CreateMockStore(t)
 	defer clean()
 	tk := testkit.NewTestKit(t, store)
 
@@ -380,7 +380,7 @@ func TestNonTransactionalDeleteMultiTables(t *testing.T) {
 }
 
 func TestNonTransactionalDeleteAlias(t *testing.T) {
-	store, clean := createMockStoreAndSetup(t)
+	store, clean := testkit.CreateMockStore(t)
 	defer clean()
 	tk := testkit.NewTestKit(t, store)
 
@@ -423,7 +423,7 @@ func TestNonTransactionalDeleteAlias(t *testing.T) {
 
 func TestNonTransactionalDeleteShardOnUnsupportedTypes(t *testing.T) {
 	// When some day the test fail because such types are supported, we can update related docs and consider remove the test.
-	store, clean := createMockStoreAndSetup(t)
+	store, clean := testkit.CreateMockStore(t)
 	defer clean()
 	tk := testkit.NewTestKit(t, store)
 	tk.MustExec("use test")

session/session_legacy_test.go

@@ -27,6 +27,7 @@ import (
 	"time"
 
 	. "github.com/pingcap/check"
+	"github.com/pingcap/errors"
 	"github.com/pingcap/failpoint"
 	"github.com/pingcap/tidb/config"
 	"github.com/pingcap/tidb/domain"
@@ -44,13 +45,11 @@ import (
 	"github.com/pingcap/tidb/privilege/privileges"
 	"github.com/pingcap/tidb/session"
 	"github.com/pingcap/tidb/sessionctx"
-	"github.com/pingcap/tidb/sessionctx/binloginfo"
 	"github.com/pingcap/tidb/sessionctx/variable"
 	"github.com/pingcap/tidb/sessiontxn"
 	"github.com/pingcap/tidb/store/copr"
 	"github.com/pingcap/tidb/store/driver"
 	"github.com/pingcap/tidb/store/mockstore"
-	"github.com/pingcap/tidb/store/mockstore/mockcopr"
 	"github.com/pingcap/tidb/table/tables"
 	"github.com/pingcap/tidb/tablecodec"
 	"github.com/pingcap/tidb/types"
@@ -59,10 +58,10 @@ import (
 	"github.com/pingcap/tidb/util/sqlexec"
 	"github.com/pingcap/tidb/util/testkit"
 	"github.com/pingcap/tidb/util/testleak"
-	"github.com/pingcap/tipb/go-binlog"
 	"github.com/tikv/client-go/v2/testutils"
 	"github.com/tikv/client-go/v2/tikv"
 	"github.com/tikv/client-go/v2/txnkv/transaction"
+	clientv3 "go.etcd.io/etcd/client/v3"
 	"google.golang.org/grpc"
 )
 
@@ -79,7 +78,6 @@ type testSessionSuiteBase struct {
 	cluster testutils.Cluster
 	store   kv.Storage
 	dom     *domain.Domain
-	pdAddr  string
 }
 
 type testSessionSuite struct {
@@ -98,16 +96,72 @@ type testSessionSerialSuite struct {
 	testSessionSuiteBase
 }
 
+func clearTiKVStorage(store kv.Storage) error {
+	txn, err := store.Begin()
+	if err != nil {
+		return errors.Trace(err)
+	}
+	iter, err := txn.Iter(nil, nil)
+	if err != nil {
+		return errors.Trace(err)
+	}
+	for iter.Valid() {
+		if err := txn.Delete(iter.Key()); err != nil {
+			return errors.Trace(err)
+		}
+		if err := iter.Next(); err != nil {
+			return errors.Trace(err)
+		}
+	}
+	return txn.Commit(context.Background())
+}
+
+func clearEtcdStorage(ebd kv.EtcdBackend) error {
+	endpoints, err := ebd.EtcdAddrs()
+	if err != nil {
+		return err
+	}
+	cli, err := clientv3.New(clientv3.Config{
+		Endpoints:        endpoints,
+		AutoSyncInterval: 30 * time.Second,
+		DialTimeout:      5 * time.Second,
+		DialOptions: []grpc.DialOption{
+			grpc.WithBackoffMaxDelay(time.Second * 3),
+		},
+		TLS: ebd.TLSConfig(),
+	})
+	if err != nil {
+		return errors.Trace(err)
+	}
+	defer cli.Close()
+
+	resp, err := cli.Get(context.Background(), "/tidb", clientv3.WithPrefix())
+	if err != nil {
+		return errors.Trace(err)
+	}
+	for _, kv := range resp.Kvs {
+		if kv.Lease != 0 {
+			if _, err := cli.Revoke(context.Background(), clientv3.LeaseID(kv.Lease)); err != nil {
+				return errors.Trace(err)
+			}
+		}
+	}
+	_, err = cli.Delete(context.Background(), "/tidb", clientv3.WithPrefix())
+	if err != nil {
+		return errors.Trace(err)
+	}
+	return nil
+}
+
 func (s *testSessionSuiteBase) SetUpSuite(c *C) {
 	testleak.BeforeTest()
 
 	if *withTiKV {
-		s.pdAddr = "127.0.0.1:2379"
 		var d driver.TiKVDriver
 		config.UpdateGlobal(func(conf *config.Config) {
 			conf.TxnLocalLatches.Enabled = false
 		})
-		store, err := d.Open(fmt.Sprintf("tikv://%s?disableGC=true", s.pdAddr))
+		store, err := d.Open("tikv://127.0.0.1:2379?disableGC=true")
 		c.Assert(err, IsNil)
 		err = clearTiKVStorage(store)
 		c.Assert(err, IsNil)
@@ -155,56 +209,6 @@ func (s *testSessionSuiteBase) TearDownTest(c *C) {
 	}
 }
 
-type mockBinlogPump struct {
-}
-
-var _ binlog.PumpClient = &mockBinlogPump{}
-
-func (p *mockBinlogPump) WriteBinlog(ctx context.Context, in *binlog.WriteBinlogReq, opts ...grpc.CallOption) (*binlog.WriteBinlogResp, error) {
-	return &binlog.WriteBinlogResp{}, nil
-}
-
-type mockPumpPullBinlogsClient struct {
-	grpc.ClientStream
-}
-
-func (m mockPumpPullBinlogsClient) Recv() (*binlog.PullBinlogResp, error) {
-	return nil, nil
-}
-
-func (p *mockBinlogPump) PullBinlogs(ctx context.Context, in *binlog.PullBinlogReq, opts ...grpc.CallOption) (binlog.Pump_PullBinlogsClient, error) {
-	return mockPumpPullBinlogsClient{mockcopr.MockGRPCClientStream()}, nil
-}
-
-func (s *testSessionSuite) TestForCoverage(c *C) {
-	// Just for test coverage.
-	tk := testkit.NewTestKitWithInit(c, s.store)
-	tk.MustExec("drop table if exists t")
-	tk.MustExec("create table t (id int auto_increment, v int, index (id))")
-	tk.MustExec("insert t values ()")
-	tk.MustExec("insert t values ()")
-	tk.MustExec("insert t values ()")
-
-	// Normal request will not cover txn.Seek.
-	tk.MustExec("admin check table t")
-
-	// Cover dirty table operations in StateTxn.
-	tk.Se.GetSessionVars().BinlogClient = binloginfo.MockPumpsClient(&mockBinlogPump{})
-	tk.MustExec("begin")
-	tk.MustExec("truncate table t")
-	tk.MustExec("insert t values ()")
-	tk.MustExec("delete from t where id = 2")
-	tk.MustExec("update t set v = 5 where id = 2")
-	tk.MustExec("insert t values ()")
-	tk.MustExec("rollback")
-
-	c.Check(tk.Se.SetCollation(mysql.DefaultCollationID), IsNil)
-
-	tk.MustExec("show processlist")
-	_, err := tk.Se.FieldList("t")
-	c.Check(err, IsNil)
-}
-
 func (s *testSessionSuite2) TestErrorRollback(c *C) {
 	tk := testkit.NewTestKitWithInit(c, s.store)
 	tk.MustExec("drop table if exists t_rollback")

session/backup_restore_test.go → tests/realtikvtest/backup_restore_test.go

@@ -15,7 +15,7 @@
 // This file contains tests about backup restore (br) which need running with real TiKV.
 // Only tests under /session will be run with real TiKV, so we put them here instead of /br.
 
-package session_test
+package realtikvtest
 
 import (
 	"os"
@@ -27,9 +27,8 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-// TODO move this test to BR integration tests.
 func TestBackupAndRestore(t *testing.T) {
-	if !*withTiKV {
+	if !*withRealTiKV {
 		t.Skip("only run BR SQL integration test with tikv store")
 	}