-
Notifications
You must be signed in to change notification settings - Fork 1.2k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Cherry pick v3.1.0 (0412-0413) (#4145)
* Add SetUsageMessage For daemons (#4134) * fix the wrong regex pattern of scientific notation (#4136) * fix updatePart (#4139) * fix updatePart * fix format check Co-authored-by: Sophie <84560950+Sophie-Xie@users.noreply.github.com> * show DATA_BALANCE in job list (#4138) Co-authored-by: Sophie <84560950+Sophie-Xie@users.noreply.github.com> * fix disk fault recovery (#4131) * fix disk fault recovery * add ut Co-authored-by: Sophie <84560950+Sophie-Xie@users.noreply.github.com> * Refactor findpath (#4095) * optimize bfs * optimizer allpath * optimizer multi-shortestpath * optimizer multi shortest path * fix validate unit test * add some comment * fix error * fix bfs error * add comment * delete conjunct * add findpath unit test * delete useless file * delete log * remove check * multi thread executor * single shortest multi thread * add some testcases * add gflags * fix bfs error * address comment * Just report errorof property pruner in debug mode (#4142) Co-authored-by: Sophie <84560950+Sophie-Xie@users.noreply.github.com> * remove Atomic Edge (#4127) * try to remove Atomic Edge * remove some space * remove zone Co-authored-by: Doodle <13706157+critical27@users.noreply.github.com> * fix show tag/edge index status (#4148) * fix multiple match (#4143) Co-authored-by: Sophie <84560950+Sophie-Xie@users.noreply.github.com> * Refactor the process of symbols in optimizer. (#4146) * Refactor the process of symbols in optimizer. * Fix variable shadowing. * Collect boundary from MatchResult directly. * Check variable in TransformResult don't referenced by outside plan nodes. Co-authored-by: Sophie <84560950+Sophie-Xie@users.noreply.github.com> Co-authored-by: Alex Xing <90179377+SuperYoko@users.noreply.github.com> Co-authored-by: jie.wang <38901892+jievince@users.noreply.github.com> Co-authored-by: Doodle <13706157+critical27@users.noreply.github.com> Co-authored-by: liwenhui-soul <38217397+liwenhui-soul@users.noreply.github.com> Co-authored-by: panda-sheep <59197347+panda-sheep@users.noreply.github.com> Co-authored-by: jimingquan <mingquan.ji@vesoft.com> Co-authored-by: lionel.liu@vesoft.com <52276794+liuyu85cn@users.noreply.github.com> Co-authored-by: kyle.cao <kyle.cao@vesoft.com> Co-authored-by: shylock <33566796+Shylock-Hg@users.noreply.github.com>
- Loading branch information
10 people
authored
Apr 13, 2022
1 parent
50159ab
commit 85b77a5
Showing
85 changed files
with
2,914 additions
and
4,144 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,52 +1,217 @@ | ||
| /* Copyright (c) 2020 vesoft inc. All rights reserved. | ||
| * | ||
| * This source code is licensed under Apache 2.0 License. | ||
| */ | ||
| // Copyright (c) 2022 vesoft inc. All rights reserved. | ||
| // | ||
| // This source code is licensed under Apache 2.0 License. | ||
|
|
||
| #include "graph/executor/algo/BFSShortestPathExecutor.h" | ||
|
|
||
| #include "graph/planner/plan/Algo.h" | ||
|
|
||
| DECLARE_int32(num_operator_threads); | ||
| namespace nebula { | ||
| namespace graph { | ||
| folly::Future<Status> BFSShortestPathExecutor::execute() { | ||
| SCOPED_TIMER(&execTime_); | ||
| auto* bfs = asNode<BFSShortestPath>(node()); | ||
| auto iter = ectx_->getResult(bfs->inputVar()).iter(); | ||
| VLOG(1) << "current: " << node()->outputVar(); | ||
| VLOG(1) << "input: " << bfs->inputVar(); | ||
| pathNode_ = asNode<BFSShortestPath>(node()); | ||
|
|
||
| if (step_ == 1) { | ||
| allRightEdges_.emplace_back(); | ||
| auto& currentEdges = allRightEdges_.back(); | ||
| auto rIter = ectx_->getResult(pathNode_->rightVidVar()).iter(); | ||
| std::unordered_set<Value> rightVids; | ||
| for (; rIter->valid(); rIter->next()) { | ||
| auto& vid = rIter->getColumn(0); | ||
| if (rightVids.emplace(vid).second) { | ||
| Edge dummy; | ||
| currentEdges.emplace(vid, std::move(dummy)); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| std::vector<folly::Future<Status>> futures; | ||
| auto leftFuture = folly::via(runner(), [this]() { return buildPath(false); }); | ||
| auto rightFuture = folly::via(runner(), [this]() { return buildPath(true); }); | ||
| futures.emplace_back(std::move(leftFuture)); | ||
| futures.emplace_back(std::move(rightFuture)); | ||
|
|
||
| return folly::collect(futures) | ||
| .via(runner()) | ||
| .thenValue([this](auto&& status) { | ||
| UNUSED(status); | ||
| return conjunctPath(); | ||
| }) | ||
| .thenValue([this](auto&& status) { | ||
| UNUSED(status); | ||
| step_++; | ||
| DataSet ds; | ||
| ds.colNames = pathNode_->colNames(); | ||
| ds.rows.swap(currentDs_.rows); | ||
| return finish(ResultBuilder().value(Value(std::move(ds))).build()); | ||
| }); | ||
| } | ||
|
|
||
| Status BFSShortestPathExecutor::buildPath(bool reverse) { | ||
| auto iter = reverse ? ectx_->getResult(pathNode_->rightInputVar()).iter() | ||
| : ectx_->getResult(pathNode_->leftInputVar()).iter(); | ||
| DCHECK(!!iter); | ||
| auto& visitedVids = reverse ? rightVisitedVids_ : leftVisitedVids_; | ||
| auto& allEdges = reverse ? allRightEdges_ : allLeftEdges_; | ||
| allEdges.emplace_back(); | ||
| auto& currentEdges = allEdges.back(); | ||
|
|
||
| auto iterSize = iter->size(); | ||
| visitedVids.reserve(visitedVids.size() + iterSize); | ||
|
|
||
| std::unordered_set<Value> uniqueDst; | ||
| uniqueDst.reserve(iterSize); | ||
| DataSet nextStepVids; | ||
| nextStepVids.colNames = {nebula::kVid}; | ||
| if (step_ == 1) { | ||
| for (; iter->valid(); iter->next()) { | ||
| auto edgeVal = iter->getEdge(); | ||
| if (UNLIKELY(!edgeVal.isEdge())) { | ||
| continue; | ||
| } | ||
| auto& edge = edgeVal.getEdge(); | ||
| auto dst = edge.dst; | ||
| visitedVids.emplace(edge.src); | ||
| if (uniqueDst.emplace(dst).second) { | ||
| nextStepVids.rows.emplace_back(Row({dst})); | ||
| } | ||
| currentEdges.emplace(std::move(dst), std::move(edge)); | ||
| } | ||
| } else { | ||
| for (; iter->valid(); iter->next()) { | ||
| auto edgeVal = iter->getEdge(); | ||
| if (UNLIKELY(!edgeVal.isEdge())) { | ||
| continue; | ||
| } | ||
| auto& edge = edgeVal.getEdge(); | ||
| auto dst = edge.dst; | ||
| if (visitedVids.find(dst) != visitedVids.end()) { | ||
| continue; | ||
| } | ||
| if (uniqueDst.emplace(dst).second) { | ||
| nextStepVids.rows.emplace_back(Row({dst})); | ||
| } | ||
| currentEdges.emplace(std::move(dst), std::move(edge)); | ||
| } | ||
| } | ||
| // set nextVid | ||
| const auto& nextVidVar = reverse ? pathNode_->rightVidVar() : pathNode_->leftVidVar(); | ||
| ectx_->setResult(nextVidVar, ResultBuilder().value(std::move(nextStepVids)).build()); | ||
|
|
||
| visitedVids.insert(std::make_move_iterator(uniqueDst.begin()), | ||
| std::make_move_iterator(uniqueDst.end())); | ||
| return Status::OK(); | ||
| } | ||
|
|
||
| folly::Future<Status> BFSShortestPathExecutor::conjunctPath() { | ||
| const auto& leftEdges = allLeftEdges_.back(); | ||
| const auto& preRightEdges = allRightEdges_[step_ - 1]; | ||
| std::unordered_set<Value> meetVids; | ||
| bool oddStep = true; | ||
| for (const auto& edge : leftEdges) { | ||
| if (preRightEdges.find(edge.first) != preRightEdges.end()) { | ||
| meetVids.emplace(edge.first); | ||
| } | ||
| } | ||
| if (meetVids.empty() && step_ * 2 <= pathNode_->steps()) { | ||
| const auto& rightEdges = allRightEdges_.back(); | ||
| for (const auto& edge : leftEdges) { | ||
| if (rightEdges.find(edge.first) != rightEdges.end()) { | ||
| meetVids.emplace(edge.first); | ||
| oddStep = false; | ||
| } | ||
| } | ||
| } | ||
| if (meetVids.empty()) { | ||
| return Status::OK(); | ||
| } | ||
| size_t i = 0; | ||
| size_t totalSize = meetVids.size(); | ||
| size_t batchSize = totalSize / static_cast<size_t>(FLAGS_num_operator_threads); | ||
| std::vector<Value> batchVids; | ||
| batchVids.reserve(batchSize); | ||
| std::vector<folly::Future<DataSet>> futures; | ||
| for (auto& vid : meetVids) { | ||
| batchVids.push_back(vid); | ||
| if (i == totalSize - 1 || batchVids.size() == batchSize) { | ||
| auto future = folly::via(runner(), [this, vids = std::move(batchVids), oddStep]() { | ||
| return doConjunct(vids, oddStep); | ||
| }); | ||
| futures.emplace_back(std::move(future)); | ||
| } | ||
| i++; | ||
| } | ||
|
|
||
| return folly::collect(futures).via(runner()).thenValue([this](auto&& resps) { | ||
| for (auto& resp : resps) { | ||
| currentDs_.append(std::move(resp)); | ||
| } | ||
| return Status::OK(); | ||
| }); | ||
| } | ||
|
|
||
| DataSet BFSShortestPathExecutor::doConjunct(const std::vector<Value>& meetVids, | ||
| bool oddStep) const { | ||
| DataSet ds; | ||
| ds.colNames = node()->colNames(); | ||
| std::unordered_multimap<Value, Value> interim; | ||
|
|
||
| for (; iter->valid(); iter->next()) { | ||
| auto edgeVal = iter->getEdge(); | ||
| if (!edgeVal.isEdge()) { | ||
| continue; | ||
| } | ||
| auto& edge = edgeVal.getEdge(); | ||
| auto visited = visited_.find(edge.dst) != visited_.end(); | ||
| if (visited) { | ||
| continue; | ||
| } | ||
|
|
||
| // save the starts. | ||
| visited_.emplace(edge.src); | ||
| VLOG(1) << "dst: " << edge.dst << " edge: " << edge; | ||
| interim.emplace(edge.dst, std::move(edgeVal)); | ||
| } | ||
| for (auto& kv : interim) { | ||
| auto dst = std::move(kv.first); | ||
| auto edge = std::move(kv.second); | ||
| Row row; | ||
| row.values.emplace_back(dst); | ||
| row.values.emplace_back(std::move(edge)); | ||
| ds.rows.emplace_back(std::move(row)); | ||
| visited_.emplace(dst); | ||
| } | ||
| return finish(ResultBuilder().value(Value(std::move(ds))).build()); | ||
| auto leftPaths = createPath(meetVids, false, oddStep); | ||
| auto rightPaths = createPath(meetVids, true, oddStep); | ||
| for (auto& leftPath : leftPaths) { | ||
| auto range = rightPaths.equal_range(leftPath.first); | ||
| for (auto& rightPath = range.first; rightPath != range.second; ++rightPath) { | ||
| Path result = leftPath.second; | ||
| result.reverse(); | ||
| result.append(rightPath->second); | ||
| Row row; | ||
| row.emplace_back(std::move(result)); | ||
| ds.rows.emplace_back(std::move(row)); | ||
| } | ||
| } | ||
| return ds; | ||
| } | ||
|
|
||
| std::unordered_multimap<Value, Path> BFSShortestPathExecutor::createPath( | ||
| std::vector<Value> meetVids, bool reverse, bool oddStep) const { | ||
| std::unordered_multimap<Value, Path> result; | ||
| auto& allEdges = reverse ? allRightEdges_ : allLeftEdges_; | ||
| for (auto& meetVid : meetVids) { | ||
| Path start; | ||
| start.src = Vertex(meetVid, {}); | ||
| std::vector<Path> interimPaths = {std::move(start)}; | ||
| auto iter = (reverse && oddStep) ? allEdges.rbegin() + 1 : allEdges.rbegin(); | ||
| auto end = reverse ? allEdges.rend() - 1 : allEdges.rend(); | ||
| if (iter == end) { | ||
| result.emplace(meetVid, std::move(start)); | ||
| return result; | ||
| } | ||
| for (; iter != end; ++iter) { | ||
| std::vector<Path> temp; | ||
| for (auto& interimPath : interimPaths) { | ||
| Value id; | ||
| if (interimPath.steps.empty()) { | ||
| id = interimPath.src.vid; | ||
| } else { | ||
| id = interimPath.steps.back().dst.vid; | ||
| } | ||
| auto range = iter->equal_range(id); | ||
| for (auto edgeIter = range.first; edgeIter != range.second; ++edgeIter) { | ||
| Path p = interimPath; | ||
| auto& edge = edgeIter->second; | ||
| p.steps.emplace_back(Step(Vertex(edge.src, {}), -edge.type, edge.name, edge.ranking, {})); | ||
|
|
||
| if (iter == end - 1) { | ||
| result.emplace(p.src.vid, std::move(p)); | ||
| } else { | ||
| temp.emplace_back(std::move(p)); | ||
| } | ||
| } // edgeIter | ||
| } // interimPath | ||
| interimPaths = std::move(temp); | ||
| } | ||
| } | ||
| return result; | ||
| } | ||
|
|
||
| } // namespace graph | ||
| } // namespace nebula |
Oops, something went wrong.