Support general storage in StorageService

src/daemons/CMakeLists.txt

@@ -121,7 +121,6 @@ nebula_add_executable(
  wangle
 )
 
-
 install(
  TARGETS
  nebula-graphd

src/interface/common.thrift

@@ -84,4 +84,9 @@ struct HostAddr {
  2: Port port,
 }
 
+struct Pair {
+ 1: string key,
+ 2: string value,
+}
+
 const ValueType kInvalidValueType = {"type" : UNKNOWN}

src/interface/meta.thrift

@@ -78,11 +78,6 @@ struct IdName {
  2: string name,
 }
 
-struct Pair {
- 1: string key,
- 2: string value,
-}
-
 struct SpaceProperties {
  1: string space_name,
  2: i32 partition_num,
@@ -302,7 +297,7 @@ struct MultiPutReq {
  // segment is used to avoid conflict with system data.
  // it should be comprised of numbers and letters.
  1: string segment,
- 2: list<Pair> pairs,
+ 2: list<common.Pair> pairs,
 }
 
 struct GetReq {

src/interface/storage.thrift

@@ -23,7 +23,8 @@ enum ErrorCode {
  E_KEY_HAS_EXISTS = -12,
  E_SPACE_NOT_FOUND = -13,
  E_PART_NOT_FOUND = -14,
- E_CONSENSUS_ERROR = -15,
+ E_KEY_NOT_FOUND = -15,
+ E_CONSENSUS_ERROR = -16,
 
  // meta failures
  E_EDGE_PROP_NOT_FOUND = -21,
@@ -292,6 +293,41 @@ struct UpdateEdgeRequest {
  7: bool insertable,
 }
 
+struct PutRequest {
+ 1: common.GraphSpaceID space_id,
+ 2: map<common.PartitionID, list<common.Pair>>(cpp.template = "std::unordered_map") parts,
+}
+
+struct RemoveRequest {
+ 1: common.GraphSpaceID space_id,
+ 2: map<common.PartitionID, list<string>>(cpp.template = "std::unordered_map") parts,
+}
+
+struct RemoveRangeRequest {
+ 1: common.GraphSpaceID space_id,
+ 2: map<common.PartitionID, list<common.Pair>>(cpp.template = "std::unordered_map") parts,
+}
+
+struct GetRequest {
+ 1: common.GraphSpaceID space_id,
+ 2: map<common.PartitionID, list<string>>(cpp.template = "std::unordered_map") parts,
+}
+
+struct PrefixRequest {
+ 1: common.GraphSpaceID space_id,
+ 2: map<common.PartitionID, string>(cpp.template = "std::unordered_map") parts,
+}
+
+struct ScanRequest {
+ 1: common.GraphSpaceID space_id,
+ 2: map<common.PartitionID, common.Pair>(cpp.template = "std::unordered_map") parts,
+}
+
+struct GeneralResponse {
+ 1: required ResponseCommon result,
+ 2: map<string, string>(cpp.template = "std::unordered_map") values,
+}
+
 struct GetUUIDReq {
  1: common.GraphSpaceID space_id,
  2: common.PartitionID part_id,
@@ -331,5 +367,11 @@ service StorageService {
  AdminExecResp memberChange(1: MemberChangeReq req);
  GetLeaderResp getLeaderPart(1: GetLeaderReq req);
 
+ // Interfaces for key-value storage
+ ExecResponse put(1: PutRequest req);
+ GeneralResponse get(1: GetRequest req);
+ ExecResponse remove(1: RemoveRequest req);
+ ExecResponse removeRange(1: RemoveRangeRequest req);
+
  GetUUIDResp getUUID(1: GetUUIDReq req);
 }

src/meta/client/MetaClient.cpp

@@ -702,7 +702,7 @@ MetaClient::multiPut(std::string segment,
  return Status::Error("arguments invalid!");
  }
  cpp2::MultiPutReq req;
- std::vector<cpp2::Pair> data;
+ std::vector<nebula::cpp2::Pair> data;
  for (auto& element : pairs) {
  data.emplace_back(apache::thrift::FragileConstructor::FRAGILE,
  std::move(element.first), std::move(element.second));

src/meta/test/ProcessorTest.cpp

@@ -517,7 +517,7 @@ TEST(ProcessorTest, KVOperationTest) {
  }
  {
  // Multi Put Test
- std::vector<cpp2::Pair> pairs;
+ std::vector<nebula::cpp2::Pair> pairs;
  for (auto i = 0; i < 10; i++) {
  pairs.emplace_back(apache::thrift::FragileConstructor::FRAGILE,
  folly::stringPrintf("key_%d", i),

src/storage/BaseProcessor.h

@@ -24,6 +24,8 @@
 namespace nebula {
 namespace storage {
 
+using PartCode = std::pair<PartitionID, kvstore::ResultCode>;
+
 template<typename RESP>
 class BaseProcessor {
 public:

src/storage/CMakeLists.txt

@@ -14,6 +14,8 @@ nebula_add_library(
  QueryEdgeKeysProcessor.cpp
  UpdateVertexProcessor.cpp
  UpdateEdgeProcessor.cpp
+ PutProcessor.cpp
+ GetProcessor.cpp
 )
 
 nebula_add_library(

src/storage/GetProcessor.cpp

@@ -0,0 +1,55 @@
+/* Copyright (c) 2019 vesoft inc. All rights reserved.
+ *
+ * This source code is licensed under Apache 2.0 License,
+ * attached with Common Clause Condition 1.0, found in the LICENSES directory.
+ */
+
+#include "storage/GetProcessor.h"
+#include "base/NebulaKeyUtils.h"
+
+namespace nebula {
+namespace storage {
+
+void GetProcessor::process(const cpp2::GetRequest& req) {
+ space_ = req.get_space_id();
+ std::vector<folly::Future<std::pair<PartitionID, kvstore::ResultCode>>> results;
+ for (auto& part : req.get_parts()) {
+ results.emplace_back(asyncProcess(part.first, part.second));
+ }
+
+ folly::collectAll(results).via(executor_)
+ .then([&] (const std::vector<folly::Try<PartCode>>& tries) mutable {
+ for (const auto& t : tries) {
+ auto ret = t.value();
+ auto part = std::get<0>(ret);
+ auto resultCode = std::get<1>(ret);
+ this->pushResultCode(this->to(resultCode), part);
+ }
+
+ resp_.set_values(std::move(pairs_));
+ this->onFinished();
+ });
+}
+
+folly::Future<std::pair<PartitionID, kvstore::ResultCode>>
+GetProcessor::asyncProcess(PartitionID part,
+ const std::vector<std::string>& keys) {
+ folly::Promise<std::pair<PartitionID, kvstore::ResultCode>> promise;
+ auto future = promise.getFuture();
+
+ executor_->add([this, p = std::move(promise), part, keys] () mutable {
+ std::vector<std::string> values;
+ auto ret = this->kvstore_->multiGet(space_, part, keys, &values);
+ if (ret == kvstore::ResultCode::SUCCEEDED) {
+ std::lock_guard<std::mutex> lg(this->lock_);
+ for (int32_t i = 0; i < static_cast<int32_t>(keys.size()); i++) {
+ pairs_.emplace(keys[i], values[i]);
+ }
+ }
+ p.setValue(std::make_pair(part, ret));
+ });
+ return future;
+}
+
+} // namespace storage
+} // namespace nebula

src/storage/GetProcessor.h

@@ -0,0 +1,43 @@
+/* Copyright (c) 2019 vesoft inc. All rights reserved.
+ *
+ * This source code is licensed under Apache 2.0 License,
+ * attached with Common Clause Condition 1.0, found in the LICENSES directory.
+ */
+
+#ifndef STORAGE_GETPROCESSOR_H_
+#define STORAGE_GETPROCESSOR_H_
+
+#include "base/Base.h"
+#include "storage/BaseProcessor.h"
+
+namespace nebula {
+namespace storage {
+
+class GetProcessor : public BaseProcessor<cpp2::GeneralResponse> {
+public:
+ static GetProcessor* instance(kvstore::KVStore* kvstore,
+ meta::SchemaManager* schemaMan,
+ folly::Executor* executor) {
+ return new GetProcessor(kvstore, schemaMan, executor);
+ }
+
+ void process(const cpp2::GetRequest& req);
+
+protected:
+ explicit GetProcessor(kvstore::KVStore* kvstore, meta::SchemaManager* schemaMan,
+ folly::Executor* executor = nullptr)
+ : BaseProcessor<cpp2::GeneralResponse>(kvstore, schemaMan), executor_(executor) {}
+
+private:
+ folly::Future<std::pair<PartitionID, kvstore::ResultCode>>
+ asyncProcess(PartitionID part, const std::vector<std::string>& keys);
+
+ folly::Executor *executor_ = nullptr;
+ std::unordered_map<std::string, std::string> pairs_;
+ GraphSpaceID space_;
+};
+
+} // namespace storage
+} // namespace nebula
+
+#endif // STORAGE_GETPROCESSOR_H_

src/storage/PutProcessor.cpp

@@ -0,0 +1,29 @@
+/* Copyright (c) 2019 vesoft inc. All rights reserved.
+ *
+ * This source code is licensed under Apache 2.0 License,
+ * attached with Common Clause Condition 1.0, found in the LICENSES directory.
+ */
+
+#include "storage/PutProcessor.h"
+
+namespace nebula {
+namespace storage {
+
+void PutProcessor::process(const cpp2::PutRequest& req) {
+ const auto& pairs = req.get_parts();
+ auto space = req.get_space_id();
+ callingNum_ = pairs.size();
+ CHECK_NOTNULL(kvstore_);
+
+ std::for_each(pairs.begin(), pairs.end(), [&](auto& value) {
+ auto part = value.first;
+ std::vector<kvstore::KV> data;
+ for (auto& pair : value.second) {
+ data.emplace_back(pair.key, pair.value);
+ }
+ doPut(space, part, std::move(data));
+ });
+}
+
+} // namespace storage
+} // namespace nebula

src/storage/PutProcessor.h

@@ -0,0 +1,33 @@
+/* Copyright (c) 2019 vesoft inc. All rights reserved.
+ *
+ * This source code is licensed under Apache 2.0 License,
+ * attached with Common Clause Condition 1.0, found in the LICENSES directory.
+ */
+
+#ifndef STORAGE_PUTPROCESSOR_H_
+#define STORAGE_PUTPROCESSOR_H_
+
+#include "base/Base.h"
+#include "storage/BaseProcessor.h"
+
+namespace nebula {
+namespace storage {
+
+class PutProcessor : public BaseProcessor<cpp2::ExecResponse> {
+public:
+ static PutProcessor* instance(kvstore::KVStore* kvstore,
+ meta::SchemaManager* schemaMan) {
+ return new PutProcessor(kvstore, schemaMan);
+ }
+
+ void process(const cpp2::PutRequest& req);
+
+private:
+ explicit PutProcessor(kvstore::KVStore* kvstore, meta::SchemaManager* schemaMan)
+ : BaseProcessor<cpp2::ExecResponse>(kvstore, schemaMan) {}
+};
+
+} // namespace storage
+} // namespace nebula
+
+#endif // STORAGE_PUTPROCESSOR_H_

src/storage/StorageServiceHandler.cpp

@@ -18,6 +18,8 @@
 #include "storage/QueryEdgeKeysProcessor.h"
 #include "storage/UpdateVertexProcessor.h"
 #include "storage/UpdateEdgeProcessor.h"
+#include "storage/PutProcessor.h"
+#include "storage/GetProcessor.h"
 #include "storage/GetUUIDProcessor.h"
 
 #define RETURN_FUTURE(processor) \
@@ -138,6 +140,18 @@ StorageServiceHandler::future_getLeaderPart(const cpp2::GetLeaderReq& req) {
  RETURN_FUTURE(processor);
 }
 
+folly::Future<cpp2::ExecResponse>
+StorageServiceHandler::future_put(const cpp2::PutRequest& req) {
+ auto* processor = PutProcessor::instance(kvstore_, schemaMan_);
+ RETURN_FUTURE(processor);
+}
+
+folly::Future<cpp2::GeneralResponse>
+StorageServiceHandler::future_get(const cpp2::GetRequest& req) {
+ auto* processor = GetProcessor::instance(kvstore_, schemaMan_, getThreadManager());
+ RETURN_FUTURE(processor);
+}
+
 folly::Future<cpp2::GetUUIDResp>
 StorageServiceHandler::future_getUUID(const cpp2::GetUUIDReq& req) {
  auto* processor = GetUUIDProcessor::instance(kvstore_);

src/storage/StorageServiceHandler.h

@@ -82,6 +82,12 @@ class StorageServiceHandler final : public cpp2::StorageServiceSvIf {
  folly::Future<cpp2::GetLeaderResp>
  future_getLeaderPart(const cpp2::GetLeaderReq& req) override;
 
+ folly::Future<cpp2::ExecResponse>
+ future_put(const cpp2::PutRequest& req) override;
+
+ folly::Future<cpp2::GeneralResponse>
+ future_get(const cpp2::GetRequest& req) override;
+
  folly::Future<cpp2::GetUUIDResp>
  future_getUUID(const cpp2::GetUUIDReq& req) override;
 

src/storage/client/StorageClient.cpp

@@ -406,5 +406,53 @@ PartitionID StorageClient::partId(GraphSpaceID spaceId, int64_t id) const {
  return s;
 }
 
+folly::SemiFuture<StorageRpcResponse<cpp2::ExecResponse>>
+StorageClient::put(GraphSpaceID space,
+ std::vector<nebula::cpp2::Pair> values,
+ folly::EventBase* evb) {
+ auto clusters = clusterIdsToHosts(space, values,
+ [] (const nebula::cpp2::Pair& v) {
+ return std::hash<std::string>{}(v.get_key());
+ });
+
+ std::unordered_map<HostAddr, cpp2::PutRequest> requests;
+ for (auto& c : clusters) {
+ auto& host = c.first;
+ auto& req = requests[host];
+ req.set_space_id(space);
+ req.set_parts(std::move(c.second));
+ }
+
+ return collectResponse(evb, std::move(requests),
+ [](cpp2::StorageServiceAsyncClient* client,
+ const cpp2::PutRequest& r) {
+ return client->future_put(r);
+ });
+}
+
+folly::SemiFuture<StorageRpcResponse<storage::cpp2::GeneralResponse>>
+StorageClient::get(GraphSpaceID space,
+ const std::vector<std::string>& keys,
+ folly::EventBase* evb) {
+ auto clusters = clusterIdsToHosts(space, keys,
+ [] (const std::string& v) {
+ return std::hash<std::string>{}(v);
+ });
+
+ std::unordered_map<HostAddr, cpp2::GetRequest> requests;
+ for (auto& c : clusters) {
+ auto& host = c.first;
+ auto& req = requests[host];
+ req.set_space_id(space);
+ req.set_parts(std::move(c.second));
+ }
+
+ return collectResponse(evb, std::move(requests),
+ [](cpp2::StorageServiceAsyncClient* client,
+ const cpp2::GetRequest& r) {
+ return client->future_get(r);
+ });
+}
+
 } // namespace storage
 } // namespace nebula