update storage client for plato (#68)

Add meta/storage client.

Co-authored-by: Shylock Hg <33566796+Shylock-Hg@users.noreply.github.com>

.github/workflows/pull_request.yml

@@ -101,7 +101,7 @@ jobs:
           centos7)
             set +e
             for i in {0..10}; do
-              wget https://oss-cdn.nebula-graph.com.cn/package/v2-nightly/$(./date.py --day_diff=$i)/nebula-graph-$(./date.py --day_diff=$i)-nightly.el7.x86_64.rpm
+              wget https://oss-cdn.nebula-graph.com.cn/package/nightly/$(./date.py --day_diff=$i)/nebula-graph-$(./date.py --day_diff=$i)-nightly.el7.x86_64.rpm
               rpm -ivh nebula-graph-$(./date.py --day_diff=$i)-nightly.el7.x86_64.rpm
               if [ $? -eq 0 ]; then
                 break;
@@ -112,7 +112,7 @@ jobs:
           ubuntu1804)
             set +e
             for i in {0..10}; do
-              wget https://oss-cdn.nebula-graph.com.cn/package/v2-nightly/$(./date.py --day_diff=$i)/nebula-graph-$(./date.py --day_diff=$i)-nightly.ubuntu1804.amd64.deb
+              wget https://oss-cdn.nebula-graph.com.cn/package/nightly/$(./date.py --day_diff=$i)/nebula-graph-$(./date.py --day_diff=$i)-nightly.ubuntu1804.amd64.deb
               dpkg -i nebula-graph-$(./date.py --day_diff=$i)-nightly.ubuntu1804.amd64.deb
               if [ $? -eq 0 ]; then
                 break;

.gitignore

@@ -37,3 +37,6 @@ modules
 
 # IDE
 .vscode/
+
+# Coredump
+core.*

CMakeLists.txt

@@ -89,7 +89,7 @@ install(
 
 add_custom_target(
   format
-  COMMAND "find" "include" "src/client" "src/datatypes" "src/time" "src/graph" "-type" "f" "\\(" "-iname" "\\*.h" "-o" "-iname" "\\*.cpp" "\\)" "|" "xargs" "clang-format" "-i"
+  COMMAND "find" "include" "src/client" "src/mclient" "src/sclient" "src/thrift" "src/datatypes" "src/time" "src/geo" "src/graph" "-type" "f" "\\(" "-iname" "\\*.h" "-o" "-iname" "\\*.cpp" "\\)" "|" "xargs" "clang-format" "-i"
   WORKING_DIRECTORY ${CMAKE_SOURCE_DIR}
 )
 

examples/CMakeLists.txt

@@ -34,7 +34,16 @@ add_executable(session_example
     SessionExample.cpp
 )
 
+add_executable(storage_client_example
+    StorageClientExample.cpp
+)
+
 target_link_libraries(session_example
     PRIVATE
         nebula_graph_client
 )
+
+target_link_libraries(storage_client_example
+    PRIVATE
+        nebula_storage_client
+)

examples/SessionExample.cpp

@@ -9,7 +9,7 @@
 
 #include <nebula/client/Config.h>
 #include <nebula/client/ConnectionPool.h>
-#include <nebula/client/Init.h>
+#include <common/Init.h>
 
 int main(int argc, char* argv[]) {
     nebula::init(&argc, &argv);

examples/StorageClientExample.cpp

@@ -0,0 +1,39 @@
+/* Copyright (c) 2020 vesoft inc. All rights reserved.
+ *
+ * This source code is licensed under Apache 2.0 License.
+ */
+
+#include <atomic>
+#include <chrono>
+#include <limits>
+#include <thread>
+
+#include <nebula/sclient/ScanEdgeIter.h>
+#include <common/Init.h>
+#include <nebula/sclient/StorageClient.h>
+
+int main(int argc, char* argv[]) {
+    nebula::init(&argc, &argv);
+
+    nebula::StorageClient c({"127.0.0.1:9559"});
+
+    nebula::ScanEdgeIter scanEdgeIter = c.scanEdgeWithPart("nba",
+                                           1,
+                                           "like",
+                                           std::vector<std::string>{"likeness"},
+                                           10,
+                                           0,
+                                           std::numeric_limits<int64_t>::max(),
+                                           "",
+                                           true,
+                                           true);
+    std::cout << "scan edge..." << std::endl;
+    while (scanEdgeIter.hasNext()) {
+        std::cout << "-------------------------" << std::endl;
+        nebula::DataSet ds = scanEdgeIter.next();
+        std::cout << ds << std::endl;
+        std::cout << "+++++++++++++++++++++++++" << std::endl;
+    }
+
+    return 0;
+}

exported-symbols.map

@@ -31,6 +31,8 @@ global:
         nebula::PlanNodeDescription::*;
         nebula::PlanDescription::*;
         nebula::ExecutionResponse::*;
+        nebula::StorageClient::*;
+        nebula::ScanEdgeIter::*;
         *::_S_empty_rep_storage;
     };
 local:

include/common/datatypes/Geography.h

@@ -5,10 +5,11 @@
 
 #pragma once
 
-#include "common/datatypes/Value.h"
-#include <vector>
-#include <variant>
 #include <string>
+#include <variant>
+#include <vector>
+
+#include "common/datatypes/Value.h"
 
 // Do not include <s2/s2polygon.h> here, it will indirectly includes a header file which defines a
 // enum `BEGIN`(not enum class). While Geography.h is indirectly included by parser.yy, which has a

include/common/geo/io/wkb/ByteOrder.h

@@ -15,4 +15,3 @@ enum class ByteOrder : uint8_t {
 
 }  // namespace geo
 }  // namespace nebula
