Add explicit join

.linters/cpp/checkKeyword.py

@@ -119,6 +119,13 @@
     'KW_GEOGRAPHY',
     'KW_DURATION',
     'KW_ACROSS',
+    'KW_JOIN',
+    'KW_LEFT',
+    'KW_RIGHT',
+    'KW_INNER',
+    'KW_OUTER',
+    'KW_SEMI',
+    'KW_ANTI',
 ]
 
 

src/common/expression/test/FunctionCallExpressionTest.cpp

@@ -44,10 +44,10 @@ TEST_F(FunctionCallExpressionTest, FunctionCallTest) {
   }
   {
     TEST_FUNCTION(substr, args_["substr"], "cdef");
-    TEST_FUNCTION(left, args_["side"], "abcde");
-    TEST_FUNCTION(right, args_["side"], "mnopq");
-    TEST_FUNCTION(left, args_["neg_side"], Value::kNullValue);
-    TEST_FUNCTION(right, args_["neg_side"], Value::kNullValue);
+    TEST_FUNCTION(`left`, args_["side"], "abcde");
+    TEST_FUNCTION(`right`, args_["side"], "mnopq");
+    TEST_FUNCTION(`left`, args_["neg_side"], Value::kNullValue);
+    TEST_FUNCTION(`right`, args_["neg_side"], Value::kNullValue);
 
     TEST_FUNCTION(lpad, args_["pad"], "1231abcdefghijkl");
     TEST_FUNCTION(rpad, args_["pad"], "abcdefghijkl1231");

src/graph/context/ast/QueryAstContext.h

@@ -54,6 +54,7 @@ struct PathContext final : AstContext {
   std::string toVidsVar;
 
   bool isShortest{false};
+  bool singleShortest{false};
   bool isWeight{false};
   bool noLoop{false};
   bool withProp{false};

src/graph/executor/algo/BFSShortestPathExecutor.cpp

@@ -13,6 +13,7 @@ folly::Future<Status> BFSShortestPathExecutor::execute() {
   // MemoryTrackerVerified
   SCOPED_TIMER(&execTime_);
   pathNode_ = asNode<BFSShortestPath>(node());
+  singleShortest_ = pathNode_->singleShortest();
   terminateEarlyVar_ = pathNode_->terminateEarlyVar();
 
   if (step_ == 1) {
@@ -43,16 +44,25 @@ folly::Future<Status> BFSShortestPathExecutor::execute() {
   futures.emplace_back(std::move(leftFuture));
   futures.emplace_back(std::move(rightFuture));
 
-  return folly::collect(futures)
+  return folly::collectAll(futures)
       .via(runner())
-      .thenValue([this](auto&& status) {
+      .thenValue([this](std::vector<folly::Try<Status>>&& resps) {
         memory::MemoryCheckGuard guard;
-        UNUSED(status);
+        for (auto& respVal : resps) {
+          if (respVal.hasException()) {
+            auto ex = respVal.exception().get_exception<std::bad_alloc>();
+            if (ex) {
+              throw std::bad_alloc();
+            } else {
+              throw std::runtime_error(respVal.exception().what().c_str());
+            }
+          }
+        }
         return conjunctPath();
       })
       .thenValue([this](auto&& status) {
         memory::MemoryCheckGuard guard;
-        UNUSED(status);
+        NG_RETURN_IF_ERROR(status);
         step_++;
         DataSet ds;
         ds.colNames = pathNode_->colNames();
@@ -166,13 +176,26 @@ folly::Future<Status> BFSShortestPathExecutor::conjunctPath() {
     }
   }
 
-  return folly::collect(futures).via(runner()).thenValue([this](auto&& resps) {
-    memory::MemoryCheckGuard guard;
-    for (auto& resp : resps) {
-      currentDs_.append(std::move(resp));
-    }
-    return Status::OK();
-  });
+  return folly::collectAll(futures).via(runner()).thenValue(
+      [this](std::vector<folly::Try<DataSet>>&& resps) {
+        memory::MemoryCheckGuard guard;
+        for (auto& respVal : resps) {
+          if (respVal.hasException()) {
+            auto ex = respVal.exception().get_exception<std::bad_alloc>();
+            if (ex) {
+              throw std::bad_alloc();
+            } else {
+              throw std::runtime_error(respVal.exception().what().c_str());
+            }
+          }
+          auto resp = std::move(respVal).value();
+          currentDs_.append(std::move(resp));
+        }
+        if (singleShortest_) {
+          currentDs_.rows.resize(1);
+        }
+        return Status::OK();
+      });
 }
 
 DataSet BFSShortestPathExecutor::doConjunct(const std::vector<Value>& meetVids,
@@ -189,6 +212,9 @@ DataSet BFSShortestPathExecutor::doConjunct(const std::vector<Value>& meetVids,
       Row row;
       row.emplace_back(std::move(result));
       ds.rows.emplace_back(std::move(row));
+      if (singleShortest_) {
+        return ds;
+      }
     }
   }
   return ds;

src/graph/executor/algo/BFSShortestPathExecutor.h

@@ -63,6 +63,7 @@ class BFSShortestPathExecutor final : public Executor {
 
  private:
   const BFSShortestPath* pathNode_{nullptr};
+  bool singleShortest_{false};
   size_t step_{1};
   HashSet leftVisitedVids_;
   HashSet rightVisitedVids_;

src/graph/executor/algo/MultiShortestPathExecutor.cpp

@@ -11,6 +11,7 @@ folly::Future<Status> MultiShortestPathExecutor::execute() {
   SCOPED_TIMER(&execTime_);
   pathNode_ = asNode<MultiShortestPath>(node());
   terminationVar_ = pathNode_->terminationVar();
+  singleShortest_ = pathNode_->singleShortest();
 
   if (step_ == 1) {
     init();
@@ -28,12 +29,21 @@ folly::Future<Status> MultiShortestPathExecutor::execute() {
   futures.emplace_back(std::move(leftFuture));
   futures.emplace_back(std::move(rightFuture));
 
-  return folly::collect(futures)
+  return folly::collectAll(futures)
       .via(runner())
-      .thenValue([this](auto&& status) {
-        memory::MemoryCheckGuard guard;
+      .thenValue([this](std::vector<folly::Try<Status>>&& resps) {
         // oddStep
-        UNUSED(status);
+        for (auto& respVal : resps) {
+          if (respVal.hasException()) {
+            auto ex = respVal.exception().get_exception<std::bad_alloc>();
+            if (ex) {
+              throw std::bad_alloc();
+            } else {
+              throw std::runtime_error(respVal.exception().what().c_str());
+            }
+          }
+        }
+        memory::MemoryCheckGuard guard;
         return conjunctPath(true);
       })
       .thenValue([this](auto&& termination) {
@@ -224,7 +234,7 @@ Status MultiShortestPathExecutor::buildPath(bool reverse) {
 DataSet MultiShortestPathExecutor::doConjunct(
     const std::vector<std::pair<Interims::iterator, Interims::iterator>>& iters) {
   auto buildPaths =
-      [](const std::vector<Path>& leftPaths, const std::vector<Path>& rightPaths, DataSet& ds) {
+      [this](const std::vector<Path>& leftPaths, const std::vector<Path>& rightPaths, DataSet& ds) {
         for (const auto& leftPath : leftPaths) {
           for (const auto& rightPath : rightPaths) {
             auto forwardPath = leftPath;
@@ -234,6 +244,9 @@ DataSet MultiShortestPathExecutor::doConjunct(
             Row row;
             row.values.emplace_back(std::move(forwardPath));
             ds.rows.emplace_back(std::move(row));
+            if (singleShortest_) {
+              return;
+            }
           }
         }
       };
@@ -247,6 +260,9 @@ DataSet MultiShortestPathExecutor::doConjunct(
       for (const auto& rightPath : rightPaths) {
         for (auto found = range.first; found != range.second; ++found) {
           if (found->second.first == rightPath.first) {
+            if (singleShortest_ && !found->second.second) {
+              break;
+            }
             buildPaths(leftPath.second, rightPath.second, ds);
             found->second.second = false;
           }
@@ -312,25 +328,55 @@ folly::Future<bool> MultiShortestPathExecutor::conjunctPath(bool oddStep) {
     futures.emplace_back(std::move(future));
   }
 
-  return folly::collect(futures).via(runner()).thenValue([this](auto&& resps) {
-    memory::MemoryCheckGuard guard;
-    for (auto& resp : resps) {
-      currentDs_.append(std::move(resp));
-    }
+  return folly::collectAll(futures).via(runner()).thenValue(
+      [this](std::vector<folly::Try<DataSet>>&& resps) {
+        memory::MemoryCheckGuard guard;
+        std::unordered_map<Value, std::unordered_set<Value>> uniquePath;
+        for (auto& respVal : resps) {
+          if (respVal.hasException()) {
+            auto ex = respVal.exception().get_exception<std::bad_alloc>();
+            if (ex) {
+              throw std::bad_alloc();
+            } else {
+              throw std::runtime_error(respVal.exception().what().c_str());
+            }
+          }
+          auto ds = std::move(respVal).value();
+          if (singleShortest_) {
+            for (auto& row : ds.rows) {
+              auto& pathVal = row.values.front();
+              auto& path = pathVal.getPath();
+              auto& src = path.src.vid;
+              auto& dst = path.steps.back().dst.vid;
+              auto findSrc = uniquePath.find(src);
+              if (findSrc == uniquePath.end()) {
+                uniquePath[src] = {dst};
+                currentDs_.rows.emplace_back(std::move(row));
+              } else {
+                if (findSrc->second.find(dst) == findSrc->second.end()) {
+                  findSrc->second.emplace(dst);
+                  currentDs_.rows.emplace_back(std::move(row));
+                }
+              }
+            }
+          } else {
+            currentDs_.append(std::move(ds));
+          }
+        }
 
-    for (auto iter = terminationMap_.begin(); iter != terminationMap_.end();) {
-      if (!iter->second.second) {
-        iter = terminationMap_.erase(iter);
-      } else {
-        ++iter;
-      }
-    }
-    if (terminationMap_.empty()) {
-      ectx_->setValue(terminationVar_, true);
-      return true;
-    }
-    return false;
-  });
+        for (auto iter = terminationMap_.begin(); iter != terminationMap_.end();) {
+          if (!iter->second.second) {
+            iter = terminationMap_.erase(iter);
+          } else {
+            ++iter;
+          }
+        }
+        if (terminationMap_.empty()) {
+          ectx_->setValue(terminationVar_, true);
+          return true;
+        }
+        return false;
+      });
 }
 
 void MultiShortestPathExecutor::setNextStepVid(const Interims& paths, const string& var) {

src/graph/executor/algo/MultiShortestPathExecutor.h

@@ -75,6 +75,7 @@ class MultiShortestPathExecutor final : public Executor {
 
  private:
   const MultiShortestPath* pathNode_{nullptr};
+  bool singleShortest_{false};
   size_t step_{1};
   std::string terminationVar_;
   // {src, <dst, true>}

src/graph/executor/query/DataCollectExecutor.cpp

@@ -112,8 +112,6 @@ Status DataCollectExecutor::rowBasedMove(const std::vector<std::string>& vars) {
       for (; seqIter->valid(); seqIter->next()) {
         ds.rows.emplace_back(seqIter->moveRow());
       }
-    } else {
-      return Status::Error("Iterator should be kind of SequentialIter.");
     }
   }
   result_.setDataSet(std::move(ds));
@@ -140,10 +138,6 @@ Status DataCollectExecutor::collectAllPaths(const std::vector<std::string>& vars
         for (; seqIter->valid(); seqIter->next()) {
           ds.rows.emplace_back(seqIter->moveRow());
         }
-      } else {
-        std::stringstream msg;
-        msg << "Iterator should be kind of SequentialIter, but was: " << iter->kind();
-        return Status::Error(msg.str());
       }
     }
   }

src/graph/planner/ngql/PathPlanner.cpp

@@ -117,6 +117,7 @@ SubPlan PathPlanner::singlePairPlan(PlanNode* left, PlanNode* right) {
   path->setRightVidVar(pathCtx_->toVidsVar);
   path->setColNames({kPathStr});
   path->setTerminateEarlyVar(terminateEarlyVar);
+  path->setSingleShortest(pathCtx_->singleShortest);
 
   auto* loopCondition = singlePairLoopCondition(steps, path->outputVar(), terminateEarlyVar);
   auto* loop = Loop::make(qctx, nullptr, path, loopCondition);
@@ -220,6 +221,7 @@ SubPlan PathPlanner::multiPairPlan(PlanNode* left, PlanNode* right) {
   path->setLeftVidVar(pathCtx_->fromVidsVar);
   path->setRightVidVar(pathCtx_->toVidsVar);
   path->setTerminationVar(terminationVar);
+  path->setSingleShortest(pathCtx_->singleShortest);
   path->setColNames({kPathStr});
 
   SubPlan loopDep = loopDepPlan();

src/graph/planner/plan/Algo.cpp

@@ -45,6 +45,7 @@ std::unique_ptr<PlanNodeDescription> BFSShortestPath::explain() const {
   addDescription("LeftNextVidVar", folly::toJson(util::toJson(leftVidVar_)), desc.get());
   addDescription("RightNextVidVar", folly::toJson(util::toJson(rightVidVar_)), desc.get());
   addDescription("steps", folly::toJson(util::toJson(steps_)), desc.get());
+  addDescription("singleShortest", folly::toJson(util::toJson(singleShortest_)), desc.get());
   return desc;
 }
 
@@ -53,6 +54,7 @@ std::unique_ptr<PlanNodeDescription> MultiShortestPath::explain() const {
   addDescription("LeftNextVidVar", folly::toJson(util::toJson(leftVidVar_)), desc.get());
   addDescription("RightNextVidVar", folly::toJson(util::toJson(rightVidVar_)), desc.get());
   addDescription("steps", folly::toJson(util::toJson(steps_)), desc.get());
+  addDescription("singleShortest", folly::toJson(util::toJson(singleShortest_)), desc.get());
   return desc;
 }
 

src/graph/planner/plan/Algo.h

@@ -39,6 +39,10 @@ class MultiShortestPath : public BinaryInputNode {
     return terminationVar_;
   }
 
+  bool singleShortest() const {
+    return singleShortest_;
+  }
+
   void setLeftVidVar(const std::string& var) {
     leftVidVar_ = var;
   }
@@ -51,6 +55,10 @@ class MultiShortestPath : public BinaryInputNode {
     terminationVar_ = var;
   }
 
+  void setSingleShortest(bool singleShortest) {
+    singleShortest_ = singleShortest;
+  }
+
   std::unique_ptr<PlanNodeDescription> explain() const override;
 
  private:
@@ -59,6 +67,7 @@ class MultiShortestPath : public BinaryInputNode {
       : BinaryInputNode(qctx, Kind::kMultiShortestPath, left, right), steps_(steps) {}
 
  private:
+  bool singleShortest_{false};
   size_t steps_{0};
   std::string leftVidVar_;
   std::string rightVidVar_;
@@ -87,6 +96,10 @@ class BFSShortestPath : public BinaryInputNode {
     return terminateEarlyVar_;
   }
 
+  bool singleShortest() const {
+    return singleShortest_;
+  }
+
   void setLeftVidVar(const std::string& var) {
     leftVidVar_ = var;
   }
@@ -99,6 +112,10 @@ class BFSShortestPath : public BinaryInputNode {
     terminateEarlyVar_ = var;
   }
 
+  void setSingleShortest(bool singleShortest) {
+    singleShortest_ = singleShortest;
+  }
+
   std::unique_ptr<PlanNodeDescription> explain() const override;
 
  private:
@@ -107,6 +124,7 @@ class BFSShortestPath : public BinaryInputNode {
       : BinaryInputNode(qctx, Kind::kBFSShortest, left, right), steps_(steps) {}
 
  private:
+  bool singleShortest_{false};
   std::string leftVidVar_;
   std::string rightVidVar_;
   std::string terminateEarlyVar_;

src/graph/planner/plan/Query.cpp

@@ -703,6 +703,10 @@ std::unique_ptr<PlanNodeDescription> DataCollect::explain() const {
 PlanNode* DataCollect::clone() const {
   auto* newDataCollect = DataCollect::make(qctx_, kind_);
   newDataCollect->cloneMembers(*this);
+  // when cloning, the number of dependencies will be lost, needs to be added manually
+  for (size_t i = 0; i < numDeps(); ++i) {
+    newDataCollect->addDep(nullptr);
+  }
   return newDataCollect;
 }