Some minor fix (#4609)

* reserve enough space when merge storage executor's result * make filter logic same as ent version * address @SuperYoko's comments Co-authored-by: Sophie <[email protected]>
vesoft-inc · Sep 2, 2022 · 3157fad · 3157fad
1 parent 7b9fc17
commit 3157fad
Show file tree

Hide file tree

Showing 7 changed files with 60 additions and 10 deletions.
diff --git a/src/storage/exec/GetPropNode.h b/src/storage/exec/GetPropNode.h
@@ -94,8 +94,13 @@ class GetTagPropNode : public QueryNode<VertexID> {
         return ret;
       }
     }
-    if (filter_ == nullptr || (QueryUtils::vTrue(filter_->eval(*expCtx_)))) {
+    if (filter_ == nullptr) {
       resultDataSet_->rows.emplace_back(std::move(row));
+    } else {
+      auto result = QueryUtils::vTrue(filter_->eval(*expCtx_));
+      if (result.ok() && result.value()) {
+        resultDataSet_->rows.emplace_back(std::move(row));
+      }
     }
     if (expCtx_ != nullptr) {
       expCtx_->clear();
@@ -172,8 +177,13 @@ class GetEdgePropNode : public QueryNode<cpp2::EdgeKey> {
         return ret;
       }
     }
-    if (filter_ == nullptr || (QueryUtils::vTrue(filter_->eval(*expCtx_)))) {
+    if (filter_ == nullptr) {
       resultDataSet_->rows.emplace_back(std::move(row));
+    } else {
+      auto result = QueryUtils::vTrue(filter_->eval(*expCtx_));
+      if (result.ok() && result.value()) {
+        resultDataSet_->rows.emplace_back(std::move(row));
+      }
     }
     if (expCtx_ != nullptr) {
       expCtx_->clear();

diff --git a/src/storage/exec/QueryUtils.h b/src/storage/exec/QueryUtils.h
@@ -18,8 +18,16 @@ namespace storage {
 
 class QueryUtils final {
  public:
-  static inline bool vTrue(const Value& v) {
-    return v.isBool() && v.getBool();
+  // The behavior keep same with filter executor
+  static inline StatusOr<bool> vTrue(const Value& val) {
+    if (val.isBadNull() || (!val.empty() && !val.isBool() && !val.isNull())) {
+      return Status::Error("Wrong type result, the type should be NULL, EMPTY or BOOL");
+    }
+    if (val.empty() || val.isNull() || !val.getBool()) {
+      return false;
+    } else {
+      return true;
+    }
   }
 
   enum class ReturnColType : uint16_t {

diff --git a/src/storage/exec/ScanNode.h b/src/storage/exec/ScanNode.h
@@ -171,9 +171,15 @@ class ScanVertexPropNode : public QueryNode<Cursor> {
           break;
         }
       }
-      if (ret == nebula::cpp2::ErrorCode::SUCCEEDED &&
-          (filter_ == nullptr || QueryUtils::vTrue(filter_->eval(*expCtx_)))) {
-        resultDataSet_->rows.emplace_back(std::move(row));
+      if (ret == nebula::cpp2::ErrorCode::SUCCEEDED) {
+        if (filter_ == nullptr) {
+          resultDataSet_->rows.emplace_back(std::move(row));
+        } else {
+          auto result = QueryUtils::vTrue(filter_->eval(*expCtx_));
+          if (result.ok() && result.value()) {
+            resultDataSet_->rows.emplace_back(std::move(row));
+          }
+        }
       }
       expCtx_->clear();
       for (auto& tagNode : tagNodes_) {
@@ -323,9 +329,15 @@ class ScanEdgePropNode : public QueryNode<Cursor> {
         break;
       }
     }
-    if (ret == nebula::cpp2::ErrorCode::SUCCEEDED &&
-        (filter_ == nullptr || QueryUtils::vTrue(filter_->eval(*expCtx_)))) {
-      resultDataSet_->rows.emplace_back(std::move(row));
+    if (ret == nebula::cpp2::ErrorCode::SUCCEEDED) {
+      if (filter_ == nullptr) {
+        resultDataSet_->rows.emplace_back(std::move(row));
+      } else {
+        auto result = QueryUtils::vTrue(filter_->eval(*expCtx_));
+        if (result.ok() && result.value()) {
+          resultDataSet_->rows.emplace_back(std::move(row));
+        }
+      }
     }
     expCtx_->clear();
     for (auto& edgeNode : edgeNodes_) {

diff --git a/src/storage/query/GetNeighborsProcessor.cpp b/src/storage/query/GetNeighborsProcessor.cpp
@@ -130,8 +130,13 @@ void GetNeighborsProcessor::runInMultipleThread(const cpp2::GetNeighborsRequest&
   folly::collectAll(futures).via(executor_).thenTry([this](auto&& t) mutable {
     CHECK(!t.hasException());
     const auto& tries = t.value();
+    size_t sum = 0;
     for (size_t j = 0; j < tries.size(); j++) {
       CHECK(!tries[j].hasException());
+      sum += results_[j].size();
+    }
+    resultDataSet_.rows.reserve(sum);
+    for (size_t j = 0; j < tries.size(); j++) {
       const auto& [code, partId] = tries[j].value();
       if (code != nebula::cpp2::ErrorCode::SUCCEEDED) {
         handleErrorCode(code, spaceId_, partId);

diff --git a/src/storage/query/GetPropProcessor.cpp b/src/storage/query/GetPropProcessor.cpp
@@ -129,8 +129,13 @@ void GetPropProcessor::runInMultipleThread(const cpp2::GetPropRequest& req) {
   folly::collectAll(futures).via(executor_).thenTry([this](auto&& t) mutable {
     CHECK(!t.hasException());
     const auto& tries = t.value();
+    size_t sum = 0;
     for (size_t j = 0; j < tries.size(); j++) {
       CHECK(!tries[j].hasException());
+      sum += results_[j].size();
+    }
+    resultDataSet_.rows.reserve(sum);
+    for (size_t j = 0; j < tries.size(); j++) {
       const auto& [code, partId] = tries[j].value();
       if (code != nebula::cpp2::ErrorCode::SUCCEEDED) {
         handleErrorCode(code, spaceId_, partId);

diff --git a/src/storage/query/ScanEdgeProcessor.cpp b/src/storage/query/ScanEdgeProcessor.cpp
@@ -179,8 +179,13 @@ void ScanEdgeProcessor::runInMultipleThread(const cpp2::ScanEdgeRequest& req) {
   folly::collectAll(futures).via(executor_).thenTry([this](auto&& t) mutable {
     CHECK(!t.hasException());
     const auto& tries = t.value();
+    size_t sum = 0;
     for (size_t j = 0; j < tries.size(); j++) {
       CHECK(!tries[j].hasException());
+      sum += results_[j].size();
+    }
+    resultDataSet_.rows.reserve(sum);
+    for (size_t j = 0; j < tries.size(); j++) {
       const auto& [code, partId] = tries[j].value();
       if (code != nebula::cpp2::ErrorCode::SUCCEEDED) {
         handleErrorCode(code, spaceId_, partId);

diff --git a/src/storage/query/ScanVertexProcessor.cpp b/src/storage/query/ScanVertexProcessor.cpp
@@ -184,8 +184,13 @@ void ScanVertexProcessor::runInMultipleThread(const cpp2::ScanVertexRequest& req
   folly::collectAll(futures).via(executor_).thenTry([this](auto&& t) mutable {
     CHECK(!t.hasException());
     const auto& tries = t.value();
+    size_t sum = 0;
     for (size_t j = 0; j < tries.size(); j++) {
       CHECK(!tries[j].hasException());
+      sum += results_[j].size();
+    }
+    resultDataSet_.rows.reserve(sum);
+    for (size_t j = 0; j < tries.size(); j++) {
       const auto& [code, partId] = tries[j].value();
       if (code != nebula::cpp2::ErrorCode::SUCCEEDED) {
         handleErrorCode(code, spaceId_, partId);