-

include/common/geo/io/wkt/WKTWriter.h

@@ -5,8 +5,8 @@
 
 #pragma once
 
-#include <vector>
 #include <string>
+#include <vector>
 
 #include "common/datatypes/Geography.h"
 

include/nebula/mclient/MConfig.h

@@ -0,0 +1,22 @@
+/* Copyright (c) 2020 vesoft inc. All rights reserved.
+ *
+ * This source code is licensed under Apache 2.0 License.
+ */
+
+#pragma once
+
+#include <cstdint>
+#include <string>
+
+namespace nebula {
+
+struct MConfig {
+  // It's as same as FLAGS_conn_timeout_ms in nebula
+  int32_t connTimeoutInMs_{1000};
+  // It's as same as FLAG_meta_client_timeout_ms in nebula
+  int32_t clientTimeoutInMs_{60 * 1000};
+  bool enableSSL_{false};
+  std::string CAPath_;
+};
+
+}  // namespace nebula

include/nebula/mclient/MetaClient.h

@@ -0,0 +1,116 @@
+/* Copyright (c) 2020 vesoft inc. All rights reserved.
+ *
+ * This source code is licensed under Apache 2.0 License.
+ */
+
+#pragma once
+
+#include <functional>
+#include <memory>
+#include <string>
+#include <thread>
+#include <unordered_map>
+#include <vector>
+
+#include "common/datatypes/HostAddr.h"
+#include "common/thrift/ThriftTypes.h"
+#include "nebula/mclient/MConfig.h"
+
+struct pair_hash {
+  template <class T1, class T2>
+  std::size_t operator()(const std::pair<T1, T2> &pair) const {
+    return std::hash<T1>()(pair.first) ^ std::hash<T2>()(pair.second);
+  }
+};
+
+namespace folly {
+
+class IOThreadPoolExecutor;
+template <class T>
+class Promise;
+
+}  // namespace folly
+
+namespace nebula {
+
+namespace thrift {
+
+template <class ClientType>
+class ThriftClientManager;
+
+}  // namespace thrift
+
+namespace meta {
+namespace cpp2 {
+
+enum class ListHostType;
+class HostItem;
+class MetaServiceAsyncClient;
+class ListSpacesReq;
+class ListSpacesResp;
+class IdName;
+class EdgeItem;
+class ListEdgesReq;
+class ListEdgesResp;
+
+}  // namespace cpp2
+}  // namespace meta
+
+using SpaceIdName = std::pair<GraphSpaceID, std::string>;
+using SpaceNameIdMap = std::unordered_map<std::string, GraphSpaceID>;
+using SpaceEdgeNameTypeMap =
+    std::unordered_map<std::pair<GraphSpaceID, std::string>, EdgeType, pair_hash>;
+
+class MetaClient {
+ public:
+  explicit MetaClient(const std::vector<std::string> &metaAddrs,
+                      const MConfig &mConfig = MConfig{});
+
+  ~MetaClient();
+
+  std::pair<bool, GraphSpaceID> getSpaceIdByNameFromCache(const std::string &name);
+
+  std::pair<bool, EdgeType> getEdgeTypeByNameFromCache(GraphSpaceID spaceId,
+                                                       const std::string &name);
+
+  std::pair<bool, std::vector<PartitionID>> getPartsFromCache(GraphSpaceID spaceId);
+
+  std::pair<bool, HostAddr> getPartLeaderFromCache(GraphSpaceID spaceId, PartitionID partId);
+
+ private:
+  bool loadData();
+
+  std::pair<bool, std::vector<SpaceIdName>> listSpaces();
+
+  std::pair<bool, std::vector<meta::cpp2::HostItem>> listHosts(meta::cpp2::ListHostType tp);
+
+  std::pair<bool, std::vector<meta::cpp2::EdgeItem>> listEdgeSchemas(GraphSpaceID spaceId);
+
+  void loadLeader(const std::vector<nebula::meta::cpp2::HostItem> &hostItems,
+                  const SpaceNameIdMap &spaceIndexByName);
+
+  std::vector<SpaceIdName> toSpaceIdName(const std::vector<meta::cpp2::IdName> &tIdNames);
+
+  template <class Request,
+            class RemoteFunc,
+            class RespGenerator,
+            class RpcResponse = typename std::result_of<RemoteFunc(
+                std::shared_ptr<meta::cpp2::MetaServiceAsyncClient>, Request)>::type::value_type,
+            class Response = typename std::result_of<RespGenerator(RpcResponse)>::type>
+  void getResponse(Request req,
+                   RemoteFunc remoteFunc,
+                   RespGenerator respGen,
+                   folly::Promise<std::pair<bool, Response>> pro);
+
+ private:
+  std::vector<HostAddr> metaAddrs_;
+  MConfig mConfig_;
+  SpaceNameIdMap spaceIndexByName_;
+  SpaceEdgeNameTypeMap spaceEdgeIndexByName_;
+  std::unordered_map<std::pair<GraphSpaceID, PartitionID>, HostAddr, pair_hash> spacePartLeaderMap_;
+  std::unordered_map<GraphSpaceID, std::vector<PartitionID>> spacePartsMap_;
+  std::shared_ptr<folly::IOThreadPoolExecutor> ioExecutor_;
+  std::shared_ptr<thrift::ThriftClientManager<meta::cpp2::MetaServiceAsyncClient>> clientsMan_;
+};
+
+}  // namespace nebula

