FLASH-275/276/277/278/281/282: Complete DDL implementation

CMakeLists.txt

@@ -265,7 +265,6 @@ include (cmake/find_capnp.cmake)
 include (cmake/find_llvm.cmake)
 include (cmake/find_grpc.cmake)
 include (cmake/find_kvproto.cmake)
-include (cmake/find_curl.cmake)
 
 
 include (cmake/find_contrib_lib.cmake)

contrib/client-c/include/common/CltException.h

@@ -1,30 +1,32 @@
 #pragma once
 
+#include <Poco/Exception.h>
 #include <exception>
 #include <string>
-#include <Poco/Exception.h>
 
-namespace pingcap {
+namespace pingcap
+{
 
 const int MismatchClusterIDCode = 1;
 const int GRPCErrorCode = 2;
 const int InitClusterIDFailed = 3;
 const int UpdatePDLeaderFailed = 4;
 const int TimeoutError = 5;
 const int RegionUnavailable = 6;
+const int LogicalError = 7;
+const int LockError = 8;
 
 class Exception : public Poco::Exception
 {
 public:
-    Exception() {}  /// For deferred initialization.
+    Exception() {} /// For deferred initialization.
     Exception(const std::string & msg, int code = 0) : Poco::Exception(msg, code) {}
     Exception(const std::string & msg, const std::string & arg, int code = 0) : Poco::Exception(msg, arg, code) {}
     Exception(const std::string & msg, const Exception & exc, int code = 0) : Poco::Exception(msg, exc, code) {}
     explicit Exception(const Poco::Exception & exc) : Poco::Exception(exc.displayText()) {}
 
     Exception * clone() const override { return new Exception(*this); }
     void rethrow() const override { throw *this; }
-
 };
 
-}
+} // namespace pingcap

contrib/client-c/include/common/Log.h

@@ -1,5 +1,6 @@
 #include <Poco/Logger.h>
 
