From 0eee7c85224cbd7ea7b6887e5b07990f2e82bbb4 Mon Sep 17 00:00:00 2001
From: disksing <i@disksing.com>
Date: Tue, 25 Oct 2016 20:57:25 +0800
Subject: [PATCH] store/tikv: add disableGC flag. (#1872)

* store/tikv: add disableGC flag.
---
 store/tikv/kv.go         | 20 ++++++++++++++------
 store/tikv/store_test.go | 16 ++++++++++++++++
 2 files changed, 30 insertions(+), 6 deletions(-)

diff --git a/store/tikv/kv.go b/store/tikv/kv.go
index 243830ea24..ac6c6d625f 100644
--- a/store/tikv/kv.go
+++ b/store/tikv/kv.go
@@ -45,12 +45,12 @@ type Driver struct {
 }
 
 // Open opens or creates an TiKV storage with given path.
-// Path example: tikv://etcd-node1:port,etcd-node2:port?cluster=1
+// Path example: tikv://etcd-node1:port,etcd-node2:port?cluster=1&disableGC=false
 func (d Driver) Open(path string) (kv.Storage, error) {
 	mc.Lock()
 	defer mc.Unlock()
 
-	etcdAddrs, clusterID, err := parsePath(path)
+	etcdAddrs, clusterID, disableGC, err := parsePath(path)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
@@ -64,7 +64,7 @@ func (d Driver) Open(path string) (kv.Storage, error) {
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
-	s, err := newTikvStore(uuid, &codecPDClient{pdCli}, newRPCClient(), true)
+	s, err := newTikvStore(uuid, &codecPDClient{pdCli}, newRPCClient(), !disableGC)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
@@ -248,7 +248,7 @@ func (s *tikvStore) SendKVReq(bo *Backoffer, req *pb.Request, regionID RegionVer
 	}
 }
 
-func parsePath(path string) (etcdAddrs []string, clusterID uint64, err error) {
+func parsePath(path string) (etcdAddrs []string, clusterID uint64, disableGC bool, err error) {
 	var u *url.URL
 	u, err = url.Parse(path)
 	if err != nil {
@@ -256,8 +256,8 @@ func parsePath(path string) (etcdAddrs []string, clusterID uint64, err error) {
 		return
 	}
 	if strings.ToLower(u.Scheme) != "tikv" {
-		log.Errorf("Uri scheme expected[tikv] but found [%s]", u.Scheme)
-		err = errors.Trace(err)
+		err = errors.Errorf("Uri scheme expected[tikv] but found [%s]", u.Scheme)
+		log.Error(err)
 		return
 	}
 	clusterID, err = strconv.ParseUint(u.Query().Get("cluster"), 10, 64)
@@ -266,6 +266,14 @@ func parsePath(path string) (etcdAddrs []string, clusterID uint64, err error) {
 		err = errors.Trace(err)
 		return
 	}
+	switch strings.ToLower(u.Query().Get("disableGC")) {
+	case "true":
+		disableGC = true
+	case "false", "":
+	default:
+		err = errors.New("disableGC flag should be true/false")
+		return
+	}
 	etcdAddrs = strings.Split(u.Host, ",")
 	return
 }
diff --git a/store/tikv/store_test.go b/store/tikv/store_test.go
index a489155bac..f5cc346c82 100644
--- a/store/tikv/store_test.go
+++ b/store/tikv/store_test.go
@@ -46,6 +46,22 @@ func (s *testStoreSuite) SetUpTest(c *C) {
 	s.store = store
 }
 
+func (s *testStoreSuite) TestParsePath(c *C) {
+	etcdAddrs, clusterID, disableGC, err := parsePath("tikv://node1:2379,node2:2379?cluster=1")
+	c.Assert(err, IsNil)
+	c.Assert(etcdAddrs, DeepEquals, []string{"node1:2379", "node2:2379"})
+	c.Assert(clusterID, Equals, uint64(1))
+	c.Assert(disableGC, IsFalse)
+
+	_, _, _, err = parsePath("tikv://node1:2379")
+	c.Assert(err, NotNil)
+	_, _, _, err = parsePath("tidb://node1:2379?cluster=1")
+	c.Assert(err, NotNil)
+	_, _, disableGC, err = parsePath("tikv://node1:2379?cluster=1&disableGC=true")
+	c.Assert(err, IsNil)
+	c.Assert(disableGC, IsTrue)
+}
+
 func (s *testStoreSuite) TestOracle(c *C) {
 	o := &mockOracle{}
 	s.store.oracle = o