include/nebula/sclient/SConfig.h

@@ -0,0 +1,22 @@
+/* Copyright (c) 2020 vesoft inc. All rights reserved.
+ *
+ * This source code is licensed under Apache 2.0 License.
+ */
+
+#pragma once
+
+#include <cstdint>
+#include <string>
+
+namespace nebula {
+
+struct SConfig {
+  // It's as same as FLAGS_conn_timeout_ms in nebula
+  int32_t connTimeoutInMs_{1000};
+  // It's as same as FLAG_meta_client_timeout_ms in nebula
+  int32_t clientTimeoutInMs_{60 * 1000};
+  bool enableSSL_{false};
+  std::string CAPath_;
+};
+
+}  // namespace nebula

include/nebula/sclient/ScanEdgeIter.h

@@ -0,0 +1,37 @@
+/* Copyright (c) 2020 vesoft inc. All rights reserved.
+ *
+ * This source code is licensed under Apache 2.0 License.
+ */
+
+#pragma once
+
+#include <string>
+#include <vector>
+
+#include "common/datatypes/DataSet.h"
+
+namespace nebula {
+class StorageClient;
+
+namespace storage {
+namespace cpp2 {
+class ScanEdgeRequest;
+}  // namespace cpp2
+}  // namespace storage
+
+struct ScanEdgeIter {
+  ScanEdgeIter(StorageClient* client, storage::cpp2::ScanEdgeRequest* req, bool hasNext = true);
+
+  ~ScanEdgeIter();
+
+  bool hasNext();
+
+  DataSet next();
+
+  StorageClient* client_;
+  storage::cpp2::ScanEdgeRequest* req_;
+  bool hasNext_;
+  std::string nextCursor_;
+};
+
+}  // namespace nebula