use libcurl

remove some useless curl

impl http client with libcurl

refactor es client

abandon multi field in single fulltext index

compile

remove useless curl

fix compile

fix unittest

src/common/expression/test/CMakeLists.txt

@@ -33,7 +33,7 @@ set(expression_test_common_libs
  $<TARGET_OBJECTS:file_based_cluster_id_man_obj>
  $<TARGET_OBJECTS:process_obj>
  $<TARGET_OBJECTS:time_utils_obj>
- $<TARGET_OBJECTS:ft_es_graph_adapter_obj>
+ $<TARGET_OBJECTS:es_adapter_obj>
  $<TARGET_OBJECTS:ws_common_obj>
  $<TARGET_OBJECTS:version_obj>
  $<TARGET_OBJECTS:graph_session_obj>
@@ -47,6 +47,7 @@ set(expression_test_common_libs
  $<TARGET_OBJECTS:ssl_obj>
  $<TARGET_OBJECTS:memory_obj>
  $<TARGET_OBJECTS:gc_obj>
+ $<TARGET_OBJECTS:http_client_obj>
 )
 
 
@@ -95,6 +96,7 @@ nebula_add_test(
  gtest
  ${THRIFT_LIBRARIES}
  ${PROXYGEN_LIBRARIES}
+ curl
 )
 
 nebula_add_executable(
@@ -110,6 +112,7 @@ nebula_add_executable(
  boost_regex
  ${THRIFT_LIBRARIES}
  ${PROXYGEN_LIBRARIES}
+ curl
 )
 
 nebula_add_executable(

src/common/http/test/CMakeLists.txt

@@ -23,5 +23,4 @@ nebula_add_test(
  ${PROXYGEN_LIBRARIES}
  gtest
  curl
-
 )

src/common/http/test/HttpClientTest.cpp

@@ -1,4 +1,4 @@
-/* Copyright (c) 2019 vesoft inc. All rights reserved.
+/* Copyright (c) 2022 vesoft inc. All rights reserved.
  *
  * This source code is licensed under Apache 2.0 License.
  */

src/common/plugin/fulltext/CMakeLists.txt

@@ -3,13 +3,9 @@
 # This source code is licensed under Apache 2.0 License.
 
 nebula_add_library(
- ft_es_graph_adapter_obj OBJECT
- elasticsearch/ESGraphAdapter.cpp
-)
-
-nebula_add_library(
- ft_es_storage_adapter_obj OBJECT
- elasticsearch/ESStorageAdapter.cpp
+ es_adapter_obj OBJECT
+ elasticsearch/ESAdapter.cpp
+ elasticsearch/ESClient.cpp
 )
 
 nebula_add_subdirectory(test)

src/common/plugin/fulltext/elasticsearch/ESAdapter.cpp

@@ -0,0 +1,245 @@
+/* Copyright (c) 2022 vesoft inc. All rights reserved.
+ *
+ * This source code is licensed under Apache 2.0 License.
+ */
+
+#include "common/plugin/fulltext/elasticsearch/ESAdapter.h"
+
+#include <random>
+
+#include "fmt/printf.h"
+#include "openssl/sha.h"
+namespace nebula::plugin {
+
+using namespace fmt::literals; // NOLINT
+
+void ESBulk::put(const std::string& indexName,
+ const std::string& vid,
+ const std::string& src,
+ const std::string& dst,
+ int64_t rank,
+ const std::string& text) {
+ folly::dynamic action = folly::dynamic::object();
+ folly::dynamic metadata = folly::dynamic::object();
+ folly::dynamic body = folly::dynamic::object();
+ auto docId = ESAdapter::genDocID(vid, src, dst, rank);
+ metadata["_id"] = docId;
+ metadata["_type"] = "_doc";
+ metadata["_index"] = indexName;
+ action["index"] = std::move(metadata);
+ body["vid"] = vid;
+ body["src"] = src;
+ body["dst"] = dst;
+ body["rank"] = rank;
+ body["content"] = text;
+ documents_[docId] = {std::move(action), std::move(body)};
+}
+
+void ESBulk::delete_(const std::string& indexName,
+ const std::string& vid,
+ const std::string& src,
+ const std::string& dst,
+ int64_t rank) {
+ folly::dynamic action = folly::dynamic::object();
+ folly::dynamic metadata = folly::dynamic::object();
+ auto docId = ESAdapter::genDocID(vid, src, dst, rank);
+ metadata["_id"] = docId;
+ metadata["_type"] = "_doc";
+ metadata["_index"] = indexName;
+ action["delete"] = std::move(metadata);
+ documents_[docId] = {std::move(action)};
+}
+
+bool ESBulk::empty() {
+ return documents_.empty();
+}
+
+ESAdapter::ESAdapter(std::vector<ESClient>&& clients) : clients_(clients) {}
+
+void ESAdapter::setClients(std::vector<ESClient>&& clients) {
+ clients_ = std::move(clients);
+}
+
+Status ESAdapter::createIndex(const std::string& name) {
+ folly::dynamic mappings = folly::parseJson(R"(
+ {
+ "mappings":{
+ "properties":{
+ "vid": {
+ "type": "keyword"
+ },
+ "src": {
+ "type": "keyword"
+ },
+ "dst": {
+ "type": "keyword"
+ },
+ "rank": {
+ "type": "long"
+ },
+ "text": {
+ "type": "text"
+ }
+ }
+ }
+ }
+ )");
+ auto result = randomClient().createIndex(name, mappings);
+ if (!result.ok()) {
+ return result.status();
+ }
+ auto resp = std::move(result).value();
+ if (resp["acknowledged"].isBool() && resp["acknowledged"].getBool()) {
+ return Status::OK();
+ }
+ auto error = resp["error"];
+ if (error.isObject()) {
+ return Status::Error(folly::toJson(error));
+ }
+ return Status::Error(folly::toJson(resp));
+}
+
+Status ESAdapter::dropIndex(const std::string& name) {
+ auto result = randomClient().dropIndex(name);
+ if (!result.ok()) {
+ return result.status();
+ }
+ auto resp = std::move(result).value();
+ if (resp["acknowledged"].isBool() && resp["acknowledged"].getBool()) {
+ return Status::OK();
+ }
+ auto error = resp["error"];
+ if (error.isObject()) {
+ return Status::Error(folly::toJson(error));
+ }
+ return Status::Error(folly::toJson(resp));
+}
+
+Status ESAdapter::clearIndex(const std::string& name) {
+ auto result = randomClient().clearIndex(name);
+ if (!result.ok()) {
+ return result.status();
+ }
+ auto resp = std::move(result).value();
+ if (resp["failures"].isArray() && resp["failures"].size() == 0) {
+ return Status::OK();
+ }
+ auto error = resp["error"];
+ if (error.isObject()) {
+ return Status::Error(folly::toJson(error));
+ }
+ return Status::Error(folly::toJson(resp));
+}
+
+StatusOr<bool> ESAdapter::isIndexExist(const std::string& name) {
+ auto result = randomClient().getIndex(name);
+ if (!result.ok()) {
+ return result.status();
+ }
+ auto resp = std::move(result).value();
+
+ if (resp[name].isObject()) {
+ return true;
+ }
+ auto error = resp["error"];
+ if (error.isObject()) {
+ if (error["type"].isString() && error["type"].getString() == "index_not_found_exception") {
+ return false;
+ }
+ return Status::Error(folly::toJson(error));
+ }
+ return Status::Error(folly::toJson(resp));
+}
+
+Status ESAdapter::bulk(const ESBulk& bulk) {
+ std::vector<folly::dynamic> jsonArray;
+ for (auto& [docId, objs] : bulk.documents_) {
+ for (auto& obj : objs) {
+ jsonArray.push_back(obj);
+ }
+ }
+ auto result = randomClient().bulk(jsonArray);
+ if (!result.ok()) {
+ return result.status();
+ }
+ auto resp = std::move(result).value();
+ if (resp["errors"].isBool() && !resp["errors"].getBool()) {
+ return Status::OK();
+ }
+ auto error = resp["error"];
+ if (error.isObject()) {
+ return Status::Error(folly::toJson(error));
+ }
+ return Status::Error(folly::toJson(resp));
+}
+
+StatusOr<ESQueryResult> ESAdapter::prefix(const std::string& index, const std::string& pattern) {
+ folly::dynamic query = folly::dynamic::object("query", folly::dynamic::object("prefix", pattern));
+ return ESAdapter::query(index, query);
+}
+
+StatusOr<ESQueryResult> ESAdapter::wildcard(const std::string& index, const std::string& pattern) {
+ folly::dynamic query =
+ folly::dynamic::object("query", folly::dynamic::object("wildcard", pattern));
+ return ESAdapter::query(index, query);
+}
+
+StatusOr<ESQueryResult> ESAdapter::regexp(const std::string& index, const std::string& pattern) {
+ folly::dynamic query = folly::dynamic::object("query", folly::dynamic::object("regexp", pattern));
+ return ESAdapter::query(index, query);
+}
+
+StatusOr<ESQueryResult> ESAdapter::fuzzy(const std::string& index, const std::string& pattern) {
+ folly::dynamic query = folly::dynamic::object("query", folly::dynamic::object("fuzzy", pattern));
+ return ESAdapter::query(index, query);
+}
+
+StatusOr<ESQueryResult> ESAdapter::query(const std::string& index, const folly::dynamic& query) {
+ auto result = randomClient().search(index, query);
+ if (!result.ok()) {
+ return std::move(result).status();
+ }
+ auto resp = std::move(result).value();
+ auto hits = resp["hits"];
+ if (hits.isObject()) {
+ ESQueryResult res;
+ for (auto& hit : hits) {
+ auto source = hit["_source"];
+ ESQueryResult::Item item;
+ item.text = source["text"].getString();
+ item.src = source["src"].getString();
+ item.dst = source["dst"].getString();
+ item.rank = source["rank"].getInt();
+ item.vid = source["vid"].getString();
+ res.items.emplace_back(std::move(item));
+ }
+ return res;
+ }
+ auto error = resp["error"];
+ if (error.isObject()) {
+ return Status::Error(folly::toJson(error));
+ }
+ return Status::Error(folly::toJson(resp));
+}
+
+std::string ESAdapter::genDocID(const std::string& vid,
+ const std::string& src,
+ const std::string& dst,
+ int64_t rank) {
+ std::string str;
+ unsigned char mdStr[33] = {0};
+ if (!vid.empty()) {
+ str = vid;
+ } else {
+ str = src + dst + std::to_string(rank);
+ }
+ SHA256(reinterpret_cast<unsigned char*>(str.data()), str.size(), mdStr);
+ return folly::hexDump(mdStr, 32);
+}
+
+ESClient& ESAdapter::randomClient() {
+ static thread_local std::default_random_engine engine;
+ static thread_local std::uniform_int_distribution<size_t> d(0, clients_.size() - 1);
+ return clients_[d(engine)];
+}
+} // namespace nebula::plugin