-namespace pingcap {
+namespace pingcap
+{
 using Poco::Logger;
 }

contrib/client-c/include/pd/Client.h

@@ -1,24 +1,28 @@
 #pragma once
 
+#include <common/Log.h>
+#include <kvproto/pdpb.grpc.pb.h>
 #include <atomic>
-#include <thread>
+#include <condition_variable>
 #include <mutex>
 #include <shared_mutex>
-#include <condition_variable>
-#include <kvproto/pdpb.grpc.pb.h>
-#include <common/Log.h>
+#include <thread>
 #include "IClient.h"
 
-namespace pingcap{
-namespace pd {
+namespace pingcap
+{
+namespace pd
+{
 
-struct SecurityOption {
+struct SecurityOption
+{
     std::string CAPath;
     std::string CertPath;
     std::string KeyPath;
 };
 
-class Client : public IClient {
+class Client : public IClient
+{
     const int max_init_cluster_retries;
 
     const std::chrono::seconds pd_timeout;
@@ -28,17 +32,14 @@ class Client : public IClient {
     const std::chrono::seconds update_leader_interval;
 
 public:
-
     Client(const std::vector<std::string> & addrs);
 
     ~Client() override;
 
     //uint64_t getClusterID() override;
 
     // only implement a weak get ts.
-    uint64_t getTS() override {
-        throw "not implemented";
-    }
+    uint64_t getTS() override;
 
     std::tuple<metapb::Region, metapb::Peer, std::vector<metapb::Peer>> getRegion(std::string key) override;
 
@@ -59,17 +60,17 @@ class Client : public IClient {
 
     void updateLeader();
 
-    void updateURLs(const ::google::protobuf::RepeatedPtrField<::pdpb::Member>& members);
+    void updateURLs(const ::google::protobuf::RepeatedPtrField<::pdpb::Member> & members);
 
     void leaderLoop();
 
-    void switchLeader(const ::google::protobuf::RepeatedPtrField<std::string>&);
+    void switchLeader(const ::google::protobuf::RepeatedPtrField<std::string> &);
 
     std::unique_ptr<pdpb::PD::Stub> leaderStub();
 
     pdpb::GetMembersResponse getMembers(std::string);
 
-    pdpb::RequestHeader* requestHeader();
+    pdpb::RequestHeader * requestHeader();
 
     std::shared_ptr<grpc::Channel> getOrCreateGRPCConn(const std::string &);
 
@@ -83,7 +84,7 @@ class Client : public IClient {
 
     std::mutex gc_safepoint_mutex;
 
-    std::unordered_map<std::string, std::shared_ptr<grpc::Channel> > channel_map;
+    std::unordered_map<std::string, std::shared_ptr<grpc::Channel>> channel_map;
 
     std::vector<std::string> urls;
 
@@ -100,9 +101,8 @@ class Client : public IClient {
     std::atomic<bool> check_leader;
 
     Logger * log;
-
 };
 
 
-}
-}
+} // namespace pd
+} // namespace pingcap

contrib/client-c/include/pd/IClient.h

@@ -1,41 +1,43 @@
 #pragma once
 
-#include<string>
-#include<vector>
+#include <string>
+#include <vector>
 
 #pragma GCC diagnostic push
 #pragma GCC diagnostic ignored "-Wunused-parameter"
 #include <kvproto/enginepb.pb.h>
 #pragma GCC diagnostic pop
 
-namespace pingcap {
-namespace pd {
+namespace pingcap
+{
+namespace pd
+{
 
-class IClient {
+class IClient
+{
 public:
-//    virtual uint64_t getClusterID() = 0;
+    //    virtual uint64_t getClusterID() = 0;
 
     virtual ~IClient() {}
 
     virtual uint64_t getTS() = 0;
 
     virtual std::tuple<metapb::Region, metapb::Peer, std::vector<metapb::Peer>> getRegion(std::string key) = 0;
 
-//    virtual std::pair<metapb::Region, metapb::Peer> getPrevRegion(std::string key) = 0;
+    //    virtual std::pair<metapb::Region, metapb::Peer> getPrevRegion(std::string key) = 0;
 
     virtual std::tuple<metapb::Region, metapb::Peer, std::vector<metapb::Peer>> getRegionByID(uint64_t region_id) = 0;
 
     virtual metapb::Store getStore(uint64_t store_id) = 0;
 
-//    virtual std::vector<metapb::Store> getAllStores() = 0;
+    //    virtual std::vector<metapb::Store> getAllStores() = 0;
 
     virtual uint64_t getGCSafePoint() = 0;
 
     virtual bool isMock() = 0;
-
 };
 
 using ClientPtr = std::shared_ptr<IClient>;
 
-}
-}
+} // namespace pd
+} // namespace pingcap

contrib/client-c/include/pd/MockPDClient.h

@@ -1,46 +1,34 @@
 #pragma once
 
-#include <limits>
 #include <pd/IClient.h>
+#include <limits>
 
-namespace pingcap {
-namespace pd {
+namespace pingcap
+{
+namespace pd
+{
 
 using Clock = std::chrono::system_clock;
-using Seconds = std::chrono::seconds;
 
-class MockPDClient : public IClient {
+class MockPDClient : public IClient
+{
 public:
     MockPDClient() = default;
 
     ~MockPDClient() override {}
 
-    uint64_t getGCSafePoint() override
-    {
-        return (Clock::now() - Seconds(2)).time_since_epoch().count();
-    }
+    uint64_t getGCSafePoint() override { return 1000000000000000; }
 
-    uint64_t getTS() override
-    {
-        return Clock::now().time_since_epoch().count();
-    }
+    uint64_t getTS() override { return Clock::now().time_since_epoch().count(); }
 
-    std::tuple<metapb::Region, metapb::Peer, std::vector<metapb::Peer>> getRegion(std::string) override {
-        throw "not implemented";
-    }
+    std::tuple<metapb::Region, metapb::Peer, std::vector<metapb::Peer>> getRegion(std::string) override { throw "not implemented"; }
 
-    std::tuple<metapb::Region, metapb::Peer, std::vector<metapb::Peer>> getRegionByID(uint64_t) override {
-        throw "not implemented";
-    }
+    std::tuple<metapb::Region, metapb::Peer, std::vector<metapb::Peer>> getRegionByID(uint64_t) override { throw "not implemented"; }
 
-    metapb::Store getStore(uint64_t) override {
-        throw "not implemented";
-    }
+    metapb::Store getStore(uint64_t) override { throw "not implemented"; }
 
-    bool isMock() override {
-        return true;
-    }
+    bool isMock() override { return true; }
 };
 
-}
-}
+} // namespace pd
+} // namespace pingcap

contrib/client-c/include/tikv/Backoff.h

@@ -1,24 +1,28 @@
 #pragma once
 
-#include <iostream>
-#include <cmath>
 #include <unistd.h>
+#include <cmath>
+#include <iostream>
 #include <map>
 #include <memory>
 
 #include <common/CltException.h>
 
-namespace pingcap {
-namespace kv {
+namespace pingcap
+{
+namespace kv
+{
 
-enum Jitter {
+enum Jitter
+{
     NoJitter = 1,
     FullJitter,
     EqualJitter,
     DecorrJitter
 };
 
-enum BackoffType {
+enum BackoffType
+{
     boTiKVRPC = 0,
     boTxnLock,
     boTxnLockFast,
@@ -28,28 +32,31 @@ enum BackoffType {
     boServerBusy
 };
 
-inline int expo(int base, int cap, int n) {
-    return std::min(double(cap), double(base) * std::pow(2.0, double(n)));
-}
+inline int expo(int base, int cap, int n) { return std::min(double(cap), double(base) * std::pow(2.0, double(n))); }
 
-struct Backoff {
+struct Backoff
+{
     int base;
     int cap;
     int jitter;
     int last_sleep;
     int attempts;
 
-    Backoff(int base_, int cap_, Jitter jitter_) : base(base_), cap(cap_), jitter(jitter_), attempts(0) {
-        if (base < 2) {
+    Backoff(int base_, int cap_, Jitter jitter_) : base(base_), cap(cap_), jitter(jitter_), attempts(0)
+    {
+        if (base < 2)
+        {
             base = 2;
         }
         last_sleep = base;
     }
 
-    int sleep () {
+    int sleep()
+    {
         int sleep_time = 0;
         int v = 0;
-        switch(jitter) {
+        switch (jitter)
+        {
             case NoJitter:
                 sleep_time = expo(base, cap, attempts);
                 break;
@@ -59,31 +66,34 @@ struct Backoff {
                 break;
             case EqualJitter:
                 v = expo(base, cap, attempts);
-                sleep_time = v/2 + rand() % (v/2);
+                sleep_time = v / 2 + rand() % (v / 2);
                 break;
             case DecorrJitter:
-                sleep_time = int(std::min(double(cap), double(base + rand() % (last_sleep*3-base))));
+                sleep_time = int(std::min(double(cap), double(base + rand() % (last_sleep * 3 - base))));
         }
         ::usleep(sleep_time);
-        attempts ++;
+        attempts++;
         last_sleep = sleep_time;
         return last_sleep;
     }
 };
 
 constexpr int readIndexMaxBackoff = 20000;
+constexpr int GetMaxBackoff = 20000;
+constexpr int scanMaxBackoff = 20000;
 
 using BackoffPtr = std::shared_ptr<Backoff>;
 
-struct Backoffer {
+struct Backoffer
+{
     std::map<BackoffType, BackoffPtr> backoff_map;
     size_t total_sleep; // ms
-    size_t max_sleep; // ms
+    size_t max_sleep;   // ms
 
     Backoffer(size_t max_sleep_) : total_sleep(0), max_sleep(max_sleep_) {}
 
     void backoff(BackoffType tp, const Exception & exc);
 };
 
-}
-}
+} // namespace kv
+} // namespace pingcap