讀 NebulaGraph源碼 ｜ 查詢語句 LOOKUP 的一生
本文由社區用户 Milittle 供稿
LOOKUP 是圖數據庫 NebulaGraph 的一個查詢語句。它依賴索引，可以查詢點或者邊的信息。在本文，我將着重從源碼的角度解析一下 LOOKUP 語句的一生是如何度過的。
本文源碼閲讀基於內核源碼的 v3.3.0 版本，詳見 GitHub http://github.com/vesoft-inc/nebula/releases/tag/v3.3.0
讀源碼之前
首先，我們需要明確 NebulaGraph 中 LOOKUP 語句的語法：
LOOKUP ON {<vertex_tag> | <edge_type>}
[WHERE <expression> [AND <expression> ...]]
YIELD <return_list> [AS <alias>]
[<clause>];
<return_list>
<prop_name> [AS <col_alias>] [, <prop_name> [AS <prop_alias>] ...];
<vertex_tag>是 Tag 的類型，比如：數據集 basketballplayer 中的 player 和 team；<edge_type>是 EdgeType 的類型，比如：數據集 basketballplayer 中的 follow 和 serve；<expression>是表達式；<return_list>是返回的列表，比如：id(vertex)，這部分內容詳細參見 nGQL 的 Schema 函數 nGQL Schema 函數詳解；<clause>是子句，可以是ORDER BY、LIMIT等子句，子句詳情參見 子句；
這裏有個 LOOKUP 使用注意事項：
- 如果已經存在點、邊，但是沒有索引。必須在新建索引後再通過
REBUILD INDEX重建索引，才能使其生效；
讀語句解析原理
為了便於大家理解這裏放一張 NebulaGraph 計算層的服務架構：
我們再來看下此次閲讀的語句，是一個比較簡單的 LOOKUP Sentence。用比較簡單的語句來解析 LOOKUP 語句的基本原理，後面可以慢慢擴展條件語句和子句：
// 我們需要分析以下語句
LOOKUP ON player YIELD id(vertex);
1. 從 Parser 開始
我們先從 Parser 入手分析 LOOKUP Sentence 的組成部分。這裏不介紹 lex 詞法分析和 yacc 語法分析，感興趣的小夥伴自己可以瞭解一下。下面，我們直接上我們關心的部分：
我們打開源碼，找到文件 src/parser/parser.yy 文件，裏面有所有語句的定義。我們 定位到 LOOKUP Sentence，是這裏 http://github.com/Milittle/nebula/blob/90a3107044ce1621c7834a0f36a4eef273ec2f31/src/parser/parser.yy#L2176。下面便是 LOOKUP 語句的定義，你也可以拷貝上面的鏈接訪問 GitHub 查看。來，我們分析分析每個部分：
/// LOOKUP 語句的語法定義
lookup_sentence
: KW_LOOKUP KW_ON name_label lookup_where_clause yield_clause {
$$ = new LookupSentence($3, $4, $5);
}
;
// KW_LOOKUP 是 LOOKUP 的關鍵字，大小寫不敏感的
// KW_ON 是 ON 的關鍵字，大小寫不敏感的
// name_label 是 LABEL 的定義，也是 strval，簡單的説就是字符串
// lookup_where_clause 是 WHERE 子句的定義，這個我們後面有機會擴展介紹，也有一個對應的語義定義
// yield_clause 這個是 YIELD 輸出數據的關鍵語句，在 v3.x 版本以後，YIELD 子句是必須要指定的，不指定會報語法錯誤
/// YIELD clause 的語法定義，其實 YIELD clause 用在了很多其他語句中，比如 GO、FIND PATH、GET SUBGRAPH
yield_clause
: %empty { $$ = nullptr; }
| KW_YIELD yield_columns {
if ($2->hasAgg()) {
delete($2);
throw nebula::GraphParser::syntax_error(@2, "Invalid use of aggregating function in yield clause.");
}
$$ = new YieldClause($2);
}
| KW_YIELD KW_DISTINCT yield_columns {
if ($3->hasAgg()) {
delete($3);
throw nebula::GraphParser::syntax_error(@3, "Invalid use of aggregating function in yield clause.");
}
$$ = new YieldClause($3, true);
}
;
// 可以為 empty，但是後面 validator 會進行校驗，不指定就會報 Error
// KW_YIELD 是 YIELD 的關鍵字，大小寫不敏感
// yield_columns 是輸出的列信息，也有對應的一個語法定義
// KW_DISTINCT 是 distinct 關鍵字，表示是否去除重複數據的語義，大小寫不敏感
// LOOKUP Sentence 就是上面所有的信息組成，都會被構造在這個類裏面，也就是 LOOKUP 語句的內容了
下面，我們繼續從 lookup_sentence 語句的定義往下規約看，可以看到它屬於 src/parser/parser.yy:2917: traverse_sentence → src/parser/parser.yy:2936: piped_sentence → src/parser/parser.yy:2942: set_sentence → src/parser/parser.yy:3924: sentence → src/parser/parser.yy:3933: seq_sentence。
其實，上面這些你可以暫時忽略，因為這些都是對 sentence 的規約抽象，有些集合語句和管道語句。這裏，我想表達的是這些語句一定會映射到 seq_sentence 上的，即，序列語句。你可以把它理解為用分號分隔的複合語句，只不過這裏面只包含了一條 lookup_sentence 而已。這樣子，就好理解為什麼下文在 seq_sentence 尋找入口代碼，而不是 lookup_sentence.
2. 從 nGQL 解析看 LOOKUP 語句
第二，從 nGQL 的解析過程繼續看 LOOKUP Sentence。其實，剛才已經強調過了，這裏解析出來的對象一定是 seq_sentence。
/// src/graph/service/QueryInstance.cpp
void QueryInstance::execute() {
Status status = validateAndOptimize(); // 1. 負責 validate、執行計劃生成、執行計劃優化等工作
if (!status.ok()) {
onError(std::move(status));
return;
}
// Sentence is explain query, finish
if (!explainOrContinue()) { // 6. 判斷是否是 explain 語句。如果是，直接輸出執行計劃，不做實際物理算子執行
onFinish();
return;
}
// The execution engine converts the physical execution plan generated by the Planner into a
// series of Executors through the Scheduler to drive the execution of the Executors.
scheduler_->schedule() // 7. 實際物理算子調度執行的部分，通過 DAG，對每一個 plan -> executor 的轉換執行（後續步驟會進行詳解）
.thenValue([this](Status s) {
if (s.ok()) {
this->onFinish(); // 8. 這裏是幹完了所有物理執行計劃，然後開始處理客户端 resp 了
} else {
this->onError(std::move(s)); // 9. 這裏是上面的過程出錯了，需要處理 Error 信息
}
}) // 10. 下面是處理一些異常情況，也是走錯誤分支
.thenError(folly::tag_t<ExecutionError>{},
[this](const ExecutionError &e) { onError(e.status()); })
.thenError(folly::tag_t<std::exception>{},
[this](const std::exception &e) { onError(Status::Error("%s", e.what())); });
}
// 這個函數執行的是註釋 1 的內容
Status QueryInstance::validateAndOptimize() {
auto *rctx = qctx()->rctx();
auto &spaceName = rctx->session()->space().name;
VLOG(1) << "Parsing query: " << rctx->query();
// Result of parsing, get the parsing tree
// 2. 第一步中的語法解析就是這裏的解釋，對 nGQL 進行詞法語法解析，出來的 result 就是 Sentence*，通過我們上面的分析，這裏吐出來的就是 seq_sentence 了
auto result = GQLParser(qctx()).parse(rctx->query());
NG_RETURN_IF_ERROR(result);
sentence_ = std::move(result).value();
// 3. 這裏是做指標的統計。這個可以在 dashboard 裏面展示
if (sentence_->kind() == Sentence::Kind::kSequential) {
size_t num = static_cast<const SequentialSentences *>(sentence_.get())->numSentences();
stats::StatsManager::addValue(kNumSentences, num);
if (FLAGS_enable_space_level_metrics && spaceName != "") {
stats::StatsManager::addValue(
stats::StatsManager::counterWithLabels(kNumSentences, {{"space", spaceName}}), num);
}
} else {
stats::StatsManager::addValue(kNumSentences);
if (FLAGS_enable_space_level_metrics && spaceName != "") {
stats::StatsManager::addValue(
stats::StatsManager::counterWithLabels(kNumSentences, {{"space", spaceName}}));
}
}
// Validate the query, if failed, return
// 4. 這個是源碼校驗 nGQL 解析出來的內容是否符合我們的預期，如果不符合預期就報語法錯誤
// validate 過程還會涉及到執行計劃的生成，重點函數
NG_RETURN_IF_ERROR(Validator::validate(sentence_.get(), qctx()));
// Optimize the query, and get the execution plan
// 5. 對上面生成的執行計劃進行 RBO 規則的優化，這個留在後面有機會再介紹
NG_RETURN_IF_ERROR(findBestPlan());
stats::StatsManager::addValue(kOptimizerLatencyUs, *(qctx_->plan()->optimizeTimeInUs()));
if (FLAGS_enable_space_level_metrics && spaceName != "") {
stats::StatsManager::addValue(
stats::StatsManager::histoWithLabels(kOptimizerLatencyUs, {{"space", spaceName}}));
}
return Status::OK();
}
我們按照上面的註釋部分進行講解，有的比較容易的部分，像註釋 1、2、3、5。我們下面重點介紹註釋 4 的部分
// src/graph/validator/Validator.cpp
// Entry of validating sentence.
// Check session, switch space of validator context, create validators and validate.
// static
// 1. 參數 sentence 就是剛才我們從語法解析器中拿到的 seq_sentence
// 2. 參數 qctx 是我們查詢上下文，一個語句進來對應一個查詢上下文，這個是在 QueryEngine 裏面生成的，感興趣可以自行閲讀一下
Status Validator::validate(Sentence* sentence, QueryContext* qctx) {
DCHECK(sentence != nullptr);
DCHECK(qctx != nullptr);
// Check if space chosen from session. if chosen, add it to context.
auto session = qctx->rctx()->session();
if (session->space().id > kInvalidSpaceID) {
auto spaceInfo = session->space();
qctx->vctx()->switchToSpace(std::move(spaceInfo));
}
// 3. 既然我們需要校驗該 sentence 是否符合我們的預期，則需要根據 sentence 的類型，創建一個 validator，記住目前是 seq_sentence
// 所以生成的就是 SequentialValidator，可以直接看下 makeValidator 函數的 switch case
auto validator = makeValidator(sentence, qctx);
// 4. 調用 validator 進行校驗，我們切換到下面的函數中
NG_RETURN_IF_ERROR(validator->validate());
auto root = validator->root();
if (!root) {
return Status::SemanticError("Get null plan from sequential validator");
}
qctx->plan()->setRoot(root);
return Status::OK();
}
// 5. 所有子類 validator，調用 validate 方法，進行校驗
// Validate current sentence.
// Check validator context, space, validate, duplicate reference columns,
// check permission according to sentence kind and privilege of user.
Status Validator::validate() {
if (!vctx_) {
VLOG(1) << "Validate context was not given.";
return Status::SemanticError("Validate context was not given.");
}
if (!sentence_) {
VLOG(1) << "Sentence was not given";
return Status::SemanticError("Sentence was not given");
}
if (!noSpaceRequired_ && !spaceChosen()) {
VLOG(1) << "Space was not chosen.";
return Status::SemanticError("Space was not chosen.");
}
if (!noSpaceRequired_) {
space_ = vctx_->whichSpace();
VLOG(1) << "Space chosen, name: " << space_.spaceDesc.space_name_ref().value()
<< " id: " << space_.id;
}
auto vidType = space_.spaceDesc.vid_type_ref().value().type_ref().value();
vidType_ = SchemaUtil::propTypeToValueType(vidType);
// 6. 調用子類 validateImpl
NG_RETURN_IF_ERROR(validateImpl());
// Check for duplicate reference column names in pipe or var statement
NG_RETURN_IF_ERROR(checkDuplicateColName());
// Execute after validateImpl because need field from it
if (FLAGS_enable_authorize) {
NG_RETURN_IF_ERROR(checkPermission());
}
// 7. 這裏是生成執行計劃調用
NG_RETURN_IF_ERROR(toPlan());
return Status::OK();
}
講了這麼久了，啥時候到 LOOKUP。只能説快了，因為第一次講源碼，一些上下文信息需要講清楚，不然大家一看就看得雲裏霧裏了。
3. 深入到 validator
下面，我們要進入 SequentialValidator.cpp 的 validateImpl() 去一探究竟。
// src/graph/validator/SequentialValidator.cpp
// Validator of sequential sentences which combine multiple sentences, e.g. GO ...; GO ...;
// Call validator of sub-sentences.
Status SequentialValidator::validateImpl() {
Status status;
if (sentence_->kind() != Sentence::Kind::kSequential) {
return Status::SemanticError(
"Sequential validator validates a SequentialSentences, but %ld is "
"given.",
static_cast<int64_t>(sentence_->kind()));
}
auto seqSentence = static_cast<SequentialSentences*>(sentence_);
auto sentences = seqSentence->sentences();
if (sentences.size() > static_cast<size_t>(FLAGS_max_allowed_statements)) {
return Status::SemanticError("The maximum number of statements allowed has been exceeded");
}
DCHECK(!sentences.empty());
// 我們的 StartNode 就是這裏創建出來的
seqAstCtx_->startNode = StartNode::make(seqAstCtx_->qctx);
// 一般序列語句中會放很多語句，也就是分號分隔的語句，這裏我們只有一條語句就是 lookup_sentence
// LOOKUP 語句創建出來 LookupValidator，終於看到曙光了
for (auto* sentence : sentences) {
auto validator = makeValidator(sentence, qctx_);
NG_RETURN_IF_ERROR(validator->validate());
seqAstCtx_->validators.emplace_back(std::move(validator));
}
return Status::OK();
}
4. 讀一讀 LookupValidator
終於，看到點 LOOKUP 的影子了，LookupValidator 駕到：
// src/graph/validator/LookupValidator.cpp
// LOOKUP 的 validateImpl 比較簡潔，直接對 From Where Yield e分別進行校驗
Status LookupValidator::validateImpl() {
lookupCtx_ = getContext<LookupContext>();
// 詳情請見下面的子函數分析
NG_RETURN_IF_ERROR(validateFrom());
// 此次不涉及，我們先不做分析
NG_RETURN_IF_ERROR(validateWhere());
// 詳情請見下面的子函數分析
NG_RETURN_IF_ERROR(validateYield());
return Status::OK();
}
// Validate specified schema(tag or edge) from sentence
Status LookupValidator::validateFrom() {
auto spaceId = lookupCtx_->space.id;
auto from = sentence()->from();
// 根據 spaceId 和指定的 label_name 查詢 Schema
auto ret = qctx_->schemaMng()->getSchemaIDByName(spaceId, from);
NG_RETURN_IF_ERROR(ret);
// 指定的是不是邊類型
lookupCtx_->isEdge = ret.value().first;
// 指定的 schemaId
lookupCtx_->schemaId = ret.value().second;
schemaIds_.emplace_back(ret.value().second);
return Status::OK();
}
// Validate yield clause.
Status LookupValidator::validateYield() {
auto yieldClause = sentence()->yieldClause();
if (yieldClause == nullptr) {
return Status::SemanticError("Missing yield clause.");
}
// 這個是判斷是否指定了 distinct 關鍵字，用於後續生成 dedup
lookupCtx_->dedup = yieldClause->isDistinct();
lookupCtx_->yieldExpr = qctx_->objPool()->makeAndAdd<YieldColumns>();
// 如果是邊類型，返回的列中，有 src、dst、rank、type
if (lookupCtx_->isEdge) {
idxReturnCols_.emplace_back(nebula::kSrc);
idxReturnCols_.emplace_back(nebula::kDst);
idxReturnCols_.emplace_back(nebula::kRank);
idxReturnCols_.emplace_back(nebula::kType);
// 校驗邊類型
NG_RETURN_IF_ERROR(validateYieldEdge());
} else { // 如果點類型、返回的列中有 vid
idxReturnCols_.emplace_back(nebula::kVid);
// 校驗點類型，這次我們介紹點類型的校驗
NG_RETURN_IF_ERROR(validateYieldTag());
}
if (exprProps_.hasInputVarProperty()) {
return Status::SemanticError("unsupport input/variable property expression in yield.");
}
if (exprProps_.hasSrcDstTagProperty()) {
return Status::SemanticError("unsupport src/dst property expression in yield.");
}
extractExprProps();
return Status::OK();
}
// Validate yield clause when lookup on tag.
// Disable invalid expressions, check schema name, rewrites expression to fit semantic,
// check type and collect properties.
Status LookupValidator::validateYieldTag() {
auto yield = sentence()->yieldClause();
auto yieldExpr = lookupCtx_->yieldExpr;
// yield 子句裏面的每一個逗號分隔的就是一個 col、我們的示例語句是 id(vertex)
// src/parser/parser.yy:1559 對 col 進行了定義
for (auto col : yield->columns()) {
// 如果發現表達式有 Edge 類型的，則直接把語義錯誤
if (ExpressionUtils::hasAny(col->expr(), {Expression::Kind::kEdge})) {
return Status::SemanticError("illegal yield clauses `%s'", col->toString().c_str());
}
// 如果是 label 屬性，則進行表達式名字的校驗，比如 yield player.name 這種語句
if (col->expr()->kind() == Expression::Kind::kLabelAttribute) {
const auto& schemaName = static_cast<LabelAttributeExpression*>(col->expr())->left()->name();
if (schemaName != sentence()->from()) {
return Status::SemanticError("Schema name error: %s", schemaName.c_str());
}
}
// 這塊應該是重寫表達式，有 label 屬性轉換為 Tag 的 prop，這裏不是特別清楚，後續精讀一下
col->setExpr(ExpressionUtils::rewriteLabelAttr2TagProp(col->expr()));
NG_RETURN_IF_ERROR(ValidateUtil::invalidLabelIdentifiers(col->expr()));
auto colExpr = col->expr();
// 推測表達式的類型
auto typeStatus = deduceExprType(colExpr);
NG_RETURN_IF_ERROR(typeStatus);
// 組織輸出，由名字和類型組成的集合對象
outputs_.emplace_back(col->name(), typeStatus.value());
yieldExpr->addColumn(col->clone().release());
NG_RETURN_IF_ERROR(deduceProps(colExpr, exprProps_, &schemaIds_));
}
return Status::OK();
}
到這裏，LOOKUP 的 validator 工作差不多完事了。
5. 語句如何變成執行計劃
介紹得不夠細緻，我還在熟悉過程，接下來就是介紹將 sentence 轉換成執行計劃的過程了。
執行計劃生成
執行計劃的生成，像是一些簡單的語句，就通過子類的 validator 的 toPlan 直接生成了，比如：SHOW HOSTS 這個語句，就是直接在 ShowHostsValidator::toPlan 方法中直接生成執行計劃。但是，對於一些比較複雜的語句來説，子類 validator 都沒有實現 toPlan 方法，也就是需要藉助父類的 toPlan 方法來生成執行計劃。比如，本文在讀的 LOOKUP 語句也屬於複雜語句：
// src/graph/validator/Validator.cpp
// 這裏就是複雜語句生成執行計劃的入口
// 需要配合 AstContext 來生成，對於 LOOKUP 語句來説，就是 LookupContext
// Call planner to get final execution plan.
Status Validator::toPlan() {
// **去子類 LookupValidator 的 getAstContext() 方法看下，是不是返回的是 LookupContext**
auto* astCtx = getAstContext();
if (astCtx != nullptr) {
astCtx->space = space_;
}
// 利用抽象語法樹上下文，借用 Planner 的 toPlan 生成具體的執行計劃
auto subPlanStatus = Planner::toPlan(astCtx);
NG_RETURN_IF_ERROR(subPlanStatus);
auto subPlan = std::move(subPlanStatus).value();
// 將返回的 subPlan 對 root 和 tail 進行填充
root_ = subPlan.root;
tail_ = subPlan.tail;
VLOG(1) << "root: " << root_->kind() << " tail: " << tail_->kind();
return Status::OK();
}
6. 進入 toPlan() 一探究竟
從章節 5. 上面獲知，需要進入 Planner 的 toPlan 方法一探究竟
// src/graph/planner/Planner.cpp
StatusOr<SubPlan> Planner::toPlan(AstContext* astCtx) {
if (astCtx == nullptr) {
return Status::Error("AstContext nullptr.");
}
const auto* sentence = astCtx->sentence;
DCHECK(sentence != nullptr);
// 從抽象語法樹的執行上下文取到我們的 sentence
// 下面的 plannerMap 是我們在 src/graph/planner/PlannersRegister.cpp 註冊好的，一些複雜的語句都在這裏註冊好了
auto planners = plannersMap().find(sentence->kind());
if (planners == plannersMap().end()) {
return Status::Error("No planners for sentence: %s", sentence->toString().c_str());
}
for (auto& planner : planners->second) { // second 是語句具體對應的 planner 的實例化對象： MatchAndInstantiate
if (planner.match(astCtx)) { // match 方法是具體 planner 的 match 方法，對應到 LookupPlaner，就是 match
// 這裏的 instantiate 是 LookupPlanner 的 make 方法
// 這裏的 transform 是拿着 lookupcontext 生成執行計劃的函數
return planner.instantiate()->transform(astCtx);
}
}
return Status::Error("No planner matches sentence: %s", sentence->toString().c_str());
}
7. 計劃中的 transform()
我們分析到這裏，使用了 Planner 的 toPlan 方法生成一些複雜語句的執行計劃。接下來，就是進去 LookupPlanner 的 transform 方法從 LookupContext 轉換到執行計劃的過程了。我們直接定位到 LookupPlanner 的 transform 方法上：
// src/graph/planner/ngql/LookupPlanner.cpp
StatusOr<SubPlan> LookupPlanner::transform(AstContext* astCtx) {
// 是不是我們上面提到的 lookupContext
auto lookupCtx = static_cast<LookupContext*>(astCtx);
auto qctx = lookupCtx->qctx;
// ON 後面的 name_label
auto from = static_cast<const LookupSentence*>(lookupCtx->sentence)->from();
SubPlan plan;
// 如果是邊的話，生成的是 EdgeIndexFullScan
if (lookupCtx->isEdge) {
auto* edgeIndexFullScan = EdgeIndexFullScan::make(qctx,
nullptr,
from,
lookupCtx->space.id,
{},
lookupCtx->idxReturnCols,
lookupCtx->schemaId,
lookupCtx->isEmptyResultSet);
edgeIndexFullScan->setYieldColumns(lookupCtx->yieldExpr);
plan.tail = edgeIndexFullScan;
plan.root = edgeIndexFullScan;
} else { // 如果是點的話，生成的是 TagIndexFullScan
auto* tagIndexFullScan = TagIndexFullScan::make(qctx,
nullptr,
from,
lookupCtx->space.id,
{},
lookupCtx->idxReturnCols,
lookupCtx->schemaId,
lookupCtx->isEmptyResultSet);
tagIndexFullScan->setYieldColumns(lookupCtx->yieldExpr);
plan.tail = tagIndexFullScan;
plan.root = tagIndexFullScan;
}
plan.tail->setColNames(lookupCtx->idxColNames);
// 我們沒有指定 where 語句，所以不會有 filter 算子
if (lookupCtx->filter) {
plan.root = Filter::make(qctx, plan.root, lookupCtx->filter);
}
// 會有 Project 算子生成：對輸出列做一個映射
plan.root = Project::make(qctx, plan.root, lookupCtx->yieldExpr);
// 這裏是 distinct 關鍵字，我們沒有指定，默認是沒有這個算子的
if (lookupCtx->dedup) {
plan.root = Dedup::make(qctx, plan.root);
}
return plan;
}
8. explain 驗證生成的執行計劃
通過我們上述的介紹，執行計劃已經生成了。那麼，我們是不是可以通過 explain 或者 profile 來驗證我們分析生成的執行計劃就是 Project→TagIndexFullScan→Start 呢。下面是我們通過 explain 生成的執行計劃，它驗證了我們分析的源碼和生成的執行計劃是一致的。 大喜😊
(root@nebula) [basketballplayer]> explain lookup on player yield id(vertex)
Execution succeeded (time spent 615µs/1.057064ms)
Execution Plan (optimize time 42 us)
-----+------------------+--------------+----------------+-----------------------------------
| id | name | dependencies | profiling data | operator info |
-----+------------------+--------------+----------------+-----------------------------------
| 2 | Project | 3 | | outputVar: { |
| | | | | "colNames": [ |
| | | | | "id(VERTEX)" |
| | | | | ], |
| | | | | "type": "DATASET", |
| | | | | "name": "__Project_2" |
| | | | | } |
| | | | | inputVar: __TagIndexFullScan_1 |
| | | | | columns: [ |
| | | | | "id(VERTEX)" |
| | | | | ] |
-----+------------------+--------------+----------------+-----------------------------------
| 3 | TagIndexFullScan | 0 | | outputVar: { |
| | | | | "colNames": [ |
| | | | | "_vid", |
| | | | | "player._tag", |
| | | | | "player.age", |
| | | | | "player.name" |
| | | | | ], |
| | | | | "type": "DATASET", |
| | | | | "name": "__TagIndexFullScan_1" |
| | | | | } |
| | | | | inputVar: |
| | | | | space: 6 |
| | | | | dedup: false |
| | | | | limit: 9223372036854775807 |
| | | | | filter: |
| | | | | orderBy: [] |
| | | | | schemaId: 7 |
| | | | | isEdge: false |
| | | | | returnCols: [ |
| | | | | "_vid", |
| | | | | "_tag", |
| | | | | "age", |
| | | | | "name" |
| | | | | ] |
| | | | | indexCtx: [ |
| | | | | { |
| | | | | "columnHints": [], |
| | | | | "filter": "", |
| | | | | "index_id": 11 |
| | | | | } |
| | | | | ] |
-----+------------------+--------------+----------------+-----------------------------------
| 0 | Start | | | outputVar: { |
| | | | | "colNames": [], |
| | | | | "type": "DATASET", |
| | | | | "name": "__Start_0" |
| | | | | } |
-----+------------------+--------------+----------------+-----------------------------------
階段小結
源碼閲讀到這裏，我們知道 Graph 層從一個 nGQL 語句，到生成執行計劃的所有過程。當中可能有一些細節沒有面面俱到，但是，我們應該整體對代碼有了初步瞭解。
9. 調度執行計劃
接下來，我們要了解執行計劃是如何被物理執行、Executor 是如何調度執行計劃的。目前，我們只涉及到三個物理算子的執行，而且 Start 節點是一個沒有實際語義的算子。這裏我們仔細分析一下 TagIndexScan 和 Project 算子。
我們需要先回到第二章節的註釋 7 那裏了。註釋 5 我們就不講了，那裏是內核語句 RBO 規則對執行計劃進行優化的子模塊，我們的簡單語句的執行計劃不涉及這塊，留下後續擴展介紹吧。
// src/graph/scheduler/AsyncMsgNotifyBasedScheduler.cpp
// 我們回到了註釋 7 那裏，對 scheduler_ 的 shcedule 方法解讀一下
// 然後我們再看 LOOKUP 語句的兩個物理算子在這裏是怎麼執行的
// 目前內核只實現了基於消息的異步調度器
folly::Future<Status> AsyncMsgNotifyBasedScheduler::schedule() {
// 拿到執行計劃的 root 節點，在這次的語句中，就是 Project
auto root = qctx_->plan()->root();
// 這塊還沒有深入解讀過，後續再擴展吧
if (FLAGS_enable_lifetime_optimize) {
// special for root
root->outputVarPtr()->userCount.store(std::numeric_limits<uint64_t>::max(),
std::memory_order_relaxed);
analyzeLifetime(root);
}
// 遞歸將執行計劃 convert 到物理執行計劃 Executor，也就是 Project->ProjectExecutor, TagindexFullScan->IndexScanExecutor
// 把物理 Executor 的拓撲結構創建出來
// ProjectExecutor 依賴 IndexScanExecutor IndexScanExecutor 的後繼是 ProjectExecutor
// IndexScanExecutor 依賴 StartExecutor StartExecutor 的後繼是 IndexScanExecutor
auto executor = Executor::create(root, qctx_);
// 這裏開始 DAG 的物理計劃執行
// 調度是基於 folly 的 Promise 和 Future 異步調用展開的
return doSchedule(executor);
}
folly::Future<Status> AsyncMsgNotifyBasedScheduler::doSchedule(Executor* root) const {
// 這個是按照算子的 id，承諾給別的算子的 promise（你可以理解為誰依賴這個算子，那麼就給誰一個 promise）
std::unordered_map<int64_t, std::vector<folly::Promise<Status>>> promiseMap;
// 這個是當前算子，被誰許諾過的 future，是從 promise 那裏或者的結果值。也就是説，如果這個算子依賴了某些算子，只有它們的許諾兑現了（promise set value），這裏的 future 才能得到處理
std::unordered_map<int64_t, std::vector<folly::Future<Status>>> futureMap;
// 這個 queue 是為了輔助算子生成 promiseMap 和 futureMap 的
std::queue<Executor*> queue;
// 這個 queue2 是為結合剛才生成的 promiseMap 和 futureMap 實際進行調度運行的
std::queue<Executor*> queue2;
// 算子節點訪問標記，避免重複遍歷
std::unordered_set<Executor*> visited;
auto* runner = qctx_->rctx()->runner();
// 首先把 root 的 promise 出來，這個對於我們的執行計劃中的算子就是 Project
folly::Promise<Status> promiseForRoot;
auto resultFuture = promiseForRoot.getFuture();
promiseMap[root->id()].emplace_back(std::move(promiseForRoot));
queue.push(root);
visited.emplace(root);
// 開始 DAG 訪問圖計算節點，生成每一個節點的 promise 和 future
while (!queue.empty()) {
auto* exe = queue.front();
queue.pop();
queue2.push(exe);
std::vector<folly::Future<Status>>& futures = futureMap[exe->id()];
if (exe->node()->kind() == PlanNode::Kind::kArgument) {
auto nodeInputVar = exe->node()->inputVar();
const auto& writtenBy = qctx_->symTable()->getVar(nodeInputVar)->writtenBy;
for (auto& node : writtenBy) {
folly::Promise<Status> p;
futures.emplace_back(p.getFuture());
auto& promises = promiseMap[node->id()];
promises.emplace_back(std::move(p));
}
} else {
for (auto* dep : exe->depends()) {
auto notVisited = visited.emplace(dep).second;
if (notVisited) {
queue.push(dep);
}
folly::Promise<Status> p;
futures.emplace_back(p.getFuture());
auto& promises = promiseMap[dep->id()];
promises.emplace_back(std::move(p));
}
}
}
// 開始調度執行，下面的 scheduleExecutor 這個方法是關鍵
// 這個方法是純異步運行的，比如運行 ProjectExecutor，它的依賴是 IndexScanExecutor
// 那麼 ProjectExecutor 的 future 就來自於 IndexScanExecutor 的 promise
// ProjectExecutor 需要在 folly::collect 出等待 IndexScanExecutor 的執行結束
// 這樣 ProjectExecutor 才可以得到執行的機會
while (!queue2.empty()) {
auto* exe = queue2.front();
queue2.pop();
auto currentFuturesFound = futureMap.find(exe->id());
DCHECK(currentFuturesFound != futureMap.end());
auto currentExeFutures = std::move(currentFuturesFound->second);
auto currentPromisesFound = promiseMap.find(exe->id());
DCHECK(currentPromisesFound != promiseMap.end());
auto currentExePromises = std::move(currentPromisesFound->second);
scheduleExecutor(std::move(currentExeFutures), exe, runner)
.thenTry([this, pros = std::move(currentExePromises)](auto&& t) mutable {
if (t.hasException()) {
notifyError(pros, Status::Error(std::move(t).exception().what()));
} else {
auto v = std::move(t).value();
if (v.ok()) {
notifyOK(pros); // **Promise填充：成功以後具體填充promise的地方**
} else {
notifyError(pros, v);
}
}
});
}
return resultFuture;
}
// 你可以把這個函數理解為異步調度器，上面把所有的算子通過這個函數進行了調度
// 第一個參數包含了該算子所有的 futures，也就是這個算子依賴算子的 promise 需要執行結束，這裏的 futures 才可以獲取到結果
// 第二個參數是該算子的 Executor
// 第三個參數是執行器，你可以理解為線程池
// 根據不同的算子類型，實現不同的分支運行，我們上面的語句是走 default 分支
// lookup on player yield id(vertex)；語句整體的調度過程
// ProjectExecutor(P)->IndexScanExecutor(I)->Start(S）執行計劃。下面我們用簡寫來表示三個算子
// 首先 P 算子調度以後，它到了 default 分支，depends 不為空，那麼走 runExecutor
// P 算子的 future 就來自於 I 算子的 promise，所以需要等待 I 算子的執行結束
// I 算子調度到這個函數以後，它到了 default 分支，depends 不為空，那麼走 runExecutor
// I 算子的 future 就來自於 S 算子的 promise，所以需要等待 S 算子的執行結束
// S 算子調度到這個函數以後，它到了 default 分支，depends 為空，那麼走 runLeafExecutor
// S 算子就開始 execute 的邏輯了，可以去看看 StartExecutor 的 executor 方法，啥也沒幹，所以之前説 start 算子沒啥語義
// S 算子結束以後，它的 promise 被填充，其實是上面那個函數的回調填充的，具體看我上面的註釋 **Promise 填充**
// 那麼 I 算子的 future 就得到了響應，去 runExecutor 看看，是不是也是有一個回調，立馬發起了 I 算子的調用
// 當 I 算子的 promise 也被上面的函數填充
// 那麼 P 算子的 executor 也得到了執行，這下就算執行完
folly::Future<Status> AsyncMsgNotifyBasedScheduler::scheduleExecutor(
std::vector<folly::Future<Status>>&& futures, Executor* exe, folly::Executor* runner) const {
switch (exe->node()->kind()) {
case PlanNode::Kind::kSelect: {
auto select = static_cast<SelectExecutor*>(exe);
return runSelect(std::move(futures), select, runner);
}
case PlanNode::Kind::kLoop: {
auto loop = static_cast<LoopExecutor*>(exe);
return runLoop(std::move(futures), loop, runner);
}
case PlanNode::Kind::kArgument: {
return runExecutor(std::move(futures), exe, runner);
}
default: {
if (exe->depends().empty()) {
return runLeafExecutor(exe, runner);
} else {
return runExecutor(std::move(futures), exe, runner);
}
}
}
}
10. LOOKUP 語句的算子在執行什麼？
上面我介紹了物理算子通過 folly 三方庫的 Promise 和 Future 異步編程模型來實現調度執行。接下來，重點介紹一下我們本次 LOOKUP 語句中兩個算子執行了什麼。源碼走起：上面的語句主要介紹了三個物理算子：ProjectExecutor、IndexScanExecutor、StartExecutor。這裏多説一句，因為和 IndexScan 有關的算子都會映射到 IndexScanExecutor。
// StartExecutor：啥也沒幹
// IndexScanExecutor：是主要幹活的，需要 graph 和 storage 的 rpc，拉取數據
// ProjectExecutor：這個物理執行算子不需要和 storage 交互，直接在 graph 層閉環計算
// 這三個算子，我們只分析後兩個算子的源碼：
// src/graph/executor/query/IndexScanExecutor.cpp
folly::Future<Status> IndexScanExecutor::execute() {
return indexScan();
}
folly::Future<Status> IndexScanExecutor::indexScan() {
// 拿到和 storage 交互的 storageClient
StorageClient *storageClient = qctx_->getStorageClient();
auto *lookup = asNode<IndexScan>(node());
if (lookup->isEmptyResultSet()) {
DataSet dataSet({"dummy"});
return finish(ResultBuilder().value(Value(std::move(dataSet))).build());
}
const auto &ictxs = lookup->queryContext();
auto iter = std::find_if(
ictxs.begin(), ictxs.end(), [](auto &ictx) { return !ictx.index_id_ref().is_set(); });
if (ictxs.empty() || iter != ictxs.end()) {
return Status::Error("There is no index to use at runtime");
}
// Req 的公共請求參數
StorageClient::CommonRequestParam param(lookup->space(),
qctx()->rctx()->session()->id(),
qctx()->plan()->id(),
qctx()->plan()->isProfileEnabled());
return storageClient
->lookupIndex(param,
ictxs,
lookup->isEdge(), // 是不是邊類型
lookup->schemaId(), // schemaId
lookup->returnColumns(), // resp 返回的列數據
lookup->orderBy(), // 是否帶有 orderBy，為了下推 TopN 算子
lookup->limit(qctx_)) // 是否帶有 limit，為了下推 limit 算子
.via(runner())
.thenValue([this](StorageRpcResponse<LookupIndexResp> &&rpcResp) {
addStats(rpcResp, otherStats_);
return handleResp(std::move(rpcResp));
});
}
// TODO(shylock) merge the handler with GetProp
template <typename Resp>
Status IndexScanExecutor::handleResp(storage::StorageRpcResponse<Resp> &&rpcResp) {
auto completeness = handleCompleteness(rpcResp, FLAGS_accept_partial_success);
if (!completeness.ok()) {
return std::move(completeness).status();
}
auto state = std::move(completeness).value();
nebula::DataSet v;
// 把每一個 resp 拉出來處理，因為我們 storage 是可以分佈式部署的
// 這裏有一個問題重點提出一下，結果集會維護在 ectx_ 中，供 ProjectExecutor 一會取
for (auto &resp : rpcResp.responses()) {
if (resp.data_ref().has_value()) {
nebula::DataSet &data = *resp.data_ref();
// TODO: convert the column name to alias.
if (v.colNames.empty()) {
v.colNames = data.colNames;
}
v.rows.insert(v.rows.end(), data.rows.begin(), data.rows.end());
} else {
state = Result::State::kPartialSuccess;
}
}
if (!node()->colNames().empty()) {
DCHECK_EQ(node()->colNames().size(), v.colNames.size());
v.colNames = node()->colNames();
}
return finish(
ResultBuilder().value(std::move(v)).iter(Iterator::Kind::kProp).state(state).build());
}
// src/graph/executor/query/ProjectExecutor.cpp
folly::Future<Status> ProjectExecutor::execute() {
SCOPED_TIMER(&execTime_);
auto *project = asNode<Project>(node());
// 剛才説從 storage 獲取的結果數據都放在 ectx_ 裏面了
auto iter = ectx_->getResult(project->inputVar()).iter();
DCHECK(!!iter);
QueryExpressionContext ctx(ectx_);
// 默認 max_job_size 是 1，我們先看 if 分支，看 handleJob 到底幹了啥
if (FLAGS_max_job_size <= 1) {
auto ds = handleJob(0, iter->size(), iter.get());
return finish(ResultBuilder().value(Value(std::move(ds))).build());
} else {
DataSet ds;
ds.colNames = project->colNames();
ds.rows.reserve(iter->size());
auto scatter = [this](size_t begin, size_t end, Iterator *tmpIter) -> StatusOr<DataSet> {
return handleJob(begin, end, tmpIter);
};
auto gather = [this, result = std::move(ds)](auto &&results) mutable {
for (auto &r : results) {
auto &&rows = std::move(r).value();
result.rows.insert(result.rows.end(),
std::make_move_iterator(rows.begin()),
std::make_move_iterator(rows.end()));
}
finish(ResultBuilder().value(Value(std::move(result))).build());
return Status::OK();
};
return runMultiJobs(std::move(scatter), std::move(gather), iter.get());
}
}
DataSet ProjectExecutor::handleJob(size_t begin, size_t end, Iterator *iter) {
auto *project = asNode<Project>(node());
auto columns = project->columns()->clone();
DataSet ds;
ds.colNames = project->colNames();
QueryExpressionContext ctx(qctx()->ectx());
ds.rows.reserve(end - begin);
// 從頭到尾遍歷數據，去除關心的數據
for (; iter->valid() && begin++ < end; iter->next()) {
Row row;
for (auto &col : columns->columns()) {
Value val = col->expr()->eval(ctx(iter)); // 這個是表達式的 eval 執行，對於我們 id(vertex) 對應的是：src/common/function/FunctionManager.cpp:1832 auto &attr = functions_["id"];
row.values.emplace_back(std::move(val)); // 這個對於 id(vertex) 的 val 來説，就是 vertex.id
ds.rows.emplace_back(std::move(row));
}
return ds;
}
11. 數據結果顯示
我們通過物理執行算子，把數據放在最後一個算子的 ProjectExecutor 的 ectx_(ExecutionContext) 裏面了。我們接下來就是要知道，哪個流程把這個執行上下文的數據取走了：給客户端的 resp 填充這些數據，最終顯示到我們的 nebula-console，或者其他客户端中。Its time to go back to 章節 2. 的註釋 8：
// 請看第二步的註釋 8：
this->onFinish(); // 8. 這裏是幹完了所有物理執行計劃，然後開始處理客户端 resp 了
// 我們進到 onFinish 函數看下：
void QueryInstance::onFinish() {
auto rctx = qctx()->rctx();
VLOG(1) << "Finish query: " << rctx->query();
auto &spaceName = rctx->session()->space().name;
rctx->resp().spaceName = std::make_unique<std::string>(spaceName);
// 這個函數做了填充結果數據到 resp 中
fillRespData(&rctx->resp());
auto latency = rctx->duration().elapsedInUSec();
rctx->resp().latencyInUs = latency;
addSlowQueryStats(latency, spaceName);
rctx->finish();
rctx->session()->deleteQuery(qctx_.get());
// The `QueryInstance' is the root node holding all resources during the
// execution. When the whole query process is done, it's safe to release this
// object, as long as no other contexts have chances to access these resources
// later on, e.g. previously launched uncompleted async sub-tasks, EVEN on
// failures.
delete this;
}
// 把執行的數據從 ectx 中取出，然後填充到執行 resp 中，這次語句執行就結束了
// Get result from query context and fill the response
void QueryInstance::fillRespData(ExecutionResponse *resp) {
auto ectx = DCHECK_NOTNULL(qctx_->ectx());
auto plan = DCHECK_NOTNULL(qctx_->plan());
const auto &name = plan->root()->outputVar();
if (!ectx->exist(name)) return;
auto &&value = ectx->moveValue(name);
if (!value.isDataSet()) return;
// Fill dataset
auto result = value.moveDataSet();
if (!result.colNames.empty()) {
// 結果填充
resp->data = std::make_unique<DataSet>(std::move(result));
} else {
// 如果有錯誤，錯誤碼和錯誤信息
resp->errorCode = ErrorCode::E_EXECUTION_ERROR;
resp->errorMsg = std::make_unique<std::string>("Internal error: empty column name list");
LOG(ERROR) << "Empty column name list";
}
}
小結
目前為止，我們把 LOOKUP 是怎麼在內核中執行的一生的源碼解讀就做完了。有很多細節沒有展開，後續的文章中我們將不斷展開。其實，對於任意一個語句，基本執行的流程和 LOOKUP 的一生都類似，其中有不同的地方就是額外的算子不同，算子之間處理的邏輯不同。而且，這次我們沒有打開 Storage 服務的代碼，可以作為一個遺留項。
祝大家都可以在 NebulaGraph 圖數據庫的源碼世界裏面翱翔，歡迎大家和我來進行交流，學習 Wey Gu 的方式，給大家留一個微信聯繫方式：echo TWlsaXR0bGVUaW1l | base64 -d Call me.
謝謝你讀完本文 (///▽///)
要來近距離體驗一把圖數據庫嗎？現在可以用用 NebulaGraph Cloud 來搭建自己的圖數據系統喲，快來節省大量的部署安裝時間來搞定業務吧~ NebulaGraph 阿里雲計算巢現 30 天免費使用中，點擊鏈接來用用圖數據庫吧~
想看源碼的小夥伴可以前往 GitHub 閲讀、使用、(^з^)-☆ star 它 -> GitHub；和其他的 NebulaGraph 用户一起交流圖數據庫技術和應用技能，留下「你的名片」一起玩耍呢~
「其他文章」
- 圖數據庫在中國移動金融風控的落地應用
- 記一次 rr 和硬件斷點解決內存踩踏問題
- 用圖技術搞定附近好友、時空交集等 7 個典型社交網絡應用
- 用圖技術搞定附近好友、時空交集等 7 個典型社交網絡應用
- 圖數據庫中的“分佈式”和“數據切分”（切圖）
- 揭祕可視化圖探索工具 NebulaGraph Explore 是如何實現圖計算的
- 連接微信羣、Slack 和 GitHub：社區開放溝通的基礎設施搭建
- 圖數據庫認證考試 NGCP 錯題解析 vol.02：這 10 道題竟無一人全部答對
- 如何判斷多賬號是同一個人？用圖技術搞定 ID Mapping
- 複雜場景下圖數據庫的 OLTP 與 OLAP 融合實踐
- 如何運維多集羣數據庫？58 同城 NebulaGraph Database 運維實踐
- 有了 ETL 數據神器 dbt，表數據秒變 NebulaGraph 中的圖數據
- 基於圖的下一代入侵檢測系統
- 從實測出發，掌握 NebulaGraph Exchange 性能最大化的祕密
- 讀 NebulaGraph源碼 ｜ 查詢語句 LOOKUP 的一生
- 當雲原生網關遇上圖數據庫，NebulaGraph 的 APISIX 最佳實踐
- 從全球頂級數據庫大會 SIGMOD 看數據庫發展趨勢
- 「實操」結合圖數據庫、圖算法、機器學習、GNN 實現一個推薦系統
- 如何輕鬆做數據治理？開源技術棧告訴你答案
- 圖算法、圖數據庫在風控場景的應用